ekg-influxdb (empty) → 0.1.0.0

raw patch · 5 files changed

+249/−0 lines, 5 filesdep +basedep +clockdep +containerssetup-changed

Dependencies added: base, clock, containers, ekg-core, libinfluxdb, text, time, unordered-containers, vector

Files

• Changelog.md +5/−0
• LICENSE +30/−0
• Setup.hs +2/−0
• ekg-influxdb.cabal +29/−0
• src/System/Remote/Monitoring/Influxdb.hs +183/−0

@@ -0,0 +1,5 @@+# Revision history for ekg-influxdb++## 0.1.0.0  -- YYYY-mm-dd++* First version. Released on an unsuspecting world.

@@ -0,0 +1,30 @@+Copyright (c) 2016, Moritz Angermann++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 Moritz Angermann 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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,29 @@+-- Initial ekg-influxdb.cabal generated by cabal init.  For further +-- documentation, see http://haskell.org/cabal/users-guide/++name:                ekg-influxdb+version:             0.1.0.0+synopsis:            An EKG backend to send statistics to influxdb+homepage:            https://github.com/angerman/ekg-influxdb+license:             BSD3+license-file:        LICENSE+author:              Moritz Angermann+maintainer:          moritz.angermann@gmail.com+build-type:          Simple+extra-source-files:  Changelog.md+cabal-version:       >=1.10++library+  exposed-modules:  System.Remote.Monitoring.Influxdb+  build-depends:    base >=4.9 && <4.10+                  , ekg-core >= 0.1+                  , libinfluxdb >= 0.0.4+                  , containers >= 0.5.7+                  , unordered-containers >= 0.2.7+                  , time >= 1.6+                  , clock >= 0.7+                  , text >= 1.2.2+                  , vector >= 0.11+  hs-source-dirs:   src+  default-language: Haskell2010+  ghc-options:      -Wall -O2

@@ -0,0 +1,183 @@+{-# LANGUAGE OverloadedStrings #-}++-- | This module lets you periodically flush metrics to a influxdb+-- backend. Example usage:+--+-- > main = do+-- >   store <- newStore+-- >   forkInfluxdb defaultInfluxdbOptions store+--+-- You probably want to include some of the predefined metrics defined+-- in the @ekg-core@ package, by calling e.g. the 'EKG.registerGcMetrics'+-- function defined in that pacakge.+--+-- NOTE: This package has been modelled after the @ekg-carbon@ package, and+--       is almost identical. The author is indebted to those fine folks who+--       wrote the @ekg-carbon@ package.++module System.Remote.Monitoring.Influxdb+  ( InfluxdbOptions(..)+  , defaultInfluxdbOptions+  , forkInfluxdb+  , forkInfluxdbRestart+  ) where++import Data.Monoid ((<>))+import qualified Data.Time.Clock.POSIX as Time+import qualified Data.Text as T+import qualified System.Metrics as EKG+import qualified System.Metrics.Distribution as Stats+import           System.Clock+import qualified Data.Map as M+import qualified Data.Vector as V+import qualified Database.InfluxDB.Writer as Influxdb+import Control.Exception (SomeException, try, bracket)+import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay, throwTo)+import Control.Monad (forever)+import qualified Data.HashMap.Strict as HashMap+import Data.Int (Int64)++--------------------------------------------------------------------------------+-- | Options to control how to connect to the Influxdb server and how often to+-- flush metrics. The flush interval should match the shortest retention rate of+-- the matching retention periods, or you risk over-riding previous samples.+data InfluxdbOptions = InfluxdbOptions+  { -- | The hostname or IP address of the server running influxdb+    host :: !T.Text+    -- | Server port of the TCP line receiver interface.+  , port :: !Int+    -- | The Influxdb database name+  , database :: !T.Text+    -- | The amount of time between sampling EKG metrics and pushing to Influxdb.+  , flushInterval :: !Int+    -- | Prefix to add to all matric names.+  , prefix :: !T.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 :: !T.Text+  } deriving (Eq, Show)++--------------------------------------------------------------------------------+-- | Defaults+--+-- * @host@ = @\"127.0.0.1\"@+--+-- * @port@ = @8086@+--+-- * @flushInterval@ = @1000@+--+-- * Empty 'database', 'prefix' and 'suffix'.+defaultInfluxdbOptions :: InfluxdbOptions+defaultInfluxdbOptions = InfluxdbOptions+    { host          = "127.0.0.1"+    , port          = 8086+    , database      = ""+    , flushInterval = 1000+    , prefix        = ""+    , suffix        = ""+    }++--------------------------------------------------------------------------------+-- | Create a thread that periodically flushes the metrics in 'EKG.Store' to+-- Influxdb. If the thread flushing statistics throws an exception (for example,+-- the network connection is lost), this exception will be thrown up to the thread+-- that called 'forkInfluxdb'. For more control, see 'forkInfluxdbRestart'.+forkInfluxdb :: InfluxdbOptions -> EKG.Store -> IO ThreadId+forkInfluxdb opts store =+  do parent <- myThreadId+     forkInfluxdbRestart opts+                         store+                         (\e _ -> throwTo parent e)++--------------------------------------------------------------------------------+-- | Create a thread that periodically flushes the metrics in 'EKG.Store' to+-- Influxdb. If the thread flushing statistics throws an exception (for example,+-- the network connection is lost), the callback function will be invoked with the+-- exception that was thrown, and an 'IO' computation to restart the handler.+--+-- For example, you can use 'forkInfluxdbRestart' to log failures and restart+-- logging:+--+-- > forkInfluxdbRestart defaultInfluxdbOptions+-- >                     store+-- >                     (\ex restart -> do hPutStrLn stderr ("ekg-influxdb: " ++ show ex)+-- >                                        restart)+forkInfluxdbRestart :: InfluxdbOptions+                    -> EKG.Store+                    -> (SomeException -> IO () -> IO ())+                    -> IO ThreadId+forkInfluxdbRestart opts store exceptionHandler = do+  eHandle <- Influxdb.createHandle (Influxdb.Config $ T.unpack ("http://" <> host opts <> ":" <> T.pack (show (port opts)) <> "/" <> database opts))+  handle <- case eHandle of+    Right h -> return $ h+    _ -> unsupportedAddressError+  let go = do+        terminated <-+          try $ bracket+          (return handle)+          (\_ -> return ())+          (loop store opts)+        case terminated of+          Left exception -> exceptionHandler exception go+          Right _ -> go+  forkIO go+  where unsupportedAddressError =+          ioError (userError ("unsupported address: " ++ T.unpack (host opts)))+++--------------------------------------------------------------------------------+loop :: EKG.Store -> InfluxdbOptions -> Influxdb.Handle -> IO ()+loop store opts handle = forever $ do+  start <- time+  sample <- EKG.sampleAll store+  flushSample sample handle opts+  end <- time+  threadDelay (flushInterval opts * 1000 - fromIntegral (end - start))++-- | Microseconds since epoch.+time :: IO Int64+time = (round . (* 1000000.0) . toDouble) `fmap` Time.getPOSIXTime+  where toDouble = realToFrac :: Real a => a -> Double++flushSample :: EKG.Sample -> Influxdb.Handle -> InfluxdbOptions -> IO ()+flushSample sample handle opts = do+  t <- getTime Realtime+  sendMetrics handle+    (V.map renamed+      (HashMap.foldlWithKey' (\ms k v -> metrics k v t <> ms)+        V.empty+        sample))++  where+  renamed (Metric n v t) =+    let p = if T.null (prefix opts) then "" else prefix opts <> "."+        s = if T.null (suffix opts) then "" else "." <> suffix opts+    in Metric (p <> n <> s) v t++  metrics n (EKG.Counter i) t = V.singleton (Metric ("counter." <> n) (fromIntegral i) t)+  metrics n (EKG.Gauge i) t = V.singleton (Metric ("gauge." <> n) (fromIntegral i) t)+  metrics _ (EKG.Label {}) _ = V.empty+  metrics n (EKG.Distribution stats) t =+    let f n' v = Metric ("dist." <> n <> "." <> n') v t+    in V.fromList [ f "mean" (Stats.mean stats)+                  , f "variance" (Stats.variance stats)+                  , f "count" (fromIntegral $ Stats.count stats)+                  , f "sum" (Stats.sum stats)+                  , f "min" (Stats.min stats)+                  , f "max" (Stats.max stats)+                  ]++data Metric = Metric+  { path  :: !T.Text+  , value :: !Double+  , timestamp :: !TimeSpec+  } deriving (Show)++sendMetrics :: Influxdb.Handle -> V.Vector Metric -> IO ()+sendMetrics handle metrics = V.forM_ metrics $ \metric -> do+  let tags   = M.empty+      values = M.singleton "value" (Influxdb.F (value metric))+      ts     = fromIntegral . toNanoSecs $ (timestamp metric)+    in Influxdb.writePoint' handle (path metric) tags values ts