@@ -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
@@ -95,6 +100,8 @@ data EsScribeCfg = EsScribeCfg {
95100-- * Pool size of 2
96101--
97102-- * Annotate types set to False
103+--
104+-- * NoIndexSharding
98105defaultEsScribeCfg :: EsScribeCfg
99106defaultEsScribeCfg = EsScribeCfg {
100107 essManagerSettings = defaultManagerSettings
@@ -103,9 +110,104 @@ defaultEsScribeCfg = EsScribeCfg {
103110 , essPoolSize = EsPoolSize 2
104111 , essAnnotateTypes = False
105112 , essIndexSettings = defaultIndexSettings
113+ , essIndexSharding = NoIndexSharding
106114 }
107115
108116
117+-------------------------------------------------------------------------------
118+-- | How should katip store your log data?
119+--
120+-- * NoIndexSharding will store all logs in one index name. This is
121+-- the simplest option but is not advised in production. In practice,
122+-- the index will grow very large and will get slower to
123+-- search. Deleting records based on some sort of retention period is
124+-- also extremely slow.
125+--
126+-- * MonthlyIndexSharding, DailyIndexSharding, HourlyIndexSharding,
127+-- EveryMinuteIndexSharding will generate indexes based on the time of
128+-- the log. Index name is treated as a prefix. So if your index name
129+-- is @foo@ and DailySharding is used, logs will be stored in
130+-- @foo-2016-2-25@, @foo-2016-2-26@ and so on. Index templating will
131+-- be used to set up mappings automatically. Deletes based on date are
132+-- very fast and queries can be restricted to date ranges for better
133+-- performance. Queries against all dates should use @foo-*@ as an
134+-- index name. Note that index aliasing's glob feature is not suitable
135+-- for these date ranges as it matches index names as they are
136+-- declared, so new dates will be excluded. DailyIndexSharding is a
137+-- reasonable choice. Changing index sharding strategies is not
138+-- advisable.
139+--
140+-- * CustomSharding: supply your own function that decomposes an item
141+-- into its index name heirarchy which will be appended to the index
142+-- name. So for instance if your function return ["arbitrary",
143+-- "prefix"], the index will be @foo-arbitrary-prefix@ and the index
144+-- template will be set to match @foo-*@. In general, you want to use
145+-- segments of increasing granularity (like year, month, day for
146+-- dates). This makes it easier to address groups of indexes
147+-- (e.g. @foo-2016-*@).
148+data IndexShardingPolicy = NoIndexSharding
149+ | MonthlyIndexSharding
150+ | DailyIndexSharding
151+ | HourlyIndexSharding
152+ | EveryMinuteIndexSharding
153+ | CustomIndexSharding (forall a. Item a -> [IndexNameSegment])
154+
155+
156+instance Show IndexShardingPolicy where
157+ show NoIndexSharding = "NoIndexSharding"
158+ show MonthlyIndexSharding = "MonthlyIndexSharding"
159+ show DailyIndexSharding = "DailyIndexSharding"
160+ show HourlyIndexSharding = "HourlyIndexSharding"
161+ show EveryMinuteIndexSharding = "EveryMinuteIndexSharding"
162+ show (CustomIndexSharding _) = "CustomIndexSharding λ"
163+
164+
165+-------------------------------------------------------------------------------
166+newtype IndexNameSegment = IndexNameSegment {
167+ indexNameSegment :: Text
168+ } deriving (Show, Eq, Ord)
169+
170+
171+-------------------------------------------------------------------------------
172+shardPolicySegs :: IndexShardingPolicy -> Item a -> [IndexNameSegment]
173+shardPolicySegs NoIndexSharding _ = []
174+shardPolicySegs MonthlyIndexSharding Item {..} = [sis y, sis m]
175+ where
176+ (y, m, _) = toGregorian (utctDay _itemTime)
177+shardPolicySegs DailyIndexSharding Item {..} = [sis y, sis m, sis d]
178+ where
179+ (y, m, d) = toGregorian (utctDay _itemTime)
180+shardPolicySegs HourlyIndexSharding Item {..} = [sis y, sis m, sis d, sis h]
181+ where
182+ (y, m, d) = toGregorian (utctDay _itemTime)
183+ (h, _) = splitTime (utctDayTime _itemTime)
184+shardPolicySegs EveryMinuteIndexSharding Item {..} = [sis y, sis m, sis d, sis h, sis mn]
185+ where
186+ (y, m, d) = toGregorian (utctDay _itemTime)
187+ (h, mn) = splitTime (utctDayTime _itemTime)
188+shardPolicySegs (CustomIndexSharding f) i = f i
189+
190+
191+-------------------------------------------------------------------------------
192+chooseIxn :: IndexName -> IndexShardingPolicy -> Item a -> IndexName
193+chooseIxn (IndexName ixn) p i =
194+ IndexName (T.intercalate "-" (ixn:segs))
195+ where
196+ segs = indexNameSegment <$> shardPolicySegs p i
197+
198+
199+-------------------------------------------------------------------------------
200+sis :: Show a => a -> IndexNameSegment
201+sis = IndexNameSegment . T.pack . show
202+
203+
204+-------------------------------------------------------------------------------
205+splitTime :: DiffTime -> (Int, Int)
206+splitTime t = asMins `divMod` 60
207+ where
208+ asMins = floor t `div` 60
209+
210+
109211-------------------------------------------------------------------------------
110212data EsScribeSetupError = CouldNotCreateIndex !Reply
111213 | CouldNotCreateMapping !Reply deriving (Typeable, Show)
@@ -118,6 +220,7 @@ mkEsScribe
118220 :: EsScribeCfg
119221 -> Server
120222 -> IndexName
223+ -- ^ Treated as a prefix if index sharding is enabled
121224 -> MappingName
122225 -> Severity
123226 -> Verbosity
@@ -139,13 +242,14 @@ mkEsScribe cfg@EsScribeCfg {..} server ix mapping sev verb = do
139242 r1 <- createIndex essIndexSettings ix
140243 unless (statusIsSuccessful (responseStatus r1)) $
141244 liftIO $ throwIO (CouldNotCreateIndex r1)
142− --TODO: throw on error
143− r2 <- putMapping ix mapping (baseMapping mapping)
245+ r2 <- if shardingEnabled
246+ then putTemplate tpl tplName
247+ else putMapping ix mapping (baseMapping mapping)
144248 unless (statusIsSuccessful (responseStatus r2)) $
145249 liftIO $ throwIO (CouldNotCreateMapping r2)
146250
147251 workers <- replicateM (unEsPoolSize essPoolSize) $ async $
148− startWorker cfg env ix mapping q
252+ startWorker cfg env mapping q
149253
150254 _ <- async $ do
151255 takeMVar endSig
@@ -155,13 +259,19 @@ mkEsScribe cfg@EsScribeCfg {..} server ix mapping sev verb = do
155259
156260 let scribe = Scribe $ \ i ->
157261 when (_itemSeverity i >= sev) $
158− void $ atomically $ tryWriteTBMQueue q (itemJson' verb i)
262+ void $ atomically $ tryWriteTBMQueue q (chooseIxn ix essIndexSharding i, itemJson' i)
159263 let finalizer = putMVar endSig () >> takeMVar endSig
160264 return (scribe, finalizer)
161265 where
162− itemJson' v i
163− | essAnnotateTypes = itemJson v (TypeAnnotated <$> i)
164− | otherwise = itemJson v i
266+ tplName = TemplateName ixn
267+ shardingEnabled = case essIndexSharding of
268+ NoIndexSharding -> False
269+ _ -> True
270+ tpl = IndexTemplate (TemplatePattern (ixn <> "-*")) (Just essIndexSettings) [toJSON (baseMapping mapping)]
271+ IndexName ixn = ix
272+ itemJson' i
273+ | essAnnotateTypes = itemJson verb (TypeAnnotated <$> i)
274+ | otherwise = itemJson verb i
165275
166276
167277-------------------------------------------------------------------------------
@@ -244,22 +354,22 @@ mkNonZero ctor n
244354startWorker
245355 :: EsScribeCfg
246356 -> BHEnv
247− -> IndexName
248357 -> MappingName
249− -> TBMQueue Value
358+ -> TBMQueue (IndexName, Value)
250359 -> IO ()
251−startWorker EsScribeCfg {..} env ix mapping q = go -- manual recursion here. is this a space leak
360+startWorker EsScribeCfg {..} env mapping q = go
252361 where
253362 go = do
254− v <- atomically $ readTBMQueue q
255− case v of
256− Just v' -> do
257− sendLog v' `catchAny` eat
363+ popped <- atomically $ readTBMQueue q
364+ case popped of
365+ Just (ixn, v) -> do
366+ sendLog ixn v `catchAny` eat
258367 go
259368 Nothing -> return ()
260− sendLog v = void $ recovering essRetryPolicy [handler] $ const $ do
369+ sendLog :: IndexName -> Value -> IO ()
370+ sendLog ixn v = void $ recovering essRetryPolicy [handler] $ const $ do
261371 did <- mkDocId
262− runBH env $ indexDocument ix mapping defaultIndexDocumentSettings v did
372+ runBH env $ indexDocument ixn mapping defaultIndexDocumentSettings v did
263373 eat _ = return ()
264374 handler _ = Handler $ \e ->
265375 case fromException e of