@@ -1,639 +0,0 @@
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"