Monitoring and Production Operations

What we're doing: Show the shape of an alert rule as executable Python — a threshold-over-a-window check a monitoring system runs on every new metric sample, independent of which vendor evaluates it.

def should_page(error_rate_samples: list[float], threshold: float = 0.05) -> bool:
    """True only if EVERY recent sample breached the threshold — avoids
    paging on one noisy data point, the same 'FOR 5m' idea alerting
    tools express as a rule."""
    if not error_rate_samples:
        return False
    return all(sample > threshold for sample in error_rate_samples)


noisy_spike = [0.01, 0.09, 0.02, 0.01]        # one bad sample among good ones
sustained_breach = [0.08, 0.07, 0.09, 0.11]   # every recent sample is bad

print("noisy spike pages:     ", should_page(noisy_spike))
print("sustained breach pages:", should_page(sustained_breach))

1

error_rate_samples is a sliding window of recent metric values — the "FOR 5m" duration an alert rule requires before it fires.

6

all(...) means one good sample in the window keeps the alert quiet — this is what filters a single blip from a real, sustained breach.

noisy spike pages:      False
sustained breach pages: True

Why this works: A single elevated sample does not page — all() requires every sample in the window to have crossed the threshold, mirroring the "FOR 5m" duration clause every real alerting tool (Alertmanager, CloudWatch, Grafana) attaches to a threshold so a one-off blip cannot trigger a page.

What we're doing: Check a real SLI against an SLO and report how much error budget is left — the exact question "how will you know the service is broken" reduces to.

def check_slo(successful: int, total: int, slo_target: float = 0.999) -> dict:
    sli = successful / total
    error_budget = 1 - slo_target        # e.g. 0.1% allowed to fail
    actual_failure = 1 - sli
    budget_remaining = error_budget - actual_failure
    return {
        "sli": round(sli, 5),
        "slo_target": slo_target,
        "budget_remaining_pct": round(budget_remaining * 100, 4),
        "breached": budget_remaining < 0,
    }


healthy = check_slo(successful=999_600, total=1_000_000)
breached = check_slo(successful=996_000, total=1_000_000)
print("healthy: ", healthy)
print("breached:", breached)

2

sli is the measured, real number — successes divided by total requests over the window.

3

error_budget is fixed by the SLO target alone (1 − 0.999 = 0.001), not by what actually happened.

5

budget_remaining goes negative exactly when actual failure exceeds what the error budget allowed — that IS the SLO breach.

healthy:  {'sli': 0.9996, 'slo_target': 0.999, 'budget_remaining_pct': 0.06, 'breached': False}
breached: {'sli': 0.996, 'slo_target': 0.999, 'budget_remaining_pct': -0.3, 'breached': True}

Why this works: budget_remaining is exactly error_budget minus how much actually failed — positive means the team has room left at the current SLO target, negative means the SLO has already been breached for the window. This is the concrete form of 'how will you know the service is broken': not a single alert firing, but the error budget going negative.

What we're doing: Show mitigation strictly ordered before full root-cause investigation, and root cause analysis modeled as contributing factors rather than one cause.

def next_incident_action(mitigated: bool, root_cause_known: bool) -> str:
    if not mitigated:
        return "mitigate now (rollback / flag off / shed traffic)"
    if not root_cause_known:
        return "investigate root cause (Five Whys / contributing factors)"
    return "write the postmortem"


def five_whys(symptom: str, answers: list[str]) -> dict:
    """Models root cause analysis as a chain of contributing factors,
    not a single root cause — the SRE book's own framing."""
    chain = [symptom, *answers]
    return {"chain": chain, "contributing_factors": answers, "depth": len(answers)}


print(next_incident_action(mitigated=False, root_cause_known=False))
print(next_incident_action(mitigated=True, root_cause_known=False))
print(next_incident_action(mitigated=True, root_cause_known=True))
print(five_whys(
    "Checkout returned 500s",
    ["DB connection pool exhausted",
     "a slow query held connections open",
     "an index was dropped in last week's migration",
     "the migration review process has no query-plan check"],
))

2

Mitigation is checked FIRST — the function refuses to suggest root-cause work while users are still impacted.

9

five_whys returns the whole chain of contributing_factors, not a single "root_cause" field — matching the SRE book's framing.

mitigate now (rollback / flag off / shed traffic)
investigate root cause (Five Whys / contributing factors)
write the postmortem
{'chain': ['Checkout returned 500s', 'DB connection pool exhausted', 'a slow query held connections open', "an index was dropped in last week's migration", 'the migration review process has no query-plan check'], 'contributing_factors': ['DB connection pool exhausted', 'a slow query held connections open', "an index was dropped in last week's migration", 'the migration review process has no query-plan check'], 'depth': 4}

Why this works: next_incident_action never recommends root-cause work before mitigation is done — that ordering is the incident-response discipline itself. five_whys stops at the last answer given (here, a process gap: no query-plan check) rather than a person, which is what keeps root cause analysis about fixable systemic conditions instead of blame.

What we're doing: Model the postmortem's action-item follow-through as data — the part most likely to silently not happen without an owner and a status.

def open_action_items(actions: list[dict]) -> list[dict]:
    """Filters a postmortem's action items down to what's still open —
    the concrete signal a postmortem produced real follow-through."""
    return [a for a in actions if a["status"] != "done"]


actions = [
    {"item": "Add a query-plan check to migration review", "owner": "dana", "status": "done"},
    {"item": "Alert on connection-pool saturation, not just errors", "owner": "priya", "status": "open"},
    {"item": "Document the rollback runbook for checkout", "owner": "priya", "status": "open"},
]

print(open_action_items(actions))

4

status != "done" is the whole check — a postmortem with items stuck at "open" for months is a documented outage that has not actually been prevented from repeating.

[{'item': 'Alert on connection-pool saturation, not just errors', 'owner': 'priya', 'status': 'open'}, {'item': 'Document the rollback runbook for checkout', 'owner': 'priya', 'status': 'open'}]

Why this works: The filter surfaces exactly the two items that still need work — a postmortem that never gets checked back against this list produces a document, not a fix, and the same incident is free to happen again.

What we're doing: Build a small circuit breaker that stops calling a failing dependency after 3 consecutive failures, instead of retrying (and timing out) forever.

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3):
        self.failure_threshold = failure_threshold
        self.failure_count = 0
        self.open = False

    def call(self, func, *args):
        if self.open:
            raise RuntimeError("circuit open - skipping call, using fallback")
        try:
            result = func(*args)
        except Exception:
            self.failure_count += 1
            if self.failure_count >= self.failure_threshold:
                self.open = True
            raise
        else:
            self.failure_count = 0
            return result


def flaky_payment_check(_order_id: str) -> str:
    raise TimeoutError("payment gateway unreachable")


breaker = CircuitBreaker(failure_threshold=3)
for attempt in range(1, 5):
    try:
        breaker.call(flaky_payment_check, "order_1")
    except TimeoutError:
        print(f"attempt {attempt}: TimeoutError, failure_count={breaker.failure_count}, open={breaker.open}")
    except RuntimeError as e:
        print(f"attempt {attempt}: {e}")

8

Once open is True, call() refuses to even try the dependency — this is what stops piling up slow, timing-out calls.

13

failure_count only increments on a real failure, and open flips True once it reaches the threshold — 3 consecutive failures here.

26

The 4th attempt never reaches flaky_payment_check at all — the breaker fails fast with RuntimeError instead of another slow TimeoutError.

attempt 1: TimeoutError, failure_count=1, open=False
attempt 2: TimeoutError, failure_count=2, open=False
attempt 3: TimeoutError, failure_count=3, open=True
attempt 4: circuit open - skipping call, using fallback

Why this works: The first three attempts genuinely call flaky_payment_check and pay its full timeout cost each time; failure_count climbs to the threshold on attempt 3 and flips open to True. The 4th attempt never touches the dependency — call() raises immediately, which is the entire point: stop paying the cost of calling something that is already known to be down.

What we're doing: Show an automated rollback trigger: the deployed version reverts on its own once the error rate crosses a threshold, without anyone deciding it by hand.

def error_rate_breached(errors: int, total: int, threshold: float = 0.05) -> bool:
    return total > 0 and (errors / total) > threshold


def choose_active_version(current: str, previous: str, errors: int, total: int) -> str:
    if error_rate_breached(errors, total):
        return previous   # roll back
    return current


print("healthy   :", choose_active_version("v2.4.0", "v2.3.1", errors=2, total=1000))
print("unhealthy :", choose_active_version("v2.4.0", "v2.3.1", errors=80, total=1000))

1

error_rate_breached is the same threshold-check shape as an alert rule (previous section concept) — rollback triggers and alerts often share logic.

6

Breaching the threshold returns previous, not current — the deploy system would read this as "route traffic back to the last known-good version".

healthy   : v2.4.0
unhealthy : v2.3.1

Why this works: With a 2/1000 (0.2%) error rate, choose_active_version keeps the new version — well under the 5% threshold. With 80/1000 (8%), it returns the previous version instead, the same decision an automated canary-rollback system makes without a human in the loop.