git.iko.soy/typeable/katip
on master help sign in
Initial work on async buffer
b53d122Michael Xavier
katip-elasticsearch/katip-elasticsearch.cabal+10−7
compare with master
unifiedsplit
@@ -1,7 +1,7 @@
1name: katip-elasticsearch
1name: katip-elasticsearch
2version: 0.1.0.0
2version: 0.1.0.0
3−-- synopsis:
3+-- synopsis:
4−-- description:
4+-- description:
5license: BSD3
5license: BSD3
6license-file: LICENSE
6license-file: LICENSE
7author: Ozgun Ataman
7author: Ozgun Ataman
@@ -12,15 +12,18 @@ build-type: Simple
12cabal-version: >=1.10
12cabal-version: >=1.10
13
13
14library
14library
15− exposed-modules:
15+ exposed-modules:
16 Katip.Scribes.ElasticSearch
16 Katip.Scribes.ElasticSearch
17− -- other-modules:
17+ -- other-modules:
18− -- other-extensions:
18+ -- other-extensions:
19− build-depends:
19+ build-depends:
20 base >=4.6 && <4.7
20 base >=4.6 && <4.7
21 , katip
21 , katip
22 , bloodhound >= 0.5 && < 0.6
22 , bloodhound >= 0.5 && < 0.6
23 , random >= 1.0 && < 1.2
23 , random >= 1.0 && < 1.2
24 , uuid >= 1.3 && < 1.4
24 , uuid >= 1.3 && < 1.4
25+ , aeson >=0.6 && <0.9
26+ , stm >= 2.4.3 && < 2.5
27+ , async >= 2.0.1.0 && < 2.1
25 hs-source-dirs: src
28 hs-source-dirs: src
26− default-language: Haskell2010
29+ default-language: Haskell2010
katip-elasticsearch/src/Katip/Scribes/ElasticSearch.hs+55−5
@@ -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
2
7
3
8
4-------------------------------------------------------------------------------
9-------------------------------------------------------------------------------
10+import Control.Applicative
11+import Control.Concurrent.Async
12+import Control.Concurrent.STM
5import Control.Monad
13import Control.Monad
14+import Data.Aeson
6import Data.UUID
15import Data.UUID
7import Database.Bloodhound
16import Database.Bloodhound
8import System.Random
17import System.Random
@@ -16,13 +25,18 @@ mkEsScribe
16 :: Server
25 :: Server
17 -> IndexName
26 -> IndexName
18 -> MappingName
27 -> MappingName
28+ -> EsQueueSize
19 -> Severity
29 -> Severity
20 -> Verbosity
30 -> Verbosity
21 -> IO Scribe
31 -> IO Scribe
22−mkEsScribe server ix mapping sev verb = return $ Scribe $ \ i -> do
32+mkEsScribe server ix mapping qs sev verb = do
23− did <- mkDocId
33+ q <- newTBQueueIO $ unEsQueueSize qs
34+ worker <- startWorker server ix mapping q
35+ link worker
36+
37+ return $ Scribe $ \ i ->
24− when (_itemSeverity i >= sev) $
38+ when (_itemSeverity i >= sev) $
25− void $ indexDocument server ix mapping (itemJson verb i) did
39+ void $ atomically $ tryWriteTBQueue q (itemJson verb i)
26
40
27
41
28-------------------------------------------------------------------------------
42-------------------------------------------------------------------------------
@@ -30,8 +44,44 @@ mkDocId :: IO DocId
30mkDocId = (DocId . toString) `fmap` randomIO
44mkDocId = (DocId . toString) `fmap` randomIO
31
45
32
46
47+-------------------------------------------------------------------------------
48+newtype EsQueueSize = EsQueueSize {
49+ unEsQueueSize :: Int
50+ } deriving (Show, Eq, Ord, Bounded)
33
51
34
52
53+mkEsQueueSize :: Int -> Maybe EsQueueSize
54+mkEsQueueSize n
55+ | n >= 0 = Just $ EsQueueSize n
56+ | otherwise = Nothing
35
57
36
58
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 ()
37
74
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