diff --git a/Data/Conduit/ElasticSearch.hs b/Data/Conduit/ElasticSearch.hs --- a/Data/Conduit/ElasticSearch.hs +++ b/Data/Conduit/ElasticSearch.hs @@ -24,7 +24,7 @@ import Data.Monoid import Network.HTTP.Types import Control.Lens hiding ((.=)) -import Control.Lens.Aeson +import Data.Aeson.Lens import Data.Default import qualified Data.HashMap.Strict as HM @@ -34,9 +34,9 @@ data SearchObject = SearchObject { _source :: LogstashMessage , _index :: T.Text - , _score :: Double - , _id :: BS.ByteString - , _type :: T.Text + , _score :: Double + , _id :: T.Text + , _type :: T.Text } deriving (Show) instance FromJSON SearchObject where @@ -108,6 +108,7 @@ when (nexti < total) (self nexti) _ -> yield (Left errbody) +-- | Use this function for \'scanning\' requests, using the scroll feature. esScan :: Maybe Request -- ^ 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) @@ -173,7 +174,7 @@ Nothing -> Left (input, object [ "error" .= String "Time was not supplied" ]) Just (UTCTime day _) -> let (y,m,d) = toGregorian day - indx = BSL.toStrict (E.encodeUtf8 (format "{}-{}.{}.{}" (prefix, y, left 2 '0' m, left 2 '0' d))) + indx = format "{}-{}.{}.{}" (prefix, y, left 2 '0' m, left 2 '0' d) in Right (input, object [ "index" .= object [ "_index" .= indx, "_type" .= logstashType input ] ], toJSON input) sendBulk :: (MonadResource m) => [Either (LogstashMessage, Value) (LogstashMessage, Value, Value)] -> m [Either (LogstashMessage, Value) Value] sendBulk input = @@ -182,12 +183,15 @@ body = BSL.unlines $ concatMap (\(_,x,y) -> [encode x, encode y]) tosend req = defR2 { requestBody = RequestBodyLBS body } in do - res <- fmap responseBody $ liftIO $ safeQuery req + 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 = decode res >>= getObject "items" + items = getObject "items" (String res) extractErrors x = if (getObject "create" (snd x) >>= getObject "ok") == Just (Bool True) then Right (snd x) else Left x diff --git a/Data/Conduit/FireHose.hs b/Data/Conduit/FireHose.hs --- a/Data/Conduit/FireHose.hs +++ b/Data/Conduit/FireHose.hs @@ -1,5 +1,5 @@ --- | A firehose sink, letting client get through a port and read the sink --- output. +-- | A firehose conduit, spawning a web server that will allow for the +-- observation of the messages. module Data.Conduit.FireHose (fireHose) where import Control.Monad.IO.Class @@ -8,7 +8,7 @@ import Data.Conduit import Data.Conduit.Network.Firehose -import Network.Wai (Request,pathInfo) +import Network.Wai (pathInfo) import qualified Data.HashSet as HS import qualified Data.Text as T @@ -17,9 +17,20 @@ import Blaze.ByteString.Builder.ByteString import Data.Monoid --- | A web server will be launched on the specified port. The request URL --- must be of the form /type1,type2,type3. The client will be fed all --- messages matching those types. +{-| 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 +'LogstashMessage' matching one of the given types. + +Here is a sample usage : + +> -- run the fire hose on port 13400 +> fh <- fireHose 13400 10 +> logstashListener lport (printErrors =$ CL.mapM (liftIO . addLogstashTime) -- add the time +> =$ fh +> =$ CL.map (BSL.toStrict . encode) -- turn into a bytestring +> =$ redisSink host port queue (Just logfunc)) -- store to redis + +-} fireHose :: MonadIO m => Int -- ^ Port -> Int -- ^ Buffer size for the fire hose threads -> IO (Conduit LogstashMessage m LogstashMessage) diff --git a/Data/Conduit/Logstash.hs b/Data/Conduit/Logstash.hs --- a/Data/Conduit/Logstash.hs +++ b/Data/Conduit/Logstash.hs @@ -1,6 +1,6 @@ -- | Receive logstash messages from the network, and process them with -- a conduit. -module Data.Conduit.Logstash (logstashListener) where +module Data.Conduit.Logstash (logstashListener,tryDecode) where import Data.Conduit import qualified Data.Conduit.Binary as CB 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.6 +version: 0.3.7 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,7 @@ 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.7, conduit >=1.0.9 && <1.1, bytestring >=0.10 && <0.11, transformers >=0.3 && <0.4, aeson >=0.6 && <0.7, parallel-io >=0.3 && <0.4, text >=0.11 && <1.1, 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.2, hedis >=0.6 && <0.7, stm >=2.4 && <2.5, attoparsec >=0.10 && <0.11, lens >= 3.9 && < 4, lens-aeson >= 0.1 && < 0.2, data-default, stm-firehose >= 0.1.1 && < 0.2, wai >= 2.0 && < 2.1, blaze-builder == 0.3.* + 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.2, hedis >=0.6 && <0.7, stm >=2.4 && <2.5, attoparsec >=0.10 && <0.11, lens >= 4 && < 4.1, data-default, stm-firehose >= 0.1.4 && < 0.2, wai >= 2.0 && < 2.1, blaze-builder == 0.3.* default-language: Haskell2010 ghc-options: -Wall