S079 · SLIs, SLOs & Error Budgets — the SRE Math

#!/usr/bin/env python3
"""slo_simulator.py — walk through the SLO math with live traffic.
 
Run: python slo_simulator.py
 
Simulates:
  - a service serving 100 rps with 0.05% baseline error rate
  - an incident partway through that spikes error rate to 5%
  - real-time SLI computation over a 30-day-equivalent window
  - error budget consumption
  - multi-burn-rate alerts (fast + slow)
"""
from __future__ import annotations
 
import random
import time
from collections import deque
from dataclasses import dataclass
 
 
@dataclass
class SLOConfig:
    target: float = 0.999  # 99.9%
    window_seconds: int = 30 * 24 * 3600  # 30 days
    # Fast burn: consume 2% of budget in 1 hour = 14.4x normal burn rate
    fast_burn_multiplier: float = 14.4
    fast_burn_window: int = 3600  # 1 hour
    # Slow burn: consume 10% of budget in 6 hours = 6x normal
    slow_burn_multiplier: float = 6.0
    slow_burn_window: int = 6 * 3600
 
 
class SLOTracker:
    def __init__(self, cfg: SLOConfig):
        self.cfg = cfg
        # Retain a rolling window of (timestamp, is_good) samples
        self.events: deque[tuple[float, bool]] = deque()
        # Rolling counts for fast/slow burn windows
        self.recent_1h: deque[tuple[float, bool]] = deque()
        self.recent_6h: deque[tuple[float, bool]] = deque()
 
    def observe(self, is_good: bool, now: float | None = None) -> None:
        now = now or time.time()
        self.events.append((now, is_good))
        self.recent_1h.append((now, is_good))
        self.recent_6h.append((now, is_good))
        # Prune each window
        cutoff = now - self.cfg.window_seconds
        while self.events and self.events[0][0] < cutoff:
            self.events.popleft()
        while self.recent_1h and self.recent_1h[0][0] < now - self.cfg.fast_burn_window:
            self.recent_1h.popleft()
        while self.recent_6h and self.recent_6h[0][0] < now - self.cfg.slow_burn_window:
            self.recent_6h.popleft()
 
    @staticmethod
    def _bad_ratio(events: deque[tuple[float, bool]]) -> float:
        if not events:
            return 0.0
        bad = sum(1 for _, g in events if not g)
        return bad / len(events)
 
    def sli(self) -> float:
        return 1.0 - self._bad_ratio(self.events)
 
    def error_budget_remaining(self) -> float:
        """Fraction of budget still available (0-1)."""
        allowed_bad = 1.0 - self.cfg.target
        actual_bad = self._bad_ratio(self.events)
        if allowed_bad == 0:
            return 0.0
        remaining = 1.0 - (actual_bad / allowed_bad)
        return max(0.0, remaining)
 
    def burn_rate(self, window_events: deque[tuple[float, bool]]) -> float:
        """How fast are we consuming budget, expressed as multiplier of nominal.
        nominal = SLO_error_ratio (e.g. 0.001 for 99.9%)."""
        allowed_bad = 1.0 - self.cfg.target
        actual_bad = self._bad_ratio(window_events)
        if allowed_bad == 0:
            return float("inf") if actual_bad > 0 else 0.0
        return actual_bad / allowed_bad
 
    def alerts(self) -> list[str]:
        alerts = []
        fast = self.burn_rate(self.recent_1h)
        slow = self.burn_rate(self.recent_6h)
        if fast >= self.cfg.fast_burn_multiplier:
            alerts.append(f"PAGE · fast burn {fast:.1f}x over 1h "
                         f"(threshold {self.cfg.fast_burn_multiplier}x)")
        if slow >= self.cfg.slow_burn_multiplier:
            alerts.append(f"TICKET · slow burn {slow:.1f}x over 6h "
                         f"(threshold {self.cfg.slow_burn_multiplier}x)")
        return alerts
 
 
def simulate() -> None:
    cfg = SLOConfig(target=0.999)
    tracker = SLOTracker(cfg)
 
    print(f"SLO target: {cfg.target*100:.2f}%")
    print(f"Error budget: {(1-cfg.target)*100:.3f}% "
          f"= {(1-cfg.target) * 30 * 24 * 60:.1f} min/month full downtime")
    print(f"Fast burn threshold: {cfg.fast_burn_multiplier}x over 1h")
    print(f"Slow burn threshold: {cfg.slow_burn_multiplier}x over 6h\n")
 
    # Simulate 8 hours of traffic. Fast-forward: 1 real second = 60 sim minutes.
    RPS = 100
    SPEEDUP = 3600  # 1 real sec = 1 sim hour
    start_real = time.time()
    now_sim = start_real  # timestamps in sim time
 
    incident_start = start_real + 3.0  # sim: hour 3
    incident_end = start_real + 3.5    # sim: hour 3.5
 
    for hour in range(8):
        # Serve one hour of RPS traffic in a burst
        error_rate = 0.0005  # baseline 0.05%
        real_now = time.time()
        if incident_start <= real_now <= incident_end:
            error_rate = 0.05  # 5% during incident
            print(f"[hour {hour}] INCIDENT — error rate 5%")
 
        for _ in range(RPS * 3600):  # requests in this hour
            is_good = random.random() > error_rate
            tracker.observe(is_good, now=now_sim)
            now_sim += 1.0 / RPS
 
        sli = tracker.sli() * 100
        budget = tracker.error_budget_remaining() * 100
        alerts = tracker.alerts()
        alert_str = " | ".join(alerts) if alerts else "healthy"
        print(f"[hour {hour}] SLI={sli:.3f}%  budget_remaining={budget:5.1f}%  "
              f"{alert_str}")
        # Advance real time to simulate an hour
        time.sleep(1.0)
 
    print("\nSimulation complete.")
    print(f"Final SLI: {tracker.sli()*100:.3f}%")
    print(f"Final budget remaining: {tracker.error_budget_remaining()*100:.1f}%")
 
 
if __name__ == "__main__":
    simulate()