diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.5.0.2 - 2026-11-07 + +- Lots of documentation. + ## 0.5.0.1 - 2025-05-09 - Bump LTS Haskell. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,6 +1,424 @@ # statsd-rupp -StatsD client library for Haskell. +[![Hackage](https://img.shields.io/hackage/v/statsd-rupp.svg)](https://hackage.haskell.org/package/statsd-rupp) +[![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](https://github.com/jprupp/statsd-rupp/blob/master/LICENSE) -It can report both individual samples like simple StatsD clients, and -statistical aggregations like StatsD servers. +statsd-rupp is a [StatsD](https://github.com/statsd/statsd) client for Haskell. + +It works in two ways, which you can use separately or together: + +- **As a client**: it sends each sample to the StatsD server as it happens, with + optional sampling. +- **As an aggregator**: it keeps samples in memory and every flush interval sends + statistics, such as counts, rates, means, medians and percentiles, like a + StatsD server does. This lets you feed backends that speak the StatsD line + protocol but do not aggregate, such as some Graphite and InfluxDB relays. + +Metrics go over UDP and send errors are ignored, so a missing StatsD server +never breaks your program. + +- [API documentation](https://hackage.haskell.org/package/statsd-rupp) +- [Changelog](https://github.com/jprupp/statsd-rupp/blob/master/CHANGELOG.md) +- [Report a bug](https://github.com/jprupp/statsd-rupp/issues) + +## Contents + +- [Why statsd-rupp](#why-statsd-rupp) +- [Design principles](#design-principles) +- [Features](#features) +- [Installation](#installation) +- [Quick start](#quick-start) +- [Tutorial: instrumenting a web service](#tutorial-instrumenting-a-web-service) +- [Concepts](#concepts) +- [Usage guide](#usage-guide) +- [Configuration](#configuration) +- [Metric names](#metric-names) +- [Module overview](#module-overview) +- [Development](#development) +- [Related projects](#related-projects) +- [License](#license) + +## Why statsd-rupp + +Metrics are how you find out what a program in production is doing: how many +requests it handles, how slow they are, how close it is to falling over. +StatsD is one of the simplest ways to get them out. Each sample is one line of +text sent over UDP to a collector, so instrumenting code costs almost nothing +and needs no connection management. + +The classic setup has a StatsD daemon next to the program that aggregates +samples into counts, rates and percentiles before they reach the time series +database. Many modern collectors accept the StatsD protocol but skip that +aggregation and store every line they receive. A busy service then floods them +with one line per request, and percentiles of response times are nowhere to be +found. + +statsd-rupp covers both setups. It can send samples like any StatsD client, +and it can do the daemon's job inside the process: keep every reading in +memory, and once per interval send a fixed set of statistics. The traffic stays +constant however busy the program gets. It was written to monitor +[Haskoin Store](https://github.com/jprupp/haskoin-store). + +## Design principles + +- **Metrics never hurt the program.** Sending is fire-and-forget UDP and errors + are ignored. A collector that is down costs lost samples, never failed + requests. +- **Mistakes fail early.** Bad keys, duplicate keys and bad configuration are + refused when you create the metric or the client, so a typo shows up at + start-up, not as a missing graph weeks later. +- **Register once, update cheaply.** Creating a metric returns a handle; + updating it is one short STM transaction and at most one UDP packet. +- **Sampling saves traffic, not accuracy.** Sampling thins out the samples you + send. Aggregated statistics always count every reading. +- **Familiar names.** Aggregated metrics follow the naming of the StatsD daemon + (`stats.counters.x.count`, `stats.timers.x.mean_90`), so existing dashboards + and alerts keep working. + +## Features + +| Area | What you get | +| ------------ | -------------------------------------------------------------------------------- | +| Metric types | Counters, gauges, timings and sets | +| Samples | Each reading sent as it happens, with per-metric sampling rates | +| Aggregation | Counts and rates; gauge values; distinct set sizes; timing count, rate, mean, median, standard deviation, minimum, maximum, sum, sum of squares and percentiles | +| Naming | Configurable namespace and prefixes; key validation | +| Transport | UDP to any host and port, with an optional newline after each line | +| Monads | Works in any `MonadIO`; `withStats` needs `MonadUnliftIO` | + +### What is not included + +- No tags (the DogStatsD extension) and no histogram or distribution types. +- No TCP transport and no batching of several lines in one packet. +- No server: this library only sends. + +## Installation + +statsd-rupp is on [Hackage](https://hackage.haskell.org/package/statsd-rupp) and +in Stackage snapshots. It has no C dependencies. It needs GHC 9.2 or later, +because it uses `OverloadedRecordDot` and `NoFieldSelectors`. + +### Step by step + +1. Install [Stack](https://docs.haskellstack.org/) or + [Cabal](https://www.haskell.org/cabal/). +2. Add `statsd-rupp` to the `build-depends` of your package (see below). +3. Start something that listens for StatsD on UDP port 8125. To see the raw + lines while developing, netcat is enough: + + ```sh + nc -u -l 8125 # some versions need: nc -u -l -p 8125 + ``` + +4. Run the [quick start](#quick-start) program with + `statsdServer = "127.0.0.1"`. You should see lines like + `myapp.requests:1|c` straight away, and a burst of `myapp.stats.` lines + every second. + +### With Cabal + +```cabal +build-depends: + base + , statsd-rupp +``` + +### With Stack + +```yaml +# package.yaml +dependencies: + - statsd-rupp +``` + +If your resolver does not include it, add it to `stack.yaml`: + +```yaml +extra-deps: + - statsd-rupp-0.5.0.1 +``` + +## Quick start + +```haskell +import System.Metrics.StatsD + +main :: IO () +main = do + let cfg = defStatConfig {namespace = "myapp", statsdServer = "statsd.local"} + withStats cfg $ \stats -> do + requests <- newStatCounter stats "requests" 1 -- send every sample + latency <- newStatTiming stats "latency" 10 -- send 1 sample in 10 + users <- newStatGauge stats "users" 0 -- start at 0 + visitors <- newStatSet stats "visitors" + + incrementCounter requests 1 + addTiming latency 42 + incrementGauge users 1 + newSetElement visitors "alice" +``` + +With the default configuration this sends `myapp.requests:1|c`, +`myapp.users:+1|g` and `myapp.visitors:alice|s` right away. Every second it +sends `myapp.stats.counters.requests.count`, `myapp.stats.timers.latency.mean` +and the other statistics listed under [Metric names](#metric-names). + +## Tutorial: instrumenting a web service + +This shows how to decide what to measure and how, for a typical request +handler. + +### 1. Decide what you need to know + +For a service, start with the four questions that most incidents come down to: +how much traffic, how many errors, how slow, and how saturated. Each maps to a +metric type: + +| Question | Metric | Type | +| -------------- | ---------------- | ------- | +| How much? | `requests` | counter | +| How many fail? | `errors` | counter | +| How slow? | `latency` | timing | +| How full? | `queue.length` | gauge | + +### 2. Create the metrics at start-up + +Create them once, next to the rest of your configuration, and pass the handles +to the code that needs them: + +```haskell +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE ScopedTypeVariables #-} + +import GHC.Clock (getMonotonicTimeNSec) +import System.Metrics.StatsD +import UnliftIO (SomeException, throwIO, try) + +data Metrics = Metrics + { requests :: StatCounter, + errors :: StatCounter, + latency :: StatTiming, + queue :: StatGauge + } + +newMetrics :: Stats -> IO Metrics +newMetrics s = + Metrics + <$> newStatCounter s "requests" 10 -- busy: send 1 sample in 10 + <*> newStatCounter s "errors" 1 -- rare: send them all + <*> newStatTiming s "latency" 10 + <*> newStatGauge s "queue.length" 0 +``` + +The sampling rates follow the traffic: frequent events can be sampled, rare and +important ones should not be. The aggregated statistics count everything +either way. + +### 3. Record around the work + +```haskell +handle :: Metrics -> Request -> IO Response +handle m req = do + incrementCounter m.requests 1 + start <- getMonotonicTimeNSec + r <- try (serve req) + end <- getMonotonicTimeNSec + addTiming m.latency (fromIntegral ((end - start) `div` 1000000)) + case r of + Left (e :: SomeException) -> incrementCounter m.errors 1 >> throwIO e + Right resp -> return resp +``` + +Use a monotonic clock (`GHC.Clock.getMonotonicTimeNSec`) for durations, so that +clock adjustments do not produce negative or huge timings. + +### 4. Choose the mode + +- If a real StatsD daemon aggregates for you, turn aggregation off with + `reportStats = False` and let it do its job. +- If your collector stores raw lines, turn samples off with + `reportSamples = False` and let the library aggregate. +- Leave both on while you find out which one your setup needs. + +### 5. Read the results + +With aggregation, `stats.timers.latency.mean_95` is the average of the fastest +95% of requests in each interval, `upper_95` is the 95th percentile, and +`count_ps` is requests per second. Alert on `upper_95` and on the ratio of +`stats.counters.errors.count` to `stats.counters.requests.count`. + +## Concepts + +### The client handle + +`withStats` resolves the server address, opens a UDP socket, and passes a +`Stats` handle to your function. If aggregation is on, it also runs a thread +that flushes statistics. Everything stops when the function returns. A `Stats` +can be shared between threads. + +### Registering metrics + +Each metric is registered once with a `newStat*` function, which returns a +handle to update it with. Keys may contain only ASCII letters, digits, `.`, `_` +and `-`. Registering a key twice, or an invalid key, throws an `IOError`. + +### Sampling + +Counters and timings take a sampling rate `n`. With samples turned on, one +reading in every `n` is sent, tagged with the rate `1/n` (`|@0.100000` for +`n = 10`) so that the server +can scale it back up. Use `1` to send every reading and `0` to send none. +Aggregated statistics always include every reading. + +### Flushing + +Every `flushInterval` milliseconds, the aggregated data of every metric is sent +and reset: counters go back to zero, timings and sets are emptied, gauges keep +their value. + +## Usage guide + +### Counters + +```haskell +c <- newStatCounter stats "jobs.done" 1 +incrementCounter c 1 +incrementCounter c 5 +``` + +### Gauges + +```haskell +g <- newStatGauge stats "queue.length" 0 +setGauge g 10 +incrementGauge g 3 +decrementGauge g 1 +``` + +The aggregated value never drops below zero and saturates at `maxBound`. In the +StatsD protocol a negative value means "decrease", so `setGauge g (-1)` is seen +by a server as a decrement. + +### Timings + +```haskell +t <- newStatTiming stats "db.query" 1 +start <- getCurrentTime +runQuery +end <- getCurrentTime +addTiming t (round (1000 * diffUTCTime end start)) +``` + +Timings are whole numbers, normally milliseconds. + +### Sets + +```haskell +s <- newStatSet stats "active_users" +newSetElement s userId +``` + +The aggregated statistic is the number of distinct elements seen in the flush +interval. + +### Samples only, or statistics only + +```haskell +-- a plain StatsD client +defStatConfig {reportStats = False} + +-- an aggregator that sends nothing until it flushes +defStatConfig {reportSamples = False} +``` + +## Configuration + +Start from `defStatConfig` and override the fields you need. + +| Field | Default | Meaning | +| ------------------- | ------------- | ---------------------------------------------------------- | +| `reportStats` | `True` | Aggregate and send statistics every flush interval | +| `reportSamples` | `True` | Send each (sampled) reading as it happens | +| `namespace` | `""` | Prefix for all metric names | +| `prefixStats` | `"stats"` | Prefix for aggregated statistics, after the namespace | +| `prefixCounter` | `"counters"` | Prefix for counter statistics | +| `prefixTimer` | `"timers"` | Prefix for timing statistics | +| `prefixGauge` | `"gauges"` | Prefix for gauge statistics | +| `prefixSet` | `"sets"` | Prefix for set statistics | +| `statsdServer` | `"127.0.0.1"` | StatsD host name or address | +| `statsdPort` | `8125` | StatsD UDP port | +| `flushInterval` | `1000` | Milliseconds between flushes; must be positive | +| `timingPercentiles` | `[90, 95]` | Percentiles to compute for timings, each from 1 to 99 | +| `appendNewline` | `False` | End each line with a newline, as some servers need | + +`withStats` fails with `error` if a prefix or the namespace is not a valid key, +the flush interval is not positive, a percentile is out of range, or the server +name cannot be resolved. + +## Metric names + +With `namespace = "myapp"` and the default prefixes: + +| What | Names | +| ------------------- | ------------------------------------------------------------- | +| Samples | `myapp.` | +| Counter statistics | `myapp.stats.counters..count`, `.rate` (per second) | +| Gauge statistics | `myapp.stats.gauges.` | +| Set statistics | `myapp.stats.sets..count` | +| Timing statistics | `myapp.stats.timers..` | + +Timing statistics are `count`, `count_ps` (per second), `std`, `mean`, `upper`, +`lower`, `sum`, `sum_squares` and `median`. For each percentile `N`, the same +statistics except `count_ps` and `std` are also sent with an `_N` suffix, +computed over the lowest `N`% of the timings: for example `mean_90` and +`upper_95`. + +Without a namespace, names start at `stats.` and samples at the key. + +## Module overview + +| Module | Contents | +| --------------------------------- | -------------------------------------------------------------- | +| `System.Metrics.StatsD` | The public API: configuration, `withStats`, metric types | +| `System.Metrics.StatsD.Internal` | Aggregation, the line format and its parser; may change between minor versions | + +## Development + +```sh +git clone https://github.com/jprupp/statsd-rupp.git +cd statsd-rupp + +stack build # build the library +stack test # run the tests +stack haddock --no-haddock-deps # build the API documentation +``` + +`cabal build` and `cabal test` work too. + +`package.yaml` is the source of the package description. Change it rather than +`statsd-rupp.cabal`, which [hpack](https://github.com/sol/hpack) generates. + +The tests are in `test/Spec.hs` and use [hspec](https://hspec.github.io/) and +QuickCheck. They check the line format against its parser, the statistics +against simple reference implementations, and what each metric type sends, +through a local socket pair. + +### Contributing + +Pull requests are welcome. + +- Format the code with [Ormolu](https://github.com/tweag/ormolu). +- Add tests for the behaviour you change. +- Describe the change in `CHANGELOG.md`, which follows + [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). Versions follow the + [Haskell Package Versioning Policy](https://pvp.haskell.org/). + +## Related projects + +- [haskoin-store](https://github.com/jprupp/haskoin-store) reports its metrics + with this library. + +## License + +statsd-rupp is released under the +[BSD 3-Clause License](https://github.com/jprupp/statsd-rupp/blob/master/LICENSE). diff --git a/src/System/Metrics/StatsD.hs b/src/System/Metrics/StatsD.hs --- a/src/System/Metrics/StatsD.hs +++ b/src/System/Metrics/StatsD.hs @@ -6,25 +6,125 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE NoFieldSelectors #-} +-- | +-- Module : System.Metrics.StatsD +-- Copyright : (c) 2023 JP Rupp +-- License : BSD3 +-- Maintainer : jprupp@protonmail.ch +-- Stability : experimental +-- Portability : POSIX +-- +-- A client. +-- +-- This library can work in two ways at the same time: +-- +-- * Like a simple StatsD client, sending each sample to the StatsD server as it +-- happens. Enable this with 'reportSamples'. +-- +-- * Like a StatsD server, aggregating samples in memory and sending statistics +-- such as counts, rates, means and percentiles every 'flushInterval' +-- milliseconds. Enable this with 'reportStats'. This mode is useful when +-- sending to a backend that accepts the StatsD protocol but does not +-- aggregate, such as a Graphite or InfluxDB relay. +-- +-- All metrics are sent over UDP. Network errors are silently ignored, so that +-- metrics never break the application. +-- +-- = Purpose +-- +-- Metrics tell you what a running program is doing: how many requests it +-- serves, how long they take, how full its queues are. StatsD is a simple and +-- widely supported way to ship them: one line of text per sample, sent over +-- UDP to a daemon that aggregates and forwards them. +-- +-- The aggregation mode exists because not every collector aggregates. Some +-- accept the StatsD line protocol but store whatever they receive, so a busy +-- service sending one line per request floods them. Aggregating in the process +-- and sending a fixed set of statistics once per interval keeps the traffic +-- constant however busy the program is, and puts percentiles in reach of +-- backends that cannot compute them. +-- +-- = Design +-- +-- * __Metrics must never hurt the program.__ Sending is fire-and-forget over +-- UDP and send errors are ignored. A missing or overloaded StatsD server +-- costs some lost samples, never a failed request. +-- +-- * __Mistakes in naming fail early.__ Invalid keys, duplicate keys and invalid +-- configurations are refused when the metric or client is created, not when +-- the first sample is sent. A typo shows up at start-up instead of as a +-- missing graph weeks later. +-- +-- * __Register first, then update cheaply.__ Each metric is created once and +-- returns a handle. Updating it is an 'STM' update of a map and, when samples +-- are on, one UDP packet, so it is cheap enough for hot paths. +-- +-- * __Sampling saves traffic, not accuracy.__ Sampling only thins out the +-- individual samples sent. The aggregated statistics always count every +-- reading. +-- +-- = Example +-- +-- @ +-- main :: IO () +-- main = +-- 'withStats' 'defStatConfig'{'namespace' = "myapp"} $ \\stats -> do +-- requests <- 'newStatCounter' stats "requests" 1 +-- latency <- 'newStatTiming' stats "latency" 10 +-- users <- 'newStatGauge' stats "users" 0 +-- visitors <- 'newStatSet' stats "visitors" +-- 'incrementCounter' requests 1 +-- 'addTiming' latency 42 +-- 'incrementGauge' users 1 +-- 'newSetElement' visitors "alice" +-- @ +-- +-- = Metric names +-- +-- Metric keys may only contain ASCII letters, digits, @.@, @_@ and @-@. Each key +-- can only be registered once per 'Stats' instance. With the default +-- configuration and a namespace of @myapp@, the following names are sent: +-- +-- [Samples] @myapp.\@ +-- +-- [Counters] @myapp.stats.counters.\.count@ and +-- @myapp.stats.counters.\.rate@ (per second) +-- +-- [Gauges] @myapp.stats.gauges.\@ +-- +-- [Sets] @myapp.stats.sets.\.count@ (distinct elements seen during the +-- interval) +-- +-- [Timings] @myapp.stats.timers.\.\@, where @\@ is one of +-- @count@, @count_ps@, @std@, @mean@, @upper@, @lower@, @sum@, @sum_squares@ +-- and @median@. For each configured percentile @N@, the same statistics except +-- @count_ps@ and @std@ are also sent with an @_N@ suffix, computed over the +-- lowest @N@% of the timings. module System.Metrics.StatsD - ( StatCounter, - StatGauge, - StatTiming, - StatSet, - Stats, + ( -- * Configuration StatConfig (..), + defStatConfig, + -- * Running + Stats, + withStats, + -- * Counters + StatCounter, newStatCounter, - newStatGauge, - newStatTiming, - newStatSet, incrementCounter, + -- * Gauges + StatGauge, + newStatGauge, setGauge, incrementGauge, decrementGauge, + -- * Timings + StatTiming, + newStatTiming, addTiming, + -- * Sets + StatSet, + newStatSet, newSetElement, - withStats, - defStatConfig, ) where @@ -50,8 +150,18 @@ import UnliftIO (MonadIO, MonadUnliftIO, throwIO) import UnliftIO.Async (link, withAsync) +-- | Name of a metric. Only ASCII letters, digits, @.@, @_@ and @-@ are allowed. type Key = String +-- | Default configuration: +-- +-- * Report both aggregated statistics and individual samples. +-- * No namespace. +-- * Prefixes @stats@, @counters@, @timers@, @gauges@ and @sets@. +-- * Send to @127.0.0.1@ port @8125@. +-- * Flush statistics every second. +-- * Compute the 90th and 95th percentiles of timings. +-- * Do not append a newline to each line sent. defStatConfig :: StatConfig defStatConfig = StatConfig @@ -70,6 +180,20 @@ appendNewline = False } +-- | Register a new counter. +-- +-- The last argument is the sampling rate for individual samples sent when +-- 'reportSamples' is on: one in every @n@ increments is sent, tagged with a rate +-- of @1\/n@ so that the StatsD server can scale it back up. Use @1@ to send +-- every increment and @0@ to send none. The aggregated statistics always count +-- every increment. +-- +-- Throws an 'IOError' if the sampling rate is negative, if the key is invalid, +-- or if a metric with the same key already exists. +-- +-- Sampling is per metric because traffic is: a counter incremented on every +-- request needs a low rate, while one incremented on every error can afford to +-- send everything. newStatCounter :: (MonadIO m) => Stats -> Key -> Int -> m StatCounter newStatCounter stats key sampling = do @@ -79,12 +203,21 @@ newMetric stats (C.pack key) (CounterData 0) return $ StatCounter stats (C.pack key) sampling +-- | Register a new gauge with an initial value. +-- +-- Throws an 'IOError' if the key is invalid or a metric with the same key +-- already exists. newStatGauge :: (MonadIO m) => Stats -> Key -> Int -> m StatGauge newStatGauge stats key ini = do newMetric stats (C.pack key) (GaugeData ini) return $ StatGauge stats (C.pack key) +-- | Register a new timing metric. The last argument is the sampling rate for +-- individual samples, with the same meaning as in 'newStatCounter'. +-- +-- Throws an 'IOError' if the sampling rate is negative, if the key is invalid, +-- or if a metric with the same key already exists. newStatTiming :: (MonadIO m) => Stats -> Key -> Int -> m StatTiming newStatTiming stats key sampling = do when (sampling < 0) $ @@ -93,34 +226,67 @@ newMetric stats (C.pack key) (TimingData []) return $ StatTiming stats (C.pack key) sampling +-- | Register a new set, which counts the distinct elements seen during each +-- flush interval. +-- +-- Throws an 'IOError' if the key is invalid or a metric with the same key +-- already exists. newStatSet :: (MonadIO m) => Stats -> Key -> m StatSet newStatSet stats key = do newMetric stats (C.pack key) (SetData HashSet.empty) return $ StatSet stats (C.pack key) +-- | Add a number to a counter. The aggregated count is reset to zero after each +-- flush. incrementCounter :: (MonadIO m) => StatCounter -> Int -> m () incrementCounter StatCounter {..} = processSample stats sampling key . Counter +-- | Set a gauge to a value. Gauges keep their value across flushes. +-- +-- Note that in the StatsD protocol a negative value is interpreted as a +-- decrement, so StatsD servers receiving samples will not set the gauge to a +-- negative value. setGauge :: (MonadIO m) => StatGauge -> Int -> m () setGauge StatGauge {..} i = processSample stats 1 key (Gauge i False) +-- | Add a number to a gauge, which may be negative. The aggregated value never +-- goes below zero and saturates at 'maxBound'. incrementGauge :: (MonadIO m) => StatGauge -> Int -> m () incrementGauge StatGauge {..} i = processSample stats 1 key (Gauge i True) +-- | Subtract a number from a gauge. The aggregated value never goes below zero. decrementGauge :: (MonadIO m) => StatGauge -> Int -> m () decrementGauge x i = incrementGauge x (negate i) +-- | Record a timing, normally a duration in milliseconds. The timings collected +-- during a flush interval are summarised and then discarded. addTiming :: (MonadIO m) => StatTiming -> Int -> m () addTiming StatTiming {..} = processSample stats sampling key . Timing +-- | Record that an element was seen in a set. Elements should follow the same +-- character rules as keys, or StatsD servers may reject the sample. newSetElement :: (MonadIO m) => StatSet -> String -> m () newSetElement StatSet {..} = processSample stats 1 key . Set . C.pack +-- | Connect to the StatsD server and pass a 'Stats' handle to the given function. +-- +-- If 'reportStats' is on, a background thread is started that flushes +-- aggregated statistics every 'flushInterval' milliseconds. It is stopped when +-- the function returns. +-- +-- Calls 'error' if the configuration is invalid: if a prefix or the namespace is +-- not a valid key, if 'flushInterval' is not positive, or if a percentile is not +-- strictly between 0 and 100. Also fails if the server address cannot be +-- resolved. +-- +-- A scope rather than a global client keeps the background flusher's lifetime +-- explicit, and makes it possible to run several independent clients in one +-- program, for example in tests. withStats :: (MonadUnliftIO m) => StatConfig -> (Stats -> m a) -> m a withStats cfg go = do metrics <- newMetrics diff --git a/src/System/Metrics/StatsD/Internal.hs b/src/System/Metrics/StatsD/Internal.hs --- a/src/System/Metrics/StatsD/Internal.hs +++ b/src/System/Metrics/StatsD/Internal.hs @@ -5,6 +5,31 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE NoFieldSelectors #-} +-- | +-- Module : System.Metrics.StatsD.Internal +-- Copyright : (c) 2023 JP Rupp +-- License : BSD3 +-- Maintainer : jprupp@protonmail.ch +-- Stability : experimental +-- Portability : POSIX +-- +-- Implementation details of "System.Metrics.StatsD". This module is exposed for +-- testing and for advanced use. Its API may change between minor versions. +-- +-- Metrics are kept in a 'HashMap' from key to 'Store' inside a 'TVar'. Every +-- reading updates the aggregated 'MetricData' of its metric and increments a +-- per-metric counter, which is used to decide which samples to send when +-- sampling. A background loop ('statsLoop') periodically turns the aggregated +-- data into 'Report's, sends them, and resets the data ('flush'). +-- +-- Keeping every metric in one 'TVar' means a reading is a single short +-- transaction, and a flush can take a consistent picture of all metrics at once +-- and reset them atomically, so no reading is counted twice or lost between +-- intervals. +-- +-- The parser ('parseReport') is not used by the client. It exists so that the +-- tests can check that every line the client formats reads back as the same +-- report. module System.Metrics.StatsD.Internal ( Stats (..), newStats, @@ -87,105 +112,182 @@ stateTVar, ) +-- | Handle to a StatsD client. Create one with +-- 'System.Metrics.StatsD.withStats'. It can be shared between threads. data Stats = Stats - { metrics :: !(TVar Metrics), + { -- | registered metrics and their aggregated data + metrics :: !(TVar Metrics), + -- | UDP socket connected to the StatsD server socket :: !Socket, + -- | settings derived from the 'StatConfig' params :: !StatParams } +-- | Settings derived from a 'StatConfig' by 'newParams', with the prefixes +-- already joined and encoded. data StatParams = StatParams - { pfx :: !ByteString, + { -- | namespace followed by a dot, or empty; prepended to samples + pfx :: !ByteString, + -- | full prefix for counter statistics, ending in a dot pfxCounter :: !ByteString, + -- | full prefix for timing statistics, ending in a dot pfxTimer :: !ByteString, + -- | full prefix for gauge statistics, ending in a dot pfxGauge :: !ByteString, + -- | full prefix for set statistics, ending in a dot pfxSet :: !ByteString, + -- | flush interval in milliseconds flush :: !Int, + -- | whether aggregated statistics are kept and reported stats :: !Bool, + -- | whether individual samples are reported samples :: !Bool, + -- | percentiles to compute for timings, always including 100 percentiles :: ![Int], + -- | whether to append a newline to each line sent newline :: !Bool } +-- | Client configuration. Start from 'System.Metrics.StatsD.defStatConfig' and +-- change the fields you need. data StatConfig = StatConfig - { reportStats :: !Bool, + { -- | aggregate samples and send statistics every 'flushInterval' + reportStats :: !Bool, + -- | send every sample (subject to sampling) as it happens reportSamples :: !Bool, + -- | prefix for every metric name; empty for none namespace :: !String, + -- | prefix for aggregated statistics, after the namespace prefixStats :: !String, + -- | prefix for counter statistics, after 'prefixStats' prefixCounter :: !String, + -- | prefix for timing statistics, after 'prefixStats' prefixTimer :: !String, + -- | prefix for gauge statistics, after 'prefixStats' prefixGauge :: !String, + -- | prefix for set statistics, after 'prefixStats' prefixSet :: !String, + -- | host name or IP address of the StatsD server statsdServer :: !String, + -- | UDP port of the StatsD server statsdPort :: !Int, + -- | milliseconds between flushes of aggregated statistics; must be + -- positive flushInterval :: !Int, + -- | percentiles to compute for timings, each strictly between 0 and 100 timingPercentiles :: ![Int], + -- | append a newline to each line sent, as some servers require appendNewline :: !Bool } deriving (Show, Read, Eq, Ord) +-- | Aggregated data of a metric during the current flush interval. data MetricData - = CounterData !Int - | GaugeData !Int - | TimingData ![Int] - | SetData !(HashSet ByteString) + = -- | sum of increments + CounterData !Int + | -- | current value + GaugeData !Int + | -- | timings recorded, most recent first + TimingData ![Int] + | -- | distinct elements seen + SetData !(HashSet ByteString) +-- | State of a registered metric. data Store = Store - { index :: !Int, + { -- | number of readings so far, used for sampling + index :: !Int, + -- | aggregated data, or 'Nothing' if statistics are not reported dat :: !(Maybe MetricData) } +-- | All registered metrics, by key. type Metrics = HashMap ByteString Store +-- | A single reading, or the value of a report. data Value - = Counter !Int - | Gauge !Int !Bool - | Timing !Int - | Set !ByteString + = -- | counter increment (@|c@) + Counter !Int + | -- | gauge value; if the flag is 'True' it is a relative change and is sent + -- with an explicit sign (@|g@) + Gauge !Int !Bool + | -- | timing in milliseconds (@|ms@) + Timing !Int + | -- | set element (@|s@) + Set !ByteString deriving (Eq, Ord, Show, Read) +-- | A line to send to the StatsD server, in the form +-- @key:value|type[|\@rate]@. data Report = Report - { key :: !ByteString, + { -- | full metric name + key :: !ByteString, + -- | value and type value :: !Value, + -- | sampling rate between 0 and 1; only sent if below 1 rate :: !Double } deriving (Eq, Ord, Show, Read) +-- | A reading together with what is needed to decide whether to send it. data Sample = Sample - { key :: !ByteString, + { -- | metric key, without namespace + key :: !ByteString, + -- | the reading value :: !Value, + -- | send one in this many readings; @0@ to send none sampling :: !Int, + -- | number of readings of this metric so far, including this one index :: !Int } deriving (Eq, Ord, Show, Read) +-- | Handle to a counter. Create with 'System.Metrics.StatsD.newStatCounter'. data StatCounter = StatCounter - { stats :: !Stats, + { -- | client the counter belongs to + stats :: !Stats, + -- | metric key key :: !ByteString, + -- | sampling rate for individual samples sampling :: !Int } +-- | Handle to a gauge. Create with 'System.Metrics.StatsD.newStatGauge'. data StatGauge = StatGauge - { stats :: !Stats, + { -- | client the gauge belongs to + stats :: !Stats, + -- | metric key key :: !ByteString } +-- | Handle to a timing metric. Create with +-- 'System.Metrics.StatsD.newStatTiming'. data StatTiming = StatTiming - { stats :: !Stats, + { -- | client the timing belongs to + stats :: !Stats, + -- | metric key key :: !ByteString, + -- | sampling rate for individual samples sampling :: !Int } +-- | Handle to a set. Create with 'System.Metrics.StatsD.newStatSet'. data StatSet = StatSet - { stats :: !Stats, + { -- | client the set belongs to + stats :: !Stats, + -- | metric key key :: !ByteString } +-- | Add a metric with its initial data to the map. The data is only kept if +-- statistics are reported. addMetric :: StatParams -> ByteString -> MetricData -> Metrics -> Metrics addMetric params key dat = HashMap.insert key $ Store 0 $ if params.stats then Just dat else Nothing +-- | Register a metric. Throws an 'IOError' if the key is invalid or already +-- registered. newMetric :: (MonadIO m) => Stats -> ByteString -> MetricData -> m () newMetric stats key store = do unless (validateKey key) $ do @@ -201,11 +303,20 @@ return False when e $ throwIO $ userError $ "StatsD key exists: " <> C.unpack key +-- | A key is valid if it is not empty and only contains ASCII letters, digits, +-- @.@, @_@ and @-@. validateKey :: ByteString -> Bool validateKey t = not (C.null t) && C.all valid t where valid c = elem c ("._-" :: [Char]) || isAscii c && isAlphaNum c +-- | Apply a reading to the aggregated data of a metric and increment its reading +-- count. Relative gauge changes saturate at 'maxBound' and never go below zero. +-- Calls 'error' if the reading does not match the type of the metric. +-- +-- Gauges saturate rather than wrap around because a gauge that suddenly jumps +-- from a huge value to zero, or below, would be worse than one stuck at its +-- maximum. addReading :: Value -> ByteString -> Metrics -> Metrics addReading reading = HashMap.adjust adjust where @@ -224,12 +335,16 @@ SetData (HashSet.insert e s) _ -> error "Stats reading mismatch" +-- | Apply a reading and return the updated reading count of the metric, or +-- @0@ if the metric does not exist. newReading :: Stats -> ByteString -> Value -> STM Int newReading stats key reading = do modifyTVar stats.metrics (addReading reading key) maybe 0 (.index) . HashMap.lookup key <$> readTVar stats.metrics +-- | Record a reading and, if samples are reported, send it subject to the +-- sampling rate. processSample :: (MonadIO m) => Stats -> Int -> ByteString -> Value -> m () processSample stats sampling key val = do @@ -238,9 +353,12 @@ submit stats $ Sample key val sampling idx +-- | Create an empty set of metrics. newMetrics :: (MonadIO m) => m (TVar Metrics) newMetrics = newTVarIO HashMap.empty +-- | Validate a configuration and derive its parameters. Calls 'error' if the +-- configuration is invalid. newParams :: StatConfig -> StatParams newParams cfg | valid = @@ -275,6 +393,7 @@ else namespace <> "." full_pfx_stats = pfx_namespace <> pfx_stats <> "." +-- | Build a client handle from its parts. newStats :: StatConfig -> TVar Metrics -> Socket -> Stats newStats cfg metrics socket = Stats @@ -283,17 +402,21 @@ params = newParams cfg } +-- | Flush statistics every flush interval, forever. statsLoop :: (MonadIO m) => Stats -> m () statsLoop stats = forever $ do threadDelay (stats.params.flush * 1000) statsFlush stats +-- | Send the statistics of all metrics and reset them. statsFlush :: (MonadIO m) => Stats -> m () statsFlush stats = do mapM_ (send stats) =<< atomically (stateTVar stats.metrics (flushStats stats.params)) +-- | Compute the reports for all metrics and the metrics reset for the next +-- interval. flushStats :: StatParams -> Metrics -> ([Report], Metrics) flushStats params metrics = let f xs key m = maybe xs ((<> xs) . statReports params key) m.dat @@ -302,9 +425,11 @@ ms = HashMap.map g metrics in (rs, ms) +-- | Join non-empty key parts with dots. catKey :: [ByteString] -> ByteString catKey = C.intercalate "." . filter (not . C.null) +-- | Reports for the aggregated data of a metric. statReports :: StatParams -> ByteString -> MetricData -> [Report] statReports params key dat = case dat of CounterData c -> @@ -335,13 +460,23 @@ ] TimingData s -> timingReports params key s +-- | Sorted timings with precomputed running totals, so that statistics for +-- any percentile can be read off a prefix of the vectors. data TimingStats = TimingStats - { timings :: !(Vector Int), + { -- | timings in ascending order + timings :: !(Vector Int), + -- | running sums of the timings cumsums :: !(Vector Int), + -- | running sums of the squares of the timings cumsquares :: !(Vector Int) } deriving (Eq, Ord, Show, Read) +-- | Sort timings and compute their running totals. +-- +-- Sorting once and keeping running sums lets every percentile's count, sum, +-- mean and sum of squares be read from a prefix of the vectors in constant time, +-- instead of recomputing them for each percentile. makeTimingStats :: [Int] -> TimingStats makeTimingStats timings = TimingStats @@ -352,12 +487,14 @@ where sorted = sort timings +-- | Reports for a timing metric, for every configured percentile. timingReports :: StatParams -> ByteString -> [Int] -> [Report] timingReports params key timings = concatMap (timingStats params key tstats) params.percentiles where tstats = makeTimingStats timings +-- | Keep only the lowest given percentage of the timings. trimPercentile :: Int -> TimingStats -> TimingStats trimPercentile pc ts = ts @@ -368,17 +505,21 @@ where f ls = V.take (length ls * pc `div` 100) ls +-- | Suffix for statistics of a percentile: empty for 100, @_N@ otherwise. percentileSuffix :: Int -> ByteString percentileSuffix pc | pc == 100 = "" | otherwise = C.pack $ printf "_%d" pc +-- | Turn a count over the flush interval into a count per second. computeRate :: StatParams -> Int -> Int computeRate params i = i * 1000 `div` params.flush +-- | Integer mean of the timings. Must not be called on empty stats. mean :: TimingStats -> Int mean ts = V.last ts.cumsums `div` length ts.timings +-- | Reports for a timing metric at one percentile. timingStats :: StatParams -> ByteString -> TimingStats -> Int -> [Report] timingStats params key tstats pc = concat @@ -400,12 +541,15 @@ pfx = params.pfxTimer <> key <> "." mkr s v = Report {key = pfx <> s <> sfx, value = v, rate = 1.0} +-- | Running sums of a list. cumulativeSums :: (Num a) => [a] -> [a] cumulativeSums = scanl1 (+) +-- | Running sums of the squares of a list. cumulativeSquares :: (Num a) => [a] -> [a] cumulativeSquares = scanl1 (+) . map (\x -> x * x) +-- | Population standard deviation of the timings, rounded. stdev :: TimingStats -> Int stdev ts = round $ sqrt var @@ -414,6 +558,7 @@ var = fromIntegral ds / fromIntegral len :: Double ds = sum $ map ((^ (2 :: Int)) . subtract (mean ts)) (V.toList ts.timings) +-- | Median of the timings, or @0@ if there are none. median :: TimingStats -> Int median ts | null ts.timings = 0 @@ -426,12 +571,14 @@ where middle = length ts.timings `div` 2 +-- | Reset aggregated data for the next interval. Gauges keep their value. flush :: MetricData -> MetricData flush (CounterData _) = CounterData 0 flush (GaugeData g) = GaugeData g flush (TimingData _) = TimingData [] flush (SetData _) = SetData HashSet.empty +-- | Turn a sample into a report if the sampling rate selects it. toReport :: StatParams -> Sample -> Maybe Report toReport params sample | sample.sampling > 0 && sample.index `mod` sample.sampling == 0 = @@ -443,6 +590,7 @@ } | otherwise = Nothing +-- | Encode a report in the StatsD line protocol. formatReport :: Report -> ByteString formatReport report = L.toStrict $ toLazyByteString builder where @@ -465,10 +613,15 @@ Set e -> byteString e <> "|s" +-- | Send a sample if the sampling rate selects it. submit :: (MonadIO m) => Stats -> Sample -> m () submit stats sample = forM_ (toReport stats.params sample) (send stats) +-- | Send a report, ignoring network errors. +-- +-- Errors are ignored on purpose: the alternative, failing the code that recorded +-- a metric because the network is down, is never what a program wants. send :: (MonadIO m) => Stats -> Report -> m () send stats report = liftIO $ @@ -478,6 +631,8 @@ stats.socket (formatReport report <> bool "" "\n" stats.params.newline) +-- | Resolve a host and port and create a UDP socket connected to it. Calls +-- 'error' if the host cannot be resolved. connectStatsD :: (MonadIO m) => String -> Int -> m Socket connectStatsD host port = liftIO $ do as <- @@ -492,6 +647,7 @@ Net.connect sock (Net.addrAddress a) return sock +-- | Parse a line in the StatsD protocol. The inverse of 'formatReport'. parseReport :: (MonadPlus m) => ByteString -> m Report parseReport bs = case C.split '|' bs of @@ -532,14 +688,17 @@ return t _ -> mzero +-- | Parse a value with its 'Read' instance. parseRead :: (MonadPlus m, Read a) => ByteString -> m a parseRead = maybe mzero return . readMaybe . C.unpack +-- | Parse a signed integer that takes up the whole string. parseInt :: (MonadPlus m) => ByteString -> m Int parseInt bs = case C.readInt bs of Just (i, bs') | B.null bs' -> return i _ -> mzero +-- | Parse a non-negative integer that takes up the whole string. parseNatural :: (MonadPlus m) => ByteString -> m Int parseNatural bs = case C.readInt bs of Just (i, bs') | B.null bs' -> guard (0 <= i) >> return i diff --git a/statsd-rupp.cabal b/statsd-rupp.cabal --- a/statsd-rupp.cabal +++ b/statsd-rupp.cabal @@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.38.0. +-- This file has been generated from package.yaml by hpack version 0.39.6. -- -- see: https://github.com/sol/hpack name: statsd-rupp -version: 0.5.0.1 +version: 0.5.0.2 synopsis: Simple StatsD Client description: Please see the README on GitHub at category: System