Configuring SLA Alerts for Serialization Pipelines

An SLA that only exists in a compliance document does not stop a bad deployment from silently missing verification response windows for six hours over a weekend. This guide turns the operational commitments behind Real-Time Monitoring & Alerting — itself part of Serialization Data Ingestion & EPCIS Event Sync — into concrete, code-defined alert rules: what to measure, when a breach counts as a breach, who gets paged, and how to stop the same incident from paging that person forty times before anyone acknowledges it.

The core design problem is hysteresis. A naive alert that fires the instant a metric crosses a threshold and clears the instant it drops back below it will flap continuously when a metric oscillates around the boundary — every flap is a fresh notification, and fresh notifications during an active incident are how on-call engineers learn to ignore alerts entirely. The diagram below shows the state machine this guide implements: escalation and recovery use different thresholds, separated by a deliberate gap, and each transition requires several consecutive confirming samples before it fires.

Prerequisites #

• Python 3.10+ — the evaluator uses X | Y union return types and dataclasses with default values.
• Pydantic v2 for the SLO rule definitions — field_validator-style typed config keeps threshold ordering and hysteresis ratios from being silently misconfigured.
• An async metric source — a snapshot feed of gauge values (ingestion lag, DLQ depth, verification latency percentiles, uncommissioned-EPC counts), typically the same aggregation layer behind Building EPCIS Ingestion Dashboards with Python.
• A notification transport — Slack webhook, PagerDuty Events API v2, or both, addressable by severity.
• DSCSA data prerequisites — a repository that can report, per polling interval, the count of ObjectEvent commissioning records missing for EPCs already observed downstream, plus current verification-router p99 latency and dead-letter queue depth.

Before writing rules, agree on units and polling cadence with the team that owns the metric source — a “lag” gauge sampled every 30 seconds behaves very differently under hysteresis math than one sampled every 5 minutes, and the window_breaches counts below assume a roughly constant sampling interval.

Step-by-Step Solution #

Step 1 — Define SLOs as typed rules, not scattered constants #

Every alert rule needs a warning threshold, a critical threshold, and a recovery ratio that opens the hysteresis gap below both. Encoding this as a Pydantic model means a rule with an inverted threshold order, or a recovery ratio outside (0, 1), fails at load time instead of at 3 a.m.

from __future__ import annotations
from enum import Enum
from pydantic import BaseModel, Field, model_validator


class Severity(str, Enum):
    OK = "ok"
    WARNING = "warning"
    CRITICAL = "critical"


class SLORule(BaseModel):
    name: str
    metric: str                       # key into the metric snapshot dict
    warn_threshold: float
    critical_threshold: float
    recovery_ratio: float = Field(0.75, gt=0.0, lt=1.0)
    window_breaches: int = Field(3, ge=1)
    runbook_url: str

    @model_validator(mode="after")
    def _thresholds_ordered(self) -> "SLORule":
        if self.critical_threshold <= self.warn_threshold:
            raise ValueError("critical_threshold must exceed warn_threshold")
        return self


SLO_RULES: list[SLORule] = [
    SLORule(name="verification_response_latency", metric="verification_p99_ms",
             warn_threshold=800.0, critical_threshold=1500.0,
             runbook_url="https://runbooks.internal/vrs-latency"),
    SLORule(name="ingestion_lag_ceiling", metric="ingestion_lag_seconds",
             warn_threshold=90.0, critical_threshold=180.0,
             runbook_url="https://runbooks.internal/ingestion-lag"),
    SLORule(name="dlq_depth_threshold", metric="dlq_depth",
             warn_threshold=200.0, critical_threshold=500.0,
             runbook_url="https://runbooks.internal/dlq-depth"),
    SLORule(name="missing_commission_detection", metric="uncommissioned_epc_count",
             warn_threshold=5.0, critical_threshold=25.0,
             runbook_url="https://runbooks.internal/missing-commission"),
]

DSCSA/GS1 note: the missing_commission_detection rule exists because a serialized unit that reaches a downstream read point without a prior ObjectEvent commissioning record — one carrying GTIN (01), serial (21), lot (10), and expiry (17) — will fail its first verification lookup. Catching the gap here, upstream of the Verification Router Service Architecture, is cheaper than explaining a false “not verified” response to a dispenser.

Step 2 — Choose burn-rate math where an error budget exists, thresholds where it doesn’t #

Verification latency has a natural error budget: “99% of verification responses under 800ms this month.” A single slow response barely dents that budget; a sustained regression burns it fast. Model that with a burn-rate helper rather than a flat threshold count:

def burn_rate(breaching_checks: int, window_checks: int, budget_fraction: float) -> float:
    """Fraction of the SLO's error budget this window's breach rate would consume.

    A burn_rate of 1.0 means the current breach rate exactly exhausts the
    budget over the SLO's full compliance period; > 2.0 is a fast burn that
    should page immediately rather than wait for window_breaches to accrue.
    """
    if window_checks == 0:
        return 0.0
    observed_bad_fraction = breaching_checks / window_checks
    return observed_bad_fraction / budget_fraction if budget_fraction else float("inf")

Ingestion lag, DLQ depth, and missing-commission counts do not have a monthly budget in the same sense — they are operational ceilings, not availability targets — so the threshold-plus-hysteresis evaluator in Step 3 handles them directly. Reserve burn_rate for latency-style SLOs where you can express “percentage of good events” cleanly; forcing every metric through burn-rate math adds a layer of abstraction that a hard ceiling like DLQ depth does not need.

DSCSA/GS1 note: a fast latency burn is treated as critical immediately in most on-call policies, because a verification-router regression risks breaching the FDA-mandated bounded response window for suspect-product and saleable-return queries, not just an internal dashboard metric.

Step 3 — Build the async, hysteresis-gated alert evaluator #

This is the piece that actually stops flapping. It tracks per-rule state across calls, only emits an Alert on a state transition, and requires window_breaches consecutive confirming samples in either direction before it commits to a new severity.

import asyncio
from dataclasses import dataclass
from datetime import datetime, timezone


@dataclass
class RuleState:
    severity: Severity = Severity.OK
    consecutive_up: int = 0
    consecutive_down: int = 0


@dataclass
class Alert:
    rule: str
    severity: Severity
    metric: str
    value: float
    threshold: float
    runbook_url: str
    fired_at: datetime
    dedup_key: str


_RANK = {Severity.OK: 0, Severity.WARNING: 1, Severity.CRITICAL: 2}


class AlertEvaluator:
    """Evaluates metric snapshots against SLO rules with hysteresis, so a
    single noisy sample never flips severity back and forth."""

    def __init__(self, rules: list[SLORule]) -> None:
        self._rules = {r.name: r for r in rules}
        self._state = {r.name: RuleState() for r in rules}

    async def evaluate(self, snapshot: dict[str, float]) -> list[Alert]:
        fired: list[Alert] = []
        for rule in self._rules.values():
            value = snapshot.get(rule.metric)
            if value is None:
                continue  # missing metric is a monitoring gap, not a breach
            alert = self._step(rule, value)
            if alert is not None:
                fired.append(alert)
        return fired

    def _step(self, rule: SLORule, value: float) -> Alert | None:
        state = self._state[rule.name]
        recover_warn = rule.warn_threshold * rule.recovery_ratio
        recover_critical = rule.critical_threshold * rule.recovery_ratio

        target = state.severity
        if value >= rule.critical_threshold:
            target = Severity.CRITICAL
        elif value >= rule.warn_threshold:
            target = Severity.WARNING
        elif state.severity is Severity.CRITICAL and value < recover_critical:
            target = Severity.WARNING
        elif state.severity is Severity.WARNING and value < recover_warn:
            target = Severity.OK

        if target == state.severity:
            state.consecutive_up = 0
            state.consecutive_down = 0
            return None

        escalating = _RANK[target] > _RANK[state.severity]
        if escalating:
            state.consecutive_up += 1
            state.consecutive_down = 0
            confirmed = state.consecutive_up
        else:
            state.consecutive_down += 1
            state.consecutive_up = 0
            confirmed = state.consecutive_down

        if confirmed < rule.window_breaches:
            return None  # not enough consecutive samples yet — hold current severity

        state.severity = target
        state.consecutive_up = 0
        state.consecutive_down = 0
        threshold = rule.critical_threshold if target is Severity.CRITICAL else rule.warn_threshold
        return Alert(
            rule=rule.name, severity=target, metric=rule.metric, value=value,
            threshold=threshold, runbook_url=rule.runbook_url,
            fired_at=datetime.now(timezone.utc),
            dedup_key=f"{rule.name}:{target.value}",
        )


async def run_evaluator(evaluator: AlertEvaluator, metric_snapshots, alert_sink) -> None:
    """Consume an async stream of metric snapshots and dispatch fired alerts."""
    async for snapshot in metric_snapshots:
        for alert in await evaluator.evaluate(snapshot):
            await alert_sink.dispatch(alert)

DSCSA/GS1 note: requiring window_breaches consecutive samples before escalating the dlq_depth_threshold rule stops a momentary batch-commit spike from paging on-call for a queue that drains itself within the next polling interval — the same discipline described for triaging entries in Draining Dead-Letter Queues Safely.

Step 4 — Route by severity and de-duplicate to protect the on-call rotation #

An AlertEvaluator only fires on transitions, which already kills most flapping. The remaining fatigue risk is a different one: several rules tripping at once during a single upstream incident, each independently valid but collectively overwhelming. Route by severity and add a cool-down keyed on the same dedup_key:

from datetime import timedelta

ROUTES: dict[Severity, list[str]] = {
    Severity.WARNING: ["slack:#serialization-alerts"],
    Severity.CRITICAL: ["pagerduty:on-call-serialization", "slack:#serialization-alerts"],
}


class DeduplicatingSink:
    """Suppresses repeat notifications for the same rule+severity within a
    cool-down window, so a rule that re-confirms every polling cycle does
    not re-notify every polling cycle."""

    def __init__(self, cooldown: timedelta = timedelta(minutes=15)) -> None:
        self._cooldown = cooldown
        self._last_sent: dict[str, datetime] = {}

    async def dispatch(self, alert: Alert) -> None:
        last = self._last_sent.get(alert.dedup_key)
        if last is not None and alert.fired_at - last < self._cooldown:
            return
        self._last_sent[alert.dedup_key] = alert.fired_at
        for route in ROUTES.get(alert.severity, []):
            await _send(route, alert)


async def _send(route: str, alert: Alert) -> None:
    print(f"[{alert.severity.value.upper()}] {route} :: {alert.rule} "
          f"= {alert.value} (threshold {alert.threshold}) -> {alert.runbook_url}")

DSCSA/GS1 note: because Alert objects only exist for state transitions, the cool-down window here is a second, independent line of defense — it protects against a transition being re-emitted by a redeployed evaluator instance, not against the flapping the hysteresis logic already handles.

Step 5 — Attach the on-call runbook link at the point of dispatch #

An alert without a runbook link sends someone searching a wiki while a packaging line waits. Because runbook_url already lives on the SLORule and flows through to the Alert, the notification payload can carry it as a first-class field rather than free text:

def to_pagerduty_payload(alert: Alert, routing_key: str) -> dict[str, object]:
    return {
        "routing_key": routing_key,
        "event_action": "trigger",
        "dedup_key": alert.dedup_key,
        "payload": {
            "summary": f"{alert.rule} breached ({alert.severity.value})",
            "severity": alert.severity.value,
            "source": "serialization-ingestion-monitor",
            "custom_details": {
                "metric": alert.metric,
                "value": alert.value,
                "threshold": alert.threshold,
                "runbook": alert.runbook_url,
            },
        },
    }

DSCSA/GS1 note: keep the runbook per-rule rather than per-severity — the remediation for a stalled ingestion_lag_ceiling (restart the consumer group) is unrelated to the remediation for missing_commission_detection (check the packaging-line PLC feed for a dropped commissioning post), and merging them into one generic runbook is how on-call engineers learn to skip reading it.

Verification #

Confirm the hysteresis logic actually holds before wiring it to a pager. A synchronous test using asyncio.run around the evaluator is enough to prove both directions without a live metric feed:

import asyncio

def test_hysteresis_prevents_single_spike_flap():
    rule = SLORule(
        name="ingestion_lag_ceiling", metric="ingestion_lag_seconds",
        warn_threshold=90.0, critical_threshold=180.0, window_breaches=2,
        runbook_url="https://runbooks.internal/ingestion-lag",
    )
    evaluator = AlertEvaluator([rule])

    async def scenario() -> list[list[Alert]]:
        lags = (95.0, 95.0, 40.0, 40.0)
        return [await evaluator.evaluate({"ingestion_lag_seconds": lag}) for lag in lags]

    first_spike, second_spike, first_drop, second_drop = asyncio.run(scenario())
    assert first_spike == []                                    # one sample: not enough
    assert len(second_spike) == 1
    assert second_spike[0].severity == Severity.WARNING           # two consecutive: fires
    assert first_drop == []                                       # one recovery sample: holds
    assert len(second_drop) == 1
    assert second_drop[0].severity == Severity.OK                 # two consecutive: clears

Beyond the unit test, run the evaluator against a replayed window of production metric snapshots from an incident that is known to have flapped under the old naive threshold logic, and confirm the new state machine collapses that incident into a single escalation and a single recovery notification. Finally, inspect the DeduplicatingSink’s dispatch log for a synthetic burst (fire all four SLO_RULES at once) and confirm exactly one notification per dedup_key lands inside the cool-down window, not one per rule per polling tick.

Gotchas & Edge Cases #

• Absent metrics are not zero. A metric source that goes silent should never resolve to 0.0 and be read as “healthy” — the evaluator here treats a missing key as a monitoring gap and skips the rule, which is a deliberate choice: alert on the gap itself with a separate staleness check rather than silently clearing severity.
• A recovery ratio too close to 1.0 defeats the whole design. If recovery_ratio is 0.98, the gap between trigger and recovery thresholds is a rounding error, and the state machine behaves almost exactly like the naive single-threshold version it was built to replace.
• Cool-down math needs timezone-aware timestamps end to end. Mixing a naive datetime.now() with the timezone-aware fired_at produced above raises at comparison time; keep every timestamp in this pipeline UTC-aware, the same discipline EPCIS eventTime handling requires.
• Dedup key granularity hides multi-partner incidents. A dedup_key of rule:severity alone will merge a DLQ-depth breach from one trading partner with an unrelated one from another. If alerts need to be actionable per trading partner or read point, fold the GLN or partition key into the dedup key before it reaches the sink.
• Threshold values drift out of sync with reality. A warn_threshold tuned for a launch-week traffic level becomes noise-generating six months later at steady-state volume; review SLO_RULES on the same cadence as the SLA itself, not only when someone complains about a pager storm.

Related #

• Up to the parent topic: Real-Time Monitoring & Alerting
• Building EPCIS Ingestion Dashboards with Python — the metric snapshots this evaluator consumes
• Draining Dead-Letter Queues Safely — the remediation path the DLQ-depth alert should link to
• Serialization Data Ingestion & EPCIS Event Sync — the pipeline these SLAs govern