hslogstash 0.3.1 → 0.3.2
raw patch · 10 files changed
+164/−34 lines, 10 filesdep +QuickCheckdep +hslogstashdep +hspecdep ~conduit
Dependencies added: QuickCheck, hslogstash, hspec, split, stm-chans
Dependency ranges changed: conduit
Files
- Data/Conduit/Branching.hs +16/−10
- Data/Conduit/ElasticSearch.hs +6/−5
- Data/Conduit/Logstash.hs +10/−1
- Data/Conduit/Misc.hs +5/−3
- Data/Conduit/Network/Retry.hs +1/−2
- Data/Conduit/Redis.hs +8/−2
- Logstash/Counter.hs +31/−6
- Logstash/Message.hs +8/−3
- hslogstash.cabal +12/−2
- tests/conduits.hs +67/−0
Data/Conduit/Branching.hs view
@@ -37,38 +37,44 @@ import Control.Monad.IO.Class (liftIO) import qualified Data.IntMap as IntMap import Data.Maybe (mapMaybe)-import Control.Concurrent.ParallelIO-import Data.Conduit.TMChan+import Control.Concurrent.ParallelIO.Local+import Control.Concurrent.STM.TBMQueue+import Data.Conduit.TQueue import GHC.Conc (atomically) mkSink :: (MonadResource m) => (a -> [Int])- -> [TBMChan a]+ -> [TBMQueue a] -> Sink a m () mkSink brfunc chans = let cmap = IntMap.fromList $ zip [0..] chans -- creates an intmap with all output channels- querymap x = IntMap.lookup x cmap -- query the intmap to get a Maybe (TBMChan a)- cleanup = mapM_ (liftIO . atomically . closeTBMChan) chans -- this is the cleanup function that closes all chans+ querymap x = IntMap.lookup x cmap -- query the intmap to get a Maybe (TQueue a)+ cleanup = mapM_ (liftIO . atomically . closeTBMQueue) chans -- this is the cleanup function that closes all chans inject input = let outchans = mapMaybe querymap (brfunc input) -- compiles the list of output channels- in mapM_ (\c -> liftIO $ atomically $ writeTBMChan c input) outchans -- and write into them+ in mapM_ (\c -> liftIO $ atomically $ writeTBMQueue c input) outchans -- and write into them in addCleanup (const cleanup) (CL.mapM_ inject) +-- | Creates the /plumbing/ that might be used to connect several conduits+-- together, based on a branching function. mkBranchingConduit :: (MonadResource m) => Int -- ^ Number of branches -> (a -> [Int]) -- ^ Branching function, where 0 is the first branch -> IO (Sink a m (), [Source m a]) -- ^ Returns a sink and N sources mkBranchingConduit nbbranches brfunction = do- chans <- replicateM nbbranches (newTBMChanIO 16)- return (mkSink brfunction chans, map sourceTBMChan chans)+ chans <- replicateM nbbranches (newTBMQueueIO 16)+ return (mkSink brfunction chans, map sourceTBMQueue chans) +-- | A higher level function. Given a source, a branching function and+-- a list of sinks, this will run the conduits until completion. branchConduits :: Source (ResourceT IO) a -- ^ The source to branch from -> (a -> [Int]) -- ^ The branching function (0 is the first sink) -> [Sink a (ResourceT IO) ()] -- ^ The destination sinks -> IO () -- ^ Results of the sinks branchConduits src brfunc sinks = do- (newsink, sources) <- mkBranchingConduit (length sinks) brfunc+ let nbsinks = length sinks+ (newsink, sources) <- mkBranchingConduit nbsinks brfunc let srcconduit = src $$ newsink dstconduits = map (uncurry ($$)) (zip sources sinks) actions = map runResourceT (srcconduit : dstconduits)- parallel_ actions+ withPool (nbsinks + 1) $ \pool -> parallel_ pool actions
Data/Conduit/ElasticSearch.hs view
@@ -1,6 +1,6 @@ {-# LANGUAGE RankNTypes #-}-{-| This module exports "Conduit" interfaces to ElasticSearch. It is-totally experimental.+{-| This module exports "Conduit" interfaces to ElasticSearch.+It has been used intensively in production for several month now, but at a single site. -} module Data.Conduit.ElasticSearch (esConduit, esSearchSource) where @@ -69,6 +69,7 @@ <*> v .: "took" parseJSON _ = mzero +-- | A source of Logstash messages generated from an ElasticSearch query. esSearchSource :: (MonadResource m) => Maybe (Request m) -- ^ Defaults parameters for the http request to ElasticSearch. Use "Nothing" for defaults. -> BS.ByteString -- ^ Hostname of the ElasticSearch server -> Int -- ^ Port of the HTTP interface (usually 9200)@@ -107,9 +108,9 @@ safeQuery :: Request (ResourceT IO) -> IO (Response BSL.ByteString) safeQuery req = catch (withManager $ httpLbs req) (\e -> print (e :: SomeException) >> threadDelay 500000 >> safeQuery req) --- | Takes a "LogstashMessage", and returns the result of the ES request--- along with the value in case of errors, or ES's values in case of--- success+-- | Takes a "LogstashMessage", and returns the result of the ElasticSearch request+-- along with the value in case of errors, or ElasticSearch's values in case of+-- success. esConduit :: (MonadResource m) => Maybe (Request m) -- ^ Defaults parameters for the http request to ElasticSearch. Use "Nothing" for defaults. -> BS.ByteString -- ^ Hostname of the ElasticSearch server -> Int -- ^ Port of the HTTP interface (usually 9200)
Data/Conduit/Logstash.hs view
@@ -1,3 +1,5 @@+-- | Receive logstash messages from the network, and process them with+-- a conduit. module Data.Conduit.Logstash (logstashListener) where import Data.Conduit@@ -11,6 +13,8 @@ import Data.Text.Encoding import Logstash.Message +-- | Decodes JSON data from ByteStrings that can be encoded in UTF-8 or+-- latin1. tryDecode :: (FromJSON a) => BS.ByteString -> Either BS.ByteString a tryDecode i = let latin1 = convert "LATIN1" "UTF-8" li@@ -22,5 +26,10 @@ Just x -> Right x Nothing -> Left i -logstashListener :: Int -> Sink (Either BS.ByteString LogstashMessage) (ResourceT IO) () -> IO ()+-- | This creates a logstash network listener, given a TCP port.+-- It will try to decode the Bytestring as UTF-8, and, if it fails, as+-- Latin1.+logstashListener :: Int -- ^ Port number+ -> Sink (Either BS.ByteString LogstashMessage) (ResourceT IO) ()+ -> IO () logstashListener port sink = runResourceT $ runTCPServer (serverSettings port HostAny) (\app -> appSource app $= CB.lines $= CL.map tryDecode $$ sink)
Data/Conduit/Misc.hs view
@@ -1,3 +1,5 @@+-- | Various conduit functions, mostly related to grouping or separating+-- the items handled by conduits. module Data.Conduit.Misc where import Data.Conduit@@ -10,12 +12,12 @@ -- | Converts a stream of [a] into a stream of (Flush a). This is done by -- sending a Flush when the input is the empty list, or that we reached -- a certain threshold-simpleConcatFlush :: (Monad m) => Integer -> Conduit [a] m (Flush a)+simpleConcatFlush :: (Monad m) => Int -> Conduit [a] m (Flush a) simpleConcatFlush mx = concatFlush 0 sendf where sendf curx ev = do yield (Chunk ev)- if curx >= mx+ if curx >= (mx - 1) then yield Flush >> return 0 else return (curx+1) @@ -53,7 +55,7 @@ groupFlush :: (Monad m) => Conduit (Flush a) m [a] groupFlush = grouper [] where- grouper lst = await >>= maybe (return ()) (handle lst)+ grouper lst = await >>= maybe (unless (null lst) (yield (reverse lst))) (handle lst) handle lst Flush = do unless (null lst) (yield (reverse lst)) grouper []
Data/Conduit/Network/Retry.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}+{-| Network conduits that will retry sending messages forever -} module Data.Conduit.Network.Retry where import Prelude hiding (catch)@@ -20,8 +21,6 @@ every time the call to 'sendAll' fails. This means that some bytes might be sent multiple times, if the socket fails in the middle of the sendAll call. This is targeted at protocols where only a full message makes sense.--This is used to send a full JSON object to Logstash. -} sinkSocketRetry :: MonadResource m => IO Socket -> Int -> IO () -> Consumer ByteString m () sinkSocketRetry mkSocket delay exeptionCallback =
Data/Conduit/Redis.hs view
@@ -48,6 +48,11 @@ nx <- fmap rights $ replicateM (n-1) (rpop l) return $ catMaybes (f:nx) +-- | This is a source that pops elements from a Redis list. It is capable+-- of poping several elements at once, and will return lists of+-- ByteStrings. You might then use 'Data.Conduit.Misc.concat' or the+-- flushing facilities in "Data.Conduit.Misc" to work with individual+-- elements. redisSource :: (MonadResource m) => HostName -- ^ Hostname of the Redis server -> Int -- ^ Port of the Redis server (usually 6379)@@ -61,12 +66,13 @@ myPipe conn = forever (liftIO (runRedis conn (popN list nb to)) >>= yield) in bracketP (connect cinfo) (\conn -> runRedis conn (void quit)) myPipe --- | Warning, this outputs strings when things go wrong!+-- | A Sink that will let you write ByteStrings to a redis queue. It can be+-- augmented with a logging function, that will be able to report errors. redisSink :: (MonadResource m) => HostName -- ^ Hostname of the Redis server -> Int -- ^ Port of the Redis server (usually 6379) -> BS.ByteString -- ^ Name of the list- -> Maybe (BS.ByteString -> IO ()) -- ^ Command used to log various errors. Defaults to BS.putStrLn. It shoud not fail !+ -> Maybe (BS.ByteString -> IO ()) -- ^ Command used to log various errors. Defaults to BS.putStrLn. It must not fail, so be careful about exceptions. -> Sink BS.ByteString m () redisSink h p list logcmd = let cinfo = defaultConnectInfo { connectHost = h, connectPort = PortNumber $ fromIntegral p }
Logstash/Counter.hs view
@@ -1,5 +1,19 @@-module Logstash.Counter where+{-| This module is not very well named, as it has almost nothing to do with+Logstash. It is used to define counters that will then be logged by collectd. +You should configure collectd to create a Unix socket :++> LoadPlugin unixsock+>+> <Plugin "unixsock">+> SocketFile "/var/run/collectd-unixsock"+> SocketGroup "collectdsocket"+> SocketPerms "0660"+> </Plugin>++-}+module Logstash.Counter (Counter, newCounter, incrementCounterConduit, incrementCounter, readCounter, counter2collectd) where+ import Control.Concurrent.STM import Data.Conduit import qualified Data.Conduit.List as CL@@ -14,25 +28,36 @@ import qualified Data.Text.IO as T import Data.Monoid +-- | The opaque counter type. It is actually just a 'TVar' 'Integer'. newtype Counter = Counter (TVar Integer) +-- | Gives you a new empty counter. newCounter :: IO Counter newCounter = fmap Counter $ newTVarIO 0 +-- | This is a conduits-specific function that will increase a counter for+-- each piece of data that traverses this conduit. It will not alter the+-- data. incrementCounterConduit :: (Monad m, MonadIO m) => Counter -> Conduit a m a incrementCounterConduit c = CL.iterM (const $ liftIO $ incrementCounter c) +-- | Increments a counter. incrementCounter :: Counter -> IO () incrementCounter (Counter c) = atomically $ modifyTVar c (+1) +-- | Retrieve the current value of a counter. readCounter :: Counter -> IO Integer readCounter (Counter c) = readTVarIO c -counter2collectd :: Counter -- the counter to export- -> FilePath -- path to the unix socket- -> String -- name of the node- -> String -- name of the plugin + instance- -> String -- name of the counter instance+{-| This registers a counter to a Collectd server. This can be used in this way :++> counter2collectd nbmsg "/var/run/collectd-unixsock" nodename "logstash-shipper" "messages"+-}+counter2collectd :: Counter -- ^ the counter to export+ -> FilePath -- ^ path to the unix socket+ -> String -- ^ name of the node, usually the server's fully qualified domain name+ -> String -- ^ name of the plugin + instance+ -> String -- ^ name of the counter instance -> IO () counter2collectd c sockpath nodename plugin vinstance = void $ forkIO work where
Logstash/Message.hs view
@@ -1,3 +1,5 @@+-- | Datatypes, helper functions, and JSON instances for Logstash+-- messages. module Logstash.Message where import Data.Aeson@@ -13,7 +15,7 @@ import Data.Attoparsec.Text {-| The Logstash message, as described in <https://github.com/logstash/logstash/wiki/logstash's-internal-message-format>.-Please not there is no timestamp, as the logstash server will add it.+Please note that it is good practice to forget about the timestamp when creating messages (set 'logstashTime' to 'Nothing'), as it should be a responsability of the Logstash server to add it. -} data LogstashMessage = LogstashMessage { logstashType :: T.Text@@ -63,7 +65,7 @@ (seconds, fminutes) = reduce fseconds 60 (minutes, hours) = reduce fminutes 60 --- | This parses the logstash format+-- | This parses the logstash time format. parseLogstashTime :: T.Text -> Maybe UTCTime parseLogstashTime t = case parseOnly prs t of Right r -> Just r@@ -124,6 +126,9 @@ curtime <- getCurrentTime return msg { logstashTime = Just curtime } -addLogstashTag :: T.Text -> LogstashMessage -> LogstashMessage+-- | Adds a tag to a logstash message.+addLogstashTag :: T.Text -- ^ The tag to add+ -> LogstashMessage+ -> LogstashMessage addLogstashTag tag msg = msg { logstashTags = tag : logstashTags msg }
hslogstash.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: hslogstash-version: 0.3.1+version: 0.3.2 synopsis: A library to work with, or as, a logstash server description: This library contains a few modules that let you work with Logstash messages, read them from a Redis list, store them in or get them from Elasticsearch, and more. license: BSD3@@ -23,4 +23,14 @@ extensions: OverloadedStrings, BangPatterns ghc-options: -Wall -- other-modules: - build-depends: base <5, aeson, network, bytestring, text, vector, unordered-containers, time, text-format, attoparsec, hedis, conduit >= 1.0.0, transformers, http-conduit, iconv, network-conduit, containers, parallel-io, stm-conduit, http-types, stm+ build-depends: base <5, aeson, network, bytestring, text, vector, unordered-containers, time, text-format, attoparsec, hedis, conduit >= 1.0.0, transformers, http-conduit, iconv, network-conduit, containers, parallel-io, stm-conduit, http-types, stm, stm-chans++Test-Suite conduits+ hs-source-dirs: tests+ type: exitcode-stdio-1.0+ ghc-options: -Wall -rtsopts -threaded+ extensions: OverloadedStrings+ build-depends: base <5, hslogstash, conduit, QuickCheck, hspec, transformers, stm, split+ main-is: conduits.hs++
+ tests/conduits.hs view
@@ -0,0 +1,67 @@+module Main where++import Test.Hspec+import Data.Conduit+import qualified Data.Conduit.List as CL+import Data.Conduit.Misc as CM+import Test.QuickCheck+import Data.Functor.Identity+import Control.Monad.IO.Class+import Data.Conduit.Branching+import Control.Concurrent.STM+import Data.List.Split (chunksOf,splitOn)+import Data.Either (lefts,rights)+import Data.Maybe (catMaybes)++main :: IO ()+main = hspec $ do+ describe "concat" $ do+ it "should work just like Prelude.concat" $ property $ \x ->+ let y = runIdentity $ CL.sourceList (x :: [[Int]]) $= CM.concat $$ CL.consume+ in y == Prelude.concat x+ describe "simpleConcatFlush" $ do+ it "should behave like expected for simple lists" $ property $ \x -> do+ chunksize <- suchThat arbitrary (>0)+ let y = runIdentity $ CL.sourceList (x :: [[Int]]) $= simpleConcatFlush chunksize $= groupFlush $$ CL.consume+ toMaybeList [] = [Nothing]+ toMaybeList k = map Just k+ z = concatMap (chunksOf chunksize . catMaybes) $ splitOn [Nothing] $ concatMap toMaybeList x+ return (y == z)+ it "should separate a list when max = 1" $ do+ x <- CL.sourceList [ [1,2,3,4,5,6] :: [Int] ] $= simpleConcatFlush 1 $= groupFlush $$ CL.consume+ x `shouldBe` map return [1,2,3,4,5,6]+ it "should pair values" $ do+ x <- CL.sourceList [ [1,2,3,4,5,6] :: [Int] ] $= simpleConcatFlush 2 $= groupFlush $$ CL.consume+ x `shouldBe` [ [1,2] , [3,4] , [5,6] ]+ it "should not alter small lists" $ do+ x <- CL.sourceList [ [1,2,3,4,5,6] :: [Int] ] $= simpleConcatFlush 10 $= groupFlush $$ CL.consume+ x `shouldBe` [ [1,2,3,4,5,6] ]+ it "should correctly group values" $ do+ x <- CL.sourceList [ [1,2] , [3,4,5,6], [7,8,9] :: [Int] ] $= simpleConcatFlush 4 $= groupFlush $$ CL.consume+ x `shouldBe` [ [1,2,3,4],[5,6,7,8],[9] ]+ it "should flush on empty input" $ do+ x <- CL.sourceList [ [1,2] , [], [3,4,5], [6], [7,8,9], [] :: [Int] ] $= simpleConcatFlush 4 $= groupFlush $$ CL.consume+ x `shouldBe` [ [1,2], [3,4,5,6],[7,8,9] ]+ describe "concatFlushSum" $ do+ it "should separate input at the right place" $ do+ x <- CL.sourceList [ [1,2,0,0,0,6,3,2,0,0,1] :: [Int] ] $= concatFlushSum id 3 $= groupFlush $$ CL.consume+ x `shouldBe` [[1,2,0,0,0],[6],[3],[2,0,0,1]]+ describe "branching" $ do+ it "should branch simple eithers" $ property $ \lst -> do+ tsink0 <- newTVarIO []+ tsink1 <- newTVarIO []+ tsink2 <- newTVarIO []+ let source = CL.sourceList (lst :: [Either Int String])+ sappend f s = CL.mapM_ (\x -> liftIO $ atomically $ modifyTVar s (f x :))+ sink0 = sappend (\(Left x) -> x) tsink0+ sink1 = sappend (\(Right x) -> x) tsink1+ sink2 = sappend id tsink2+ branching (Left _) = [0,2]+ branching (Right _) = [1,2]+ branchConduits source branching [sink0, sink1, sink2]+ o0 <- atomically (readTVar tsink0)+ o1 <- atomically (readTVar tsink1)+ o2 <- atomically (readTVar tsink2)+ let result = (o0,o1,o2)+ expected = (reverse (lefts lst),reverse (rights lst),reverse lst)+ result `shouldBe` expected