From 04dc8e4546ea10c4f9b254a985bb23747c0f45c9 Mon Sep 17 00:00:00 2001 From: Michael Otteni Date: Sun, 31 May 2026 14:25:18 -0400 Subject: [PATCH] feat(metrics): collect client stats periodically, not only on scrape Per-client gauges were refreshed only when /metrics was scraped, so a client that connected and disconnected between two scrapes never showed up at all: the on-scrape gather runs at scrape time, finds the client already gone, and so never sets its gauge. Add a periodic collector (collectClientStats) started from Serve that gathers connected-client stats every 15s, independent of scrapes. A client connected when a tick fires has its gauge written to the sink; because the sink expires entries only after 60s (not on disconnect), that value survives to be emitted on the next scrape even though the client has since disconnected. This is the only path by which a client whose whole lifetime falls between two scrapes is recorded at all. Note this does NOT improve freshness for connected clients: the pre-existing on-scrape gather already re-sets every connected client immediately before the sink is collected, so scraped values are already current and the ticker's writes for them are overwritten before any read. The benefit is therefore conditional on the scrape interval. It narrows the short-lived-client miss window from the scrape interval down to min(15s, scrape_interval): it helps when Prometheus scrapes less often than every 15s (e.g. 30s or 60s), and is effectively a no-op when the scrape interval is <=15s, since the on-scrape gather then captures the same clients at equal-or-finer granularity. The 15s constant is a fixed guess at 'shorter than a typical scrape'; it cannot see the actual scrape interval. Minor tradeoff: because ticks keep refreshing each gauge's updatedAt, a disconnected client's last value can linger slightly longer before expiring than under scrape-only collection, bounded by the 60s Expiration either way. Collection remains intentionally incomplete: a client whose entire lifetime falls between two ticks is still never recorded. The per-client gauges are a best-effort, cardinality-bounded snapshot of currently- connected clients - the ClientIdentifier label embeds a per-connection hash, so departed clients must age out via the sink Expiration rather than accumulate forever. Promoting these to per-client counters to catch short-lived clients would defeat that bound and grow cardinality without limit. Where exact throughput is needed it is already captured per message by the aggregate counters arke_recvmsg_total / arke_sendmsg_total in the gRPC interceptors. Replaces the FIXME with notes describing the sampling strategy, its known gaps, and why they are acceptable. Fixes #111 Signed-off-by: Michael Otteni --- internal/metrics/prometheus/metrics.go | 69 ++++++++++++++++++--- internal/metrics/prometheus/metrics_test.go | 23 +++++++ 2 files changed, 83 insertions(+), 9 deletions(-) diff --git a/internal/metrics/prometheus/metrics.go b/internal/metrics/prometheus/metrics.go index bafff41..d1dddfa 100644 --- a/internal/metrics/prometheus/metrics.go +++ b/internal/metrics/prometheus/metrics.go @@ -36,6 +36,14 @@ var ( const EnvPprofEnabled = "ARKE_PPROF_ENABLED" +// clientStatsInterval controls how often connected-client stats are refreshed +// independent of Prometheus scrapes. It is shorter than the sink Expiration +// (60s) so that gauges for clients that connect and disconnect between two +// scrapes are still captured at least once, as long as the client is connected +// when a refresh fires. Clients whose entire lifetime falls between two refresh +// ticks are still not captured; see gatherClientStats. +const clientStatsInterval = 15 * time.Second + func init() { Stats = &stats{} @@ -104,11 +112,39 @@ func Serve(ctx context.Context, lis *net.Listener) { metricsServer := setupServer() go metricsServer.Serve(*lis) //nolint:errcheck + go collectClientStats(ctx, clientStatsInterval) <-ctx.Done() metricsServer.Shutdown(ctx) //nolint:errcheck } +// collectClientStats periodically gathers per-client stats until ctx is +// cancelled, independent of /metrics scrapes. Its purpose is to surface a +// client that connects and disconnects between two scrapes: a client connected +// when a tick fires has its gauge written to the sink, which (because the sink +// expires entries only after Expiration, not on disconnect) survives to be +// emitted on the next scrape even though the client is already gone. +// +// This only adds coverage when the scrape interval is longer than the tick +// interval; if Prometheus scrapes more frequently, the on-scrape gather already +// captures the same clients at equal-or-finer granularity and this is a no-op. +// It does not guarantee every client is recorded — a client whose lifetime fits +// entirely between two ticks is still missed. Disconnected clients age out of +// the sink via its Expiration. +func collectClientStats(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + gatherClientStats() + } + } +} + func gatherClientStatsHandler(h http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { gatherClientStats() @@ -116,15 +152,30 @@ func gatherClientStatsHandler(h http.Handler) http.Handler { }) } -// FIXME? -// I don't particularly like only collecting these client level stats when the metrics handler -// is called because it can and will lead to us never logging stats for some clients. -// We can even have incorrect statistics if part way through a client producing/consuming the -// metrics are collected, but they disconnect before the next metrics collection. -// I also don't like the idea of keeping a separate record of stats from every for an extended -// period of time even if they have disconnected. -// Maybe client level stats don't matter much as long as they're regarded as an incomplete look -// into the current state of only connected clients. +// gatherClientStats refreshes the per-client gauges for every client that is +// connected at the moment it runs. It is called from two places: on each +// /metrics scrape (so a scrape always reflects the latest values) and on a +// periodic interval via collectClientStats (so clients that connect and +// disconnect between scrapes are still refreshed, provided they are connected +// when a tick fires). +// +// This is a sampling approach and is inherently incomplete: a client whose +// entire lifetime falls between two collections is never recorded, because by +// the next collection it is already gone from the provider's connection set. +// This is by design. The per-client gauges are a best-effort snapshot of +// *currently connected* clients, kept deliberately cardinality-bounded: the +// ClientIdentifier label embeds a per-connection hash (see +// util.SetClientIdentifier), so series for departed clients must age out of the +// sink via its Expiration rather than accumulate forever. Promoting these to +// per-client counters to capture short-lived clients would defeat that bound +// and grow cardinality without limit. +// +// Exact, complete throughput is already available without sampling from the +// aggregate counters arke_recvmsg_total / arke_sendmsg_total, which are +// incremented per message in the gRPC interceptors and never miss a client. +// +// Stats for disconnected clients are not removed here; they age out of the sink +// via its Expiration. func gatherClientStats() { providers := provider.RegisteredProviders() for _, providerName := range provider.RegisteredProviders().GetList() { diff --git a/internal/metrics/prometheus/metrics_test.go b/internal/metrics/prometheus/metrics_test.go index d43879e..af8c56b 100644 --- a/internal/metrics/prometheus/metrics_test.go +++ b/internal/metrics/prometheus/metrics_test.go @@ -11,6 +11,7 @@ import ( "os" "strings" "testing" + "time" "github.com/sassoftware/arke/internal/metrics" "github.com/sassoftware/arke/internal/provider" @@ -95,6 +96,28 @@ func Test_gatherClientStats(t *testing.T) { assert.NotEmpty(t, Stats) } +func Test_collectClientStats(t *testing.T) { + _, _ = provider.GetProvider("amqp091") + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + collectClientStats(ctx, time.Millisecond) + close(done) + }() + + // Allow a few ticks to fire so the periodic gather runs. + time.Sleep(20 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("collectClientStats did not return after context cancel") + } + assert.NotEmpty(t, Stats) +} + func Test_Serve(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel()