instrument 0.6.0.0 → 0.6.1.0

raw patch · 18 files changed

+1215/−1047 lines, 18 filesdep +atomic-primopssetup-changedPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: atomic-primops

API changes (from Hackage documentation)

Files

• Setup.hs +1/−0
• instrument.cabal +3/−1
• src/Instrument.hs +12/−11
• src/Instrument/Client.hs +187/−179
• src/Instrument/ClientClass.hs +51/−51
• src/Instrument/Counter.hs +36/−21
• src/Instrument/Measurement.hs +16/−17
• src/Instrument/Sampler.hs +28/−38
• src/Instrument/Types.hs +290/−220
• src/Instrument/Utils.hs +60/−66
• src/Instrument/Worker.hs +235/−222
• test/src/Instrument/Tests/Arbitrary.hs +9/−9
• test/src/Instrument/Tests/Client.hs +89/−86
• test/src/Instrument/Tests/Counter.hs +59/−0
• test/src/Instrument/Tests/Types.hs +24/−20
• test/src/Instrument/Tests/Utils.hs +47/−49
• test/src/Instrument/Tests/Worker.hs +54/−50
• test/src/Main.hs +14/−7

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

@@ -1,5 +1,5 @@ name:                instrument-version:             0.6.0.0+version:             0.6.1.0 synopsis:            Easy stats/metrics instrumentation for Haskell programs license:             BSD3 license-file:        LICENSE@@ -60,6 +60,7 @@     , exceptions     , safe-exceptions     , safecopy+    , atomic-primops  test-suite test   default-language: Haskell2010@@ -76,6 +77,7 @@     Instrument.Tests.Utils     Instrument.Tests.Arbitrary     Instrument.Tests.Worker+    Instrument.Tests.Counter   build-depends:      base    , data-default

@@ -1,17 +1,18 @@ module Instrument-    ( -- * Data Collection (Client) Side-      module Instrument.Client+  ( -- * Data Collection (Client) Side+    module Instrument.Client, -      -- * Data Processing (Backend) Side-    , module Instrument.Worker+    -- * Data Processing (Backend) Side+    module Instrument.Worker, -      -- *-    , module Instrument.Types-    ) where+    -- *+    module Instrument.Types,+  )+where  --------------------------------------------------------------------------------import           Instrument.Client-import           Instrument.Types-import           Instrument.Worker--------------------------------------------------------------------------------+import Instrument.Client+import Instrument.Types+import Instrument.Worker +-------------------------------------------------------------------------------

@@ -1,164 +1,171 @@-{-# LANGUAGE BangPatterns      #-}-{-# LANGUAGE CPP               #-}+{-# 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+  ( 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+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.Counter as C import qualified Instrument.Measurement as TM-import qualified Instrument.Sampler     as S-import           Instrument.Types-import           Instrument.Utils---------------------------------------------------------------------------------+import qualified Instrument.Sampler as S+import Instrument.Types+import Instrument.Utils+import Network.HostName +-------------------------------------------------------------------------------  -- | 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 ::+  -- | Redis connection info+  ConnectInfo ->+  -- | Instrument configuration. Use "def" if you don't have specific needs+  InstrumentConfig ->+  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+  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 ::+  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 ::+  MetricName ->+  Dimensions ->+  Integer ->+  IO SubmissionPacket mkCounterSubmission m dims i = do-    ts <- TM.getTime-    return $ SP ts m (Counter i) dims-+  ts <- TM.getTime+  return $ SP ts m (Counter i) dims  -- | Flush all samplers in Instrument-submitSamplers-  :: IORef Samplers-  -> Connection-  -> InstrumentConfig-  -> IO ()+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 ::+  IORef Counters ->+  Connection ->+  InstrumentConfig ->+  IO () submitCounters cs r cfg = do-    ss <- M.toList `liftM` readIORef cs-    mapM_ (flushCounter r cfg) ss-+  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]-+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 ::+  Connection ->+  InstrumentConfig ->+  ((MetricName, Dimensions), C.Counter) ->+  IO () flushCounter r cfg ((m, dims), c) =-    C.resetCounter c >>=-    mkCounterSubmission m dims >>=-    submitPacket r m (redisQueueBound cfg)-+  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 ::+  Connection ->+  InstrumentConfig ->+  ((MetricName, Dimensions), S.Sampler) ->+  IO () flushSampler r cfg ((name, dims), sampler) = do   vals <- S.get sampler   case vals of@@ -167,58 +174,57 @@       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 ::+  (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+      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 ::+  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+      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 ::+  (MonadIO m) =>+  MetricName ->+  HostDimensionPolicy ->+  Dimensions ->+  Instrument ->+  m a ->+  m a timeI nm hostDimPolicy rawDims = do   timeI' (const (pure (Just (nm, hostDimPolicy, rawDims)))) @@ -228,12 +234,12 @@ -- -- * 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' ::+  (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@@ -248,12 +254,12 @@ -- 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 ::+  (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@@ -262,23 +268,21 @@ timerMetricName :: MetricName -> MetricName timerMetricName name@(MetricName nameS) =   if timerMetricNamePrefix `isPrefixOf` nameS-     then name-     else MetricName (timerMetricNamePrefix Monoid.<> 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-+  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'.@@ -288,22 +292,22 @@ -- -- 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 ::+  (MonadIO m) =>+  MetricName ->+  HostDimensionPolicy ->+  Dimensions ->+  -- | Time in seconds+  Double ->+  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,@@ -313,49 +317,53 @@ -- something like: -- -- >>> sampleI \"uploadQueue\" 27 inst-sampleI-  :: MonadIO m-  => MetricName-  -> HostDimensionPolicy-  -> Dimensions-  -> Double-  -> Instrument-  -> m ()+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+      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.+-- Note mapRef is append only, so we can use double-checked locking+-- to avoid synchronization on reads. That makes hot-path lock free.+-- We'll only synchronize the first time metric is inserted. getRef :: Ord k => IO b -> k -> IORef (M.Map k b) -> IO b getRef f name mapRef = do-    empty <- f-    atomicModifyIORef mapRef $ \ m ->+  mapRef' <- readIORef mapRef+  case M.lookup name mapRef' of+    Just ref -> pure ref+    Nothing -> 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 #-}+          Nothing ->+            let m' = M.insert name empty m+             in (m', empty)+          Just ref -> (m, ref)+{-# INLINEABLE getRef #-}  -- | Bounded version of lpush which truncates *new* data first. This -- effectively stops accepting data until the queue shrinks below the@@ -363,4 +371,4 @@ lpushBoundedTxn :: B.ByteString -> [B.ByteString] -> Integer -> RedisTx (Queued ()) lpushBoundedTxn k vs mx = do   _ <- lpush k vs-  fmap (() <$) (ltrim k (-mx) (-1))+  fmap (() <$) (ltrim k (- mx) (-1))

@@ -1,9 +1,12 @@-{-# LANGUAGE FlexibleContexts      #-}-{-# LANGUAGE FlexibleInstances     #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE UndecidableInstances  #-}+{-# LANGUAGE UndecidableInstances #-}  -----------------------------------------------------------------------------++----------------------------------------------------------------------------+ -- | -- Module      :  Instrument.ClientClass -- Copyright   :  Soostone Inc@@ -16,26 +19,25 @@ -- 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+  ( 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+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@@ -43,52 +45,50 @@ 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 ::+  (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 ::+  (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 ::+  ( 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 ::+  ( MonadIO m,+    HasInstrument m+  ) =>+  MetricName ->+  HostDimensionPolicy ->+  Dimensions ->+  Int ->+  m () countI m hostDimPolicy dims v =   I.countI m hostDimPolicy dims v =<< getInstrument

@@ -1,44 +1,59 @@ {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TypeApplications #-}  module Instrument.Counter-    ( Counter-    , newCounter-    , readCounter-    , resetCounter-    , add-    , increment-    ) where+  ( Counter,+    newCounter,+    readCounter,+    resetCounter,+    add,+    increment,+  )+where  --------------------------------------------------------------------------------import           Control.Monad-import           Data.IORef+import Data.IORef+import qualified Data.Atomics.Counter as A ------------------------------------------------------------------------------- -newtype Counter = Counter { unCounter :: IORef Int }+data Counter = Counter {_atomicCounter :: A.AtomicCounter, _lastResetValue :: IORef Int}  ------------------------------------------------------------------------------- newCounter :: IO Counter-newCounter = Counter `liftM` newIORef 0-+newCounter = Counter <$> A.newCounter 0 <*> newIORef 0  --------------------------------------------------------------------------------readCounter :: Counter -> IO Int-readCounter (Counter i) = readIORef i +-- | Reads current counter value+readCounter :: Counter -> IO Integer+readCounter (Counter i lastReset) = calculateDelta <$> readIORef lastReset <*> A.readCounter i +-- | Our counters are represented as Int (with machine word size) and increment+-- | only, with no possibility to reset. Since we report deltas, we need facility+-- | to reset counter: we do that by storing last reported value in IORef.+-- | When application will run for long-enough counters will inevitably overflow.+-- | In such case our delta would be negative – this does not make sense for increment only.+-- | To overcome this issue we apply heuristic: when current value of counter+-- | is lower last value the counter has been reset, rollover happened and we need to+-- | take it into account. Otherwise, we can simply subtract values.+calculateDelta :: Int -> Int -> Integer+calculateDelta lastReset current | current < lastReset =+  fromIntegral (maxBound @Int - lastReset) + fromIntegral (current - minBound @Int) + 1+calculateDelta lastReset current = fromIntegral current - fromIntegral lastReset+ -------------------------------------------------------------------------------+ -- | Reset the counter while reading it-resetCounter :: Counter -> IO Int-resetCounter (Counter i) = atomicModifyIORef i f-    where f i' = (0, i')+resetCounter :: Counter -> IO Integer+resetCounter (Counter i lastReset) = do+  ctrValue <- A.readCounter i+  oldLast <- atomicModifyIORef' lastReset $ \oldLast -> (ctrValue, oldLast)+  pure $ calculateDelta oldLast ctrValue  ------------------------------------------------------------------------------- increment :: Counter -> IO () increment = add 1 - ------------------------------------------------------------------------------- add :: Int -> Counter -> IO ()-add x c = atomicModifyIORef (unCounter c) f-    where-      f !i = (i + x, ())+add x (Counter i _) = A.incrCounter_ x i

@@ -1,6 +1,6 @@-{-# LANGUAGE BangPatterns        #-}+{-# LANGUAGE BangPatterns #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeOperators       #-}+{-# LANGUAGE TypeOperators #-}  -- | -- Module      : Instrument.Measurement@@ -8,25 +8,25 @@ --               (c) 2012, Ozgun Ataman -- -- License     : BSD-style- module Instrument.Measurement-    (-      getTime-    , time-    , time_-    , timeEx-    ) where+  ( 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)-------------------------------------------------------------------------------- +import Control.Exception (SomeException)+import Control.Exception.Safe (MonadCatch, tryAny)+import Control.Monad.IO.Class+import Data.Time.Clock.POSIX (getPOSIXTime) +-------------------------------------------------------------------------------+ -- | Measure how long action took, in seconds along with its result-time :: MonadIO m =>  m a -> m (Double, a)+time :: MonadIO m => m a -> m (Double, a) time act = do   start <- liftIO getTime   !result <- act@@ -40,8 +40,7 @@  -- | Just measure how long action takes, discard its result time_ :: MonadIO m => m a -> m Double-time_  = fmap fst . time-+time_ = fmap fst . time  ------------------------------------------------------------------------------- getTime :: IO Double

@@ -1,4 +1,7 @@ -----------------------------------------------------------------------------++----------------------------------------------------------------------------+ -- | -- Module      :  Instrument.Sampler -- Copyright   :  Soostone Inc@@ -10,40 +13,39 @@ -- 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+  ( 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--------------------------------------------------------------------------------+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-     }+-- | '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 :: 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+    (a', b) <- io a `onException` putMVar m a     putMVar m $! a'     return b @@ -55,15 +57,13 @@     a' <- io a `onException` putMVar m a     putMVar m $! a' - ------------------------------------------------------------------------------- newBuffer :: Int -> IO (Buffer a) newBuffer lim = do-  pos  <- newMVar 0+  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.@@ -71,45 +71,35 @@ writeBuffer (B size contents wposMV) x = modifyMVar_mask_ wposMV $   \wpos ->     case wpos >= size of-      True -> return wpos       -- buffer full, don't do anything+      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)--+  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 }-+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 ()

@@ -1,152 +1,214 @@-{-# LANGUAGE CPP                        #-}-{-# LANGUAGE DeriveGeneric              #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings          #-}-{-# LANGUAGE RecordWildCards            #-}-{-# LANGUAGE TemplateHaskell            #-}-{-# LANGUAGE TypeFamilies               #-}+{-# 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+  ( createInstrumentPool,+    Samplers,+    Counters,+    Instrument (..),+    InstrumentConfig (..),+    SubmissionPacket (..),+    MetricName (..),+    DimensionName (..),+    DimensionValue (..),+    Dimensions,+    Payload (..),+    Aggregated (..),+    AggPayload (..),+    Stats (..),+    hostDimension,+    HostDimensionPolicy (..),+    Quantile (..),+  )+where  --------------------------------------------------------------------------------import           Control.Applicative   as A+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 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 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+import GHC.Generics -------------------------------------------------------------------------------+import qualified Instrument.Counter as C+import qualified Instrument.Sampler as S+import Network.HostName +-------------------------------------------------------------------------------  ------------------------------------------------------------------------------- createInstrumentPool :: ConnectInfo -> IO Connection-createInstrumentPool ci = do-  c <- connect ci {-         connectMaxIdleTime = 15-       , connectMaxConnections = 1 }-  return c+createInstrumentPool ci =+    connect+      ci+        { connectMaxIdleTime = 15,+          connectMaxConnections = 1+        } +------------------------------------------------------------------------------- +newtype DimensionName = DimensionName+  { dimensionName :: Text+  }+  deriving (Eq, Ord, Show, Generic, Serialize, IsString)++$(SC.deriveSafeCopy 0 'SC.base ''DimensionName)++-- | Convention for the dimension of the hostname. Used in the client+-- to inject hostname into the parameters map+hostDimension :: DimensionName+hostDimension = "host"++newtype DimensionValue = DimensionValue+  { dimensionValue :: Text+  }+  deriving (Eq, Ord, Show, Generic, Serialize, IsString)++$(SC.deriveSafeCopy 0 'SC.base ''DimensionValue)++newtype MetricName = MetricName+  { metricName :: String+  }+  deriving (Eq, Show, Generic, Ord, IsString, Serialize)++$(SC.deriveSafeCopy 0 'SC.base ''MetricName)++-------------------------------------------------------------------------------+ -- 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 +data Payload_v0+  = Samples_v0 {unSamples_v0 :: [Double]}+  | Counter_v0 {unCounter_v0 :: Int}+  deriving (Eq, Show, Generic) -newtype MetricName = MetricName {-      metricName :: String-    } deriving (Eq,Show,Generic,Ord,IsString,Serialize)+instance Serialize Payload_v0 +data Payload+  = Samples {unSamples :: [Double]}+  | Counter {unCounter :: Integer}+  deriving (Eq, Show, Generic) -data Instrument = I {-      hostName :: HostName-    , samplers :: !(IORef Samplers)-    , counters :: !(IORef Counters)-    , redis    :: Connection-    }+instance Serialize Payload -data InstrumentConfig = ICfg {-      redisQueueBound :: Maybe Integer-    }+$(SC.deriveSafeCopy 0 'SC.base ''Payload_v0)+$(SC.deriveSafeCopy 1 'SC.extension ''Payload) +instance SC.Migrate Payload where+  type MigrateFrom Payload = Payload_v0+  migrate (Samples_v0 n) = Samples n+  migrate (Counter_v0 n) = Counter $ fromIntegral n++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)-+data SubmissionPacket_v0 = SP_v0+  { -- | Timing of this submission+    spTimeStamp_v0 :: !Double,+    -- | Who sent it+    spHostName_v0 :: !HostName,+    -- | Metric name+    spName_v0 :: String,+    -- | Collected values+    spPayload_v0 :: Payload+  }+  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+data SubmissionPacket_v1 = SP_v1+  { -- | Timing of this submission+    spTimeStamp_v1 :: !Double,+    -- | Metric name+    spName_v1 :: !MetricName,+    -- | Collected values+    spPayload_v1 :: !Payload_v0,+    -- | 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)+    spDimensions_v1 :: !Dimensions+  }+  deriving (Eq, Show, Generic) +instance Serialize SubmissionPacket_v1 +data SubmissionPacket = SP+  { -- | Timing of this submission+    spTimeStamp :: !Double,+    -- | Metric name+    spName :: !MetricName,+    -- | Collected values+    spPayload :: !Payload,+    -- | 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"+    spDimensions :: !Dimensions+  }+  deriving (Eq, Show, Generic)+ instance Serialize SubmissionPacket where-  get = (to <$> gGet) <|> (upgradeSP0 <$> Ser.get)+  get = (to <$> gGet)  <|> (upgradeSP1 <$> Ser.get) <|> (upgradeSP0 <$> Ser.get) +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))+    } +upgradeSP1 :: SubmissionPacket_v1 -> SubmissionPacket+upgradeSP1 SP_v1 {..} =+  SP+    { spTimeStamp = spTimeStamp_v1,+      spName = spName_v1,+      spPayload = case spPayload_v1 of+        Samples_v0 n -> Samples n+        Counter_v0 n -> Counter $ fromIntegral n,+      spDimensions = spDimensions_v1+    }++$(SC.deriveSafeCopy 0 'SC.base ''SubmissionPacket_v0)+$(SC.deriveSafeCopy 1 'SC.extension ''SubmissionPacket)+ 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@@ -154,164 +216,172 @@ -- 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)-+data HostDimensionPolicy+  = AddHostDimension+  | DoNotAddHostDimension+  deriving (Show, Eq)  --------------------------------------------------------------------------------newtype DimensionName = DimensionName {-    dimensionName :: Text-  } deriving (Eq,Ord,Show,Generic,Serialize,IsString)+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 ---------------------------------------------------------------------------------newtype DimensionValue = DimensionValue {-    dimensionValue :: Text-  } deriving (Eq,Ord,Show,Generic,Serialize,IsString)+instance Serialize Stats +$(SC.deriveSafeCopy 0 'SC.base ''Stats)  --------------------------------------------------------------------------------data Payload-    = Samples { unSamples :: [Double] }-    | Counter { unCounter :: Int }+-- | Resulting payload for metrics aggregation+data AggPayload_v0+  = AggStats_v0 Stats+  | AggCount_v0 Int   deriving (Eq, Show, Generic) -instance Serialize Payload-+instance Serialize AggPayload_v0 ---------------------------------------------------------------------------------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)+data AggPayload+  = AggStats Stats+  | AggCount Integer+  deriving (Eq, Show, Generic) +instance Serialize AggPayload -instance Serialize Aggregated_v0+instance Default AggPayload where+  def = AggStats def +$(SC.deriveSafeCopy 0 'SC.base ''AggPayload_v0)+$(SC.deriveSafeCopy 1 'SC.extension ''AggPayload) -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 AggPayload where+  type MigrateFrom AggPayload = AggPayload_v0+  migrate (AggStats_v0 n) = AggStats n+  migrate (AggCount_v0 n) = AggCount $ fromIntegral n +-------------------------------------------------------------------------------+data Aggregated = Aggregated+  { -- | Timestamp for this aggregation+    aggTS :: Double,+    -- | Name of the metric+    aggName :: MetricName,+    -- | Calculated stats for the metric+    aggPayload :: AggPayload,+    aggDimensions :: Dimensions+  }+  deriving (Eq, Show, Generic) -instance SC.Migrate Aggregated_v1 where-  type MigrateFrom Aggregated_v1 = Aggregated_v0-  migrate = upgradeAggregated_v0+instance Serialize Aggregated where+  get = (to <$> gGet) <|> (upgradeAggregated_v2 <$> Ser.get) <|> (upgradeAggregated_v2 . upgradeAggregated_v1 <$> Ser.get) +instance Default Aggregated where+  def = Aggregated 0 "" def mempty -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)+data Aggregated_v2 = Aggregated_v2+  { -- | Timestamp for this aggregation+    aggTS_v2 :: Double,+    -- | Name of the metric+    aggName_v2 :: MetricName,+    -- | Calculated stats for the metric+    aggPayload_v2 :: AggPayload_v0,+    aggDimensions_v2 :: Dimensions   }--instance Serialize Aggregated_v1 where-  get = (to <$> gGet) <|> (upgradeAggregated_v0 <$> Ser.get)-+  deriving (Eq, Show, Generic) -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)+instance Serialize Aggregated_v2 +upgradeAggregated_v2 :: Aggregated_v2 -> Aggregated+upgradeAggregated_v2 a =+  Aggregated+    { aggTS = aggTS_v2 a,+      aggName = aggName_v2 a,+      aggPayload = SC.migrate (aggPayload_v2 a),+      aggDimensions = aggDimensions_v2 a+    } -upgradeAggregated_v1 :: Aggregated_v1 -> Aggregated-upgradeAggregated_v1 a = Aggregated-  { aggTS = aggTS_v1 a-  , aggName = aggName_v1 a-  , aggPayload = aggPayload_v1 a-  , aggDimensions = Monoid.mempty+data Aggregated_v1 = Aggregated_v1+  { -- | Timestamp for this aggregation+    aggTS_v1 :: Double,+    -- | Name of the metric+    aggName_v1 :: MetricName,+    -- | The aggregation level/group for this stat+    aggGroup_v1 :: B.ByteString,+    -- | Calculated stats for the metric+    aggPayload_v1 :: AggPayload_v0   }-+  deriving (Eq, Show, Generic) -instance SC.Migrate Aggregated where-  type MigrateFrom Aggregated = Aggregated_v1-  migrate = upgradeAggregated_v1+upgradeAggregated_v1 :: Aggregated_v1 -> Aggregated_v2+upgradeAggregated_v1 a =+  Aggregated_v2+    { aggTS_v2 = aggTS_v1 a,+      aggName_v2 = aggName_v1 a,+      aggPayload_v2 = aggPayload_v1 a,+      aggDimensions_v2 = Monoid.mempty+    } +data Aggregated_v0 = Aggregated_v0+  { -- | Timestamp for this aggregation+    aggTS_v0 :: Double,+    -- | Name of the metric+    aggName_v0 :: String,+    -- | The aggregation level/group for this stat+    aggGroup_v0 :: B.ByteString,+    -- | Calculated stats for the metric+    aggPayload_v0 :: AggPayload_v0+  }+  deriving (Eq, Show, Generic) -instance Serialize Aggregated where-  get = (to <$> gGet) <|> (upgradeAggregated_v1 <$> Ser.get)+instance Serialize Aggregated_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+    } --- | Resulting payload for metrics aggregation-data AggPayload-    = AggStats Stats-    | AggCount Int-    deriving (Eq,Show, Generic)+instance Serialize Aggregated_v1 where+  get = (to <$> gGet) <|> (upgradeAggregated_v0 <$> Ser.get) -instance Serialize AggPayload+$(SC.deriveSafeCopy 0 'SC.base ''Aggregated_v0) +instance SC.Migrate Aggregated_v1 where+  type MigrateFrom Aggregated_v1 = Aggregated_v0+  migrate = upgradeAggregated_v0 +$(SC.deriveSafeCopy 1 'SC.extension ''Aggregated_v1) +instance SC.Migrate Aggregated_v2 where+  type MigrateFrom Aggregated_v2 = Aggregated_v1+  migrate = upgradeAggregated_v1 -instance Default AggPayload where-    def = AggStats def+$(SC.deriveSafeCopy 2 'SC.extension ''Aggregated_v2) -instance Default Aggregated where-    def = Aggregated 0 "" def mempty+instance SC.Migrate Aggregated where+  type MigrateFrom Aggregated = Aggregated_v2+  migrate = upgradeAggregated_v2 +$(SC.deriveSafeCopy 3 'SC.extension ''Aggregated)  --------------------------------------------------------------------------------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)+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)

@@ -1,60 +1,58 @@+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE NoMonomorphismRestriction #-}-{-# LANGUAGE OverloadedStrings         #-}  module Instrument.Utils-    ( formatDecimal-    , formatInt-    , showT-    , showBS-    , collect-    , noDots-    , encodeCompress-    , decodeCompress-    , indefinitely-    , seconds-    , milliseconds-    , for-    ) where-+  ( 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---------------------------------------------------------------------------------+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 ::+  (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-+  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@@ -66,38 +64,36 @@ 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 ::+  RealFloat a =>+  -- | Digits after the point+  Int ->+  -- | Add thousands sep?+  Bool ->+  -- | Number+  a ->+  Text formatDecimal n th i =-    let res = T.pack . showFFloat (Just n) i $ ""-    in if th then addThousands res else res--+  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-+  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@@ -107,8 +103,8 @@   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@@ -118,7 +114,6 @@   where     delayed = (>> threadDelay n) - ------------------------------------------------------------------------------- logAndBackoff :: String -> IO () -> IO () logAndBackoff ctx = recovering policy [h] . const@@ -130,18 +125,17 @@       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

@@ -1,81 +1,88 @@-{-# LANGUAGE BangPatterns        #-}-{-# LANGUAGE FlexibleContexts    #-}-{-# LANGUAGE OverloadedStrings   #-}-{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}+ module Instrument.Worker-    ( initWorkerCSV-    , initWorkerCSV'-    , initWorkerGraphite-    , initWorkerGraphite'-    , work-    , initWorker-    , AggProcess(..)+  ( initWorkerCSV,+    initWorkerCSV',+    initWorkerGraphite,+    initWorkerGraphite',+    work,+    initWorker,+    AggProcess (..),+     -- * Configuring agg processes-    , AggProcessConfig(..)-    , standardQuantiles-    , noQuantiles-    , quantileMap-    , defAggProcessConfig+    AggProcessConfig (..),+    standardQuantiles,+    noQuantiles,+    quantileMap,+    defAggProcessConfig,+     -- * Exported for testing-    , expandDims-    ) where+    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 Control.Error+import Control.Monad+import Control.Monad.IO.Class+import qualified Data.ByteString.Char8 as B+import Data.CSV.Conduit+import Data.Conduit (runConduit, (.|))+import qualified Data.Conduit.List as CL+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           Instrument.Client      (packetsKey, stripTimerPrefix,-                                         timerMetricName)+import Instrument.Client+  ( packetsKey,+    stripTimerPrefix,+    timerMetricName,+  ) import qualified Instrument.Measurement as TM-import           Instrument.Types-import           Instrument.Utils---------------------------------------------------------------------------------+import Instrument.Types+import Instrument.Utils+import Network.Socket as N+import qualified Statistics.Quantile as Q+import Statistics.Sample+import System.IO+import System.Posix +-------------------------------------------------------------------------------  -------------------------------------------------------------------------------+ -- | 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 ::+  ConnectInfo ->+  -- | Target file name+  FilePath ->+  -- | Aggregation period / flush interval in seconds+  Int ->+  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' ::+  -- | Target file name+  FilePath ->+  AggProcessConfig ->+  IO AggProcess initWorkerCSV' fp cfg = do   !res <- fileExist fp   !h <- openFile fp AppendMode@@ -84,34 +91,34 @@     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 ::+  -- | Redis connection+  ConnectInfo ->+  -- | Aggregation period / flush interval in seconds+  Int ->+  -- | Graphite host+  HostName ->+  -- | Graphite port+  Int ->+  AggProcessConfig ->+  IO () initWorkerGraphite conn n server port cfg =-    initWorker "Graphite Worker" conn n =<< initWorkerGraphite' 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' ::+  -- | Graphite host+  HostName ->+  -- | Graphite port+  Int ->+  AggProcessConfig ->+  IO AggProcess initWorkerGraphite' server port cfg = do   addr <- resolve server (fromIntegral port)   sock <- open addr@@ -122,79 +129,82 @@     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))+      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)+      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+  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 }+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)+  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+        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 ::+  -- | Flush interval - determines resolution+  Int ->+  -- | What to do with aggregation results+  AggProcess ->+  -- | Redis buffer for this metric+  B.ByteString ->+  Redis () processSampler n (AggProcess cfg f) k = do   packets <- popLAll k   case packets of@@ -206,13 +216,15 @@           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+            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@@ -221,8 +233,8 @@   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@@ -241,10 +253,11 @@ -- , {d2=>d2v2} => p2 -- , {} => p1 + p2 -- }-expandDims-  :: forall packets. (Monoid packets, Eq packets)-  => M.Map Dimensions packets-  -> M.Map Dimensions packets+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@@ -255,191 +268,191 @@     mkIsolatedMap :: (DimensionName, DimensionValue) -> M.Map Dimensions packets     mkIsolatedMap dPair =       let matches = snd <$> filter ((== dPair) . fst) mFlat-      in if matches == mempty+       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]+    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 ()+  { 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'.+  { -- | What quantiles should we calculate for any given metric, if+    -- any? We offer some common patterns for this in 'quantileMap',+    -- 'standardQuantiles', and 'noQuantiles'.+    metricQuantiles :: MetricName -> Set.Set Quantile   } - instance Semigroup AggProcessConfig where   AggProcessConfig f1 <> AggProcessConfig f2 =     let f3 = f1 <> f2-    in AggProcessConfig f3-+     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]-+  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.Map MetricName (Set.Set Quantile) ->+  -- | What to return on miss+  Set.Set Quantile ->+  (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-+   in liftIO $ T.hPutStrLn h d  typePrefix :: AggPayload -> T.Text-typePrefix AggStats{} = "samples"-typePrefix AggCount{} = "counts"-+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-+  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-+    _ -> (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-+  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+aggToCSV agg@Aggregated {..} = els <> defFields <> dimFields   where     els :: MapRow T.Text-    els = M.fromList $-            ("metric", T.pack (metricName aggName)) :-            ("timestamp", ts) :-            fields+    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]-+    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)]+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+    mkQ (k, v) = (T.concat ["percentile_", showT k], formatDecimal 6 False v)+    ts = formatInt aggTS

@@ -1,24 +1,24 @@ {-# OPTIONS_GHC -fno-warn-orphans #-}-module Instrument.Tests.Arbitrary-    (-    ) where +module Instrument.Tests.Arbitrary+  (+  )+where  --------------------------------------------------------------------------------import           Test.QuickCheck-import           Test.QuickCheck.Instances ()---------------------------------------------------------------------------------import           Instrument.Types+ -------------------------------------------------------------------------------+import Instrument.Types+import Test.QuickCheck+import Test.QuickCheck.Instances () +-------------------------------------------------------------------------------  instance Arbitrary DimensionName where   arbitrary = DimensionName <$> arbitrary - instance Arbitrary DimensionValue where   arbitrary = DimensionValue <$> arbitrary-  instance Arbitrary MetricName where   arbitrary = MetricName <$> arbitrary

@@ -1,65 +1,70 @@ {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards   #-}+{-# LANGUAGE RecordWildCards #-}+ module Instrument.Tests.Client-    ( tests-    ) where+  ( 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 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           Instrument.Client-import           Instrument.Tests.Arbitrary ()-import           Instrument.Types-import           Instrument.Worker+import Instrument.Client+import Instrument.Tests.Arbitrary ()+import Instrument.Types+import Instrument.Worker+import System.Timeout+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+ -------------------------------------------------------------------------------  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+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)-+  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+data TimeException+  = TimeException   deriving (Show)  instance Exception TimeException@@ -97,35 +102,35 @@    sleepFlush -  let collectAggs = work conn 1 (AggProcess defAggProcessConfig (\agg -> liftIO (atomically (modifyTVar' aggsRef (agg:)))))+  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+    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+                (AggStats stats) -> scount stats+                (AggCount n) -> fromIntegral 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")+              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-+        length (filter ((== MetricName "time.instrument-test-time-error") . aggName) aggs) @?= 0  ------------------------------------------------------------------------------- multi_key_test :: IO Connection -> IO ()@@ -136,12 +141,13 @@   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:)))))+  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+    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!"@@ -151,51 +157,48 @@               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 (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"-  ]-+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-+    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 }-+icfg = def {redisQueueBound = Just 2}  key :: MetricName key = MetricName "instrument-test"-  dims :: Dimensions dims = M.fromList [(DimensionName "server", DimensionValue "app1")]

@@ -0,0 +1,59 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}++module Instrument.Tests.Counter+  ( tests,+  )+where++import Instrument.Counter+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+import Control.Monad++tests :: TestTree+tests =+  testGroup+    "Instrument.Counter"+    [ testCase "is zero after initialized" $ do+        c <- newCounter+        val <- readCounter c+        assertEqual "should be zero after initialized" 0 val,++      testProperty "registers arbitrary number of hits: increment" $ \(NonNegative (n :: Integer)) -> ioProperty $ do+        c <- newCounter+        forM_ [1..n] (\_ -> increment c)+        val <- readCounter c+        pure $ fromIntegral n === val,++      testProperty "registers arbitrary number of hits: add" $ \(NonNegative (n :: Integer)) -> ioProperty $ do+        c <- newCounter+        add (fromIntegral n) c+        val <- readCounter c+        pure $ fromIntegral n === val,++      testProperty "reset brings back to zero" $ \(NonNegative (n :: Integer)) -> ioProperty $ do+        c <- newCounter+        add (fromIntegral n) c+        val <- resetCounter c+        assertEqual "should match" (fromIntegral n) val++        val <- readCounter c+        pure $ 0 === val,++      testProperty "registers arbitrary number of hits close to rollover" $ \(NonNegative (n :: Integer)) -> ioProperty $ do+        c <- newCounter+        let offset = maxBound - 5+        add offset c+        val <- resetCounter c+        assertEqual "offset should match" (fromIntegral offset) val++        val <- readCounter c+        assertEqual "should be zero after reset" 0 val++        forM_ [1..n] (\_ -> increment c)+        val <- readCounter c+        pure $ fromIntegral n === val+    ]

@@ -1,33 +1,37 @@ {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell   #-}-module Instrument.Tests.Types-    ( tests-    ) where+{-# 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 qualified Data.Map as M --------------------------------------------------------------------------------import           Instrument.Types+import Instrument.Types+import Path+import Test.HUnit.SafeCopy+import Test.Tasty+import Test.Tasty.HUnit+ -------------------------------------------------------------------------------  --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-  ]+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]

@@ -1,51 +1,55 @@-{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TemplateHaskell     #-}+{-# LANGUAGE TemplateHaskell #-}+ module Instrument.Tests.Utils-    ( tests-    ) where+  ( 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+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 Instrument.Types+import Instrument.Utils+import Path+import qualified Path.IO as PIO+import System.Timeout+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck +-------------------------------------------------------------------------------  tests :: TestTree-tests = testGroup "Instrument.Utils"-    [ encodeDecodeTests-    , withResource spawnWorker killWorker $ testCase "indefinitely retries" . indefinitely_retry+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-  ]+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]@@ -60,23 +64,20 @@       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+  (called, _signal, _) <- setup+  wasCalled <- timeout (milliseconds 100) $ takeMVar called+  assertEqual "retries until success" (Just ()) wasCalled   where - ------------------------------------------------------------------------------- spawnWorker :: IO (MVar (), IO (), ThreadId) spawnWorker = do@@ -84,21 +85,18 @@   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)+  counter <- newIORef (1 :: Int)+  mv <- newEmptyMVar+  return $ (mv, go counter mv)   where     go counter mv = do       count <- readIORef counter-      modifyIORef' counter (+1)+      modifyIORef' counter (+ 1)       when (count > 2) $ void $ tryPutMVar mv ()-

@@ -1,62 +1,68 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings          #-}-module Instrument.Tests.Worker-    ( tests-    ) where+{-# 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+import qualified Data.Map as M+import Data.Monoid as Monoid -------------------------------------------------------------------------------+import Instrument.Tests.Arbitrary ()+import Instrument.Types+import Instrument.Worker+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck +-------------------------------------------------------------------------------  tests :: TestTree-tests = testGroup "Instrument.Worker"-  [ expandDimsTests-  ]-+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-  ]+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"@@ -67,8 +73,6 @@     p1 = "p1"     p2 :: String     p2 = "p2"--  -- | Fixes the packet type for type inference newtype Dims = Dims (M.Map Dimensions [Char])

@@ -1,18 +1,25 @@ 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+import qualified Instrument.Tests.Counter+import Test.Tasty+ -------------------------------------------------------------------------------  main :: IO ()-main = defaultMain $ testGroup "tests"-    [ Instrument.Tests.Utils.tests-    , Instrument.Tests.Client.tests-    , Instrument.Tests.Types.tests-    , Instrument.Tests.Worker.tests-    ]+main =+  defaultMain $+    testGroup+      "tests"+      [ Instrument.Tests.Utils.tests,+        Instrument.Tests.Client.tests,+        Instrument.Tests.Types.tests,+        Instrument.Tests.Worker.tests,+        Instrument.Tests.Counter.tests+      ]