@@ -1,3 +1,8 @@
1+{-# LANGUAGE BangPatterns #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3+{-# LANGUAGE OverloadedStrings #-}
4+{-# LANGUAGE RankNTypes #-}
5+{-# LANGUAGE TypeApplications #-}
16-- | This is a log scribe that writes logs to logz.io's bulk
27-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS HTTPS
38-- API>.
@@ -7,24 +12,56 @@ module Katip.Scribes.LogzIO.HTTPS
712 -- * Types
813 , BulkAPIError(..)
914 , LogzIOScribeConfiguration(..)
15+ , Scheme(..)
16+ , APIToken(..)
17+ , LoggingError(..)
18+ -- ** Presets for configuration
19+ , usRegionHost
20+ , euRegionHost
21+ , httpsPort
22+ , httpPort
23+ , defaultRetryPolicy
24+ , defaultLogzIOScribeConfiguration
25+ -- * Internal API exported for testing
26+ , renderLineTruncated
27+ , renderLineTruncated'
28+ , maxPayloadBytes
29+ , maxLogLineLength
30+ , BulkBuffer (..)
31+ , LogAction (..)
32+ , bufferItem
33+ , bufferItem'
34+ , forceFlush
35+ , Bytes (..)
1036 ) where
1137
1238
1339-------------------------------------------------------------------------------
1440import 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
41+import qualified Control.Concurrent.Async as Async
42+import qualified Control.Concurrent.STM as STM
43+import qualified Control.Error as E
44+import qualified Control.Exception.Safe as EX
45+import qualified Control.Retry as Retry
46+import qualified Data.Aeson as A
47+import qualified Data.ByteString.Builder as BB
48+import qualified Data.ByteString.Lazy as LBS
49+import qualified Data.ByteString.Lazy.Char8 as LBS8
50+import qualified Data.HashMap.Strict as HM
2151import 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
52+import Data.String (IsString)
53+import qualified Data.Text as T
54+import qualified Data.Text.Encoding as TE
55+import qualified Data.Text.Lazy as TL
56+import qualified Data.Text.Lazy.Builder as TB
57+import qualified Data.Time as Time
58+import qualified Katip as K
59+import Katip.Core (LocJs (..))
60+import qualified Network.HTTP.Client as HTTP
61+import qualified Network.HTTP.Client.TLS as HTTPS
62+import qualified Network.HTTP.Types as HTypes
63+import qualified System.Posix.Types as POSIX
64+import qualified URI.ByteString as URIBS
2865-------------------------------------------------------------------------------
2966-------------------------------------------------------------------------------
3067
@@ -47,7 +84,21 @@ data BulkAPIError = BulkAPIError
4784 -- concept of a truly empty log line in katip, so this most likely
4885 -- indicates a __library bug__ and should be reported as an issue on
4986 -- GitHub.
50− }
87+ } deriving (Show, Eq)
88+
89+
90+instance A.FromJSON BulkAPIError where
91+ parseJSON = A.withObject "BulkAPIError" $ \o -> do
92+ malformedLines <- o A..: "malformedLines"
93+ successfulLines <- o A..: "successfulLines"
94+ oversizedLines <- o A..: "oversizedLines"
95+ emptyLogLines <- o A..: "emptyLogLines"
96+ pure $ BulkAPIError
97+ { bulkAPIError_malformedLines = malformedLines
98+ , bulkAPIError_successfulLines = successfulLines
99+ , bulkAPIError_oversizedLines = oversizedLines
100+ , bulkAPIError_emptyLogLines = emptyLogLines
101+ }
51102
52103
53104-------------------------------------------------------------------------------
@@ -59,10 +110,94 @@ data LogzIOScribeConfiguration = LogzIOScribeConfiguration
59110 , logzIOScribeConfiguration_bufferTimeout :: Time.NominalDiffTime
60111 -- ^ Will flush the buffer if it has been this long since the last
61112 -- flush __or__ 'logzIOScribeConfiguration_bufferItems' items are
62− -- accumulated, whichever is first.
63− --TODO: more
113+ -- accumulated, whichever is first. NominalDiffTime has a Num
114+ -- instance, so you can use a literal to specify seconds, e.g. 30 =
115+ -- 30s.
116+ , logzIOScribeConfiguration_scheme :: Scheme
117+ , logzIOScribeConfiguration_host :: URIBS.Host
118+ , logzIOScribeConfiguration_port :: URIBS.Port
119+ , logzIOScribeConfiguration_token :: APIToken
120+ , logzIOScribeConfiguration_retry :: Retry.RetryPolicyM IO
121+ -- ^ How should exceptions during writes be retried?
122+ , logzIOScribeConfiguration_onError :: LoggingError -> IO ()
123+ }
124+
125+
126+-- | A default configuration:
127+--
128+-- * 100 item buffering
129+--
130+-- * 30 second buffer timeout
131+--
132+-- * US, HTTPS logging endpoint (listener.logz.io:8071)
133+--
134+-- * 'defaultRetryPolicy' of 25ms exponential backoff, 5 retries
135+--
136+-- * Ignore logging errors
137+defaultLogzIOScribeConfiguration :: APIToken -> LogzIOScribeConfiguration
138+defaultLogzIOScribeConfiguration token = LogzIOScribeConfiguration
139+ { logzIOScribeConfiguration_bufferItems = 100
140+ , logzIOScribeConfiguration_bufferTimeout = 30
141+ , logzIOScribeConfiguration_scheme = HTTPS
142+ , logzIOScribeConfiguration_host = usRegionHost
143+ , logzIOScribeConfiguration_port = httpsPort
144+ , logzIOScribeConfiguration_token = token
145+ , logzIOScribeConfiguration_retry = defaultRetryPolicy
146+ , logzIOScribeConfiguration_onError = const (pure ())
64147 }
65148
149+
150+-------------------------------------------------------------------------------
151+-- | You can retrieve your account or sub-account's API token on the
152+-- <https://app.logz.io/#/dashboard/settings/manage-accounts manage
153+-- accounts page>. Note that APIToken has an IsString instance,
154+-- meaning that you can use a string literal with OverloadedStrings
155+-- enabled.
156+newtype APIToken = APIToken
157+ { apiToken :: T.Text
158+ } deriving (Show, Eq, IsString)
159+
160+
161+-- | This particular bulk API only supports HTTP or HTTPS. HTTPS is
162+-- strongly recommended for security reasons.
163+data Scheme
164+ = HTTPS
165+ -- ^ HTTPs should always be used except for local testing.
166+ | HTTP
167+ deriving (Show, Eq)
168+
169+
170+-- | See
171+-- <https://support.logz.io/hc/en-us/articles/210206365-What-IP-addresses-should-I-open-in-my-firewall-to-ship-logs-to-Logz-io-
172+-- this> for a list of listeners. This is the US region host,
173+-- listener.logz.io
174+usRegionHost :: URIBS.Host
175+usRegionHost = URIBS.Host "listener.logz.io"
176+
177+
178+-- | See
179+-- <https://support.logz.io/hc/en-us/articles/210206365-What-IP-addresses-should-I-open-in-my-firewall-to-ship-logs-to-Logz-io-
180+-- this> for a list of listeners. This is the EU region host,
181+-- listener.logz.io
182+euRegionHost :: URIBS.Host
183+euRegionHost = URIBS.Host "listener-eu.logz.io"
184+
185+
186+-- | Logz.io uses port 8071 for HTTPS
187+httpsPort :: URIBS.Port
188+httpsPort = URIBS.Port 8071
189+
190+
191+-- | Logz.io uses port 8070 for HTTP
192+httpPort :: URIBS.Port
193+httpPort = URIBS.Port 8070
194+
195+
196+-- | A reasonable retry policy: exponential backoff with 25ms base
197+-- delay up to 5 retries, for a total cumulative delay of 775ms.
198+defaultRetryPolicy :: (Monad m) => Retry.RetryPolicyM m
199+defaultRetryPolicy = Retry.exponentialBackoff 25000 <> Retry.limitRetries 5
200+
66201-------------------------------------------------------------------------------
67202mkLogzIOScribe
68203 :: LogzIOScribeConfiguration
@@ -73,7 +208,8 @@ mkLogzIOScribe config permitItem verbosity = do
73208 -- Accumulating buffer for each item
74209 bufferState <- STM.newTVarIO (mempty :: BulkBuffer)
75210 -- Single-slot TMVar when something is ready to write. Katip buffers
76− -- for us so we can simplify shutdown and flush guarantees this way
211+ -- for us upstream so we can simplify shutdown and flush guarantees
212+ -- by synchronously writing when we need to.
77213 readyMVar <- STM.newEmptyTMVarIO :: IO (STM.TMVar BulkBuffer)
78214
79215 -- We create a TVar that is set to ShutDown when its time to stop
@@ -82,38 +218,39 @@ mkLogzIOScribe config permitItem verbosity = do
82218 -- Create a mutable ref to whether the timer has expired. This lets
83219 -- us replace the timer when its time to start timing again. We
84220 -- 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
221+ timerRef <- STM.newTVarIO =<< STM.registerDelay itemBufferTimeoutMicros
222+
223+ -- Set up a connection manager for requests
224+ mgr <- case scheme of
225+ HTTPS -> HTTPS.newTlsManager
226+ HTTP -> HTTP.newManager HTTP.defaultManagerSettings
227+
228+ -- In STM: resolve an action with the buffers
229+ let processLogAction :: LogAction BulkBuffer -> STM.STM ()
230+ processLogAction logAction = case logAction of
231+ -- No write needed right now
232+ Buffered !newBuffer -> STM.writeTVar bufferState newBuffer
233+ -- Time to flush and put excess in the next buffer
234+ FlushNow !toFlush !newBuffer -> do
99235 STM.writeTVar bufferState newBuffer
100236 STM.putTMVar readyMVar toFlush
101237
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
238+ -- Recompute the new buffer state from a new log item and take action
239+ let push :: (K.LogItem a) => K.Item a -> STM.STM ()
240+ push !logItem = do
241+ !curBuffer <- STM.readTVar bufferState
105242 processLogAction (bufferItem itemBufferSize verbosity logItem curBuffer)
106243
107244
108245 -- A transaction that retries until its time to shutdown
109− let waitShutdown = do
246+ let waitShutdown :: STM.STM ()
247+ waitShutdown = do
110248 killStatus <- STM.readTVar killSwitch
111249 STM.check (killStatus == ShutDown)
112250
113− --TODO: exception handling
114−
115− -- Retry transaction until timer has expired
116− let waitTimer = do
251+ -- Retry transaction until timer has expired. When it does, take action on the buffer.
252+ let waitTimer :: STM.STM ()
253+ waitTimer = do
117254 -- get the timer we'll be using
118255 timer <- STM.readTVar timerRef
119256 -- do not proceed until its ready
@@ -122,12 +259,13 @@ mkLogzIOScribe config permitItem verbosity = do
122259 -- do not wake again until this message is out the door
123260 STM.writeTVar timer False
124261 -- we waited long enough, grab the current buffer
125− curBuffer <- STM.readTVar bufferState
262+ !curBuffer <- STM.readTVar bufferState
126263 -- and tell it to flush right now
127264 processLogAction (forceFlush curBuffer)
128265
129− -- loop until its time to shut down
130− let timedFlushLoop = do
266+ -- run the timer in a loop until its time to shut down
267+ let timedFlushLoop :: IO ()
268+ timedFlushLoop = do
131269 res <- STM.atomically $
132270 raceAlt
133271 waitShutdown
@@ -138,7 +276,9 @@ mkLogzIOScribe config permitItem verbosity = do
138276
139277 timedFlusher <- Async.async timedFlushLoop
140278
141− let flushLoop = do
279+ -- flush the buffer until its time to shutdown and we're out of items to log
280+ let flushLoop :: IO ()
281+ flushLoop = do
142282 -- 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
143283 readyBufferOrDie <- STM.atomically $
144284 raceAlt
@@ -149,16 +289,24 @@ mkLogzIOScribe config permitItem verbosity = do
149289 -- There's some logs to write
150290 Left readyBuffer -> do
151291 -- flush in IO
152− flushBuffer readyBuffer
292+ res <- flushBuffer config mgr readyBuffer
293+ case res of
294+ Left ex -> onErrorSafe ex
295+ Right () -> pure ()
153296 -- reset the timeout before the next flush on each write
154− newTimer <- STM.registerDelay itemBufferTimeoutMicros
297+ !newTimer <- STM.registerDelay itemBufferTimeoutMicros
155298 STM.atomically $ do
156299 STM.writeTVar timerRef newTimer
157300 -- start over
158301 flushLoop
159302 flusher <- Async.async flushLoop
160303
161− let shutdown = do
304+ -- Blocking call which sends a poison pill to the workers and waits
305+ -- for them to exit, swallowing exceptions. This follows the
306+ -- semantics that when katip calls the finalizer, no more log
307+ -- entries will be sent.
308+ let shutdown :: IO ()
309+ shutdown = do
162310 -- tell workers this is the last loop
163311 STM.atomically (STM.writeTVar killSwitch ShutDown)
164312 -- wait for workers to finish
@@ -169,13 +317,18 @@ mkLogzIOScribe config permitItem verbosity = do
169317
170318
171319 pure $ K.Scribe
172− { liPush = STM.atomically . push
173− , scribeFinalizer = shutdown
174− , scribePermitItem = permitItem
320+ { K.liPush = STM.atomically . push
321+ , K.scribeFinalizer = shutdown
322+ , K.scribePermitItem = permitItem
175323 }
176324 where
177325 itemBufferSize = logzIOScribeConfiguration_bufferItems config
178326 itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
327+ scheme = logzIOScribeConfiguration_scheme config
328+ -- should this be async?
329+ onErrorSafe ex = do
330+ _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
331+ pure ()
179332
180333
181334-------------------------------------------------------------------------------
@@ -196,6 +349,9 @@ ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
196349
197350
198351-------------------------------------------------------------------------------
352+-- | An internal mode flag used to shut down worker
353+-- threads. Internally, we assume htere is no going back from
354+-- ShutDown.
199355data KillSwitch =
200356 Continue
201357 | ShutDown
@@ -203,14 +359,84 @@ data KillSwitch =
203359
204360
205361-------------------------------------------------------------------------------
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.
206364raceAlt :: (Alternative m) => m a -> m b -> m (Either a b)
207365raceAlt a b = (Left <$> a) <|> (Right <$> b)
208366
209367
210368-------------------------------------------------------------------------------
211−-- TODO: noop on empty
212−flushBuffer :: BulkBuffer -> IO ()
213−flushBuffer = undefined
369+data LoggingError =
370+ URIError HTTP.HttpException
371+ -- ^ The URI generated was invalid. Check your configuration
372+ | RequestError HTTP.HttpException
373+ -- ^ We encountered an exception while sending the batch request
374+ | PartialFailure BulkAPIError
375+ -- ^ Some or all of the request was rejected. Check the logz.io UI
376+ -- for indexing errors.
377+ | BadToken
378+ -- ^ Your API token was rejected.
379+ | UnknownFailureResponse HTypes.Status LBS.ByteString
380+ -- ^ An error returned, but it could not be decoded into a
381+ -- 'BulkAPIError'. This may indicate a __library bug__, which should
382+ -- be reported to the issue tracker.
383+ deriving (Show)
384+
385+-------------------------------------------------------------------------------
386+flushBuffer
387+ :: LogzIOScribeConfiguration
388+ -> HTTP.Manager
389+ -> BulkBuffer
390+ -> IO (Either LoggingError ())
391+flushBuffer config mgr bulkBuffer
392+ | bulkBuffer_itemCount bulkBuffer <= 0 = pure (Right ())
393+ | otherwise = do
394+ E.runExceptT $ do
395+ req <- E.fmapLT URIError (E.ExceptT (EX.try (configureRequest <$> HTTP.parseRequest uriStr)))
396+ resp <- E.fmapLT RequestError $ E.ExceptT $ EX.try $
397+ Retry.recovering retryPolicy [\_stat -> EX.Handler handleHttpException] $ \_stat ->
398+ HTTP.httpLbs req mgr
399+ let respLBS = HTTP.responseBody resp
400+ let respStatus = HTTP.responseStatus resp
401+ if HTypes.statusIsSuccessful respStatus
402+ then pure ()
403+ else case A.decode @BulkAPIError respLBS of
404+ Nothing
405+ | HTypes.statusCode respStatus == 401 -> E.throwE BadToken
406+ | otherwise -> E.throwE (UnknownFailureResponse respStatus respLBS)
407+ Just bulkError -> E.throwE (PartialFailure bulkError)
408+ where
409+ configureRequest req = req
410+ { HTTP.method = HTypes.methodPost
411+ , HTTP.requestBody = HTTP.RequestBodyLBS (BB.toLazyByteString (bulkBuffer_payload bulkBuffer))
412+ }
413+
414+ retryPolicy = logzIOScribeConfiguration_retry config
415+ apiTokenBS = TE.encodeUtf8 (apiToken (logzIOScribeConfiguration_token config))
416+
417+ handleHttpException :: (Applicative m) => HTTP.HttpException -> m Bool
418+ handleHttpException _ = pure True
419+
420+ uriStr = LBS8.unpack (BB.toLazyByteString (URIBS.serializeURIRef uri))
421+
422+ authority = URIBS.Authority
423+ { URIBS.authorityUserInfo = Nothing
424+ , URIBS.authorityHost = logzIOScribeConfiguration_host config
425+ , URIBS.authorityPort = Just (logzIOScribeConfiguration_port config)
426+ }
427+
428+ uri = URIBS.URI
429+ { URIBS.uriScheme = case logzIOScribeConfiguration_scheme config of
430+ HTTPS -> URIBS.Scheme "https"
431+ HTTP -> URIBS.Scheme "http"
432+ , URIBS.uriAuthority = Just authority
433+ , URIBS.uriPath = "/"
434+ , URIBS.uriQuery = URIBS.Query
435+ [ ("token", apiTokenBS)
436+ ]
437+ , URIBS.uriFragment = Nothing
438+ }
439+
214440
215441-------------------------------------------------------------------------------
216442newtype Bytes = Bytes
@@ -231,10 +457,10 @@ maxLogLineLength :: Bytes
231457maxLogLineLength = 500000
232458
233459
234−measureJSON :: A.ToJSON a => a -> (BB.Builder, Bytes)
235−measureJSON a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
460+measureJSONLine :: A.ToJSON a => a -> (BB.Builder, Bytes)
461+measureJSONLine a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
236462 where
237− lbs = A.encode a
463+ lbs = A.encode a <> "\n"
238464
239465
240466-- | Fully-rendered JSON object for an item
@@ -249,32 +475,48 @@ fullItemObject verbosity item = HM.fromList
249475 , "data" A..= K.payloadObject verbosity (K._itemPayload item)
250476 -- Slight deviation, logz.io uses "message" instead of "msg"
251477 , "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?
478+ -- Another slight deviation, logz.io uses "@timestamp" instead of
479+ -- "at". Note, your logs should be sent roughly close to when they
480+ -- are created. They are assigned to indexes based on index date, so
481+ -- time searches act strange if you backfill too far from the
482+ -- past. They seem to support 3 decimal places of
483+ -- precision. Formatting like this requires time 1.8.0.2, which is
484+ -- reflected in the cabal file.
485+ , "@timestamp" A..= A.String (T.pack (Time.formatTime Time.defaultTimeLocale "%Y-%m-%dT%H:%M:%03QZ" (K._itemTime item)))
258486 , "ns" A..= K._itemNamespace item
259487 , "loc" A..= (LocJs <$> K._itemLoc item)
260488 ]
261489 where
262490 POSIX.CPid pidInt = K._itemProcess item
263491
264−--TODO: property test
265492
266493-- | A version of 'renderLine' which renders a line and stays under
267494-- the maximum line size of 500,000 bytes. If the default rendering is
268495-- too large, the log will be reduced to a timestamp and a potentially
269496-- truncated message.
270497renderLineTruncated :: K.LogItem a => K.Verbosity -> K.Item a -> (BB.Builder, Bytes)
271−renderLineTruncated verbosity item =
272− if fullSize <= maxLogLineLength
498+renderLineTruncated = renderLineTruncated' maxLogLineLength
499+
500+
501+-- | A generalized renderLineTruncated that takes a custom line length
502+-- limit. This is exclusively for testing.
503+renderLineTruncated'
504+ :: K.LogItem a
505+ => Bytes
506+ -- ^ Custom max log line length. Be careful, too low and no amount
507+ -- of shrinking can get under the size, breaking the invariant. For
508+ -- the production limit, we are guraanteed always able to come in
509+ -- under the limit.
510+ -> K.Verbosity
511+ -> K.Item a
512+ -> (BB.Builder, Bytes)
513+renderLineTruncated' customMaxLogLineLength verbosity item =
514+ if fullSize <= customMaxLogLineLength
273515 then (fullLine, fullSize)
274516 else (fallbackLine, fallbackSize)
275517 where
276518 fullObject = fullItemObject verbosity item
277− (fullLine, fullSize) = measureJSON fullObject
519+ (fullLine, fullSize) = measureJSONLine fullObject
278520 -- only the absolutely necessary keys, with message stripped out
279521 -- to help us calculate how much of the message we can keep. these
280522 -- are lazily evaluated and as such won't be computed unless the
@@ -284,9 +526,9 @@ renderLineTruncated verbosity item =
284526 [ "message" A..= A.String "" -- we'll start with a blank message
285527 , "@timestamp" A..= K._itemTime item
286528 ]
287− (_, blankObjectSize) = measureJSON blankObject
529+ (_, blankObjectSize) = measureJSONLine blankObject
288530 messageBytesAllowed = maxLogLineLength - blankObjectSize
289− (fallbackLine, fallbackSize) = measureJSON fallbackObject
531+ (fallbackLine, fallbackSize) = measureJSONLine fallbackObject
290532 fallbackObject :: A.Object
291533 fallbackObject = HM.fromList
292534 [ "message" A..= A.toJSON (TL.take (bytes messageBytesAllowed) (TB.toLazyText (K.unLogStr (K._itemMessage item))))
@@ -332,11 +574,30 @@ bufferItem
332574 -> K.Item a
333575 -> BulkBuffer
334576 -> LogAction BulkBuffer
335−bufferItem maxItems verb item bulkBuffer =
336− let (encodedLine, spaceNeeded) = renderLineTruncated verb item
577+bufferItem = bufferItem' maxPayloadBytes maxLogLineLength
578+
579+
580+-- | An internal version with a configurable max item size and max
581+-- message size for testing. Be careful with this: if you set the
582+-- values too low, some of the guarantees about reducability of
583+-- messages break down.
584+bufferItem'
585+ :: (K.LogItem a)
586+ => Bytes
587+ -- ^ Max payload size
588+ -> Bytes
589+ -- ^ Max item size
590+ -> Int
591+ -- ^ Maximum items before flushing
592+ -> K.Verbosity
593+ -> K.Item a
594+ -> BulkBuffer
595+ -> LogAction BulkBuffer
596+bufferItem' customMaxPayload customMaxItem maxItems verb item bulkBuffer =
597+ let (encodedLine, spaceNeeded) = renderLineTruncated' customMaxItem verb item
337598 newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded
338599 newItemCount = bulkBuffer_itemCount bulkBuffer + 1
339− in if newItemCount >= maxItems || newBytesUsed >= maxPayloadBytes
600+ in if newItemCount > maxItems || newBytesUsed > customMaxPayload
340601 then FlushNow
341602 bulkBuffer
342603 BulkBuffer