@@ -0,0 +1,639 @@
1+{-# LANGUAGE BangPatterns #-}
2+{-# LANGUAGE GADTs #-}
3+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
4+{-# LANGUAGE OverloadedStrings #-}
5+{-# LANGUAGE RankNTypes #-}
6+{-# LANGUAGE TypeApplications #-}
7+-- | This is a log scribe that writes logs to logz.io's bulk
8+-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS HTTPS
9+-- API>.
10+module Katip.Scribes.LogzIO.HTTPS
11+ ( -- * Scribe construction
12+ mkLogzIOScribe
13+ -- * Types
14+ , BulkAPIError(..)
15+ , LogzIOScribeConfiguration(..)
16+ , Scheme(..)
17+ , APIToken(..)
18+ , LoggingError(..)
19+ -- ** Presets for configuration
20+ , usRegionHost
21+ , euRegionHost
22+ , httpsPort
23+ , httpPort
24+ , defaultRetryPolicy
25+ , defaultLogzIOScribeConfiguration
26+ -- * Internal API exported for testing
27+ , renderLineTruncated
28+ , renderLineTruncated'
29+ , maxPayloadBytes
30+ , maxLogLineLength
31+ , BulkBuffer (..)
32+ , LogAction (..)
33+ , bufferItem
34+ , bufferItem'
35+ , forceFlush
36+ , Bytes (..)
37+ ) where
38+
39+
40+-------------------------------------------------------------------------------
41+import Control.Applicative
42+import qualified Control.Concurrent.Async as Async
43+import qualified Control.Concurrent.STM as STM
44+import qualified Control.Concurrent.STM.TBMQueue as STM
45+import qualified Control.Error as E
46+import qualified Control.Exception.Safe as EX
47+import Control.Monad
48+import qualified Control.Retry as Retry
49+import qualified Data.Aeson as A
50+import qualified Data.ByteString.Builder as BB
51+import qualified Data.ByteString.Lazy as LBS
52+import qualified Data.ByteString.Lazy.Char8 as LBS8
53+import qualified Data.HashMap.Strict as HM
54+import Data.Int
55+import qualified Data.Scientific as Scientific
56+import Data.Semigroup as Semigroup
57+import Data.String (IsString)
58+import qualified Data.Text as T
59+import qualified Data.Text.Encoding as TE
60+import qualified Data.Text.Lazy as TL
61+import qualified Data.Text.Lazy.Builder as TB
62+import qualified Data.Time as Time
63+import qualified Katip as K
64+import Katip.Core (LocJs (..))
65+import qualified Network.HTTP.Client as HTTP
66+import qualified Network.HTTP.Client.TLS as HTTPS
67+import qualified Network.HTTP.Types as HTypes
68+import qualified System.Posix.Types as POSIX
69+import qualified URI.ByteString as URIBS
70+-------------------------------------------------------------------------------
71+
72+
73+
74+-- | This is returned when the bulk import was a partial success
75+data BulkAPIError = BulkAPIError
76+ { bulkAPIError_malformedLines :: Int
77+ -- ^ The number of log lines which are not well-formed JSON. This
78+ -- indicates a __library bug__. Please file an issue on GitHub.
79+ , bulkAPIError_successfulLines :: Int
80+ -- ^ The number of log lines received successfully.
81+ , bulkAPIError_oversizedLines :: Int
82+ -- ^ The number of log lines which exceed the line length
83+ -- limit. katip-logzio makes a best effort to truncate logs that
84+ -- exceed the size limit. This probably indicates a __library bug__
85+ -- and should be reported as an issue on GitHub.
86+ , bulkAPIError_emptyLogLines :: Int
87+ -- ^ The number of log lines which were empty. There isnt' really a
88+ -- concept of a truly empty log line in katip, so this most likely
89+ -- indicates a __library bug__ and should be reported as an issue on
90+ -- GitHub.
91+ } deriving (Show, Eq)
92+
93+
94+instance A.FromJSON BulkAPIError where
95+ parseJSON = A.withObject "BulkAPIError" $ \o -> do
96+ malformedLines <- o A..: "malformedLines"
97+ successfulLines <- o A..: "successfulLines"
98+ oversizedLines <- o A..: "oversizedLines"
99+ emptyLogLines <- o A..: "emptyLogLines"
100+ pure $ BulkAPIError
101+ { bulkAPIError_malformedLines = malformedLines
102+ , bulkAPIError_successfulLines = successfulLines
103+ , bulkAPIError_oversizedLines = oversizedLines
104+ , bulkAPIError_emptyLogLines = emptyLogLines
105+ }
106+
107+
108+-------------------------------------------------------------------------------
109+data LogzIOScribeConfiguration = LogzIOScribeConfiguration
110+ { logzIOScribeConfiguration_bufferItems :: Int
111+ -- ^ Will flush the log buffer if this many items is in the buffer
112+ -- __or__ 'logzIOScribeConfiguration_bufferTimeout' is reached,
113+ -- whichever is first
114+ , logzIOScribeConfiguration_bufferTimeout :: Time.NominalDiffTime
115+ -- ^ Will flush the buffer if it has been this long since the last
116+ -- flush __or__ 'logzIOScribeConfiguration_bufferItems' items are
117+ -- accumulated, whichever is first. NominalDiffTime has a Num
118+ -- instance, so you can use a literal to specify seconds, e.g. 30 =
119+ -- 30s.
120+ , logzIOScribeConfiguration_scheme :: Scheme
121+ , logzIOScribeConfiguration_host :: URIBS.Host
122+ , logzIOScribeConfiguration_port :: URIBS.Port
123+ , logzIOScribeConfiguration_token :: APIToken
124+ , logzIOScribeConfiguration_retry :: Retry.RetryPolicyM IO
125+ -- ^ How should exceptions during writes be retried?
126+ , logzIOScribeConfiguration_onError :: LoggingError -> IO ()
127+ }
128+
129+
130+-- | A default configuration:
131+--
132+-- * 100 item buffering
133+--
134+-- * 30 second buffer timeout
135+--
136+-- * US, HTTPS logging endpoint (listener.logz.io:8071)
137+--
138+-- * 'defaultRetryPolicy' of 25ms exponential backoff, 5 retries
139+--
140+-- * Ignore logging errors
141+defaultLogzIOScribeConfiguration :: APIToken -> LogzIOScribeConfiguration
142+defaultLogzIOScribeConfiguration token = LogzIOScribeConfiguration
143+ { logzIOScribeConfiguration_bufferItems = 100
144+ , logzIOScribeConfiguration_bufferTimeout = 30
145+ , logzIOScribeConfiguration_scheme = HTTPS
146+ , logzIOScribeConfiguration_host = usRegionHost
147+ , logzIOScribeConfiguration_port = httpsPort
148+ , logzIOScribeConfiguration_token = token
149+ , logzIOScribeConfiguration_retry = defaultRetryPolicy
150+ , logzIOScribeConfiguration_onError = const (pure ())
151+ }
152+
153+
154+-------------------------------------------------------------------------------
155+-- | You can retrieve your account or sub-account's API token on the
156+-- <https://app.logz.io/#/dashboard/settings/manage-accounts manage
157+-- accounts page>. Note that APIToken has an IsString instance,
158+-- meaning that you can use a string literal with OverloadedStrings
159+-- enabled.
160+newtype APIToken = APIToken
161+ { apiToken :: T.Text
162+ } deriving (Show, Eq, IsString)
163+
164+
165+-- | This particular bulk API only supports HTTP or HTTPS. HTTPS is
166+-- strongly recommended for security reasons.
167+data Scheme
168+ = HTTPS
169+ -- ^ HTTPs should always be used except for local testing.
170+ | HTTP
171+ deriving (Show, Eq)
172+
173+
174+-- | See
175+-- <https://support.logz.io/hc/en-us/articles/210206365-What-IP-addresses-should-I-open-in-my-firewall-to-ship-logs-to-Logz-io-
176+-- this> for a list of listeners. This is the US region host,
177+-- listener.logz.io
178+usRegionHost :: URIBS.Host
179+usRegionHost = URIBS.Host "listener.logz.io"
180+
181+
182+-- | See
183+-- <https://support.logz.io/hc/en-us/articles/210206365-What-IP-addresses-should-I-open-in-my-firewall-to-ship-logs-to-Logz-io-
184+-- this> for a list of listeners. This is the EU region host,
185+-- listener.logz.io
186+euRegionHost :: URIBS.Host
187+euRegionHost = URIBS.Host "listener-eu.logz.io"
188+
189+
190+-- | Logz.io uses port 8071 for HTTPS
191+httpsPort :: URIBS.Port
192+httpsPort = URIBS.Port 8071
193+
194+
195+-- | Logz.io uses port 8070 for HTTP
196+httpPort :: URIBS.Port
197+httpPort = URIBS.Port 8070
198+
199+
200+-- | A reasonable retry policy: exponential backoff with 25ms base
201+-- delay up to 5 retries, for a total cumulative delay of 775ms.
202+defaultRetryPolicy :: (Monad m) => Retry.RetryPolicyM m
203+defaultRetryPolicy = Retry.exponentialBackoff 25000 `mappend` Retry.limitRetries 5
204+
205+-------------------------------------------------------------------------------
206+mkLogzIOScribe
207+ :: LogzIOScribeConfiguration
208+ -> K.PermitFunc
209+ -> K.Verbosity
210+ -> IO K.Scribe
211+mkLogzIOScribe config permitItem verbosity = do
212+ -- This size is actually somewhat arbitrary. We're just making sure
213+ -- it isn't infinite so we don't consume the whole backlog right
214+ -- away and take up lots of memory.. Katip above us does bounded
215+ -- buffering and load-shedding when there's too much of a
216+ -- backlog. Our writes are blocking to apply backpressure up to
217+ -- katip. This value just holds onto values temporarily while we
218+ -- concatenate them into a buffer.
219+ ingestionQueue <- STM.newTBMQueueIO (itemBufferSize * 10)
220+
221+ let newTimer = STM.registerDelay itemBufferTimeoutMicros
222+ timerRef <- STM.newTVarIO =<< newTimer
223+
224+ -- Set up a connection manager for requests
225+ mgr <- case scheme of
226+ HTTPS -> HTTPS.newTlsManager
227+ HTTP -> HTTP.newManager HTTP.defaultManagerSettings
228+
229+ -- An STM transaction that will return true when writes are stopped
230+ -- and backlog is emptied.
231+ let workExhausted :: STM.STM Bool
232+ workExhausted = (&&)
233+ <$> STM.isClosedTBMQueue ingestionQueue
234+ <*> STM.isEmptyTBMQueue ingestionQueue
235+ -- Block until there's no more work
236+ let waitWorkExhausted :: STM.STM ()
237+ waitWorkExhausted = STM.check =<< workExhausted
238+ let pop :: STM.STM (Tick AnyLogItem)
239+ pop = maybe WorkExhausted NewItem <$> STM.readTBMQueue ingestionQueue
240+
241+ let timeExpired :: STM.STM (Tick a)
242+ timeExpired = do
243+ isExpired <- STM.readTVar =<< STM.readTVar timerRef
244+ STM.check isExpired
245+ pure TimeExpired
246+
247+ -- Circular transaction that checks for completion of work, time
248+ -- expiration, or new events.
249+ let nextTick :: STM.STM (Tick AnyLogItem)
250+ nextTick =
251+ timeExpired <|>
252+ pop <|>
253+ (WorkExhausted <$ waitWorkExhausted)
254+
255+ -- Blocking push. This applies backpressure upstream to katip, which does its own buffering
256+ let push :: AnyLogItem -> STM.STM ()
257+ push = void . STM.writeTBMQueue ingestionQueue
258+
259+ let sealQueue = STM.atomically (STM.closeTBMQueue ingestionQueue)
260+
261+ -- Replace the timer with a new one
262+ let resetTimer = STM.atomically . STM.writeTVar timerRef =<< newTimer
263+
264+ -- Send the buffer and then reset the timer
265+ let flush curBuffer = do
266+ res <- flushBuffer config mgr curBuffer
267+ case res of
268+ Left e -> onErrorSafe e
269+ Right () -> pure ()
270+ resetTimer
271+
272+ let flushLoop :: BulkBuffer -> IO ()
273+ flushLoop curBuffer = do
274+ tick <- STM.atomically nextTick
275+ case tick of
276+ WorkExhausted -> do
277+ flush curBuffer -- flush what you've got
278+ pure () -- stop looping
279+ TimeExpired -> do
280+ flush curBuffer
281+ flushLoop mempty
282+ NewItem (AnyLogItem item) -> do
283+ case bufferItem (logzIOScribeConfiguration_bufferItems config) verbosity item curBuffer of
284+ Buffered newBuffer -> flushLoop newBuffer
285+ FlushNow flushThis newBuffer -> do
286+ flush flushThis
287+ flushLoop newBuffer
288+
289+ flushThread <- Async.async (flushLoop mempty)
290+
291+ let close = do
292+ sealQueue
293+ _ <- Async.waitCatch flushThread
294+ pure ()
295+
296+ pure $ K.Scribe
297+ { K.liPush = STM.atomically . push . AnyLogItem
298+ , K.scribeFinalizer = close
299+ , K.scribePermitItem = permitItem
300+ }
301+ where
302+ itemBufferSize = logzIOScribeConfiguration_bufferItems config
303+ itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
304+ scheme = logzIOScribeConfiguration_scheme config
305+ onErrorSafe ex = do
306+ _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
307+ pure ()
308+
309+
310+data AnyLogItem where
311+ AnyLogItem :: K.LogItem a => K.Item a -> AnyLogItem
312+
313+
314+-------------------------------------------------------------------------------
315+data Tick a =
316+ TimeExpired
317+ | NewItem !a
318+ | WorkExhausted
319+
320+
321+-------------------------------------------------------------------------------
322+-- Match the native resolution of NominalDiffTime, which is excessive
323+-- for our needs
324+ndtPicos :: Time.NominalDiffTime -> Int64
325+ndtPicos = round . (* picos)
326+ where
327+ picos :: Time.NominalDiffTime
328+ picos = 10 ^ (9 :: Int)
329+
330+
331+-------------------------------------------------------------------------------
332+ndtToMicros :: Time.NominalDiffTime -> Int
333+ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
334+ where
335+ picosInMicro = 10 ^ (3 :: Int)
336+
337+
338+-------------------------------------------------------------------------------
339+data LoggingError =
340+ URIError HTTP.HttpException
341+ -- ^ The URI generated was invalid. Check your configuration
342+ | RequestError HTTP.HttpException
343+ -- ^ We encountered an exception while sending the batch request
344+ | PartialFailure BulkAPIError
345+ -- ^ Some or all of the request was rejected. Check the logz.io UI
346+ -- for indexing errors.
347+ | BadToken
348+ -- ^ Your API token was rejected.
349+ | UnknownFailureResponse HTypes.Status LBS.ByteString
350+ -- ^ An error returned, but it could not be decoded into a
351+ -- 'BulkAPIError'. This may indicate a __library bug__, which should
352+ -- be reported to the issue tracker.
353+ deriving (Show)
354+
355+-------------------------------------------------------------------------------
356+flushBuffer
357+ :: LogzIOScribeConfiguration
358+ -> HTTP.Manager
359+ -> BulkBuffer
360+ -> IO (Either LoggingError ())
361+flushBuffer config mgr bulkBuffer
362+ | bulkBuffer_itemCount bulkBuffer <= 0 = pure (Right ())
363+ | otherwise = do
364+ E.runExceptT $ do
365+ req <- E.fmapLT URIError (E.ExceptT (EX.try (configureRequest <$> HTTP.parseRequest uriStr)))
366+ resp <- E.fmapLT RequestError $ E.ExceptT $ EX.try $
367+ Retry.recovering retryPolicy [\_stat -> EX.Handler handleHttpException] $ \_stat ->
368+ HTTP.httpLbs req mgr
369+ let respLBS = HTTP.responseBody resp
370+ let respStatus = HTTP.responseStatus resp
371+ if HTypes.statusIsSuccessful respStatus
372+ then pure ()
373+ else case A.decode @BulkAPIError respLBS of
374+ Nothing
375+ | HTypes.statusCode respStatus == 401 -> E.throwE BadToken
376+ | otherwise -> E.throwE (UnknownFailureResponse respStatus respLBS)
377+ Just bulkError -> E.throwE (PartialFailure bulkError)
378+ where
379+ configureRequest req = req
380+ { HTTP.method = HTypes.methodPost
381+ , HTTP.requestBody = HTTP.RequestBodyLBS (BB.toLazyByteString (bulkBuffer_payload bulkBuffer))
382+ }
383+
384+ retryPolicy = logzIOScribeConfiguration_retry config
385+ apiTokenBS = TE.encodeUtf8 (apiToken (logzIOScribeConfiguration_token config))
386+
387+ handleHttpException :: (Applicative m) => HTTP.HttpException -> m Bool
388+ handleHttpException _ = pure True
389+
390+ uriStr = LBS8.unpack (BB.toLazyByteString (URIBS.serializeURIRef uri))
391+
392+ authority = URIBS.Authority
393+ { URIBS.authorityUserInfo = Nothing
394+ , URIBS.authorityHost = logzIOScribeConfiguration_host config
395+ , URIBS.authorityPort = Just (logzIOScribeConfiguration_port config)
396+ }
397+
398+ uri = URIBS.URI
399+ { URIBS.uriScheme = case logzIOScribeConfiguration_scheme config of
400+ HTTPS -> URIBS.Scheme "https"
401+ HTTP -> URIBS.Scheme "http"
402+ , URIBS.uriAuthority = Just authority
403+ , URIBS.uriPath = "/"
404+ , URIBS.uriQuery = URIBS.Query
405+ [ ("token", apiTokenBS)
406+ ]
407+ , URIBS.uriFragment = Nothing
408+ }
409+
410+
411+-------------------------------------------------------------------------------
412+newtype Bytes = Bytes
413+ { bytes :: Int64
414+ } deriving (Show, Eq, Num, Ord, Bounded)
415+
416+
417+-- | How big of a body can we send? The limit is defined
418+-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS here>.
419+maxPayloadBytes :: Bytes
420+maxPayloadBytes = 10485760
421+
422+
423+-- | How long can each serialized payload be (let's assume including
424+-- the trailing newline). The limit is defined
425+-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS here>.
426+maxLogLineLength :: Bytes
427+maxLogLineLength = 500000
428+
429+
430+measureJSONLine :: A.ToJSON a => a -> (BB.Builder, Bytes)
431+measureJSONLine a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
432+ where
433+ lbs = A.encode a <> "\n"
434+
435+
436+-- | Fully-rendered JSON object for an item
437+fullItemObject :: K.LogItem a => K.Verbosity -> K.Item a -> A.Object
438+fullItemObject verbosity item = HM.fromList
439+ [ "app" A..= K._itemApp item
440+ , "env" A..= K._itemEnv item
441+ , "sev" A..= K._itemSeverity item
442+ , "thread" A..= K.getThreadIdText (K._itemThread item)
443+ , "host" A..= K._itemHost item
444+ , "pid" A..= pidInt
445+ , "data" A..= annotateKeys (K.payloadObject verbosity (K._itemPayload item))
446+ -- Slight deviation, logz.io uses "message" instead of "msg"
447+ , "message" A..= TB.toLazyText (K.unLogStr (K._itemMessage item))
448+ -- Another slight deviation, logz.io uses "@timestamp" instead of
449+ -- "at". Note, your logs should be sent roughly close to when they
450+ -- are created. They are assigned to indexes based on index date, so
451+ -- time searches act strange if you backfill too far from the
452+ -- past. They seem to support 3 decimal places of
453+ -- precision. Formatting like this requires time 1.8.0.2, which is
454+ -- 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)))
456+ , "ns" A..= K._itemNamespace item
457+ , "loc" A..= (LocJs <$> K._itemLoc item)
458+ ]
459+ where
460+ POSIX.CPid pidInt = K._itemProcess item
461+
462+
463+-- | A version of 'renderLine' which renders a line and stays under
464+-- the maximum line size of 500,000 bytes. If the default rendering is
465+-- too large, the log will be reduced to a timestamp and a potentially
466+-- truncated message.
467+renderLineTruncated :: K.LogItem a => K.Verbosity -> K.Item a -> (BB.Builder, Bytes)
468+renderLineTruncated = renderLineTruncated' maxLogLineLength
469+
470+
471+-- | A generalized renderLineTruncated that takes a custom line length
472+-- limit. This is exclusively for testing.
473+renderLineTruncated'
474+ :: K.LogItem a
475+ => Bytes
476+ -- ^ Custom max log line length. Be careful, too low and no amount
477+ -- of shrinking can get under the size, breaking the invariant. For
478+ -- the production limit, we are guraanteed always able to come in
479+ -- under the limit.
480+ -> K.Verbosity
481+ -> K.Item a
482+ -> (BB.Builder, Bytes)
483+renderLineTruncated' customMaxLogLineLength verbosity item =
484+ if fullSize <= customMaxLogLineLength
485+ then (fullLine, fullSize)
486+ else (fallbackLine, fallbackSize)
487+ where
488+ fullObject = fullItemObject verbosity item
489+ (fullLine, fullSize) = measureJSONLine fullObject
490+ -- only the absolutely necessary keys, with message stripped out
491+ -- to help us calculate how much of the message we can keep. these
492+ -- are lazily evaluated and as such won't be computed unless the
493+ -- item is too big
494+ blankObject :: A.Object
495+ blankObject = HM.fromList
496+ [ "message" A..= A.String "" -- we'll start with a blank message
497+ , "@timestamp" A..= K._itemTime item
498+ ]
499+ (_, blankObjectSize) = measureJSONLine blankObject
500+ messageBytesAllowed = maxLogLineLength - blankObjectSize
501+ (fallbackLine, fallbackSize) = measureJSONLine fallbackObject
502+ fallbackObject :: A.Object
503+ fallbackObject = HM.fromList
504+ [ "message" A..= A.toJSON (TL.take (bytes messageBytesAllowed) (TB.toLazyText (K.unLogStr (K._itemMessage item))))
505+ , "@timestamp" A..= A.toJSON (K._itemTime item)
506+ ]
507+
508+
509+data BulkBuffer = BulkBuffer
510+ { bulkBuffer_bytesUsed :: !Bytes
511+ , bulkBuffer_payload :: !BB.Builder
512+ , bulkBuffer_itemCount :: !Int
513+ }
514+
515+
516+instance Semigroup.Semigroup BulkBuffer where
517+ (BulkBuffer bytesUsedA bufferA itemCountA) <>
518+ (BulkBuffer bytesUsedB bufferB itemCountB) = BulkBuffer
519+ (bytesUsedA + bytesUsedB)
520+ (bufferA <> bufferB)
521+ (itemCountA + itemCountB)
522+
523+
524+instance Monoid BulkBuffer where
525+ mempty = BulkBuffer 0 mempty 0
526+ mappend = (<>)
527+
528+
529+data LogAction buf =
530+ Buffered
531+ buf
532+ -- ^ New buffer
533+ | FlushNow
534+ buf
535+ -- ^ Buffer to flush
536+ buf
537+ -- ^ New buffer
538+
539+bufferItem
540+ :: K.LogItem a
541+ => Int
542+ -- ^ Maximum items before flushing
543+ -> K.Verbosity
544+ -> K.Item a
545+ -> BulkBuffer
546+ -> LogAction BulkBuffer
547+bufferItem = bufferItem' maxPayloadBytes maxLogLineLength
548+
549+
550+-- | An internal version with a configurable max item size and max
551+-- message size for testing. Be careful with this: if you set the
552+-- values too low, some of the guarantees about reducability of
553+-- messages break down.
554+bufferItem'
555+ :: (K.LogItem a)
556+ => Bytes
557+ -- ^ Max payload size
558+ -> Bytes
559+ -- ^ Max item size
560+ -> Int
561+ -- ^ Maximum items before flushing
562+ -> K.Verbosity
563+ -> K.Item a
564+ -> BulkBuffer
565+ -> LogAction BulkBuffer
566+bufferItem' customMaxPayload customMaxItem maxItems verb item bulkBuffer =
567+ let (encodedLine, spaceNeeded) = renderLineTruncated' customMaxItem verb item
568+ newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded
569+ newItemCount = bulkBuffer_itemCount bulkBuffer + 1
570+ in if newItemCount >= maxItems || newBytesUsed >= customMaxPayload
571+ then FlushNow
572+ bulkBuffer
573+ BulkBuffer
574+ { bulkBuffer_bytesUsed = spaceNeeded
575+ , bulkBuffer_payload = encodedLine
576+ , bulkBuffer_itemCount = 1
577+ }
578+ else Buffered $
579+ BulkBuffer
580+ { bulkBuffer_bytesUsed = newBytesUsed
581+ , bulkBuffer_payload = bulkBuffer_payload bulkBuffer <> encodedLine
582+ , bulkBuffer_itemCount = newItemCount
583+ }
584+
585+
586+-- | When time has run out on a flush, splits the buffer. n.b. this
587+-- will always return 'FlushNow' with an empty new buffer.
588+forceFlush :: (Monoid buf) => buf -> LogAction buf
589+forceFlush buf = FlushNow buf mempty
590+
591+
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
600+-- happen.
601+
602+annotateValue :: A.Value -> A.Value
603+annotateValue (A.Object o) = A.Object (annotateKeys o)
604+annotateValue (A.Array a) = A.Array (annotateValue <$> a)
605+annotateValue x = x
606+
607+
608+annotateKeys :: A.Object -> A.Object
609+annotateKeys = HM.fromList . map go . HM.toList
610+ where
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)
619+
620+
621+-------------------------------------------------------------------------------
622+-- Annotation Constants
623+-------------------------------------------------------------------------------
624+
625+
626+stringAnn :: T.Text
627+stringAnn = "::s"
628+
629+doubleAnn :: T.Text
630+doubleAnn = "::d"
631+
632+longAnn :: T.Text
633+longAnn = "::l"
634+
635+booleanAnn :: T.Text
636+booleanAnn = "::b"
637+
638+nullAnn :: T.Text
639+nullAnn = "::n"