impl/python.py

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)