packages feed

instrument-cloudwatch (empty) → 0.2.1.0

raw patch · 6 files changed

+434/−0 lines, 6 filesdep +QuickCheckdep +amazonkadep +amazonka-cloudwatchsetup-changed

Dependencies added: QuickCheck, amazonka, amazonka-cloudwatch, async, base, containers, instrument, instrument-cloudwatch, lens, retry, safe-exceptions, semigroups, stm, stm-chans, tasty, tasty-hunit, tasty-quickcheck, text, time, transformers

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Michael Xavier (c) 2015++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 Michael Xavier nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,3 @@+import Distribution.Simple++main = defaultMain
+ instrument-cloudwatch.cabal view
@@ -0,0 +1,68 @@+name:                instrument-cloudwatch+version:             0.2.1.0+synopsis:            Adds a worker for the instrument package that exports to Amazon CloudWatch+description:         Please see README.md+homepage:            http://github.com/soostone/instrument-cloudwatch#readme+license:             BSD3+license-file:        LICENSE+author:              Michael Xavier+maintainer:          michael.xavier@soostone.com+copyright:           2015 Soostone Inc+category:            Web+build-type:          Simple+cabal-version:       >=1.10++flag lib-Werror+  default: False+  manual: True++library+  hs-source-dirs:      src+  exposed-modules:     Instrument.CloudWatch+  build-depends:       base >= 4.6 && < 5+                     , amazonka-cloudwatch >= 2+                     , instrument >= 0.4.0.0+                     , lens >= 4.7 && <= 5.2 +                     , text+                     , time >= 1.4.2+                     , async >= 2.0.2+                     , stm-chans >= 2.0+                     , stm >= 2.4+                     , transformers+                     , amazonka >= 1.6.1+                     , retry >= 0.7+                     , semigroups >= 0.5+                     , containers+                     , safe-exceptions+  default-language:    Haskell2010++  if flag(lib-Werror)+    ghc-options: -Werror++  ghc-options: -Wall+++test-suite test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Main.hs+  other-modules:       Instrument.Tests.CloudWatch+  build-depends:       base+                     , instrument-cloudwatch+                     , tasty >= 0.10+                     , tasty-hunit >= 0.9+                     , stm+                     , stm-chans+                     , semigroups+                     , tasty-quickcheck >= 0.8.4+                     , QuickCheck+  default-language:    Haskell2010++  if flag(lib-Werror)+    ghc-options: -Werror++  ghc-options: -Wall++source-repository head+  type:     git+  location: https://github.com/soostone/instrument-cloudwatch
+ src/Instrument/CloudWatch.hs view
@@ -0,0 +1,234 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Instrument.CloudWatch+  ( CloudWatchICfg (..),+    mkDefCloudWatchICfg,+    QueueSize,+    queueSize,+    cloudWatchAggProcess,++    -- * Exported for testing+    slurpTBMQueue,+    splitNE,+  )+where++-------------------------------------------------------------------------------+import Control.Applicative as A+import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Concurrent.STM.TBMQueue+import qualified Control.Exception.Safe as EX+import Control.Lens+import Control.Monad+import Control.Monad.IO.Class+import Control.Retry+import qualified Data.Foldable as FT+import Data.List.NonEmpty (NonEmpty (..))+import qualified Data.List.NonEmpty as NE+import qualified Data.Map as M+import Data.Monoid as Monoid+import Data.Semigroup (sconcat)+import qualified Data.Text as T+import Data.Time.Clock+import Data.Time.Clock.POSIX+import Instrument+import Amazonka+import qualified Amazonka.CloudWatch as CW+import qualified Amazonka.CloudWatch.Lens as CW++-------------------------------------------------------------------------------++-- | Construct with @preview queueSize@+newtype QueueSize = QueueSize Int deriving (Show, Eq, Ord)++-------------------------------------------------------------------------------++-- | Construct a queue size. Accepts value > 0+queueSize :: Prism' Int QueueSize+queueSize = prism' f t+  where+    t i+      | i > 0 = Just (QueueSize i)+      | otherwise = Nothing+    f (QueueSize i) = i++-------------------------------------------------------------------------------+data CloudWatchICfg = CloudWatchICfg+  { cwiNamespace :: Text,+    cwiQueueSize :: QueueSize,+    cwiEnv :: Env,+    -- | Note: you should probably limit the quantiles you publish with+    -- this backend. Every quantile you decide to publish for a metric+    -- has to be published as a *separate* metric because of the way+    -- cloudwatch works. So if you use something like+    -- 'standardQuantiles', you're going to see (and pay for) 11 metrics+    -- per metric you publish.+    cwiAggProcessConfig :: AggProcessConfig,+    -- | This hook will be executed on any unexpected exceptions so that you can+    -- log, for example.+    cwiOnError :: EX.SomeException -> IO (),+    -- | Delay this long on error in microseconds. This can be used to avoid log+    -- flooding+    cwiErrorDelay :: Maybe Int+  }++-- | Constructor for CloudWatchICfg. If or when new fields are added to the+-- record, they can be defaulted to avoid unnecessary breakage. Defaults to+-- 10,000 queue size, defAggProcessConfig, no-op on error and no delay on error.+mkDefCloudWatchICfg ::+  -- | Metric namespace+  Text ->+  -- | AWS Environment+  Env ->+  CloudWatchICfg+mkDefCloudWatchICfg ns env =+  CloudWatchICfg+    { cwiNamespace = ns,+      cwiQueueSize = QueueSize 10000,+      cwiEnv = env,+      cwiAggProcessConfig = defAggProcessConfig,+      cwiOnError = const (pure ()),+      cwiErrorDelay = Nothing+    }++-------------------------------------------------------------------------------+cloudWatchAggProcess ::+  CloudWatchICfg ->+  -- | Returns the function to push metrics and a+  -- finalizer. Finalizer blocks until workers are terminated.+  IO (AggProcess, IO ())+cloudWatchAggProcess cfg@CloudWatchICfg {..} = do+  q <- newTBMQueueIO (review queueSize cwiQueueSize)+  endSig <- newEmptyMVar+  worker <- async (startWorker cfg q)++  _ <- async $ do+    takeMVar endSig+    atomically $ closeTBMQueue q+    _ <- waitCatch worker+    putMVar endSig ()++  let writer agg = liftIO (atomically (void (tryWriteTBMQueue q agg)))++  let finalizer = putMVar endSig () >> takeMVar endSig+  return (AggProcess cwiAggProcessConfig writer, finalizer)++-------------------------------------------------------------------------------+startWorker :: CloudWatchICfg -> TBMQueue Aggregated -> IO ()+startWorker CloudWatchICfg {..} q = go+  where+    go = do+      maggs <- atomically (slurpTBMQueue q)+      case maggs of+        Just rawAggs -> do+          let datums = sconcat (toDatum A.<$> rawAggs)+          FT.forM_ (splitNE maxDatums datums) $ \datumPage -> do+            let pmd = CW.newPutMetricData cwiNamespace & CW.putMetricData_metricData .~ FT.toList datumPage+            res <- EX.tryAny (runResourceT (awsRetry (send cwiEnv pmd)))+            case res of+              Left e -> do+                void (EX.tryAny (cwiOnError e))+                maybe (pure ()) threadDelay cwiErrorDelay+              Right _ -> pure ()+          go+        Nothing -> return ()+    maxDatums = 20++-------------------------------------------------------------------------------+splitNE :: Int -> NonEmpty a -> NonEmpty (NonEmpty a)+splitNE n xs+  | n > 0 = NE.reverse (unsafeNE (unsafeNE <$> go [] (NE.toList xs)))+  | otherwise = xs :| []+  where+    go acc [] = acc+    go acc remaining =+      let (toAdd, remaining') = splitAt n remaining+       in go (toAdd : acc) remaining'+    unsafeNE (a : as) = a :| as+    unsafeNE _ = error "Impossible empty list passed to unsafeNE"++-------------------------------------------------------------------------------++-- | Expands the aggregated stats into datums. In most cases, this+-- will result in 1 datum. If the payload is an 'AggStats' and+-- contains quantiles, those will be emitted as individual metrics+-- with the quantile appended, e.g. metricName.p90+toDatum :: Aggregated -> NonEmpty CW.MetricDatum+toDatum a =+  baseDatum :| quantileDatums+  where+    baseDatum =+      mkDatum baseMetricName $ case aggPayload a of+        AggStats stats -> Right (toSS stats)+        AggCount n -> Left (fromIntegral n)+    mkDatum name dValOrStats =+      let base =+            CW.newMetricDatum (T.pack name)+              & CW.metricDatum_timestamp ?~ ts+              & CW.metricDatum_dimensions ?~ dims+       in -- Value and stats are mutually exclusive+          case dValOrStats of+            Left dVal -> base & CW.metricDatum_value ?~ dVal+            Right dStats -> base & CW.metricDatum_statisticValues ?~ dStats+    quantileDatums = uncurry mkQuantileDatum <$> quantiles+    mkQuantileDatum :: Int -> Double -> CW.MetricDatum+    mkQuantileDatum quantile val =+      mkDatum (baseMetricName Monoid.<> ".p" <> show quantile) (Left val)+    quantiles = case aggPayload a of+      AggStats stats -> M.toList (squantiles stats)+      AggCount _ -> []+    baseMetricName = (metricName (aggName a))+    ts = aggTS a ^. timeDouble+    dims = uncurry mkDim <$> take maxDimensions (M.toList (aggDimensions a))+    mkDim (DimensionName dn) (DimensionValue dv) = CW.newDimension dn dv+    maxDimensions = 10++-------------------------------------------------------------------------------+timeDouble :: Iso' Double UTCTime+timeDouble = iso toT fromT+  where+    toT :: Double -> UTCTime+    toT = posixSecondsToUTCTime . realToFrac+    fromT :: UTCTime -> Double+    fromT = realToFrac . utcTimeToPOSIXSeconds++-------------------------------------------------------------------------------+toSS :: Stats -> CW.StatisticSet+toSS Stats {..} = CW.newStatisticSet (fromIntegral scount) ssum smin smax++-------------------------------------------------------------------------------++-- | Nothing when closed and empty, retries when just empty+slurpTBMQueue :: TBMQueue a -> STM (Maybe (NonEmpty a))+slurpTBMQueue q = do+  mh <- readTBMQueue q+  case mh of+    Just h -> Just <$> go (h :| [])+    Nothing -> return Nothing+  where+    go acc = do+      ma <- tryReadTBMQueue q+      case ma of+        Just (Just a) -> go (NE.cons a acc)+        _ -> return acc++-------------------------------------------------------------------------------+awsRetry :: (MonadIO m, EX.MonadMask m) => m a -> m a+awsRetry = recovering policy [httpRetryH, networkRetryH] . const+  where+    policy = constantDelay 50000 <> limitRetries 5++-------------------------------------------------------------------------------++-- | Which exceptions should we retry?+httpRetryH :: Monad m => a -> EX.Handler m Bool+httpRetryH = const $ EX.Handler $ \(_ :: HttpException) -> return True++-------------------------------------------------------------------------------++-- | 'IOException's should be retried+networkRetryH :: Monad m => a -> EX.Handler m Bool+networkRetryH = const $ EX.Handler $ \(_ :: EX.IOException) -> return True
+ test/Instrument/Tests/CloudWatch.hs view
@@ -0,0 +1,77 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++module Instrument.Tests.CloudWatch+  ( tests,+  )+where++-------------------------------------------------------------------------------+import Control.Concurrent.STM+import Control.Concurrent.STM.TBMQueue+import qualified Data.Foldable as FT+import Data.List.NonEmpty (NonEmpty (..))+import qualified Data.List.NonEmpty as NE+import Data.Semigroup+-------------------------------------------------------------------------------+import Instrument.CloudWatch+import Test.QuickCheck+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck++-------------------------------------------------------------------------------++tests :: TestTree+tests =+  testGroup+    "Instrument.CloudWatch"+    [ slurpTBMQueueTests,+      splitNETests+    ]++-------------------------------------------------------------------------------+slurpTBMQueueTests :: TestTree+slurpTBMQueueTests =+  testGroup+    "slurpTBMQueue"+    [ testCase "returns Nothing immediately on a closed, empty queue" $ do+        q <- mkQueue 2+        atomically (closeTBMQueue q)+        res <- atomically (slurpTBMQueue q)+        res @?= Nothing,+      testCase "Returns the rest of the queue when closed" $ do+        q <- mkQueue 2+        atomically (writeTBMQueue q "one")+        atomically (closeTBMQueue q)+        res <- atomically (slurpTBMQueue q)+        res @?= Just ("one" :| []),+      testCase "Returns the whole queue when its open" $ do+        q <- mkQueue 2+        atomically (writeTBMQueue q "one")+        res <- atomically (slurpTBMQueue q)+        res @?= Just ("one" :| [])+    ]++-------------------------------------------------------------------------------+mkQueue :: Int -> IO (TBMQueue String)+mkQueue = newTBMQueueIO++-------------------------------------------------------------------------------+splitNETests :: TestTree+splitNETests =+  testGroup+    "splitNE"+    [ testProperty "0 or negative count" $ \(NonEmpty nel) n ->+        n <= 0+          ==> let ne = NE.fromList nel :: NonEmpty ()+               in splitNE n ne === ne :| [],+      testProperty "positive count, no items exceed length" $ \(NonEmpty nel) (Positive n) ->+        let ne = NE.fromList nel :: NonEmpty ()+            res = splitNE n ne+         in FT.all ((<= n) . NE.length) res,+      testProperty "positive count, loses no items and preserves order" $ \(NonEmpty nel) (Positive n) ->+        let ne = NE.fromList nel :: NonEmpty ()+            res = splitNE n ne+         in sconcat res === ne+    ]
+ test/Main.hs view
@@ -0,0 +1,22 @@+module Main+  ( main,+  )+where++-------------------------------------------------------------------------------++-------------------------------------------------------------------------------+import qualified Instrument.Tests.CloudWatch+import Test.Tasty++-------------------------------------------------------------------------------++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests =+  testGroup+    "instrument-cloudwatch"+    [ Instrument.Tests.CloudWatch.tests+    ]