ekg-push (empty) → 0.0.2

raw patch · 9 files changed

+390/−0 lines, 9 filesdep +basedep +bytestringdep +ekg-coresetup-changed

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

Files

CHANGES.md +7/−0
LICENSE +23/−0
LICENSE.Tibell +30/−0
Makefile +3/−0
README.md +51/−0
Setup.hs +2/−0
System/Remote/Monitoring/Push.hs +185/−0
ekg-push.cabal +40/−0
examples/basic.hs +49/−0

+ CHANGES.md view

@@ -0,0 +1,7 @@+## 0.0.1 (2015-03-20)++ * Push agents "subscribe" via IO Action callbacks.++## 0.0.2 (2015-04-05)++ * Push agents subscribe via a broadcast channel.

+ LICENSE view

@@ -0,0 +1,23 @@+Copyright (c) 2015, Andrew Darqui+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.++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 HOLDER 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.

+ LICENSE.Tibell view

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

+ Makefile view

@@ -0,0 +1,3 @@+all:+	cabal sandbox init+	cabal install

+ README.md view

@@ -0,0 +1,51 @@+# Modifications of ekg-statsd by Johan Tibell to provide a generic "push" framework.++You will notice this is almost identical to ekg-statsd. It's just me trying to abstract away the basic functionality found in ekg-statsd, so not to have to duplicate this code anywhere else. Eventually I may try to clean up this "abstraction" and create a PR in ekg-core.++# TODO: Soon++So, the idea of ekg-push is to create a simple framework so that I can create agents such as:+- ekg-push-redis+- ekg-push-file+- ekg-push-statsd+- ekg-push-uberlog++etc..++Push agents simply 'subscribe' to the push handle returned by forkPush. Once subscribed, agents call consume and handle the Metric.Sample data how they want.++# Installation++You can just type 'make' to install ekg-push into a local sandbox. Or you can use cabal:++```+cabal install ekg-push+```++# Getting started++See examples/basic.hs++```+main :: IO ()+main = do+    store <- newStore+    registerGcMetrics store+    iters <- createCounter "iterations" store+    push <- forkPush defaultPushOptions { prefix = "pfx", suffix = "sfx" } store++    ch1 <- subscribe push+    _ <- forkIO $ forever $ do+            msg <- consume ch1+            putStrLn $ "subscription #1: " ++ show  msg++    let loop n = do+            evaluate $ mean [1..n]+            Counter.inc iters+            threadDelay 2000+            loop n++    loop 1000000+```++-- adarqui

+ Setup.hs view

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

+ System/Remote/Monitoring/Push.hs view

@@ -0,0 +1,185 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- | This library lets you push metric samples to a broadcast channel.+-- Consumers can then persist the metrics samples as they wish.+-- ekg-push is based heavily off of the ekg-statsd package which+-- can be found at: https://github.com/tibbe/ekg-statsd+--+-- Example usage:+--+-- > main = do+-- >     store <- newStore+-- >     push <- forkPush defaultPushOptions store+-- >     ch <- subscribe push+-- >     sample <- consume ch+-- >     putStrLn $ show sample+--+-- 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.Push+    (+      Push+    , PushChan+    , PushOptions(..)+    , pushThreadId+    , forkPush+    , defaultPushOptions+    , subscribe+    , consume+    ) where++import Control.Concurrent (ThreadId, myThreadId, threadDelay, throwTo)+import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan, dupChan)+import qualified Data.HashMap.Strict as M+import Data.Int (Int64)+import qualified Data.Text as T+import Data.Time.Clock.POSIX (getPOSIXTime)+import qualified System.Metrics as Metrics++#if __GLASGOW_HASKELL__ >= 706+import Control.Concurrent (forkFinally)+#else+import Control.Concurrent (forkIO)+import Control.Exception (SomeException, mask, try)+import Prelude hiding (catch)+#endif++-- | A handle that can be used to control the push sync thread.+-- Created by 'forkPush'.+data Push = Push+    { threadId :: {-# UNPACK #-} !ThreadId+    , mainCh :: Chan (Metrics.Sample)+    }++-- | A new PushChan is created on every call to subscribe.+-- This is essentially a dupChan of our main channel (mainCh).+data PushChan = PushChan+    { ch :: PushChanType }+    +type PushChanType = Chan Metrics.Sample++-- | The thread ID of the push sync thread. You can stop the sync by+-- killing this thread (i.e. by throwing it an asynchronous+-- exception.)+pushThreadId :: Push -> ThreadId+pushThreadId = threadId++-- | Options to control how to connect to the push server and how+-- often to flush metrics. The flush interval should be shorter than+-- the flush interval push itself uses to flush data to its+-- backends.+data PushOptions = PushOptions+    {+      -- | Data push interval, in ms.+      flushInterval :: !Int++      -- | Print debug output to stderr.+    , debug :: !Bool++      -- | 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+    }++-- | Defaults:+--+-- * @flushInterval@ = @1000@+--+-- * @debug@ = @False@+defaultPushOptions :: PushOptions+defaultPushOptions = PushOptions+    { flushInterval = 1000+    , debug         = False+    , prefix        = ""+    , suffix        = ""+    }++-- | Create a thread that periodically flushes the metrics in the+-- store to push.+forkPush :: PushOptions  -- ^ Options+           -> Metrics.Store  -- ^ Metric store+           -> IO Push      -- ^ Push sync handle+forkPush opts store = do+    me <- myThreadId+    ch <- newChan+    tid <- forkFinally (loop ch store emptySample opts) $ \ r -> do+        case r of+            Left e  -> throwTo me e+            Right _ -> return ()+    return $ Push tid ch+    where+        emptySample = M.empty++loop :: PushChanType -- ^ ekg-push clients subscribe to this channel+     -> Metrics.Store   -- ^ Metric Store+     -> Metrics.Sample  -- ^ Last sampled metrics+     -> PushOptions  -- ^ Options+     -> IO ()+loop ch store lastSample opts = do+    start <- time+    sample <- Metrics.sampleAll store+    let !diff = diffSamples opts lastSample sample+    writeChan ch diff -- Write the Metrics.Sample to our broadcast channel.+    end <- time+    threadDelay (flushInterval opts * 1000 - fromIntegral (end - start))+    loop ch store sample opts++-- | Subscribe to the push broadcast channel.+subscribe :: Push -> IO PushChan+subscribe Push{..} = do+    ch' <- dupChan mainCh+    return $ PushChan {+        ch = ch'+    }++-- | Consume a Metrics.Sample message from a subscribed channel.+consume :: PushChan -> IO Metrics.Sample+consume PushChan{..} = readChan ch++-- | Microseconds since epoch.+time :: IO Int64+time = (round . (* 1000000.0) . toDouble) `fmap` getPOSIXTime+  where toDouble = realToFrac :: Real a => a -> Double++diffSamples :: PushOptions -> Metrics.Sample -> Metrics.Sample -> Metrics.Sample+diffSamples opts 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+        where+            name' = T.append (prefix opts) (T.append name (suffix opts))++    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++------------------------------------------------------------------------+-- Backwards compatibility shims++#if __GLASGOW_HASKELL__ < 706+forkFinally :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId+forkFinally action and_then =+  mask $ \restore ->+    forkIO $ try (restore action) >>= and_then+#endif

+ ekg-push.cabal view

@@ -0,0 +1,40 @@+name:                   ekg-push+version:                0.0.2+synopsis:               Small framework to push metric deltas to a broadcast channel using the ekg-core library.+description:+  This library lets you push metric samples to a broadcast channel. Consumers can then persist the samples as they wish.+homepage:               https://github.com/adarqui/ekg-push+bug-reports:            https://github.com/adarqui/ekg-push/issues+license:                BSD3+license-files:          LICENSE, LICENSE.Tibell+author:                 Andrew Darqui+maintainer:             andrew.darqui@gmail.com+category:               System+build-type:             Simple+extra-source-files:     CHANGES.md+cabal-version:          >=1.18+extra-source-files:     Makefile, README.md++library+    exposed-modules:     +                        System.Remote.Monitoring.Push+    build-depends:+                        base >= 4.5 && < 4.8,+                        bytestring < 1.0,+                        ekg-core == 0.1.0.3,+                        text < 1.3,+                        time < 1.5,+                        unordered-containers < 0.3+    default-language:   Haskell2010+    ghc-options:        -Wall++executable basic+    main-is:            basic.hs+    hs-source-dirs:     examples+    build-depends:      base >= 4.5 && <= 4.9, ekg-core == 0.1.0.3, ekg-push == 0.0.2+    default-language:   Haskell2010+    ghc-options:        -Wall++source-repository head+  type:                 git+  location:             https://github.com/adarqui/ekg-push.git

+ examples/basic.hs view

@@ -0,0 +1,49 @@+{-# LANGUAGE OverloadedStrings #-}+module Main+    ( main+    ) where++import Control.Concurrent (threadDelay, forkIO)+import Control.Exception (evaluate)+import Control.Monad (forever)+import Data.List (foldl')+import System.Metrics (newStore, registerGcMetrics, createCounter)+import qualified System.Metrics.Counter as Counter (inc)+import System.Remote.Monitoring.Push (PushOptions (..), forkPush, defaultPushOptions, subscribe, consume)++-- 'sum' is using a non-strict lazy fold and will blow the stack.+sum' :: Num a => [a] -> a+sum' = foldl' (+) 0++mean :: Fractional a => [a] -> a+mean xs = sum' xs / fromIntegral (length xs)++main :: IO ()+main = do+    store <- newStore+    registerGcMetrics store+    iters <- createCounter "iterations" store++    -- Register our main ekg push channel+    push <- forkPush defaultPushOptions { prefix = "pfx_", suffix = "_sfx" } store++    -- Create two different subscriptions (ch1 & ch2)+    ch1 <- subscribe push+    _ <- forkIO $ forever $ do+            msg <- consume ch1+            putStrLn $ "subscription #1: " ++ show  msg++    ch2 <- subscribe push+    _ <- forkIO $ forever $ do+            msg <- consume ch2+            putStrLn $ "subscription #2: " ++ show msg++    let +        loop :: Double -> IO ()+        loop n = do+            _ <- evaluate $ mean [1..n]+            Counter.inc iters+            threadDelay 2000+            loop n++    loop 1000000