@@ -1,3 +1,9 @@
1+{-# LANGUAGE BangPatterns #-}
2+{-# LANGUAGE GADTs #-}
3+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
4+{-# LANGUAGE OverloadedStrings #-}
5+{-# LANGUAGE RankNTypes #-}
6+{-# LANGUAGE TypeApplications #-}
17-- | This is a log scribe that writes logs to logz.io's bulk
28-- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS HTTPS
39-- API>.
@@ -7,25 +13,60 @@ module Katip.Scribes.LogzIO.HTTPS
713 -- * Types
814 , BulkAPIError(..)
915 , 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 (..)
1037 ) where
1138
1239
1340-------------------------------------------------------------------------------
1441import 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
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
2154import 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
28−-------------------------------------------------------------------------------
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
2970-------------------------------------------------------------------------------
3071
3172
@@ -47,7 +88,21 @@ data BulkAPIError = BulkAPIError
4788 -- concept of a truly empty log line in katip, so this most likely
4889 -- indicates a __library bug__ and should be reported as an issue on
4990 -- GitHub.
50− }
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+ }
51106
52107
53108-------------------------------------------------------------------------------
@@ -59,10 +114,94 @@ data LogzIOScribeConfiguration = LogzIOScribeConfiguration
59114 , logzIOScribeConfiguration_bufferTimeout :: Time.NominalDiffTime
60115 -- ^ Will flush the buffer if it has been this long since the last
61116 -- flush __or__ 'logzIOScribeConfiguration_bufferItems' items are
62− -- accumulated, whichever is first.
63− --TODO: more
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 ()
64127 }
65128
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+
66205-------------------------------------------------------------------------------
67206mkLogzIOScribe
68207 :: LogzIOScribeConfiguration
@@ -70,112 +209,113 @@ mkLogzIOScribe
70209 -> K.Verbosity
71210 -> IO K.Scribe
72211mkLogzIOScribe config permitItem verbosity = do
73− -- Accumulating buffer for each item
74− bufferState <- STM.newTVarIO (mempty :: BulkBuffer)
75− -- Single-slot TMVar when something is ready to write. Katip buffers
76− -- for us so we can simplify shutdown and flush guarantees this way
77− readyMVar <- STM.newEmptyTMVarIO :: IO (STM.TMVar BulkBuffer)
78−
79− -- We create a TVar that is set to ShutDown when its time to stop
80− killSwitch <- STM.newTVarIO Continue
81−
82− -- Create a mutable ref to whether the timer has expired. This lets
83− -- us replace the timer when its time to start timing again. We
84− -- 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
99− STM.writeTVar bufferState newBuffer
100− STM.putTMVar readyMVar toFlush
101−
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
105− processLogAction (bufferItem itemBufferSize verbosity logItem curBuffer)
106−
107−
108− -- A transaction that retries until its time to shutdown
109− let waitShutdown = do
110− killStatus <- STM.readTVar killSwitch
111− STM.check (killStatus == ShutDown)
112−
113− --TODO: exception handling
114−
115− -- Retry transaction until timer has expired
116− let waitTimer = do
117− -- get the timer we'll be using
118− timer <- STM.readTVar timerRef
119− -- do not proceed until its ready
120− timerExpired <- STM.readTVar timer
121− STM.check timerExpired
122− -- do not wake again until this message is out the door
123− STM.writeTVar timer False
124− -- we waited long enough, grab the current buffer
125− curBuffer <- STM.readTVar bufferState
126− -- and tell it to flush right now
127− processLogAction (forceFlush curBuffer)
128−
129− -- loop until its time to shut down
130− let timedFlushLoop = do
131− res <- STM.atomically $
132− raceAlt
133− waitShutdown
134− waitTimer
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
135267 case res of
136− Left () -> pure () -- exit loop
137− Right () -> timedFlushLoop -- wait for another timer or shutdown
138−
139− timedFlusher <- Async.async timedFlushLoop
140−
141− let flushLoop = do
142− -- 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
143− readyBufferOrDie <- STM.atomically $
144− raceAlt
145− (STM.takeTMVar readyMVar)
146− waitShutdown
147− case readyBufferOrDie of
148− Right () -> pure () -- we caught shutdown before any new data arrived
149− -- There's some logs to write
150− Left readyBuffer -> do
151− -- flush in IO
152− flushBuffer readyBuffer
153− -- reset the timeout before the next flush on each write
154− newTimer <- STM.registerDelay itemBufferTimeoutMicros
155− STM.atomically $ do
156− STM.writeTVar timerRef newTimer
157− -- start over
158− flushLoop
159− flusher <- Async.async flushLoop
160−
161− let shutdown = do
162− -- tell workers this is the last loop
163− STM.atomically (STM.writeTVar killSwitch ShutDown)
164− -- wait for workers to finish
165− _ <- Async.waitCatch timedFlusher
166− _ <- Async.waitCatch flusher
167− -- stop blocking
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
168294 pure ()
169295
170−
171296 pure $ K.Scribe
172− { liPush = STM.atomically . push
173− , scribeFinalizer = shutdown
174− , scribePermitItem = permitItem
297+ { K.liPush = STM.atomically . push . AnyLogItem
298+ , K.scribeFinalizer = close
299+ , K.scribePermitItem = permitItem
175300 }
176301 where
177302 itemBufferSize = logzIOScribeConfiguration_bufferItems config
178303 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
179319
180320
181321-------------------------------------------------------------------------------
@@ -196,26 +336,82 @@ ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
196336
197337
198338-------------------------------------------------------------------------------
199−data KillSwitch =
200− Continue
201− | ShutDown
202− deriving (Eq)
203−
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)
204354
205355-------------------------------------------------------------------------------
206−raceAlt :: (Alternative m) => m a -> m b -> m (Either a b)
207−raceAlt a b = (Left <$> a) <|> (Right <$> b)
208−
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+ }
209409
210−-------------------------------------------------------------------------------
211−-- TODO: noop on empty
212−flushBuffer :: BulkBuffer -> IO ()
213−flushBuffer = undefined
214410
215411-------------------------------------------------------------------------------
216412newtype Bytes = Bytes
217413 { bytes :: Int64
218− } deriving (Show, Eq, Num, Ord)
414+ } deriving (Show, Eq, Num, Ord, Bounded)
219415
220416
221417-- | How big of a body can we send? The limit is defined
@@ -231,10 +427,10 @@ maxLogLineLength :: Bytes
231427maxLogLineLength = 500000
232428
233429
234−measureJSON :: A.ToJSON a => a -> (BB.Builder, Bytes)
235−measureJSON a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
430+measureJSONLine :: A.ToJSON a => a -> (BB.Builder, Bytes)
431+measureJSONLine a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
236432 where
237− lbs = A.encode a
433+ lbs = A.encode a <> "\n"
238434
239435
240436-- | Fully-rendered JSON object for an item
@@ -246,35 +442,51 @@ fullItemObject verbosity item = HM.fromList
246442 , "thread" A..= K.getThreadIdText (K._itemThread item)
247443 , "host" A..= K._itemHost item
248444 , "pid" A..= pidInt
249− , "data" A..= K.payloadObject verbosity (K._itemPayload item)
445+ , "data" A..= annotateKeys (K.payloadObject verbosity (K._itemPayload item))
250446 -- Slight deviation, logz.io uses "message" instead of "msg"
251447 , "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?
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)))
258456 , "ns" A..= K._itemNamespace item
259457 , "loc" A..= (LocJs <$> K._itemLoc item)
260458 ]
261459 where
262460 POSIX.CPid pidInt = K._itemProcess item
263461
264−--TODO: property test
265462
266463-- | A version of 'renderLine' which renders a line and stays under
267464-- the maximum line size of 500,000 bytes. If the default rendering is
268465-- too large, the log will be reduced to a timestamp and a potentially
269466-- truncated message.
270467renderLineTruncated :: K.LogItem a => K.Verbosity -> K.Item a -> (BB.Builder, Bytes)
271−renderLineTruncated verbosity item =
272− if fullSize <= maxLogLineLength
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
273485 then (fullLine, fullSize)
274486 else (fallbackLine, fallbackSize)
275487 where
276488 fullObject = fullItemObject verbosity item
277− (fullLine, fullSize) = measureJSON fullObject
489+ (fullLine, fullSize) = measureJSONLine fullObject
278490 -- only the absolutely necessary keys, with message stripped out
279491 -- to help us calculate how much of the message we can keep. these
280492 -- are lazily evaluated and as such won't be computed unless the
@@ -284,9 +496,9 @@ renderLineTruncated verbosity item =
284496 [ "message" A..= A.String "" -- we'll start with a blank message
285497 , "@timestamp" A..= K._itemTime item
286498 ]
287− (_, blankObjectSize) = measureJSON blankObject
499+ (_, blankObjectSize) = measureJSONLine blankObject
288500 messageBytesAllowed = maxLogLineLength - blankObjectSize
289− (fallbackLine, fallbackSize) = measureJSON fallbackObject
501+ (fallbackLine, fallbackSize) = measureJSONLine fallbackObject
290502 fallbackObject :: A.Object
291503 fallbackObject = HM.fromList
292504 [ "message" A..= A.toJSON (TL.take (bytes messageBytesAllowed) (TB.toLazyText (K.unLogStr (K._itemMessage item))))
@@ -301,7 +513,7 @@ data BulkBuffer = BulkBuffer
301513 }
302514
303515
304−instance Semigroup BulkBuffer where
516+instance Semigroup.Semigroup BulkBuffer where
305517 (BulkBuffer bytesUsedA bufferA itemCountA) <>
306518 (BulkBuffer bytesUsedB bufferB itemCountB) = BulkBuffer
307519 (bytesUsedA + bytesUsedB)
@@ -332,11 +544,30 @@ bufferItem
332544 -> K.Item a
333545 -> BulkBuffer
334546 -> LogAction BulkBuffer
335−bufferItem maxItems verb item bulkBuffer =
336− let (encodedLine, spaceNeeded) = renderLineTruncated verb item
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
337568 newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded
338569 newItemCount = bulkBuffer_itemCount bulkBuffer + 1
339− in if newItemCount >= maxItems || newBytesUsed >= maxPayloadBytes
570+ in if newItemCount >= maxItems || newBytesUsed >= customMaxPayload
340571 then FlushNow
341572 bulkBuffer
342573 BulkBuffer
@@ -356,3 +587,53 @@ bufferItem maxItems verb item bulkBuffer =
356587-- will always return 'FlushNow' with an empty new buffer.
357588forceFlush :: (Monoid buf) => buf -> LogAction buf
358589forceFlush 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"