diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2018, Soostone Inc + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Ozgun Ataman nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,4 @@ +# Katip Logz.IO [![Build Status](https://travis-ci.org/Soostone/katip.svg?branch=master)](https://travis-ci.org/Soostone/katip) + +katip-logzio is a scribe for the Katip logging framework that sends +structured logs to the logz.io centralized logging platform. diff --git a/changelog.md b/changelog.md new file mode 100644 --- /dev/null +++ b/changelog.md @@ -0,0 +1,3 @@ +0.1.0.0 +========== +* Initial release diff --git a/katip-logzio.cabal b/katip-logzio.cabal new file mode 100644 --- /dev/null +++ b/katip-logzio.cabal @@ -0,0 +1,93 @@ +name: katip-logzio +version: 0.1.0.0 +synopsis: Logz.IO scribe for the Katip logging framework +description: See README.md for more details. +license: BSD3 +license-file: LICENSE +author: Michael Xavier +maintainer: michael.xavier@soostone.com +copyright: Soostone Inc, 2018 +category: Data, Text, Logging +homepage: https://github.com/Soostone/katip +bug-reports: https://github.com/Soostone/katip/issues +build-type: Simple +cabal-version: >=1.10 +extra-source-files: + README.md + changelog.md + +tested-with: GHC == 8.4.3 + +source-repository head + type: git + location: https://github.com/Soostone/katip.git + +flag lib-Werror + default: False + manual: True + + +library + exposed-modules: + Katip.Scribes.LogzIO.HTTPS + build-depends: + base >=4.8.0.0 && <5 + , katip >= 0.8.0.0 + , stm >= 2.1.1 + , async + , time >= 1.8.0.2 + , aeson >= 0.10.0.0 + , unix + , unordered-containers + , http-client >= 0.1.0 + , http-client-tls >= 0.3.4 + , uri-bytestring + , errors >= 2.0.0 + , safe-exceptions + , retry + , http-types + , text + , bytestring + , scientific + hs-source-dirs: src + default-language: Haskell2010 + hs-source-dirs: src + ghc-options: -Wall -Wall -Wnoncanonical-monad-instances + if flag(lib-Werror) + ghc-options: -Werror + + +test-suite test + type: exitcode-stdio-1.0 + main-is: Main.hs + hs-source-dirs: test + default-language: Haskell2010 + other-modules: + Katip.Tests.Scribes.LogzIO.HTTPS + build-depends: base + , katip + , katip-logzio + , tasty + , tasty-hedgehog + , tasty-hunit + , hedgehog + , vector + , unordered-containers + , text + , bytestring + , time + , template-haskell + , aeson + , scientific + , unix + , hostname + , scotty + , stm + , async + , safe-exceptions + , uri-bytestring + , warp + , http-types + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -Wnoncanonical-monad-instances + if flag(lib-Werror) + ghc-options: -Werror diff --git a/src/Katip/Scribes/LogzIO/HTTPS.hs b/src/Katip/Scribes/LogzIO/HTTPS.hs new file mode 100644 --- /dev/null +++ b/src/Katip/Scribes/LogzIO/HTTPS.hs @@ -0,0 +1,671 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE TypeApplications #-} +-- | This is a log scribe that writes logs to logz.io's bulk +-- . +module Katip.Scribes.LogzIO.HTTPS + ( -- * 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 + + +------------------------------------------------------------------------------- +import Control.Applicative +import qualified Control.Concurrent.Async as Async +import qualified Control.Concurrent.STM as STM +import qualified Control.Error as E +import qualified Control.Exception.Safe as EX +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 +------------------------------------------------------------------------------- +------------------------------------------------------------------------------- + + + +-- | 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) + + +instance A.FromJSON BulkAPIError where + parseJSON = A.withObject "BulkAPIError" $ \o -> do + 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 + } + + +------------------------------------------------------------------------------- +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 () + } + + +-- | A default configuration: +-- +-- * 100 item buffering +-- +-- * 30 second buffer timeout +-- +-- * US, HTTPS logging endpoint (listener.logz.io:8071) +-- +-- * 'defaultRetryPolicy' of 25ms exponential backoff, 5 retries +-- +-- * 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 ()) + } + + +------------------------------------------------------------------------------- +-- | You can retrieve your account or sub-account's API token on the +-- . Note that APIToken has an IsString instance, +-- meaning that you can use a string literal with OverloadedStrings +-- enabled. +newtype APIToken = APIToken + { apiToken :: T.Text + } 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. + | HTTP + deriving (Show, Eq) + + +-- | See +-- for a list of listeners. This is the US region host, +-- listener.logz.io +usRegionHost :: URIBS.Host +usRegionHost = URIBS.Host "listener.logz.io" + + +-- | See +-- for a list of listeners. This is the EU region host, +-- listener.logz.io +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 config permitItem verbosity = do + -- Accumulating buffer for each item + bufferState <- STM.newTVarIO (mempty :: BulkBuffer) + -- Single-slot TMVar when something is ready to write. Katip buffers + -- for us upstream so we can simplify shutdown and flush guarantees + -- by synchronously writing when we need to. + readyMVar <- STM.newEmptyTMVarIO :: IO (STM.TMVar BulkBuffer) + + -- We create a TVar that is set to ShutDown when its time to stop + killSwitch <- STM.newTVarIO Continue + + -- Create a mutable ref to whether the timer has expired. This lets + -- us replace the timer when its time to start timing again. We + -- reset the timer __after__ a write goes out, not as soon as it expires + timerRef <- STM.newTVarIO =<< STM.registerDelay itemBufferTimeoutMicros + + -- Set up a connection manager for requests + mgr <- case scheme of + HTTPS -> HTTPS.newTlsManager + HTTP -> HTTP.newManager HTTP.defaultManagerSettings + + -- In STM: resolve an action with the buffers + let processLogAction :: LogAction BulkBuffer -> STM.STM () + processLogAction logAction = case logAction of + -- No write needed right now + Buffered !newBuffer -> STM.writeTVar bufferState newBuffer + -- Time to flush and put excess in the next buffer + FlushNow !toFlush !newBuffer -> do + STM.writeTVar bufferState newBuffer + STM.putTMVar readyMVar toFlush + + -- Recompute the new buffer state from a new log item and take action + let push :: (K.LogItem a) => K.Item a -> STM.STM () + push !logItem = do + !curBuffer <- STM.readTVar bufferState + processLogAction (bufferItem itemBufferSize verbosity logItem curBuffer) + + + -- A transaction that retries until its time to shutdown + let waitShutdown :: STM.STM () + waitShutdown = do + killStatus <- STM.readTVar killSwitch + STM.check (killStatus == ShutDown) + + -- Retry transaction until timer has expired. When it does, take action on the buffer. + let waitTimer :: STM.STM () + waitTimer = do + -- get the timer we'll be using + timer <- STM.readTVar timerRef + -- do not proceed until its ready + timerExpired <- STM.readTVar timer + STM.check timerExpired + -- do not wake again until this message is out the door + STM.writeTVar timer False + -- we waited long enough, grab the current buffer + !curBuffer <- STM.readTVar bufferState + -- and tell it to flush right now + processLogAction (forceFlush curBuffer) + + -- run the timer in a loop until its time to shut down + let timedFlushLoop :: IO () + timedFlushLoop = do + res <- STM.atomically $ + raceAlt + waitShutdown + waitTimer + case res of + Left () -> pure () -- exit loop + Right () -> timedFlushLoop -- wait for another timer or shutdown + + timedFlusher <- Async.async timedFlushLoop + + -- flush the buffer until its time to shutdown and we're out of items to log + let flushLoop :: IO () + flushLoop = do + -- hold until something comes in to flush or we're told to stop. If we're told to stop but get a log first, we'll catch it on the next loop + readyBufferOrDie <- STM.atomically $ + raceAlt + (STM.takeTMVar readyMVar) + waitShutdown + case readyBufferOrDie of + Right () -> pure () -- we caught shutdown before any new data arrived + -- There's some logs to write + Left readyBuffer -> do + -- flush in IO + res <- flushBuffer config mgr readyBuffer + case res of + Left ex -> onErrorSafe ex + Right () -> pure () + -- reset the timeout before the next flush on each write + !newTimer <- STM.registerDelay itemBufferTimeoutMicros + STM.atomically $ do + STM.writeTVar timerRef newTimer + -- start over + flushLoop + flusher <- Async.async flushLoop + + -- Blocking call which sends a poison pill to the workers and waits + -- for them to exit, swallowing exceptions. This follows the + -- semantics that when katip calls the finalizer, no more log + -- entries will be sent. + let shutdown :: IO () + shutdown = do + -- tell workers this is the last loop + STM.atomically (STM.writeTVar killSwitch ShutDown) + -- wait for workers to finish + _ <- Async.waitCatch timedFlusher + _ <- Async.waitCatch flusher + -- stop blocking + pure () + + + pure $ K.Scribe + { K.liPush = STM.atomically . push + , K.scribeFinalizer = shutdown + , K.scribePermitItem = permitItem + } + where + itemBufferSize = logzIOScribeConfiguration_bufferItems config + itemBufferTimeoutMicros = ndtToMicros (logzIOScribeConfiguration_bufferTimeout config) + scheme = logzIOScribeConfiguration_scheme config + -- should this be async? + onErrorSafe ex = do + _ <- EX.tryAny (logzIOScribeConfiguration_onError config ex) + pure () + + +------------------------------------------------------------------------------- +-- Match the native resolution of NominalDiffTime, which is excessive +-- for our needs +ndtPicos :: Time.NominalDiffTime -> Int64 +ndtPicos = round . (* picos) + where + picos :: Time.NominalDiffTime + picos = 10 ^ (9 :: Int) + + +------------------------------------------------------------------------------- +ndtToMicros :: Time.NominalDiffTime -> Int +ndtToMicros t = round (((fromIntegral (ndtPicos t)) :: Double) / picosInMicro) + where + picosInMicro = 10 ^ (3 :: Int) + + +------------------------------------------------------------------------------- +-- | An internal mode flag used to shut down worker +-- threads. Internally, we assume htere is no going back from +-- ShutDown. +data KillSwitch = + Continue + | ShutDown + deriving (Eq) + + +------------------------------------------------------------------------------- +-- | When applied to STM, it will try the left side first and if that +-- retries, the right, and if that retries, start from the beginning. +raceAlt :: (Alternative m) => m a -> m b -> m (Either a b) +raceAlt a b = (Left <$> a) <|> (Right <$> b) + + +------------------------------------------------------------------------------- +data LoggingError = + 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. + deriving (Show) + +------------------------------------------------------------------------------- +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) + where + 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)) + + handleHttpException :: (Applicative m) => HTTP.HttpException -> m Bool + handleHttpException _ = pure True + + 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 + } + + +------------------------------------------------------------------------------- +newtype Bytes = Bytes + { bytes :: Int64 + } deriving (Show, Eq, Num, Ord, Bounded) + + +-- | How big of a body can we send? The limit is defined +-- . +maxPayloadBytes :: Bytes +maxPayloadBytes = 10485760 + + +-- | How long can each serialized payload be (let's assume including +-- the trailing newline). The limit is defined +-- . +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) + ] + 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 +-- truncated message. +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 + -- 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) +renderLineTruncated' customMaxLogLineLength verbosity item = + if fullSize <= customMaxLogLineLength + then (fullLine, fullSize) + else (fallbackLine, fallbackSize) + where + fullObject = fullItemObject verbosity item + (fullLine, fullSize) = measureJSONLine fullObject + -- only the absolutely necessary keys, with message stripped out + -- to help us calculate how much of the message we can keep. these + -- 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 + ] + (_, 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) + ] + + +data BulkBuffer = BulkBuffer + { 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) + + +instance Monoid BulkBuffer where + mempty = BulkBuffer 0 mempty 0 + mappend = (<>) + + +data LogAction buf = + 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 = 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' 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 + } + + +-- | 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 +-- thus won't need annotation. However, any code in userland may +-- choose to add akey to their log's metadata with a type that's +-- incompatible with the mapping. If the first value that's picked up +-- restrictive (e.g. a long), subsequent values will be rejected. By +-- differentiating types by their name, this is guaranteed not to +-- happen. + +annotateValue :: A.Value -> A.Value +annotateValue (A.Object o) = A.Object (annotateKeys o) +annotateValue (A.Array a) = A.Array (annotateValue <$> a) +annotateValue x = x + + +annotateKeys :: A.Object -> A.Object +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, 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, b@(A.Bool _)) = (k <> booleanAnn, b) + go (k, A.Null) = (k <> nullAnn, A.Null) + + +------------------------------------------------------------------------------- +-- Annotation Constants +------------------------------------------------------------------------------- + + +stringAnn :: T.Text +stringAnn = "::s" + +doubleAnn :: T.Text +doubleAnn = "::d" + +longAnn :: T.Text +longAnn = "::l" + +booleanAnn :: T.Text +booleanAnn = "::b" + +nullAnn :: T.Text +nullAnn = "::n" diff --git a/test/Katip/Tests/Scribes/LogzIO/HTTPS.hs b/test/Katip/Tests/Scribes/LogzIO/HTTPS.hs new file mode 100644 --- /dev/null +++ b/test/Katip/Tests/Scribes/LogzIO/HTTPS.hs @@ -0,0 +1,457 @@ +{-# LANGUAGE OverloadedStrings #-} +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 +------------------------------------------------------------------------------- + + +tests :: TestTree +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") + + 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" + + 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) + ] + where + second = 1000000 + timeoutFailure waitTime f = do + res <- timeout waitTime f + case res of + Nothing -> assertFailure ("Operation timed out after " <> show waitTime <> "us") + Just x -> pure x + withLock = withResource (atomically (newTSem 1)) (const (return ())) + withServer sem f = bracket_ (atomically (waitTSem sem)) (atomically (signalTSem sem)) $ do + 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 + } + startServer receivedMessages = Scotty.scottyOpts serverOpts $ do + Scotty.post "/" $ do + token <- APIToken <$> Scotty.param "token" + if token == testAPIToken + then do + bodyLines <- LBS8.lines <$> Scotty.body + receivedLogs <- mapM (either error pure . A.eitherDecode) bodyLines + 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 + } + 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 ()) + } + + + +data RequestLog = RequestLog + { 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 + } + + +------------------------------------------------------------------------------- +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" + ] + where + assertFlushNow (FlushNow _ _) = pure () + 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)]}) + + +------------------------------------------------------------------------------- +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) + ] + + +------------------------------------------------------------------------------- +builderLength :: BB.Builder -> Int64 +builderLength = LBS.length . BB.toLazyByteString + +------------------------------------------------------------------------------- +genVerbosity :: Gen Verbosity +genVerbosity = Gen.enumBounded + + +------------------------------------------------------------------------------- +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 + payloadMultiplier <- Gen.int (Range.linear 1 100) + 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 + } + + +------------------------------------------------------------------------------- +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 + app <- genNamespace + env <- genEnvironment + severity <- genSeverity + thread <- genThreadIdText + host <- genHostName + process <- genProcessID + payload <- genSimpleLogPayload + message <- genLogStr + 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 + } + + +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) + + +genLoc :: Gen Loc +genLoc = Loc + <$> genSmallString + <*> genSmallString + <*> genSmallString + <*> genCharPos + <*> genCharPos + where + 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 + where + genPair = (,) + <$> genSmallText + <*> genAnyLogPayload + + +genAnyLogPayload :: Gen AnyLogPayload +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 + ] + recursiveGens = + [ 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 + + + +genScientific :: Gen Scientific +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 +mkDay y m d = day + where + Just day = fromGregorianValid y m d diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,22 @@ +module Main + ( main + ) where + + +------------------------------------------------------------------------------- +import Test.Tasty +------------------------------------------------------------------------------- +import qualified Katip.Tests.Scribes.LogzIO.HTTPS +------------------------------------------------------------------------------- + + + +main :: IO () +main = defaultMain tests + + +------------------------------------------------------------------------------- +tests :: TestTree +tests = testGroup "katip-logzio" + [ Katip.Tests.Scribes.LogzIO.HTTPS.tests + ]