@@ -0,0 +1,192 @@
1+-- | Creates a scribe as a Custom Forwarder for Datadog that sends log
2+-- messages over TCP via their TLS endpoint.
3+module Katip.Scribes.Datadog.TCP
4+ ( -- * Types
5+ APIKey(..)
6+ , DataDogScribeSettings(..)
7+ , DataDogAuth(..)
8+ , directAPIConnectionParams
9+ , localAgentConnectionParams
10+ -- * Scribe construction
11+ , mkDataDogScribeSettings
12+ , mkDataDogScribe
13+ ) where
14+
15+
16+
17+-------------------------------------------------------------------------------
18+import qualified Control.Concurrent as Conc
19+import qualified Control.Exception.Safe as EX
20+import Control.Monad (void)
21+import qualified Control.Retry as Retry
22+import qualified Data.Aeson as A
23+import qualified Data.Binary.Builder as BB
24+import qualified Data.ByteString.Lazy as BSL
25+import Data.Monoid
26+import qualified Data.Pool as Pool
27+import Data.Text (Text)
28+import qualified Data.Text.Encoding as T
29+import qualified Data.Text.Lazy.Builder as B
30+import Data.Time (NominalDiffTime)
31+import Debug.Trace --TODO: drop
32+import Katip
33+import Katip.Core (LocJs (..))
34+import qualified Network as Net
35+import qualified Network.Connection as C
36+import qualified System.Posix.Types as POSIX
37+-------------------------------------------------------------------------------
38+
39+
40+-- | When writing directly to the main intake API, you must specify an
41+-- API key
42+newtype APIKey = APIKey
43+ { apiKey :: Text
44+ } deriving (Show, Eq)
45+
46+
47+--TODO: retry policy
48+
49+data DataDogAuth =
50+ NoAuthLocal
51+ -- ^ Writing to the local agent, no auth required
52+ | DirectAuth APIKey
53+ -- ^ Writing directly to the main intake API over TLS . API Key is
54+ -- required. Using a local agent is recommended.
55+
56+
57+data DataDogScribeSettings = DataDogScribeSettings
58+ { dataDogScribeSettings_connectionParams :: C.ConnectionParams
59+ -- ^ Specify where the logs should go. If writing directly to the
60+ -- main intake API, you can use 'directAPIConnectionParams'. If
61+ -- writing to a local agent, we recommend
62+ -- 'localAgentConnectionParams'
63+ , dataDogScribeSettings_auth :: DataDogAuth
64+ , dataDogScribeSettings_poolStripes :: Int
65+ -- ^ How many stripes should be used in the connection pool. 1 is a reasonable default
66+ , dataDogScribeSettings_connsPerStripe :: Int
67+ -- ^ How many TCP connections per stripe should be maintained in the pool.
68+ , dataDogScribeSettings_connectionKeepalive :: NominalDiffTime
69+ -- ^ 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.
70+ , dataDogScribeSettings_retry :: Retry.RetryPolicyM IO
71+ -- ^ How should exceptions during write be retried?
72+ }
73+
74+
75+-- | Configured for secure TLS communication to the main intake API at
76+-- intake.logs.datadoghq.com:10516. Must be combined with DirectAuth
77+-- authentication.
78+directAPIConnectionParams :: C.ConnectionParams
79+directAPIConnectionParams = C.ConnectionParams
80+ { connectionHostname = "intake.logs.datadoghq.com"
81+ , connectionPort = 10516
82+ , connectionUseSecure = Just $ C.TLSSettingsSimple
83+ { settingDisableCertificateValidation = False
84+ , settingDisableSession = False
85+ , settingUseServerName = False
86+ }
87+ , connectionUseSocks = Nothing
88+ }
89+
90+
91+-- | Configure to talk to a local DataDog agent at 127.0.0.1. TLS is not
92+-- utilized. This is the preferred type of connection because
93+-- authentication, buffering, and other more advanced features are
94+-- handled by the agent configuration.
95+localAgentConnectionParams :: Net.PortNumber -> C.ConnectionParams
96+localAgentConnectionParams port = C.ConnectionParams
97+ { connectionHostname = "127.0.0.1"
98+ , connectionPort = port
99+ , connectionUseSecure = Nothing
100+ , connectionUseSocks = Nothing
101+ }
102+
103+
104+-- | Reasonable defaults. Sets a 30s connection timeout, 1 stripe of
105+-- connections with a number of connections per stripe equal to
106+-- 'getNumCapabilities'. Retry policy will do an exponential backoff
107+-- with a 25ms base delay for up to 5 retries for a total cumulative delay of 775ms
108+mkDataDogScribeSettings :: C.ConnectionParams -> DataDogAuth -> IO DataDogScribeSettings
109+mkDataDogScribeSettings connectionParams auth = do
110+ capabilities <- Conc.getNumCapabilities
111+ pure $ DataDogScribeSettings
112+ { dataDogScribeSettings_connectionParams = connectionParams
113+ , dataDogScribeSettings_auth = auth
114+ , dataDogScribeSettings_poolStripes = 1
115+ , dataDogScribeSettings_connsPerStripe = capabilities
116+ , dataDogScribeSettings_connectionKeepalive = 30
117+ , dataDogScribeSettings_retry = Retry.exponentialBackoff 25000 <> Retry.limitRetries 5
118+ }
119+
120+
121+-------------------------------------------------------------------------------
122+mkDataDogScribe
123+ :: DataDogScribeSettings
124+ -> Severity
125+ -- ^ Severity level to observe
126+ -> Verbosity
127+ -- ^ Verbosity level to observe
128+ -> IO Scribe
129+mkDataDogScribe settings sev verb = do
130+ connectionContext <- C.initConnectionContext
131+ pool <- Pool.createPool
132+ (C.connectTo connectionContext (dataDogScribeSettings_connectionParams settings))
133+ C.connectionClose
134+ (dataDogScribeSettings_poolStripes settings)
135+ (dataDogScribeSettings_connectionKeepalive settings)
136+ (dataDogScribeSettings_connsPerStripe settings)
137+ pure $ Scribe
138+ { liPush = pushPool (dataDogScribeSettings_retry settings) pool keyBuilder sev verb
139+ , scribeFinalizer = Pool.destroyAllResources pool
140+ }
141+ where
142+ !keyBuilder = case dataDogScribeSettings_auth settings of
143+ NoAuthLocal -> Nothing
144+ DirectAuth (APIKey k) -> Just (BB.fromByteString (T.encodeUtf8 k))
145+
146+
147+pushPool
148+ :: LogItem a
149+ => Retry.RetryPolicyM IO
150+ -> Pool.Pool C.Connection
151+ -> Maybe BB.Builder
152+ -- ^ Rendered API key
153+ -> Severity
154+ -> Verbosity
155+ -> Item a
156+ -> IO ()
157+pushPool retryPolicy pool token sev verb item
158+ | permitItem sev item = void $ Retry.retrying retryPolicy (\_status shouldRetry -> pure (traceShowId shouldRetry)) $ \_status -> do
159+ res <- EX.tryAny $ Pool.withResource pool $ \conn -> do
160+ C.connectionPut conn rendered
161+ pure $ case res of
162+ Left e -> traceShow e True
163+ Right _ -> False
164+ | otherwise = pure ()
165+ where
166+ payloadBuilder = A.fromEncoding (encodeDatadog verb item) <> "\n"
167+ messageBuilder = case token of
168+ --TODO: lots of conversion going on here. could use a bench
169+ Nothing -> payloadBuilder
170+ Just directKey -> directKey <> " " <> payloadBuilder
171+ rendered = BSL.toStrict (BB.toLazyByteString messageBuilder)
172+
173+
174+encodeDatadog :: LogItem a => Verbosity -> Item a -> A.Encoding
175+encodeDatadog verb i = A.pairs $
176+ -- datadog seems to use "service" to denote appname
177+ "service" A..= _itemApp i <>
178+ -- datadog uses severity (or some other alternatives for sev)
179+ "severity" A..= _itemSeverity i <>
180+ "env" A..= _itemEnv i <>
181+ "thread" A..= getThreadIdText (_itemThread i) <>
182+ "host" A..= _itemHost i <>
183+ "pid" A..= pidInt <>
184+ "data" A..= payloadObject verb (_itemPayload i) <>
185+ -- datadog uses "message" instead of msg
186+ "message" A..= (B.toLazyText (unLogStr (_itemMessage i))) <>
187+ -- datadog uses "date" instead of "at"
188+ "date" A..= _itemTime i <>
189+ "ns" A..= _itemNamespace i <>
190+ "loc" A..= fmap LocJs (_itemLoc i)
191+ where
192+ POSIX.CPid pidInt = _itemProcess i