Functional Weave
Code in Rust

monitor.mttr@1.0.0

impl/python.py

2,444 bytes · the Python implementation · view raw

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Optional, Sequence

from .monitor_incidents_types import Incident
from .math_round_div import round_div  ← from math.round-div ^1.0.0 · built alongside by fune
from .monitor_mttr_types import ReliabilityStats


def reliability_stats(incidents: Sequence[Incident], from_: int, to: int) -> ReliabilityStats:
    """Reliability figures for the window from <= t < to. An incident counts
    when it overlaps the window; downtime is clipped to it, but MTTR and the
    longest incident use full durations, because an outage that began last
    month still took as long as it took to fix."""
    for v in (from_, to):
        if not isinstance(v, int) or isinstance(v, bool):
            raise ValueError("from and to must be whole seconds")
    if from_ > to:
        raise ValueError("from must not be after to: %d > %d" % (from_, to))
    prev_end: Optional[int] = None
    count = 0
    resolved = 0
    downtime = 0
    repair_total = 0
    longest = 0
    for x in incidents:
        if x.duration_seconds < 0:
            raise ValueError("durationSeconds must not be negative, received %d" % (x.duration_seconds,))
        if x.end is not None:
            if x.end < x.start:
                raise ValueError("incident end must not be before its start: %d < %d" % (x.end, x.start))
            if x.duration_seconds != x.end - x.start:
                raise ValueError("durationSeconds must equal end - start: %d != %d" % (x.duration_seconds, x.end - x.start))
        # An ongoing incident ends, for now, where its duration says.
        end = x.start + x.duration_seconds
        if prev_end is not None and x.start < prev_end:
            raise ValueError("incidents must be in time order and must not overlap: %d is before %d" % (x.start, prev_end))
        prev_end = end
        if not (x.start < to and (x.start >= from_ or end > from_)):
            continue
        count += 1
        downtime += max(0, min(end, to) - max(x.start, from_))
        if x.end is not None:
            resolved += 1
            repair_total += x.duration_seconds
        if x.duration_seconds > longest:
            longest = x.duration_seconds
    return ReliabilityStats(
        incidents=count,
        resolved=resolved,
        downtime_seconds=downtime,
        mttr_seconds=round_div(repair_total, resolved, "half-up") if resolved > 0 else None,
        mtbf_seconds=round_div(to - from_ - downtime, count, "half-up") if count > 0 else None,
        longest_seconds=longest,
    )