from typing import Optional from .monitor_alert_state_types import AlertPolicy, AlertState, AlertTransition def next_alert_state(previous: AlertState, condition: bool, now: int, policy: AlertPolicy) -> AlertTransition: """One step of an alert's lifecycle. Pure: the caller stores the returned state and passes it back at the next evaluation. Firing needs the condition to hold for_seconds (Prometheus `for`); resolving needs it to stay false clear_for_seconds, so a flapping check does not send resolved-firing-resolved storms. last_notified_at records the last notification of any kind; it drives the repeat interval while firing.""" if not isinstance(now, int) or isinstance(now, bool): raise ValueError("now must be a whole number of seconds") for_seconds = policy.for_seconds clear_for = policy.clear_for_seconds renotify = policy.renotify_seconds if for_seconds < 0: raise ValueError("forSeconds must not be negative, received %d" % (for_seconds,)) if clear_for < 0: raise ValueError("clearForSeconds must not be negative, received %d" % (clear_for,)) if renotify is not None and renotify < 1: raise ValueError("renotifySeconds must be null or at least 1, received %d" % (renotify,)) phase = previous.phase if phase not in ("inactive", "pending", "firing", "resolved"): raise ValueError("unknown alert phase: %s" % (phase,)) since = previous.since last = previous.last_notified_at clear_since = previous.clear_since if phase in ("pending", "firing") and since is None: raise ValueError("since must be set in phase %s" % (phase,)) for name, t in (("since", since), ("lastNotifiedAt", last), ("clearSince", clear_since)): if t is not None and now < t: raise ValueError("now %d is earlier than %s %d" % (now, name, t)) same = AlertState(phase=phase, since=since, last_notified_at=last, clear_since=clear_since) if phase in ("inactive", "resolved"): if not condition: return AlertTransition(state=same, notify=None, changed=False) if for_seconds == 0: return AlertTransition(state=AlertState(phase="firing", since=now, last_notified_at=now, clear_since=None), notify="firing", changed=True) return AlertTransition(state=AlertState(phase="pending", since=now, last_notified_at=last, clear_since=None), notify=None, changed=True) if phase == "pending": if not condition: return AlertTransition(state=AlertState(phase="inactive", since=now, last_notified_at=last, clear_since=None), notify=None, changed=True) assert since is not None if now - since >= for_seconds: return AlertTransition(state=AlertState(phase="firing", since=now, last_notified_at=now, clear_since=None), notify="firing", changed=True) return AlertTransition(state=same, notify=None, changed=False) # firing if condition: if renotify is not None and (last is None or now - last >= renotify): return AlertTransition(state=AlertState(phase=phase, since=since, last_notified_at=now, clear_since=None), notify="repeat", changed=False) return AlertTransition(state=AlertState(phase=phase, since=since, last_notified_at=last, clear_since=None), notify=None, changed=False) clearing: Optional[int] = now if clear_since is None else clear_since if now - clearing >= clear_for: return AlertTransition(state=AlertState(phase="resolved", since=now, last_notified_at=now, clear_since=None), notify="resolved", changed=True) return AlertTransition(state=AlertState(phase=phase, since=since, last_notified_at=last, clear_since=clearing), notify=None, changed=False)