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