Metrics aggregator with time buckets

Q: How would this look in Go, and in production?

In Go: map[string]*series where series holds a type tag and a fixed-size ring of buckets indexed by (start / bucket) % retention, avoiding allocation after warm-up, all under a mutex or sharded by name. On hot paths counters are often atomic.Int64 values updated without a lock and swapped out at each bucket boundary by a flusher goroutine. Production agents such as StatsD aggregate like this and flush each closed bucket to a backend, rather than keeping history in the process.

mediumMetrics and health

Build the in-memory core of a metrics agent. Application code reports counters (things that only add up, like requests served) and gauges (a current reading, like queue depth). The agent rolls them into fixed time buckets, say 10 seconds each, and keeps only a limited history, so memory stays bounded no matter how long the process runs.

API (Java: setGauge, values are long, series returns List<Point> where a Point prints as (start, value))

◈ DIAGRAM

MetricsAggregator(bucket_seconds: int, retention_buckets: int, clock=time.monotonic)
incr(name: str, amount=1) -> None
set_gauge(name: str, value) -> None
series(name: str) -> list[(bucket_start, value)]   # oldest first

Rules

The bucket for time t starts at floor(t / bucket_seconds) * bucket_seconds.
Counters add every incr in a bucket together. Gauges keep the last value set in the bucket.
A name's type is fixed by its first use. Using a counter as a gauge, or the other way round, raises TypeError with the message "<name> is a <type>".
Only the newest retention_buckets buckets are kept, counting the current bucket even if it is empty. Older ones are dropped on the next write or read of that metric.
series lists only buckets that received data. An unknown name returns an empty list.
incr with a negative amount raises ValueError, because counters only go up.
Values are integers in this exercise so outputs are exact.

Examples use bucket_seconds=10 and retention_buckets=3. Example 2 continues from example 1.

Examples

Example 1

Input: t=1 m.incr("requests") t=4 m.incr("requests", 2) t=5 m.set_gauge("queue_depth", 7) t=12 m.incr("requests") t=18 m.set_gauge("queue_depth", 3) t=19 m.set_gauge("queue_depth", 4) t=19 m.series("requests") t=19 m.series("queue_depth")

Output: [(0, 3), (10, 1)] [(0, 7), (10, 4)]

Explanation: Counters sum within a bucket (1 + 2). The gauge keeps the last value in each bucket (4, not 3).

Example 2

Input: t=35 m.incr("requests") t=35 m.series("requests") t=35 m.series("queue_depth") t=35 m.series("unknown") t=35 m.set_gauge("requests", 1)

Output: [(10, 1), (30, 1)] [(10, 4)] [] error: requests is a counter

Explanation: At t=35 the current bucket is 30, and three buckets of retention keep 10, 20 and 30, so bucket 0 is dropped. Bucket 20 had no data and is not listed.

Hints

Approach

Optimal

Per metric name, keep its type and a deque of [bucket_start, value] pairs, oldest first.

_current_start(): floor(now / bucket) * bucket.
_prune(q, current): the oldest bucket worth keeping starts at current - (retention - 1) * bucket. Pop from the left while the front is older than that.
_bucket_for(name, kind): check or record the type, prune, and, if the newest bucket is older than the current one, append a fresh [current, 0]. Return the newest bucket.
incr adds to it, and set_gauge overwrites it.
series prunes before reading, so a metric nobody has written to recently still reports correct retention.

Because time moves forward, buckets are created in order, and both ends of the deque are O(1). Memory per metric is at most retention_buckets pairs.

ComplexityTime O(1) amortised per write; O(retention) per seriesSpace O(metrics × retention)

Python

import time
from collections import deque
 
 
class MetricsAggregator:
    COUNTER, GAUGE = "counter", "gauge"
 
    def __init__(self, bucket_seconds, retention_buckets, clock=time.monotonic):
        if bucket_seconds <= 0 or retention_buckets < 1:
            raise ValueError("bucket_seconds must be > 0 and retention_buckets >= 1")
        self.bucket = bucket_seconds
        self.retention = retention_buckets
        self.clock = clock
        self.kind = {}             # name -> "counter" | "gauge"
        self.buckets = {}          # name -> deque of [bucket_start, value], oldest first
 
    def _current_start(self):
        return int(self.clock() // self.bucket) * self.bucket
 
    def _prune(self, q, current):
        oldest_kept = current - (self.retention - 1) * self.bucket
        while q and q[0][0] < oldest_kept:
            q.popleft()
 
    def _bucket_for(self, name, kind):
        known = self.kind.setdefault(name, kind)
        if known != kind:
            raise TypeError(f"{name} is a {known}")
        q = self.buckets.setdefault(name, deque())
        current = self._current_start()
        self._prune(q, current)
        if not q or q[-1][0] < current:
            q.append([current, 0])
        return q[-1]               # writes land in the newest bucket
 
    def incr(self, name, amount=1):
        if amount < 0:
            raise ValueError("counters only go up")
        self._bucket_for(name, self.COUNTER)[1] += amount
 
    def set_gauge(self, name, value):
        self._bucket_for(name, self.GAUGE)[1] = value      # last write in the bucket wins
 
    def series(self, name):
        q = self.buckets.get(name)
        if q is None:
            return []
        self._prune(q, self._current_start())
        return [(start, value) for start, value in q]

Follow-up questions

Add labels (requests{status="500"}). What happens to memory if someone puts a user id in a label?
Add a histogram metric type with fixed latency buckets.
Flush each bucket to a remote backend when it closes, without blocking callers of incr.

Frequently asked questions

Per-bucket sums answer "how many requests in the last 10 seconds?" directly. Prometheus does the opposite: clients expose a monotonically increasing total, and the server computes rate() from differences between scrapes. That survives lost scrapes, because the next total still includes the missed increments, but it needs special handling when a process restarts and the total resets to zero. Both are valid; know which one you are building.

In Go: map[string]*series where series holds a type tag and a fixed-size ring of buckets indexed by (start / bucket) % retention, avoiding allocation after warm-up, all under a mutex or sharded by name. On hot paths counters are often atomic.Int64 values updated without a lock and swapped out at each bucket boundary by a flusher goroutine. Production agents such as StatsD aggregate like this and flush each closed bucket to a backend, rather than keeping history in the process.