1 The Problem
We want a live metrics dashboard: services report numbers (requests, errors, response times), the server aggregates them into rolling statistics, and a web page shows them updating in real time. It teaches aggregating streaming data and computing rolling metrics — the engine behind every monitoring dashboard.
Where this shows up: Grafana, Datadog, application monitoring, trading dashboards, IoT telemetry, live analytics. Turning a stream of events into live, meaningful numbers is core to observability.
2 How to Think About It
Think about turning events into live stats, before any code:
The plan — in plain English
1. Services report metric events (a request took 120ms). → 2. The collector keeps a rolling window of recent events. → 3. On demand, it computes stats over that window: count, average, max, error rate. → 4. The dashboard polls these stats and redraws. Rolling windows keep the stats current and bounded.
3 The Build — explained part by part
Here is the complete metrics collector. The stats logic is fully testable. Each part is explained below.
Pythondashboard.py
from collections import deque
class MetricsCollector:
def __init__(self, window=100):
# A deque with maxlen auto-drops old events.
self.response_times = deque(maxlen=window)
self.total_requests = 0
self.total_errors = 0
def record(self, response_time_ms, is_error=False):
self.response_times.append(response_time_ms)
self.total_requests += 1
if is_error:
self.total_errors += 1
def stats(self):
times = list(self.response_times)
if not times:
return {"count": 0, "avg_ms": 0, "max_ms": 0, "error_rate": 0}
return {
"count": self.total_requests,
"avg_ms": round(sum(times) / len(times), 1),
"max_ms": max(times),
"error_rate": round(self.total_errors / self.total_requests, 3),
}
if __name__ == "__main__":
metrics = MetricsCollector()
metrics.record(120)
metrics.record(95)
metrics.record(300, is_error=True)
print(metrics.stats())
What each part does — in plain words
deque(maxlen=window) — a double-ended queue with a maximum length. When it is full and you add another item, the oldest is automatically dropped. This gives us a rolling window of recent events for free.record(response_time_ms, is_error) — log one event: append its response time, bump the total counters. Errors are counted separately for the error rate.
stats() — compute the live numbers from the current window: average and max response time over recent requests, plus the overall count and error rate.
the empty guard — if no events yet, return zeros instead of dividing by zero.
why a rolling window? — the average response time should reflect now, not all history. The window keeps the metric current and uses bounded memory no matter how long the server runs.
Common mistakes — and how to avoid them
✗ Keeping every event forever — memory grows without bound and old data skews the average.
✓ Use a rolling window (
deque(maxlen=...)) for bounded, current stats.✗ Dividing by zero before any events arrive — the dashboard crashes on startup.
✓ Guard the empty case and return zeros.
✗ Recomputing all history on every stats call — slow for long-running servers.
✓ Keep running totals and a bounded window.
4 Test & Prove Each Part
We test the stats computation with known events, so every number can be verified.
The average response time is computed correctly
The error rate reflects errors over total
An empty collector returns safe zeros
Pythontest_dashboard.py
from collections import deque
class MetricsCollector:
def __init__(self, window=100):
self.response_times = deque(maxlen=window)
self.total_requests = 0
self.total_errors = 0
def record(self, response_time_ms, is_error=False):
self.response_times.append(response_time_ms)
self.total_requests += 1
if is_error:
self.total_errors += 1
def stats(self):
times = list(self.response_times)
if not times:
return {"count": 0, "avg_ms": 0, "max_ms": 0, "error_rate": 0}
return {
"count": self.total_requests,
"avg_ms": round(sum(times) / len(times), 1),
"max_ms": max(times),
"error_rate": round(self.total_errors / self.total_requests, 3),
}
def test_average():
"""Average of 100 and 200 is 150."""
m = MetricsCollector()
m.record(100)
m.record(200)
assert m.stats()["avg_ms"] == 150
def test_error_rate():
"""One error in four requests is 0.25."""
m = MetricsCollector()
for _ in range(3):
m.record(100)
m.record(100, is_error=True)
assert m.stats()["error_rate"] == 0.25
def test_empty_safe():
"""An empty collector returns zeros, not an error."""
assert MetricsCollector().stats()["count"] == 0
Run with pytest -v. The empty-safe test guards the divide-by-zero case — a dashboard must not crash before any data arrives. Testing the empty state is a habit that prevents real outages.
5 API Documentation
recordcollector.record(ms, is_error)log a metric event
GET/statscurrent rolling metrics
Returns
{"count": 1024, "avg_ms": 142.5,
"max_ms": 980, "error_rate": 0.012}6 Run It & Automate It
Run it locally
python3 dashboard.pySee live stats from sample events. Wire it to the REST API project so a web page can poll
/stats and redraw.Jenkins runs the tests automatically — each line explained below.
What you should see when it works
Terminala real run
{'count': 3, 'avg_ms': 171.7, 'max_ms': 300, 'error_rate': 0.333}If it breaks — how to fix it
🚨
ZeroDivisionError.Add the empty-window guard so stats work before data arrives.
🚨 Average never reflects recent changes.
Make sure the window has a maxlen so old events drop off.
GroovyJenkinsfile
// Jenkinsfile — automated tests on every change.
pipeline {
agent any
stages {
stage('Get the code') { steps { checkout scm } } // download the code
stage('Install') { steps { sh 'python3 -m venv venv && . venv/bin/activate && pip install pytest' } } // test tool
stage('Test') { steps { sh '. venv/bin/activate && pytest -v' } } // run every test
}
post {
success { echo 'All tests passed.' }
failure { echo 'A test failed — look above.' }
}
}
🎯 Try this next — make it yours
- Percentiles. Add p95/p99 response times, what ops teams really watch. (Teaches: percentiles.)
- Per-endpoint metrics. Track stats per route. (Teaches: grouped metrics.)
- Live web view. Serve a page that auto-refreshes the stats. (Teaches: combining with the server.)
What you learned
You built a real-time metrics collector with a rolling window (deque) and live aggregate stats — the engine behind Grafana, Datadog, and every monitoring dashboard. Turning a stream of events into current numbers is core to observability. Related: collections, statistics.