packages feed

instrument (empty) → 0.6.0.0

raw patch · 18 files changed

+2184/−0 lines, 18 filesdep +HUnitdep +QuickCheckdep +arraysetup-changed

Dependencies added: HUnit, QuickCheck, array, async, base, bytestring, cereal, cereal-text, conduit, containers, csv-conduit, data-default, errors, exceptions, hedis, hostname, instrument, mtl, network, path, path-io, quickcheck-instances, retry, safe-exceptions, safecopy, safecopy-hunit, statistics, stm, tasty, tasty-hunit, tasty-quickcheck, text, time, transformers, unix, vector, zlib

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2012, Ozgun Ataman, Soostone Inc++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 Ozgun Ataman 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
+ instrument.cabal view
@@ -0,0 +1,100 @@+name:                instrument+version:             0.6.0.0+synopsis:            Easy stats/metrics instrumentation for Haskell programs+license:             BSD3+license-file:        LICENSE+author:              Ozgun Ataman+maintainer:          ozgun.ataman@soostone.com+copyright:           Soostone Inc+category:            Data+build-type:          Simple+cabal-version:       >=1.10+++flag lib-Werror+  default: False+  manual: True++library+  default-language: Haskell2010+  hs-source-dirs: src+  exposed-modules:+    Instrument+    Instrument.Client+    Instrument.ClientClass+    Instrument.Counter+    Instrument.Measurement+    Instrument.Sampler+    Instrument.Types+    Instrument.Utils+    Instrument.Worker++  if flag(lib-Werror)+    ghc-options: -Werror++  ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates++  build-depends:+      base >= 4.5 && < 5+    , array+    , time+    , transformers+    , mtl+    , hostname+    , hedis >= 0.10.0+    , text+    , bytestring+    , statistics+    , vector+    , containers+    , cereal+    , cereal-text+    , data-default+    , csv-conduit+    , conduit >= 1.2.8+    , unix+    , errors+    , network+    , zlib          >= 0.6+    , retry         >= 0.7+    , exceptions+    , safe-exceptions+    , safecopy++test-suite test+  default-language: Haskell2010+  type: exitcode-stdio-1.0+  main-is: Main.hs+  if flag(lib-Werror)+    ghc-options: -Werror+  ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates -threaded -rtsopts -O0+  hs-source-dirs:+    test/src+  other-modules:+    Instrument.Tests.Client+    Instrument.Tests.Types+    Instrument.Tests.Utils+    Instrument.Tests.Arbitrary+    Instrument.Tests.Worker+  build-depends:+     base+   , data-default+   , hedis+   , HUnit+   , instrument+   , safecopy-hunit+   , tasty+   , tasty-quickcheck+   , tasty-hunit+   , transformers+   , QuickCheck+   , path >= 0.6.0+   , path-io+   , containers+   , cereal+   , bytestring+   , safecopy+   , quickcheck-instances+   , stm+   , async+    , safe-exceptions
+ src/Instrument.hs view
@@ -0,0 +1,17 @@+module Instrument+    ( -- * Data Collection (Client) Side+      module Instrument.Client++      -- * Data Processing (Backend) Side+    , module Instrument.Worker++      -- *+    , module Instrument.Types+    ) where++-------------------------------------------------------------------------------+import           Instrument.Client+import           Instrument.Types+import           Instrument.Worker+-------------------------------------------------------------------------------+
+ src/Instrument/Client.hs view
@@ -0,0 +1,366 @@+{-# LANGUAGE BangPatterns      #-}+{-# LANGUAGE CPP               #-}+{-# LANGUAGE OverloadedStrings #-}+module Instrument.Client+    ( Instrument+    , initInstrument+    , sampleI+    , timeI+    , timeI'+    , timeExI+    , TM.time+    , TM.timeEx+    , submitTime+    , incrementI+    , countI+    , timerMetricName+    , stripTimerPrefix+    , timerMetricNamePrefix+    , packetsKey+    ) where++-------------------------------------------------------------------------------+import           Control.Concurrent     (forkIO)+import           Control.Exception      (throw)+import           Control.Exception.Safe (MonadCatch, SomeException, tryAny)+import           Control.Monad+import           Control.Monad.IO.Class+import qualified Data.ByteString.Char8  as B+import           Data.IORef             (IORef, atomicModifyIORef, newIORef,+                                         readIORef)+import           Data.List              (isPrefixOf, stripPrefix)+import qualified Data.Map               as M+import           Data.Monoid            as Monoid+import qualified Data.SafeCopy          as SC+import qualified Data.Text              as T+#if MIN_VERSION_hedis(0,12,0)+import           Database.Redis         as R+#else+import           Database.Redis         as R hiding (HostName, time)+#endif+import           Network.HostName+-------------------------------------------------------------------------------+import qualified Instrument.Counter     as C+import qualified Instrument.Measurement as TM+import qualified Instrument.Sampler     as S+import           Instrument.Types+import           Instrument.Utils+-------------------------------------------------------------------------------++++-- | Initialize an instrument for measurement and feeding data into the system.+--+-- The resulting opaque 'Instrument' is meant to be threaded around in+-- your application to be later used in conjunction with 'sample' and+-- 'time'.+initInstrument :: ConnectInfo+               -- ^ Redis connection info+               -> InstrumentConfig+               -- ^ Instrument configuration. Use "def" if you don't have specific needs+               -> IO Instrument+initInstrument conn cfg = do+    p <- createInstrumentPool conn+    h        <- getHostName+    smplrs <- newIORef M.empty+    ctrs <- newIORef M.empty+    void $ forkIO $ indefinitely' $ submitSamplers smplrs p cfg+    void $ forkIO $ indefinitely' $ submitCounters ctrs p cfg+    return $ I h smplrs ctrs p+  where+    indefinitely' = indefinitely "Client" (seconds 1)++-------------------------------------------------------------------------------+mkSampledSubmission :: MetricName+                    -> Dimensions+                    -> [Double]+                    -> IO SubmissionPacket+mkSampledSubmission nm dims vals = do+  ts <- TM.getTime+  return $ SP ts nm (Samples vals) dims+++-------------------------------------------------------------------------------+addHostDimension :: HostName -> Dimensions -> Dimensions+addHostDimension host = M.insert hostDimension (DimensionValue (T.pack host))+++-------------------------------------------------------------------------------+mkCounterSubmission :: MetricName+                    -> Dimensions+                    -> Int+                    -> IO SubmissionPacket+mkCounterSubmission m dims i = do+    ts <- TM.getTime+    return $ SP ts m (Counter i) dims+++-- | Flush all samplers in Instrument+submitSamplers+  :: IORef Samplers+  -> Connection+  -> InstrumentConfig+  -> IO ()+submitSamplers smplrs rds cfg = do+  ss <- getSamplers smplrs+  mapM_ (flushSampler rds cfg) ss+++-- | Flush all samplers in Instrument+submitCounters+  :: IORef Counters+  -> Connection+  -> InstrumentConfig+  -> IO ()+submitCounters cs r cfg = do+    ss <- M.toList `liftM` readIORef cs+    mapM_ (flushCounter r cfg) ss+++-------------------------------------------------------------------------------+submitPacket :: (SC.SafeCopy a) => R.Connection -> MetricName -> Maybe Integer -> a -> IO ()+submitPacket r m mbound sp = void $ R.runRedis r $ R.multiExec $ do+  -- Write key with the stat contents+  _ <- push+  -- Remember the key we wrote to so we can retrieve it later without key scanning+  rememberKey+  where rk = B.concat [B.pack "_sq_", B.pack (metricName m)]+        push = case mbound of+          Just n  -> lpushBoundedTxn rk [encodeCompress sp] n+          Nothing -> (() <$) <$> R.lpush rk [encodeCompress sp]+        rememberKey = sadd packetsKey [rk]+++-------------------------------------------------------------------------------+-- | A key pointing to a SET of keys with _sq_ prefix, which contain+-- data packets. These are processed by worker.+packetsKey :: B.ByteString+packetsKey = "_sqkeys"+++-------------------------------------------------------------------------------+-- | Flush given counter to remote service and reset in-memory counter+-- back to 0.+flushCounter+  :: Connection+  -> InstrumentConfig+  -> ((MetricName, Dimensions), C.Counter)+  -> IO ()+flushCounter r cfg ((m, dims), c) =+    C.resetCounter c >>=+    mkCounterSubmission m dims >>=+    submitPacket r m (redisQueueBound cfg)+++-------------------------------------------------------------------------------+-- | Flush given sampler to remote service and flush in-memory queue+flushSampler+  :: Connection+  -> InstrumentConfig+  -> ((MetricName, Dimensions), S.Sampler)+  -> IO ()+flushSampler r cfg ((name, dims), sampler) = do+  vals <- S.get sampler+  case vals of+    [] -> return ()+    _ -> do+      S.reset sampler+      submitPacket r name (redisQueueBound cfg) =<< mkSampledSubmission name dims vals+++-------------------------------------------------------------------------------+-- | Increment a counter by one. Same as calling 'countI' with 1.+--+-- >>> incrementI \"uploadedFiles\" instr+incrementI+  :: (MonadIO m)+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> Instrument+  -> m ()+incrementI m hostDimPolicy rawDims i =+  liftIO $ C.increment =<< getCounter m dims i+  where+    dims = case hostDimPolicy of+      AddHostDimension      -> addHostDimension (hostName i) rawDims+      DoNotAddHostDimension -> rawDims+++-------------------------------------------------------------------------------+-- | Increment a counter by n.+--+-- >>> countI \"uploadedFiles\" 1 instr+countI+  :: MonadIO m+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> Int+  -> Instrument+  -> m ()+countI m hostDimPolicy rawDims n i =+  liftIO $ C.add n =<< getCounter m dims i+  where+    dims = case hostDimPolicy of+      AddHostDimension      -> addHostDimension (hostName i) rawDims+      DoNotAddHostDimension -> rawDims+++-- | Run a monadic action while measuring its runtime. Push the+-- measurement into the instrument system.+--+-- >>> timeI \"fileUploadTime\" policy dims instr $ uploadFile file+timeI+  :: (MonadIO m)+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> Instrument+  -> m a+  -> m a+timeI nm hostDimPolicy rawDims = do+  timeI' (const (pure (Just (nm, hostDimPolicy, rawDims))))++-- | like timeI but with maximum flexibility: it uses the result and+-- can use the monad to determine the metric name, host dimension+-- policy, and dimensions or even not emit a timing at all. Some use cases include:+--+-- * Emit different metrics or suppress metrics on error+-- * Fetch some dimension info from the environment+timeI'+  :: (MonadIO m)+  => (a -> m (Maybe (MetricName, HostDimensionPolicy, Dimensions)))+  -> Instrument+  -> m a+  -> m a+timeI' toMetric i act = do+  (!secs, !res) <- TM.time act+  metricMay <- toMetric res+  case metricMay of+    Just (nm, hostDimPolicy, rawDims) -> do+      submitTime nm hostDimPolicy rawDims secs i+    Nothing -> pure ()+  return res++-- | Run a monadic action while measuring its runtime. Push the measurement into+-- the instrument system. rethrows exceptions and sends a different Metric on+-- failure+--+-- >>> timeExI \"fileUploadTimeError\" \"fileUploadTime\" policy dims instr $ uploadFile file+timeExI+  :: (MonadIO m, MonadCatch m)+  => (Either SomeException a -> (MetricName, HostDimensionPolicy, Dimensions))+  -> Instrument+  -> m a+  -> m a+timeExI toMetric i act = do+  resE <- timeI' (pure . Just . toMetric) i (tryAny act)+  either throw pure resE++-------------------------------------------------------------------------------+timerMetricName :: MetricName -> MetricName+timerMetricName name@(MetricName nameS) =+  if timerMetricNamePrefix `isPrefixOf` nameS+     then name+     else MetricName (timerMetricNamePrefix Monoid.<> nameS)+++-------------------------------------------------------------------------------+stripTimerPrefix :: MetricName -> MetricName+stripTimerPrefix (MetricName n) = case stripPrefix timerMetricNamePrefix n of+  Just unprefixed -> MetricName unprefixed+  Nothing         -> MetricName n+++-------------------------------------------------------------------------------+timerMetricNamePrefix :: String+timerMetricNamePrefix = "time."+++-------------------------------------------------------------------------------+-- | Sometimes dimensions are determined within a code block that+-- you're measuring. In that case, you can use 'time' to measure it+-- and when you're ready to submit, use 'submitTime'.+--+-- Also, you may be pulling time details from some external source+-- that you can't measure with 'timeI' yourself.+--+-- Note: for legacy purposes, metric name will have "time." prepended+-- to it.+submitTime+  :: (MonadIO m)+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> Double+  -- ^ Time in seconds+  -> Instrument+  -> m ()+submitTime nameRaw hostDimPolicy rawDims secs i =+  liftIO $ sampleI nm hostDimPolicy rawDims secs i+  where+    nm = timerMetricName nameRaw+++-------------------------------------------------------------------------------+-- | Record given measurement under the given label.+--+-- Instrument will automatically capture useful stats like min, max,+-- count, avg, stdev and percentiles within a single flush interval.+--+-- Say we check our upload queue size every minute and record+-- something like:+--+-- >>> sampleI \"uploadQueue\" 27 inst+sampleI+  :: MonadIO m+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> Double+  -> Instrument+  -> m ()+sampleI name hostDimPolicy rawDims v i =+  liftIO $ S.sample v =<< getSampler name dims i+  where+    dims = case hostDimPolicy of+      AddHostDimension      -> addHostDimension (hostName i) rawDims+      DoNotAddHostDimension -> rawDims+++-------------------------------------------------------------------------------+getCounter :: MetricName -> Dimensions -> Instrument -> IO C.Counter+getCounter nm dims i = getRef C.newCounter (nm, dims) (counters i)+++-- | Get or create a sampler under given name+getSampler :: MetricName -> Dimensions -> Instrument -> IO S.Sampler+getSampler name dims i = getRef (S.new 1000) (name, dims) (samplers i)+++-- | Get a list of current samplers present+getSamplers :: IORef Samplers -> IO [((MetricName, Dimensions), S.Sampler)]+getSamplers ss = M.toList `fmap` readIORef ss+++-- | Lookup a 'Ref' by name in the given map.  If no 'Ref' exists+-- under the given name, create a new one, insert it into the map and+-- return it.+getRef :: Ord k => IO b -> k -> IORef (M.Map k b) -> IO b+getRef f name mapRef = do+    empty <- f+    atomicModifyIORef mapRef $ \ m ->+        case M.lookup name m of+            Nothing  -> let m' = M.insert name empty m+                        in (m', empty)+            Just ref -> (m, ref)+{-# INLINABLE getRef #-}++-- | Bounded version of lpush which truncates *new* data first. This+-- effectively stops accepting data until the queue shrinks below the+-- bound. Occurs in a transaction for composibility with larger transactions.+lpushBoundedTxn :: B.ByteString -> [B.ByteString] -> Integer -> RedisTx (Queued ())+lpushBoundedTxn k vs mx = do+  _ <- lpush k vs+  fmap (() <$) (ltrim k (-mx) (-1))
+ src/Instrument/ClientClass.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE FlexibleInstances     #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances  #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Instrument.ClientClass+-- Copyright   :  Soostone Inc+-- License     :  BSD3+--+-- Maintainer  :  Ozgun Ataman+-- Stability   :  experimental+--+-- This module mimics the functionality of Instrument.Client but+-- instead exposes a typeclass facilitated interface. Once you define+-- the typeclass for your application's main monad, you can call all+-- the mesaurement functions directly.+----------------------------------------------------------------------------++module Instrument.ClientClass+    ( I.Instrument+    , I.initInstrument+    , HasInstrument (..)+    , sampleI+    , timeI+    , countI+    , incrementI+    ) where++-------------------------------------------------------------------------------+import           Control.Monad.IO.Class+import           Control.Monad.Reader+-------------------------------------------------------------------------------+import qualified Instrument.Client      as I+import           Instrument.Types+-------------------------------------------------------------------------------+++class HasInstrument m where+  getInstrument :: m I.Instrument++instance (Monad m) => HasInstrument (ReaderT I.Instrument m) where+  getInstrument = ask+++-- | Run a monadic action while measuring its runtime+timeI :: (MonadIO m, HasInstrument m)+     => MetricName+     -> HostDimensionPolicy+     -> Dimensions+     -> m a+     -> m a+timeI name hostDimPolicy dims act = do+  i <- getInstrument+  I.timeI name hostDimPolicy dims i act+++-- | Record a measurement sample+sampleI :: (MonadIO m, HasInstrument m )+       => MetricName+       -> HostDimensionPolicy+       -> Dimensions+       -> Double+       -> m ()+sampleI name hostDimPolicy dims val =+  I.sampleI name hostDimPolicy dims val =<< getInstrument+++-------------------------------------------------------------------------------+incrementI+  :: ( MonadIO m+     , HasInstrument m+     )+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> m ()+incrementI m hostDimPolicy dims =+  I.incrementI m hostDimPolicy dims =<< getInstrument+++-------------------------------------------------------------------------------+countI+  :: ( MonadIO m+     , HasInstrument m+     )+  => MetricName+  -> HostDimensionPolicy+  -> Dimensions+  -> Int+  -> m ()+countI m hostDimPolicy dims v =+  I.countI m hostDimPolicy dims v =<< getInstrument
+ src/Instrument/Counter.hs view
@@ -0,0 +1,44 @@+{-# LANGUAGE BangPatterns #-}++module Instrument.Counter+    ( Counter+    , newCounter+    , readCounter+    , resetCounter+    , add+    , increment+    ) where++-------------------------------------------------------------------------------+import           Control.Monad+import           Data.IORef+-------------------------------------------------------------------------------++newtype Counter = Counter { unCounter :: IORef Int }++-------------------------------------------------------------------------------+newCounter :: IO Counter+newCounter = Counter `liftM` newIORef 0+++-------------------------------------------------------------------------------+readCounter :: Counter -> IO Int+readCounter (Counter i) = readIORef i+++-------------------------------------------------------------------------------+-- | Reset the counter while reading it+resetCounter :: Counter -> IO Int+resetCounter (Counter i) = atomicModifyIORef i f+    where f i' = (0, i')++-------------------------------------------------------------------------------+increment :: Counter -> IO ()+increment = add 1+++-------------------------------------------------------------------------------+add :: Int -> Counter -> IO ()+add x c = atomicModifyIORef (unCounter c) f+    where+      f !i = (i + x, ())
+ src/Instrument/Measurement.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE BangPatterns        #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators       #-}++-- |+-- Module      : Instrument.Measurement+-- Copyright   : (c) 2009, 2010 Bryan O'Sullivan+--               (c) 2012, Ozgun Ataman+--+-- License     : BSD-style++module Instrument.Measurement+    (+      getTime+    , time+    , time_+    , timeEx+    ) where++-------------------------------------------------------------------------------+import           Control.Monad.IO.Class+import           Data.Time.Clock.POSIX  (getPOSIXTime)+import           Control.Exception (SomeException)+import           Control.Exception.Safe (MonadCatch, tryAny)+-------------------------------------------------------------------------------+++-- | Measure how long action took, in seconds along with its result+time :: MonadIO m =>  m a -> m (Double, a)+time act = do+  start <- liftIO getTime+  !result <- act+  end <- liftIO getTime+  let !delta = end - start+  return (delta, result)++-- | Measure how long action took, even if they fail+timeEx :: (MonadCatch m, MonadIO m) => m a -> m (Double, Either SomeException a)+timeEx = time . tryAny++-- | Just measure how long action takes, discard its result+time_ :: MonadIO m => m a -> m Double+time_  = fmap fst . time+++-------------------------------------------------------------------------------+getTime :: IO Double+getTime = realToFrac `fmap` getPOSIXTime
+ src/Instrument/Sampler.hs view
@@ -0,0 +1,116 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  Instrument.Sampler+-- Copyright   :  Soostone Inc+-- License     :  BSD3+--+-- Maintainer  :  Ozgun Ataman+-- Stability   :  experimental+--+-- A container that can capture actual numeric values, efficiently and+-- concurrently buffering them in memory. We can impose a cap on how+-- many values are captured at a time.+----------------------------------------------------------------------------++module Instrument.Sampler+    ( Sampler (..)+    , new+    , sample+    , get+    , reset+    ) where++-------------------------------------------------------------------------------+import           Control.Concurrent.MVar+import           Control.Exception       (mask_, onException)+import           Control.Monad+import qualified Data.Vector.Mutable     as V+-------------------------------------------------------------------------------+++-- |'BoundedChan' is an abstract data type representing a bounded channel.+data Buffer a = B {+       _size     :: Int+     , _contents :: V.IOVector a+     , _writePos :: MVar Int+     }++-- Versions of modifyMVar and withMVar that do not 'restore' the previous mask state when running+-- 'io', with added modification strictness.  The lack of 'restore' may make these perform better+-- than the normal version.  Moving strictness here makes using them more pleasant.+{-# INLINE modifyMVar_mask #-}+modifyMVar_mask :: MVar a -> (a -> IO (a,b)) -> IO b+modifyMVar_mask m io =+  mask_ $ do+    a <- takeMVar m+    (a',b) <- io a `onException` putMVar m a+    putMVar m $! a'+    return b++{-# INLINE modifyMVar_mask_ #-}+modifyMVar_mask_ :: MVar a -> (a -> IO a) -> IO ()+modifyMVar_mask_ m io =+  mask_ $ do+    a <- takeMVar m+    a' <- io a `onException` putMVar m a+    putMVar m $! a'+++-------------------------------------------------------------------------------+newBuffer :: Int -> IO (Buffer a)+newBuffer lim = do+  pos  <- newMVar 0+  entries <- V.new lim+  return (B lim entries pos)+++-- | Write an element to the channel. If the channel is full, nothing+-- will happen and the function will return immediately. We don't want+-- to disturb production code.+writeBuffer :: Buffer a -> a -> IO ()+writeBuffer (B size contents wposMV) x = modifyMVar_mask_ wposMV $+  \wpos ->+    case wpos >= size of+      True -> return wpos       -- buffer full, don't do anything+      False -> do+        V.write contents wpos x+        return (succ wpos)+++-------------------------------------------------------------------------------+getBuffer :: Buffer a -> IO [a]+getBuffer (B _size contents pos) = do+  wpos <- modifyMVar_mask pos $ \ wpos -> return (wpos, wpos)+  forM [0.. (wpos - 1)] $ \ i -> (V.read contents i)++++-------------------------------------------------------------------------------+resetBuffer :: Buffer a -> IO ()+resetBuffer (B _size _els pos) = modifyMVar_mask_ pos (const (return 0))+++++-- | An in-memory, bounded buffer for measurement samples.+newtype Sampler = S { unS :: Buffer Double }+++-- | Create a new, empty sampler+new :: Int -> IO Sampler+new i = S `fmap` newBuffer i+++-------------------------------------------------------------------------------+sample :: Double -> Sampler -> IO ()+sample v s = writeBuffer (unS s) v+++-------------------------------------------------------------------------------+get :: Sampler -> IO [Double]+get (S buffer) = getBuffer buffer+++-------------------------------------------------------------------------------+reset :: Sampler -> IO ()+reset (S buf) = resetBuffer buf
+ src/Instrument/Types.hs view
@@ -0,0 +1,317 @@+{-# LANGUAGE CPP                        #-}+{-# LANGUAGE DeriveGeneric              #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE TemplateHaskell            #-}+{-# LANGUAGE TypeFamilies               #-}++module Instrument.Types+  ( createInstrumentPool+  , Samplers+  , Counters+  , Instrument(..)+  , InstrumentConfig(..)+  , SubmissionPacket(..)+  , MetricName(..)+  , DimensionName(..)+  , DimensionValue(..)+  , Dimensions+  , Payload(..)+  , Aggregated(..)+  , AggPayload(..)+  , Stats(..)+  , hostDimension+  , HostDimensionPolicy(..)+  , Quantile(..)+  ) where++-------------------------------------------------------------------------------+import           Control.Applicative   as A+import qualified Data.ByteString.Char8 as B+import           Data.Default+import           Data.IORef+import qualified Data.Map              as M+import           Data.Monoid           as Monoid+import qualified Data.SafeCopy         as SC+import           Data.Serialize        as Ser+import           Data.Serialize.Text   ()+import           Data.String+import           Data.Text             (Text)+import qualified Data.Text             as T+#if MIN_VERSION_hedis(0,12,0)+import           Database.Redis        as R+#else+import           Database.Redis        as R hiding (HostName, time)+#endif+import           GHC.Generics+import           Network.HostName+-------------------------------------------------------------------------------+import qualified Instrument.Counter    as C+import qualified Instrument.Sampler    as S+-------------------------------------------------------------------------------+++-------------------------------------------------------------------------------+createInstrumentPool :: ConnectInfo -> IO Connection+createInstrumentPool ci = do+  c <- connect ci {+         connectMaxIdleTime = 15+       , connectMaxConnections = 1 }+  return c+++-- Map of user-defined samplers.+type Samplers = M.Map (MetricName, Dimensions) S.Sampler++-- Map of user-defined counters.+type Counters = M.Map (MetricName, Dimensions) C.Counter+++type Dimensions = M.Map DimensionName DimensionValue+++newtype MetricName = MetricName {+      metricName :: String+    } deriving (Eq,Show,Generic,Ord,IsString,Serialize)+++data Instrument = I {+      hostName :: HostName+    , samplers :: !(IORef Samplers)+    , counters :: !(IORef Counters)+    , redis    :: Connection+    }++data InstrumentConfig = ICfg {+      redisQueueBound :: Maybe Integer+    }++instance Default InstrumentConfig where+  def = ICfg Nothing+++-- | Submitted package of collected samples+data SubmissionPacket_v0 = SP_v0 {+      spTimeStamp_v0 :: !Double+    -- ^ Timing of this submission+    , spHostName_v0  :: !HostName+    -- ^ Who sent it+    , spName_v0      :: String+    -- ^ Metric name+    , spPayload_v0   :: Payload+    -- ^ Collected values+    } deriving (Eq,Show,Generic)+++instance Serialize SubmissionPacket_v0+++data SubmissionPacket = SP {+      spTimeStamp  :: !Double+    -- ^ Timing of this submission+    , spName       :: !MetricName+    -- ^ Metric name+    , spPayload    :: !Payload+    -- ^ Collected values+    , spDimensions :: !Dimensions+    -- ^ Defines slices that this packet belongs to. This allows+    -- drill-down on the backends. For instance, you could do+    -- "server_name" "app1" or "queue_name" "my_queue"+    } deriving (Eq,Show,Generic)+++instance Serialize SubmissionPacket where+  get = (to <$> gGet) <|> (upgradeSP0 <$> Ser.get)+++instance SC.Migrate SubmissionPacket where+  type MigrateFrom SubmissionPacket = SubmissionPacket_v0+  migrate = upgradeSP0+++upgradeSP0 :: SubmissionPacket_v0 -> SubmissionPacket+upgradeSP0 SP_v0 {..} = SP+  { spTimeStamp = spTimeStamp_v0+  , spName = MetricName spName_v0+  , spPayload = spPayload_v0+  , spDimensions = M.singleton hostDimension (DimensionValue (T.pack spHostName_v0))+  }+++-------------------------------------------------------------------------------+-- | Convention for the dimension of the hostname. Used in the client+-- to inject hostname into the parameters map+hostDimension :: DimensionName+hostDimension = "host"+++-------------------------------------------------------------------------------+-- | Should we automatically pull the host and add it as a+-- dimension. Used at the call site of the various metrics ('timeI',+-- 'sampleI', etc). Hosts are basically grandfathered in as a+-- dimension and the functionality of automatically injecting them is+-- useful, but it is not relevant to some metrics and actually makes+-- some metrics difficult to use depending on the backend, so we made+-- them opt-in.+data HostDimensionPolicy = AddHostDimension+                         | DoNotAddHostDimension+                         deriving (Show, Eq)+++-------------------------------------------------------------------------------+newtype DimensionName = DimensionName {+    dimensionName :: Text+  } deriving (Eq,Ord,Show,Generic,Serialize,IsString)+++-------------------------------------------------------------------------------+newtype DimensionValue = DimensionValue {+    dimensionValue :: Text+  } deriving (Eq,Ord,Show,Generic,Serialize,IsString)+++-------------------------------------------------------------------------------+data Payload+    = Samples { unSamples :: [Double] }+    | Counter { unCounter :: Int }+  deriving (Eq, Show, Generic)++instance Serialize Payload+++-------------------------------------------------------------------------------+data Aggregated_v0 = Aggregated_v0 {+      aggTS_v0      :: Double+      -- ^ Timestamp for this aggregation+    , aggName_v0    :: String+    -- ^ Name of the metric+    , aggGroup_v0   :: B.ByteString+    -- ^ The aggregation level/group for this stat+    , aggPayload_v0 :: AggPayload+    -- ^ Calculated stats for the metric+    } deriving (Eq,Show,Generic)+++instance Serialize Aggregated_v0+++data Aggregated_v1 = Aggregated_v1 {+      aggTS_v1      :: Double+      -- ^ Timestamp for this aggregation+    , aggName_v1    :: MetricName+    -- ^ Name of the metric+    , aggGroup_v1   :: B.ByteString+    -- ^ The aggregation level/group for this stat+    , aggPayload_v1 :: AggPayload+    -- ^ Calculated stats for the metric+    } deriving (Eq,Show,Generic)+++instance SC.Migrate Aggregated_v1 where+  type MigrateFrom Aggregated_v1 = Aggregated_v0+  migrate = upgradeAggregated_v0+++upgradeAggregated_v0 :: Aggregated_v0 -> Aggregated_v1+upgradeAggregated_v0 a = Aggregated_v1+  { aggTS_v1 = aggTS_v0 a+  , aggName_v1 = MetricName (aggName_v0 a)+  , aggGroup_v1 = (aggGroup_v0 a)+  , aggPayload_v1 = (aggPayload_v0 a)+  }++instance Serialize Aggregated_v1 where+  get = (to <$> gGet) <|> (upgradeAggregated_v0 <$> Ser.get)+++data Aggregated = Aggregated {+      aggTS         :: Double+      -- ^ Timestamp for this aggregation+    , aggName       :: MetricName+    -- ^ Name of the metric+    , aggPayload    :: AggPayload+    -- ^ Calculated stats for the metric+    , aggDimensions :: Dimensions+    } deriving (Eq,Show, Generic)+++upgradeAggregated_v1 :: Aggregated_v1 -> Aggregated+upgradeAggregated_v1 a = Aggregated+  { aggTS = aggTS_v1 a+  , aggName = aggName_v1 a+  , aggPayload = aggPayload_v1 a+  , aggDimensions = Monoid.mempty+  }+++instance SC.Migrate Aggregated where+  type MigrateFrom Aggregated = Aggregated_v1+  migrate = upgradeAggregated_v1+++instance Serialize Aggregated where+  get = (to <$> gGet) <|> (upgradeAggregated_v1 <$> Ser.get)+++-- | Resulting payload for metrics aggregation+data AggPayload+    = AggStats Stats+    | AggCount Int+    deriving (Eq,Show, Generic)++instance Serialize AggPayload+++++instance Default AggPayload where+    def = AggStats def++instance Default Aggregated where+    def = Aggregated 0 "" def mempty+++-------------------------------------------------------------------------------+data Stats = Stats {+      smean      :: Double+    , ssum       :: Double+    , scount     :: Int+    , smax       :: Double+    , smin       :: Double+    , srange     :: Double+    , sstdev     :: Double+    , sskewness  :: Double+    , skurtosis  :: Double+    , squantiles :: M.Map Int Double+    } deriving (Eq, Show, Generic)++++instance Default Stats where+    def = Stats 0 0 0 0 0 0 0 0 0 mempty++instance Serialize Stats+++-------------------------------------------------------------------------------+-- | Integer quantile, valid values range from 1-99, inclusive.+newtype Quantile = Q { quantile :: Int } deriving (Show, Eq, Ord)++instance Bounded Quantile where+  minBound = Q 1+  maxBound = Q 99+++-------------------------------------------------------------------------------+$(SC.deriveSafeCopy 0 'SC.base ''Payload)+$(SC.deriveSafeCopy 0 'SC.base ''SubmissionPacket_v0)+$(SC.deriveSafeCopy 1 'SC.extension ''SubmissionPacket)+$(SC.deriveSafeCopy 0 'SC.base ''AggPayload)+$(SC.deriveSafeCopy 0 'SC.base ''Aggregated_v0)+$(SC.deriveSafeCopy 1 'SC.extension ''Aggregated_v1)+$(SC.deriveSafeCopy 2 'SC.extension ''Aggregated)+$(SC.deriveSafeCopy 0 'SC.base ''Stats)+$(SC.deriveSafeCopy 0 'SC.base ''DimensionName)+$(SC.deriveSafeCopy 0 'SC.base ''DimensionValue)+$(SC.deriveSafeCopy 0 'SC.base ''MetricName)
+ src/Instrument/Utils.hs view
@@ -0,0 +1,148 @@+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE OverloadedStrings         #-}++module Instrument.Utils+    ( formatDecimal+    , formatInt+    , showT+    , showBS+    , collect+    , noDots+    , encodeCompress+    , decodeCompress+    , indefinitely+    , seconds+    , milliseconds+    , for+    ) where+++-------------------------------------------------------------------------------+import           Codec.Compression.GZip+import           Control.Applicative    ((<|>))+import           Control.Concurrent     (threadDelay)+import           Control.Exception      (SomeException)+import           Control.Monad+import           Control.Monad.Catch    (Handler (..))+import           Control.Retry+import qualified Data.ByteString.Char8  as B+import           Data.ByteString.Lazy   (fromStrict, toStrict)+import qualified Data.Map               as M+import qualified Data.Map.Strict        as MS+import qualified Data.SafeCopy          as SC+import           Data.Serialize+import           Data.Text              (Text)+import qualified Data.Text              as T+import           Numeric+import           System.IO+-------------------------------------------------------------------------------++++-------------------------------------------------------------------------------+collect :: (Ord b)+        => [a]+        -> (a -> b)+        -> (a -> c)+        -> M.Map b [c]+collect as mkKey mkVal = foldr step M.empty as+    where+      step x acc = MS.insertWith (++) (mkKey x) ([mkVal x]) acc+++-------------------------------------------------------------------------------+noDots :: Text -> Text+noDots = T.intercalate "_" . T.splitOn "."+++-------------------------------------------------------------------------------+showT :: Show a => a -> Text+showT = T.pack . show++showBS :: Show a => a -> B.ByteString+showBS = B.pack . show++-------------------------------------------------------------------------------+formatInt :: RealFrac a => a -> Text+formatInt i = showT ((floor i) :: Int)+++-------------------------------------------------------------------------------+formatDecimal+    :: RealFloat a+    => Int+    -- ^ Digits after the point+    -> Bool+    -- ^ Add thousands sep?+    -> a+    -- ^ Number+    -> Text+formatDecimal n th i =+    let res = T.pack . showFFloat (Just n) i $ ""+    in if th then addThousands res else res++++-------------------------------------------------------------------------------+addThousands :: Text -> Text+addThousands t = T.concat [n', dec]+    where+      (n,dec) = T.span (/= '.') t+      n' = T.reverse . T.intercalate "," . T.chunksOf 3 . T.reverse $ n+++-------------------------------------------------------------------------------+-- | Serialize and compress with GZip in that order. This is the only+-- function we use for serializing to Redis.+encodeCompress :: SC.SafeCopy a => a -> B.ByteString+encodeCompress = toStrict . compress . runPutLazy . SC.safePut++-------------------------------------------------------------------------------+-- | Decompress from GZip and deserialize in that order. Tries to+-- decode SafeCopy first and falls back to Serialize if that fails to+-- account for old data. Note that encodeCompress only serializes to+-- SafeCopy so writes will be updated.+decodeCompress :: (SC.SafeCopy a, Serialize a) => B.ByteString -> Either String a+decodeCompress = decodeWithFallback . decompress . fromStrict+  where+    decodeWithFallback lbs = runGetLazy SC.safeGet lbs <|> decodeLazy lbs+++-------------------------------------------------------------------------------+-- | Run an IO repeatedly with the given delay in microseconds. If+-- there are exceptions in the inner loop, they are logged to stderr,+-- prefixed with the given string context and retried at an exponential+-- backoff capped at 60 seconds between.+indefinitely :: String -> Int -> IO () -> IO ()+indefinitely ctx n = forever . delayed . logAndBackoff ctx+  where+    delayed = (>> threadDelay n)+++-------------------------------------------------------------------------------+logAndBackoff :: String -> IO () -> IO ()+logAndBackoff ctx = recovering policy [h] . const+  where+    policy = capDelay (seconds 60) (exponentialBackoff (milliseconds 50))+    h _ = Handler (\e -> logError e >> return True)+    logError :: SomeException -> IO ()+    logError e = hPutStrLn stderr msg+      where+        msg = "Caught exception in " ++ ctx ++ ": " ++ show e ++ ". Retrying..."+++-------------------------------------------------------------------------------+-- | Convert seconds to microseconds+seconds :: Int -> Int+seconds = (* milliseconds 1000)+++-------------------------------------------------------------------------------+-- | Convert milliseconds to microseconds+milliseconds :: Int -> Int+milliseconds = (* 1000)+++-------------------------------------------------------------------------------+for :: (Functor f) => f a -> (a -> b) -> f b+for = flip fmap
+ src/Instrument/Worker.hs view
@@ -0,0 +1,445 @@+{-# LANGUAGE BangPatterns        #-}+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Instrument.Worker+    ( initWorkerCSV+    , initWorkerCSV'+    , initWorkerGraphite+    , initWorkerGraphite'+    , work+    , initWorker+    , AggProcess(..)+    -- * Configuring agg processes+    , AggProcessConfig(..)+    , standardQuantiles+    , noQuantiles+    , quantileMap+    , defAggProcessConfig+    -- * Exported for testing+    , expandDims+    ) where++-------------------------------------------------------------------------------+import           Control.Error+import           Control.Monad+import           Control.Monad.IO.Class+import qualified Data.ByteString.Char8  as B+import           Data.Conduit           (runConduit, (.|))+import qualified Data.Conduit.List      as CL+import           Data.CSV.Conduit+import           Data.Default+import qualified Data.Map               as M+import qualified Data.SafeCopy          as SC+import           Data.Semigroup         as Semigroup+import           Data.Serialize+import qualified Data.Set               as Set+import qualified Data.Text              as T+import qualified Data.Text.IO           as T+import qualified Data.Vector.Unboxed    as V+import           Database.Redis         as R hiding (decode)+import           Network.Socket         as N+import qualified Statistics.Quantile    as Q+import           Statistics.Sample+import           System.IO+import           System.Posix+-------------------------------------------------------------------------------+import           Instrument.Client      (packetsKey, stripTimerPrefix,+                                         timerMetricName)+import qualified Instrument.Measurement as TM+import           Instrument.Types+import           Instrument.Utils+-------------------------------------------------------------------------------++++-------------------------------------------------------------------------------+-- | A CSV backend to store aggregation results in a CSV+initWorkerCSV+  :: ConnectInfo+  -> FilePath+  -- ^ Target file name+  -> Int+  -- ^ Aggregation period / flush interval in seconds+  -> AggProcessConfig+  -> IO ()+initWorkerCSV conn fp n cfg =+  initWorker "CSV Worker" conn n =<< initWorkerCSV' fp cfg+++-------------------------------------------------------------------------------+-- | Create an AggProcess that dumps to CSV. Use this to compose with+-- other AggProcesses+initWorkerCSV'+  :: FilePath+  -- ^ Target file name+  -> AggProcessConfig+  -> IO AggProcess+initWorkerCSV' fp cfg = do+  !res <- fileExist fp+  !h <- openFile fp AppendMode+  hSetBuffering h LineBuffering+  unless res $+    T.hPutStrLn h $ rowToStr defCSVSettings . M.keys $ aggToCSV def+  return $ putAggregateCSV h cfg+++-------------------------------------------------------------------------------+-- | Initialize a Graphite backend+initWorkerGraphite+    :: ConnectInfo+    -- ^ Redis connection+    -> Int+    -- ^ Aggregation period / flush interval in seconds+    -> HostName+    -- ^ Graphite host+    -> Int+    -- ^ Graphite port+    -> AggProcessConfig+    -> IO ()+initWorkerGraphite conn n server port cfg =+    initWorker "Graphite Worker" conn n =<< initWorkerGraphite' server port cfg+++-------------------------------------------------------------------------------+-- | Crete an AggProcess that dumps to graphite. Use this to compose+-- with other AggProcesses+initWorkerGraphite'+    :: HostName+    -- ^ Graphite host+    -> Int+    -- ^ Graphite port+    -> AggProcessConfig+    -> IO AggProcess+initWorkerGraphite' server port cfg = do+  addr <- resolve server (fromIntegral port)+  sock <- open addr+  h <- N.socketToHandle sock ReadWriteMode+  hSetBuffering h LineBuffering+  return $ putAggregateGraphite h cfg+  where+    portNumberToServiceName :: N.PortNumber -> N.ServiceName+    portNumberToServiceName = show+    resolve host portNumber = do+      let hints = N.defaultHints { N.addrSocketType = N.Stream }+      addr:_ <- N.getAddrInfo+        (Just hints)+        (Just host)+        (Just (portNumberToServiceName portNumber))+      return addr+    open addr = do+      sock <- N.socket+        (N.addrFamily addr)+        (N.addrSocketType addr)+        (N.addrProtocol addr)+      N.connect sock (N.addrAddress addr)+      return sock+++-------------------------------------------------------------------------------+-- | Generic utility for making worker backends. Will retry+-- indefinitely with exponential backoff.+initWorker :: String -> ConnectInfo -> Int -> AggProcess -> IO ()+initWorker wname conn n f = do+    p <- createInstrumentPool conn+    indefinitely' $ work p n f+  where+    indefinitely' = indefinitely wname (seconds n)++++-------------------------------------------------------------------------------+-- | Extract statistics out of the given sample for this flush period+mkStats :: Set.Set Quantile -> Sample -> Stats+mkStats qs s = Stats { smean = mean s+                     , ssum = V.sum s+                     , scount = V.length s+                     , smax = V.maximum s+                     , smin = V.minimum s+                     , srange = range s+                     , sstdev = stdDev s+                     , sskewness = skewness s+                     , skurtosis = kurtosis s+                     , squantiles = quantiles }+  where+    quantiles = M.fromList (mkQ 100 . quantile <$> Set.toList qs)+    mkQ mx i = (i, Q.weightedAvg i mx s)+++-------------------------------------------------------------------------------+-- | Go over all pending stats buffers in redis.+work :: R.Connection -> Int -> AggProcess -> IO ()+work r n f = runRedis r $ do+    dbg "entered work block"+    estimate <- either (const 0) id <$> scard packetsKey+    runConduit $+      CL.unfoldM nextKey estimate .|+        CL.mapM_ (processSampler n f)+  where+    nextKey estRemaining+      | estRemaining > 0 = do+         mk <- spop packetsKey+         return $ case mk of+           Right (Just k) -> Just (k, estRemaining - 1)+           _              -> Nothing+      | otherwise = return Nothing+++-------------------------------------------------------------------------------+processSampler+    :: Int+    -- ^ Flush interval - determines resolution+    -> AggProcess+    -- ^ What to do with aggregation results+    -> B.ByteString+    -- ^ Redis buffer for this metric+    -> Redis ()+processSampler n (AggProcess cfg f) k = do+  packets <- popLAll k+  case packets of+    [] -> return ()+    _ -> do+      let nm = spName . head $ packets+          -- with and without timer prefix+          qs = quantilesFn (stripTimerPrefix nm) <> quantilesFn (timerMetricName nm)+          byDims :: M.Map Dimensions [SubmissionPacket]+          byDims = collect packets spDimensions id+          mkAgg xs =+              case spPayload $ head xs of+                Samples _ -> AggStats . mkStats qs . V.fromList .+                             concatMap (unSamples . spPayload) $+                             xs+                Counter _ -> AggCount . sum .+                             map (unCounter . spPayload) $+                             xs+      t <- (fromIntegral . (* n) . (`div` n) . round) `liftM` liftIO TM.getTime+      let aggs = map mkDimsAgg $ M.toList $ expandDims $ byDims+          mkDimsAgg (dims, ps) = Aggregated t nm (mkAgg ps) dims+      mapM_ f aggs+      return ()+  where+    quantilesFn = metricQuantiles cfg+++-------------------------------------------------------------------------------+-- | Take a map of packets by dimensions and *add* aggregations of the+-- existing dims that isolate each distinct dimension/dimensionvalue+-- pair + one more entry with an empty dimension set that aggregates+-- the whole thing.+-- worked example:+--+-- Given:+-- { {d1=>d1v1,d2=>d2v1} => p1+-- , {d1=>d1v1,d2=>d2v2} => p2+-- }+-- Produces:+-- { {d1=>d1v1,d2=>d2v1} => p1+-- , {d1=>d1v1,d2=>d2v2} => p2+-- , {d1=>d1v1} => p1 + p2+-- , {d2=>d2v1} => p1+-- , {d2=>d2v2} => p2+-- , {} => p1 + p2+-- }+expandDims+  :: forall packets. (Monoid packets, Eq packets)+  => M.Map Dimensions packets+  -> M.Map Dimensions packets+expandDims m =+  -- left-biased so technically if we have anything occupying the aggregated spots, leave them be+  m <> additions <> fullAggregation+  where+    distinctPairs :: Set.Set (DimensionName, DimensionValue)+    distinctPairs = Set.fromList (mconcat (M.toList <$> M.keys m))+    additions = foldMap mkIsolatedMap distinctPairs+    mkIsolatedMap :: (DimensionName, DimensionValue) -> M.Map Dimensions packets+    mkIsolatedMap dPair =+      let matches = snd <$> filter ((== dPair) . fst) mFlat+      in if matches == mempty+            then mempty+            else M.singleton (uncurry M.singleton dPair) (mconcat matches)+    mFlat :: [((DimensionName, DimensionValue), packets)]+    mFlat = [ ((dn, dv), packets)+            | (dimensionsMap, packets) <- M.toList m+            , (dn, dv) <- M.toList dimensionsMap]+    -- All packets across any combination of dimensions+    fullAggregation = M.singleton mempty (mconcat (M.elems m))+++-- | A function that does something with the aggregation results. Can+-- implement multiple backends simply using this. Note that Semigroup and Monoid instances are provided for defaulting and combining agg processes.+data AggProcess = AggProcess+  { apConfig :: AggProcessConfig+  , apProc   :: Aggregated -> Redis ()+  }+++instance Semigroup.Semigroup AggProcess where+  (AggProcess cfg1 prc1) <> (AggProcess cfg2 prc2) =+    AggProcess (cfg1 <> cfg2) (\agg -> prc1 agg >> prc2 agg)+++instance Monoid AggProcess where+  mempty = AggProcess mempty (const (pure ()))+  mappend = (<>)+++-------------------------------------------------------------------------------+-- | General configuration for agg processes. Defaulted with 'def',+-- 'defAggProcessConfig', and 'mempty'. Configurations can be combined+-- with (<>) from Monoid or Semigroup.+data AggProcessConfig = AggProcessConfig+  { metricQuantiles :: MetricName -> Set.Set Quantile+  -- ^ What quantiles should we calculate for any given metric, if+  -- any? We offer some common patterns for this in 'quantileMap',+  -- 'standardQuantiles', and 'noQuantiles'.+  }+++instance Semigroup AggProcessConfig where+  AggProcessConfig f1 <> AggProcessConfig f2 =+    let f3 = f1 <> f2+    in AggProcessConfig f3+++instance Monoid AggProcessConfig where+  mempty = AggProcessConfig mempty+  mappend = (<>)+++-- | Uses 'standardQuantiles'.+defAggProcessConfig :: AggProcessConfig+defAggProcessConfig = AggProcessConfig standardQuantiles+++instance Default AggProcessConfig where+  def = defAggProcessConfig+++-- | Regardless of metric, produce no quantiles.+noQuantiles :: MetricName -> Set.Set Quantile+noQuantiles = const mempty+++-- | This is usually a good, comprehensive default. Produces quantiles+-- 10,20,30,40,50,60,70,80,90,99. *Note:* for some backends like+-- cloudwatch, each quantile produces an additional metric, so you+-- should probably consider using something more limited than this.+standardQuantiles :: MetricName -> Set.Set Quantile+standardQuantiles _ =+  Set.fromList [Q 10,Q 20,Q 30,Q 40,Q 50,Q 60,Q 70,Q 80,Q 90,Q 99]+++-- | If you have a fixed set of metric names, this is often a+-- convenient way to express quantiles-per-metric.+quantileMap+  :: M.Map MetricName (Set.Set Quantile)+  -> Set.Set Quantile+  -- ^ What to return on miss+  -> (MetricName -> Set.Set Quantile)+quantileMap m qdef mn = fromMaybe qdef (M.lookup mn m)+++-------------------------------------------------------------------------------+-- | Store aggregation results in a CSV file+putAggregateCSV :: Handle -> AggProcessConfig -> AggProcess+putAggregateCSV h cfg = AggProcess cfg $ \agg ->+  let d = rowToStr defCSVSettings $ aggToCSV agg+  in liftIO $ T.hPutStrLn h d+++typePrefix :: AggPayload -> T.Text+typePrefix AggStats{} = "samples"+typePrefix AggCount{} = "counts"+++-------------------------------------------------------------------------------+-- | Push data into a Graphite database using the plaintext protocol+putAggregateGraphite :: Handle -> AggProcessConfig -> AggProcess+putAggregateGraphite h cfg = AggProcess cfg $ \agg ->+    let (ss, ts) = mkStatsFields agg+        -- Expand dimensions into one datum per dimension pair as the group+        mkLines (m, val) = for (M.toList (aggDimensions agg)) $ \(DimensionName dimName, DimensionValue dimVal) -> T.concat+            [ "inst."+            , typePrefix (aggPayload agg), "."+            ,  T.pack (metricName (aggName agg)), "."+            , m, "."+            , dimName, "."+            , dimVal, " "+            , val, " "+            , ts ]+    in liftIO $ mapM_ (mapM_ (T.hPutStrLn h) . mkLines) ss+++-------------------------------------------------------------------------------+-- | Pop all keys in a redis List+popLAll :: (Serialize a, SC.SafeCopy a) => B.ByteString -> Redis [a]+popLAll k = do+  res <- popLMany k 100+  case res of+    [] -> return res+    _  -> (res ++ ) `liftM` popLAll k+++-------------------------------------------------------------------------------+-- | Pop up to N items from a queue. It will pop from left and preserve order.+popLMany :: (Serialize a, SC.SafeCopy a) => B.ByteString -> Int -> Redis [a]+popLMany k n = do+    res <- replicateM n pop+    case sequence res of+      Left _   -> return []+      Right xs -> return $ mapMaybe conv $ catMaybes xs+    where+      pop = R.lpop k+      conv x =  hush $ decodeCompress x+++-------------------------------------------------------------------------------+-- | Need to pull in a debugging library here.+dbg :: (Monad m) => String -> m ()+dbg _ = return ()+++-- ------------------------------------------------------------------------------+-- dbg :: (MonadIO m) => String -> m ()+-- dbg s = debug $ "Instrument.Worker: " ++ s+++-------------------------------------------------------------------------------+-- | Expand count aggregation to have the full columns+aggToCSV :: Aggregated -> M.Map T.Text T.Text+aggToCSV agg@Aggregated{..} = els <> defFields <> dimFields+  where+    els :: MapRow T.Text+    els = M.fromList $+            ("metric", T.pack (metricName aggName)) :+            ("timestamp", ts) :+            fields+    (fields, ts) = mkStatsFields agg+    defFields = M.fromList $ fst $ mkStatsFields $ agg { aggPayload =  (AggStats def) }+    dimFields = M.fromList [(k,v) | (DimensionName k, DimensionValue v) <- M.toList aggDimensions]+++-------------------------------------------------------------------------------+-- | Get agg results into a form ready to be output+mkStatsFields :: Aggregated -> ([(T.Text, T.Text)], T.Text)+mkStatsFields Aggregated{..}  = (els, ts)+    where+      els =+        case aggPayload of+          AggStats Stats{..} ->+              [ ("mean", formatDecimal 6 False smean)+              , ("count", showT scount)+              , ("max", formatDecimal 6 False smax)+              , ("min", formatDecimal 6 False smin)+              , ("srange", formatDecimal 6 False srange)+              , ("stdDev", formatDecimal 6 False sstdev)+              , ("sum", formatDecimal 6 False ssum)+              , ("skewness", formatDecimal 6 False sskewness)+              , ("kurtosis", formatDecimal 6 False skurtosis)+              ] ++ (map mkQ $ M.toList squantiles)+          AggCount i ->+              [ ("count", showT i)]++      mkQ (k,v) = (T.concat ["percentile_", showT k], formatDecimal 6 False v)+      ts = formatInt aggTS
+ test/src/Instrument/Tests/Arbitrary.hs view
@@ -0,0 +1,24 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Instrument.Tests.Arbitrary+    (+    ) where+++-------------------------------------------------------------------------------+import           Test.QuickCheck+import           Test.QuickCheck.Instances ()+-------------------------------------------------------------------------------+import           Instrument.Types+-------------------------------------------------------------------------------+++instance Arbitrary DimensionName where+  arbitrary = DimensionName <$> arbitrary+++instance Arbitrary DimensionValue where+  arbitrary = DimensionValue <$> arbitrary+++instance Arbitrary MetricName where+  arbitrary = MetricName <$> arbitrary
+ test/src/Instrument/Tests/Client.hs view
@@ -0,0 +1,201 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards   #-}+module Instrument.Tests.Client+    ( tests+    ) where++-------------------------------------------------------------------------------+import           Control.Concurrent+import           Control.Concurrent.Async+import           Control.Concurrent.STM+import           Control.Exception.Safe     (Exception, throwM, tryAny)+import           Control.Monad+import           Control.Monad.IO.Class+import           Data.Default+import           Data.List                  (find)+import qualified Data.Map                   as M+import           Data.Monoid                as Monoid+import           Database.Redis+import           System.Timeout+import           Test.Tasty+import           Test.Tasty.HUnit+import           Test.Tasty.QuickCheck+-------------------------------------------------------------------------------+import           Instrument.Client+import           Instrument.Tests.Arbitrary ()+import           Instrument.Types+import           Instrument.Worker+-------------------------------------------------------------------------------++tests :: TestTree+tests = testGroup "Instrument.Client"+    [ withRedisCleanup $ testCase "queue bounding works" . queue_bounding_test+    , withRedisCleanup $ testCase "multiple keys eventually get flushed" . multi_key_test+    , withRedisCleanup $ testCaseSteps "also times exceptions" . time_test+    , timerMetricNameTests+    ]++-------------------------------------------------------------------------------+queue_bounding_test :: IO Connection -> IO ()+queue_bounding_test mkConn = do+    conn <- mkConn+    instr <- initInstrument redisCI icfg+    agg <- newEmptyMVar+    replicateM_ 2 (sampleI key DoNotAddHostDimension dims 1 instr >> sleepFlush)+    -- redis queue slots now full of the above aggregate+    -- bounds will drop these+    sampleI key DoNotAddHostDimension dims 100 instr+    sleepFlush+    let worker = work conn 1 (AggProcess defAggProcessConfig (liftIO . putMVar agg))+    withAsync worker $ \_ -> do+      Aggregated { aggPayload = AggStats Stats {..} } <- takeMVar agg+      assertEqual "throws away newer data exceeding bounds"+                  (2, 1, 1, 2)+                  (scount, smin, smax, ssum)+++-------------------------------------------------------------------------------+sleepFlush :: IO ()+sleepFlush = threadDelay 1100000++data TimeException =+  TimeException+  deriving (Show)++instance Exception TimeException++-------------------------------------------------------------------------------+time_test :: IO Connection -> (String -> IO a) -> IO ()+time_test mkConn step = do+  conn <- mkConn+  instr <- initInstrument redisCI icfg+  aggsRef <- newTVarIO []++  let toMetric resE =+        case resE of+          Left _ -> ("instrument-test-timeex-error", DoNotAddHostDimension, Monoid.mempty)+          Right _ -> ("instrument-test-timeex", DoNotAddHostDimension, Monoid.mempty)++  void . tryAny . timeExI toMetric instr $ throwM TimeException++  timeExI toMetric instr $ pure ()++  void . tryAny+    . timeI+      "instrument-test-time-error"+      DoNotAddHostDimension+      Monoid.mempty+      instr+    $ throwM TimeException++  timeI+    "instrument-test-time"+    DoNotAddHostDimension+    Monoid.mempty+    instr+    $ pure ()++  sleepFlush++  let collectAggs = work conn 1 (AggProcess defAggProcessConfig (\agg -> liftIO (atomically (modifyTVar' aggsRef (agg:)))))++  withAsync collectAggs $ \worker -> do+    maggs <- timeout 2000000 $ atomically $ do+      aggs <- readTVar aggsRef+      check (length aggs == 3)+      return aggs+    cancel worker+    case maggs of+      Nothing -> assertFailure "Waited 2 seconds and never received any aggs!"+      Just aggs -> do+        let getCount payload =+              case payload of+              (AggStats stats) -> scount stats+              (AggCount n) -> n++            countMatches name n = do+              void $ step name+              case filter ((== MetricName ("time." <> name)). aggName) aggs of+                  [agg] -> getCount (aggPayload agg) @?= n+                  _ -> assertFailure (name <> "has multiple aggregations")++        countMatches "instrument-test-timeex" 1+        countMatches "instrument-test-timeex-error" 1+        countMatches "instrument-test-time" 1++        void $ step "instrument-test-time-error"+        length (filter ((== MetricName "time.instrument-test-time-error"). aggName) aggs) @?= 0+++-------------------------------------------------------------------------------+multi_key_test :: IO Connection -> IO ()+multi_key_test mkConn = do+  conn <- mkConn+  instr <- initInstrument redisCI icfg+  aggsRef <- newTVarIO []+  sampleI "instrument-test1" DoNotAddHostDimension Monoid.mempty 1 instr+  sampleI "instrument-test2" DoNotAddHostDimension mempty 2 instr+  sleepFlush+  let collectAggs = work conn 1 (AggProcess defAggProcessConfig (\agg -> liftIO (atomically (modifyTVar' aggsRef (agg:)))))+  withAsync collectAggs $ \worker -> do+    maggs <- timeout 2000000 $ atomically $ do+      aggs <- readTVar aggsRef+      check (length aggs >= 2)+      return aggs+    cancel worker+    case maggs of+      Nothing -> assertFailure "Waited 2 seconds and never received any aggs!"+      Just aggs -> do+        length aggs @?= 2+        let findAgg n expectedMean = do+              let magg = find ((== MetricName n) . aggName) aggs+              case magg of+                Nothing -> assertFailure ("Expected to find agg with name " <> n <> " but could not")+                Just (Aggregated { aggPayload = AggStats (Stats { smean = actualMean})}) -> actualMean @?= expectedMean+                Just _ -> assertFailure ("Expected agg with name " <> n <> " to contain Stats but it did not.")+        findAgg "instrument-test1" 1.0+        findAgg "instrument-test2" 2.0+++-------------------------------------------------------------------------------+timerMetricNameTests :: TestTree+timerMetricNameTests = testGroup "timerMetricName"+  [ testProperty "is idempotent" $ \mn ->+      let r1 = timerMetricName mn+          r2 =  timerMetricName r1+      in r1 === r2+  , testProperty "is idempotent for timers" $ \(MetricName mnBase) ->+      let mn = MetricName (timerMetricNamePrefix <> mnBase)+          r1 = timerMetricName mn+          r2 =  timerMetricName r1+      in r1 === r2+  , testCase "adds time. prefix" $ do+      timerMetricName (MetricName "foo") @?= MetricName "time.foo"+  ]+++-------------------------------------------------------------------------------+withRedisCleanup :: (IO Connection -> TestTree) -> TestTree+withRedisCleanup = withResource (connect redisCI) cleanup+  where+    cleanup conn = void $ runRedis conn $ do+      ks <- either mempty id <$> smembers packetsKey+      _ <- del (packetsKey:ks)+      quit+++-------------------------------------------------------------------------------+redisCI :: ConnectInfo+redisCI = defaultConnectInfo+++icfg :: InstrumentConfig+icfg = def { redisQueueBound = Just 2 }+++key :: MetricName+key = MetricName "instrument-test"+++dims :: Dimensions+dims = M.fromList [(DimensionName "server", DimensionValue "app1")]
+ test/src/Instrument/Tests/Types.hs view
@@ -0,0 +1,35 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell   #-}+module Instrument.Tests.Types+    ( tests+    ) where+++-------------------------------------------------------------------------------+import qualified Data.Map            as M+import           Path+import           Test.HUnit.SafeCopy+import           Test.Tasty+import           Test.Tasty.HUnit+-------------------------------------------------------------------------------+import           Instrument.Types+-------------------------------------------------------------------------------++--TODO: test parse of .serialize files++tests :: TestTree+tests = testGroup "Instrument.Types"+  [ testCase "Stats SafeCopy" $+      testSafeCopy FailMissingFiles $(mkRelFile "test/data/Instrument/Types/Stats.safecopy") stats+  , testCase "Payload SafeCopy" $+      testSafeCopy FailMissingFiles $(mkRelFile "test/data/Instrument/Types/Payload.safecopy") payload+  , testCase "SubmissionPacket SafeCopy" $+      testSafeCopy FailMissingFiles $(mkRelFile "test/data/Instrument/Types/SubmissionPacket.safecopy") submissionPacket+  , testCase "Aggregated SafeCopy" $+      testSafeCopy FailMissingFiles $(mkRelFile "test/data/Instrument/Types/Aggregated.safecopy") aggregated+  ]+  where+    stats = Stats 1 2 3 4 5 6 7 8 9 (M.singleton 10 11)+    payload = Samples [1.2, 2.3, 4.5]+    submissionPacket = SP 3.4 "metric" payload (M.singleton hostDimension "example.org")+    aggregated = Aggregated 3.4 "metric" (AggStats stats) mempty
+ test/src/Instrument/Tests/Utils.hs view
@@ -0,0 +1,104 @@+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell     #-}+module Instrument.Tests.Utils+    ( tests+    ) where++-------------------------------------------------------------------------------+import           Control.Concurrent+import           Control.Monad+import qualified Data.ByteString       as B+import           Data.IORef+import qualified Data.Map              as M+import           Data.SafeCopy+import           Data.Serialize+import           Path+import qualified Path.IO               as PIO+import           System.Timeout+import           Test.Tasty+import           Test.Tasty.HUnit+import           Test.Tasty.QuickCheck+-------------------------------------------------------------------------------+import           Instrument.Types+import           Instrument.Utils+-------------------------------------------------------------------------------++++tests :: TestTree+tests = testGroup "Instrument.Utils"+    [ encodeDecodeTests+    , withResource spawnWorker killWorker $ testCase "indefinitely retries" . indefinitely_retry+    ]+++-------------------------------------------------------------------------------+encodeDecodeTests :: TestTree+encodeDecodeTests = testGroup "encodeCompress/decodeCompress"+  [ testProperty "encode/decode compress roundtrip" encode_decode_roundtrip+  , testCase "Stats decode" $+      testDecode $(mkRelFile "test/data/Instrument/Types/Stats.gz") stats+  , testCase "Payload decode" $+      testDecode $(mkRelFile "test/data/Instrument/Types/Payload.gz") payload+  , testCase "SubmissionPacket decode" $+      testDecode $(mkRelFile "test/data/Instrument/Types/SubmissionPacket.gz") submissionPacket+  , testCase "Aggregated decode" $+      testDecode $(mkRelFile "test/data/Instrument/Types/Aggregated.gz") aggregated+  ]+  where+    stats = Stats 1 2 3 4 5 6 7 8 9 (M.singleton 10 11)+    payload = Samples [1.2, 2.3, 4.5]+    submissionPacket = SP 3.4 "metric" payload (M.singleton hostDimension "example.org")+    aggregated = Aggregated 3.4 "metric" (AggStats stats) mempty+    testDecode :: forall a b. (Serialize a, SafeCopy a, Eq a, Show a) => Path b File -> a -> Assertion+    testDecode fp v = do+      exists <- PIO.doesFileExist fp+      unless exists $ do+        PIO.ensureDir (parent fp)+        B.writeFile (toFilePath fp) (encodeCompress v)+      res <- decodeCompress <$> B.readFile (toFilePath fp)+      (res :: Either String a) @?= Right v+++-------------------------------------------------------------------------------+encode_decode_roundtrip :: String -> Property+encode_decode_roundtrip a = roundtrip a === Right a+  where+    roundtrip = decodeCompress . encodeCompress+++-------------------------------------------------------------------------------+indefinitely_retry :: IO (MVar (), t1, t) -> IO ()+indefinitely_retry setup = do+    (called, _signal, _) <- setup+    wasCalled <- timeout (milliseconds 100) $ takeMVar called+    assertEqual "retries until success" (Just ()) wasCalled+  where+++-------------------------------------------------------------------------------+spawnWorker :: IO (MVar (), IO (), ThreadId)+spawnWorker = do+  (called, signal) <- nopeNopeYep+  tid <- forkIO $ indefinitely "Test Worker" 0 signal+  return (called, signal, tid)+++-------------------------------------------------------------------------------+killWorker :: (t1, t, ThreadId) -> IO ()+killWorker (_, _, tid) = killThread tid+++-------------------------------------------------------------------------------+nopeNopeYep :: IO (MVar (), IO ())+nopeNopeYep = do+    counter <- newIORef (1 :: Int)+    mv <- newEmptyMVar+    return $ (mv, go counter mv)+  where+    go counter mv = do+      count <- readIORef counter+      modifyIORef' counter (+1)+      when (count > 2) $ void $ tryPutMVar mv ()+
+ test/src/Instrument/Tests/Worker.hs view
@@ -0,0 +1,75 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}+module Instrument.Tests.Worker+    ( tests+    ) where+++-------------------------------------------------------------------------------+import qualified Data.Map                   as M+import           Data.Monoid                as Monoid+import           Test.Tasty+import           Test.Tasty.HUnit+import           Test.Tasty.QuickCheck+-------------------------------------------------------------------------------+import           Instrument.Tests.Arbitrary ()+import           Instrument.Types+import           Instrument.Worker+-------------------------------------------------------------------------------+++tests :: TestTree+tests = testGroup "Instrument.Worker"+  [ expandDimsTests+  ]+++-------------------------------------------------------------------------------+expandDimsTests :: TestTree+expandDimsTests = testGroup "expandDims"+  [ testProperty "input is a submap of output" $ \(Dims dims) ->+      let res = expandDims dims+      in counterexample+           ("Expected " <> show dims <> " to be a submap of " <> show res)+           (dims `M.isSubmapOf` res)+  , testProperty "always includes an aggregate with no dimensions" $ \(Dims dims) ->+      let res = expandDims dims+      in M.member Monoid.mempty res+  --TODO: more and then a test of the worked example+  , testProperty "no one member exceeds the total number of packets" $ \(Dims dims) ->+      let totalPacketCount = sum (length <$> dims)+          res = expandDims dims+          lengths = length <$> res+      in all (<= totalPacketCount) lengths+  , testCase "worked example from the readme" $ do+      let m = M.fromList [+             (M.fromList [(d1, d1v1), (d2, d2v1)], [p1])+           , (M.fromList [(d1, d1v1), (d2, d2v2)], [p2])+           ]+      let expected = M.fromList [+             (M.fromList [(d1, d1v1), (d2, d2v1)], [p1])+           , (M.fromList [(d1, d1v1), (d2, d2v2)], [p2])+           -- additions+           , (M.fromList [(d1, d1v1)], [p1, p2])+           , (M.fromList [(d2, d2v1)], [p1])+           , (M.fromList [(d2, d2v2)], [p2])+           , ((M.fromList []), [p1, p2])+           ]+      expandDims m @?= expected+  ]+  where+    d1 = DimensionName "d1"+    d2 = DimensionName "d2"+    d1v1 = DimensionValue "d1v1"+    d2v1 = DimensionValue "d2v1"+    d2v2 = DimensionValue "d2v2"+    p1 :: String+    p1 = "p1"+    p2 :: String+    p2 = "p2"++++-- | Fixes the packet type for type inference+newtype Dims = Dims (M.Map Dimensions [Char])+  deriving (Show, Eq, Arbitrary)
+ test/src/Main.hs view
@@ -0,0 +1,18 @@+module Main (main) where++-------------------------------------------------------------------------------+import           Test.Tasty+-------------------------------------------------------------------------------+import qualified Instrument.Tests.Client+import qualified Instrument.Tests.Types+import qualified Instrument.Tests.Utils+import qualified Instrument.Tests.Worker+-------------------------------------------------------------------------------++main :: IO ()+main = defaultMain $ testGroup "tests"+    [ Instrument.Tests.Utils.tests+    , Instrument.Tests.Client.tests+    , Instrument.Tests.Types.tests+    , Instrument.Tests.Worker.tests+    ]