Documentation
¶
Overview ¶
Package metronome drives a unit of work at a controlled, live-adjustable rate across N workers and measures latency and errors. It is protocol-agnostic: it knows nothing about HTTP, gRPC, Prometheus, configuration, or UI.
The two seams are Runner (what to send) and RateController (how fast). Driver paces the work. Stats aggregates the resulting Result stream into percentile [Snapshot]s.
Pacing model ¶
Driver.Pacing selects between two modes.
ClosedLoop (the default, and v0.1's only behaviour) gives each worker a rate-limiter token, runs the work, and only then asks for the next token. It is self-throttling, but when the target slows the achieved rate sags below the offered rate — silently.
OpenLoop paces from a single dispatcher that never blocks on the target. A unit that finds no free worker is delivered immediately as a Result whose Err matches ErrSaturated, so saturation is counted rather than absorbed. Workers becomes the maximum in-flight cap. The dispatcher sleeps once per unit, so above roughly 4,000 rps it becomes the bottleneck itself; see the README's measured accuracy table.
Compare Snapshot.RPS against the rate you asked for on every run, in either mode.
Raw and corrected percentiles ¶
Every Result produced by a Driver carries Result.Scheduled, the time the unit was due. The schedule is anchored: Scheduled is the run's origin plus one interval per unit dispatched, advanced whether or not the generator got there on time, so a late generator is measured against the schedule rather than redefining it.
Stats reports raw percentiles (P50/P95/P99, from when work started) alongside coordinated-omission-corrected ones (CorrectedP50/P95/P99, from when the work was due). Read them as a pair: a large gap means the generator queued, and the raw numbers understate what a real client would have suffered by roughly that much.
For a run being watched while it happens, use RollingStats instead of Stats. It records into both a lifetime aggregate and a ring of trailing buckets, so Snapshot keeps its cumulative meaning while Window reports only the recent past. A ring bucket stores its latencies sparsely until it holds enough distinct values to earn a dense histogram, so Rolling.Bytes is the ceiling a configuration budgets for and RollingStats.Bytes is what a run actually holds — usually far less. The representation is invisible: every Window and Snapshot field reads the same either way. The distinction matters most for Snapshot.MaxScheduleLag, which is a lifetime maximum: one early stall pins it for the rest of the run, and a target that stops answering never moves any cumulative number at all, because there are no new Results to move it.
Recording, breaking down, and composing ¶
Recorder names what Stats and RollingStats both do. LabeledStats splits a Result stream on one Result.Labels key into a child Recorder per value plus a total, so a Mix of endpoints reports per-endpoint percentiles instead of one aggregate that describes no endpoint. It is generic over the child, so LabeledStats[*RollingStats] gives a trailing window per endpoint.
A series is a whole aggregate, so cardinality is capped: past MaxSeries distinct values, Results land in a single overflow series. Price a configuration with Bytes before paying for it.
Recorders compose. Multi fans one Result to several of them, Filter keeps only the Results a predicate accepts, and After, AfterTime and AfterN exclude a warmup period so that cold connection pools and unwarmed targets stay out of a threshold. Drain is the canonical loop over a Driver's result channel. Each is a pure Recorder transform: none owns a goroutine, a channel or a clock.
Diagnosing a run that fell short ¶
Snapshot.Saturated counts units the target had no free worker for; Snapshot.MaxScheduleLag is how far the generator itself fell behind. Saturation with no lag means the target could not keep up. Lag with no saturation means metronome could not — lower the rate, or use ClosedLoop.
Composing a load shape ¶
A RateController is a pure function of elapsed time, and four of them compose into the rest: Sine for a smooth periodic curve, and Sum, Repeat and Scale over any controller. A spike on a baseline is a constant plus a repeating phase table, which is why there is no Burst type.
The Driver samples the controller ten times a second, so shape features shorter than about 200ms do not exist and a cycle near that length aliases. Keep periods and spikes at a second or more.
Index ¶
- Constants
- Variables
- func Drain(ch <-chan Result, rec Recorder)
- type Adaptive
- type Clock
- type Constant
- type Driver
- type Labeled
- type LabeledStats
- type ManualClock
- type PacingMode
- type PanicError
- type Phase
- type Phased
- type Ramp
- type RateController
- type Recorder
- type Result
- type Rolling
- type RollingStats
- type Runner
- type RunnerFunc
- type Sine
- type Skipper
- type Snapshot
- type Stats
- type Weighted
Examples ¶
Constants ¶
const DefaultBuckets = 10
DefaultBuckets is the ring size NewRollingStats uses when Rolling.Buckets is zero.
const DefaultMaxSeries = 100
DefaultMaxSeries is the number of distinct label values a LabeledStats gives their own child recorder before further values land in the overflow series.
const DefaultOverflowLabel = "__overflow__"
DefaultOverflowLabel names the series that absorbs label values past MaxSeries.
const DefaultResultBuffer = 1024
DefaultResultBuffer is the capacity of Run's result channel when Driver.ResultBuffer is zero.
const DefaultUnlabeledLabel = "__none__"
DefaultUnlabeledLabel names the series for Results carrying no usable value for the configured key.
const DefaultWindow = 10 * time.Second
DefaultWindow is the trailing window NewRollingStats uses when Rolling.Window is zero.
const DefaultWorkers = 10
DefaultWorkers is the worker count Run uses when Driver.Workers is not positive.
Variables ¶
var ErrSaturated = errors.New("metronome: no worker free at scheduled time")
ErrSaturated marks an open-loop unit of work that found no free worker at its scheduled time. It is recorded rather than silently dropped, so saturation appears in Snapshot.ErrorRate instead of as invisible rate sag. Test for it with errors.Is.
Functions ¶
func Drain ¶ added in v0.6.1
Drain records every Result from ch into rec until ch is closed. It is the canonical consumption loop, named so that consumers stop rewriting the one place where getting it wrong leaks goroutines.
It deliberately takes no context. The Driver's contract on its result channel is blunt — abandoning a live one leaks the units still in flight for the lifetime of the process — so a Drain that returned early on cancellation would be a leak generator in the shape of good practice. Cancel the Driver's context instead: it stops, closes the channel, and this returns because the range ended.
It panics if rec is nil, rather than on the first Result to arrive — which, on a live run, would be after the Driver had already started generating load.
Types ¶
type Adaptive ¶
type Adaptive struct {
// contains filtered or unexported fields
}
Adaptive is a RateController whose rate is set externally. Safe for concurrent SetRate/Rate.
func NewAdaptive ¶
NewAdaptive returns an Adaptive controller starting at initial rps.
func (*Adaptive) Rate ¶
Rate reports the rate most recently passed to SetRate, ignoring elapsed time.
func (*Adaptive) SetRate ¶
SetRate sets the rate reported from the next Rate call onward. It is safe to call concurrently with Rate and from any goroutine.
A rate of zero or less is floored by the Driver to a token every ~10,000 seconds: effectively paused, and resumable — a later SetRate takes effect within one rate-update interval (see maxReservationWait), so pausing is not a one-way door. NaN is floored the same way rather than treated as "unlimited", because a control loop dividing by an empty PromQL vector produces NaN and flooding the target is the wrong direction to fail; see sanitizeRate.
type Clock ¶
type Clock interface {
// Now reports the current time.
Now() time.Time
// Sleep blocks for d, returning nil, or ctx.Err() if ctx is done first. A
// non-positive d returns immediately with ctx.Err() (nil for a live ctx).
Sleep(ctx context.Context, d time.Duration) error
}
Clock supplies time to the Driver. It is injected so that pacing is testable: with a ManualClock the whole pacing path — reservations, sleeps and the rate update cadence — advances only when the test says so.
func SystemClock ¶
func SystemClock() Clock
SystemClock returns a Clock backed by the wall clock. It is what the Driver uses when Driver.Clock is nil.
type Driver ¶
type Driver struct {
Runner Runner
Rate RateController
Workers int
MaxRequests int
// Clock supplies time to the whole pacing path: the rate schedule, the
// sleeps between units of work, and the rate-update cadence. Leave nil for
// the wall clock; inject a ManualClock to drive pacing exactly in tests.
Clock Clock
// ResultBuffer is the capacity of the channel Run returns. Zero selects
// DefaultResultBuffer; a negative value selects an unbuffered channel.
//
// Buffering matters for measurement, not just throughput: on an unbuffered
// channel a consumer that pauses blocks a worker that is holding a
// rate-limiter token, producing rate sag the target did not cause.
ResultBuffer int
// Pacing selects ClosedLoop (default) or OpenLoop. See PacingMode.
Pacing PacingMode
// Burst is the rate limiter's burst size; 0 means 1. Burst 1 gives the
// smoothest schedule. A larger burst lets the generator catch up in a bunch
// after a stall, which is sometimes what you want and is never smooth.
Burst int
}
Driver runs a Runner under a RateController across Workers goroutines, emitting Results on the channel Run returns until ctx is cancelled or MaxRequests results have been produced (MaxRequests == 0 means unlimited).
Contract: when MaxRequests is set, exactly MaxRequests Results are delivered unless ctx is cancelled first (cancellation aborts promptly and may drop in-flight results). In OpenLoop mode a saturated attempt counts as one of them. Results are NOT ordered — in OpenLoop a saturated unit is emitted immediately while an earlier unit is still running. The channel is closed once all internal goroutines exit; the caller must drain it or cancel ctx.
Example ¶
package main
import (
"context"
"fmt"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
runner := metronome.RunnerFunc(func(context.Context) metronome.Result {
return metronome.Result{Start: time.Now(), Latency: time.Millisecond}
})
d := metronome.Driver{Runner: runner, Rate: metronome.Constant(500), Workers: 4, MaxRequests: 100}
stats := metronome.NewStats()
for r := range d.Run(context.Background()) {
stats.Record(r)
}
fmt.Println(stats.Snapshot().Count)
}
Output: 100
func (*Driver) Run ¶
Run starts the workers and returns the channel Results are delivered on.
Run has a pointer receiver, so it cannot be called on a composite literal: bind the Driver to a variable first (d := Driver{...}; d.Run(ctx)).
The caller MUST drain the returned channel until it closes, or cancel ctx. Abandoning a live channel leaves the workers blocked on the send and leaks them for the lifetime of the process.
Run panics if Runner or Rate is nil — programmer errors, not runtime conditions. Workers <= 0 defaults to DefaultWorkers.
type Labeled ¶ added in v0.6.0
type Labeled[R Recorder] struct { // Key is the Result.Labels key to split on. Required. // // One key, not a set of them: a series here is a whole aggregate — at // least 272 KiB as a *Stats and about 3.2 MiB as a *RollingStats — so // keying on combinations would multiply memory by the cardinality of every // label a Result happens to carry. Key string // New constructs one child recorder. Required: there is no expression for // the zero value of an arbitrary R that would produce a working recorder, // so this cannot be defaulted. NewStats has exactly this signature: // // Labeled[*Stats]{Key: "endpoint", New: NewStats} // // It is called lazily — once per distinct label value, plus once for the // total and once for overflow — so a run with four endpoints allocates six // children, not MaxSeries of them. New func() R // MaxSeries caps the number of distinct label values that get their own // child. 0 means DefaultMaxSeries. // // This is a memory bound, not a tuning knob. A label value that turns out // to carry a request ID rather than a route name would otherwise allocate a // child per request; past the cap those Results land in the Overflow series // instead, so a mislabelled run loses reporting detail rather than memory. // // The Unlabeled series counts against the cap, because it is created by the // same lazy path a real value takes. A run at the default that carries any // unlabeled traffic therefore breaks down 99 label values, not 100. The // Overflow series does not count against it — it is where the cap sends // things, so charging the cap for it would be circular. MaxSeries int // Overflow names the series absorbing label values past MaxSeries. // "" means DefaultOverflowLabel. A real label value equal to it shares that // series. Overflow string // Unlabeled names the series for Results whose Labels is nil, lacks Key, or // holds an empty string for it. "" means DefaultUnlabeledLabel. Such // Results get a named series rather than being dropped, because "which // requests were not labelled?" is the question being asked, and dropping // them would stop the series reconciling with the total. Unlabeled string }
Labeled configures a LabeledStats. Key and New are required.
type LabeledStats ¶ added in v0.6.0
type LabeledStats[R Recorder] struct { // contains filtered or unexported fields }
LabeledStats splits a Result stream on one Result.Labels key, accumulating a child Recorder per distinct value alongside a total fed by every Result.
Snapshot reports the total, and is exactly what a plain Stats fed the same stream would have produced — so existing code holding a *Stats can hold one of these instead and see its numbers unchanged, gaining the breakdown rather than trading for it.
It is generic over the child so that a breakdown of rolling windows keeps its type: LabeledStats[*RollingStats].Series()["get_user"].Window() needs no type assertion.
Nesting typechecks — *LabeledStats[R] is itself a Recorder — and is almost always a mistake: LabeledStats[*LabeledStats[*Stats]] multiplies the child count by both caps, which at the defaults is 100 x 100 x 272 KiB. A second dimension usually wants to be a second LabeledStats fed the same stream through Multi, not a nested one.
Safe for concurrent Record.
Example ¶
package main
import (
"context"
"fmt"
"sort"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
endpoint := func(name string, lat time.Duration) metronome.Runner {
return metronome.RunnerFunc(func(context.Context) metronome.Result {
return metronome.Result{
Latency: lat,
Labels: map[string]string{"endpoint": name},
}
})
}
d := metronome.Driver{
Runner: metronome.Mix(
metronome.Weighted{Runner: endpoint("list", 5*time.Millisecond), Weight: 3},
metronome.Weighted{Runner: endpoint("search", 80*time.Millisecond), Weight: 1},
),
Rate: metronome.Constant(5000),
Workers: 8,
MaxRequests: 400,
}
stats := metronome.NewLabeledStats(metronome.Labeled[*metronome.Stats]{
Key: "endpoint",
New: metronome.NewStats,
})
for r := range d.Run(context.Background()) {
stats.Record(r)
}
names := make([]string, 0, stats.Len())
for name := range stats.Series() {
names = append(names, name)
}
sort.Strings(names)
series := stats.Series()
list, search := series["list"].Snapshot(), series["search"].Snapshot()
total := stats.Snapshot()
fmt.Println("series:", names)
fmt.Println("total count:", total.Count)
// The motivating defect, shown rather than described. list is three
// quarters of the traffic and sixteen times faster, but the aggregate P99
// is search's alone — so the one number a flat Stats reports describes a
// quarter of the requests and hides the rest.
fmt.Println("list p99:", list.P99.Round(time.Millisecond))
fmt.Println("search p99:", search.P99.Round(time.Millisecond))
fmt.Println("total p99:", total.P99.Round(time.Millisecond))
// And the series always reconcile with the total.
fmt.Println("series sum == total:", list.Count+search.Count == total.Count)
}
Output: series: [list search] total count: 400 list p99: 5ms search p99: 80ms total p99: 80ms series sum == total: true
func NewLabeledStats ¶ added in v0.6.0
func NewLabeledStats[R Recorder](cfg Labeled[R]) *LabeledStats[R]
NewLabeledStats returns a LabeledStats configured by cfg. It panics on the programmer errors listed on Labeled.
func (*LabeledStats[R]) Bytes ¶ added in v0.6.0
func (ls *LabeledStats[R]) Bytes() int64
Bytes reports the histogram memory currently held: the total plus every live series. It returns -1 if the child type does not report its own size, which a custom Recorder is not required to do — -1 rather than 0, because 0 is a plausible-looking number that would read as "free".
Series are lazy, so this is what is held now. The worst case is (MaxSeries + 2) x one child: the cap, plus overflow, plus the total.
func (*LabeledStats[R]) Len ¶ added in v0.6.0
func (ls *LabeledStats[R]) Len() int
Len reports how many series exist so far, including the overflow series once it has been created. Series are lazy, so this grows as label values are seen.
func (*LabeledStats[R]) Record ¶ added in v0.6.0
func (ls *LabeledStats[R]) Record(r Result)
Record adds r to the total and to exactly one series. It is safe to call concurrently.
Every Result lands in the total and in exactly one series, so once recording has stopped the sum of the series' Counts equals the total's. That reconciles at rest, not at every instant: the two child calls below are made in sequence and outside this LabeledStats' lock, so a reader polling a live run can catch a Result counted in the total and not yet in its series, or a series that exists with a Count of zero. Serialising them would make the invariant hold continuously at the cost of funnelling every series through one mutex, which is the contention this type is shaped to avoid.
func (*LabeledStats[R]) Series ¶ added in v0.6.0
func (ls *LabeledStats[R]) Series() map[string]R
Series returns the per-label recorders, including the overflow and unlabeled series once they exist.
The map is a copy the caller owns; the recorders in it are live and shared with this LabeledStats. Read them — Snapshot, and Window when the child type has one — but do not Record into them, which would count a Result in a series without counting it in the total.
func (*LabeledStats[R]) Snapshot ¶ added in v0.6.0
func (ls *LabeledStats[R]) Snapshot() Snapshot
Snapshot returns the total: the aggregate over every Result recorded, regardless of label. It is exactly what a plain Stats fed the same stream would have produced.
func (*LabeledStats[R]) Total ¶ added in v0.6.0
func (ls *LabeledStats[R]) Total() R
Total returns the recorder behind Snapshot, typed. Use it to reach a child method the Recorder seam does not carry — most usefully LabeledStats[*RollingStats].Total().Window().
type ManualClock ¶
type ManualClock struct {
// contains filtered or unexported fields
}
ManualClock is a Clock whose time only advances when Advance is called. Goroutines sleeping on it wake when Advance moves the clock to or past their deadline, which makes pacing tests exact instead of tolerance-based.
func NewManualClock ¶
func NewManualClock(t time.Time) *ManualClock
NewManualClock returns a ManualClock reading t.
func (*ManualClock) Advance ¶
func (m *ManualClock) Advance(d time.Duration)
Advance moves the clock forward by d, waking every sleeper whose deadline the clock has now reached, in deadline order.
func (*ManualClock) BlockUntilSleepers ¶ added in v0.2.0
func (m *ManualClock) BlockUntilSleepers(n int)
BlockUntilSleepers blocks until at least n goroutines are asleep on this clock. It exists to remove the race between a goroutine reaching Sleep and a test calling Advance: without it, an Advance that lands first is simply lost and the test hangs.
It has no timeout on purpose — if the count is never reached, go test's own deadline reports it with every goroutine's stack, which is more useful than a bespoke error.
func (*ManualClock) Now ¶
func (m *ManualClock) Now() time.Time
Now reports the clock's current time.
type PacingMode ¶ added in v0.2.0
type PacingMode int
PacingMode selects how the Driver reacts when the target cannot keep up.
const ( // ClosedLoop is the default: a worker does not ask for its next token until // the current unit of work has completed. Simple and self-throttling, but a // slow target reduces the achieved rate, and latency percentiles are subject // to coordinated omission (Stats' Corrected* fields quantify how much). ClosedLoop PacingMode = iota // OpenLoop keeps the schedule regardless of the target: one dispatcher paces // and hands each unit to a free worker, or — if all Workers are busy — // records a Result carrying ErrSaturated and moves on. Saturation becomes a // counted signal instead of silent rate sag. Workers is the maximum // in-flight cap in this mode. OpenLoop )
type PanicError ¶ added in v0.2.0
PanicError reports a panic raised inside a Runner. The Driver recovers it so that one bad unit of work cannot abort a whole load run; the panic is delivered as a failed Result instead, and Stack holds the stack captured at the point of recovery.
func (*PanicError) Error ¶ added in v0.2.0
func (e *PanicError) Error() string
type Phased ¶
type Phased struct {
Phases []Phase
}
Phased steps through phases by elapsed time, holding the last phase's rate past the end.
func (Phased) Duration ¶ added in v0.6.1
Duration reports the elapsed time at which the last phase ends, or zero when there are no phases.
Rate holds the last phase's rate past this point, so this is a measurement boundary rather than a stop condition — a Driver runs until its context ends or MaxRequests is reached, whatever the controller says.
func (Phased) PhaseEnd ¶ added in v0.6.1
PhaseEnd reports the elapsed time at which phase i ends: the sum of the durations of phases 0 through i. It panics if i is out of range.
Pair it with After to keep one source of truth for a warmup boundary:
measured := metronome.After(rate.PhaseEnd(0), stats)
rather than repeating the first phase's duration, where the two can drift and only one of them fails loudly.
type RateController ¶
RateController decides the target rate (requests/sec) over elapsed time.
func Repeat ¶ added in v0.8.0
func Repeat(c RateController, period time.Duration) RateController
Repeat returns a RateController that cycles c with the given period, reporting c.Rate(elapsed mod period). It panics if c is nil.
This is what makes a finite shape endless, and cyclic is what diurnal traffic, periodic spikes and sawtooth all are. Pair it with a phase table's own length so the two cannot drift:
metronome.Repeat(p, p.Duration())
Sine is already periodic and is not wrapped in Repeat.
A non-positive period delegates straight to c and does not cycle: there is no cycle length to place elapsed on, and unlike a nil controller a zero period plausibly means "do not repeat". Ramp sets the same precedent, where a non-positive Over reports End immediately.
The Driver samples the controller ten times a second, so a period near 200ms aliases into a shape unrelated to the one asked for, and anything shorter does not survive sampling at all. Keep the period, and any feature inside it, at a second or more.
Example ¶
Repeat makes a finite shape endless. Pairing a phase table with its own Duration keeps the cycle length in one place.
package main
import (
"fmt"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
p := metronome.Phased{Phases: []metronome.Phase{
{Duration: 2 * time.Second, TargetRPS: 10},
{Duration: 3 * time.Second, TargetRPS: 40},
}}
c := metronome.Repeat(p, p.Duration())
for _, at := range []time.Duration{0, 2 * time.Second, 5 * time.Second, 7 * time.Second} {
fmt.Printf("%v: %.0f rps\n", at, c.Rate(at))
}
}
Output: 0s: 10 rps 2s: 40 rps 5s: 10 rps 7s: 40 rps
func Scale ¶ added in v0.8.0
func Scale(factor float64, c RateController) RateController
Scale returns a RateController reporting factor times c's rate. It panics if c is nil.
It exists for reuse rather than arithmetic: a Sine can be rescaled by hand through its Min and Max, but a Phased table cannot without rewriting every phase. Scale runs an agreed profile at a fraction of its intensity — a canary, or a reduced-blast-radius rerun of the same shape.
A negative factor is not rejected here; the Driver floors the result, as it does for any controller.
func Sum ¶ added in v0.8.0
func Sum(cs ...RateController) RateController
Sum returns a RateController reporting the sum of every controller's rate at a given elapsed. It panics if cs is empty or if any controller is nil.
This is how a spike on a baseline is expressed, which is why the package ships no Burst type:
metronome.Sum(metronome.Constant(100), metronome.Repeat(spike, 10*time.Minute))
It also puts a floor under an Adaptive controller, so a control signal that collapses cannot drive the run down to nothing.
Nothing clamps the total, in either direction. A sum large enough to exceed what the target or the generator can serve shows up as ErrSaturated and shortfall in the Snapshot, the same as any other over-ambitious rate; a total that a negative member drags to zero or below is floored by the Driver, as any controller's result is.
Example ¶
A spike on a baseline needs no Burst type: it is a constant plus a repeating phase table. Sum adds the two, so the composed shape is 100 rps rising to 500 for one minute in every ten.
package main
import (
"fmt"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
c := metronome.Sum(metronome.Constant(100),
metronome.Repeat(metronome.Phased{Phases: []metronome.Phase{
{Duration: 9 * time.Minute, TargetRPS: 0},
{Duration: 1 * time.Minute, TargetRPS: 400},
}}, 10*time.Minute))
for _, at := range []time.Duration{0, 9 * time.Minute, 10 * time.Minute} {
fmt.Printf("%v: %.0f rps\n", at, c.Rate(at))
}
}
Output: 0s: 100 rps 9m0s: 500 rps 10m0s: 100 rps
type Recorder ¶ added in v0.6.0
Recorder accumulates Results into an aggregate.
*Stats, *RollingStats and *LabeledStats all satisfy it. It is the seam the per-label breakdown and the recorder combinators are written against, so a caller can substitute any of them — or their own — wherever one is taken.
Window is deliberately absent: it is meaningful for *RollingStats and for nothing else, and the honest answers a lifetime aggregate could give (a zero Snapshot, or the lifetime one relabelled) are both worse than not being asked. LabeledStats recovers typed access to it by being generic over the child type instead.
Bytes is absent for a different reason: a Recorder that writes CSV has no meaningful size to report, and should not owe one to satisfy an interface about recording. LabeledStats.Bytes reaches it through an optional interface.
func Filter ¶ added in v0.6.1
Filter returns a Recorder that records only the Results for which keep reports true. It panics if keep or rec is nil.
keep is called once per Result, on the calling goroutine, and may keep state — the warmup skippers are exactly this with a stateful predicate. It must not mutate the Result.
Snapshot delegates to rec, so the filtered aggregate reads normally.
func Multi ¶ added in v0.6.1
Multi returns a Recorder that records each Result into every rec, in argument order. It panics if given no recorders.
Use it to keep several views of one run without plumbing a second channel: a flat Stats beside a LabeledStats beside a RollingStats, all fed from one drain loop.
Snapshot delegates to the first recorder. There is no non-arbitrary aggregate answer — the recorders may be different types measuring different populations — so Snapshot exists here to satisfy Recorder, which is what lets a Multi nest inside After. Hold each recorder by name and read it directly.
It is synchronous, on the calling goroutine, and that has a sharp edge: time spent in Record is time not spent receiving from the result channel, so a slow recorder fills the channel, then the Driver's delivery backlog, and in open loop surfaces as ErrSaturated — which reads as "the generator ran out of workers" and is counted inside Errors and ErrorRate. A slow writer in a Multi can therefore present itself as a target problem. Everything in a Multi should be an in-memory aggregate; anything doing I/O belongs behind the caller's own buffered goroutine, where the caller owns the policy for falling behind.
Every recorder receives the same Result, and Result holds Labels by reference, so a recorder that mutates the map corrupts what the others see. Treat Results as read-only.
type Result ¶
type Result struct {
// Scheduled is the time this unit of work was due per the pacing schedule —
// the run's origin plus one interval for each unit dispatched before it,
// advanced independently of when the generator actually got to it.
//
// Start minus Scheduled is therefore the generator's own queueing delay: the
// wait a client that never fell behind would have suffered before its
// request even started. It is what Stats uses to correct for coordinated
// omission, and it is aggregated as Snapshot.MaxScheduleLag.
//
// It may be in the past relative to Start (the generator was late — the
// whole point) or in the future when Burst > 1 lets several units go at once
// and the schedule says the later ones were not due yet; Stats floors the
// correction at zero for that case. Zero if the Result did not come from a
// Driver.
Scheduled time.Time
Start time.Time
Latency time.Duration
Err error
Code string // status/gRPC code, caller-supplied
Bytes int64
// Labels is caller-side metadata carried alongside the Result, e.g.
// {"endpoint": "get_user"}.
//
// Stats ignores it; LabeledStats is what reads it back, splitting the stream
// on one key into a child Recorder per value plus a total:
//
// NewLabeledStats(Labeled[*Stats]{Key: "endpoint", New: NewStats})
//
// Leave it nil if unused. It is a map, so stamping one costs an allocation
// and roughly 150ns per request — measured, and the reason Mix does not
// stamp a label of its own.
Labels map[string]string
}
Result is the outcome of one unit of work.
type Rolling ¶ added in v0.5.0
type Rolling struct {
// Window is the trailing duration Window() reports over. Zero selects
// DefaultWindow.
//
// It is truncated to a whole multiple of Buckets: 10s over 3 buckets gives
// 3.333s buckets spanning 9.999s. Snapshot.Window always reports what the
// ring actually covers, so the difference is visible rather than assumed.
Window time.Duration
// Buckets is the ring size — the resolution at which the window retires old
// data. Zero selects DefaultBuckets. More buckets means a smoother window.
//
// It is also a memory multiplier, which is the part that surprises people.
// Each bucket holds a raw and a corrected HDR histogram, and a RollingStats
// allocates Buckets+2 of them: the ring, the lifetime aggregate and a scratch
// merge target. At the default range that is ~272 KiB per bucket, so the
// default Rolling is ~3.2 MiB and Buckets: 1000 is ~270 MiB. Widening Lo/Hi
// or raising Sigfigs multiplies again — 5 significant figures over a
// microsecond-to-hour range is ~16 MiB per histogram.
//
// Call Bytes to price a configuration before building it. Narrow the
// histogram range when you want a large ring: 1,000 buckets over
// 1ms-1s at 1 significant figure is ~2 MiB.
Buckets int
// Clock supplies the time that rotates the ring. Nil selects the wall clock;
// inject a ManualClock to make window tests exact. Wire it to the same clock
// as Driver.Clock when both are in play.
Clock Clock
// Lo, Hi and Sigfigs are the histogram range every bucket, the lifetime
// aggregate and the scratch target are built with, exactly as in
// NewStatsRange. Zero values select that function's defaults: 1µs, 60s and
// 3 significant digits.
Lo, Hi time.Duration
Sigfigs int
}
Rolling configures a RollingStats. The zero value is valid and gives a 10s window in 10 buckets on the wall clock over the NewStats histogram range.
func (Rolling) Bytes ¶ added in v0.5.0
Bytes reports the memory NewRollingStats(cfg) would allocate for its histograms, so a configuration can be priced before it is built:
if cfg.Bytes() > budget {
cfg.Buckets /= 2
}
Buckets is a memory multiplier as much as a resolution knob — Buckets+2 Stats, each holding a raw and a corrected histogram — and this is the number that says so. It counts the histogram count arrays, which dominate; the surrounding structs, the codes maps and the recorded Results are not included.
It panics on exactly the configurations NewRollingStats panics on, so pricing an unbuildable config reports the problem rather than a number for it.
This is the ceiling, reached once every ring bucket has promoted to a dense histogram. Buckets start sparse, so a live RollingStats.Bytes reports what is actually held and is typically well under this. Budget with this one; read the other to see what a run costs.
type RollingStats ¶ added in v0.5.0
type RollingStats struct {
// contains filtered or unexported fields
}
RollingStats aggregates Results into two views at once: the cumulative lifetime aggregate Stats gives, and a trailing window over the recent past.
Use Snapshot for an end-of-run report and Window for anything watching a run while it happens. Every number Stats produces is cumulative, so one early stall pins Snapshot.MaxScheduleLag at its worst value for the rest of the run and a target that stops answering entirely never changes Snapshot.RPS — there are no new Results to move it. Window has neither property.
Safe for concurrent Record.
func NewRollingStats ¶ added in v0.5.0
func NewRollingStats(cfg Rolling) *RollingStats
NewRollingStats returns a RollingStats configured by cfg. The zero Rolling is valid.
It panics on configuration no defaulting can rescue — a negative field, a histogram range NewStatsRange rejects, or a Window narrower than one nanosecond per bucket. Those are programmer errors, and there is no error path on a constructor two consumers call at startup.
func (*RollingStats) Bytes ¶ added in v0.6.0
func (rs *RollingStats) Bytes() int64
Bytes reports the memory this RollingStats currently holds for its latency distributions: the lifetime aggregate, the scratch merge target, and every ring bucket in whichever mode it is presently in.
This is an actual figure, not a projection, and it can sit far below the Rolling.Bytes of the config it was built from. A bucket starts as a sparse value-to-count store and only allocates a dense histogram once it holds enough distinct values to be worth one, so a run with few distinct latencies per bucket never pays the full price. Rolling.Bytes remains the ceiling: what this returns once every bucket has promoted.
It takes the lock, which the version that returned a constant did not need. This one walks the ring while Record may be promoting a bucket out from under it.
func (*RollingStats) Record ¶ added in v0.5.0
func (rs *RollingStats) Record(r Result)
Record adds one Result to both the lifetime aggregate and the current bucket. It is safe to call concurrently.
Which bucket a Result lands in is decided by the clock at the moment Record is called, not by Result.Start. A caller that buffers Results and drains them in batches therefore attributes a batch to the window it drained in rather than the window the work happened in. Record as you receive, which is what ranging over the Driver's channel already does.
func (*RollingStats) Snapshot ¶ added in v0.5.0
func (rs *RollingStats) Snapshot() Snapshot
Snapshot returns the cumulative aggregate over the whole run, with exactly the meaning Stats.Snapshot has: RPS inferred from the Result timestamps and Snapshot.Window zero. Use Window for the trailing view.
func (*RollingStats) Window ¶ added in v0.5.0
func (rs *RollingStats) Window() Snapshot
Window returns the aggregate over the trailing Rolling.Window — or over however much of it the run has covered so far, which Snapshot.Window reports.
It rotates the ring before reading, so a run that has stopped producing Results drains to zero instead of freezing at its last healthy numbers. That is the difference from Snapshot, and the reason to call this one from a control loop: Snapshot.MaxScheduleLag is a lifetime maximum that cannot tell "the generator is behind now" from "the generator was behind once".
Safe to call concurrently with Record. The returned Snapshot is a value and its Codes map is a copy, so it is the caller's to keep.
type RunnerFunc ¶
RunnerFunc adapts a function to a Runner.
type Sine ¶ added in v0.8.0
Sine oscillates smoothly between Min and Max over Period, starting at Min.
Rate(0) is Min, Max at Period/2 and Min again at Period. Starting at the trough rather than mid-curve means a run opens at its lowest load and rises, the way Ramp opens at Start — a curve that began at half load would put the cold-start cost of the run straight into the measurement.
Sine is already periodic, so it is not wrapped in Repeat. Repeat is for finite shapes such as Phased and Ramp.
Min need not be below Max: the curve runs Min to Max and back either way. Negative values are floored by the Driver and the Min/Max parameterisation is chosen so the literal a caller reaches for cannot drift negative the way an amplitude-and-offset one can.
The Driver samples the controller ten times a second, so a Period below about a second renders as visible steps rather than a curve and one near 200ms does not survive sampling at all. Keep Period at a second or more.
Example ¶
A Sine gives a smooth periodic load rather than the steps a phase table produces. It starts at Min, peaks at Period/2, and returns to Min.
package main
import (
"fmt"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
s := metronome.Sine{Min: 100, Max: 300, Period: 10 * time.Second}
for _, at := range []time.Duration{0, 2500 * time.Millisecond, 5 * time.Second} {
fmt.Printf("%v: %.0f rps\n", at, s.Rate(at))
}
}
Output: 0s: 100 rps 2.5s: 200 rps 5s: 300 rps
type Skipper ¶ added in v0.6.1
type Skipper struct {
// contains filtered or unexported fields
}
Skipper is a Recorder that excludes some Results from the recorder it wraps and counts what it excluded.
It exists so that an exclusion is auditable: Snapshot().Count plus Skipped() is the whole population the Skipper was offered. A report that says "measured 4,500 of 5,000 — 500 excluded as warmup" is honest; one that says "4,500 requests" invites the reader to wonder where the rest went.
Skipped is not a Snapshot field: Stats, RollingStats and LabeledStats never skip anything, so it would be zero in every existing use and non-zero only when something upstream — which a Snapshot cannot see — happened to be a Skipper. Snapshot describes what was recorded; this describes what was not.
Safe for concurrent Record.
func After ¶ added in v0.6.1
After returns a Skipper that excludes Results scheduled within d of the run's start, then admits the rest. It panics if d is negative or rec is nil.
Use it to keep a warmup out of a measurement: cold connection pools, TLS handshakes that will be reused, an empty page cache and a target that has not warmed up are not the system under test, and a percentile threshold asserted over them is asserted over the handshake. Phased can drive a gentler opening phase, but it changes the load rather than the measurement — those Results still land in the same Stats.
The boundary is anchored on the first Result's Scheduled stamp, falling back to Start, so it is fixed by the stream rather than by when this Skipper was constructed, and it is measured against the schedule rather than against when the generator got there. A Result carrying neither stamp is skipped and counted, and does not set the anchor; under a Driver that never happens, because the Driver stamps Start even when a Runner forgets.
Admission is decided per Result, not as a one-way switch, so a unit that was due inside the warmup is excluded even if it arrives after one that was due outside it — which happens routinely, because Results arrive off a channel fed by many workers.
The anchor can be up to one reordering window late, since the first Result seen need not be the earliest scheduled. That is negligible against a warmup measured in seconds; use AfterTime when the boundary must be exact.
After(0) admits every placeable Result and never anchors. A zero window has no boundary, so anchoring one on the first Result seen would exclude every later Result that was scheduled before it — a warmup the caller did not ask for. It is the natural value for a --warmup flag left unset, so it means exactly nothing rather than nearly nothing.
Example ¶
package main
import (
"fmt"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
// A hand-fed stream, so the schedule is exact and the example is
// deterministic: ten units due one second apart, the first three of them
// cold. After anchors on the first Result's Scheduled stamp, so the warmup
// boundary comes from the run rather than from when this was constructed.
origin := time.Unix(0, 0)
stream := make([]metronome.Result, 0, 10)
for i := range 10 {
lat := 5 * time.Millisecond
if i < 3 {
lat = 200 * time.Millisecond // cold: pools, handshakes, an empty cache
}
at := origin.Add(time.Duration(i) * time.Second)
stream = append(stream, metronome.Result{Scheduled: at, Start: at, Latency: lat})
}
warm := metronome.NewStats()
measured := metronome.After(3*time.Second, warm)
everything := metronome.NewStats()
for _, r := range stream {
everything.Record(r)
measured.Record(r)
}
// Rounded, because the exact figures are HDR bucket boundaries and this
// example is about the exclusion, not about histogram resolution.
fmt.Println("whole run p99: ", everything.Snapshot().P99.Round(time.Millisecond))
fmt.Println("warmed p99: ", measured.Snapshot().P99.Round(time.Millisecond))
fmt.Printf("measured %d of %d, excluded %d as warmup\n",
measured.Snapshot().Count, everything.Snapshot().Count, measured.Skipped())
}
Output: whole run p99: 200ms warmed p99: 5ms measured 7 of 10, excluded 3 as warmup
func AfterN ¶ added in v0.6.1
AfterN returns a Skipper that excludes the first n Results it sees, then admits the rest. It panics if n is negative or rec is nil.
The count is in arrival order, not schedule order — use After when the boundary is a point on the run's timeline rather than a number of units. AfterN(0) admits everything.
Example ¶
package main
import (
"context"
"fmt"
"time"
"github.com/RomanAgaltsev/metronome"
)
func main() {
// AfterN counts units instead of time, which is the right shape when the
// warmup is "the first N requests" rather than "the first N seconds".
n := 0
runner := metronome.RunnerFunc(func(context.Context) metronome.Result {
n++
lat := 5 * time.Millisecond
if n <= 100 {
lat = 200 * time.Millisecond
}
return metronome.Result{Latency: lat, Labels: map[string]string{"endpoint": "list"}}
})
d := metronome.Driver{
Runner: runner,
Rate: metronome.Constant(10000),
Workers: 1, // one worker, so the cold units are the first ones
MaxRequests: 500,
}
report := metronome.NewStats()
byRoute := metronome.NewLabeledStats(metronome.Labeled[*metronome.Stats]{
Key: "endpoint",
New: metronome.NewStats,
})
// Warmup exclusion wraps the fan-out, so both recorders see the same
// measured population. Multi(AfterN(100, report), byRoute) would be a
// different, also-valid program — and the difference is in the expression.
measured := metronome.AfterN(100, metronome.Multi(report, byRoute))
metronome.Drain(d.Run(context.Background()), measured)
fmt.Println("measured:", report.Snapshot().Count)
fmt.Println("excluded:", measured.Skipped())
fmt.Println("warm p99 under 10ms:", report.Snapshot().P99 < 10*time.Millisecond)
}
Output: measured: 400 excluded: 100 warm p99 under 10ms: true
func AfterTime ¶ added in v0.6.1
AfterTime returns a Skipper that excludes Results scheduled before t, then admits the rest. It panics if rec is nil.
It is the exact form of After, for a caller who knows the run's origin: record clock.Now() immediately before Driver.Run and pass origin.Add(warmup). Unlike After it does not infer the boundary from the stream, so it is unaffected by which Result arrives first.
Results carrying neither a Scheduled nor a Start stamp are skipped and counted, as with After.
func (*Skipper) Record ¶ added in v0.6.1
Record admits r to the wrapped recorder, or skips and counts it.
type Snapshot ¶
type Snapshot struct {
// Window is the trailing duration these numbers cover. Zero means they are
// cumulative over the whole run — the lifetime view, and what Stats.Snapshot
// always returns. Non-zero comes from RollingStats.Window and qualifies every
// other field here: they describe only the last Window of the run.
//
// During a run's first window it is less than the configured window, and
// during steady state it sawtooths across [window-interval, window] because
// the newest bucket is always partial. The numbers are correct for the period
// they cover; this field is what that period is, so a reader never has to
// guess whether the window had filled.
Window time.Duration
Count int64
Errors int64
RPS float64
ErrorRate float64
P50, P95, P99 time.Duration
Max time.Duration
// Clamped counts Results whose Latency fell outside the histogram's range
// and was clamped to a bound — at most one per Result. Non-zero means P50,
// P95 and P99 understate reality at one end; widen the range with
// NewStatsRange. Max always reports the true maximum regardless.
Clamped int64
// CorrectedClamped is the same count for the corrected histogram, which
// records Latency plus queueing delay and therefore overflows a bound the
// raw latency never reaches. It is counted separately because it says
// something about CorrectedP50/P95/P99 only — a run can have
// CorrectedClamped > 0 with Clamped == 0 and percentiles that are fine.
CorrectedClamped int64
// MaxScheduleLag is the largest gap between when a unit of work was due and
// when it actually started. It is the generator's own lateness: non-zero
// means metronome did not keep the schedule it offered, whatever the target
// did. Read it against Saturated — a large lag with Saturated == 0 says the
// generator, not the target, is the bottleneck.
MaxScheduleLag time.Duration
// Saturated counts Results carrying ErrSaturated: open-loop units that found
// no free worker at their scheduled time. They are included in Errors and
// ErrorRate, and this field is what separates "my generator ran out of
// workers" from "the target failed". Always zero in ClosedLoop.
Saturated int64
// Bytes is the sum of Result.Bytes.
Bytes int64
// Throughput is bytes/sec: RPS multiplied by the mean Bytes per Result, so
// it is the same kind of estimate as RPS rather than a raw total over the
// observed span. (Bytes/span would spend all N samples' bytes over the N-1
// intervals N timestamps actually bound — the bias RPS avoids.)
Throughput float64
// Codes counts Results by Result.Code. Results with an empty Code are
// counted in Count but not here. The map is a copy owned by the caller.
Codes map[string]int64
// CorrectedP* are the coordinated-omission-corrected percentiles: each
// Result's latency plus the time it spent waiting past its Scheduled send
// time. They answer "what would a client that kept to the schedule have
// seen?", which is the honest number when the target stalls. They are zero
// when no Result carried a Scheduled stamp.
//
// Read them against the raw P* above: a large gap means the generator
// queued, so the raw numbers understate what a real client would suffer.
CorrectedP50, CorrectedP95, CorrectedP99 time.Duration
// CorrectedCount is how many Results carried a Scheduled stamp and are
// therefore represented in CorrectedP*.
CorrectedCount int64
}
Snapshot is an aggregated view of many Results over a window.
func (Snapshot) String ¶ added in v0.4.0
String renders the numbers a run is usually judged on, in the order they should be read: what was achieved, what it cost, what a schedule-faithful client would have seen, and whether the generator itself kept up.
It is a summary, not a serialisation — Codes, Bytes, Throughput and the clamping counters are omitted. Reach for the fields directly when you need them.
A windowed Snapshot is prefixed "last <duration>: " so the two views are never confused in a log.
type Stats ¶
type Stats struct {
// contains filtered or unexported fields
}
Stats aggregates Results into percentile Snapshots. Safe for concurrent Record.
func NewStats ¶
func NewStats() *Stats
NewStats returns Stats recording latencies from 1µs to 60s with 3 significant digits — a sensible default for HTTP and gRPC. Use NewStatsRange when your work is slower or faster than that.
func NewStatsRange ¶ added in v0.2.0
NewStatsRange returns Stats recording latencies in [lo, hi] with sigfigs significant digits. Latencies outside the range are clamped to the nearest bound and counted in Snapshot.Clamped rather than dropped; Snapshot.Max always reports the true maximum.
It panics if lo <= 0, hi <= lo, or sigfigs is outside [1, 5] — programmer errors, and there is no error path on a constructor two consumers call at startup.
func (*Stats) Bytes ¶ added in v0.6.0
Bytes reports the memory this Stats holds for its latency distributions: two count arrays of int64. The surrounding struct, the codes map and the recorded Results are not included — the count arrays dominate and are the term that scales with the range and significant figures.
Every Stats a caller can construct is dense, so this is fixed at construction and needs no lock. The sparse form exists only inside a RollingStats ring, where RollingStats.Bytes takes the lock and totals the buckets.
func (*Stats) Merge ¶ added in v0.6.0
Merge folds src into s, so that s reports what it would have reported had it recorded every Result src did. Percentiles cannot be combined after the fact, so this merges the underlying histograms rather than the Snapshots.
Use it to roll a LabeledStats breakdown up into an ad-hoc subtotal, to combine per-worker aggregates recorded in parallel, or to combine separate runs.
Both Stats must have been built with the same range and significant figures; Merge panics otherwise, and panics if src is s. Range equality is a property of construction, so a mismatch fires on the first call rather than later and only for some values — it is a programmer error, and it is reported as one.
It is safe to call concurrently with Record on either Stats, and safe to call in both directions from different goroutines.