How to mock Prometheus alert rule thresholds in Python
Simulate a Prometheus alert rule with a configurable threshold and duration window, firing only when the metric exceeds the threshold long enough.
Python code
50 linesimport time
import random
class MetricsStore:
def __init__(self):
self.metrics = {}
def set_metric(self, name, value, labels=None):
key = (name, tuple(sorted((labels or {}).items())))
self.metrics[key] = value
def get_metric(self, name, labels=None):
key = (name, tuple(sorted((labels or {}).items())))
return self.metrics.get(key, 0.0)
class PrometheusAlertRule:
def __init__(self, expr_name, threshold, duration_seconds, metrics_store):
self.metric_name = expr_name
self.threshold = threshold
self.duration = duration_seconds
self.metrics = metrics_store
self.firing_since = None
def evaluate(self, current_time=None):
current_time = current_time or time.time()
value = self.metrics.get_metric(self.metric_name)
if value > self.threshold:
if self.firing_since is None:
self.firing_since = current_time
sustained = current_time - self.firing_since
if sustained >= self.duration:
return f"FIRING: {self.metric_name}={value} exceeded {self.threshold} for {sustained:.1f}s"
else:
self.firing_since = None
return f"OK: {self.metric_name}={value}"
if __name__ == "__main__":
store = MetricsStore()
rule = PrometheusAlertRule(expr_name="http_requests_total", threshold=100, duration_seconds=5, metrics_store=store)
start = time.time()
for tick in range(8):
# Simulate increasing traffic
store.set_metric("http_requests_total", 50 + tick * 30)
print(f"t={tick}: {rule.evaluate(current_time=start + tick)}")
time.sleep(0.5)
Output
t=0: OK: http_requests_total=50
t=1: OK: http_requests_total=80
t=2: OK: http_requests_total=110
t=3: FIRING: http_requests_total=140 exceeded 100 for 1.0s
t=4: FIRING: http_requests_total=170 exceeded 100 for 2.0s
t=5: FIRING: http_requests_total=200 exceeded 100 for 3.0s
t=6: FIRING: http_requests_total=230 exceeded 100 for 4.0s
t=7: FIRING: http_requests_total=260 exceeded 100 for 5.0s
How it works
The MetricsStore acts as a minimal stand-in for Prometheus storage, holding the latest metric value under a (name, labels) key. The PrometheusAlertRule.evaluate method tracks when the threshold was first crossed via firing_since, and only reports FIRING after the sustained duration elapses. Time is injected as a parameter so tests can control the clock without real sleeps. When the value drops back under the threshold, firing_since resets to None, mimicking Prometheus's pending-to-firing transition. This design keeps the alert logic pure and unit-testable.
Common mistakes
- Comparing with >= instead of > for the threshold, which can cause edge-case firing differences
- Using real `time.sleep` calls that slow tests down instead of injecting a fake clock
- Forgetting to reset `firing_since` when the metric recovers, causing stale alerts
- Ignoring labels — the same metric name with different labels has different values in real Prometheus
Variations
- Use `time.monotonic()` instead of `time.time()` to avoid clock adjustments during evaluation
- Wrap the rule in a loop that integrates with the real `prometheus_client` library for live exports
Real-world use cases
- Unit-testing alerting logic in an SRE tool before deploying the rule to a real Prometheus instance.
- Simulating load spikes in a staging environment to verify dashboard notifications fire at the correct threshold.
- Teaching the pending/firing semantics of Prometheus alerts in an internal workshop or on-call training.
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.