@@ -0,0 +1,358 @@
1+-- | This is a log scribe that writes logs to logz.io's bulk
2+-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS HTTPS
3+-- API>.
4+module Katip.Scribes.LogzIO.HTTPS
5+ ( -- * Scribe construction
6+ mkLogzIOScribe
7+ -- * Types
8+ , BulkAPIError(..)
9+ , LogzIOScribeConfiguration(..)
10+ ) where
11+
12+
13+-------------------------------------------------------------------------------
14+import Control.Applicative
15+import qualified Control.Concurrent.Async as Async
16+import qualified Control.Concurrent.STM as STM
17+import qualified Data.Aeson as A
18+import qualified Data.ByteString.Builder as BB
19+import qualified Data.ByteString.Lazy as LBS
20+import qualified Data.HashMap.Strict as HM
21+import Data.Int
22+import qualified Data.Text.Lazy as TL
23+import qualified Data.Text.Lazy.Builder as TB
24+import qualified Data.Time as Time
25+import qualified Katip as K
26+import Katip.Core (LocJs (..))
27+import qualified System.Posix.Types as POSIX
28+-------------------------------------------------------------------------------
29+-------------------------------------------------------------------------------
30+
31+
32+
33+-- | This is returned when the bulk import was a partial success
34+data BulkAPIError = BulkAPIError
35+ { bulkAPIError_malformedLines :: Int
36+ -- ^ The number of log lines which are not well-formed JSON. This
37+ -- indicates a __library bug__. Please file an issue on GitHub.
38+ , bulkAPIError_successfulLines :: Int
39+ -- ^ The number of log lines received successfully.
40+ , bulkAPIError_oversizedLines :: Int
41+ -- ^ The number of log lines which exceed the line length
42+ -- limit. katip-logzio makes a best effort to truncate logs that
43+ -- exceed the size limit. This probably indicates a __library bug__
44+ -- and should be reported as an issue on GitHub.
45+ , bulkAPIError_emptyLogLines :: Int
46+ -- ^ The number of log lines which were empty. There isnt' really a
47+ -- concept of a truly empty log line in katip, so this most likely
48+ -- indicates a __library bug__ and should be reported as an issue on
49+ -- GitHub.
50+ }
51+
52+
53+-------------------------------------------------------------------------------
54+data LogzIOScribeConfiguration = LogzIOScribeConfiguration
55+ { logzIOScribeConfiguration_bufferItems :: Int
56+ -- ^ Will flush the log buffer if this many items is in the buffer
57+ -- __or__ 'logzIOScribeConfiguration_bufferTimeout' is reached,
58+ -- whichever is first
59+ , logzIOScribeConfiguration_bufferTimeout :: Time.NominalDiffTime
60+ -- ^ Will flush the buffer if it has been this long since the last
61+ -- flush __or__ 'logzIOScribeConfiguration_bufferItems' items are
62+ -- accumulated, whichever is first.
63+ --TODO: more
64+ }
65+
66+-------------------------------------------------------------------------------
67+mkLogzIOScribe
68+ :: LogzIOScribeConfiguration
69+ -> K.PermitFunc
70+ -> K.Verbosity
71+ -> IO K.Scribe
72+mkLogzIOScribe config permitItem verbosity = do
73+ -- Accumulating buffer for each item
74+ bufferState <- STM.newTVarIO (mempty :: BulkBuffer)
75+ -- Single-slot TMVar when something is ready to write. Katip buffers
76+ -- for us so we can simplify shutdown and flush guarantees this way
77+ readyMVar <- STM.newEmptyTMVarIO :: IO (STM.TMVar BulkBuffer)
78+
79+ -- We create a TVar that is set to ShutDown when its time to stop
80+ killSwitch <- STM.newTVarIO Continue
81+
82+ -- Create a mutable ref to whether the timer has expired. This lets
83+ -- us replace the timer when its time to start timing again. We
84+ -- reset the timer __after__ a write goes out, not as soon as it expires
85+ timerRef <- STM.atomically $ do
86+ -- If there's never been a log, no need to start the timer
87+ never <- STM.newTVar False
88+ STM.newTVar never
89+
90+
91+ -- on every push, extend deadline
92+ -- thread waits for deadline and pushes. need a way to be atomic though, cancelthread is heavy handed
93+ --TODO: strictness
94+
95+ -- in STM: resolve an action with the buffers
96+ let processLogAction logAction = case logAction of
97+ Buffered newBuffer -> STM.writeTVar bufferState newBuffer
98+ FlushNow toFlush newBuffer -> do
99+ STM.writeTVar bufferState newBuffer
100+ STM.putTMVar readyMVar toFlush
101+
102+ -- In recompute the new buffer state from a new log item and take action
103+ let push logItem = do
104+ curBuffer <- STM.readTVar bufferState
105+ processLogAction (bufferItem itemBufferSize verbosity logItem curBuffer)
106+
107+
108+ -- A transaction that retries until its time to shutdown
109+ let waitShutdown = do
110+ killStatus <- STM.readTVar killSwitch
111+ STM.check (killStatus == ShutDown)
112+
113+ --TODO: exception handling
114+
115+ -- Retry transaction until timer has expired
116+ let waitTimer = do
117+ -- get the timer we'll be using
118+ timer <- STM.readTVar timerRef
119+ -- do not proceed until its ready
120+ timerExpired <- STM.readTVar timer
121+ STM.check timerExpired
122+ -- do not wake again until this message is out the door
123+ STM.writeTVar timer False
124+ -- we waited long enough, grab the current buffer
125+ curBuffer <- STM.readTVar bufferState
126+ -- and tell it to flush right now
127+ processLogAction (forceFlush curBuffer)
128+
129+ -- loop until its time to shut down
130+ let timedFlushLoop = do
131+ res <- STM.atomically $
132+ raceAlt
133+ waitShutdown
134+ waitTimer
135+ case res of
136+ Left () -> pure () -- exit loop
137+ Right () -> timedFlushLoop -- wait for another timer or shutdown
138+
139+ timedFlusher <- Async.async timedFlushLoop
140+
141+ let flushLoop = do
142+ -- hold until something comes in to flush or we're told to stop. If we're told to stop but get a log first, we'll catch it on the next loop
143+ readyBufferOrDie <- STM.atomically $
144+ raceAlt
145+ (STM.takeTMVar readyMVar)
146+ waitShutdown
147+ case readyBufferOrDie of
148+ Right () -> pure () -- we caught shutdown before any new data arrived
149+ -- There's some logs to write
150+ Left readyBuffer -> do
151+ -- flush in IO
152+ flushBuffer readyBuffer
153+ -- reset the timeout before the next flush on each write
154+ newTimer <- STM.registerDelay itemBufferTimeoutMicros
155+ STM.atomically $ do
156+ STM.writeTVar timerRef newTimer
157+ -- start over
158+ flushLoop
159+ flusher <- Async.async flushLoop
160+
161+ let shutdown = do
162+ -- tell workers this is the last loop
163+ STM.atomically (STM.writeTVar killSwitch ShutDown)
164+ -- wait for workers to finish
165+ _ <- Async.waitCatch timedFlusher
166+ _ <- Async.waitCatch flusher
167+ -- stop blocking
168+ pure ()
169+
170+
171+ pure $ K.Scribe
172+ { liPush = STM.atomically . push
173+ , scribeFinalizer = shutdown
174+ , scribePermitItem = permitItem
175+ }
176+ where
177+ itemBufferSize = logzIOScribeConfiguration_bufferItems config
178+ itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
179+
180+
181+-------------------------------------------------------------------------------
182+-- Match the native resolution of NominalDiffTime, which is excessive
183+-- for our needs
184+ndtPicos :: Time.NominalDiffTime -> Int64
185+ndtPicos = round . (* picos)
186+ where
187+ picos :: Time.NominalDiffTime
188+ picos = 10 ^ (9 :: Int)
189+
190+
191+-------------------------------------------------------------------------------
192+ndtToMicros :: Time.NominalDiffTime -> Int
193+ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
194+ where
195+ picosInMicro = 10 ^ (3 :: Int)
196+
197+
198+-------------------------------------------------------------------------------
199+data KillSwitch =
200+ Continue
201+ | ShutDown
202+ deriving (Eq)
203+
204+
205+-------------------------------------------------------------------------------
206+raceAlt :: (Alternative m) => m a -> m b -> m (Either a b)
207+raceAlt a b = (Left <$> a) <|> (Right <$> b)
208+
209+
210+-------------------------------------------------------------------------------
211+-- TODO: noop on empty
212+flushBuffer :: BulkBuffer -> IO ()
213+flushBuffer = undefined
214+
215+-------------------------------------------------------------------------------
216+newtype Bytes = Bytes
217+ { bytes :: Int64
218+ } deriving (Show, Eq, Num, Ord)
219+
220+
221+-- | How big of a body can we send? The limit is defined
222+-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS here>.
223+maxPayloadBytes :: Bytes
224+maxPayloadBytes = 10485760
225+
226+
227+-- | How long can each serialized payload be (let's assume including
228+-- the trailing newline). The limit is defined
229+-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS here>.
230+maxLogLineLength :: Bytes
231+maxLogLineLength = 500000
232+
233+
234+measureJSON :: A.ToJSON a => a -> (BB.Builder, Bytes)
235+measureJSON a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
236+ where
237+ lbs = A.encode a
238+
239+
240+-- | Fully-rendered JSON object for an item
241+fullItemObject :: K.LogItem a => K.Verbosity -> K.Item a -> A.Object
242+fullItemObject verbosity item = HM.fromList
243+ [ "app" A..= K._itemApp item
244+ , "env" A..= K._itemEnv item
245+ , "sev" A..= K._itemSeverity item
246+ , "thread" A..= K.getThreadIdText (K._itemThread item)
247+ , "host" A..= K._itemHost item
248+ , "pid" A..= pidInt
249+ , "data" A..= K.payloadObject verbosity (K._itemPayload item)
250+ -- Slight deviation, logz.io uses "message" instead of "msg"
251+ , "message" A..= TB.toLazyText (K.unLogStr (K._itemMessage item))
252+ -- Another slight deviation, logz.io uses "@timestamp" instead
253+ -- of "at". Note, your logs should be sent roughly close to when
254+ -- they are created. They are assigned to indexes based on index
255+ -- date, so time searches act strange if you backfill too far
256+ -- from the past.
257+ , "@timestamp" A..= K._itemTime item --TODO: explicit formatting as per spec?
258+ , "ns" A..= K._itemNamespace item
259+ , "loc" A..= (LocJs <$> K._itemLoc item)
260+ ]
261+ where
262+ POSIX.CPid pidInt = K._itemProcess item
263+
264+--TODO: property test
265+
266+-- | A version of 'renderLine' which renders a line and stays under
267+-- the maximum line size of 500,000 bytes. If the default rendering is
268+-- too large, the log will be reduced to a timestamp and a potentially
269+-- truncated message.
270+renderLineTruncated :: K.LogItem a => K.Verbosity -> K.Item a -> (BB.Builder, Bytes)
271+renderLineTruncated verbosity item =
272+ if fullSize <= maxLogLineLength
273+ then (fullLine, fullSize)
274+ else (fallbackLine, fallbackSize)
275+ where
276+ fullObject = fullItemObject verbosity item
277+ (fullLine, fullSize) = measureJSON fullObject
278+ -- only the absolutely necessary keys, with message stripped out
279+ -- to help us calculate how much of the message we can keep. these
280+ -- are lazily evaluated and as such won't be computed unless the
281+ -- item is too big
282+ blankObject :: A.Object
283+ blankObject = HM.fromList
284+ [ "message" A..= A.String "" -- we'll start with a blank message
285+ , "@timestamp" A..= K._itemTime item
286+ ]
287+ (_, blankObjectSize) = measureJSON blankObject
288+ messageBytesAllowed = maxLogLineLength - blankObjectSize
289+ (fallbackLine, fallbackSize) = measureJSON fallbackObject
290+ fallbackObject :: A.Object
291+ fallbackObject = HM.fromList
292+ [ "message" A..= A.toJSON (TL.take (bytes messageBytesAllowed) (TB.toLazyText (K.unLogStr (K._itemMessage item))))
293+ , "@timestamp" A..= A.toJSON (K._itemTime item)
294+ ]
295+
296+
297+data BulkBuffer = BulkBuffer
298+ { bulkBuffer_bytesUsed :: !Bytes
299+ , bulkBuffer_payload :: !BB.Builder
300+ , bulkBuffer_itemCount :: !Int
301+ }
302+
303+
304+instance Semigroup BulkBuffer where
305+ (BulkBuffer bytesUsedA bufferA itemCountA) <>
306+ (BulkBuffer bytesUsedB bufferB itemCountB) = BulkBuffer
307+ (bytesUsedA + bytesUsedB)
308+ (bufferA <> bufferB)
309+ (itemCountA + itemCountB)
310+
311+
312+instance Monoid BulkBuffer where
313+ mempty = BulkBuffer 0 mempty 0
314+ mappend = (<>)
315+
316+
317+data LogAction buf =
318+ Buffered
319+ buf
320+ -- ^ New buffer
321+ | FlushNow
322+ buf
323+ -- ^ Buffer to flush
324+ buf
325+ -- ^ New buffer
326+
327+bufferItem
328+ :: K.LogItem a
329+ => Int
330+ -- ^ Maximum items before flushing
331+ -> K.Verbosity
332+ -> K.Item a
333+ -> BulkBuffer
334+ -> LogAction BulkBuffer
335+bufferItem maxItems verb item bulkBuffer =
336+ let (encodedLine, spaceNeeded) = renderLineTruncated verb item
337+ newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded
338+ newItemCount = bulkBuffer_itemCount bulkBuffer + 1
339+ in if newItemCount >= maxItems || newBytesUsed >= maxPayloadBytes
340+ then FlushNow
341+ bulkBuffer
342+ BulkBuffer
343+ { bulkBuffer_bytesUsed = spaceNeeded
344+ , bulkBuffer_payload = encodedLine
345+ , bulkBuffer_itemCount = 1
346+ }
347+ else Buffered $
348+ BulkBuffer
349+ { bulkBuffer_bytesUsed = newBytesUsed
350+ , bulkBuffer_payload = bulkBuffer_payload bulkBuffer <> encodedLine
351+ , bulkBuffer_itemCount = newItemCount
352+ }
353+
354+
355+-- | When time has run out on a flush, splits the buffer. n.b. this
356+-- will always return 'FlushNow' with an empty new buffer.
357+forceFlush :: (Monoid buf) => buf -> LogAction buf
358+forceFlush buf = FlushNow buf mempty