from typing import Dict, List, Optional, Sequence, Tuple from .monitor_burn_rate import burn_rate from .monitor_burn_rate_alert_data import SRE_WORKBOOK 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)