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