diff --git a/Changelog b/Changelog --- a/Changelog +++ b/Changelog @@ -1,3 +1,10 @@ +hslogstash (0.3.4) precise; urgency=low + + * Fix collectd2nagios problems. + * New scanning search for ElasticSearch. + + -- Bartavelle Tue, 17 Dec 2013 12:12:24 +0100 + hslogstash (0.3.3) precise; urgency=low * Fixed the .cabal to follow the PVP. diff --git a/Data/Conduit/ElasticSearch.hs b/Data/Conduit/ElasticSearch.hs --- a/Data/Conduit/ElasticSearch.hs +++ b/Data/Conduit/ElasticSearch.hs @@ -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" + diff --git a/Logstash/Counter.hs b/Logstash/Counter.hs --- a/Logstash/Counter.hs +++ b/Logstash/Counter.hs @@ -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 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.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