@@ -19,6 +19,9 @@ module Katip.Scribes.ElasticSearch
1919 , essPoolSize
2020 , essAnnotateTypes
2121 , essIndexSettings
22+ , essIndexSharding
23+ , IndexShardingPolicy(..)
24+ , IndexNameSegment(..)
2225 , defaultEsScribeCfg
2326 -- * Utilities
2427 , mkDocId
@@ -27,8 +30,6 @@ module Katip.Scribes.ElasticSearch
2730
2831
2932-------------------------------------------------------------------------------
30−import Control.Monad.IO.Class
31−
3233import Control.Applicative
3334import Control.Concurrent
3435import Control.Concurrent.Async
@@ -37,6 +38,7 @@ import Control.Exception.Base
3738import Control.Exception.Enclosed
3839import Control.Monad
3940import Control.Monad.Catch
41+import Control.Monad.IO.Class
4042import Control.Monad.STM
4143import Control.Retry (RetryPolicy,
4244 exponentialBackoff,
@@ -45,7 +47,9 @@ import Control.Retry (RetryPolicy,
4547import Data.Aeson
4648import Data.Monoid ((<>))
4749import Data.Text (Text)
50+import qualified Data.Text as T
4851import qualified Data.Text.Encoding as T
52+import Data.Time
4953import Data.Typeable
5054import Data.UUID
5155import Database.Bloodhound
@@ -81,6 +85,7 @@ data EsScribeCfg = EsScribeCfg {
8185 -- querying transparently remove the type annotations if this is
8286 -- enabled.
8387 , essIndexSettings :: IndexSettings
88+ , essIndexSharding :: IndexShardingPolicy
8489 } deriving (Typeable)
8590
8691
@@ -103,9 +108,101 @@ defaultEsScribeCfg = EsScribeCfg {
103108 , essPoolSize = EsPoolSize 2
104109 , essAnnotateTypes = False
105110 , essIndexSettings = defaultIndexSettings
111+ , essIndexSharding = NoIndexSharding
106112 }
107113
108114
115+-------------------------------------------------------------------------------
116+-- | How should katip store your log data?
117+--
118+-- * NoIndexSharding will store all logs in one index name. This is
119+-- the simplest option but is not advised in production. In practice,
120+-- the index will grow very large and will get slower to
121+-- search. Deleting records based on some sort of retention period is
122+-- also extremely slow.
123+--
124+-- * MonthlyIndexSharding, DailyIndexSharding, HourlyIndexSharding,
125+-- EveryMinuteIndexSharding will generate indexes based on the time of
126+-- the log. Index name is treated as a prefix. So if your index name
127+-- is @foo@ and DailySharding is used, logs will be stored in
128+-- @foo-2016-02-25@, @foo-2016-02-26@ and so on. Index templating will
129+-- be used to set up mappings automatically. Deletes based on date are
130+-- very fast and queries can be restricted to date ranges for better
131+-- performance. Queries against all dates should use @foo-*@ as an
132+-- index name. Note that index aliasing's glob feature is not suitable
133+-- for these date ranges as it matches index names as they are
134+-- declared, so new dates will be excluded. DailyIndexSharding is a
135+-- reasonable choice. Changing index sharding strategies is a *bad
136+-- idea*.
137+--
138+-- * CustomSharding: supply your own function that decomposes an item
139+-- into its index name heirarchy which will be appended to the index
140+-- name. So for instance if your function return ["arbitrary",
141+-- "prefix"], the index will be @foo-arbitrary-prefix@ and the index
142+-- template will be set to match @foo-*@
143+data IndexShardingPolicy = NoIndexSharding
144+ | MonthlyIndexSharding
145+ | DailyIndexSharding
146+ | HourlyIndexSharding
147+ | EveryMinuteIndexSharding
148+ | CustomIndexSharding (forall a. Item a -> [IndexNameSegment])
149+
150+
151+instance Show IndexShardingPolicy where
152+ show NoIndexSharding = "NoIndexSharding"
153+ show MonthlyIndexSharding = "MonthlyIndexSharding"
154+ show DailyIndexSharding = "DailyIndexSharding"
155+ show HourlyIndexSharding = "HourlyIndexSharding"
156+ show EveryMinuteIndexSharding = "EveryMinuteIndexSharding"
157+ show (CustomIndexSharding _) = "CustomIndexSharding λ"
158+
159+
160+-------------------------------------------------------------------------------
161+newtype IndexNameSegment = IndexNameSegment {
162+ indexNameSegment :: Text
163+ } deriving (Show, Eq, Ord)
164+
165+
166+-------------------------------------------------------------------------------
167+shardPolicySegs :: IndexShardingPolicy -> Item a -> [IndexNameSegment]
168+shardPolicySegs NoIndexSharding _ = []
169+shardPolicySegs MonthlyIndexSharding Item {..} = [sis y, sis m]
170+ where
171+ (y, m, _) = toGregorian (utctDay _itemTime)
172+shardPolicySegs DailyIndexSharding Item {..} = [sis y, sis m, sis d]
173+ where
174+ (y, m, d) = toGregorian (utctDay _itemTime)
175+shardPolicySegs HourlyIndexSharding Item {..} = [sis y, sis m, sis d, sis h]
176+ where
177+ (y, m, d) = toGregorian (utctDay _itemTime)
178+ (h, _) = splitTime (utctDayTime _itemTime)
179+shardPolicySegs EveryMinuteIndexSharding Item {..} = [sis y, sis m, sis d, sis h, sis mn]
180+ where
181+ (y, m, d) = toGregorian (utctDay _itemTime)
182+ (h, mn) = splitTime (utctDayTime _itemTime)
183+shardPolicySegs (CustomIndexSharding f) i = f i
184+
185+
186+-------------------------------------------------------------------------------
187+chooseIxn :: IndexName -> IndexShardingPolicy -> Item a -> IndexName
188+chooseIxn (IndexName ixn) p i =
189+ IndexName (T.intercalate "-" (ixn:segs))
190+ where
191+ segs = indexNameSegment <$> shardPolicySegs p i
192+
193+
194+-------------------------------------------------------------------------------
195+sis :: Show a => a -> IndexNameSegment
196+sis = IndexNameSegment . T.pack . show
197+
198+
199+-------------------------------------------------------------------------------
200+splitTime :: DiffTime -> (Int, Int)
201+splitTime t = asMins `divMod` 60
202+ where
203+ asMins = floor t `div` 60
204+
205+
109206-------------------------------------------------------------------------------
110207data EsScribeSetupError = CouldNotCreateIndex !Reply
111208 | CouldNotCreateMapping !Reply deriving (Typeable, Show)
@@ -118,6 +215,7 @@ mkEsScribe
118215 :: EsScribeCfg
119216 -> Server
120217 -> IndexName
218+ -- ^ Treated as a prefix if index sharding is enabled
121219 -> MappingName
122220 -> Severity
123221 -> Verbosity
@@ -140,12 +238,14 @@ mkEsScribe cfg@EsScribeCfg {..} server ix mapping sev verb = do
140238 unless (statusIsSuccessful (responseStatus r1)) $
141239 liftIO $ throwIO (CouldNotCreateIndex r1)
142240 --TODO: throw on error
143− r2 <- putMapping ix mapping (baseMapping mapping)
241+ r2 <- if shardingEnabled
242+ then putTemplate tpl tplName
243+ else putMapping ix mapping (baseMapping mapping)
144244 unless (statusIsSuccessful (responseStatus r2)) $
145245 liftIO $ throwIO (CouldNotCreateMapping r2)
146246
147247 workers <- replicateM (unEsPoolSize essPoolSize) $ async $
148− startWorker cfg env ix mapping q
248+ startWorker cfg env mapping q
149249
150250 _ <- async $ do
151251 takeMVar endSig
@@ -155,13 +255,19 @@ mkEsScribe cfg@EsScribeCfg {..} server ix mapping sev verb = do
155255
156256 let scribe = Scribe $ \ i ->
157257 when (_itemSeverity i >= sev) $
158− void $ atomically $ tryWriteTBMQueue q (itemJson' verb i)
258+ void $ atomically $ tryWriteTBMQueue q (chooseIxn ix essIndexSharding i, itemJson' i)
159259 let finalizer = putMVar endSig () >> takeMVar endSig
160260 return (scribe, finalizer)
161261 where
162− itemJson' v i
163− | essAnnotateTypes = itemJson v (TypeAnnotated <$> i)
164− | otherwise = itemJson v i
262+ tplName = TemplateName ixn
263+ shardingEnabled = case essIndexSharding of
264+ NoIndexSharding -> False
265+ _ -> True
266+ tpl = IndexTemplate (TemplatePattern (ixn <> "-*")) (Just essIndexSettings) [toJSON (baseMapping mapping)]
267+ IndexName ixn = ix
268+ itemJson' i
269+ | essAnnotateTypes = itemJson verb (TypeAnnotated <$> i)
270+ | otherwise = itemJson verb i
165271
166272
167273-------------------------------------------------------------------------------
@@ -242,24 +348,24 @@ mkNonZero ctor n
242348
243349-------------------------------------------------------------------------------
244350startWorker
245− :: EsScribeCfg
351+ :: forall a. EsScribeCfg
246352 -> BHEnv
247− -> IndexName
248353 -> MappingName
249− -> TBMQueue Value
354+ -> TBMQueue (IndexName, Value)
250355 -> IO ()
251−startWorker EsScribeCfg {..} env ix mapping q = go -- manual recursion here. is this a space leak
356+startWorker EsScribeCfg {..} env mapping q = go
252357 where
253358 go = do
254− v <- atomically $ readTBMQueue q
255− case v of
256− Just v' -> do
257− sendLog v' `catchAny` eat
359+ popped <- atomically $ readTBMQueue q
360+ case popped of
361+ Just (ixn, v) -> do
362+ sendLog ixn v `catchAny` eat
258363 go
259364 Nothing -> return ()
260− sendLog v = void $ recovering essRetryPolicy [handler] $ const $ do
365+ sendLog :: IndexName -> Value -> IO ()
366+ sendLog ixn v = void $ recovering essRetryPolicy [handler] $ const $ do
261367 did <- mkDocId
262− runBH env $ indexDocument ix mapping defaultIndexDocumentSettings v did
368+ runBH env $ indexDocument ixn mapping defaultIndexDocumentSettings v did
263369 eat _ = return ()
264370 handler _ = Handler $ \e ->
265371 case fromException e of