@@ -1,5 +1,4 @@
1{-# LANGUAGE BangPatterns #-}
1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE GeneralizedNewtypeDeriving #-}
2{-# LANGUAGE GeneralizedNewtypeDeriving #-}
4{-# LANGUAGE OverloadedStrings #-}
3{-# LANGUAGE OverloadedStrings #-}
5{-# LANGUAGE RankNTypes #-}
4{-# LANGUAGE RankNTypes #-}
@@ -39,34 +38,31 @@ module Katip.Scribes.LogzIO.HTTPS
40-------------------------------------------------------------------------------
39-------------------------------------------------------------------------------
41import Control.Applicative
40import Control.Applicative
42−import qualified Control.Concurrent.Async as Async
41+import qualified Control.Concurrent.Async as Async
43−import qualified Control.Concurrent.STM as STM
42+import qualified Control.Concurrent.STM as STM
44−import qualified Control.Concurrent.STM.TBMQueue as STM
45−import qualified Control.Error as E
43+import qualified Control.Error as E
46−import qualified Control.Exception.Safe as EX
44+import qualified Control.Exception.Safe as EX
48−import qualified Control.Retry as Retry
45+import qualified Control.Retry as Retry
49−import qualified Data.Aeson as A
46+import qualified Data.Aeson as A
50−import qualified Data.ByteString.Builder as BB
47+import qualified Data.ByteString.Builder as BB
51−import qualified Data.ByteString.Lazy as LBS
48+import qualified Data.ByteString.Lazy as LBS
52−import qualified Data.ByteString.Lazy.Char8 as LBS8
49+import qualified Data.ByteString.Lazy.Char8 as LBS8
53−import qualified Data.HashMap.Strict as HM
50+import qualified Data.HashMap.Strict as HM
54import Data.Int
51import Data.Int
55−import qualified Data.Scientific as Scientific
56−import Data.Semigroup as Semigroup
57−import Data.String (IsString)
52+import Data.String (IsString)
58−import qualified Data.Text as T
53+import qualified Data.Text as T
59−import qualified Data.Text.Encoding as TE
54+import qualified Data.Text.Encoding as TE
60−import qualified Data.Text.Lazy as TL
55+import qualified Data.Text.Lazy as TL
61−import qualified Data.Text.Lazy.Builder as TB
56+import qualified Data.Text.Lazy.Builder as TB
62−import qualified Data.Time as Time
57+import qualified Data.Time as Time
63−import qualified Katip as K
58+import qualified Katip as K
64−import Katip.Core (LocJs (..))
59+import Katip.Core (LocJs (..))
65−import qualified Network.HTTP.Client as HTTP
60+import qualified Network.HTTP.Client as HTTP
66−import qualified Network.HTTP.Client.TLS as HTTPS
61+import qualified Network.HTTP.Client.TLS as HTTPS
67−import qualified Network.HTTP.Types as HTypes
62+import qualified Network.HTTP.Types as HTypes
68−import qualified System.Posix.Types as POSIX
63+import qualified System.Posix.Types as POSIX
69−import qualified URI.ByteString as URIBS
64+import qualified URI.ByteString as URIBS
65+-------------------------------------------------------------------------------
70-------------------------------------------------------------------------------
66-------------------------------------------------------------------------------
@@ -200,7 +196,7 @@ httpPort = URIBS.Port 8070
200-- | A reasonable retry policy: exponential backoff with 25ms base
196-- | A reasonable retry policy: exponential backoff with 25ms base
201-- delay up to 5 retries, for a total cumulative delay of 775ms.
197-- delay up to 5 retries, for a total cumulative delay of 775ms.
202defaultRetryPolicy :: (Monad m) => Retry.RetryPolicyM m
198defaultRetryPolicy :: (Monad m) => Retry.RetryPolicyM m
203−defaultRetryPolicy = Retry.exponentialBackoff 25000 `mappend` Retry.limitRetries 5
199+defaultRetryPolicy = Retry.exponentialBackoff 25000 <> Retry.limitRetries 5
205-------------------------------------------------------------------------------
201-------------------------------------------------------------------------------
206mkLogzIOScribe
202mkLogzIOScribe
@@ -209,115 +205,132 @@ mkLogzIOScribe
209 -> K.Verbosity
205 -> K.Verbosity
210 -> IO K.Scribe
206 -> IO K.Scribe
211mkLogzIOScribe config permitItem verbosity = do
207mkLogzIOScribe config permitItem verbosity = do
212− -- This size is actually somewhat arbitrary. We're just making sure
208+ -- Accumulating buffer for each item
213− -- it isn't infinite so we don't consume the whole backlog right
209+ bufferState <- STM.newTVarIO (mempty :: BulkBuffer)
214− -- away and take up lots of memory.. Katip above us does bounded
210+ -- Single-slot TMVar when something is ready to write. Katip buffers
215− -- buffering and load-shedding when there's too much of a
211+ -- for us upstream so we can simplify shutdown and flush guarantees
216− -- backlog. Our writes are blocking to apply backpressure up to
212+ -- by synchronously writing when we need to.
217− -- katip. This value just holds onto values temporarily while we
213+ readyMVar <- STM.newEmptyTMVarIO :: IO (STM.TMVar BulkBuffer)
218− -- concatenate them into a buffer.
214+
219− ingestionQueue <- STM.newTBMQueueIO (itemBufferSize * 10)
215+ -- We create a TVar that is set to ShutDown when its time to stop
220−
216+ killSwitch <- STM.newTVarIO Continue
218+ -- Create a mutable ref to whether the timer has expired. This lets
219+ -- us replace the timer when its time to start timing again. We
220+ -- reset the timer __after__ a write goes out, not as soon as it expires
221− let newTimer = STM.registerDelay itemBufferTimeoutMicros
221+ timerRef <- STM.newTVarIO =<< STM.registerDelay itemBufferTimeoutMicros
222− timerRef <- STM.newTVarIO =<< newTimer
224 -- Set up a connection manager for requests
223 -- Set up a connection manager for requests
225 mgr <- case scheme of
224 mgr <- case scheme of
226 HTTPS -> HTTPS.newTlsManager
225 HTTPS -> HTTPS.newTlsManager
227 HTTP -> HTTP.newManager HTTP.defaultManagerSettings
226 HTTP -> HTTP.newManager HTTP.defaultManagerSettings
229− -- An STM transaction that will return true when writes are stopped
228+ -- In STM: resolve an action with the buffers
230− -- and backlog is emptied.
229+ let processLogAction :: LogAction BulkBuffer -> STM.STM ()
231− let workExhausted :: STM.STM Bool
230+ processLogAction logAction = case logAction of
232− workExhausted = (&&)
231+ -- No write needed right now
233− <$> STM.isClosedTBMQueue ingestionQueue
232+ Buffered !newBuffer -> STM.writeTVar bufferState newBuffer
234− <*> STM.isEmptyTBMQueue ingestionQueue
233+ -- Time to flush and put excess in the next buffer
235− -- Block until there's no more work
234+ FlushNow !toFlush !newBuffer -> do
236− let waitWorkExhausted :: STM.STM ()
235+ STM.writeTVar bufferState newBuffer
237− waitWorkExhausted = STM.check =<< workExhausted
236+ STM.putTMVar readyMVar toFlush
238− let pop :: STM.STM (Tick AnyLogItem)
237+
239− pop = maybe WorkExhausted NewItem <$> STM.readTBMQueue ingestionQueue
238+ -- Recompute the new buffer state from a new log item and take action
240−
239+ let push :: (K.LogItem a) => K.Item a -> STM.STM ()
241− let timeExpired :: STM.STM (Tick a)
240+ push !logItem = do
242− timeExpired = do
241+ !curBuffer <- STM.readTVar bufferState
243− isExpired <- STM.readTVar =<< STM.readTVar timerRef
242+ processLogAction (bufferItem itemBufferSize verbosity logItem curBuffer)
244− STM.check isExpired
243+
245− pure TimeExpired
244+
246−
245+ -- A transaction that retries until its time to shutdown
247− -- Circular transaction that checks for completion of work, time
246+ let waitShutdown :: STM.STM ()
248− -- expiration, or new events.
247+ waitShutdown = do
249− let nextTick :: STM.STM (Tick AnyLogItem)
248+ killStatus <- STM.readTVar killSwitch
250− nextTick =
249+ STM.check (killStatus == ShutDown)
252− pop <|>
251+ -- Retry transaction until timer has expired. When it does, take action on the buffer.
253− (WorkExhausted <$ waitWorkExhausted)
255− -- Blocking push. This applies backpressure upstream to katip, which does its own buffering
256− let push :: AnyLogItem -> STM.STM ()
252+ let waitTimer :: STM.STM ()
257− push = void . STM.writeTBMQueue ingestionQueue
253+ waitTimer = do
258−
254+ -- get the timer we'll be using
259− let sealQueue = STM.atomically (STM.closeTBMQueue ingestionQueue)
255+ timer <- STM.readTVar timerRef
260−
256+ -- do not proceed until its ready
261− -- Replace the timer with a new one
257+ timerExpired <- STM.readTVar timer
262− let resetTimer = STM.atomically . STM.writeTVar timerRef =<< newTimer
258+ STM.check timerExpired
263−
259+ -- do not wake again until this message is out the door
264− -- Send the buffer and then reset the timer
260+ STM.writeTVar timer False
265− let flush curBuffer = do
261+ -- we waited long enough, grab the current buffer
266− res <- flushBuffer config mgr curBuffer
262+ !curBuffer <- STM.readTVar bufferState
263+ -- and tell it to flush right now
264+ processLogAction (forceFlush curBuffer)
266+ -- run the timer in a loop until its time to shut down
267+ let timedFlushLoop :: IO ()
269+ res <- STM.atomically $
267 case res of
273 case res of
274+ Left () -> pure () -- exit loop
275+ Right () -> timedFlushLoop -- wait for another timer or shutdown
277+ timedFlusher <- Async.async timedFlushLoop
279+ -- flush the buffer until its time to shutdown and we're out of items to log
280+ let flushLoop :: IO ()
282+ -- 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
283+ readyBufferOrDie <- STM.atomically $
285+ (STM.takeTMVar readyMVar)
287+ case readyBufferOrDie of
288+ Right () -> pure () -- we caught shutdown before any new data arrived
289+ -- There's some logs to write
290+ Left readyBuffer -> do
292+ res <- flushBuffer config mgr readyBuffer
268− Left e -> onErrorSafe e
294+ Left ex -> onErrorSafe ex
269− Right () -> pure ()
295+ Right () -> pure ()
270− resetTimer
296+ -- reset the timeout before the next flush on each write
271−
297+ !newTimer <- STM.registerDelay itemBufferTimeoutMicros
272− let flushLoop :: BulkBuffer -> IO ()
273− flushLoop curBuffer = do
274− tick <- STM.atomically nextTick
298+ STM.atomically $ do
275− case tick of
299+ STM.writeTVar timerRef newTimer
276− WorkExhausted -> do
300+ -- start over
277− flush curBuffer -- flush what you've got
301+ flushLoop
278− pure () -- stop looping
302+ flusher <- Async.async flushLoop
279− TimeExpired -> do
303+
280− flush curBuffer
304+ -- Blocking call which sends a poison pill to the workers and waits
281− flushLoop mempty
305+ -- for them to exit, swallowing exceptions. This follows the
282− NewItem (AnyLogItem item) -> do
306+ -- semantics that when katip calls the finalizer, no more log
283− case bufferItem (logzIOScribeConfiguration_bufferItems config) verbosity item curBuffer of
307+ -- entries will be sent.
284− Buffered newBuffer -> flushLoop newBuffer
308+ let shutdown :: IO ()
285− FlushNow flushThis newBuffer -> do
309+ shutdown = do
286− flush flushThis
310+ -- tell workers this is the last loop
287− flushLoop newBuffer
311+ STM.atomically (STM.writeTVar killSwitch ShutDown)
288−
312+ -- wait for workers to finish
289− flushThread <- Async.async (flushLoop mempty)
313+ _ <- Async.waitCatch timedFlusher
293− _ <- Async.waitCatch flushThread
314+ _ <- Async.waitCatch flusher
296 pure $ K.Scribe
319 pure $ K.Scribe
297− { K.liPush = STM.atomically . push . AnyLogItem
320+ { K.liPush = STM.atomically . push
298− , K.scribeFinalizer = close
321+ , K.scribeFinalizer = shutdown
299 , K.scribePermitItem = permitItem
322 , K.scribePermitItem = permitItem
302 itemBufferSize = logzIOScribeConfiguration_bufferItems config
325 itemBufferSize = logzIOScribeConfiguration_bufferItems config
303 itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
326 itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
304 scheme = logzIOScribeConfiguration_scheme config
327 scheme = logzIOScribeConfiguration_scheme config
328+ -- should this be async?
305 onErrorSafe ex = do
329 onErrorSafe ex = do
306 _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
330 _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
310−data AnyLogItem where
311− AnyLogItem :: K.LogItem a => K.Item a -> AnyLogItem
314−-------------------------------------------------------------------------------
321-------------------------------------------------------------------------------
334-------------------------------------------------------------------------------
322-- Match the native resolution of NominalDiffTime, which is excessive
335-- Match the native resolution of NominalDiffTime, which is excessive
323-- for our needs
336-- for our needs
@@ -335,6 +348,23 @@ ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
335 picosInMicro = 10 ^ (3 :: Int)
348 picosInMicro = 10 ^ (3 :: Int)
351+-------------------------------------------------------------------------------
352+-- | An internal mode flag used to shut down worker
353+-- threads. Internally, we assume htere is no going back from
361+-------------------------------------------------------------------------------
362+-- | When applied to STM, it will try the left side first and if that
363+-- retries, the right, and if that retries, start from the beginning.
364+raceAlt :: (Alternative m) => m a -> m b -> m (Either a b)
365+raceAlt a b = (Left <$> a) <|> (Right <$> b)
338-------------------------------------------------------------------------------
368-------------------------------------------------------------------------------
339data LoggingError =
369data LoggingError =
340 URIError HTTP.HttpException
370 URIError HTTP.HttpException
@@ -411,7 +441,7 @@ flushBuffer config mgr bulkBuffer
411-------------------------------------------------------------------------------
441-------------------------------------------------------------------------------
412newtype Bytes = Bytes
442newtype Bytes = Bytes
413 { bytes :: Int64
443 { bytes :: Int64
414− } deriving (Show, Eq, Num, Ord, Bounded)
444+ } deriving (Show, Eq, Num, Ord)
417-- | How big of a body can we send? The limit is defined
447-- | How big of a body can we send? The limit is defined
@@ -442,7 +472,7 @@ fullItemObject verbosity item = HM.fromList
442 , "thread" A..= K.getThreadIdText (K._itemThread item)
472 , "thread" A..= K.getThreadIdText (K._itemThread item)
443 , "host" A..= K._itemHost item
473 , "host" A..= K._itemHost item
444 , "pid" A..= pidInt
474 , "pid" A..= pidInt
445− , "data" A..= annotateKeys (K.payloadObject verbosity (K._itemPayload item))
475+ , "data" A..= K.payloadObject verbosity (K._itemPayload item)
446 -- Slight deviation, logz.io uses "message" instead of "msg"
476 -- Slight deviation, logz.io uses "message" instead of "msg"
447 , "message" A..= TB.toLazyText (K.unLogStr (K._itemMessage item))
477 , "message" A..= TB.toLazyText (K.unLogStr (K._itemMessage item))
448 -- Another slight deviation, logz.io uses "@timestamp" instead of
478 -- Another slight deviation, logz.io uses "@timestamp" instead of
@@ -452,7 +482,7 @@ fullItemObject verbosity item = HM.fromList
452 -- past. They seem to support 3 decimal places of
482 -- past. They seem to support 3 decimal places of
453 -- precision. Formatting like this requires time 1.8.0.2, which is
483 -- precision. Formatting like this requires time 1.8.0.2, which is
454 -- reflected in the cabal file.
484 -- reflected in the cabal file.
455− , "@timestamp" A..= A.String (T.pack (Time.formatTime Time.defaultTimeLocale "%Y-%m-%dT%H:%M:%S%03QZ" (K._itemTime item)))
485+ , "@timestamp" A..= A.String (T.pack (Time.formatTime Time.defaultTimeLocale "%Y-%m-%dT%H:%M:%03QZ" (K._itemTime item)))
456 , "ns" A..= K._itemNamespace item
486 , "ns" A..= K._itemNamespace item
457 , "loc" A..= (LocJs <$> K._itemLoc item)
487 , "loc" A..= (LocJs <$> K._itemLoc item)
@@ -513,7 +543,7 @@ data BulkBuffer = BulkBuffer
516−instance Semigroup.Semigroup BulkBuffer where
546+instance Semigroup BulkBuffer where
517 (BulkBuffer bytesUsedA bufferA itemCountA) <>
547 (BulkBuffer bytesUsedA bufferA itemCountA) <>
518 (BulkBuffer bytesUsedB bufferB itemCountB) = BulkBuffer
548 (BulkBuffer bytesUsedB bufferB itemCountB) = BulkBuffer
519 (bytesUsedA + bytesUsedB)
549 (bytesUsedA + bytesUsedB)
@@ -567,7 +597,7 @@ bufferItem' customMaxPayload customMaxItem maxItems verb item bulkBuffer =
567 let (encodedLine, spaceNeeded) = renderLineTruncated' customMaxItem verb item
597 let (encodedLine, spaceNeeded) = renderLineTruncated' customMaxItem verb item
568 newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded
598 newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded
569 newItemCount = bulkBuffer_itemCount bulkBuffer + 1
599 newItemCount = bulkBuffer_itemCount bulkBuffer + 1
570− in if newItemCount >= maxItems || newBytesUsed >= customMaxPayload
600+ in if newItemCount > maxItems || newBytesUsed > customMaxPayload
571 then FlushNow
601 then FlushNow
572 bulkBuffer
602 bulkBuffer
573 BulkBuffer
603 BulkBuffer
@@ -587,53 +617,3 @@ bufferItem' customMaxPayload customMaxItem maxItems verb item bulkBuffer =
587-- will always return 'FlushNow' with an empty new buffer.
617-- will always return 'FlushNow' with an empty new buffer.
588forceFlush :: (Monoid buf) => buf -> LogAction buf
618forceFlush :: (Monoid buf) => buf -> LogAction buf
589forceFlush buf = FlushNow buf mempty
619forceFlush buf = FlushNow buf mempty
592−-------------------------------------------------------------------------------
593−-- Annotation borrowed from katip-elasticsearch. There are fixed
594−-- fields in katip logs which should stay the same type forever and
595−-- thus won't need annotation. However, any code in userland may
596−-- choose to add akey to their log's metadata with a type that's
597−-- incompatible with the mapping. If the first value that's picked up
598−-- restrictive (e.g. a long), subsequent values will be rejected. By
599−-- differentiating types by their name, this is guaranteed not to
602−annotateValue :: A.Value -> A.Value
603−annotateValue (A.Object o) = A.Object (annotateKeys o)
604−annotateValue (A.Array a) = A.Array (annotateValue <$> a)
608−annotateKeys :: A.Object -> A.Object
609−annotateKeys = HM.fromList . map go . HM.toList
611− go (k, A.Object o) = (k, A.Object (annotateKeys o))
612− go (k, A.Array a) = (k, A.Array (annotateValue <$> a))
613− go (k, s@(A.String _)) = (k <> stringAnn, s)
614− go (k, n@(A.Number sci)) = if Scientific.isFloating sci
615− then (k <> doubleAnn, n)
616− else (k <> longAnn, n)
617− go (k, b@(A.Bool _)) = (k <> booleanAnn, b)
618− go (k, A.Null) = (k <> nullAnn, A.Null)
621−-------------------------------------------------------------------------------
622−-- Annotation Constants
623−-------------------------------------------------------------------------------