diff --git a/CHANGES.md b/CHANGES.md new file mode 100644 --- /dev/null +++ b/CHANGES.md @@ -0,0 +1,4 @@ +## 0.1.0.0 (2013-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/Remote/Monitoring/Statsd.hs b/System/Remote/Monitoring/Statsd.hs new file mode 100644 --- /dev/null +++ b/System/Remote/Monitoring/Statsd.hs @@ -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 () diff --git a/ekg-statsd.cabal b/ekg-statsd.cabal new file mode 100644 --- /dev/null +++ b/ekg-statsd.cabal @@ -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