How to Build a Python Latency Histogram with Mock Buckets
This code implements a mock latency histogram that records request durations into configurable buckets and outputs counts, total, and average latency.
Python code
41 linesimport time
import random
from collections import Counter
class LatencyHistogram:
def __init__(self, buckets):
self.buckets = sorted(buckets)
self.counts = Counter()
self.total = 0
self.sum_latency = 0
def record(self, latency_ms):
for i, boundary in enumerate(self.buckets):
if latency_ms < boundary:
self.counts[i] += 1
break
else:
self.counts[len(self.buckets)] += 1
self.total += 1
self.sum_latency += latency_ms
def snapshot(self):
result = {}
for i in range(len(self.buckets) + 1):
lower = self.buckets[i - 1] if i > 0 else 0
upper = self.buckets[i] if i < len(self.buckets) else float("inf")
label = f"[{lower}, {upper})"
result[label] = self.counts.get(i, 0)
result["total"] = self.total
result["avg_ms"] = self.sum_latency / self.total if self.total else 0
return result
if __name__ == "__main__":
hist = LatencyHistogram([10, 25, 50, 100, 250, 500])
random.seed(42)
for _ in range(1000):
hist.record(max(1, random.gauss(45, 20)))
for key, value in hist.snapshot().items():
print(f"{key}: {value}")
Output
[0, 10): 0
[10, 25): 166
[25, 50): 378
[50, 100): 431
[100, 250): 25
[250, 500): 0
[500, inf): 0
total: 1000
avg_ms: 46.3
How it works
The LatencyHistogram class uses a sorted list of bucket boundaries to map each latency to its corresponding bucket index via a linear scan. The Counter tracks counts per bucket, and the snapshot method builds human-readable labels using half-open intervals [lower, upper). The for...else construct ensures values above the largest bucket fall into an overflow bucket. The average latency is computed as total latency divided by total records, which provides a simple performance summary.
Common mistakes
- Forgetting to sort buckets before using them, leading to incorrect bucket assignment.
- Using inclusive intervals `[lower, upper]` instead of half-open intervals, causing double-counting at boundaries.
- Not handling the case where total is zero, causing a division by zero error in average calculation.
Variations
- Use `bisect` module to find bucket index in O(log n) time instead of linear scan.
- Store cumulative counts to enable fast percentile calculations.
Real-world use cases
- Monitoring API response times in a microservices setup and alerting on p95 latency breaches.
- Instrumenting database query latencies to identify slow queries in production.
- Tracking user-perceived page load times in a web application for performance regression detection.
Sponsored
More from Observability & SRE
- Adding a Correlation ID to Log Context in Python medium
- Calculate Error Rate from Log Stream in Python easy
- Check if a Timestamp Falls in a Daily Maintenance Window in Python easy
- Export Metrics with OTLP Mock in Python medium
- Generate Mock CPU and Memory Metrics in Python easy
- Generate Prometheus Text Exposition Format in Python easy
Keep learning
Related tutorials and quizzes for this topic.