EasyObservabilityPython 3
Sliding Window Error Counter
Track top recent errors while every read path lazily removes expired events.
25m3 sample tests7 hidden tests
Build a deterministic counter for recent errors in a rolling time window.
Requirements
- Define
ErrorCounter(window_seconds, now). record(message)stores one error occurrence at the current time.count()returns the number of non-expired errors.top(k)returns(message, count)pairs for the topkmessages, ordered by count descending then message ascending.- If
k <= 0,top(k)returns an empty list. - Every read path must treat expired events as missing.
recordmay also purge expired events, preventing unbounded storage growth. - An event expires when
now() - event_time >= window_seconds. - Ties in
top(k)sort alphabetically by message (already implied by count-desc then message-asc).
Example
python
clock = {"t": 0}
counter = ErrorCounter(10, now=lambda: clock["t"])
counter.record("timeout")
counter.record("timeout")
counter.record("rate_limit")
assert counter.count() == 3
assert counter.top(2) == [("timeout", 2), ("rate_limit", 1)]
clock["t"] = 10
assert counter.count() == 0Constraints
- Use deterministic injected time.
- Assume injected timestamps move forward monotonically; an ordered deque can't safely expire events after the clock moves backward.
window_secondsis finite and positive.now()returns finite numeric timestamps in seconds and may repeat a timestamp. Use one clock snapshot per operation. Messages are strings counted exactly, including empty strings;kis an integer.- Use standard-library Python only.
- Don't scan expired data forever if many events are recorded.
- Retain events in the active window rather than the entire history. Memory still grows with the number of events inside that window; a high event rate requires a separate limit or aggregation policy.
Editor