@@ -1,4 +1,5 @@
1{-# LANGUAGE BangPatterns #-}
1{-# LANGUAGE BangPatterns #-}
2{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3{-# LANGUAGE OverloadedStrings #-}
4{-# LANGUAGE OverloadedStrings #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE RankNTypes #-}
@@ -38,33 +39,34 @@ module Katip.Scribes.LogzIO.HTTPS
39-------------------------------------------------------------------------------
40-------------------------------------------------------------------------------
40import Control.Applicative
41import Control.Applicative
41−import qualified Control.Concurrent.Async as Async
42+import qualified Control.Concurrent.Async as Async
42−import qualified Control.Concurrent.STM as STM
43+import qualified Control.Concurrent.STM as STM
44+import qualified Control.Concurrent.STM.TBMQueue as STM
43−import qualified Control.Error as E
45+import qualified Control.Error as E
44−import qualified Control.Exception.Safe as EX
46+import qualified Control.Exception.Safe as EX
45−import qualified Control.Retry as Retry
48+import qualified Control.Retry as Retry
46−import qualified Data.Aeson as A
49+import qualified Data.Aeson as A
47−import qualified Data.ByteString.Builder as BB
50+import qualified Data.ByteString.Builder as BB
48−import qualified Data.ByteString.Lazy as LBS
51+import qualified Data.ByteString.Lazy as LBS
49−import qualified Data.ByteString.Lazy.Char8 as LBS8
52+import qualified Data.ByteString.Lazy.Char8 as LBS8
50−import qualified Data.HashMap.Strict as HM
53+import qualified Data.HashMap.Strict as HM
51import Data.Int
54import Data.Int
52−import qualified Data.Scientific as Scientific
55+import qualified Data.Scientific as Scientific
53−import Data.Semigroup as Semigroup
56+import Data.Semigroup as Semigroup
54−import Data.String (IsString)
57+import Data.String (IsString)
55−import qualified Data.Text as T
58+import qualified Data.Text as T
56−import qualified Data.Text.Encoding as TE
59+import qualified Data.Text.Encoding as TE
57−import qualified Data.Text.Lazy as TL
60+import qualified Data.Text.Lazy as TL
58−import qualified Data.Text.Lazy.Builder as TB
61+import qualified Data.Text.Lazy.Builder as TB
59−import qualified Data.Time as Time
62+import qualified Data.Time as Time
60−import qualified Katip as K
63+import qualified Katip as K
61−import Katip.Core (LocJs (..))
64+import Katip.Core (LocJs (..))
62−import qualified Network.HTTP.Client as HTTP
65+import qualified Network.HTTP.Client as HTTP
63−import qualified Network.HTTP.Client.TLS as HTTPS
66+import qualified Network.HTTP.Client.TLS as HTTPS
64−import qualified Network.HTTP.Types as HTypes
67+import qualified Network.HTTP.Types as HTypes
65−import qualified System.Posix.Types as POSIX
68+import qualified System.Posix.Types as POSIX
66−import qualified URI.ByteString as URIBS
69+import qualified URI.ByteString as URIBS
67−-------------------------------------------------------------------------------
68-------------------------------------------------------------------------------
70-------------------------------------------------------------------------------
@@ -207,132 +209,115 @@ mkLogzIOScribe
207 -> K.Verbosity
209 -> K.Verbosity
208 -> IO K.Scribe
210 -> IO K.Scribe
209mkLogzIOScribe config permitItem verbosity = do
211mkLogzIOScribe config permitItem verbosity = do
210− -- Accumulating buffer for each item
212+ -- This size is actually somewhat arbitrary. We're just making sure
211− bufferState <- STM.newTVarIO (mempty :: BulkBuffer)
213+ -- it isn't infinite so we don't consume the whole backlog right
212− -- Single-slot TMVar when something is ready to write. Katip buffers
214+ -- away and take up lots of memory.. Katip above us does bounded
213− -- for us upstream so we can simplify shutdown and flush guarantees
215+ -- buffering and load-shedding when there's too much of a
214− -- by synchronously writing when we need to.
216+ -- backlog. Our writes are blocking to apply backpressure up to
215− readyMVar <- STM.newEmptyTMVarIO :: IO (STM.TMVar BulkBuffer)
217+ -- katip. This value just holds onto values temporarily while we
216−
218+ -- concatenate them into a buffer.
217− -- We create a TVar that is set to ShutDown when its time to stop
219+ ingestionQueue <- STM.newTBMQueueIO (itemBufferSize * 10)
218− killSwitch <- STM.newTVarIO Continue
220+
219−
221+ let newTimer = STM.registerDelay itemBufferTimeoutMicros
220− -- Create a mutable ref to whether the timer has expired. This lets
221− -- us replace the timer when its time to start timing again. We
222− -- reset the timer __after__ a write goes out, not as soon as it expires
223− timerRef <- STM.newTVarIO =<< STM.registerDelay itemBufferTimeoutMicros
222+ timerRef <- STM.newTVarIO =<< newTimer
225 -- Set up a connection manager for requests
224 -- Set up a connection manager for requests
226 mgr <- case scheme of
225 mgr <- case scheme of
227 HTTPS -> HTTPS.newTlsManager
226 HTTPS -> HTTPS.newTlsManager
228 HTTP -> HTTP.newManager HTTP.defaultManagerSettings
227 HTTP -> HTTP.newManager HTTP.defaultManagerSettings
230− -- In STM: resolve an action with the buffers
229+ -- An STM transaction that will return true when writes are stopped
231− let processLogAction :: LogAction BulkBuffer -> STM.STM ()
230+ -- and backlog is emptied.
232− processLogAction logAction = case logAction of
231+ let workExhausted :: STM.STM Bool
233− -- No write needed right now
232+ workExhausted = (&&)
234− Buffered !newBuffer -> STM.writeTVar bufferState newBuffer
233+ <$> STM.isClosedTBMQueue ingestionQueue
235− -- Time to flush and put excess in the next buffer
234+ <*> STM.isEmptyTBMQueue ingestionQueue
236− FlushNow !toFlush !newBuffer -> do
235+ -- Block until there's no more work
237− STM.writeTVar bufferState newBuffer
236+ let waitWorkExhausted :: STM.STM ()
238− STM.putTMVar readyMVar toFlush
237+ waitWorkExhausted = STM.check =<< workExhausted
239−
238+ let pop :: STM.STM (Tick AnyLogItem)
240− -- Recompute the new buffer state from a new log item and take action
239+ pop = maybe WorkExhausted NewItem <$> STM.readTBMQueue ingestionQueue
241− let push :: (K.LogItem a) => K.Item a -> STM.STM ()
240+
242− push !logItem = do
241+ let timeExpired :: STM.STM (Tick a)
243− !curBuffer <- STM.readTVar bufferState
242+ timeExpired = do
244− processLogAction (bufferItem itemBufferSize verbosity logItem curBuffer)
243+ isExpired <- STM.readTVar =<< STM.readTVar timerRef
245−
244+ STM.check isExpired
246−
245+ pure TimeExpired
247− -- A transaction that retries until its time to shutdown
246+
248− let waitShutdown :: STM.STM ()
247+ -- Circular transaction that checks for completion of work, time
249− waitShutdown = do
248+ -- expiration, or new events.
250− killStatus <- STM.readTVar killSwitch
249+ let nextTick :: STM.STM (Tick AnyLogItem)
251− STM.check (killStatus == ShutDown)
250+ nextTick =
253− -- Retry transaction until timer has expired. When it does, take action on the buffer.
252+ pop <|>
253+ (WorkExhausted <$ waitWorkExhausted)
255+ -- Blocking push. This applies backpressure upstream to katip, which does its own buffering
254− let waitTimer :: STM.STM ()
256+ let push :: AnyLogItem -> STM.STM ()
255− waitTimer = do
257+ push = void . STM.writeTBMQueue ingestionQueue
256− -- get the timer we'll be using
258+
257− timer <- STM.readTVar timerRef
259+ let sealQueue = STM.atomically (STM.closeTBMQueue ingestionQueue)
258− -- do not proceed until its ready
260+
259− timerExpired <- STM.readTVar timer
261+ -- Replace the timer with a new one
260− STM.check timerExpired
262+ let resetTimer = STM.atomically . STM.writeTVar timerRef =<< newTimer
261− -- do not wake again until this message is out the door
263+
262− STM.writeTVar timer False
264+ -- Send the buffer and then reset the timer
263− -- we waited long enough, grab the current buffer
265+ let flush curBuffer = do
264− !curBuffer <- STM.readTVar bufferState
266+ res <- flushBuffer config mgr curBuffer
265− -- and tell it to flush right now
266− processLogAction (forceFlush curBuffer)
268− -- run the timer in a loop until its time to shut down
269− let timedFlushLoop :: IO ()
271− res <- STM.atomically $
275 case res of
267 case res of
276− Left () -> pure () -- exit loop
268+ Left e -> onErrorSafe e
277− Right () -> timedFlushLoop -- wait for another timer or shutdown
269+ Right () -> pure ()
279− timedFlusher <- Async.async timedFlushLoop
271+
281− -- flush the buffer until its time to shutdown and we're out of items to log
282− let flushLoop :: IO ()
272+ let flushLoop :: BulkBuffer -> IO ()
283− flushLoop = do
273+ flushLoop curBuffer = do
284− -- 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
285− readyBufferOrDie <- STM.atomically $
274+ tick <- STM.atomically nextTick
286− raceAlt
275+ case tick of
287− (STM.takeTMVar readyMVar)
276+ WorkExhausted -> do
288− waitShutdown
277+ flush curBuffer -- flush what you've got
289− case readyBufferOrDie of
278+ pure () -- stop looping
290− Right () -> pure () -- we caught shutdown before any new data arrived
279+ TimeExpired -> do
291− -- There's some logs to write
280+ flush curBuffer
292− Left readyBuffer -> do
281+ flushLoop mempty
293− -- flush in IO
282+ NewItem (AnyLogItem item) -> do
294− res <- flushBuffer config mgr readyBuffer
283+ case bufferItem (logzIOScribeConfiguration_bufferItems config) verbosity item curBuffer of
295− case res of
284+ Buffered newBuffer -> flushLoop newBuffer
296− Left ex -> onErrorSafe ex
285+ FlushNow flushThis newBuffer -> do
297− Right () -> pure ()
286+ flush flushThis
298− -- reset the timeout before the next flush on each write
287+ flushLoop newBuffer
299− !newTimer <- STM.registerDelay itemBufferTimeoutMicros
288+
301− STM.writeTVar timerRef newTimer
304− flusher <- Async.async flushLoop
289+ flushThread <- Async.async (flushLoop mempty)
306− -- Blocking call which sends a poison pill to the workers and waits
291+ let close = do
307− -- for them to exit, swallowing exceptions. This follows the
292+ sealQueue
308− -- semantics that when katip calls the finalizer, no more log
309− -- entries will be sent.
310− let shutdown :: IO ()
312− -- tell workers this is the last loop
313− STM.atomically (STM.writeTVar killSwitch ShutDown)
314− -- wait for workers to finish
315− _ <- Async.waitCatch timedFlusher
293+ _ <- Async.waitCatch flushThread
316− _ <- Async.waitCatch flusher
321 pure $ K.Scribe
296 pure $ K.Scribe
322− { K.liPush = STM.atomically . push
297+ { K.liPush = STM.atomically . push . AnyLogItem
323− , K.scribeFinalizer = shutdown
298+ , K.scribeFinalizer = close
324 , K.scribePermitItem = permitItem
299 , K.scribePermitItem = permitItem
327 itemBufferSize = logzIOScribeConfiguration_bufferItems config
302 itemBufferSize = logzIOScribeConfiguration_bufferItems config
328 itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
303 itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
329 scheme = logzIOScribeConfiguration_scheme config
304 scheme = logzIOScribeConfiguration_scheme config
330− -- should this be async?
331 onErrorSafe ex = do
305 onErrorSafe ex = do
332 _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
306 _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
310+data AnyLogItem where
311+ AnyLogItem :: K.LogItem a => K.Item a -> AnyLogItem
314+-------------------------------------------------------------------------------
336-------------------------------------------------------------------------------
321-------------------------------------------------------------------------------
337-- Match the native resolution of NominalDiffTime, which is excessive
322-- Match the native resolution of NominalDiffTime, which is excessive
338-- for our needs
323-- for our needs
@@ -350,23 +335,6 @@ ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
350 picosInMicro = 10 ^ (3 :: Int)
335 picosInMicro = 10 ^ (3 :: Int)
353−-------------------------------------------------------------------------------
354−-- | An internal mode flag used to shut down worker
355−-- threads. Internally, we assume htere is no going back from
363−-------------------------------------------------------------------------------
364−-- | When applied to STM, it will try the left side first and if that
365−-- retries, the right, and if that retries, start from the beginning.
366−raceAlt :: (Alternative m) => m a -> m b -> m (Either a b)
367−raceAlt a b = (Left <$> a) <|> (Right <$> b)
370-------------------------------------------------------------------------------
338-------------------------------------------------------------------------------
371data LoggingError =
339data LoggingError =
372 URIError HTTP.HttpException
340 URIError HTTP.HttpException