hslogstash 0.2.3.1 → 0.3.0
raw patch · 5 files changed
+127/−20 lines, 5 filesdep +stm
Dependencies added: stm
Files
- Data/Conduit/Branching.hs +29/−2
- Data/Conduit/Misc.hs +31/−6
- Data/Conduit/Redis.hs +11/−9
- Logstash/Counter.hs +53/−0
- hslogstash.cabal +3/−3
Data/Conduit/Branching.hs view
@@ -1,6 +1,33 @@-{-| Branching conduits+{-| @WARNING: executables using this function must be compiled with -threaded@ -WARNING: executables using this function must be compiled with -threaded+These functions let you connect several sinks to a single source, according to a branching strategy. For example :++@+module Main where++import Data.Conduit.Branching+import Data.Conduit+import qualified Data.Conduit.List as CL+import Control.Monad.IO.Class++src :: Monad m => Producer m (Either Int String)+src = CL.sourceList [Left 5, Left 4, Right \"five\", Right \"four\"]++sinkString :: (Monad m, MonadIO m) => Sink (Either Int String) m ()+sinkString = CL.mapM_ $ \(Right x) -> liftIO (putStrLn (\"This is a string: \" ++ x))++sinkInt :: (Monad m, MonadIO m) => Sink (Either Int String) m ()+sinkInt = CL.mapM_ $ \(Left x) -> liftIO (putStrLn (\"This is an integer: \" ++ show x))++sinkLog :: (Monad m, MonadIO m) => Sink (Either Int String) m ()+sinkLog = CL.mapM_ (liftIO . putStrLn . (\"Raw logging: \" ++) . show)++main :: IO ()+main = branchConduits src branching [sinkInt, sinkString, sinkLog]+ where+ branching (Left _) = [0,2]+ branching (Right _) = [1,2]+@ -} module Data.Conduit.Branching (mkBranchingConduit, branchConduits) where
Data/Conduit/Misc.hs view
@@ -10,18 +10,43 @@ -- | 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-concatFlush :: (Monad m) => Integer -> Conduit [a] m (Flush a)-concatFlush mx = concatFlush' 0+simpleConcatFlush :: (Monad m) => Integer -> Conduit [a] m (Flush a)+simpleConcatFlush mx = concatFlush 0 sendf where- concatFlush' x = await >>= maybe (return ()) (\input -> if null input- then yield Flush >> concatFlush' 0- else foldM sendf x input >>= concatFlush'- ) sendf curx ev = do yield (Chunk ev) if curx >= mx then yield Flush >> return 0 else return (curx+1)++-- | This is a more general version of 'simpleConcatFlush', where you+-- provide your own fold.+concatFlush :: (Monad m) => b -> (b -> a -> ConduitM [a] (Flush a) m b) -> Conduit [a] m (Flush a)+concatFlush initial foldfunc = concatFlush' initial+ where+ concatFlush' x = await >>= maybe (return ()) (\input -> if null input+ then yield Flush >> concatFlush' initial+ else foldM foldfunc x input >>= concatFlush')++-- | A generalized version of 'simpleConcatFlush' where some value is+-- summed and the 'Flush' is sent when it reaches a threshold.+concatFlushSum :: Monad m+ => (a -> Integer) -- ^ Convert your input value into an Integer, usually a size+ -> Integer -- ^ The threshold value+ -> Conduit [a] m (Flush a)+concatFlushSum tolength maxlength = concatFlush 0 foldfunc+ where+ foldfunc curlength element = do+ let nextlength = curlength + elementlength+ elementlength = tolength element+ if nextlength > maxlength+ then do+ yield Flush+ yield (Chunk element)+ return elementlength+ else do+ yield (Chunk element)+ return nextlength -- | Regroup a stream of (Flush a) into a stream of lists, using "Flush" as -- the separator
Data/Conduit/Redis.hs view
@@ -11,7 +11,7 @@ import Control.Monad (void,replicateM, forever) import Control.Monad.IO.Class (liftIO, MonadIO) import Data.Either (rights)-import Data.Maybe (catMaybes)+import Data.Maybe (catMaybes,fromMaybe,isNothing) import Control.Exception import Control.Concurrent hiding (yield) @@ -22,27 +22,27 @@ mp (Right Nothing) = Nothing mp x = trace (show x) Nothing -safePush :: BS.ByteString -> MVar Connection -> ConnectInfo -> BS.ByteString -> IO ()-safePush list mconn cinfo input = catch mypush (\SomeException{} -> resetMVar)+safePush :: BS.ByteString -> MVar Connection -> ConnectInfo -> (BS.ByteString -> IO ()) -> BS.ByteString -> IO ()+safePush list mconn cinfo logfn input = catch mypush (\SomeException{} -> resetMVar) where resetMVar = do void $ takeMVar mconn connect cinfo >>= putMVar mconn- BS.putStrLn "reconnecting to redis server ..."+ logfn "reconnecting to redis server ..." threadDelay 500000- safePush list mconn cinfo input+ safePush list mconn cinfo logfn input mypush = do conn <- readMVar mconn x <- runRedis conn (lpush list [input]) case x of Left (SingleLine "OK") -> return () Right _ -> return ()- err -> BS.putStrLn ("retrying ... " `BS.append` BS.pack (show err)) >> threadDelay 500000 >> safePush list mconn cinfo input+ err -> logfn ("retrying ... " `BS.append` BS.pack (show err)) >> threadDelay 500000 >> safePush list mconn cinfo logfn input popN :: BS.ByteString -> Int -> Integer -> Redis [BS.ByteString] popN l n to = do f <- fmap mp (brpop [l] to)- if f == Nothing+ if isNothing f then return [] -- short circuiting else do nx <- fmap rights $ replicateM (n-1) (rpop l)@@ -66,8 +66,10 @@ => 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 ! -> Sink BS.ByteString m ()-redisSink h p list =+redisSink h p list logcmd = let cinfo = defaultConnectInfo { connectHost = h, connectPort = PortNumber $ fromIntegral p }- in bracketP (connect cinfo >>= newMVar) (const $ return ()) (\mconn -> CL.mapM_ (liftIO . safePush list mconn cinfo))+ logfunc = fromMaybe BS.putStrLn logcmd+ in bracketP (connect cinfo >>= newMVar) (const $ return ()) (\mconn -> CL.mapM_ (liftIO . safePush list mconn cinfo logfunc))
+ Logstash/Counter.hs view
@@ -0,0 +1,53 @@+module Logstash.Counter where++import Control.Concurrent.STM+import Data.Conduit+import qualified Data.Conduit.List as CL+import Control.Monad.IO.Class+import Control.Concurrent+import System.IO+import Network.Socket+import Control.Monad+import Control.Exception+import Data.Time.Clock.POSIX+import qualified Data.Text as T+import qualified Data.Text.IO as T+import Data.Monoid++newtype Counter = Counter (TVar Integer)++newCounter :: IO Counter+newCounter = fmap Counter $ newTVarIO 0++incrementCounterConduit :: (Monad m, MonadIO m) => Counter -> Conduit a m a+incrementCounterConduit c = CL.iterM (const $ liftIO $ incrementCounter c)++incrementCounter :: Counter -> IO ()+incrementCounter (Counter c) = atomically $ modifyTVar c (+1)++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+ -> IO ()+counter2collectd c sockpath nodename plugin vinstance = void $ forkIO work+ where+ hdr = "PUTVAL " <> T.pack nodename <> "/" <> T.pack plugin <> "/derive-" <> T.pack vinstance <> " interval=10 "+ collectdConnect :: IO Handle+ collectdConnect = do+ soc <- socket AF_UNIX Stream 0+ connect soc (SockAddrUnix sockpath)+ socketToHandle soc ReadWriteMode+ work = do+ bracket collectdConnect hClose $ \h -> do+ v <- readCounter c+ tt <- fmap (T.pack . show . (truncate :: (RealFrac a) => a -> Integer)) getPOSIXTime+ T.hPutStrLn h (hdr <> tt <> ":" <> T.pack (show v))+ void $ T.hGetLine h+ threadDelay 10000000+ work+
hslogstash.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: hslogstash-version: 0.2.3.1+version: 0.3.0 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@@ -19,8 +19,8 @@ location: git://github.com/bartavelle/hslogstash.git library- exposed-modules: Logstash.Message, Logstash.IO, Data.Conduit.Redis, Data.Conduit.ElasticSearch, Data.Conduit.Logstash, Data.Conduit.Network.Retry, Data.Conduit.Branching, Data.Conduit.Misc+ exposed-modules: Logstash.Message, Logstash.IO, Data.Conduit.Redis, Data.Conduit.ElasticSearch, Data.Conduit.Logstash, Data.Conduit.Network.Retry, Data.Conduit.Branching, Data.Conduit.Misc, Logstash.Counter 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+ 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