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,6 @@ +# Katip Datadog [![Build Status](https://travis-ci.org/Soostone/katip.svg?branch=master)](https://travis-ci.org/Soostone/katip) + +katip-datadog is a scribe for the Katip logging framework that sends +structured logs to the Datadog logging service. It supports both +directly sending to the ingestion API and through a locally installed +datadog agent. 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-datadog.cabal b/katip-datadog.cabal new file mode 100644 --- /dev/null +++ b/katip-datadog.cabal @@ -0,0 +1,78 @@ +name: katip-datadog +version: 0.1.0.0 +synopsis: Datadog 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 + +--TODO: fill this in +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.Datadog.TCP + build-depends: + base >=4.4.0.0 && <5 + , aeson >= 1.0.0.0 + , bytestring + , katip >= 0.8.0.0 + , text + , connection + , resource-pool + , time + , binary + , network + , safe-exceptions + , retry >= 0.7 + hs-source-dirs: src + default-language: Haskell2010 + hs-source-dirs: src + ghc-options: -Wall + 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.Datadog.TCP + build-depends: base + , katip + , katip-datadog + , conduit + , conduit-extra + , tasty + , tasty-hunit + , aeson + , unordered-containers + , containers + , safe-exceptions + , async + , attoparsec + , text + ghc-options: -Wall + if flag(lib-Werror) + ghc-options: -Werror diff --git a/src/Katip/Scribes/Datadog/TCP.hs b/src/Katip/Scribes/Datadog/TCP.hs new file mode 100644 --- /dev/null +++ b/src/Katip/Scribes/Datadog/TCP.hs @@ -0,0 +1,191 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +-- | Creates a scribe as a Custom Forwarder for Datadog that sends log +-- messages over TCP via their TLS endpoint. +module Katip.Scribes.Datadog.TCP + ( -- * Types + APIKey(..) + , DatadogScribeSettings(..) + , DatadogAuth(..) + , directAPIConnectionParams + , localAgentConnectionParams + -- * Scribe construction + , mkDatadogScribeSettings + , mkDatadogScribe + ) where + + + +------------------------------------------------------------------------------- +import qualified Control.Concurrent as Conc +import qualified Control.Exception.Safe as EX +import Control.Monad (void) +import qualified Control.Retry as Retry +import qualified Data.Aeson as A +import qualified Data.Binary.Builder as BB +import qualified Data.ByteString.Lazy as BSL +import Data.Monoid as Monoid +import qualified Data.Pool as Pool +import Data.Text (Text) +import qualified Data.Text.Encoding as T +import qualified Data.Text.Lazy.Builder as B +import Data.Time (NominalDiffTime) +import Katip +import Katip.Core (LocJs (..)) +import qualified Network as Net +import qualified Network.Connection as C +import qualified System.Posix.Types as POSIX +------------------------------------------------------------------------------- + + +-- | When writing directly to the main intake API, you must specify an +-- API key +newtype APIKey = APIKey + { apiKey :: Text + } deriving (Show, Eq) + + +data DatadogAuth = + NoAuthLocal + -- ^ Writing to the local agent, no auth required + | DirectAuth APIKey + -- ^ Writing directly to the main intake API over TLS . API Key is + -- required. Using a local agent is recommended. + + +data DatadogScribeSettings = DatadogScribeSettings + { datadogScribeSettings_connectionParams :: C.ConnectionParams + -- ^ Specify where the logs should go. If writing directly to the + -- main intake API, you can use 'directAPIConnectionParams'. If + -- writing to a local agent, we recommend + -- 'localAgentConnectionParams' + , datadogScribeSettings_auth :: DatadogAuth + , datadogScribeSettings_poolStripes :: Int + -- ^ How many stripes should be used in the connection pool. 1 is a reasonable default + , datadogScribeSettings_connsPerStripe :: Int + -- ^ How many TCP connections per stripe should be maintained in the pool. + , datadogScribeSettings_connectionKeepalive :: NominalDiffTime + -- ^ How long should a TCP connection be kept open. 30 (seconds) is a reasonable default. Anything around 60 will tend to time out a lot. + , datadogScribeSettings_retry :: Retry.RetryPolicyM IO + -- ^ How should exceptions during write be retried? + } + + +-- | Configured for secure TLS communication to the main intake API at +-- intake.logs.datadoghq.com:10516. Must be combined with DirectAuth +-- authentication. +directAPIConnectionParams :: C.ConnectionParams +directAPIConnectionParams = C.ConnectionParams + { C.connectionHostname = "intake.logs.datadoghq.com" + , C.connectionPort = 10516 + , C.connectionUseSecure = Just $ C.TLSSettingsSimple + { C.settingDisableCertificateValidation = False + , C.settingDisableSession = False + , C.settingUseServerName = False + } + , C.connectionUseSocks = Nothing + } + + +-- | Configure to talk to a local Datadog agent at 127.0.0.1. TLS is not +-- utilized. This is the preferred type of connection because +-- authentication, buffering, and other more advanced features are +-- handled by the agent configuration. +localAgentConnectionParams :: Net.PortNumber -> C.ConnectionParams +localAgentConnectionParams port = C.ConnectionParams + { C.connectionHostname = "127.0.0.1" + , C.connectionPort = port + , C.connectionUseSecure = Nothing + , C.connectionUseSocks = Nothing + } + + +-- | Reasonable defaults. Sets a 30s connection timeout, 1 stripe of +-- connections with a number of connections per stripe equal to +-- 'getNumCapabilities'. Retry policy will do an exponential backoff +-- with a 25ms base delay for up to 5 retries for a total cumulative delay of 775ms +mkDatadogScribeSettings :: C.ConnectionParams -> DatadogAuth -> IO DatadogScribeSettings +mkDatadogScribeSettings connectionParams auth = do + capabilities <- Conc.getNumCapabilities + pure $ DatadogScribeSettings + { datadogScribeSettings_connectionParams = connectionParams + , datadogScribeSettings_auth = auth + , datadogScribeSettings_poolStripes = 1 + , datadogScribeSettings_connsPerStripe = capabilities + , datadogScribeSettings_connectionKeepalive = 30 + , datadogScribeSettings_retry = Retry.exponentialBackoff 25000 <> Retry.limitRetries 5 + } + + +------------------------------------------------------------------------------- +mkDatadogScribe + :: DatadogScribeSettings + -> PermitFunc + -- ^ Function on whether to permit items, e.g. @permitItem InfoS@ + -> Verbosity + -- ^ Verbosity level to observe + -> IO Scribe +mkDatadogScribe settings permit verb = do + connectionContext <- C.initConnectionContext + pool <- Pool.createPool + (C.connectTo connectionContext (datadogScribeSettings_connectionParams settings)) + C.connectionClose + (datadogScribeSettings_poolStripes settings) + (datadogScribeSettings_connectionKeepalive settings) + (datadogScribeSettings_connsPerStripe settings) + pure $ Scribe + { liPush = pushPool (datadogScribeSettings_retry settings) pool keyBuilder verb + , scribeFinalizer = Pool.destroyAllResources pool + , scribePermitItem = permit + } + where + !keyBuilder = case datadogScribeSettings_auth settings of + NoAuthLocal -> Nothing + DirectAuth (APIKey k) -> Just (BB.fromByteString (T.encodeUtf8 k)) + + +pushPool + :: LogItem a + => Retry.RetryPolicyM IO + -> Pool.Pool C.Connection + -> Maybe BB.Builder + -- ^ Rendered API key + -> Verbosity + -> Item a + -> IO () +pushPool retryPolicy pool token verb item = + void $ Retry.retrying retryPolicy (\_status shouldRetry -> pure shouldRetry) $ \_status -> do + res <- EX.tryAny $ Pool.withResource pool $ \conn -> do + C.connectionPut conn rendered + pure $ case res of + Left _ -> True + Right _ -> False + where + payloadBuilder = A.fromEncoding (encodeDatadog verb item) Monoid.<> "\n" + messageBuilder = case token of + --TODO: lots of conversion going on here. could use a bench + Nothing -> payloadBuilder + Just directKey -> directKey <> " " <> payloadBuilder + rendered = BSL.toStrict (BB.toLazyByteString messageBuilder) + + +encodeDatadog :: LogItem a => Verbosity -> Item a -> A.Encoding +encodeDatadog verb i = A.pairs $ + -- datadog seems to use "service" to denote appname + "service" A..= _itemApp i <> + -- datadog uses severity (or some other alternatives for sev) + "severity" A..= _itemSeverity i <> + "env" A..= _itemEnv i <> + "thread" A..= getThreadIdText (_itemThread i) <> + "host" A..= _itemHost i <> + "pid" A..= pidInt <> + "data" A..= payloadObject verb (_itemPayload i) <> + -- datadog uses "message" instead of msg + "message" A..= (B.toLazyText (unLogStr (_itemMessage i))) <> + -- datadog uses "date" instead of "at" + "date" A..= _itemTime i <> + "ns" A..= _itemNamespace i <> + "loc" A..= fmap LocJs (_itemLoc i) + where + POSIX.CPid pidInt = _itemProcess i diff --git a/test/Katip/Tests/Scribes/Datadog/TCP.hs b/test/Katip/Tests/Scribes/Datadog/TCP.hs new file mode 100644 --- /dev/null +++ b/test/Katip/Tests/Scribes/Datadog/TCP.hs @@ -0,0 +1,105 @@ +{-# LANGUAGE OverloadedStrings #-} +module Katip.Tests.Scribes.Datadog.TCP + ( tests + ) where + + +------------------------------------------------------------------------------- +import Control.Concurrent.Async +import qualified Control.Exception.Safe as EX +import qualified Data.Aeson as A +import qualified Data.Char as Char +import qualified Data.Conduit as C +import qualified Data.Conduit.Binary as CB +import qualified Data.Conduit.List as CL +import qualified Data.Conduit.Network as CN +import Data.Foldable (toList) +import qualified Data.HashMap.Strict as HM +import Data.IORef +import Data.Sequence ((|>)) +import Data.Text (Text) +import qualified Data.Text as T +import Test.Tasty +import Test.Tasty.HUnit +------------------------------------------------------------------------------- +import Katip +import Katip.Scribes.Datadog.TCP +------------------------------------------------------------------------------- + + + +tests :: TestTree +tests = testGroup "Katip.Scribes.Datadog.TCP" + [ testCase "logs well-formed messages in order" $ do + logs <- collectLogs $ do + logItem (sl "foo" ("bar" :: Text)) "mynamespace" noLoc InfoS "a message" + length logs @?= 1 + let (CapturedLog ns dta sev msg thread) = head logs + ns @?= Namespace ["katip-datadog-tests", "mynamespace"] + dta @?= A.Object (HM.singleton "foo" (A.String "bar")) + sev @?= InfoS + msg @?= "a message" + if T.all Char.isNumber thread && T.length thread > 0 + then pure () + else assertFailure ("Expected thread id to just be a number but it was " <> T.unpack thread) + ] + where + noLoc = Nothing + + +------------------------------------------------------------------------------- +-- | A log type parsed out of the fake datadog agent's buffer +data CapturedLog = CapturedLog + { capturedLog_namespace :: Namespace + , capturedLog_data :: A.Value + , capturedLog_sev :: Severity + , capturedLog_message :: Text + , capturedLog_thread :: Text + } deriving (Show, Eq) + + +instance A.FromJSON CapturedLog where + parseJSON = A.withObject "CapturedLog" $ \o -> do + message <- o A..: "message" + sev <- o A..: "severity" + dta <- o A..: "data" + namespace <- o A..: "ns" + thread <- o A..: "thread" + pure $ CapturedLog + { capturedLog_namespace = namespace + , capturedLog_data = dta + , capturedLog_sev = sev + , capturedLog_message = message + , capturedLog_thread = thread + } + + +------------------------------------------------------------------------------- +collectLogs :: KatipT IO () -> IO [CapturedLog] +collectLogs f = do + capturedRef <- newIORef mempty + withAsync (startServer capturedRef) $ \_worker -> do + EX.bracket mkLE closeLE $ \le -> do + runKatipT le f + toList <$> readIORef capturedRef + where + closeLE = closeScribes + port = 1337 + mkLE = do + datadogScribeSettings <- mkDatadogScribeSettings (localAgentConnectionParams (fromInteger port)) NoAuthLocal + scribe <- mkDatadogScribe datadogScribeSettings (permitItem DebugS) V3 + registerScribe "datadog" scribe defaultScribeSettings =<< initLogEnv "katip-datadog-tests" "test" + startServer capturedRef = CN.runTCPServer (CN.serverSettings (fromInteger port) "*") $ \appData -> C.runConduit $ + CN.appSource appData C..| + CB.lines C..| + CL.map A.eitherDecodeStrict C..| + CL.mapM_ (\parseResult -> case parseResult of + Left e -> EX.throwM (ParseError e) + (Right capturedLog) -> modifyIORef' capturedRef (\captured -> captured |> capturedLog)) + + +------------------------------------------------------------------------------- +newtype ParseError = ParseError String + deriving (Show) + +instance EX.Exception ParseError 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.Datadog.TCP +------------------------------------------------------------------------------- + + + +main :: IO () +main = defaultMain tests + + +------------------------------------------------------------------------------- +tests :: TestTree +tests = testGroup "katip-datadog" + [ Katip.Tests.Scribes.Datadog.TCP.tests + ]