@@ -1,10 +1,3 @@
1−{-# LANGUAGE DeriveDataTypeable #-}
2−{-# LANGUAGE FlexibleContexts #-}
3−{-# LANGUAGE OverloadedStrings #-}
4−{-# LANGUAGE RankNTypes #-}
5−{-# LANGUAGE RecordWildCards #-}
6−{-# LANGUAGE ScopedTypeVariables #-}
7−{-# LANGUAGE TypeFamilies #-}
81-- | Includes a scribe that can be used to log structured, JSON log
92-- messages to ElasticSearch. These logs can be explored easily using
103-- <https://www.elastic.co/products/kibana kibana> or your tool of
@@ -44,8 +37,6 @@ module Katip.Scribes.ElasticSearch
4437 , defaultEsScribeCfg
4538 , defaultEsScribeCfgV1
4639 , defaultEsScribeCfgV5
47− , ESV1
48− , ESV5
4940 -- * Utilities
5041 , mkDocId
5142 , module Katip.Scribes.ElasticSearch.Annotations
@@ -54,488 +45,6 @@ module Katip.Scribes.ElasticSearch
5445
5546
5647-------------------------------------------------------------------------------
57−import Control.Applicative as A
58−import Control.Concurrent
59−import Control.Concurrent.Async
60−import Control.Concurrent.STM.TBMQueue
61−import Control.Exception.Base
62−import Control.Exception.Enclosed
63−import Control.Monad
64−import Control.Monad.Catch
65−import Control.Monad.IO.Class
66−import Control.Monad.STM
67−import Control.Retry (RetryPolicy,
68− exponentialBackoff,
69− limitRetries,
70− recovering)
71−import Data.Aeson
72−import Data.ByteString.Lazy (ByteString)
73−import Data.Monoid ((<>))
74−import Data.Text (Text)
75−import qualified Data.Text as T
76−import qualified Data.Text.Encoding as T
77−import Data.Time
78−import Data.Time.Calendar.WeekDate
79−import Data.Typeable as Typeable
80−import Data.UUID
81−import qualified Data.UUID.V4 as UUID4
82−import qualified Database.V1.Bloodhound as V1
83−import qualified Database.V5.Bloodhound as V5
84−import Network.HTTP.Client
85−import Network.HTTP.Types.Status
86−import Text.Printf (printf)
87−-------------------------------------------------------------------------------
88−import Katip.Core
8948import Katip.Scribes.ElasticSearch.Annotations
49+import Katip.Scribes.ElasticSearch.Internal
9050-------------------------------------------------------------------------------
91−
92−
93−data EsScribeCfg v = EsScribeCfg {
94− essRetryPolicy :: RetryPolicy
95− -- ^ Retry policy when there are errors sending logs to the server
96− , essQueueSize :: EsQueueSize
97− -- ^ Maximum size of the bounded log queue
98− , essPoolSize :: EsPoolSize
99− -- ^ Worker pool size limit for sending data to the
100− , essAnnotateTypes :: Bool
101− -- ^ Different payload items coexist in the "data" attribute in
102− -- ES. It is possible for different payloads to have different
103− -- types for the same key, e.g. an "id" key that is sometimes a
104− -- number and sometimes a string. If you're having ES do dynamic
105− -- mapping, the first log item will set the type and any that
106− -- don't conform will be *discarded*. If you set this to True,
107− -- keys will recursively be appended with their ES core
108− -- type. e.g. "id" would become "id::l" and "id::s"
109− -- automatically, so they won't conflict. When this library
110− -- exposes a querying API, we will try to make deserialization and
111− -- querying transparently remove the type annotations if this is
112− -- enabled.
113− , essIndexSettings :: IndexSettings v
114− , essIndexSharding :: IndexShardingPolicy
115− } deriving (Typeable)
116−
117−
118−-- | Reasonable defaults for a config:
119−--
120−-- * defaultManagerSettings
121−--
122−-- * exponential backoff with 25ms base delay up to 5 retries
123−--
124−-- * Queue size of 1000
125−--
126−-- * Pool size of 2
127−--
128−-- * Annotate types set to False
129−--
130−-- * DailyIndexSharding
131−defaultEsScribeCfg :: forall v. ESVersion v => EsScribeCfg v
132−defaultEsScribeCfg = EsScribeCfg {
133− essRetryPolicy = exponentialBackoff 25 <> limitRetries 5
134− , essQueueSize = EsQueueSize 1000
135− , essPoolSize = EsPoolSize 2
136− , essAnnotateTypes = False
137− , essIndexSettings = defaultIndexSettings prx
138− , essIndexSharding = DailyIndexSharding
139− }
140− where
141− prx :: Typeable.Proxy v
142− prx = Typeable.Proxy
143−
144−
145−-------------------------------------------------------------------------------
146−-- | EsScribeCfg that will use ElasticSearch V1
147−defaultEsScribeCfgV1 :: EsScribeCfg ESV1
148−defaultEsScribeCfgV1 = defaultEsScribeCfg
149−
150−
151−-------------------------------------------------------------------------------
152−-- | EsScribeCfg that will use ElasticSearch V5
153−defaultEsScribeCfgV5 :: EsScribeCfg ESV5
154−defaultEsScribeCfgV5 = defaultEsScribeCfg
155−
156−
157−-------------------------------------------------------------------------------
158−-- | How should katip store your log data?
159−--
160−-- * NoIndexSharding will store all logs in one index name. This is
161−-- the simplest option but is not advised in production. In practice,
162−-- the index will grow very large and will get slower to
163−-- search. Deleting records based on some sort of retention period is
164−-- also extremely slow.
165−--
166−-- * MonthlyIndexSharding, DailyIndexSharding, HourlyIndexSharding,
167−-- EveryMinuteIndexSharding will generate indexes based on the time of
168−-- the log. Index name is treated as a prefix. So if your index name
169−-- is @foo@ and DailySharding is used, logs will be stored in
170−-- @foo-2016-2-25@, @foo-2016-2-26@ and so on. Index templating will
171−-- be used to set up mappings automatically. Deletes based on date are
172−-- very fast and queries can be restricted to date ranges for better
173−-- performance. Queries against all dates should use @foo-*@ as an
174−-- index name. Note that index aliasing's glob feature is not suitable
175−-- for these date ranges as it matches index names as they are
176−-- declared, so new dates will be excluded. DailyIndexSharding is a
177−-- reasonable choice. Changing index sharding strategies is not
178−-- advisable.
179−--
180−-- * CustomSharding: supply your own function that decomposes an item
181−-- into its index name hierarchy which will be appended to the index
182−-- name. So for instance if your function return ["arbitrary",
183−-- "prefix"], the index will be @foo-arbitrary-prefix@ and the index
184−-- template will be set to match @foo-*@. In general, you want to use
185−-- segments of increasing granularity (like year, month, day for
186−-- dates). This makes it easier to address groups of indexes
187−-- (e.g. @foo-2016-*@).
188−data IndexShardingPolicy = NoIndexSharding
189− | MonthlyIndexSharding
190− | WeeklyIndexSharding
191− -- ^ A special case of daily which shards to sunday
192− | DailyIndexSharding
193− | HourlyIndexSharding
194− | EveryMinuteIndexSharding
195− | CustomIndexSharding (forall a. Item a -> [IndexNameSegment])
196−
197−
198−instance Show IndexShardingPolicy where
199− show NoIndexSharding = "NoIndexSharding"
200− show MonthlyIndexSharding = "MonthlyIndexSharding"
201− show WeeklyIndexSharding = "WeeklyIndexSharding"
202− show DailyIndexSharding = "DailyIndexSharding"
203− show HourlyIndexSharding = "HourlyIndexSharding"
204− show EveryMinuteIndexSharding = "EveryMinuteIndexSharding"
205− show (CustomIndexSharding _) = "CustomIndexSharding λ"
206−
207−
208−-------------------------------------------------------------------------------
209−newtype IndexNameSegment = IndexNameSegment {
210− indexNameSegment :: Text
211− } deriving (Show, Eq, Ord)
212−
213−
214−-------------------------------------------------------------------------------
215−shardPolicySegs :: IndexShardingPolicy -> Item a -> [IndexNameSegment]
216−shardPolicySegs NoIndexSharding _ = []
217−shardPolicySegs MonthlyIndexSharding Item {..} = [sis y, sis m]
218− where
219− (y, m, _) = toGregorian (utctDay _itemTime)
220−shardPolicySegs WeeklyIndexSharding Item {..} = [sis y, sis m, sis d]
221− where
222− (y, m, d) = toGregorian (roundToSunday (utctDay _itemTime))
223−shardPolicySegs DailyIndexSharding Item {..} = [sis y, sis m, sis d]
224− where
225− (y, m, d) = toGregorian (utctDay _itemTime)
226−shardPolicySegs HourlyIndexSharding Item {..} = [sis y, sis m, sis d, sis h]
227− where
228− (y, m, d) = toGregorian (utctDay _itemTime)
229− (h, _) = splitTime (utctDayTime _itemTime)
230−shardPolicySegs EveryMinuteIndexSharding Item {..} = [sis y, sis m, sis d, sis h, sis mn]
231− where
232− (y, m, d) = toGregorian (utctDay _itemTime)
233− (h, mn) = splitTime (utctDayTime _itemTime)
234−shardPolicySegs (CustomIndexSharding f) i = f i
235−
236−
237−-------------------------------------------------------------------------------
238−-- | If the given day is sunday, returns the input, otherwise returns
239−-- the previous sunday
240−roundToSunday :: Day -> Day
241−roundToSunday d
242− | dow == 7 = d
243− | w > 1 = fromWeekDate y (w - 1) 7
244− | otherwise = fromWeekDate (y - 1) 53 7
245− where
246− (y, w, dow) = toWeekDate d
247−
248−
249−-------------------------------------------------------------------------------
250−chooseIxn :: ESVersion v => proxy v -> IndexName v -> IndexShardingPolicy -> Item a -> IndexName v
251−chooseIxn prx ixn p i =
252− toIndexName prx (T.intercalate "-" ((fromIndexName prx ixn):segs))
253− where
254− segs = indexNameSegment A.<$> shardPolicySegs p i
255−
256−
257−-------------------------------------------------------------------------------
258−sis :: Integral a => a -> IndexNameSegment
259−sis = IndexNameSegment . T.pack . fmt
260− where
261− fmt = printf "%02d" . toInteger
262−
263−
264−-------------------------------------------------------------------------------
265−splitTime :: DiffTime -> (Int, Int)
266−splitTime t = asMins `divMod` 60
267− where
268− asMins = floor t `div` 60
269−
270−
271−-------------------------------------------------------------------------------
272−data EsScribeSetupError = CouldNotCreateIndex !(Response ByteString)
273− | CouldNotCreateMapping !(Response ByteString) deriving (Typeable, Show)
274−
275−
276−instance Exception EsScribeSetupError
277−
278−
279−-------------------------------------------------------------------------------
280−mkEsScribe
281− :: forall v. ( ESVersion v
282− , MonadIO (BH v IO)
283− )
284− => EsScribeCfg v
285− -> BHEnv v
286− -> IndexName v
287− -- ^ Treated as a prefix if index sharding is enabled
288− -> MappingName v
289− -> Severity
290− -> Verbosity
291− -> IO Scribe
292−mkEsScribe cfg@EsScribeCfg {..} env ix mapping sev verb = do
293− q <- newTBMQueueIO $ unEsQueueSize essQueueSize
294− endSig <- newEmptyMVar
295−
296− runBH prx env $ do
297− chk <- indexExists prx ix
298− -- note that this doesn't update settings. That's not available
299− -- through the Bloodhound API yet
300− unless chk $ void $ do
301− r1 <- createIndex prx essIndexSettings ix
302− unless (statusIsSuccessful (responseStatus r1)) $
303− liftIO $ throwIO (CouldNotCreateIndex r1)
304− r2 <- if shardingEnabled
305− then putTemplate prx tpl tplName
306− else putMapping prx ix mapping base
307− unless (statusIsSuccessful (responseStatus r2)) $
308− liftIO $ throwIO (CouldNotCreateMapping r2)
309−
310− workers <- replicateM (unEsPoolSize essPoolSize) $ async $
311− startWorker cfg env mapping q
312−
313− _ <- async $ do
314− takeMVar endSig
315− atomically $ closeTBMQueue q
316− mapM_ waitCatch workers
317− putMVar endSig ()
318−
319− let finalizer = putMVar endSig () >> takeMVar endSig
320− return (Scribe (logger q) finalizer)
321− where
322− logger :: forall a. LogItem a => TBMQueue (IndexName v, Value) -> Item a -> IO ()
323− logger q i = when (_itemSeverity i >= sev) $
324− void $ atomically $ tryWriteTBMQueue q (chooseIxn prx ix essIndexSharding i, itemJson' i)
325− prx :: Typeable.Proxy v
326− prx = Typeable.Proxy
327− tplName = toTemplateName prx ixn
328− shardingEnabled = case essIndexSharding of
329− NoIndexSharding -> False
330− _ -> True
331− tpl = toIndexTemplate prx (toTemplatePattern prx (ixn <> "-*")) (Just essIndexSettings) [toJSON base]
332− base = baseMapping prx mapping
333− ixn = fromIndexName prx ix
334− itemJson' :: LogItem a => Item a -> Value
335− itemJson' i
336− | essAnnotateTypes = itemJson verb (TypeAnnotated <$> i)
337− | otherwise = itemJson verb i
338−
339−
340−-------------------------------------------------------------------------------
341−baseMapping :: ESVersion v => proxy v -> MappingName v -> Value
342−baseMapping prx mn =
343− object [ fromMappingName prx mn .= object ["properties" .= object prs] ]
344− where prs = [ str "thread"
345− , str "sev"
346− , str "pid"
347− , str "ns"
348− , str "msg"
349− , "loc" .= locType
350− , str "host"
351− , str "env"
352− , "at" .= dateType
353− , str "app"
354− ]
355− str k = k .= object ["type" .= String "string"]
356− locType = object ["properties" .= object locPairs]
357− locPairs = [ str "loc_pkg"
358− , str "loc_mod"
359− , str "loc_ln"
360− , str "loc_fn"
361− , str "loc_col"
362− ]
363− dateType = object [ "format" .= esDateFormat
364− , "type" .= String "date"
365− ]
366−
367−
368−-------------------------------------------------------------------------------
369−-- | Handle both old-style aeson and picosecond-level precision
370−esDateFormat :: Text
371−esDateFormat = "yyyy-MM-dd'T'HH:mm:ssZ||yyyy-MM-dd'T'HH:mm:ss.SSSZ||yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSSSSSZ"
372−
373−
374−-------------------------------------------------------------------------------
375−mkDocId :: ESVersion v => proxy v -> IO (DocId v)
376−mkDocId prx = (toDocId prx . T.decodeUtf8 . toASCIIBytes) `fmap` UUID4.nextRandom
377−
378−
379−-------------------------------------------------------------------------------
380−newtype EsQueueSize = EsQueueSize {
381− unEsQueueSize :: Int
382− } deriving (Show, Eq, Ord)
383−
384−
385−instance Bounded EsQueueSize where
386− minBound = EsQueueSize 1
387− maxBound = EsQueueSize maxBound
388−
389−
390−mkEsQueueSize :: Int -> Maybe EsQueueSize
391−mkEsQueueSize = mkNonZero EsQueueSize
392−
393−
394−-------------------------------------------------------------------------------
395−newtype EsPoolSize = EsPoolSize {
396− unEsPoolSize :: Int
397− } deriving (Show, Eq, Ord)
398−
399−
400−instance Bounded EsPoolSize where
401− minBound = EsPoolSize 1
402− maxBound = EsPoolSize maxBound
403−
404−
405−mkEsPoolSize :: Int -> Maybe EsPoolSize
406−mkEsPoolSize = mkNonZero EsPoolSize
407−
408−
409−-------------------------------------------------------------------------------
410−mkNonZero :: (Int -> a) -> Int -> Maybe a
411−mkNonZero ctor n
412− | n > 0 = Just $ ctor n
413− | otherwise = Nothing
414−
415−
416−-------------------------------------------------------------------------------
417−startWorker
418− :: forall v. (ESVersion v)
419− => EsScribeCfg v
420− -> BHEnv v
421− -> MappingName v
422− -> TBMQueue (IndexName v, Value)
423− -> IO ()
424−startWorker EsScribeCfg {..} env mapping q = go
425− where
426− go = do
427− popped <- atomically $ readTBMQueue q
428− case popped of
429− Just (ixn, v) -> do
430− sendLog ixn v `catchAny` eat
431− go
432− Nothing -> return ()
433− prx :: Typeable.Proxy v
434− prx = Typeable.Proxy
435− sendLog :: IndexName v -> Value -> IO ()
436− sendLog ixn v = void $ recovering essRetryPolicy [handler] $ const $ do
437− did <- mkDocId prx
438− res <- runBH prx env $ indexDocument prx ixn mapping (defaultIndexDocumentSettings prx) v did
439− return res
440− eat _ = return ()
441− handler _ = Handler $ \e ->
442− case fromException e of
443− Just (_ :: AsyncException) -> return False
444− _ -> return True
445−
446−
447−-------------------------------------------------------------------------------
448−-- We are spanning multiple versions of ES which use completely
449−-- separate types and APIs, but the subset we use is the same for both
450−-- versions. This will be kept up to date with bloodhound's supported
451−-- versions and should be minimally visible to the end user.
452−class ESVersion v where
453− -- Types
454− type BHEnv v
455− type IndexSettings v
456− defaultIndexSettings :: proxy v -> IndexSettings v
457− type IndexName v
458− toIndexName :: proxy v -> Text -> IndexName v
459− fromIndexName :: proxy v -> IndexName v -> Text
460− type MappingName v
461− fromMappingName :: proxy v -> MappingName v -> Text
462− type DocId v
463− toDocId :: proxy v -> Text -> DocId v
464− type BH v :: (* -> *) -> * -> *
465− runBH :: proxy v -> BHEnv v -> BH v m a -> m a
466− type TemplateName v
467− toTemplateName :: proxy v -> Text -> TemplateName v
468− type TemplatePattern v
469− toTemplatePattern :: proxy v -> Text -> TemplatePattern v
470− type IndexTemplate v
471− toIndexTemplate :: proxy v -> TemplatePattern v -> Maybe (IndexSettings v) -> [Value] -> IndexTemplate v
472− type IndexDocumentSettings v
473− defaultIndexDocumentSettings :: proxy v -> IndexDocumentSettings v
474−
475− -- Operations
476− -- We're deciding on IO here, but it isn't necessary
477− indexExists :: proxy v -> IndexName v -> BH v IO Bool
478− indexDocument :: ToJSON doc => proxy v -> IndexName v -> MappingName v -> IndexDocumentSettings v -> doc -> DocId v -> BH v IO (Response ByteString)
479− createIndex :: proxy v -> IndexSettings v -> IndexName v -> BH v IO (Response ByteString)
480− putTemplate :: proxy v -> IndexTemplate v -> TemplateName v -> BH v IO (Response ByteString)
481− putMapping :: (ToJSON a) => proxy v -> IndexName v -> MappingName v -> a -> BH v IO (Response ByteString)
482−
483−
484−data ESV1
485−
486−instance ESVersion ESV1 where
487− type BHEnv ESV1 = V1.BHEnv
488− type IndexSettings ESV1 = V1.IndexSettings
489− defaultIndexSettings _ = V1.defaultIndexSettings
490− type IndexName ESV1 = V1.IndexName
491− toIndexName _ = V1.IndexName
492− fromIndexName _ (V1.IndexName x) = x
493− type MappingName ESV1 = V1.MappingName
494− fromMappingName _ (V1.MappingName x) = x
495− type DocId ESV1 = V1.DocId
496− toDocId _ = V1.DocId
497− type BH ESV1 = V1.BH
498− runBH _ = V1.runBH
499− type TemplateName ESV1 = V1.TemplateName
500− toTemplateName _ = V1.TemplateName
501− type TemplatePattern ESV1 = V1.TemplatePattern
502− toTemplatePattern _ = V1.TemplatePattern
503− type IndexTemplate ESV1 = V1.IndexTemplate
504− toIndexTemplate _ = V1.IndexTemplate
505− type IndexDocumentSettings ESV1 = V1.IndexDocumentSettings
506− defaultIndexDocumentSettings _ = V1.defaultIndexDocumentSettings
507− indexExists _ = V1.indexExists
508− indexDocument _ = V1.indexDocument
509− createIndex _ = V1.createIndex
510− putTemplate _ = V1.putTemplate
511− putMapping _ = V1.putMapping
512−
513−
514−data ESV5
515−
516−instance ESVersion ESV5 where
517− type BHEnv ESV5 = V5.BHEnv
518− type IndexSettings ESV5 = V5.IndexSettings
519− defaultIndexSettings _ = V5.defaultIndexSettings
520− type IndexName ESV5 = V5.IndexName
521− toIndexName _ = V5.IndexName
522− fromIndexName _ (V5.IndexName x) = x
523− type MappingName ESV5 = V5.MappingName
524− fromMappingName _ (V5.MappingName x) = x
525− type DocId ESV5 = V5.DocId
526− toDocId _ = V5.DocId
527− type BH ESV5 = V5.BH
528− runBH _ = V5.runBH
529− type TemplateName ESV5 = V5.TemplateName
530− toTemplateName _ = V5.TemplateName
531− type TemplatePattern ESV5 = V5.TemplatePattern
532− toTemplatePattern _ = V5.TemplatePattern
533− type IndexTemplate ESV5 = V5.IndexTemplate
534− toIndexTemplate _ = V5.IndexTemplate
535− type IndexDocumentSettings ESV5 = V5.IndexDocumentSettings
536− defaultIndexDocumentSettings _ = V5.defaultIndexDocumentSettings
537− indexExists _ = V5.indexExists
538− indexDocument _ = V5.indexDocument
539− createIndex _ = V5.createIndex
540− putTemplate _ = V5.putTemplate
541− putMapping _ = V5.putMapping