ekg-statsd (empty) → 0.1.0.0

raw patch · 5 files changed

+238/−0 lines, 5 filesdep +basedep +bytestringdep +ekg-coresetup-changed

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

Files

• CHANGES.md +4/−0
• LICENSE +30/−0
• Setup.hs +2/−0
• System/Remote/Monitoring/Statsd.hs +166/−0
• ekg-statsd.cabal +36/−0

@@ -0,0 +1,4 @@+## 0.1.0.0 (2013-05-01)++ * Initial release.+

@@ -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.

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

@@ -0,0 +1,166 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- | This module lets you periodically flush metrics to a statsd+-- backend. Example usage:+--+-- > main = do+-- >     store <- newStore+-- >     forkStatsd defaultStatsdOptions 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.Statsd+    (+      -- * The statsd syncer+      Statsd+    , statsdThreadId+    , forkStatsd+    , StatsdOptions(..)+    , defaultStatsdOptions+    ) where++import Control.Concurrent (ThreadId, forkFinally, myThreadId, threadDelay,+                           throwTo)+import Control.Exception (IOException, catch)+import Control.Monad (forM_, when)+import qualified Data.ByteString.Char8 as B8+import qualified Data.HashMap.Strict as M+import Data.Int (Int64)+import Data.Monoid ((<>))+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.Text.IO as T+import Data.Time.Clock.POSIX (getPOSIXTime)+import qualified Network.Socket as Socket+import qualified Network.Socket.ByteString as Socket+import qualified System.Metrics as Metrics+import System.IO (stderr)++-- | A handle that can be used to control the statsd sync thread.+-- Created by 'forkStatsd'.+data Statsd = Statsd+    { threadId :: {-# UNPACK #-} !ThreadId+    }++-- | The thread ID of the statsd sync thread. You can stop the sync by+-- killing this thread (i.e. by throwing it an asynchronous+-- exception.)+statsdThreadId :: Statsd -> ThreadId+statsdThreadId = threadId++-- | Options to control how to connect to the statsd server and how+-- often to flush metrics. The flush interval should be shorter than+-- the flush interval statsd itself uses to flush data to its+-- backends.+data StatsdOptions = StatsdOptions+    { host          :: !T.Text  -- ^ Server hostname or IP address+    , port          :: !Int     -- ^ Server port+    , flushInterval :: !Int     -- ^ Data push interval, in ms.+    , debug         :: !Bool    -- ^ Print debug output to stderr.+    }++-- | Defaults:+--+-- * @host@ = @\"127.0.0.1\"@+--+-- * @port@ = @8125@+--+-- * @flushInterval@ = @1000@+--+-- * @debug@ = @False@+defaultStatsdOptions :: StatsdOptions+defaultStatsdOptions = StatsdOptions+    { host          = "127.0.0.1"+    , port          = 8125+    , flushInterval = 1000+    , debug         = False+    }++-- | Create a thread that periodically flushes the metrics in the+-- store to statsd.+forkStatsd :: StatsdOptions  -- ^ Options+           -> Metrics.Store  -- ^ Metric store+           -> IO Statsd      -- ^ Statsd sync handle+forkStatsd opts store = do+    addrInfos <- Socket.getAddrInfo Nothing (Just $ T.unpack $ host opts)+                 (Just $ show $ port opts)+    socket <- case addrInfos of+        [] -> unsupportedAddressError+        (addrInfo:_) -> do+            socket <- Socket.socket (Socket.addrFamily addrInfo)+                      Socket.Datagram Socket.defaultProtocol+            Socket.connect socket (Socket.addrAddress addrInfo)+            return socket+    me <- myThreadId+    tid <- forkFinally (loop store emptySample socket opts) $ \ r -> do+        Socket.close socket+        case r of+            Left e  -> throwTo me e+            Right _ -> return ()+    return $ Statsd tid+  where+    unsupportedAddressError = ioError $ userError $+        "unsupported address: " ++ T.unpack (host opts)+    emptySample = M.empty++loop :: Metrics.Store   -- ^ Metric store+     -> Metrics.Sample  -- ^ Last sampled metrics+     -> Socket.Socket   -- ^ Connected socket+     -> StatsdOptions   -- ^ Options+     -> IO ()+loop store lastSample socket opts = do+    start <- time+    sample <- Metrics.sampleAll store+    let !diff = diffSamples lastSample sample+    flushSample diff socket opts+    end <- time+    threadDelay (flushInterval opts * 1000 - fromIntegral (end - start))+    loop store sample socket opts++-- | Microseconds since epoch.+time :: IO Int64+time = (round . (* 1000000.0) . toDouble) `fmap` getPOSIXTime+  where toDouble = realToFrac :: Real a => a -> Double++diffSamples :: Metrics.Sample -> Metrics.Sample -> Metrics.Sample+diffSamples prev curr = M.foldlWithKey' combine M.empty curr+  where+    combine m name new = case M.lookup name prev of+        Just old -> case diffMetric old new of+            Just val -> M.insert name val m+            Nothing  -> m+        _        -> M.insert name new m++    diffMetric :: Metrics.Value -> Metrics.Value -> Maybe Metrics.Value+    diffMetric (Metrics.Counter n1) (Metrics.Counter n2)+        | n1 == n2  = Nothing+        | otherwise = Just $! Metrics.Counter $ n2 - n1+    diffMetric (Metrics.Gauge n1) (Metrics.Gauge n2)+        | n1 == n2  = Nothing+        | otherwise = Just $ Metrics.Gauge n2+    diffMetric (Metrics.Label n1) (Metrics.Label n2)+        | n1 == n2  = Nothing+        | otherwise = Just $ Metrics.Label n2+    -- Distributions are assumed to be non-equal.+    diffMetric _ _  = Nothing++flushSample :: Metrics.Sample -> Socket.Socket -> StatsdOptions -> IO ()+flushSample sample socket opts = do+    forM_ (M.toList $ sample) $ \ (name, val) ->+        flushMetric name val+  where+    flushMetric name (Metrics.Counter n) = send "|c" name (show n)+    flushMetric name (Metrics.Gauge n)   = send "|g" name (show n)+    flushMetric _ _                      = return ()++    isDebug = debug opts+    send ty name val = do+        let !msg = B8.concat [T.encodeUtf8 name, ":", B8.pack val, ty]+        when isDebug $ B8.hPutStrLn stderr $ B8.concat [ "DEBUG: ", msg]+        Socket.sendAll socket msg `catch` \ (e :: IOException) -> do+            T.hPutStrLn stderr $ "ERROR: Couldn't send message: " <>+                T.pack (show e)+            return ()

@@ -0,0 +1,36 @@+name:                ekg-statsd+version:             0.1.0.0+synopsis:            Push metrics to statsd+description:+  This library lets you push system metrics to a statsd server.+homepage:            https://github.com/tibbe/ekg-statsd+bug-reports:         https://github.com/tibbe/ekg-statsd/issues+license:             BSD3+license-file:        LICENSE+author:              Johan Tibell+maintainer:          johan.tibell@gmail.com+category:            System+build-type:          Simple+extra-source-files:  CHANGES.md+cabal-version:       >=1.10++library+  exposed-modules:     +    System.Remote.Monitoring.Statsd++  build-depends:+    base >= 4.5 && < 4.8,+    bytestring < 1.0,+    ekg-core >= 0.1 && < 1.0,+    network < 2.6,+    text < 1.2,+    time < 1.5,+    unordered-containers < 0.3++  default-language:    Haskell2010++  ghc-options: -Wall++source-repository head+  type:     git+  location: https://github.com/tibbe/ekg-statsd.git