diff --git a/CHANGES.md b/CHANGES.md new file mode 100644 --- /dev/null +++ b/CHANGES.md @@ -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. + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/System/Metrics/Json.hs b/System/Metrics/Json.hs new file mode 100644 --- /dev/null +++ b/System/Metrics/Json.hs @@ -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 +-- 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 +-- +-- +-- 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 diff --git a/System/Remote/Monitoring/ElasticSearch.hs b/System/Remote/Monitoring/ElasticSearch.hs new file mode 100644 --- /dev/null +++ b/System/Remote/Monitoring/ElasticSearch.hs @@ -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 diff --git a/ekg-elasticsearch.cabal b/ekg-elasticsearch.cabal new file mode 100644 --- /dev/null +++ b/ekg-elasticsearch.cabal @@ -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