hslogstash 0.3.6 → 0.3.7
raw patch · 4 files changed
+31/−16 lines, 4 filesdep −lens-aesondep ~aesondep ~basedep ~lens
Dependencies removed: lens-aeson
Dependency ranges changed: aeson, base, lens, stm-firehose, text
Files
- Data/Conduit/ElasticSearch.hs +11/−7
- Data/Conduit/FireHose.hs +17/−6
- Data/Conduit/Logstash.hs +1/−1
- hslogstash.cabal +2/−2
Data/Conduit/ElasticSearch.hs view
@@ -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
Data/Conduit/FireHose.hs view
@@ -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)
Data/Conduit/Logstash.hs view
@@ -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
hslogstash.cabal view
@@ -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