hslogstash 0.3.7.1 → 0.4.2
raw patch · 9 files changed
+102/−48 lines, 9 filesdep +conduit-extradep +http-clientdep +lens-aesondep −network-conduitdep ~aesondep ~attoparsecdep ~base
Dependencies added: conduit-extra, http-client, lens-aeson, resourcet
Dependencies removed: network-conduit
Dependency ranges changed: aeson, attoparsec, base, blaze-builder, bytestring, conduit, containers, hedis, http-conduit, http-types, iconv, lens, network, parallel-io, stm, stm-chans, stm-conduit, stm-firehose, text, text-format, time, transformers, unordered-containers, vector, wai
Files
- Data/Conduit/Branching.hs +1/−0
- Data/Conduit/ElasticSearch.hs +16/−18
- Data/Conduit/FireHose.hs +2/−1
- Data/Conduit/Logstash.hs +2/−1
- Data/Conduit/Network/Retry.hs +18/−6
- Data/Conduit/Redis.hs +5/−3
- Logstash/Counter.hs +2/−2
- Logstash/Message.hs +21/−15
- hslogstash.cabal +35/−2
Data/Conduit/Branching.hs view
@@ -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])
Data/Conduit/ElasticSearch.hs view
@@ -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
Data/Conduit/FireHose.hs view
@@ -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"
Data/Conduit/Logstash.hs view
@@ -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))
Data/Conduit/Network/Retry.hs view
@@ -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+ )
Data/Conduit/Redis.hs view
@@ -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
Logstash/Counter.hs view
@@ -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
Logstash/Message.hs view
@@ -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
hslogstash.cabal view
@@ -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