ekg-elasticsearch (empty) → 0.3.0.0
raw patch · 6 files changed
+535/−0 lines, 6 filesdep +aesondep +basedep +bytestringsetup-changed
Dependencies added: aeson, base, bytestring, ekg-core, hostname, http-client, lens, text, time, unordered-containers, wreq
Files
- CHANGES.md +30/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- System/Metrics/Json.hs +245/−0
- System/Remote/Monitoring/ElasticSearch.hs +188/−0
- ekg-elasticsearch.cabal +40/−0
+ CHANGES.md view
@@ -0,0 +1,30 @@+## 0.2.1.0 (2016-08-11)++ * Send distributions as gauges and counters.+ * Update examples.++## 0.2.0.4 (2016-05-28)++ * GHC 8.0 support.++## 0.2.0.3 (2015-06-06)++ * Support GHC 7.10.++## 0.2.0.2 (2015-04-09)++ * Add support for network-2.6.++## 0.2.0.1 (2014-09-30)++ * Add support for text-1.2.++## 0.2.0.0 (2014-05-27)++ * Add configurable metric name prefix and suffix.+ * Add support for GHC 7.4.++## 0.1.0.0 (2014-05-01)++ * Initial release.+
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2014, Johan Tibell++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Johan Tibell nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ System/Metrics/Json.hs view
@@ -0,0 +1,245 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++-- | Encoding of ekg metrics as metricbeats compliant JSON. This module is+-- originally from egk-json with some tweaks for metricbeats+module System.Metrics.Json+ ( -- * Converting metrics to JSON values+ -- ** Types+ BeatEvent(..)+ , Beat(..)+ , BulkRequest(..)+ , CreateBulk(..)++ -- ** Conversion functions+ , sampleToJson+ , valueToJson++ -- ** Generated Lenses+ , beat, timestamp, beatTags, ekg, rtt++ -- ** Newtype wrappers with instances+ , Sample(..)+ , Value(..)+ ) where++import Control.Lens hiding ((.=))+import Data.Aeson ((.=))+import qualified Data.Aeson as A+import qualified Data.Aeson.Types as A+import qualified Data.ByteString.Lazy as LBS+import qualified Data.HashMap.Strict as M+import Data.Int (Int64)+import Data.Monoid ((<>))+import Data.Text (Text)+import qualified Data.Text as T+import Data.Time.Clock.POSIX (POSIXTime)+import GHC.Generics (Generic)++import Network.HTTP.Client (RequestBody (..), requestBody)+import Network.Wreq.Types+import qualified System.Metrics as Metrics+import qualified System.Metrics.Distribution as Distribution++--------------------------------------------------------------------------------+-- * Converting metrics to JSON values+--+-- egk-elastic transfers all the metrics in the 'Metrics.Store' as a single+-- 'BulkRequest'. Each individual metric is one document in the bulk request and+-- the operation is index. The index is controlled by the 'CreateBulk'. See+-- <https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html+-- ElasticSearch Bulk Request Documents> for more on the format.++--------------------------------------------------------------------------------+-- | The @beat@ key of the 'BeatEvent'+data Beat = Beat {+ hostname :: !Text -- ^ Hostname of metric source+ , name :: !Text -- ^ Name of the beat (hardcoded to "ekg" at the moment)+ , version :: !Text -- ^ Beat version (hardcoded to @0.1@ at the moment)+ } deriving (Generic)++instance A.ToJSON Beat++--------------------------------------------------------------------------------+-- | Newtype around the index to send events to+newtype CreateBulk = CreateBulk { _index :: Text }++instance A.ToJSON CreateBulk where+ toJSON (CreateBulk idx) = A.object+ [ "index" .= A.object [ "_index" .= idx+ , "_type" .= ("metricsets" :: Text)+ ]+ ]++--------------------------------------------------------------------------------+-- | Encode a 'Metrics.Sample' as a+-- <https://www.elastic.co/guide/en/beats/metricbeat/5.3/metricbeat-event-structure.html metricbeat event>+--+-- For reference this is the event structure+--+-- > {+-- > "@timestamp": "2016-06-22T22:05:53.291Z",+-- > "beat": {+-- > "hostname": "host.example.com",+-- > "name": "host.example.com"+-- > },+-- > "metricset": {+-- > "module": "system",+-- > "name": "process",+-- > "rtt": 7419+-- > },+-- > .+-- > .+-- > .+-- > "type": "metricsets"+-- > }+data BeatEvent = BeatEvent {+ _beat :: !Beat -- ^ The 'Beat'+ , _timestamp :: !POSIXTime -- ^ Timestamp of this event+ , _beatTags :: ![Text] -- ^ Extra tags to add to the event+ , _rtt :: !Int -- ^ Round trip time to generate the event+ , _ekg :: !Metrics.Sample -- ^ The snapshot of the 'Metrics.Store'+ }++makeLenses ''BeatEvent++instance A.ToJSON BeatEvent where+ toJSON b =+ A.object+ [ "beat" .= (b ^. beat)+ , "metricset" .= metricset+ , "@timestamp" .= (floor . (*1000) $ b ^. timestamp :: Integer)+ , "ekg" .= sampleToJson (b ^. ekg)+ , "type" .= ("metricsets" :: Text)+ ]+ where+ metricset = A.object+ [ "module" .= ("ekg" :: Text)+ , "name" .= ("ekg" :: Text)+ , "rtt" .= (b ^. rtt)+ ]+++--------------------------------------------------------------------------------+-- | Make a bulk submission to elasticsearch+newtype BulkRequest = BulkRequest [(CreateBulk, BeatEvent)]++instance Postable BulkRequest where+ postPayload (BulkRequest docs) req = return $ req { requestBody = RequestBodyLBS body}+ where+ body = (<> "\n") . LBS.intercalate "\n" . concatMap encodeBoth $ docs+ encodeBoth (cb, be) = [A.encode cb, A.encode be]++--------------------------------------------------------------------------------+-- | Generate the nested json for a 'Mertics.Sample'+--+-- Each "." in the metric name introduces a new level of nesting. For example,+-- the metrics @[("foo.bar", 10), ("foo.baz", "label")]@ are encoded as+--+-- > {+-- > "foo": {+-- > "bar": {+-- > "type:", "c",+-- > "val": 10+-- > },+-- > "baz": {+-- > "type": "l",+-- > "val": "label"+-- > }+-- > }+-- > }+sampleToJson :: Metrics.Sample -> A.Value+sampleToJson metrics =+ buildOne metrics A.emptyObject+ where+ buildOne :: M.HashMap T.Text Metrics.Value -> A.Value -> A.Value+ buildOne m o = M.foldlWithKey' build o m++ build :: A.Value -> T.Text -> Metrics.Value -> A.Value+ build m key = go m (T.splitOn "." key)++ go :: A.Value -> [T.Text] -> Metrics.Value -> A.Value+ go (A.Object m) [str] val = A.Object $ M.insert str metric m+ where metric = valueToJson val+ go (A.Object m) (str:rest) val = case M.lookup str m of+ Nothing -> A.Object $ M.insert str (go A.emptyObject rest val) m+ Just m' -> A.Object $ M.insert str (go m' rest val) m+ go v _ _ = typeMismatch "Object" v++typeMismatch :: String -- ^ The expected type+ -> A.Value -- ^ The actual value encountered+ -> a+typeMismatch expected actual =+ error $ "when expecting a " ++ expected ++ ", encountered " ++ typ +++ " instead"+ where+ typ = case actual of+ A.Object _ -> "Object"+ A.Array _ -> "Array"+ A.String _ -> "String"+ A.Number _ -> "Number"+ A.Bool _ -> "Boolean"+ A.Null -> "Null"++-- | Encodes a single 'Metrics.Value' as a JSON object. Examples:+--+-- > { "count": 89460 }+-- > { "gauge": 300 }+valueToJson :: Metrics.Value -> A.Value+valueToJson (Metrics.Counter n) = scalarToJson n CounterType+valueToJson (Metrics.Gauge n) = scalarToJson n GaugeType+valueToJson (Metrics.Label l) = scalarToJson l LabelType+valueToJson (Metrics.Distribution l) = distrubtionToJson l++-- | Convert a scalar metric (i.e. counter, gauge, or label) to a JSON+-- value.+scalarToJson :: A.ToJSON a => a -> MetricType -> A.Value+scalarToJson val ty = A.object+ [metricType ty .= val]+{-# SPECIALIZE scalarToJson :: Int64 -> MetricType -> A.Value #-}+{-# SPECIALIZE scalarToJson :: T.Text -> MetricType -> A.Value #-}++data MetricType =+ CounterType+ | GaugeType+ | LabelType+ | DistributionType++metricType :: MetricType -> T.Text+metricType CounterType = "count"+metricType GaugeType = "gauge"+metricType LabelType = "label"+metricType DistributionType = "dist"++-- | Convert a distribution to a JSON value.+distrubtionToJson :: Distribution.Stats -> A.Value+distrubtionToJson stats = A.object+ [ "mean" .= Distribution.mean stats+ , "variance" .= Distribution.variance stats+ , "count" .= Distribution.count stats+ , "sum" .= Distribution.sum stats+ , "min" .= Distribution.min stats+ , "max" .= Distribution.max stats+ ]++------------------------------------------------------------------------+-- ** Newtype wrappers with instances++-- | Newtype wrapper that provides a 'A.ToJSON' instances for the+-- underlying 'Metrics.Sample' without creating an orphan instance.+newtype Sample = Sample Metrics.Sample+ deriving Show++-- | Uses 'sampleToJson'.+instance A.ToJSON Sample where+ toJSON (Sample s) = sampleToJson s++-- | Newtype wrapper that provides a 'A.ToJSON' instances for the+-- underlying 'Metrics.Value' without creating an orphan instance.+newtype Value = Value Metrics.Value+ deriving Show++-- | Uses 'valueToJson'.+instance A.ToJSON Value where+ toJSON (Value v) = valueToJson v
+ System/Remote/Monitoring/ElasticSearch.hs view
@@ -0,0 +1,188 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TupleSections #-}+-- | This module lets you periodically flush metrics to elasticsearch. Example+-- usage:+--+-- > main = do+-- > store <- newStore+-- > forkElasticSearch defaultESOptions store+--+-- You probably want to include some of the predefined metrics defined+-- in the ekg-core package, by calling e.g. the 'registerGcStats'+-- function defined in that package.+module System.Remote.Monitoring.ElasticSearch+ (+ -- * The elasticsearch syncer+ ElasticSearch+ , elasticSearchThreadId+ , forkElasticSearch+ -- * ElasticSearch options+ , ESOptions(..)+ , defaultESOptions+ ) where++import Control.Concurrent (ThreadId, forkIO, threadDelay)+import Control.Lens+import Control.Monad (forever, void)+import qualified Data.HashMap.Strict as M+import Data.Int (Int64)+import Data.Monoid ((<>))+import Data.Text (Text)+import qualified Data.Text as T+import Data.Text.Lens+import Data.Time.Clock (getCurrentTime)+import Data.Time.Clock.POSIX (getPOSIXTime)+import Data.Time.Format (defaultTimeLocale, formatTime)+import Network.HostName (getHostName)+import Network.Wreq as Wreq+import qualified System.Metrics as Metrics++import System.Metrics.Json++--------------------------------------------------------------------------------+-- | A handle that can be used to control the elasticsearch sync thread.+-- Created by 'forkElasticSearch'.+newtype ElasticSearch = ElasticSearch { threadId :: ThreadId }++-- | The thread ID of the elasticsearch sync thread. You can stop the sync by+-- killing this thread (i.e. by throwing it an asynchronous+-- exception.)+elasticSearchThreadId :: ElasticSearch -> ThreadId+elasticSearchThreadId = threadId++--------------------------------------------------------------------------------+-- | Options to control how to connect to the elasticsearch server and how+-- often to flush metrics. The flush interval should be shorter than+-- the flush interval elasticsearch itself uses to flush data to its+-- backends.+data ESOptions = ESOptions+ { -- | Server hostname or IP address+ _host :: !Text++ -- | Server port+ , _port :: !Int++ -- | The elasticsearch index to insert into+ , _indexBase :: !Text++ -- | Append "-YYYY.MM.DD" onto index?+ , _indexByDate :: !Bool++ -- | What to put in the @beat.name@ field+ , _beatName :: !Text++ -- | Data push interval, in ms.+ , _flushInterval :: !Int++ -- | Print debug output to stderr.+ , _debug :: !Bool++ -- | Prefix to add to all metric names.+ , _prefix :: !Text++ -- | Suffix to add to all metric names. This is particularly+ -- useful for sending per host stats by settings this value to:+ -- @takeWhile (/= \'.\') \<$\> getHostName@, using @getHostName@+ -- from the @Network.BSD@ module in the network package.+ , _suffix :: !Text+ -- | Extra tags to add to events+ , _tags :: ![Text]+ }++makeClassy ''ESOptions++-- | Defaults:+--+-- * @host@ = @\"127.0.0.1\"@+--+-- * @port@ = @8125@+--+-- * @indexBase@ = @metricbeats@+--+-- * @indexByDate@ = @True@+--+-- * @beatName@ = @\"ekg\"@+--+-- * @flushInterval@ = @1000@+--+-- * @debug@ = @False@+defaultESOptions :: ESOptions+defaultESOptions = ESOptions+ { _host = "127.0.0.1"+ , _port = 9200+ , _indexBase = "metricbeat"+ , _indexByDate = True+ , _beatName = "ekg"+ , _flushInterval = 1000+ , _debug = False+ , _prefix = ""+ , _suffix = ""+ , _tags = []+ }++--------------------------------------------------------------------------------+-- | Create a thread that flushes the metrics in the store to elasticsearch.+forkElasticSearch :: ESOptions -- ^ Options+ -> Metrics.Store -- ^ Metric store+ -> IO ElasticSearch -- ^ ElasticSearch sync handle+forkElasticSearch opts store = ElasticSearch <$> forkIO (loop store opts)++loop :: Metrics.Store -- ^ Metric store+ -> ESOptions -- ^ Options+ -> IO ()+loop store opts = forever $ do+ start <- time+ flushSample store opts+ end <- time+ threadDelay ((opts ^. flushInterval) * 1000 - fromIntegral (end - start))+ loop store opts++-- | Microseconds since epoch.+time :: IO Int64+time = (round . (* 1000000.0) . toDouble) `fmap` getPOSIXTime+ where toDouble = realToFrac :: Real a => a -> Double+++--------------------------------------------------------------------------------+-- | Construct the correct URL to send metrics too from '_host' and '_port'+elasticURL :: ESOptions -> String+elasticURL eo = "http://" ++ (eo ^. host.unpacked) ++ ":" ++ show (eo ^. port) ++ "/_bulk"++-- | Construct the index to send to+--+-- if '_indexByDate' is @True@ this will be '_indexBase'-YYYY.MM.DD otherwise it+-- will just be '_indexBase'+mkIndex :: ESOptions -> IO CreateBulk+mkIndex eo =+ CreateBulk <$> if eo ^. indexByDate+ then appendDate (eo ^. indexBase)+ else return (eo ^. indexBase)+ where+ appendDate base = do+ day <- formatTime defaultTimeLocale "%Y.%m.%d" <$> getCurrentTime+ return $ base <> "-" <> (day ^. packed)++--------------------------------------------------------------------------------+-- | Generate a 'BeatEvent' for each metric in the 'Metrics.Store'+sampleBeatEvents :: Metrics.Store -> ESOptions -> IO [BeatEvent]+sampleBeatEvents store eo = do+ now <- getPOSIXTime+ sample <- Metrics.sampleAll store+ hostName <- T.pack <$> getHostName+ finish <- getPOSIXTime+ let took = floor $ (finish - now) * 1000+ theBeat = Beat hostName (eo ^. beatName) "0.1"+ mkBeatEvt evts k v = BeatEvent theBeat now(eo ^. tags) took (M.singleton k v) : evts+ return $ M.foldlWithKey' mkBeatEvt [] sample++--------------------------------------------------------------------------------+-- | Create a 'BulkRequest' and send it elasticsearch+flushSample :: Metrics.Store -> ESOptions -> IO ()+flushSample store eo = do+ createBulk <- mkIndex eo+ bulkEvts <- sampleBeatEvents store eo+ void $ Wreq.post (elasticURL eo) $ BulkRequest $ (createBulk,) <$> bulkEvts
+ ekg-elasticsearch.cabal view
@@ -0,0 +1,40 @@+name: ekg-elasticsearch+version: 0.3.0.0+synopsis: Push metrics to elasticsearch+description:+ This library lets you push system metrics to a elasticsearch server.+homepage: https://github.com/cdodev/ekg-elasticsearch+bug-reports: https://github.com/cdodev/ekg-elasticsearch/issues+license: BSD3+license-file: LICENSE+author: Ben Ford+maintainer: ben@perurbis.com+category: System+build-type: Simple+extra-source-files: CHANGES.md+cabal-version: >=1.10++library+ exposed-modules: + System.Remote.Monitoring.ElasticSearch+ System.Metrics.Json++ build-depends: base >= 4.5 && < 4.10+ , bytestring < 1.0+ , ekg-core >= 0.1 && < 1.0+ , aeson >= 1.0+ , hostname >= 1.0+ , http-client >= 0.5+ , lens >= 4.15.1+ , text < 1.3+ , time < 1.7+ , unordered-containers < 0.3+ , wreq++ default-language: Haskell2010++ ghc-options: -Wall++source-repository head+ type: git+ location: https://github.com/cdodev/ekg-elastic.git