@@ -28,7 +28,7 @@ import Control.Retry (RetryPolicy,
2828import Data.Aeson
2929import Data.ByteString.Lazy (ByteString)
3030import Data.List.NonEmpty (NonEmpty (..))
31−import Data.Monoid ((<>))
31+import Data.Monoid as Monoid
3232import Data.Text (Text)
3333import qualified Data.Text as T
3434import qualified Data.Text.Encoding as T
@@ -37,8 +37,12 @@ import Data.Time.Calendar.WeekDate
3737import Data.Typeable as Typeable
3838import Data.UUID
3939import qualified Data.UUID.V4 as UUID4
40+#if MIN_VERSION_bloodhound(0,17,0)
41+import qualified Database.Bloodhound as V5
42+#else
4043import qualified Database.V1.Bloodhound as V1
4144import qualified Database.V5.Bloodhound as V5
45+#endif
4246import Network.HTTP.Client
4347import Network.HTTP.Types.Status
4448import Text.Printf (printf)
@@ -53,26 +57,30 @@ import Katip.Scribes.ElasticSearch.Annotations
5357-- 'defaultEsScribeCfgV1' and 'defaultESScribeCfgV5' for a good
5458-- starting point depending on the ES version you have.
5559data EsScribeCfg v = EsScribeCfg {
56− essRetryPolicy :: RetryPolicy
60+ essRetryPolicy :: RetryPolicy
5761 -- ^ Retry policy when there are errors sending logs to the server
58− , essQueueSize :: EsQueueSize
62+ , essQueueSize :: EsQueueSize
5963 -- ^ Maximum size of the bounded log queue
60− , essPoolSize :: EsPoolSize
64+ , essPoolSize :: EsPoolSize
6165 -- ^ Worker pool size limit for sending data to the
62− , essItemFormatter :: Verbosity -> ItemFunc Value
63− -- ^ The formatter of an item. Must format any item to json
64− -- Value. You can use 'annotateValue' to get an effect of
65− -- @essAnnotateTypes@ (removed) here.
66− , essIndexName :: IndexName v
67− , essIndexMapping :: MappingName v
68− , essIndexMappingValue :: Value
69− -- ^ The value of index mapping. The default is 'baseMapping' but
70− -- user can override.
71− , essIndexSettings :: IndexSettings v
66+ , essAnnotateTypes :: Bool
67+ -- ^ Different payload items coexist in the "data" attribute in
68+ -- ES. It is possible for different payloads to have different
69+ -- types for the same key, e.g. an "id" key that is sometimes a
70+ -- number and sometimes a string. If you're having ES do dynamic
71+ -- mapping, the first log item will set the type and any that
72+ -- don't conform will be *discarded*. If you set this to True,
73+ -- keys will recursively be appended with their ES core
74+ -- type. e.g. "id" would become "id::l" and "id::s"
75+ -- automatically, so they won't conflict. When this library
76+ -- exposes a querying API, we will try to make deserialization and
77+ -- querying transparently remove the type annotations if this is
78+ -- enabled.
79+ , essIndexSettings :: IndexSettings v
7280 -- ^ This will be the IndexSettings type from the appropriate
7381 -- bloodhound module, either @Database.V1.Bloodhound@ or
7482 -- @Database.V5.Bloodhound@
75− , essIndexSharding :: IndexShardingPolicy
83+ , essIndexSharding :: IndexShardingPolicy
7684 } deriving (Typeable)
7785
7886
@@ -89,47 +97,37 @@ data EsScribeCfg v = EsScribeCfg {
8997-- * Annotate types set to False
9098--
9199-- * DailyIndexSharding
92−defaultEsScribeCfg'
93− :: ESVersion v
94− => proxy v
95− -> IndexName v
96− -> MappingName v
97− -> EsScribeCfg v
98−defaultEsScribeCfg' prx index mapping = EsScribeCfg
99− { essRetryPolicy = exponentialBackoff 25000 <> limitRetries 5
100− , essQueueSize = EsQueueSize 1000
101− , essPoolSize = EsPoolSize 2
102− , essItemFormatter = defaultItemFormatter
103− , essIndexName = index
104− , essIndexMapping = mapping
105− , essIndexMappingValue = baseMapping prx mapping
106− , essIndexSettings = defaultIndexSettings prx
107− , essIndexSharding = DailyIndexSharding
108− }
109−
110−defaultItemFormatter :: Verbosity -> ItemFunc Value
111−defaultItemFormatter verb = ItemFunc $ \item -> itemJson verb item
112−
113−annotatedItemFormatter :: Verbosity -> ItemFunc Value
114−annotatedItemFormatter verb = ItemFunc $ \item -> itemJson verb (TypeAnnotated <$> item)
100+defaultEsScribeCfg' :: ESVersion v => proxy v -> EsScribeCfg v
101+defaultEsScribeCfg' prx = EsScribeCfg {
102+ essRetryPolicy = exponentialBackoff 25000 Monoid.<> limitRetries 5
103+ , essQueueSize = EsQueueSize 1000
104+ , essPoolSize = EsPoolSize 2
105+ , essAnnotateTypes = False
106+ , essIndexSettings = defaultIndexSettings prx
107+ , essIndexSharding = DailyIndexSharding
108+ }
109+
110+
111+
115112
116113-------------------------------------------------------------------------------
117−-- | Alias of 'defaultEsScribeCfgV1' to minimize API
114+-- | Alias of 'defaultEsScribeCfgV5' to minimize API
118115-- breakage. Previous versions of katip-elasticsearch only supported
119−-- ES version 1.
120−defaultEsScribeCfg :: IndexName ESV1 -> MappingName ESV1 -> EsScribeCfg ESV1
121−defaultEsScribeCfg = defaultEsScribeCfgV1
116+-- ES version 1 and defaulted to it.
117+defaultEsScribeCfg :: EsScribeCfg ESV5
118+defaultEsScribeCfg = defaultEsScribeCfgV5
122119
123120
124121-------------------------------------------------------------------------------
122+#if !MIN_VERSION_bloodhound(0,17,0)
125123-- | EsScribeCfg that will use ElasticSearch V1
126−defaultEsScribeCfgV1 :: IndexName ESV1 -> MappingName ESV1 -> EsScribeCfg ESV1
124+defaultEsScribeCfgV1 :: EsScribeCfg ESV1
127125defaultEsScribeCfgV1 = defaultEsScribeCfg' (Typeable.Proxy :: Typeable.Proxy ESV1)
128−
126+#endif
129127
130128-------------------------------------------------------------------------------
131129-- | EsScribeCfg that will use ElasticSearch V5
132−defaultEsScribeCfgV5 :: IndexName ESV5 -> MappingName ESV5 -> EsScribeCfg ESV5
130+defaultEsScribeCfgV5 :: EsScribeCfg ESV5
133131defaultEsScribeCfgV5 = defaultEsScribeCfg' (Typeable.Proxy :: Typeable.Proxy ESV5)
134132
135133
@@ -274,62 +272,66 @@ mkEsScribe
274272 )
275273 => EsScribeCfg v
276274 -> BHEnv v
275+ -> IndexName v
276+ -- ^ Treated as a prefix if index sharding is enabled
277+ -> MappingName v
277278 -> PermitFunc
278279 -> Verbosity
279280 -> IO Scribe
280−mkEsScribe cfg@EsScribeCfg {..} env permit verb = error "fuck"
281− -- q <- newTBMQueueIO $ unEsQueueSize essQueueSize
282− -- endSig <- newEmptyMVar
283−
284− -- runBH prx env $ do
285− -- if shardingEnabled
286− -- then do
287− -- -- create or update
288− -- res <- putTemplate prx tpl tplName
289− -- unless (statusIsSuccessful (responseStatus res)) $
290− -- liftIO $ EX.throwIO (CouldNotPutTemplate res)
291− -- else do
292− -- ixExists <- indexExists prx ix
293− -- if ixExists
294− -- then do
295− -- res <- updateIndexSettings prx (toUpdatabaleIndexSettings prx essIndexSettings) ix
296− -- unless (statusIsSuccessful (responseStatus res)) $
297− -- liftIO $ EX.throwIO (CouldNotUpdateIndexSettings res)
298− -- else do
299− -- r1 <- createIndex prx essIndexSettings ix
300− -- unless (statusIsSuccessful (responseStatus r1)) $
301− -- liftIO $ EX.throwIO (CouldNotCreateIndex r1)
302− -- r2 <- putMapping prx ix mapping mappingValue
303− -- unless (statusIsSuccessful (responseStatus r2)) $
304− -- liftIO $ EX.throwIO (CouldNotCreateMapping r2)
305−
306− -- workers <- replicateM (unEsPoolSize essPoolSize) $ async $
307− -- startWorker cfg env mapping q
308−
309− -- _ <- async $ do
310− -- takeMVar endSig
311− -- atomically $ closeTBMQueue q
312− -- mapM_ waitCatch workers
313− -- putMVar endSig ()
314−
315− -- let finalizer = putMVar endSig () >> takeMVar endSig
316− -- return (Scribe (logger q) finalizer permit)
317− -- where
318− -- logger :: forall a. LogItem a => TBMQueue (IndexName v, Value) -> Item a -> IO ()
319− -- logger q i =
320− -- void $ atomically $ tryWriteTBMQueue q (chooseIxn prx ix essIndexSharding i, itemJson' i)
321− -- prx :: Typeable.Proxy v
322− -- prx = Typeable.Proxy
323− -- tplName = toTemplateName prx ixn
324− -- shardingEnabled = case essIndexSharding of
325− -- NoIndexSharding -> False
326− -- _ -> True
327− -- tpl = toIndexTemplate prx (toTemplatePattern prx (ixn <> "-*")) (Just essIndexSettings) [mappingValue]
328− -- ixn = fromIndexName prx ix
329− -- itemJson' :: LogItem a => Item a -> Value
330− -- itemJson' i
331− -- | essAnnotateTypes = itemJson verb (TypeAnnotated <$> i)
332− -- | otherwise = itemJson verb i
281+mkEsScribe cfg@EsScribeCfg {..} env ix mapping permit verb = do
282+ q <- newTBMQueueIO $ unEsQueueSize essQueueSize
283+ endSig <- newEmptyMVar
284+
285+ runBH prx env $ do
286+ if shardingEnabled
287+ then do
288+ -- create or update
289+ res <- putTemplate prx tpl tplName
290+ unless (statusIsSuccessful (responseStatus res)) $
291+ liftIO $ EX.throwIO (CouldNotPutTemplate res)
292+ else do
293+ ixExists <- indexExists prx ix
294+ if ixExists
295+ then do
296+ res <- updateIndexSettings prx (toUpdatabaleIndexSettings prx essIndexSettings) ix
297+ unless (statusIsSuccessful (responseStatus res)) $
298+ liftIO $ EX.throwIO (CouldNotUpdateIndexSettings res)
299+ else do
300+ r1 <- createIndex prx essIndexSettings ix
301+ unless (statusIsSuccessful (responseStatus r1)) $
302+ liftIO $ EX.throwIO (CouldNotCreateIndex r1)
303+ r2 <- putMapping prx ix mapping base
304+ unless (statusIsSuccessful (responseStatus r2)) $
305+ liftIO $ EX.throwIO (CouldNotCreateMapping r2)
306+
307+ workers <- replicateM (unEsPoolSize essPoolSize) $ async $
308+ startWorker cfg env mapping q
309+
310+ _ <- async $ do
311+ takeMVar endSig
312+ atomically $ closeTBMQueue q
313+ mapM_ waitCatch workers
314+ putMVar endSig ()
315+
316+ let finalizer = putMVar endSig () >> takeMVar endSig
317+ return (Scribe (logger q) finalizer permit)
318+ where
319+ logger :: forall a. LogItem a => TBMQueue (IndexName v, Value) -> Item a -> IO ()
320+ logger q i =
321+ void $ atomically $ tryWriteTBMQueue q (chooseIxn prx ix essIndexSharding i, itemJson' i)
322+ prx :: Typeable.Proxy v
323+ prx = Typeable.Proxy
324+ tplName = toTemplateName prx ixn
325+ shardingEnabled = case essIndexSharding of
326+ NoIndexSharding -> False
327+ _ -> True
328+ tpl = toIndexTemplate prx (toTemplatePattern prx (ixn <> "-*")) (Just essIndexSettings) [toJSON base]
329+ base = baseMapping prx mapping
330+ ixn = fromIndexName prx ix
331+ itemJson' :: LogItem a => Item a -> Value
332+ itemJson' i
333+ | essAnnotateTypes = itemJson verb (TypeAnnotated <$> i)
334+ | otherwise = itemJson verb i
333335
334336
335337-------------------------------------------------------------------------------
@@ -489,7 +491,10 @@ class ESVersion v where
489491
490492
491493
494+#if !MIN_VERSION_bloodhound(0,17,0)
492495data ESV1 = ESV1
496+{-# DEPRECATED ESV1 "ESV1 is deprecated and removed in bloodhound >= 0.17.0" #-}
497+
493498
494499instance ESVersion ESV1 where
495500 type BHEnv ESV1 = V1.BHEnv
@@ -529,6 +534,7 @@ instance ESVersion ESV1 where
529534 [ "type" .= String "string"
530535 , "index" .= String "analyzed"
531536 ]
537+#endif
532538
533539
534540data ESV5 = ESV5