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