diff --git a/Data/Conduit/Branching.hs b/Data/Conduit/Branching.hs --- a/Data/Conduit/Branching.hs +++ b/Data/Conduit/Branching.hs @@ -41,6 +41,7 @@ import Control.Concurrent.STM.TBMQueue import Data.Conduit.TQueue import GHC.Conc (atomically) +import Control.Monad.Trans.Resource mkSink :: (MonadResource m) => (a -> [Int]) diff --git a/Data/Conduit/ElasticSearch.hs b/Data/Conduit/ElasticSearch.hs --- a/Data/Conduit/ElasticSearch.hs +++ b/Data/Conduit/ElasticSearch.hs @@ -9,6 +9,7 @@ import Data.Conduit import qualified Data.Conduit.List as CL import Network.HTTP.Conduit +import Network.HTTP.Client () import Data.Aeson import Control.Applicative import qualified Data.ByteString.Char8 as BS @@ -26,9 +27,8 @@ import Control.Lens hiding ((.=)) import Data.Aeson.Lens import Data.Default +import Control.Monad.Trans.Resource -import qualified Data.HashMap.Strict as HM -import qualified Data.Vector as V import qualified Data.Text as T import qualified Data.Text.Encoding as T @@ -81,7 +81,7 @@ -> Int -- ^ Maximum size of each response -> Int -- ^ start -> Producer m (Either Value [LogstashMessage]) -esSearchSource r h p prefix req maxsize start = self start +esSearchSource r h p prefix req maxsize = self where defR1 = fromMaybe def r defR2 = defR1 { host = h @@ -166,7 +166,7 @@ , port = p , path = "/_bulk" , method = "POST" - , checkStatus = (\_ _ _ -> Nothing) + , checkStatus = \_ _ _ -> Nothing } prepareBS :: LogstashMessage -> Either (LogstashMessage, Value) (LogstashMessage, Value, Value) prepareBS input = @@ -184,18 +184,16 @@ req = defR2 { requestBody = RequestBodyLBS body } in do res' <- liftIO (safeQuery req) - let genericError er = return (Left (emptyLSMessage "error", object [ "error" .= T.pack er, "data" .= res ]) : lerrors) - res = case E.decodeUtf8' (responseBody res') of - Right x -> x ^. strict - Left _ -> "Error, would not decode" - getObject st (Object hh) = HM.lookup st hh - getObject _ _ = Nothing - fst3 (a,_,_) = a - items = getObject "items" (String res) - extractErrors x = if (getObject "create" (snd x) >>= getObject "ok") == Just (Bool True) - then Right (snd x) - else Left x - case items of - Just (Array v) -> return $ map extractErrors (zip (map fst3 tosend) (V.toList v)) ++ lerrors - _ -> genericError "Can't find items" + let genericError er = return (Left (emptyLSMessage "error", object [ ("error", String er), ("data", String text) ]) : lerrors) + items = responseBody res' ^.. _Object . ix "items" . _Array . traverse + text = case E.decodeUtf8' (responseBody res') of + Right x -> x ^. strict + Left _ -> "Error, would not decode" + extractErrors :: (LogstashMessage, Value, Value) -> Value -> Either (LogstashMessage, Value) Value + extractErrors v' res = case res ^? _Object . ix "create" . _Object . ix "status" . _Number of + Just 201 -> Right (v' ^. _3) + _ -> Left (v' ^. _1, res) + if length items /= length tosend + then genericError ("Can't find items " <> text) + else return $ zipWith extractErrors tosend items ++ lerrors diff --git a/Data/Conduit/FireHose.hs b/Data/Conduit/FireHose.hs --- a/Data/Conduit/FireHose.hs +++ b/Data/Conduit/FireHose.hs @@ -16,6 +16,7 @@ import Data.Aeson import Blaze.ByteString.Builder.ByteString import Data.Monoid +import Data.Maybe (fromMaybe) {-| A web server will be launched on the specified port. Clients can request URLs of the form /type1,type2,type3. They will be fed all @@ -39,6 +40,6 @@ serialize = (<> fromByteString "\n") . fromLazyByteString . encode getFilter r = case pathInfo r of [p] -> let set = HS.fromList $ T.splitOn "," p - in flip HS.member set . logstashType + in flip HS.member set . fromMaybe "empty" . logstashType _ -> error "invalid url" diff --git a/Data/Conduit/Logstash.hs b/Data/Conduit/Logstash.hs --- a/Data/Conduit/Logstash.hs +++ b/Data/Conduit/Logstash.hs @@ -12,6 +12,7 @@ import Codec.Text.IConv import Data.Text.Encoding import Logstash.Message +import Control.Monad.Trans.Resource -- | Decodes JSON data from ByteStrings that can be encoded in UTF-8 or -- latin1. @@ -32,4 +33,4 @@ 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) +logstashListener port sink = runTCPServer (serverSettings port "*") (\app -> runResourceT (appSource app $= CB.lines $= CL.map tryDecode $$ sink)) diff --git a/Data/Conduit/Network/Retry.hs b/Data/Conduit/Network/Retry.hs --- a/Data/Conduit/Network/Retry.hs +++ b/Data/Conduit/Network/Retry.hs @@ -6,8 +6,6 @@ import Prelude hiding (catch) import Data.Conduit -import Data.Conduit.Network -import Network.Socket (Socket, close) import Network.Socket.ByteString (sendAll) import Control.Monad.IO.Class (MonadIO (liftIO)) import Control.Exception @@ -16,6 +14,9 @@ import Control.Concurrent.MVar import Control.Monad ((>=>)) import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Resource +import Network.Socket +import Network.BSD {-| Tentative /safe/ "Sink" for a "Socket". It should try reopening the "Socket" every time the call to 'sendAll' fails. This means that some bytes might be sent @@ -23,10 +24,10 @@ targeted at protocols where only a full message makes sense. -} sinkSocketRetry :: MonadResource m => IO Socket -> Int -> IO () -> Consumer ByteString m () -sinkSocketRetry mkSocket delay exeptionCallback = +sinkSocketRetry mkSock delay exeptionCallback = let safeMkSocket :: IO Socket - safeMkSocket = catch mkSocket (\SomeException{} -> exeptionCallback >> threadDelay delay >> safeMkSocket) + safeMkSocket = catch mkSock (\SomeException{} -> exeptionCallback >> threadDelay delay >> safeMkSocket) safeSend :: MVar Socket -> ByteString -> IO () safeSend s o = do sock <- takeMVar s @@ -40,6 +41,17 @@ in bracketP (safeMkSocket >>= newMVar) (takeMVar >=> close) push -- | A specialization of the previous Sink that opens a TCP connection. -tcpSinkRetry :: MonadResource m => ByteString -> Int -> Int -> IO () -> Consumer ByteString m () -tcpSinkRetry host port = sinkSocketRetry (fmap fst (getSocket host port)) +tcpSinkRetry :: MonadResource m => String -> Int -> Int -> IO () -> Consumer ByteString m () +tcpSinkRetry host port = sinkSocketRetry getSock + where + getSock = do + proto <- getProtocolNumber "tcp" + bracketOnError + (socket AF_INET Stream proto) + sClose + (\sock -> do + he <- getHostByName host + Network.Socket.connect sock (SockAddrInet (fromIntegral port) (hostAddress he)) + return sock + ) diff --git a/Data/Conduit/Redis.hs b/Data/Conduit/Redis.hs --- a/Data/Conduit/Redis.hs +++ b/Data/Conduit/Redis.hs @@ -2,18 +2,20 @@ -} module Data.Conduit.Redis (redisSource, redisSink) where -import Data.Conduit +import Data.Conduit hiding (connect) import qualified Data.Conduit.List as CL import qualified Data.ByteString.Char8 as BS import Network import Control.Concurrent.MVar import Database.Redis hiding (String, decode) import Control.Monad (void,replicateM, forever) -import Control.Monad.IO.Class (liftIO, MonadIO) +import Control.Monad.IO.Class (liftIO) import Data.Either (rights) import Data.Maybe (catMaybes,fromMaybe,isNothing) import Control.Exception import Control.Concurrent hiding (yield) +import Control.Monad.Trans.Resource +import Control.Applicative import Debug.Trace @@ -45,7 +47,7 @@ if isNothing f then return [] -- short circuiting else do - nx <- fmap rights $ replicateM (n-1) (rpop l) + nx <- rights <$> replicateM (n-1) (rpop l) return $ catMaybes (f:nx) -- | This is a source that pops elements from a Redis list. It is capable diff --git a/Logstash/Counter.hs b/Logstash/Counter.hs --- a/Logstash/Counter.hs +++ b/Logstash/Counter.hs @@ -34,7 +34,7 @@ -- | Gives you a new empty counter. newCounter :: IO Counter -newCounter = fmap Counter $ newTVarIO 0 +newCounter = 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 @@ -67,7 +67,7 @@ threadDelay 10000000 soc <- socket AF_UNIX Stream 0 eh <- try $ do - connect soc (SockAddrUnix sockpath) + Network.Socket.connect soc (SockAddrUnix sockpath) socketToHandle soc ReadWriteMode case eh of Left (_ :: SomeException) -> sClose soc diff --git a/Logstash/Message.hs b/Logstash/Message.hs --- a/Logstash/Message.hs +++ b/Logstash/Message.hs @@ -18,21 +18,21 @@ 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 - , logstashSource :: T.Text + { logstashType :: Maybe T.Text + , logstashSource :: Maybe T.Text , logstashTags :: [T.Text] - , logstashFields :: Value + , logstashFields :: Maybe Value , logstashContent :: T.Text , logstashTime :: Maybe UTCTime } deriving (Show, Eq) instance FromJSON LogstashMessage where parseJSON (Object v) = LogstashMessage - <$> v .: "@type" - <*> v .: "@source" - <*> v .: "@tags" - <*> v .: "@fields" - <*> v .: "@message" + <$> v .:? "@type" + <*> v .:? "@source" + <*> v .: "tags" + <*> v .:? "@fields" + <*> v .: "message" <*> v .:? "@timestamp" parseJSON _ = mzero @@ -40,14 +40,20 @@ updating the message field. -} emptyLSMessage :: T.Text -> LogstashMessage -emptyLSMessage m = LogstashMessage "empty" "dummy" [] (object []) m Nothing +emptyLSMessage m = LogstashMessage + (Just "empty") + (Just "dummy") + [] + (Just $ object []) + m + Nothing instance ToJSON LogstashMessage where toJSON (LogstashMessage ty s ta f c ts) = object $ [ "@type" .= ty , "@source" .= s - , "@tags" .= ta + , "tags" .= ta , "@fields" .= f - , "@message" .= c + , "message" .= c ] ++ case ts of Nothing -> [] Just t -> [ "@timestamp" .= t ] @@ -95,10 +101,10 @@ mflds = case HM.lookup "@fields" m of Just x -> x Nothing -> Null - mtags = case HM.lookup "@tags" m of + mtags = case HM.lookup "tags" m of Just (Array v) -> toTags (V.toList v) _ -> Nothing - mmsg = case HM.lookup "@message" m of + mmsg = case HM.lookup "message" m of Just (String x) -> x _ -> "" mts = case HM.lookup "@timestamp" m of @@ -110,11 +116,11 @@ isString _ = False toText (String x) = x toText _ = "" - in if null (filter (not . isString) v) + in if not (any (not . isString) v) then Just (map toText v) else Nothing in case (mtype, msrc, mtags) of - (Just (String t), Just (String s), Just tags) -> Just $ LogstashMessage t s tags mflds mmsg mts + (Just (String t), Just (String s), Just tags) -> Just $ LogstashMessage (Just t) (Just s) tags (Just mflds) mmsg mts _ -> Nothing value2logstash _ = Nothing 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.3.7.1 +version: 0.4.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. homepage: https://github.com/bartavelle/hslogstash @@ -23,7 +23,40 @@ 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, Logstash.Counter, Data.Conduit.FireHose default-extensions: OverloadedStrings, BangPatterns - build-depends: base >=4.6 && <4.8, conduit >=1.0.9 && <1.1, bytestring >=0.10 && <0.11, transformers >=0.3 && <0.4, aeson >=0.7 && <0.8, parallel-io >=0.3 && <0.4, text >=0.11 && <1.2, network >=2.4 && <2.5, network-conduit >=1.0 && <1.1, iconv >=0.4 && <0.5, http-conduit >= 2.0 && < 2.1, time >=1.4 && <1.5, text-format >=0.3 && <0.4, http-types >=0.8 && <0.9, unordered-containers >=0.2 && <0.3, vector >=0.10 && <0.11, containers >=0.5 && <0.6, stm-chans >=3.0 && <3.1, stm-conduit >=2.1 && <2.3, hedis >=0.6 && <0.7, stm >=2.4 && <2.5, attoparsec >=0.11 && <0.12, lens >= 4 && < 4.1, data-default, stm-firehose >= 0.1.4 && < 0.2, wai >= 2.0 && < 2.1, blaze-builder == 0.3.* + build-depends: + base >=4.6 && <5, + conduit >= 1.1.7 && < 1.3, + bytestring, + transformers >= 0.3 && < 0.5, + aeson >= 0.8 && < 0.11, + parallel-io >= 0.3.3 && < 0.4, + text, + network, + iconv, + http-conduit, + time >= 1.5 && < 1.6, + text-format, + http-types, + unordered-containers, + vector, + containers, + stm-chans, + stm-conduit, + hedis, + stm, + attoparsec >= 0.12 && < 0.14, + lens >= 4.12 && < 4.14, + data-default, + stm-firehose >= 0.3.0, + wai, + blaze-builder, + conduit-extra, + resourcet, + http-client, + conduit-extra >= 1.1.9.1 && < 1.2, + lens-aeson >= 1.0.0.5 && < 1.1 + + default-language: Haskell2010 ghc-options: -Wall