@@ -1,18 +1,34 @@
1+{-# LANGUAGE DeriveDataTypeable #-}
2+{-# LANGUAGE RecordWildCards #-}
13module Katip.Scribes.ElasticSearch
2− ( mkEsScribe
3− , mkDocId
4+ (-- * Building a scribe
5+ mkEsScribe
6+ -- * Scribe configuration
47 , EsQueueSize
58 , mkEsQueueSize
9+ , EsPoolSize
10+ , mkEsPoolSize
11+ , EsScribeCfg
12+ , essManagerSettings
13+ , essRetryPolicy
14+ , essQueueSize
15+ , essPoolSize
16+ , defaultEsScribeCfg
17+ -- * Utilities
18+ , mkDocId
619 ) where
720
821
922-------------------------------------------------------------------------------
1023import Control.Applicative
11−import Control.Concurrent.Async
24+import Control.Concurrent
1225import Control.Concurrent.STM
26+import Control.Exception.Enclosed
1327import Control.Monad
28+import Control.Retry
1429import Data.Aeson
15−import qualified Data.Text.Encoding as T
30+import qualified Data.Text.Encoding as T
31+import Data.Typeable
1632import Data.UUID
1733import Database.Bloodhound
1834import Network.HTTP.Client
@@ -22,26 +38,55 @@ import Katip.Core
2238-------------------------------------------------------------------------------
2339
2440
41+data EsScribeCfg = EsScribeCfg {
42+ essManagerSettings :: ManagerSettings
43+ -- ^ Connection manager settings
44+ , essRetryPolicy :: RetryPolicy
45+ -- ^ Retry policy when there are errors sending logs to the server
46+ , essQueueSize :: EsQueueSize
47+ -- ^ Maximum size of the bounded log queue
48+ , essPoolSize :: EsPoolSize
49+ -- ^ Worker pool size limit for sending data to the
50+ } deriving (Typeable)
51+
52+
53+-- | Reasonable defaults for a config:
54+--
55+-- * defaultManagerSettings
56+--
57+-- * exponential backoff with 25ms base delay up to 5 retries
58+--
59+-- * Queue size of 1000
60+--
61+-- * Pool size of 2
62+defaultEsScribeCfg :: EsScribeCfg
63+defaultEsScribeCfg = EsScribeCfg {
64+ essManagerSettings = defaultManagerSettings
65+ , essRetryPolicy = exponentialBackoff 25 <> limitRetries 5
66+ , essQueueSize = EsQueueSize 1000
67+ , essPoolSize = EsPoolSize 2
68+ }
69+
70+
2571-------------------------------------------------------------------------------
2672mkEsScribe
27− :: ManagerSettings
73+ :: EsScribeCfg
2874 -> Server
2975 -> IndexName
3076 -> MappingName
31− -> EsQueueSize
3277 -> Severity
3378 -> Verbosity
3479 -> IO Scribe
35−mkEsScribe ms server ix mapping qs sev verb = do
36− q <- newTBQueueIO $ unEsQueueSize qs
37− withManager ms $ \mgr -> do
80+mkEsScribe cfg@EsScribeCfg {..} server ix mapping sev verb = do
81+ q <- newTBQueueIO $ unEsQueueSize essQueueSize
82+ withManager essManagerSettings $ \mgr -> do
3883 let env = BHEnv { bhServer = server
3984 , bhManager = mgr
4085 }
41− worker <- startWorker env ix mapping q
42− link worker
43− return $ Scribe $ \ i -> do
44− did <- mkDocId
86+ replicateM_ (unEsPoolSize essPoolSize) $ forkIO $
87+ startWorker cfg env ix mapping q
88+
89+ return $ Scribe $ \ i ->
4590 when (_itemSeverity i >= sev) $
4691 void $ atomically $ tryWriteTBQueue q (itemJson verb i)
4792
@@ -54,32 +99,56 @@ mkDocId = (DocId . T.decodeUtf8 . toASCIIBytes) `fmap` randomIO
5499-------------------------------------------------------------------------------
55100newtype EsQueueSize = EsQueueSize {
56101 unEsQueueSize :: Int
57− } deriving (Show, Eq, Ord, Bounded)
102+ } deriving (Show, Eq, Ord)
103+
104+
105+instance Bounded EsQueueSize where
106+ minBound = EsQueueSize 1
107+ maxBound = EsQueueSize maxBound
58108
59109
60110mkEsQueueSize :: Int -> Maybe EsQueueSize
61−mkEsQueueSize n
62− | n >= 0 = Just $ EsQueueSize n
111+mkEsQueueSize = mkNonZero EsQueueSize
112+
113+
114+-------------------------------------------------------------------------------
115+newtype EsPoolSize = EsPoolSize {
116+ unEsPoolSize :: Int
117+ } deriving (Show, Eq, Ord)
118+
119+
120+instance Bounded EsPoolSize where
121+ minBound = EsPoolSize 1
122+ maxBound = EsPoolSize maxBound
123+
124+
125+mkEsPoolSize :: Int -> Maybe EsPoolSize
126+mkEsPoolSize = mkNonZero EsPoolSize
127+
128+
129+-------------------------------------------------------------------------------
130+mkNonZero :: (Int -> a) -> Int -> Maybe a
131+mkNonZero ctor n
132+ | n > 0 = Just $ ctor n
63133 | otherwise = Nothing
64134
65135
66136-------------------------------------------------------------------------------
67137startWorker
68− :: BHEnv
138+ :: EsScribeCfg
139+ -> BHEnv
69140 -> IndexName
70141 -> MappingName
71142 -> TBQueue Value
72− -> IO (Async ())
73−startWorker env ix mapping q = async $ forever $ do
143+ -> IO ()
144+startWorker EsScribeCfg {..} env ix mapping q = forever $ do
74145 v <- atomically $ readTBQueue q
75− res <- waitCatch =<< async (sendLog v)
76− when (isLeft res) inCaseOfEmergency
146+ sendLog v `catchAny` eat
77147 where
78− sendLog v = void $ do
148+ sendLog v = void $ recoverAll essRetryPolicy $ do
79149 did <- mkDocId
80150 runBH env $ indexDocument ix mapping v did
81− --TODO reenqueue? drop? would be nice to log but we don't have that context
82− inCaseOfEmergency = return ()
151+ eat _ = return ()
83152
84153
85154-------------------------------------------------------------------------------
@@ -88,9 +157,3 @@ tryWriteTBQueue q v = do
88157 ok <- not <$> isFullTBQueue q
89158 when ok $ writeTBQueue q v
90159 return ok
91−
92−
93−-------------------------------------------------------------------------------
94−isLeft :: Either a b -> Bool
95−isLeft (Left _) = True
96−isLeft _ = False