impl/python.py

from typing import List, Optional, Sequence

from .monitor_check_status_types import Check, CheckStatus
from .monitor_incidents_types import Incident


def find_incidents(checks: Sequence[Check], incident_status: CheckStatus, confirm_checks: int, now: int) -> List[Incident]:
    """The incidents in a check series. A run of consecutive bad checks is
    one incident, from its first bad check to the first good check after it,
    but only when it holds at least confirm_checks checks: a single failed
    probe is noise, not an outage, and would otherwise page and skew MTTR."""
    if not isinstance(now, int) or isinstance(now, bool):
        raise ValueError("now must be a whole number of seconds")
    if incident_status not in ("down", "degraded"):
        raise ValueError("incidentStatus must be down or degraded, received %s" % (incident_status,))
    if not isinstance(confirm_checks, int) or isinstance(confirm_checks, bool) or confirm_checks < 1:
        raise ValueError("confirmChecks must be at least 1, received %r" % (confirm_checks,))
    for i, c in enumerate(checks):
        if c.status not in ("up", "degraded", "down"):
            raise ValueError("unknown check status: %s" % (c.status,))
        if i > 0 and c.at <= checks[i - 1].at:
            raise ValueError("checks must be in strictly ascending time order: %d follows %d" % (c.at, checks[i - 1].at))
    if len(checks) > 0 and now < checks[-1].at:
        raise ValueError("now %d is before the last check at %d" % (now, checks[-1].at))

    def bad(s: str) -> bool:
        return s == "down" or (incident_status == "degraded" and s == "degraded")

    out: List[Incident] = []
    i = 0
    n = len(checks)
    while i < n:
        if not bad(checks[i].status):
            i += 1
            continue
        start = checks[i].at
        count = 0
        worst = "degraded"
        while i < n and bad(checks[i].status):
            if checks[i].status == "down":
                worst = "down"
            count += 1
            i += 1
        end: Optional[int] = checks[i].at if i < n else None
        if count >= confirm_checks:
            out.append(Incident(start=start, end=end, duration_seconds=(now if end is None else end) - start, worst=worst, checks=count))
    return out