hslogstash 0.3.3 → 0.3.4
raw patch · 4 files changed
+75/−20 lines, 4 filesdep +lensdep +lens-aesondep ~conduit
Dependencies added: lens, lens-aeson
Dependency ranges changed: conduit
Files
- Changelog +7/−0
- Data/Conduit/ElasticSearch.hs +48/−6
- Logstash/Counter.hs +18/−12
- hslogstash.cabal +2/−2
Changelog view
@@ -1,3 +1,10 @@+hslogstash (0.3.4) precise; urgency=low++ * Fix collectd2nagios problems.+ * New scanning search for ElasticSearch.++ -- Bartavelle <bartavelle@gmail.com> Tue, 17 Dec 2013 12:12:24 +0100+ hslogstash (0.3.3) precise; urgency=low * Fixed the .cabal to follow the PVP.
Data/Conduit/ElasticSearch.hs view
@@ -2,9 +2,8 @@ {-| This module exports "Conduit" interfaces to ElasticSearch. It has been used intensively in production for several month now, but at a single site. -}-module Data.Conduit.ElasticSearch (esConduit, esSearchSource) where+module Data.Conduit.ElasticSearch (esConduit, esSearchSource, esScan) where -import Prelude hiding (catch) import Data.Maybe (fromMaybe) import Control.Exception import Data.Conduit@@ -12,7 +11,7 @@ import Network.HTTP.Conduit import Data.Aeson import Control.Applicative-import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as BS import qualified Data.ByteString.Lazy.Char8 as BSL import Data.Time import qualified Data.Text.Lazy.Encoding as E@@ -24,10 +23,13 @@ import Data.Either (partitionEithers) import Data.Monoid import Network.HTTP.Types+import Control.Lens hiding ((.=))+import Control.Lens.Aeson 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 data SearchObject = SearchObject { _source :: LogstashMessage , _index :: T.Text@@ -105,8 +107,47 @@ when (nexti < total) (self nexti) _ -> yield (Left errbody) +esScan :: Maybe (Request (ResourceT IO)) -- ^ 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)+ -> BS.ByteString -- ^ Name of the index (Something like logstash-2013.12.06)+ -> Int -- ^ Maximum size of each response+ -> Producer (ResourceT IO) (Either Value [Value])+esScan r h p indx maxsize = do+ resp <- liftIO (safeQuery defRScan)+ let respbody = decode (responseBody resp) :: Maybe Value+ code = statusCode $ responseStatus resp+ case (code, respbody ^? _Just . key "_scroll_id") of+ (200, Just (String scrid)) -> self scrid+ _ -> yield (Left (fromMaybe "could not parse json" respbody))+ where+ defR1 = fromMaybe def r+ defRScan = defR1 { host = h+ , port = p+ , path = indx <> "/_search"+ , queryString = "search_type=scan&scroll=5m&size=" <> BS.pack (show maxsize)+ , method = "GET"+ , checkStatus = \_ _ _ -> Nothing+ }+ self :: (MonadResource m) => T.Text -> Producer m (Either Value [Value])+ self scrollid = do+ let req = defRScan { path = "/_search/scroll"+ , queryString = "scroll=5m&scroll_id=" <> T.encodeUtf8 scrollid+ }+ resp <- liftIO (safeQuery req)+ let respbody = decode (responseBody resp) :: Maybe Value+ mscrollid = respbody ^? _Just . key "_scroll_id"+ mvals = respbody ^.. _Just . key "hits" . key "hits" . _Array . traverse . key "_source"+ -- liftIO (print respbody)+ case (mscrollid, mvals) of+ (Just _, []) -> return ()+ (Just (String nscr), hts) -> yield (Right hts) >> self nscr+ _ -> yield (Left (fromMaybe "could not parse json" respbody))+ safeQuery :: Request (ResourceT IO) -> IO (Response BSL.ByteString)-safeQuery req = catch (withManager $ httpLbs req) (\e -> print (e :: SomeException) >> threadDelay 500000 >> safeQuery req)+safeQuery req = catch (withManager $ httpLbs ureq ) (\e -> print (e :: SomeException) >> threadDelay 500000 >> safeQuery req)+ where+ ureq = req { responseTimeout = Nothing } -- | Takes a "LogstashMessage", and returns the result of the ElasticSearch request -- along with the value in case of errors, or ElasticSearch's values in case of@@ -131,8 +172,8 @@ Nothing -> Left (input, object [ "error" .= String "Time was not supplied" ]) Just (UTCTime day _) -> let (y,m,d) = toGregorian day- index = BSL.toStrict (E.encodeUtf8 (format "{}-{}.{}.{}" (prefix, y, left 2 '0' m, left 2 '0' d)))- in Right (input, object [ "index" .= object [ "_index" .= index, "_type" .= logstashType input ] ], toJSON input)+ indx = BSL.toStrict (E.encodeUtf8 (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 = let (errors, tosend) = partitionEithers input@@ -152,3 +193,4 @@ case items of Just (Array v) -> return $ map extractErrors (zip (map fst3 tosend) (V.toList v)) ++ lerrors _ -> genericError "Can't find items"+
Logstash/Counter.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE ScopedTypeVariables #-} {-| This module is not very well named, as it has almost nothing to do with Logstash. It is used to define counters that will then be logged by collectd. @@ -59,20 +60,25 @@ -> String -- ^ name of the plugin + instance -> String -- ^ name of the counter instance -> IO ()-counter2collectd c sockpath nodename plugin vinstance = void $ forkIO work+counter2collectd c sockpath nodename plugin vinstance = void $ forkIO $ forever 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+ soc <- socket AF_UNIX Stream 0+ eh <- try $ do+ connect soc (SockAddrUnix sockpath)+ socketToHandle soc ReadWriteMode+ case eh of+ Left (_ :: SomeException) -> sClose soc+ Right h -> do+ o <- try $ 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+ hClose h+ case o of+ Right () -> return ()+ Left (_ :: SomeException) -> hClose h
hslogstash.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: hslogstash-version: 0.3.3+version: 0.3.4 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 default-extensions: OverloadedStrings, BangPatterns- build-depends: base >=4.6 && <4.7, conduit >=1.0 && <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 && <0.12, network >=2.4 && <2.5, network-conduit >=1.0 && <1.1, iconv >=0.4 && <0.5, http-conduit >=1.9 && <1.10, 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+ 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 && <0.12, network >=2.4 && <2.5, network-conduit >=1.0 && <1.1, iconv >=0.4 && <0.5, http-conduit >=1.9 && <1.10, 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 default-language: Haskell2010 ghc-options: -Wall