Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import Dict, List, Optional, Sequence, Tuple
from .monitor_burn_rate import burn_rate ← from monitor.burn-rate ^1.0.0 · built alongside by fune
from .monitor_burn_rate_alert_data import SRE_WORKBOOK ← this capability’s own data, compiled from data/sre-workbook.json into the same file by fune build
from .monitor_burn_rate_alert_types import BurnAlert, BurnRule, BurnRuleResult, WindowCount
def _whole(value: object) -> bool:
return isinstance(value, int) and not isinstance(value, bool)
def _at_or_above(w: WindowCount, threshold_milli: int, target_basis_points: int) -> bool:
# bad/total >= threshold/1000 x (10000 - target)/10000, cross-multiplied so a
# burn of 14.3995x never fires a 14.4x rule by rounding up.
if w.total_events == 0:
return False
return w.bad_events * 10000000 >= threshold_milli * w.total_events * (10000 - target_basis_points)
def burn_rate_alert(target_basis_points: int, windows: Sequence[WindowCount], rules: Optional[Sequence[BurnRule]], min_events: Optional[int]) -> BurnAlert:
"""Multiwindow, multi-burn-rate SLO alerting, as the Google SRE workbook
recommends: a rule fires only when both its long window (enough budget
spent to matter) and its short window (still happening now) burn at or
above its threshold. With rules None it uses the workbook's Table 5-8.
min_events guards small samples: a rule whose long window saw fewer
events stays quiet, since 2 bad checks out of 11 is a 36x burn of a
99.5% budget but proves little. The short window is not guarded: it
only confirms the burn is still going on."""
if not _whole(target_basis_points) or target_basis_points < 1 or target_basis_points > 9999:
raise ValueError("targetBasisPoints must be a whole number from 1 to 9999 (10000 leaves no error budget), received %s" % (target_basis_points,))
if min_events is not None and (not _whole(min_events) or min_events < 0):
raise ValueError("minEvents must be null or a whole number of at least 0, received %s" % (min_events,))
minimum = 0 if min_events is None else min_events
counts: Dict[int, Tuple[WindowCount, int]] = {}
for w in windows:
if not _whole(w.window_seconds) or w.window_seconds < 1:
raise ValueError("windowSeconds must be at least 1, received %s" % (w.window_seconds,))
if w.window_seconds in counts:
raise ValueError("duplicate counts for a %d-second window" % w.window_seconds)
# burn_rate checks the counts, so every window is checked, used or not.
counts[w.window_seconds] = (w, burn_rate(target_basis_points, w.total_events, w.bad_events))
chosen: Sequence[BurnRule] = rules if rules is not None else [
BurnRule(
severity=r.severity,
long_window_seconds=r.long_window_seconds,
short_window_seconds=r.short_window_seconds,
burn_rate_milli=r.burn_rate_milli,
)
for r in SRE_WORKBOOK
]
if len(chosen) == 0:
raise ValueError("rules must not be empty; pass null for the SRE workbook rules")
results: List[BurnRuleResult] = []
severity: Optional[str] = None
for rule in chosen:
if rule.severity == "":
raise ValueError("severity must not be empty")
if not _whole(rule.short_window_seconds) or rule.short_window_seconds < 1:
raise ValueError("shortWindowSeconds must be at least 1, received %s" % (rule.short_window_seconds,))
if not _whole(rule.long_window_seconds) or rule.short_window_seconds > rule.long_window_seconds:
raise ValueError("shortWindowSeconds must not exceed longWindowSeconds: %s > %s" % (rule.short_window_seconds, rule.long_window_seconds))
if not _whole(rule.burn_rate_milli) or rule.burn_rate_milli < 1:
raise ValueError("burnRateMilli must be at least 1, received %s" % (rule.burn_rate_milli,))
if rule.long_window_seconds not in counts:
raise ValueError("no counts for a %d-second window" % rule.long_window_seconds)
if rule.short_window_seconds not in counts:
raise ValueError("no counts for a %d-second window" % rule.short_window_seconds)
long_count, long_burn = counts[rule.long_window_seconds]
short_count, short_burn = counts[rule.short_window_seconds]
enough = long_count.total_events >= minimum
firing = enough and _at_or_above(long_count, rule.burn_rate_milli, target_basis_points) and _at_or_above(short_count, rule.burn_rate_milli, target_basis_points)
if firing and severity is None:
severity = rule.severity
results.append(BurnRuleResult(
severity=rule.severity,
long_window_seconds=rule.long_window_seconds,
short_window_seconds=rule.short_window_seconds,
threshold_milli=rule.burn_rate_milli,
long_burn_milli=long_burn,
short_burn_milli=short_burn,
enough_events=enough,
firing=firing,
))
return BurnAlert(firing=severity is not None, severity=severity, rules=results)