Functional Weave
Code in Python

monitor.uptime@1.0.0

impl/python.py

2,318 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_check_status import Check  ← from monitor.check-status ^1.0.0 · built alongside by fune
from .monitor_uptime_types import UptimeReport


def _whole(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def uptime(checks: Sequence[Check], from_: int, to: int, max_gap_seconds: int) -> UptimeReport:
    """Seconds up, degraded, down and unknown over [from, to). Each check's
    status holds from its `at` until the next check, but for at most
    max_gap_seconds: a monitor that stopped reporting has not proved the
    target was up."""
    if not _whole(from_) or not _whole(to):
        raise ValueError("from and to must be whole seconds")
    if from_ > to:
        raise ValueError("from must not be after to: %d > %d" % (from_, to))
    if not _whole(max_gap_seconds) or max_gap_seconds < 1:
        raise ValueError("maxGapSeconds must be a whole number of at least 1, received %s" % (max_gap_seconds,))
    up = degraded = down = 0
    count = down_count = 0
    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 from_ <= c.at < to:
            count += 1
            if c.status == "down":
                down_count += 1
        # The span this check vouches for, clipped to the window.
        end = c.at + max_gap_seconds
        if i + 1 < len(checks) and checks[i + 1].at < end:
            end = checks[i + 1].at
        s = max(c.at, from_)
        e = min(end, to)
        if e > s:
            if c.status == "up":
                up += e - s
            elif c.status == "degraded":
                degraded += e - s
            else:
                down += e - s
    known = up + degraded + down
    # Floored: never 100.00% while a down second exists.
    bp: Optional[int] = None if known == 0 else (up + degraded) * 10000 // known
    return UptimeReport(
        up_seconds=up,
        degraded_seconds=degraded,
        down_seconds=down,
        unknown_seconds=to - from_ - known,
        uptime_basis_points=bp,
        checks=count,
        down_checks=down_count,
    )