"""SLO (Service Level Objective) targets for scheduled jobs. Each registered handler can declare an ``SLOTarget`` describing the production expectation it commits to: * ``success_rate_floor`` — minimum acceptable ok/total ratio over the evaluation window (e.g. 0.99 for "less than 1% failure"). * ``latency_p95_ms_ceiling`` — p95 duration ceiling. None = unbounded. * ``max_consecutive_failures`` — most recent N runs all ``failed`` is a hard breach regardless of overall rate (catches a freshly broken job before it drags down the window average). * ``tier`` — operational severity: ``P0`` (page immediately, cc management), ``P1`` (notify IT admins, dedup signature), ``P2`` (notify IT admins, lower urgency). The tier drives ``failure_notifier`` card emoji + recipient fan-out (P0 also goes to ``management``), and the Prometheus exposition. Handlers without an explicit ``slo=`` parameter fall through to ``DEFAULT_SLO`` (P2, success_rate >= 0.80, no latency ceiling, max 3 consecutive failures) so every cron has *some* SLO line — the bar is just lower until an owner declares stricter targets. """ from __future__ import annotations from dataclasses import dataclass from typing import Literal # Operational severity tier — drives failure-notifier card emoji, # recipient fan-out, and the SLO panel color. SLOTier = Literal["P0", "P1", "P2"] @dataclass(frozen=True, slots=True) class SLOTarget: """One handler's SLO commitment. Immutable + frozen so it's safe to share across cron ticks. """ tier: SLOTier = "P2" success_rate_floor: float = 0.80 latency_p95_ms_ceiling: int | None = None max_consecutive_failures: int = 3 def __post_init__(self) -> None: if not (0.0 <= self.success_rate_floor <= 1.0): raise ValueError( f"success_rate_floor must be in [0.0, 1.0]; got " f"{self.success_rate_floor}" ) if self.max_consecutive_failures < 1: raise ValueError( f"max_consecutive_failures must be >= 1; got " f"{self.max_consecutive_failures}" ) if self.latency_p95_ms_ceiling is not None and self.latency_p95_ms_ceiling <= 0: raise ValueError( f"latency_p95_ms_ceiling must be > 0 when set; got " f"{self.latency_p95_ms_ceiling}" ) if self.tier not in ("P0", "P1", "P2"): raise ValueError(f"tier must be P0/P1/P2; got {self.tier!r}") # Default — every registered cron gets *some* SLO line. Owners can # tighten by passing an explicit ``slo=SLOTarget(...)`` to the # decorator. DEFAULT_SLO = SLOTarget( tier="P2", success_rate_floor=0.80, latency_p95_ms_ceiling=None, max_consecutive_failures=3, ) # Pre-defined SLOs for the cron handlers whose breach materially hurts # the platform. Imported by the handler module at registration time. # T+0 data-freshness driver — every 15 min, must succeed nearly always. # Two consecutive failures = data is now visibly stale to IT, page. SLO_PROCESS_ADVANCE = SLOTarget( tier="P0", success_rate_floor=0.99, latency_p95_ms_ceiling=120_000, # 2 minutes max_consecutive_failures=2, ) # Notification retry — the platform's last-mile to humans. Breach = # users stop seeing alerts. SLO_NOTIFICATION_OUTBOX_RETRY = SLOTarget( tier="P1", success_rate_floor=0.95, latency_p95_ms_ceiling=60_000, max_consecutive_failures=3, ) # Daily ops digest — non-urgent but if it fails for days admins are # flying blind. SLO_DAILY_DIGEST = SLOTarget( tier="P1", success_rate_floor=0.95, latency_p95_ms_ceiling=300_000, # 5 minutes max_consecutive_failures=2, ) # HRIS overdue check — drives the auto-flip to `left`. Soft tier # because the gate is already conservative. SLO_OVERDUE_LEAVERS_CHECK = SLOTarget( tier="P1", success_rate_floor=0.95, latency_p95_ms_ceiling=120_000, max_consecutive_failures=3, ) # Cloud cost facts sync — daily, expensive, often partial. Wider # tolerance. SLO_CMDB_COST_FACTS_SYNC = SLOTarget( tier="P2", success_rate_floor=0.85, latency_p95_ms_ceiling=15 * 60_000, # 15 minutes max_consecutive_failures=2, ) # Leaver escalation tick — hourly, MUST keep firing. SLO_LEAVER_ESCALATION_TICK = SLOTarget( tier="P0", success_rate_floor=0.99, latency_p95_ms_ceiling=60_000, max_consecutive_failures=2, ) __all__ = [ "DEFAULT_SLO", "SLO_CMDB_COST_FACTS_SYNC", "SLO_DAILY_DIGEST", "SLO_LEAVER_ESCALATION_TICK", "SLO_NOTIFICATION_OUTBOX_RETRY", "SLO_OVERDUE_LEAVERS_CHECK", "SLO_PROCESS_ADVANCE", "SLOTarget", "SLOTier", ]