ekg-carbon (empty) → 1.0.0

raw patch · 5 files changed

+209/−0 lines, 5 filesdep +basedep +ekg-coredep +networksetup-changed

Dependencies added: base, ekg-core, network, network-carbon, text, time, unordered-containers, vector

Files

Changelog.md +3/−0
LICENSE +30/−0
Setup.hs +2/−0
ekg-carbon.cabal +30/−0
src/System/Remote/Monitoring/Carbon.hs +144/−0

+ Changelog.md view

@@ -0,0 +1,3 @@+## 1.0.0++* Initial release.

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright (c) 2014, Oliver Charles++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 Oliver Charles 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

+ ekg-carbon.cabal view

@@ -0,0 +1,30 @@+name: ekg-carbon+version: 1.0.0+synopsis: An EKG backend to send statistics to Carbon (part of Graphite monitoring tools)+homepage: http://github.com/ocharles/ekg-carbon+license: BSD3+license-file: LICENSE+author: Oliver Charles+maintainer: ollie@ocharles.org.uk+build-type: Simple+cabal-version: >=1.10+extra-source-files: Changelog.md++library+  exposed-modules:+    System.Remote.Monitoring.Carbon++  build-depends:+    base >=4.6 && <4.8,+    ekg-core >= 0.1.0.1 && < 0.2,+    network >= 2.5.0.0 && < 2.6,+    network-carbon >= 1.0.0 && <1.1,+    text >= 1.1.1.3 && < 1.2,+    time >= 1.4.2 && < 1.5,+    unordered-containers,+    vector >= 0.10.11.0 && < 0.11++  hs-source-dirs: src+  default-language: Haskell2010++  ghc-options: -Wall -O2 -fwarn-incomplete-uni-patterns

+ src/System/Remote/Monitoring/Carbon.hs view

@@ -0,0 +1,144 @@+{-# LANGUAGE OverloadedStrings #-}++-- | This module lets you periodically flush metrics to a Graphite Carbon+-- backend. Example usage:+--+-- > main = do+-- >   store <- newStore+-- >   forkCarbon defaultCarbonOptions 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 package.+module System.Remote.Monitoring.Carbon+  ( CarbonOptions(..)+  , defaultCarbonOptions+  , forkCarbon+  ) where++import Control.Concurrent (ThreadId, forkFinally, myThreadId, threadDelay, throwTo)+import Control.Monad (forever)+import Data.Int (Int64)+import Data.Monoid ((<>))++import qualified Data.HashMap.Strict as HashMap+import qualified Data.Text as T+import qualified Data.Time as Time+import qualified Data.Time.Clock.POSIX as Time+import qualified Data.Vector as V+import qualified Network.Carbon.Plaintext as Carbon+import qualified Network.Socket as Network+import qualified System.Metrics as EKG+import qualified System.Metrics.Distribution as Stats++--------------------------------------------------------------------------------+-- | Options to control how to connect to the Carbon 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 CarbonOptions = CarbonOptions+  { -- | The hostname or IP address of the server running Carbon.+    host :: !T.Text++    -- | Server port of the TCP line receiver interface.+  , port :: !Int++    -- | The amount of time between sampling EKG metrics and pushing to Carbon.+  , flushInterval :: !Int++    -- | Prefix to add to all metric 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@ = @2003@+--+-- * @flushInterval@ = @1000@+--+-- * Empty 'prefix' and 'suffix'.+defaultCarbonOptions :: CarbonOptions+defaultCarbonOptions = CarbonOptions+    { host          = "127.0.0.1"+    , port          = 2003+    , flushInterval = 1000+    , prefix        = ""+    , suffix        = ""+    }+++--------------------------------------------------------------------------------+-- | Create a thread that periodically flushes the metrics in 'EKG.Store' to+-- Carbon.+forkCarbon :: CarbonOptions -> EKG.Store -> IO (ThreadId)+forkCarbon opts store = do+  addrInfos <- Network.getAddrInfo Nothing+                                   (Just $ T.unpack $ host opts)+                                   (Just $ show $ port opts)+  c <- case addrInfos of+    (addrInfo : _) -> Carbon.connect (Network.addrAddress addrInfo)+    _ -> unsupportedAddressError++  parent <- myThreadId+  forkFinally (loop store c opts)+              (\r -> do Carbon.disconnect c+                        case r of+                          Left e  -> throwTo parent e+                          Right _ -> return ())++  where+  unsupportedAddressError = ioError $ userError $+      "unsupported address: " ++ T.unpack (host opts)+++--------------------------------------------------------------------------------+loop :: EKG.Store -> Carbon.Connection -> CarbonOptions -> IO ()+loop store socket opts = forever $ do+  start <- time+  sample <- EKG.sampleAll store+  flushSample sample socket 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 -> Carbon.Connection -> CarbonOptions -> IO ()+flushSample sample conn opts = do+  t <- Time.getCurrentTime+  Carbon.sendMetrics conn+                     (V.map renamed+                            (HashMap.foldlWithKey' (\ms k v -> metrics k v t <> ms)+                                                   V.empty+                                                   sample))++  where+  renamed (Carbon.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 Carbon.Metric (p <> n <> s) v t++  metrics n (EKG.Counter i) t = V.singleton (Carbon.Metric n (fromIntegral i) t)+  metrics n (EKG.Gauge i) t = V.singleton (Carbon.Metric n (fromIntegral i) t)+  metrics _ (EKG.Label {}) _ = V.empty+  metrics n (EKG.Distribution stats) t =+    let f n' v = Carbon.Metric (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)+                  ]