@@ -1,8 +1,17 @@
1−module Katip.Scribes.ElasticSearch where
1+module Katip.Scribes.ElasticSearch
2+ ( mkEsScribe
3+ , mkDocId
4+ , EsQueueSize
5+ , mkEsQueueSize
6+ ) where
27
38
49-------------------------------------------------------------------------------
10+import Control.Applicative
11+import Control.Concurrent.Async
12+import Control.Concurrent.STM
513import Control.Monad
14+import Data.Aeson
615import Data.UUID
716import Database.Bloodhound
817import System.Random
@@ -16,13 +25,18 @@ mkEsScribe
1625 :: Server
1726 -> IndexName
1827 -> MappingName
28+ -> EsQueueSize
1929 -> Severity
2030 -> Verbosity
2131 -> IO Scribe
22−mkEsScribe server ix mapping sev verb = return $ Scribe $ \ i -> do
23− did <- mkDocId
24− when (_itemSeverity i >= sev) $
25− void $ indexDocument server ix mapping (itemJson verb i) did
32+mkEsScribe server ix mapping qs sev verb = do
33+ q <- newTBQueueIO $ unEsQueueSize qs
34+ worker <- startWorker server ix mapping q
35+ link worker
36+
37+ return $ Scribe $ \ i ->
38+ when (_itemSeverity i >= sev) $
39+ void $ atomically $ tryWriteTBQueue q (itemJson verb i)
2640
2741
2842-------------------------------------------------------------------------------
@@ -30,8 +44,44 @@ mkDocId :: IO DocId
3044mkDocId = (DocId . toString) `fmap` randomIO
3145
3246
47+-------------------------------------------------------------------------------
48+newtype EsQueueSize = EsQueueSize {
49+ unEsQueueSize :: Int
50+ } deriving (Show, Eq, Ord, Bounded)
3351
3452
53+mkEsQueueSize :: Int -> Maybe EsQueueSize
54+mkEsQueueSize n
55+ | n >= 0 = Just $ EsQueueSize n
56+ | otherwise = Nothing
3557
3658
59+-------------------------------------------------------------------------------
60+startWorker
61+ :: Server
62+ -> IndexName
63+ -> MappingName
64+ -> TBQueue Value
65+ -> IO (Async ())
66+startWorker server ix mapping q = async $ forever $ do
67+ v <- atomically $ readTBQueue q
68+ res <- waitCatch =<< async (sendLog v)
69+ when (isLeft res) $ inCaseOfEmergency
70+ where
71+ sendLog v = void $ indexDocument server ix mapping v =<< mkDocId
72+ --TODO reenqueue? drop? would be nice to log but we don't have that context
73+ inCaseOfEmergency = return ()
3774
75+
76+-------------------------------------------------------------------------------
77+tryWriteTBQueue :: TBQueue a -> a -> STM Bool
78+tryWriteTBQueue q v = do
79+ ok <- not <$> isFullTBQueue q
80+ when ok $ writeTBQueue q v
81+ return ok
82+
83+
84+-------------------------------------------------------------------------------
85+isLeft :: Either a b -> Bool
86+isLeft (Left _) = True
87+isLeft _ = False