diff --git a/Data/Conduit/Branching.hs b/Data/Conduit/Branching.hs --- a/Data/Conduit/Branching.hs +++ b/Data/Conduit/Branching.hs @@ -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 diff --git a/Data/Conduit/Misc.hs b/Data/Conduit/Misc.hs --- a/Data/Conduit/Misc.hs +++ b/Data/Conduit/Misc.hs @@ -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 diff --git a/Data/Conduit/Redis.hs b/Data/Conduit/Redis.hs --- a/Data/Conduit/Redis.hs +++ b/Data/Conduit/Redis.hs @@ -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)) diff --git a/Logstash/Counter.hs b/Logstash/Counter.hs new file mode 100644 --- /dev/null +++ b/Logstash/Counter.hs @@ -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 + diff --git a/hslogstash.cabal b/hslogstash.cabal --- a/hslogstash.cabal +++ b/hslogstash.cabal @@ -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