@@ -0,0 +1,639 @@
1+{-# LANGUAGE BangPatterns #-}
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
10+module Katip.Scribes.LogzIO.HTTPS
11+ ( -- * Scribe construction
15+ , LogzIOScribeConfiguration(..)
19+ -- ** Presets for configuration
25+ , defaultLogzIOScribeConfiguration
26+ -- * Internal API exported for testing
27+ , renderLineTruncated
28+ , renderLineTruncated'
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
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
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+-------------------------------------------------------------------------------
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
91+ } deriving (Show, Eq)
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"
101+ { bulkAPIError_malformedLines = malformedLines
102+ , bulkAPIError_successfulLines = successfulLines
103+ , bulkAPIError_oversizedLines = oversizedLines
104+ , bulkAPIError_emptyLogLines = emptyLogLines
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 =
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 ()
130+-- | A default configuration:
132+-- * 100 item buffering
134+-- * 30 second buffer timeout
136+-- * US, HTTPS logging endpoint (listener.logz.io:8071)
138+-- * 'defaultRetryPolicy' of 25ms exponential backoff, 5 retries
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 ())
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
160+newtype APIToken = APIToken
161+ { apiToken :: T.Text
162+ } deriving (Show, Eq, IsString)
165+-- | This particular bulk API only supports HTTP or HTTPS. HTTPS is
166+-- strongly recommended for security reasons.
169+ -- ^ HTTPs should always be used except for local testing.
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,
178+usRegionHost :: URIBS.Host
179+usRegionHost = URIBS.Host "listener.logz.io"
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,
186+euRegionHost :: URIBS.Host
187+euRegionHost = URIBS.Host "listener-eu.logz.io"
190+-- | Logz.io uses port 8071 for HTTPS
191+httpsPort :: URIBS.Port
192+httpsPort = URIBS.Port 8071
195+-- | Logz.io uses port 8070 for HTTP
196+httpPort :: URIBS.Port
197+httpPort = URIBS.Port 8070
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
205+-------------------------------------------------------------------------------
207+ :: LogzIOScribeConfiguration
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)
221+ let newTimer = STM.registerDelay itemBufferTimeoutMicros
222+ timerRef <- STM.newTVarIO =<< newTimer
224+ -- Set up a connection manager for requests
225+ mgr <- case scheme of
226+ HTTPS -> HTTPS.newTlsManager
227+ HTTP -> HTTP.newManager HTTP.defaultManagerSettings
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
241+ let timeExpired :: STM.STM (Tick a)
243+ isExpired <- STM.readTVar =<< STM.readTVar timerRef
247+ -- Circular transaction that checks for completion of work, time
248+ -- expiration, or new events.
249+ let nextTick :: STM.STM (Tick AnyLogItem)
253+ (WorkExhausted <$ waitWorkExhausted)
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
259+ let sealQueue = STM.atomically (STM.closeTBMQueue ingestionQueue)
261+ -- Replace the timer with a new one
262+ let resetTimer = STM.atomically . STM.writeTVar timerRef =<< newTimer
264+ -- Send the buffer and then reset the timer
265+ let flush curBuffer = do
266+ res <- flushBuffer config mgr curBuffer
268+ Left e -> onErrorSafe e
272+ let flushLoop :: BulkBuffer -> IO ()
273+ flushLoop curBuffer = do
274+ tick <- STM.atomically nextTick
277+ flush curBuffer -- flush what you've got
278+ pure () -- stop looping
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
289+ flushThread <- Async.async (flushLoop mempty)
293+ _ <- Async.waitCatch flushThread
297+ { K.liPush = STM.atomically . push . AnyLogItem
298+ , K.scribeFinalizer = close
299+ , K.scribePermitItem = permitItem
302+ itemBufferSize = logzIOScribeConfiguration_bufferItems config
303+ itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)
304+ scheme = logzIOScribeConfiguration_scheme config
306+ _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex)
310+data AnyLogItem where
311+ AnyLogItem :: K.LogItem a => K.Item a -> AnyLogItem
314+-------------------------------------------------------------------------------
321+-------------------------------------------------------------------------------
322+-- Match the native resolution of NominalDiffTime, which is excessive
324+ndtPicos :: Time.NominalDiffTime -> Int64
325+ndtPicos = round . (* picos)
327+ picos :: Time.NominalDiffTime
328+ picos = 10 ^ (9 :: Int)
331+-------------------------------------------------------------------------------
332+ndtToMicros :: Time.NominalDiffTime -> Int
333+ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)
335+ picosInMicro = 10 ^ (3 :: Int)
338+-------------------------------------------------------------------------------
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.
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.
355+-------------------------------------------------------------------------------
357+ :: LogzIOScribeConfiguration
360+ -> IO (Either LoggingError ())
361+flushBuffer config mgr bulkBuffer
362+ | bulkBuffer_itemCount bulkBuffer <= 0 = pure (Right ())
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
373+ else case A.decode @BulkAPIError respLBS of
375+ | HTypes.statusCode respStatus == 401 -> E.throwE BadToken
376+ | otherwise -> E.throwE (UnknownFailureResponse respStatus respLBS)
377+ Just bulkError -> E.throwE (PartialFailure bulkError)
379+ configureRequest req = req
380+ { HTTP.method = HTypes.methodPost
381+ , HTTP.requestBody = HTTP.RequestBodyLBS (BB.toLazyByteString (bulkBuffer_payload bulkBuffer))
384+ retryPolicy = logzIOScribeConfiguration_retry config
385+ apiTokenBS = TE.encodeUtf8 (apiToken (logzIOScribeConfiguration_token config))
387+ handleHttpException :: (Applicative m) => HTTP.HttpException -> m Bool
388+ handleHttpException _ = pure True
390+ uriStr = LBS8.unpack (BB.toLazyByteString (URIBS.serializeURIRef uri))
392+ authority = URIBS.Authority
393+ { URIBS.authorityUserInfo = Nothing
394+ , URIBS.authorityHost = logzIOScribeConfiguration_host config
395+ , URIBS.authorityPort = Just (logzIOScribeConfiguration_port config)
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)
407+ , URIBS.uriFragment = Nothing
411+-------------------------------------------------------------------------------
412+newtype Bytes = Bytes
414+ } deriving (Show, Eq, Num, Ord, Bounded)
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
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
430+measureJSONLine :: A.ToJSON a => a -> (BB.Builder, Bytes)
431+measureJSONLine a = (BB.lazyByteString lbs, Bytes (LBS.length lbs))
433+ lbs = A.encode a <> "\n"
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
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)
460+ POSIX.CPid pidInt = K._itemProcess item
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
471+-- | A generalized renderLineTruncated that takes a custom line length
472+-- limit. This is exclusively for testing.
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
482+ -> (BB.Builder, Bytes)
483+renderLineTruncated' customMaxLogLineLength verbosity item =
484+ if fullSize <= customMaxLogLineLength
485+ then (fullLine, fullSize)
486+ else (fallbackLine, fallbackSize)
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
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
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)
509+data BulkBuffer = BulkBuffer
510+ { bulkBuffer_bytesUsed :: !Bytes
511+ , bulkBuffer_payload :: !BB.Builder
512+ , bulkBuffer_itemCount :: !Int
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)
524+instance Monoid BulkBuffer where
525+ mempty = BulkBuffer 0 mempty 0
535+ -- ^ Buffer to flush
542+ -- ^ Maximum items before flushing
546+ -> LogAction BulkBuffer
547+bufferItem = bufferItem' maxPayloadBytes maxLogLineLength
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.
557+ -- ^ Max payload size
561+ -- ^ Maximum items before flushing
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
574+ { bulkBuffer_bytesUsed = spaceNeeded
575+ , bulkBuffer_payload = encodedLine
576+ , bulkBuffer_itemCount = 1
580+ { bulkBuffer_bytesUsed = newBytesUsed
581+ , bulkBuffer_payload = bulkBuffer_payload bulkBuffer <> encodedLine
582+ , bulkBuffer_itemCount = newItemCount
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
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
602+annotateValue :: A.Value -> A.Value
603+annotateValue (A.Object o) = A.Object (annotateKeys o)
604+annotateValue (A.Array a) = A.Array (annotateValue <$> a)
608+annotateKeys :: A.Object -> A.Object
609+annotateKeys = HM.fromList . map go . HM.toList
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)
621+-------------------------------------------------------------------------------
622+-- Annotation Constants
623+-------------------------------------------------------------------------------