diff --git a/Changelog.md b/Changelog.md new file mode 100644 --- /dev/null +++ b/Changelog.md @@ -0,0 +1,3 @@ +## 1.0.0 + +* Initial release. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. 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/ekg-carbon.cabal b/ekg-carbon.cabal new file mode 100644 --- /dev/null +++ b/ekg-carbon.cabal @@ -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 diff --git a/src/System/Remote/Monitoring/Carbon.hs b/src/System/Remote/Monitoring/Carbon.hs new file mode 100644 --- /dev/null +++ b/src/System/Remote/Monitoring/Carbon.hs @@ -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) + ]