@@ -53,30 +53,26 @@ import Katip.Scribes.ElasticSearch.Annotations
5353-- 'defaultEsScribeCfgV1' and 'defaultESScribeCfgV5' for a good
5454-- starting point depending on the ES version you have.
5555data EsScribeCfg v = EsScribeCfg {
56− essRetryPolicy :: RetryPolicy
56+ essRetryPolicy :: RetryPolicy
5757 -- ^ Retry policy when there are errors sending logs to the server
58− , essQueueSize :: EsQueueSize
58+ , essQueueSize :: EsQueueSize
5959 -- ^ Maximum size of the bounded log queue
60− , essPoolSize :: EsPoolSize
60+ , essPoolSize :: EsPoolSize
6161 -- ^ Worker pool size limit for sending data to the
62− , essAnnotateTypes :: Bool
63− -- ^ Different payload items coexist in the "data" attribute in
64− -- ES. It is possible for different payloads to have different
65− -- types for the same key, e.g. an "id" key that is sometimes a
66− -- number and sometimes a string. If you're having ES do dynamic
67− -- mapping, the first log item will set the type and any that
68− -- don't conform will be *discarded*. If you set this to True,
69− -- keys will recursively be appended with their ES core
70− -- type. e.g. "id" would become "id::l" and "id::s"
71− -- automatically, so they won't conflict. When this library
72− -- exposes a querying API, we will try to make deserialization and
73− -- querying transparently remove the type annotations if this is
74− -- enabled.
75− , 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
7672 -- ^ This will be the IndexSettings type from the appropriate
7773 -- bloodhound module, either @Database.V1.Bloodhound@ or
7874 -- @Database.V5.Bloodhound@
79− , essIndexSharding :: IndexShardingPolicy
75+ , essIndexSharding :: IndexShardingPolicy
8076 } deriving (Typeable)
8177
8278
@@ -93,34 +89,47 @@ data EsScribeCfg v = EsScribeCfg {
9389-- * Annotate types set to False
9490--
9591-- * DailyIndexSharding
96−defaultEsScribeCfg' :: ESVersion v => proxy v -> EsScribeCfg v
97−defaultEsScribeCfg' prx = EsScribeCfg {
98− essRetryPolicy = exponentialBackoff 25000 <> limitRetries 5
99− , essQueueSize = EsQueueSize 1000
100− , essPoolSize = EsPoolSize 2
101− , essAnnotateTypes = False
102− , essIndexSettings = defaultIndexSettings prx
103− , essIndexSharding = DailyIndexSharding
104− }
105−
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)
106115
107116-------------------------------------------------------------------------------
108117-- | Alias of 'defaultEsScribeCfgV1' to minimize API
109118-- breakage. Previous versions of katip-elasticsearch only supported
110119-- ES version 1.
111−defaultEsScribeCfg :: EsScribeCfg ESV1
120+defaultEsScribeCfg :: IndexName ESV1 -> MappingName ESV1 -> EsScribeCfg ESV1
112121defaultEsScribeCfg = defaultEsScribeCfgV1
113122
114123
115124-------------------------------------------------------------------------------
116125-- | EsScribeCfg that will use ElasticSearch V1
117−defaultEsScribeCfgV1 :: EsScribeCfg ESV1
126+defaultEsScribeCfgV1 :: IndexName ESV1 -> MappingName ESV1 -> EsScribeCfg ESV1
118127defaultEsScribeCfgV1 = defaultEsScribeCfg' (Typeable.Proxy :: Typeable.Proxy ESV1)
119128
120129
121130-------------------------------------------------------------------------------
122131-- | EsScribeCfg that will use ElasticSearch V5
123−defaultEsScribeCfgV5 :: EsScribeCfg ESV5
132+defaultEsScribeCfgV5 :: IndexName ESV5 -> MappingName ESV5 -> EsScribeCfg ESV5
124133defaultEsScribeCfgV5 = defaultEsScribeCfg' (Typeable.Proxy :: Typeable.Proxy ESV5)
125134
126135
@@ -265,68 +274,62 @@ mkEsScribe
265274 )
266275 => EsScribeCfg v
267276 -> BHEnv v
268− -> IndexName v
269− -- ^ Treated as a prefix if index sharding is enabled
270− -> MappingName v
271− -> Value
272− -- ^ Index mapping value. Use 'baseMapping' as default value
273− -- constructor.
274277 -> PermitFunc
275278 -> Verbosity
276279 -> IO Scribe
277−mkEsScribe cfg@EsScribeCfg {..} env ix mapping mappingValue permit verb = do
278− q <- newTBMQueueIO $ unEsQueueSize essQueueSize
279− endSig <- newEmptyMVar
280−
281− runBH prx env $ do
282− if shardingEnabled
283− then do
284− -- create or update
285− res <- putTemplate prx tpl tplName
286− unless (statusIsSuccessful (responseStatus res)) $
287− liftIO $ EX.throwIO (CouldNotPutTemplate res)
288− else do
289− ixExists <- indexExists prx ix
290− if ixExists
291− then do
292− res <- updateIndexSettings prx (toUpdatabaleIndexSettings prx essIndexSettings) ix
293− unless (statusIsSuccessful (responseStatus res)) $
294− liftIO $ EX.throwIO (CouldNotUpdateIndexSettings res)
295− else do
296− r1 <- createIndex prx essIndexSettings ix
297− unless (statusIsSuccessful (responseStatus r1)) $
298− liftIO $ EX.throwIO (CouldNotCreateIndex r1)
299− r2 <- putMapping prx ix mapping mappingValue
300− unless (statusIsSuccessful (responseStatus r2)) $
301− liftIO $ EX.throwIO (CouldNotCreateMapping r2)
302−
303− workers <- replicateM (unEsPoolSize essPoolSize) $ async $
304− startWorker cfg env mapping q
305−
306− _ <- async $ do
307− takeMVar endSig
308− atomically $ closeTBMQueue q
309− mapM_ waitCatch workers
310− putMVar endSig ()
311−
312− let finalizer = putMVar endSig () >> takeMVar endSig
313− return (Scribe (logger q) finalizer permit)
314− where
315− logger :: forall a. LogItem a => TBMQueue (IndexName v, Value) -> Item a -> IO ()
316− logger q i =
317− void $ atomically $ tryWriteTBMQueue q (chooseIxn prx ix essIndexSharding i, itemJson' i)
318− prx :: Typeable.Proxy v
319− prx = Typeable.Proxy
320− tplName = toTemplateName prx ixn
321− shardingEnabled = case essIndexSharding of
322− NoIndexSharding -> False
323− _ -> True
324− tpl = toIndexTemplate prx (toTemplatePattern prx (ixn <> "-*")) (Just essIndexSettings) [mappingValue]
325− ixn = fromIndexName prx ix
326− itemJson' :: LogItem a => Item a -> Value
327− itemJson' i
328− | essAnnotateTypes = itemJson verb (TypeAnnotated <$> i)
329− | 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
330333
331334
332335-------------------------------------------------------------------------------