katip-logzio 0.1.1.0 → 0.1.2.0
raw patch · 5 files changed
+630/−604 lines, 5 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- changelog.md +4/−0
- katip-logzio.cabal +1/−1
- src/Katip/Scribes/LogzIO/HTTPS.hs +331/−308
- test/Katip/Tests/Scribes/LogzIO/HTTPS.hs +283/−285
- test/Main.hs +11/−10
changelog.md view
@@ -1,3 +1,7 @@+0.1.2.0+=======+* Add aeson 2 support [#146](https://github.com/Soostone/katip/pull/146)+ 0.1.1.0 ======= * Fixed bug where on scribe close, any items left in the buffer that
katip-logzio.cabal view
@@ -1,5 +1,5 @@ name: katip-logzio-version: 0.1.1.0+version: 0.1.2.0 synopsis: Logz.IO scribe for the Katip logging framework description: See README.md for more details. license: BSD3
src/Katip/Scribes/LogzIO/HTTPS.hs view
@@ -1,132 +1,139 @@-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE GADTs #-}+{-# LANGUAGE GADTs #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE CPP #-}+ -- | This is a log scribe that writes logs to logz.io's bulk -- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS HTTPS -- API>. module Katip.Scribes.LogzIO.HTTPS- ( -- * Scribe construction- mkLogzIOScribe- -- * Types- , BulkAPIError(..)- , LogzIOScribeConfiguration(..)- , Scheme(..)- , APIToken(..)- , LoggingError(..)+ ( -- * Scribe construction+ mkLogzIOScribe,++ -- * Types+ BulkAPIError (..),+ LogzIOScribeConfiguration (..),+ Scheme (..),+ APIToken (..),+ LoggingError (..),+ -- ** Presets for configuration- , usRegionHost- , euRegionHost- , httpsPort- , httpPort- , defaultRetryPolicy- , defaultLogzIOScribeConfiguration- -- * Internal API exported for testing- , renderLineTruncated- , renderLineTruncated'- , maxPayloadBytes- , maxLogLineLength- , BulkBuffer (..)- , LogAction (..)- , bufferItem- , bufferItem'- , forceFlush- , Bytes (..)- ) where+ usRegionHost,+ euRegionHost,+ httpsPort,+ httpPort,+ defaultRetryPolicy,+ defaultLogzIOScribeConfiguration, + -- * Internal API exported for testing+ renderLineTruncated,+ renderLineTruncated',+ maxPayloadBytes,+ maxLogLineLength,+ BulkBuffer (..),+ LogAction (..),+ bufferItem,+ bufferItem',+ forceFlush,+ Bytes (..),+ )+where --------------------------------------------------------------------------------import Control.Applicative-import qualified Control.Concurrent.Async as Async-import qualified Control.Concurrent.STM as STM+import Control.Applicative+import qualified Control.Concurrent.Async as Async+import qualified Control.Concurrent.STM as STM import qualified Control.Concurrent.STM.TBMQueue as STM-import qualified Control.Error as E-import qualified Control.Exception.Safe as EX-import Control.Monad-import qualified Control.Retry as Retry-import qualified Data.Aeson as A-import qualified Data.ByteString.Builder as BB-import qualified Data.ByteString.Lazy as LBS-import qualified Data.ByteString.Lazy.Char8 as LBS8-import qualified Data.HashMap.Strict as HM-import Data.Int-import qualified Data.Scientific as Scientific-import Data.Semigroup as Semigroup-import Data.String (IsString)-import qualified Data.Text as T-import qualified Data.Text.Encoding as TE-import qualified Data.Text.Lazy as TL-import qualified Data.Text.Lazy.Builder as TB-import qualified Data.Time as Time-import qualified Katip as K-import Katip.Core (LocJs (..))-import qualified Network.HTTP.Client as HTTP-import qualified Network.HTTP.Client.TLS as HTTPS-import qualified Network.HTTP.Types as HTypes-import qualified System.Posix.Types as POSIX-import qualified URI.ByteString as URIBS---------------------------------------------------------------------------------+import qualified Control.Error as E+import qualified Control.Exception.Safe as EX+import Control.Monad+import qualified Control.Retry as Retry+import qualified Data.Aeson as A+import qualified Data.ByteString.Builder as BB+import qualified Data.ByteString.Lazy as LBS+import qualified Data.ByteString.Lazy.Char8 as LBS8+#if MIN_VERSION_aeson (2, 0, 0)+import qualified Data.Aeson.Key as A+import qualified Data.Aeson.KeyMap as A+#else+import qualified Data.HashMap.Strict as HM+#endif+import Data.Int+import qualified Data.Scientific as Scientific+import Data.Semigroup as Semigroup+import Data.String (IsString)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import qualified Data.Text.Lazy as TL+import qualified Data.Text.Lazy.Builder as TB+import qualified Data.Time as Time+import qualified Katip as K+import Katip.Core (LocJs (..))+import qualified Network.HTTP.Client as HTTP+import qualified Network.HTTP.Client.TLS as HTTPS+import qualified Network.HTTP.Types as HTypes+import qualified System.Posix.Types as POSIX+import qualified URI.ByteString as URIBS +------------------------------------------------------------------------------- -- | This is returned when the bulk import was a partial success data BulkAPIError = BulkAPIError- { bulkAPIError_malformedLines :: Int- -- ^ The number of log lines which are not well-formed JSON. This- -- indicates a __library bug__. Please file an issue on GitHub.- , bulkAPIError_successfulLines :: Int- -- ^ The number of log lines received successfully.- , bulkAPIError_oversizedLines :: Int- -- ^ The number of log lines which exceed the line length- -- limit. katip-logzio makes a best effort to truncate logs that- -- exceed the size limit. This probably indicates a __library bug__- -- and should be reported as an issue on GitHub.- , bulkAPIError_emptyLogLines :: Int- -- ^ The number of log lines which were empty. There isnt' really a- -- concept of a truly empty log line in katip, so this most likely- -- indicates a __library bug__ and should be reported as an issue on- -- GitHub.- } deriving (Show, Eq)-+ { -- | The number of log lines which are not well-formed JSON. This+ -- indicates a __library bug__. Please file an issue on GitHub.+ bulkAPIError_malformedLines :: Int,+ -- | The number of log lines received successfully.+ bulkAPIError_successfulLines :: Int,+ -- | The number of log lines which exceed the line length+ -- limit. katip-logzio makes a best effort to truncate logs that+ -- exceed the size limit. This probably indicates a __library bug__+ -- and should be reported as an issue on GitHub.+ bulkAPIError_oversizedLines :: Int,+ -- | The number of log lines which were empty. There isnt' really a+ -- concept of a truly empty log line in katip, so this most likely+ -- indicates a __library bug__ and should be reported as an issue on+ -- GitHub.+ bulkAPIError_emptyLogLines :: Int+ }+ deriving (Show, Eq) instance A.FromJSON BulkAPIError where parseJSON = A.withObject "BulkAPIError" $ \o -> do- malformedLines <- o A..: "malformedLines"+ malformedLines <- o A..: "malformedLines" successfulLines <- o A..: "successfulLines"- oversizedLines <- o A..: "oversizedLines"- emptyLogLines <- o A..: "emptyLogLines"- pure $ BulkAPIError- { bulkAPIError_malformedLines = malformedLines- , bulkAPIError_successfulLines = successfulLines- , bulkAPIError_oversizedLines = oversizedLines- , bulkAPIError_emptyLogLines = emptyLogLines- }-+ oversizedLines <- o A..: "oversizedLines"+ emptyLogLines <- o A..: "emptyLogLines"+ pure $+ BulkAPIError+ { bulkAPIError_malformedLines = malformedLines,+ bulkAPIError_successfulLines = successfulLines,+ bulkAPIError_oversizedLines = oversizedLines,+ bulkAPIError_emptyLogLines = emptyLogLines+ } ------------------------------------------------------------------------------- data LogzIOScribeConfiguration = LogzIOScribeConfiguration- { logzIOScribeConfiguration_bufferItems :: Int- -- ^ Will flush the log buffer if this many items is in the buffer- -- __or__ 'logzIOScribeConfiguration_bufferTimeout' is reached,- -- whichever is first- , logzIOScribeConfiguration_bufferTimeout :: Time.NominalDiffTime- -- ^ Will flush the buffer if it has been this long since the last- -- flush __or__ 'logzIOScribeConfiguration_bufferItems' items are- -- accumulated, whichever is first. NominalDiffTime has a Num- -- instance, so you can use a literal to specify seconds, e.g. 30 =- -- 30s.- , logzIOScribeConfiguration_scheme :: Scheme- , logzIOScribeConfiguration_host :: URIBS.Host- , logzIOScribeConfiguration_port :: URIBS.Port- , logzIOScribeConfiguration_token :: APIToken- , logzIOScribeConfiguration_retry :: Retry.RetryPolicyM IO- -- ^ How should exceptions during writes be retried?- , logzIOScribeConfiguration_onError :: LoggingError -> IO ()+ { -- | Will flush the log buffer if this many items is in the buffer+ -- __or__ 'logzIOScribeConfiguration_bufferTimeout' is reached,+ -- whichever is first+ logzIOScribeConfiguration_bufferItems :: Int,+ -- | Will flush the buffer if it has been this long since the last+ -- flush __or__ 'logzIOScribeConfiguration_bufferItems' items are+ -- accumulated, whichever is first. NominalDiffTime has a Num+ -- instance, so you can use a literal to specify seconds, e.g. 30 =+ -- 30s.+ logzIOScribeConfiguration_bufferTimeout :: Time.NominalDiffTime,+ logzIOScribeConfiguration_scheme :: Scheme,+ logzIOScribeConfiguration_host :: URIBS.Host,+ logzIOScribeConfiguration_port :: URIBS.Port,+ logzIOScribeConfiguration_token :: APIToken,+ -- | How should exceptions during writes be retried?+ logzIOScribeConfiguration_retry :: Retry.RetryPolicyM IO,+ logzIOScribeConfiguration_onError :: LoggingError -> IO () } - -- | A default configuration: -- -- * 100 item buffering@@ -139,19 +146,20 @@ -- -- * Ignore logging errors defaultLogzIOScribeConfiguration :: APIToken -> LogzIOScribeConfiguration-defaultLogzIOScribeConfiguration token = LogzIOScribeConfiguration- { logzIOScribeConfiguration_bufferItems = 100- , logzIOScribeConfiguration_bufferTimeout = 30- , logzIOScribeConfiguration_scheme = HTTPS- , logzIOScribeConfiguration_host = usRegionHost- , logzIOScribeConfiguration_port = httpsPort- , logzIOScribeConfiguration_token = token- , logzIOScribeConfiguration_retry = defaultRetryPolicy- , logzIOScribeConfiguration_onError = const (pure ())- }-+defaultLogzIOScribeConfiguration token =+ LogzIOScribeConfiguration+ { logzIOScribeConfiguration_bufferItems = 100,+ logzIOScribeConfiguration_bufferTimeout = 30,+ logzIOScribeConfiguration_scheme = HTTPS,+ logzIOScribeConfiguration_host = usRegionHost,+ logzIOScribeConfiguration_port = httpsPort,+ logzIOScribeConfiguration_token = token,+ logzIOScribeConfiguration_retry = defaultRetryPolicy,+ logzIOScribeConfiguration_onError = const (pure ())+ } -------------------------------------------------------------------------------+ -- | You can retrieve your account or sub-account's API token on the -- <https://app.logz.io/#/dashboard/settings/manage-accounts manage -- accounts page>. Note that APIToken has an IsString instance,@@ -159,18 +167,17 @@ -- enabled. newtype APIToken = APIToken { apiToken :: T.Text- } deriving (Show, Eq, IsString)-+ }+ deriving (Show, Eq, IsString) -- | This particular bulk API only supports HTTP or HTTPS. HTTPS is -- strongly recommended for security reasons. data Scheme- = HTTPS- -- ^ HTTPs should always be used except for local testing.+ = -- | HTTPs should always be used except for local testing.+ HTTPS | HTTP deriving (Show, Eq) - -- | See -- <https://support.logz.io/hc/en-us/articles/210206365-What-IP-addresses-should-I-open-in-my-firewall-to-ship-logs-to-Logz-io- -- this> for a list of listeners. This is the US region host,@@ -178,7 +185,6 @@ usRegionHost :: URIBS.Host usRegionHost = URIBS.Host "listener.logz.io" - -- | See -- <https://support.logz.io/hc/en-us/articles/210206365-What-IP-addresses-should-I-open-in-my-firewall-to-ship-logs-to-Logz-io- -- this> for a list of listeners. This is the EU region host,@@ -186,28 +192,25 @@ euRegionHost :: URIBS.Host euRegionHost = URIBS.Host "listener-eu.logz.io" - -- | Logz.io uses port 8071 for HTTPS httpsPort :: URIBS.Port httpsPort = URIBS.Port 8071 - -- | Logz.io uses port 8070 for HTTP httpPort :: URIBS.Port httpPort = URIBS.Port 8070 - -- | A reasonable retry policy: exponential backoff with 25ms base -- delay up to 5 retries, for a total cumulative delay of 775ms. defaultRetryPolicy :: (Monad m) => Retry.RetryPolicyM m defaultRetryPolicy = Retry.exponentialBackoff 25000 `mappend` Retry.limitRetries 5 --------------------------------------------------------------------------------mkLogzIOScribe- :: LogzIOScribeConfiguration- -> K.PermitFunc- -> K.Verbosity- -> IO K.Scribe+mkLogzIOScribe ::+ LogzIOScribeConfiguration ->+ K.PermitFunc ->+ K.Verbosity ->+ IO K.Scribe mkLogzIOScribe config permitItem verbosity = do -- This size is actually somewhat arbitrary. We're just making sure -- it isn't infinite so we don't consume the whole backlog right@@ -224,14 +227,15 @@ -- Set up a connection manager for requests mgr <- case scheme of HTTPS -> HTTPS.newTlsManager- HTTP -> HTTP.newManager HTTP.defaultManagerSettings+ HTTP -> HTTP.newManager HTTP.defaultManagerSettings -- An STM transaction that will return true when writes are stopped -- and backlog is emptied. let workExhausted :: STM.STM Bool- workExhausted = (&&)- <$> STM.isClosedTBMQueue ingestionQueue- <*> STM.isEmptyTBMQueue ingestionQueue+ workExhausted =+ (&&)+ <$> STM.isClosedTBMQueue ingestionQueue+ <*> STM.isEmptyTBMQueue ingestionQueue -- Block until there's no more work let waitWorkExhausted :: STM.STM () waitWorkExhausted = STM.check =<< workExhausted@@ -248,9 +252,9 @@ -- expiration, or new events. let nextTick :: STM.STM (Tick AnyLogItem) nextTick =- timeExpired <|>- pop <|>- (WorkExhausted <$ waitWorkExhausted)+ timeExpired+ <|> pop+ <|> (WorkExhausted <$ waitWorkExhausted) -- Blocking push. This applies backpressure upstream to katip, which does its own buffering let push :: AnyLogItem -> STM.STM ()@@ -265,7 +269,7 @@ let flush curBuffer = do res <- flushBuffer config mgr curBuffer case res of- Left e -> onErrorSafe e+ Left e -> onErrorSafe e Right () -> pure () resetTimer @@ -293,11 +297,12 @@ _ <- Async.waitCatch flushThread pure () - pure $ K.Scribe- { K.liPush = STM.atomically . push . AnyLogItem- , K.scribeFinalizer = close- , K.scribePermitItem = permitItem- }+ pure $+ K.Scribe+ { K.liPush = STM.atomically . push . AnyLogItem,+ K.scribeFinalizer = close,+ K.scribePermitItem = permitItem+ } where itemBufferSize = logzIOScribeConfiguration_bufferItems config itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config)@@ -306,18 +311,15 @@ _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex) pure () - data AnyLogItem where AnyLogItem :: K.LogItem a => K.Item a -> AnyLogItem - --------------------------------------------------------------------------------data Tick a =- TimeExpired+data Tick a+ = TimeExpired | NewItem !a | WorkExhausted - ------------------------------------------------------------------------------- -- Match the native resolution of NominalDiffTime, which is excessive -- for our needs@@ -327,59 +329,60 @@ picos :: Time.NominalDiffTime picos = 10 ^ (9 :: Int) - ------------------------------------------------------------------------------- ndtToMicros :: Time.NominalDiffTime -> Int-ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro)+ndtToMicros t = round ((fromIntegral (ndtPicos t) :: Double) / picosInMicro) where picosInMicro = 10 ^ (3 :: Int) - --------------------------------------------------------------------------------data LoggingError =+data LoggingError+ = -- | The URI generated was invalid. Check your configuration URIError HTTP.HttpException- -- ^ The URI generated was invalid. Check your configuration- | RequestError HTTP.HttpException- -- ^ We encountered an exception while sending the batch request- | PartialFailure BulkAPIError- -- ^ Some or all of the request was rejected. Check the logz.io UI- -- for indexing errors.- | BadToken- -- ^ Your API token was rejected.- | UnknownFailureResponse HTypes.Status LBS.ByteString- -- ^ An error returned, but it could not be decoded into a- -- 'BulkAPIError'. This may indicate a __library bug__, which should- -- be reported to the issue tracker.+ | -- | We encountered an exception while sending the batch request+ RequestError HTTP.HttpException+ | -- | Some or all of the request was rejected. Check the logz.io UI+ -- for indexing errors.+ PartialFailure BulkAPIError+ | -- | Your API token was rejected.+ BadToken+ | -- | An error returned, but it could not be decoded into a+ -- 'BulkAPIError'. This may indicate a __library bug__, which should+ -- be reported to the issue tracker.+ UnknownFailureResponse HTypes.Status LBS.ByteString deriving (Show) --------------------------------------------------------------------------------flushBuffer- :: LogzIOScribeConfiguration- -> HTTP.Manager- -> BulkBuffer- -> IO (Either LoggingError ())+flushBuffer ::+ LogzIOScribeConfiguration ->+ HTTP.Manager ->+ BulkBuffer ->+ IO (Either LoggingError ()) flushBuffer config mgr bulkBuffer | bulkBuffer_itemCount bulkBuffer <= 0 = pure (Right ()) | otherwise = do- E.runExceptT $ do- req <- E.fmapLT URIError (E.ExceptT (EX.try (configureRequest <$> HTTP.parseRequest uriStr)))- resp <- E.fmapLT RequestError $ E.ExceptT $ EX.try $- Retry.recovering retryPolicy [\_stat -> EX.Handler handleHttpException] $ \_stat ->- HTTP.httpLbs req mgr- let respLBS = HTTP.responseBody resp- let respStatus = HTTP.responseStatus resp- if HTypes.statusIsSuccessful respStatus- then pure ()- else case A.decode @BulkAPIError respLBS of- Nothing- | HTypes.statusCode respStatus == 401 -> E.throwE BadToken- | otherwise -> E.throwE (UnknownFailureResponse respStatus respLBS)- Just bulkError -> E.throwE (PartialFailure bulkError)+ E.runExceptT $ do+ req <- E.fmapLT URIError (E.ExceptT (EX.try (configureRequest <$> HTTP.parseRequest uriStr)))+ resp <- E.fmapLT RequestError $+ E.ExceptT $+ EX.try $+ Retry.recovering retryPolicy [\_stat -> EX.Handler handleHttpException] $ \_stat ->+ HTTP.httpLbs req mgr+ let respLBS = HTTP.responseBody resp+ let respStatus = HTTP.responseStatus resp+ if HTypes.statusIsSuccessful respStatus+ then pure ()+ else case A.decode @BulkAPIError respLBS of+ Nothing+ | HTypes.statusCode respStatus == 401 -> E.throwE BadToken+ | otherwise -> E.throwE (UnknownFailureResponse respStatus respLBS)+ Just bulkError -> E.throwE (PartialFailure bulkError) where- configureRequest req = req- { HTTP.method = HTypes.methodPost- , HTTP.requestBody = HTTP.RequestBodyLBS (BB.toLazyByteString (bulkBuffer_payload bulkBuffer))- }+ configureRequest req =+ req+ { HTTP.method = HTypes.methodPost,+ HTTP.requestBody = HTTP.RequestBodyLBS (BB.toLazyByteString (bulkBuffer_payload bulkBuffer))+ } retryPolicy = logzIOScribeConfiguration_retry config apiTokenBS = TE.encodeUtf8 (apiToken (logzIOScribeConfiguration_token config))@@ -389,77 +392,80 @@ uriStr = LBS8.unpack (BB.toLazyByteString (URIBS.serializeURIRef uri)) - authority = URIBS.Authority- { URIBS.authorityUserInfo = Nothing- , URIBS.authorityHost = logzIOScribeConfiguration_host config- , URIBS.authorityPort = Just (logzIOScribeConfiguration_port config)- }-- uri = URIBS.URI- { URIBS.uriScheme = case logzIOScribeConfiguration_scheme config of- HTTPS -> URIBS.Scheme "https"- HTTP -> URIBS.Scheme "http"- , URIBS.uriAuthority = Just authority- , URIBS.uriPath = "/"- , URIBS.uriQuery = URIBS.Query- [ ("token", apiTokenBS)- ]- , URIBS.uriFragment = Nothing- }+ authority =+ URIBS.Authority+ { URIBS.authorityUserInfo = Nothing,+ URIBS.authorityHost = logzIOScribeConfiguration_host config,+ URIBS.authorityPort = Just (logzIOScribeConfiguration_port config)+ } + uri =+ URIBS.URI+ { URIBS.uriScheme = case logzIOScribeConfiguration_scheme config of+ HTTPS -> URIBS.Scheme "https"+ HTTP -> URIBS.Scheme "http",+ URIBS.uriAuthority = Just authority,+ URIBS.uriPath = "/",+ URIBS.uriQuery =+ URIBS.Query+ [ ("token", apiTokenBS)+ ],+ URIBS.uriFragment = Nothing+ } ------------------------------------------------------------------------------- newtype Bytes = Bytes { bytes :: Int64- } deriving (Show, Eq, Num, Ord, Bounded)-+ }+ deriving (Show, Eq, Num, Ord, Bounded) -- | How big of a body can we send? The limit is defined -- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS here>. maxPayloadBytes :: Bytes maxPayloadBytes = 10485760 - -- | How long can each serialized payload be (let's assume including -- the trailing newline). The limit is defined -- <https://app.logz.io/#/dashboard/data-sources/Bulk-HTTPS here>. maxLogLineLength :: Bytes maxLogLineLength = 500000 - measureJSONLine :: A.ToJSON a => a -> (BB.Builder, Bytes) measureJSONLine a = (BB.lazyByteString lbs, Bytes (LBS.length lbs)) where lbs = A.encode a <> "\n" - -- | Fully-rendered JSON object for an item fullItemObject :: K.LogItem a => K.Verbosity -> K.Item a -> A.Object-fullItemObject verbosity item = HM.fromList- [ "app" A..= K._itemApp item- , "env" A..= K._itemEnv item- , "sev" A..= K._itemSeverity item- , "thread" A..= K.getThreadIdText (K._itemThread item)- , "host" A..= K._itemHost item- , "pid" A..= pidInt- , "data" A..= annotateKeys (K.payloadObject verbosity (K._itemPayload item))- -- Slight deviation, logz.io uses "message" instead of "msg"- , "message" A..= TB.toLazyText (K.unLogStr (K._itemMessage item))- -- Another slight deviation, logz.io uses "@timestamp" instead of- -- "at". Note, your logs should be sent roughly close to when they- -- are created. They are assigned to indexes based on index date, so- -- time searches act strange if you backfill too far from the- -- past. They seem to support 3 decimal places of- -- precision. Formatting like this requires time 1.8.0.2, which is- -- reflected in the cabal file.- , "@timestamp" A..= A.String (T.pack (Time.formatTime Time.defaultTimeLocale "%Y-%m-%dT%H:%M:%S%03QZ" (K._itemTime item)))- , "ns" A..= K._itemNamespace item- , "loc" A..= (LocJs <$> K._itemLoc item)- ]+fullItemObject verbosity item =+#if MIN_VERSION_aeson (2, 0, 0)+ A.fromList+#else+ HM.fromList+#endif+ [ "app" A..= K._itemApp item,+ "env" A..= K._itemEnv item,+ "sev" A..= K._itemSeverity item,+ "thread" A..= K.getThreadIdText (K._itemThread item),+ "host" A..= K._itemHost item,+ "pid" A..= pidInt,+ "data" A..= annotateKeys (K.payloadObject verbosity (K._itemPayload item)),+ -- Slight deviation, logz.io uses "message" instead of "msg"+ "message" A..= TB.toLazyText (K.unLogStr (K._itemMessage item)),+ -- Another slight deviation, logz.io uses "@timestamp" instead of+ -- "at". Note, your logs should be sent roughly close to when they+ -- are created. They are assigned to indexes based on index date, so+ -- time searches act strange if you backfill too far from the+ -- past. They seem to support 3 decimal places of+ -- precision. Formatting like this requires time 1.8.0.2, which is+ -- reflected in the cabal file.+ "@timestamp" A..= A.String (T.pack (Time.formatTime Time.defaultTimeLocale "%Y-%m-%dT%H:%M:%S%03QZ" (K._itemTime item))),+ "ns" A..= K._itemNamespace item,+ "loc" A..= (LocJs <$> K._itemLoc item)+ ] where POSIX.CPid pidInt = K._itemProcess item - -- | A version of 'renderLine' which renders a line and stays under -- the maximum line size of 500,000 bytes. If the default rendering is -- too large, the log will be reduced to a timestamp and a potentially@@ -467,19 +473,18 @@ renderLineTruncated :: K.LogItem a => K.Verbosity -> K.Item a -> (BB.Builder, Bytes) renderLineTruncated = renderLineTruncated' maxLogLineLength - -- | A generalized renderLineTruncated that takes a custom line length -- limit. This is exclusively for testing.-renderLineTruncated'- :: K.LogItem a- => Bytes- -- ^ Custom max log line length. Be careful, too low and no amount+renderLineTruncated' ::+ K.LogItem a =>+ -- | Custom max log line length. Be careful, too low and no amount -- of shrinking can get under the size, breaking the invariant. For -- the production limit, we are guraanteed always able to come in -- under the limit.- -> K.Verbosity- -> K.Item a- -> (BB.Builder, Bytes)+ Bytes ->+ K.Verbosity ->+ K.Item a ->+ (BB.Builder, Bytes) renderLineTruncated' customMaxLogLineLength verbosity item = if fullSize <= customMaxLogLineLength then (fullLine, fullSize)@@ -492,103 +497,109 @@ -- are lazily evaluated and as such won't be computed unless the -- item is too big blankObject :: A.Object- blankObject = HM.fromList- [ "message" A..= A.String "" -- we'll start with a blank message- , "@timestamp" A..= K._itemTime item- ]+ blankObject =+#if MIN_VERSION_aeson (2, 0, 0)+ A.fromList+#else+ HM.fromList+#endif+ [ "message" A..= A.String "", -- we'll start with a blank message+ "@timestamp" A..= K._itemTime item+ ] (_, blankObjectSize) = measureJSONLine blankObject messageBytesAllowed = maxLogLineLength - blankObjectSize (fallbackLine, fallbackSize) = measureJSONLine fallbackObject fallbackObject :: A.Object- fallbackObject = HM.fromList- [ "message" A..= A.toJSON (TL.take (bytes messageBytesAllowed) (TB.toLazyText (K.unLogStr (K._itemMessage item))))- , "@timestamp" A..= A.toJSON (K._itemTime item)- ]-+ fallbackObject =+#if MIN_VERSION_aeson (2, 0, 0)+ A.fromList+#else+ HM.fromList+#endif+ [ "message" A..= A.toJSON (TL.take (bytes messageBytesAllowed) (TB.toLazyText (K.unLogStr (K._itemMessage item)))),+ "@timestamp" A..= A.toJSON (K._itemTime item)+ ] data BulkBuffer = BulkBuffer- { bulkBuffer_bytesUsed :: !Bytes- , bulkBuffer_payload :: !BB.Builder- , bulkBuffer_itemCount :: !Int+ { bulkBuffer_bytesUsed :: !Bytes,+ bulkBuffer_payload :: !BB.Builder,+ bulkBuffer_itemCount :: !Int } - instance Semigroup.Semigroup BulkBuffer where- (BulkBuffer bytesUsedA bufferA itemCountA) <>- (BulkBuffer bytesUsedB bufferB itemCountB) = BulkBuffer- (bytesUsedA + bytesUsedB)- (bufferA <> bufferB)- (itemCountA + itemCountB)-+ (BulkBuffer bytesUsedA bufferA itemCountA)+ <> (BulkBuffer bytesUsedB bufferB itemCountB) =+ BulkBuffer+ (bytesUsedA + bytesUsedB)+ (bufferA <> bufferB)+ (itemCountA + itemCountB) instance Monoid BulkBuffer where mempty = BulkBuffer 0 mempty 0 mappend = (<>) --data LogAction buf =+data LogAction buf+ = -- | New buffer Buffered buf- -- ^ New buffer | FlushNow buf -- ^ Buffer to flush buf -- ^ New buffer -bufferItem- :: K.LogItem a- => Int- -- ^ Maximum items before flushing- -> K.Verbosity- -> K.Item a- -> BulkBuffer- -> LogAction BulkBuffer+bufferItem ::+ K.LogItem a =>+ -- | Maximum items before flushing+ Int ->+ K.Verbosity ->+ K.Item a ->+ BulkBuffer ->+ LogAction BulkBuffer bufferItem = bufferItem' maxPayloadBytes maxLogLineLength - -- | An internal version with a configurable max item size and max -- message size for testing. Be careful with this: if you set the -- values too low, some of the guarantees about reducability of -- messages break down.-bufferItem'- :: (K.LogItem a)- => Bytes- -- ^ Max payload size- -> Bytes- -- ^ Max item size- -> Int- -- ^ Maximum items before flushing- -> K.Verbosity- -> K.Item a- -> BulkBuffer- -> LogAction BulkBuffer+bufferItem' ::+ (K.LogItem a) =>+ -- | Max payload size+ Bytes ->+ -- | Max item size+ Bytes ->+ -- | Maximum items before flushing+ Int ->+ K.Verbosity ->+ K.Item a ->+ BulkBuffer ->+ LogAction BulkBuffer bufferItem' customMaxPayload customMaxItem maxItems verb item bulkBuffer = let (encodedLine, spaceNeeded) = renderLineTruncated' customMaxItem verb item newBytesUsed = bulkBuffer_bytesUsed bulkBuffer + spaceNeeded newItemCount = bulkBuffer_itemCount bulkBuffer + 1- in if newItemCount >= maxItems || newBytesUsed >= customMaxPayload- then FlushNow- bulkBuffer- BulkBuffer- { bulkBuffer_bytesUsed = spaceNeeded- , bulkBuffer_payload = encodedLine- , bulkBuffer_itemCount = 1- }- else Buffered $- BulkBuffer- { bulkBuffer_bytesUsed = newBytesUsed- , bulkBuffer_payload = bulkBuffer_payload bulkBuffer <> encodedLine- , bulkBuffer_itemCount = newItemCount- }-+ in if newItemCount >= maxItems || newBytesUsed >= customMaxPayload+ then+ FlushNow+ bulkBuffer+ BulkBuffer+ { bulkBuffer_bytesUsed = spaceNeeded,+ bulkBuffer_payload = encodedLine,+ bulkBuffer_itemCount = 1+ }+ else+ Buffered $+ BulkBuffer+ { bulkBuffer_bytesUsed = newBytesUsed,+ bulkBuffer_payload = bulkBuffer_payload bulkBuffer <> encodedLine,+ bulkBuffer_itemCount = newItemCount+ } -- | When time has run out on a flush, splits the buffer. n.b. this -- will always return 'FlushNow' with an empty new buffer. forceFlush :: (Monoid buf) => buf -> LogAction buf forceFlush buf = FlushNow buf mempty - ------------------------------------------------------------------------------- -- Annotation borrowed from katip-elasticsearch. There are fixed -- fields in katip logs which should stay the same type forever and@@ -601,27 +612,39 @@ annotateValue :: A.Value -> A.Value annotateValue (A.Object o) = A.Object (annotateKeys o)-annotateValue (A.Array a) = A.Array (annotateValue <$> a)-annotateValue x = x-+annotateValue (A.Array a) = A.Array (annotateValue <$> a)+annotateValue x = x annotateKeys :: A.Object -> A.Object+#if MIN_VERSION_aeson (2, 0, 0)+annotateKeys = A.fromList . map go . A.toList+ where+ go (k, A.Object o) = (k, A.Object (annotateKeys o))+ go (k, A.Array a) = (k, A.Array (annotateValue <$> a))+ go (k, s@(A.String _)) = (A.fromText (A.toText k <> stringAnn), s)+ go (k, n@(A.Number sci)) =+ if Scientific.isFloating sci+ then (A.fromText (A.toText k <> doubleAnn), n)+ else (A.fromText (A.toText k <> longAnn), n)+ go (k, b@(A.Bool _)) = (A.fromText (A.toText k <> booleanAnn), b)+ go (k, A.Null) = (A.fromText (A.toText k <> nullAnn), A.Null)+#else annotateKeys = HM.fromList . map go . HM.toList where go (k, A.Object o) = (k, A.Object (annotateKeys o))- go (k, A.Array a) = (k, A.Array (annotateValue <$> a))+ go (k, A.Array a) = (k, A.Array (annotateValue <$> a)) go (k, s@(A.String _)) = (k <> stringAnn, s)- go (k, n@(A.Number sci)) = if Scientific.isFloating sci- then (k <> doubleAnn, n)- else (k <> longAnn, n)+ go (k, n@(A.Number sci)) =+ if Scientific.isFloating sci+ then (k <> doubleAnn, n)+ else (k <> longAnn, n) go (k, b@(A.Bool _)) = (k <> booleanAnn, b) go (k, A.Null) = (k <> nullAnn, A.Null)-+#endif ------------------------------------------------------------------------------- -- Annotation Constants -------------------------------------------------------------------------------- stringAnn :: T.Text stringAnn = "::s"
test/Katip/Tests/Scribes/LogzIO/HTTPS.hs view
@@ -1,123 +1,127 @@ {-# LANGUAGE OverloadedStrings #-}-module Katip.Tests.Scribes.LogzIO.HTTPS- ( tests- ) where +module Katip.Tests.Scribes.LogzIO.HTTPS+ ( tests,+ )+where --------------------------------------------------------------------------------import Control.Concurrent.Async-import Control.Concurrent.STM-import Control.Concurrent.STM.TSem-import Control.Exception.Safe-import Control.Monad-import Control.Monad.IO.Class-import Data.Aeson as A-import qualified Data.ByteString.Builder as BB-import qualified Data.ByteString.Lazy as LBS-import qualified Data.ByteString.Lazy.Char8 as LBS8-import qualified Data.HashMap.Strict as HM-import Data.Int-import Data.Scientific-import Data.Text (Text)-import Data.Time-import Data.Time.Clock.POSIX-import qualified Data.Vector as V-import Hedgehog as HH-import qualified Hedgehog.Gen as Gen-import qualified Hedgehog.Range as Range-import Katip-import Katip.Core-import Language.Haskell.TH (Loc (..))-import Network.HostName-import qualified Network.HTTP.Types.Status as HTypes-import qualified Network.Wai.Handler.Warp as Warp-import System.Posix-import System.Timeout-import Test.Tasty-import Test.Tasty.Hedgehog-import Test.Tasty.HUnit-import URI.ByteString-import qualified Web.Scotty as Scotty---------------------------------------------------------------------------------import Katip.Scribes.LogzIO.HTTPS+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Concurrent.STM.TSem+import Control.Exception.Safe+import Control.Monad+import Control.Monad.IO.Class+import Data.Aeson as A+import qualified Data.ByteString.Builder as BB+import qualified Data.ByteString.Lazy as LBS+import qualified Data.ByteString.Lazy.Char8 as LBS8+import qualified Data.HashMap.Strict as HM+import Data.Int+import Data.Scientific+import Data.Text (Text)+import Data.Time+import Data.Time.Clock.POSIX+import qualified Data.Vector as V+import Hedgehog as HH+import qualified Hedgehog.Gen as Gen+import qualified Hedgehog.Range as Range+import Katip+import Katip.Core -------------------------------------------------------------------------------+import Katip.Scribes.LogzIO.HTTPS+import Language.Haskell.TH (Loc (..))+import qualified Network.HTTP.Types.Status as HTypes+import Network.HostName+import qualified Network.Wai.Handler.Warp as Warp+import System.Posix+import System.Timeout+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.Hedgehog+import URI.ByteString+import qualified Web.Scotty as Scotty +------------------------------------------------------------------------------- tests :: TestTree-tests = testGroup "Katip.Scribes.LogzIO.HTTP"- [ bufferItemTests- , renderLineTruncatedTests- , scribeTests- ]-+tests =+ testGroup+ "Katip.Scribes.LogzIO.HTTP"+ [ bufferItemTests,+ renderLineTruncatedTests,+ scribeTests+ ] ------------------------------------------------------------------------------- -- We bind to a system port so we use a lock on these tests to avoid interleaved, concurrent requests scribeTests :: TestTree-scribeTests = withLock $ \mkSem -> testGroup "scribe"- [ testCase "writes well-formed log items as lines to the given API" $ do- sem <- mkSem- withServer sem $ \receivedMessages -> do- withScribe (\cfg -> cfg { logzIOScribeConfiguration_bufferTimeout = 1}) $ \scribe -> do- (liPush scribe) (genericLog "a message")+scribeTests = withLock $ \mkSem ->+ testGroup+ "scribe"+ [ testCase "writes well-formed log items as lines to the given API" $ do+ sem <- mkSem+ withServer sem $ \receivedMessages -> do+ withScribe (\cfg -> cfg {logzIOScribeConfiguration_bufferTimeout = 1}) $ \scribe -> do+ (liPush scribe) (genericLog "a message") - requestLog <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))- requestLog_token requestLog @?= testAPIToken- let logs = requestLog_logs requestLog- length logs @?= 1- let firstLog = head logs- receivedLog_message firstLog @?= "a message"- , testCase "respects the items limit" $ do- sem <- mkSem- withServer sem $ \receivedMessages -> do- withScribe (\cfg -> cfg { logzIOScribeConfiguration_bufferItems = 1, logzIOScribeConfiguration_bufferTimeout = 1}) $ \scribe -> do- (liPush scribe) (genericLog "message 1")- (liPush scribe) (genericLog "message 2")+ requestLog <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))+ requestLog_token requestLog @?= testAPIToken+ let logs = requestLog_logs requestLog+ length logs @?= 1+ let firstLog = head logs+ receivedLog_message firstLog @?= "a message",+ testCase "respects the items limit" $ do+ sem <- mkSem+ withServer sem $ \receivedMessages -> do+ withScribe (\cfg -> cfg {logzIOScribeConfiguration_bufferItems = 1, logzIOScribeConfiguration_bufferTimeout = 1}) $ \scribe -> do+ (liPush scribe) (genericLog "message 1")+ (liPush scribe) (genericLog "message 2") - log1 <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))- requestLog_token log1 @?= testAPIToken- let logs1 = requestLog_logs log1- length logs1 @?= 1- let firstLog1 = head logs1- receivedLog_message firstLog1 @?= "message 1"+ log1 <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))+ requestLog_token log1 @?= testAPIToken+ let logs1 = requestLog_logs log1+ length logs1 @?= 1+ let firstLog1 = head logs1+ receivedLog_message firstLog1 @?= "message 1" - log2 <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))- let logs2 = requestLog_logs log2- length logs2 @?= 1- let firstLog2 = head logs2- receivedLog_message firstLog2 @?= "message 2"- , testCase "respects the time limit" $ do- sem <- mkSem- withServer sem $ \receivedMessages -> do- withScribe (\cfg -> cfg { logzIOScribeConfiguration_bufferItems = 2, logzIOScribeConfiguration_bufferTimeout = 1}) $ \scribe -> do- (liPush scribe) (genericLog "a message")- log1 <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))- requestLog_token log1 @?= testAPIToken- let logs1 = requestLog_logs log1- length logs1 @?= 1- let firstLog1 = head logs1- receivedLog_message firstLog1 @?= "a message"- , testCase "reports errors" $ do- sem <- mkSem- let badToken = APIToken "badtoken"- errorSignal <- newEmptyTMVarIO- withServer sem $ \_receivedMessages -> do- withScribe (\cfg -> cfg { logzIOScribeConfiguration_bufferItems = 1, logzIOScribeConfiguration_bufferTimeout = 1, logzIOScribeConfiguration_token = badToken, logzIOScribeConfiguration_onError = void . atomically . tryPutTMVar errorSignal}) $ \scribe -> do- (liPush scribe) (genericLog "a message")- loggingError <- timeoutFailure (20 * second) (atomically (takeTMVar errorSignal))- case loggingError of- BadToken -> pure ()- _ -> assertFailure ("Expected BadToken error but got " <> show loggingError)- , testCase "flushes all queued logs on close" $ do- sem <- mkSem- withServer sem $ \receivedMessages -> do- withScribe (\cfg -> cfg { logzIOScribeConfiguration_bufferItems = 100, logzIOScribeConfiguration_bufferTimeout = 100}) $ \scribe -> do- (liPush scribe) (genericLog "message 1")- (liPush scribe) (genericLog "message 2")- messages <- atomically (flushTQueue receivedMessages)- let allLogs = receivedLog_message <$> mconcat (requestLog_logs <$> messages)- allLogs @?= ["message 1", "message 2"]- ]+ log2 <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))+ let logs2 = requestLog_logs log2+ length logs2 @?= 1+ let firstLog2 = head logs2+ receivedLog_message firstLog2 @?= "message 2",+ testCase "respects the time limit" $ do+ sem <- mkSem+ withServer sem $ \receivedMessages -> do+ withScribe (\cfg -> cfg {logzIOScribeConfiguration_bufferItems = 2, logzIOScribeConfiguration_bufferTimeout = 1}) $ \scribe -> do+ (liPush scribe) (genericLog "a message")+ log1 <- timeoutFailure (20 * second) (atomically (readTQueue receivedMessages))+ requestLog_token log1 @?= testAPIToken+ let logs1 = requestLog_logs log1+ length logs1 @?= 1+ let firstLog1 = head logs1+ receivedLog_message firstLog1 @?= "a message",+ testCase "reports errors" $ do+ sem <- mkSem+ let badToken = APIToken "badtoken"+ errorSignal <- newEmptyTMVarIO+ withServer sem $ \_receivedMessages -> do+ withScribe (\cfg -> cfg {logzIOScribeConfiguration_bufferItems = 1, logzIOScribeConfiguration_bufferTimeout = 1, logzIOScribeConfiguration_token = badToken, logzIOScribeConfiguration_onError = void . atomically . tryPutTMVar errorSignal}) $ \scribe -> do+ (liPush scribe) (genericLog "a message")+ loggingError <- timeoutFailure (20 * second) (atomically (takeTMVar errorSignal))+ case loggingError of+ BadToken -> pure ()+ _ -> assertFailure ("Expected BadToken error but got " <> show loggingError),+ testCase "flushes all queued logs on close" $ do+ sem <- mkSem+ withServer sem $ \receivedMessages -> do+ withScribe (\cfg -> cfg {logzIOScribeConfiguration_bufferItems = 100, logzIOScribeConfiguration_bufferTimeout = 100}) $ \scribe -> do+ (liPush scribe) (genericLog "message 1")+ (liPush scribe) (genericLog "message 2")+ messages <- atomically (flushTQueue receivedMessages)+ let allLogs = receivedLog_message <$> mconcat (requestLog_logs <$> messages)+ allLogs @?= ["message 1", "message 2"]+ ] where second = 1000000 timeoutFailure waitTime f = do@@ -130,19 +134,20 @@ receivedMessages <- atomically newTQueue withAsync (startServer receivedMessages) $ \_async -> f receivedMessages testAPIToken = "bigSECRET"- genericLog message = Item- { _itemApp = "katip-logzio"- , _itemEnv = "test"- , _itemSeverity = InfoS- , _itemThread = ThreadIdText "42"- , _itemHost = "localhost"- , _itemProcess = 111- , _itemPayload = ()- , _itemMessage = message- , _itemTime = mkUTCTime 2019 1 2 3 4 5- , _itemNamespace = "test"- , _itemLoc = Nothing- }+ genericLog message =+ Item+ { _itemApp = "katip-logzio",+ _itemEnv = "test",+ _itemSeverity = InfoS,+ _itemThread = ThreadIdText "42",+ _itemHost = "localhost",+ _itemProcess = 111,+ _itemPayload = (),+ _itemMessage = message,+ _itemTime = mkUTCTime 2019 1 2 3 4 5,+ _itemNamespace = "test",+ _itemLoc = Nothing+ } startServer receivedMessages = Scotty.scottyOpts serverOpts $ do Scotty.post "/" $ do token <- APIToken <$> Scotty.param "token"@@ -150,143 +155,149 @@ then do bodyLines <- LBS8.lines <$> Scotty.body receivedLogs <- mapM (either error pure . A.eitherDecode) bodyLines- let requestLog = RequestLog- { requestLog_token = token- , requestLog_logs = receivedLogs- }+ let requestLog =+ RequestLog+ { requestLog_token = token,+ requestLog_logs = receivedLogs+ } liftIO (atomically (writeTQueue receivedMessages requestLog)) else Scotty.status HTypes.unauthorized401- serverOpts = Scotty.Options- { Scotty.verbose = 0- , Scotty.settings = Warp.setPort 1337 Warp.defaultSettings- }+ serverOpts =+ Scotty.Options+ { Scotty.verbose = 0,+ Scotty.settings = Warp.setPort 1337 Warp.defaultSettings+ } withScribe modConfig = bracket (mkLogzIOScribe (modConfig scribeConfig) (const (pure True)) V3) scribeFinalizer- scribeConfig = LogzIOScribeConfiguration- { logzIOScribeConfiguration_bufferItems = 5- , logzIOScribeConfiguration_bufferTimeout = 10- , logzIOScribeConfiguration_scheme = HTTP- , logzIOScribeConfiguration_host = Host "127.0.0.1"- , logzIOScribeConfiguration_port = Port 1337- , logzIOScribeConfiguration_token = testAPIToken- , logzIOScribeConfiguration_retry = defaultRetryPolicy- , logzIOScribeConfiguration_onError = const (pure ())- }--+ scribeConfig =+ LogzIOScribeConfiguration+ { logzIOScribeConfiguration_bufferItems = 5,+ logzIOScribeConfiguration_bufferTimeout = 10,+ logzIOScribeConfiguration_scheme = HTTP,+ logzIOScribeConfiguration_host = Host "127.0.0.1",+ logzIOScribeConfiguration_port = Port 1337,+ logzIOScribeConfiguration_token = testAPIToken,+ logzIOScribeConfiguration_retry = defaultRetryPolicy,+ logzIOScribeConfiguration_onError = const (pure ())+ } data RequestLog = RequestLog- { requestLog_token :: APIToken- , requestLog_logs :: [ReceivedLog]+ { requestLog_token :: APIToken,+ requestLog_logs :: [ReceivedLog] } - -- | Simplified log that only parses the fields we want to assert on data ReceivedLog = ReceivedLog { receivedLog_message :: Text } - instance FromJSON ReceivedLog where parseJSON = withObject "ReceivedLog" $ \o -> do message <- o .: "message"- pure $ ReceivedLog- { receivedLog_message = message- }-+ pure $+ ReceivedLog+ { receivedLog_message = message+ } ------------------------------------------------------------------------------- bufferItemTests :: TestTree-bufferItemTests = testGroup "bufferItem"- [ testProperty "0 or negative maximum items always flushes immediately" $ property $ do- item <- forAllWith showSimplePayloadItem genSimplePayloadItem- verbosity <- forAll genVerbosity- maxItems <- forAll (Gen.int (Range.linear minBound 0))- bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer- let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer- assertFlushNow logAction- , testProperty "never buffers too many items" $ property $ do- item <- forAllWith showSimplePayloadItem genSimplePayloadItem- verbosity <- forAll genVerbosity- maxItems <- forAll (Gen.int (Range.linear 0 100))- bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer- let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer- case logAction of- Buffered newBuffer -> HH.assert (bulkBuffer_itemCount newBuffer <= maxItems)- _ -> pure ()- , testProperty "never produces too many bytes" $ property $ do- item <- forAllWith showSimplePayloadItem genSimplePayloadItem- verbosity <- forAll genVerbosity- maxItems <- forAll (Gen.int (Range.linear 0 100))- bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer- let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer- case logAction of- Buffered newBuffer -> HH.assert (bulkBuffer_bytesUsed newBuffer <= reducedMaxPayloadBytes)- _ -> pure ()- , testProperty "never flushes more than 1 item" $ property $ do- item <- forAllWith showSimplePayloadItem genSimplePayloadItem- verbosity <- forAll genVerbosity- maxItems <- forAll (Gen.int (Range.linear minBound 0))- bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer- let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer- case logAction of- FlushNow _ newBuffer -> bulkBuffer_itemCount newBuffer === 1- _ -> pure ()- , testProperty "flushes when the buffer is exactly at the limit" $ property $ do- maxItems <- forAll (Gen.int (Range.linear 1 20))- item <- forAllWith showSimplePayloadItem genSimplePayloadItem- bulkBuffer <- forAllWith inspectBulkBuffer ((\b -> b { bulkBuffer_itemCount = maxItems - 1 }) <$> genBulkBuffer)- verbosity <- forAll genVerbosity- let logAction = bufferItem' maxBound maxBound maxItems verbosity item bulkBuffer- case logAction of- FlushNow _ _ -> pure ()- Buffered _ -> fail "Expected a FlushNow but got a Buffered"- ]+bufferItemTests =+ testGroup+ "bufferItem"+ [ testProperty "0 or negative maximum items always flushes immediately" $+ property $ do+ item <- forAllWith showSimplePayloadItem genSimplePayloadItem+ verbosity <- forAll genVerbosity+ maxItems <- forAll (Gen.int (Range.linear minBound 0))+ bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer+ let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer+ assertFlushNow logAction,+ testProperty "never buffers too many items" $+ property $ do+ item <- forAllWith showSimplePayloadItem genSimplePayloadItem+ verbosity <- forAll genVerbosity+ maxItems <- forAll (Gen.int (Range.linear 0 100))+ bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer+ let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer+ case logAction of+ Buffered newBuffer -> HH.assert (bulkBuffer_itemCount newBuffer <= maxItems)+ _ -> pure (),+ testProperty "never produces too many bytes" $+ property $ do+ item <- forAllWith showSimplePayloadItem genSimplePayloadItem+ verbosity <- forAll genVerbosity+ maxItems <- forAll (Gen.int (Range.linear 0 100))+ bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer+ let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer+ case logAction of+ Buffered newBuffer -> HH.assert (bulkBuffer_bytesUsed newBuffer <= reducedMaxPayloadBytes)+ _ -> pure (),+ testProperty "never flushes more than 1 item" $+ property $ do+ item <- forAllWith showSimplePayloadItem genSimplePayloadItem+ verbosity <- forAll genVerbosity+ maxItems <- forAll (Gen.int (Range.linear minBound 0))+ bulkBuffer <- forAllWith inspectBulkBuffer genBulkBuffer+ let logAction = bufferItem' reducedMaxPayloadBytes reducedMaxLogLineLength maxItems verbosity item bulkBuffer+ case logAction of+ FlushNow _ newBuffer -> bulkBuffer_itemCount newBuffer === 1+ _ -> pure (),+ testProperty "flushes when the buffer is exactly at the limit" $+ property $ do+ maxItems <- forAll (Gen.int (Range.linear 1 20))+ item <- forAllWith showSimplePayloadItem genSimplePayloadItem+ bulkBuffer <- forAllWith inspectBulkBuffer ((\b -> b {bulkBuffer_itemCount = maxItems - 1}) <$> genBulkBuffer)+ verbosity <- forAll genVerbosity+ let logAction = bufferItem' maxBound maxBound maxItems verbosity item bulkBuffer+ case logAction of+ FlushNow _ _ -> pure ()+ Buffered _ -> fail "Expected a FlushNow but got a Buffered"+ ] where assertFlushNow (FlushNow _ _) = pure ()- assertFlushNow (Buffered _) = fail ("Expceted FlushNow but got a Buffered")+ assertFlushNow (Buffered _) = fail ("Expceted FlushNow but got a Buffered") inspectBulkBuffer (BulkBuffer used _ itemCount) = "BulkBuffer (" <> show used <> ") ... " <> show itemCount - -------------------------------------------------------------------------------+ -- | Reduced max log line length for more space and time efficient tests reducedMaxLogLineLength :: Bytes reducedMaxLogLineLength = Bytes (bytes maxLogLineLength `div` limitScalingFactor) - reducedMaxPayloadBytes :: Bytes reducedMaxPayloadBytes = Bytes (bytes maxPayloadBytes `div` limitScalingFactor) - limitScalingFactor :: Int64 limitScalingFactor = 10 ------------------------------------------------------------------------------- showSimplePayloadItem :: Item SimpleLogPayload -> String showSimplePayloadItem item =- show (item { _itemPayload = [(k, toJSON v) | (k, AnyLogPayload v) <- unSimpleLogPayload (_itemPayload item)]})-+ show (item {_itemPayload = [(k, toJSON v) | (k, AnyLogPayload v) <- unSimpleLogPayload (_itemPayload item)]}) ------------------------------------------------------------------------------- renderLineTruncatedTests :: TestTree-renderLineTruncatedTests = testGroup "renderLineTruncated"- [ testProperty "always produces an accurate byte count" $ property $ do- item <- forAllWith showSimplePayloadItem genSimplePayloadItem- verbosity <- forAll genVerbosity- let (builder, Bytes byteCount) = renderLineTruncated' reducedMaxLogLineLength verbosity item- builderLength builder === byteCount- , testProperty "always produces a payload smaller than the line limit" $ property $ do- let tinyMaxLineLength = 1024 -- set a really low one for speed- let lowerBound = fromIntegral (bytes tinyMaxLineLength - 100)- let upperBound = fromIntegral (bytes tinyMaxLineLength + 100)- bigMessage <- forAll (genLogStr' (Range.linear lowerBound upperBound))- item <- (\item -> item { _itemMessage = bigMessage}) <$> forAllWith showSimplePayloadItem genSimplePayloadItem- verbosity <- forAll genVerbosity- let (_, byteCount) = renderLineTruncated' tinyMaxLineLength verbosity item- HH.assert (byteCount <= reducedMaxLogLineLength)- ]-+renderLineTruncatedTests =+ testGroup+ "renderLineTruncated"+ [ testProperty "always produces an accurate byte count" $+ property $ do+ item <- forAllWith showSimplePayloadItem genSimplePayloadItem+ verbosity <- forAll genVerbosity+ let (builder, Bytes byteCount) = renderLineTruncated' reducedMaxLogLineLength verbosity item+ builderLength builder === byteCount,+ testProperty "always produces a payload smaller than the line limit" $+ property $ do+ let tinyMaxLineLength = 1024 -- set a really low one for speed+ let lowerBound = fromIntegral (bytes tinyMaxLineLength - 100)+ let upperBound = fromIntegral (bytes tinyMaxLineLength + 100)+ bigMessage <- forAll (genLogStr' (Range.linear lowerBound upperBound))+ item <- (\item -> item {_itemMessage = bigMessage}) <$> forAllWith showSimplePayloadItem genSimplePayloadItem+ verbosity <- forAll genVerbosity+ let (_, byteCount) = renderLineTruncated' tinyMaxLineLength verbosity item+ HH.assert (byteCount <= reducedMaxLogLineLength)+ ] ------------------------------------------------------------------------------- builderLength :: BB.Builder -> Int64@@ -296,9 +307,8 @@ genVerbosity :: Gen Verbosity genVerbosity = Gen.enumBounded - --------------------------------------------------------------------------------genBulkBuffer :: Gen BulkBuffer+genBulkBuffer :: Gen BulkBuffer genBulkBuffer = do bytesUsed <- genBytes -- generating a large bytestring string with hedghog seems quite slow, instead we'll generate a smaller string and replicate and merge it@@ -306,23 +316,21 @@ payloadSegment <- genBuilder (Range.linear 0 (round (toRational (bytes reducedMaxPayloadBytes) * 0.05))) let payload = mconcat (replicate payloadMultiplier payloadSegment) itemCount <- Gen.int (Range.linear 0 maxBound)- pure $ BulkBuffer- { bulkBuffer_bytesUsed = bytesUsed- , bulkBuffer_payload = payload- , bulkBuffer_itemCount = itemCount- }-+ pure $+ BulkBuffer+ { bulkBuffer_bytesUsed = bytesUsed,+ bulkBuffer_payload = payload,+ bulkBuffer_itemCount = itemCount+ } ------------------------------------------------------------------------------- genBytes :: Gen Bytes genBytes = Bytes <$> Gen.int64 (Range.linear 0 maxBound) - ------------------------------------------------------------------------------- genBuilder :: Range Int -> Gen BB.Builder genBuilder size = BB.byteString <$> Gen.bytes size - ------------------------------------------------------------------------------- genSimplePayloadItem :: Gen (Item SimpleLogPayload) genSimplePayloadItem = do@@ -337,127 +345,117 @@ time <- genUTCTime namespace <- genNamespace loc <- Gen.maybe genLoc- pure $ Item- { _itemApp = app- , _itemEnv = env- , _itemSeverity = severity- , _itemThread = thread- , _itemHost = host- , _itemProcess = process- , _itemPayload = payload- , _itemMessage = message- , _itemTime = time- , _itemNamespace = namespace- , _itemLoc = loc- }-+ pure $+ Item+ { _itemApp = app,+ _itemEnv = env,+ _itemSeverity = severity,+ _itemThread = thread,+ _itemHost = host,+ _itemProcess = process,+ _itemPayload = payload,+ _itemMessage = message,+ _itemTime = time,+ _itemNamespace = namespace,+ _itemLoc = loc+ } genNamespace :: Gen Namespace genNamespace = Namespace <$> Gen.list (Range.linear 0 20) genSmallText - genEnvironment :: Gen Environment genEnvironment = Environment <$> genSmallText - genSeverity :: Gen Severity genSeverity = Gen.enumBounded - genThreadIdText :: Gen ThreadIdText genThreadIdText = ThreadIdText <$> genSmallText - genHostName :: Gen HostName genHostName = genSmallString - genProcessID :: Gen ProcessID genProcessID = Gen.enumBounded - genLogStr :: Gen LogStr genLogStr = genLogStr' (Range.linear 0 1024) - genLogStr' :: Range Int -> Gen LogStr genLogStr' sizeRange = ls <$> Gen.text sizeRange Gen.unicode - genUTCTime :: Gen UTCTime-genUTCTime = posixSecondsToUTCTime . fromIntegral- <$> Gen.int (Range.linear 0 maxBound)-+genUTCTime =+ posixSecondsToUTCTime . fromIntegral+ <$> Gen.int (Range.linear 0 maxBound) genLoc :: Gen Loc-genLoc = Loc- <$> genSmallString- <*> genSmallString- <*> genSmallString- <*> genCharPos- <*> genCharPos+genLoc =+ Loc+ <$> genSmallString+ <*> genSmallString+ <*> genSmallString+ <*> genCharPos+ <*> genCharPos where- genCharPos = (,)- <$> Gen.int (Range.linear 0 maxBound)- <*> Gen.int (Range.linear 0 maxBound)-+ genCharPos =+ (,)+ <$> Gen.int (Range.linear 0 maxBound)+ <*> Gen.int (Range.linear 0 maxBound) genSimpleLogPayload :: Gen SimpleLogPayload-genSimpleLogPayload = SimpleLogPayload- <$> Gen.list (Range.linear 0 20) genPair+genSimpleLogPayload =+ SimpleLogPayload+ <$> Gen.list (Range.linear 0 20) genPair where- genPair = (,)- <$> genSmallText- <*> genAnyLogPayload-+ genPair =+ (,)+ <$> genSmallText+ <*> genAnyLogPayload genAnyLogPayload :: Gen AnyLogPayload-genAnyLogPayload = AnyLogPayload- <$> genValue-+genAnyLogPayload =+ AnyLogPayload+ <$> genValue genValue :: Gen Value genValue = Gen.recursive Gen.choice nonRecursiveGens recursiveGens where nonRecursiveGens =- [ pure A.Null- , A.String <$> genSmallText- , A.Number <$> genScientific- , A.Bool <$> Gen.bool+ [ pure A.Null,+ A.String <$> genSmallText,+ A.Number <$> genScientific,+ A.Bool <$> Gen.bool ] recursiveGens =- [ A.Object <$> genObject- , A.Array <$> genArray+ [ A.Object <$> genObject,+ A.Array <$> genArray ] genObject = HM.fromList <$> Gen.list (Range.linear 0 20) genPair where genPair = (,) <$> genSmallText <*> genValue- genArray = V.fromList <$>- Gen.list (Range.linear 0 20) genValue--+ genArray =+ V.fromList+ <$> Gen.list (Range.linear 0 20) genValue genScientific :: Gen Scientific-genScientific = scientific- <$> Gen.integral (Range.linear (-999999) 999999)- <*> Gen.int (Range.linear (-999) (999))-+genScientific =+ scientific+ <$> Gen.integral (Range.linear (-999999) 999999)+ <*> Gen.int (Range.linear (-999) (999)) genSmallText :: Gen Text genSmallText = Gen.text (Range.linear 0 128) Gen.unicode - genSmallString :: Gen String genSmallString = Gen.string (Range.linear 0 128) Gen.unicode - ------------------------------------------------------------------------------- mkUTCTime :: Integer -> Int -> Int -> DiffTime -> DiffTime -> DiffTime -> UTCTime mkUTCTime y m d hr mn s = UTCTime day dt where day = mkDay y m d dt = s + 60 * mn + 60 * 60 * hr- ------------------------------------------------------------------------------- mkDay :: Integer -> Int -> Int -> Day
test/Main.hs view
@@ -1,22 +1,23 @@ module Main- ( main- ) where-+ ( main,+ )+where --------------------------------------------------------------------------------import Test.Tasty+ ------------------------------------------------------------------------------- import qualified Katip.Tests.Scribes.LogzIO.HTTPS---------------------------------------------------------------------------------+import Test.Tasty +------------------------------------------------------------------------------- main :: IO () main = defaultMain tests - ------------------------------------------------------------------------------- tests :: TestTree-tests = testGroup "katip-logzio"- [ Katip.Tests.Scribes.LogzIO.HTTPS.tests- ]+tests =+ testGroup+ "katip-logzio"+ [ Katip.Tests.Scribes.LogzIO.HTTPS.tests+ ]