@@ -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 as Monoid
31+import Data.Monoid ((<>))
3232import Data.Text (Text)
3333import qualified Data.Text as T
3434import qualified Data.Text.Encoding as T
@@ -37,12 +37,8 @@ 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
4340import qualified Database.V1.Bloodhound as V1
4441import qualified Database.V5.Bloodhound as V5
45−#endif
4642import Network.HTTP.Client
4743import Network.HTTP.Types.Status
4844import Text.Printf (printf)
@@ -57,30 +53,26 @@ import Katip.Scribes.ElasticSearch.Annotations
5753-- 'defaultEsScribeCfgV1' and 'defaultESScribeCfgV5' for a good
5854-- starting point depending on the ES version you have.
5955data EsScribeCfg v = EsScribeCfg {
60− essRetryPolicy :: RetryPolicy
56+ essRetryPolicy :: RetryPolicy
6157 -- ^ Retry policy when there are errors sending logs to the server
62− , essQueueSize :: EsQueueSize
58+ , essQueueSize :: EsQueueSize
6359 -- ^ Maximum size of the bounded log queue
64− , essPoolSize :: EsPoolSize
60+ , essPoolSize :: EsPoolSize
6561 -- ^ Worker pool size limit for sending data to the
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
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
8072 -- ^ This will be the IndexSettings type from the appropriate
8173 -- bloodhound module, either @Database.V1.Bloodhound@ or
8274 -- @Database.V5.Bloodhound@
83− , essIndexSharding :: IndexShardingPolicy
75+ , essIndexSharding :: IndexShardingPolicy
8476 } deriving (Typeable)
8577
8678
@@ -97,37 +89,47 @@ data EsScribeCfg v = EsScribeCfg {
9789-- * Annotate types set to False
9890--
9991-- * DailyIndexSharding
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−
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)
112115
113116-------------------------------------------------------------------------------
114−-- | Alias of 'defaultEsScribeCfgV5' to minimize API
117+-- | Alias of 'defaultEsScribeCfgV1' to minimize API
115118-- breakage. Previous versions of katip-elasticsearch only supported
116−-- ES version 1 and defaulted to it.
117−defaultEsScribeCfg :: EsScribeCfg ESV5
118−defaultEsScribeCfg = defaultEsScribeCfgV5
119+-- ES version 1.
120+defaultEsScribeCfg :: IndexName ESV1 -> MappingName ESV1 -> EsScribeCfg ESV1
121+defaultEsScribeCfg = defaultEsScribeCfgV1
119122
120123
121124-------------------------------------------------------------------------------
122−#if !MIN_VERSION_bloodhound(0,17,0)
123125-- | EsScribeCfg that will use ElasticSearch V1
124−defaultEsScribeCfgV1 :: EsScribeCfg ESV1
126+defaultEsScribeCfgV1 :: IndexName ESV1 -> MappingName ESV1 -> EsScribeCfg ESV1
125127defaultEsScribeCfgV1 = defaultEsScribeCfg' (Typeable.Proxy :: Typeable.Proxy ESV1)
126−#endif
128+
127129
128130-------------------------------------------------------------------------------
129131-- | EsScribeCfg that will use ElasticSearch V5
130−defaultEsScribeCfgV5 :: EsScribeCfg ESV5
132+defaultEsScribeCfgV5 :: IndexName ESV5 -> MappingName ESV5 -> EsScribeCfg ESV5
131133defaultEsScribeCfgV5 = defaultEsScribeCfg' (Typeable.Proxy :: Typeable.Proxy ESV5)
132134
133135
@@ -272,66 +274,62 @@ mkEsScribe
272274 )
273275 => EsScribeCfg v
274276 -> BHEnv v
275− -> IndexName v
276− -- ^ Treated as a prefix if index sharding is enabled
277− -> MappingName v
278277 -> PermitFunc
279278 -> Verbosity
280279 -> IO Scribe
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
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
335333
336334
337335-------------------------------------------------------------------------------
@@ -491,10 +489,7 @@ class ESVersion v where
491489
492490
493491
494−#if !MIN_VERSION_bloodhound(0,17,0)
495492data ESV1 = ESV1
496−{-# DEPRECATED ESV1 "ESV1 is deprecated and removed in bloodhound >= 0.17.0" #-}
497−
498493
499494instance ESVersion ESV1 where
500495 type BHEnv ESV1 = V1.BHEnv
@@ -534,7 +529,6 @@ instance ESVersion ESV1 where
534529 [ "type" .= String "string"
535530 , "index" .= String "analyzed"
536531 ]
537−#endif
538532
539533
540534data ESV5 = ESV5