net.device-status

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 List

from .net_latency_summary import LatencySummary  ← from net.latency-summary ^1.0.0 · built alongside by fune
from .net_device_status_types import StatusResult, StatusThresholds


# Integer arithmetic only, so the reason text is identical in every language.
def _percent(basis_points: int) -> str:
    return "%d.%02d%%" % (basis_points // 100, basis_points % 100)


def _ms(us: int) -> str:
    return "%d.%03d ms" % (us // 1000, us % 1000)


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


def device_status(summary: LatencySummary, thresholds: StatusThresholds) -> StatusResult:
    """Up, degraded or down, and every reason, from one latency summary.

    Down wins outright: once a device is down, its p95 and jitter are beside
    the point. Otherwise every threshold crossed is listed, loss first, so an
    alert says all that is wrong at once.
    """
    sent, received, lost = summary.sent, summary.received, summary.lost
    if sent < 1 or received < 0 or lost < 0 or received + lost != sent:
        raise ValueError("summary is inconsistent: sent %d, received %d, lost %d" % (sent, received, lost))
    deg_loss = thresholds.degraded_loss_basis_points
    down_loss = thresholds.down_loss_basis_points
    p95_limit = thresholds.degraded_p95_us
    jitter_limit = thresholds.degraded_jitter_us
    if not _whole(down_loss) or down_loss < 1 or down_loss > 10000:
        raise ValueError("downLossBasisPoints must be 1 to 10000, received %r" % (down_loss,))
    if not _whole(deg_loss) or deg_loss < 1 or deg_loss > down_loss:
        raise ValueError("degradedLossBasisPoints must be 1 to downLossBasisPoints (%d), received %r" % (down_loss, deg_loss))
    if not _whole(p95_limit) or p95_limit < 1:
        raise ValueError("degradedP95Us must be at least 1, received %r" % (p95_limit,))
    if jitter_limit is not None and (not _whole(jitter_limit) or jitter_limit < 1):
        raise ValueError("degradedJitterUs must be null or at least 1, received %r" % (jitter_limit,))

    if received == 0:
        return StatusResult(status="down", reasons=["no replies: %d of %d probes lost" % (lost, sent)])
    loss = summary.loss_basis_points
    if loss >= down_loss:
        return StatusResult(status="down", reasons=["loss %s at or above the down threshold %s" % (_percent(loss), _percent(down_loss))])
    reasons: List[str] = []
    if loss >= deg_loss:
        reasons.append("loss %s at or above %s" % (_percent(loss), _percent(deg_loss)))
    if summary.p95_us is not None and summary.p95_us >= p95_limit:
        reasons.append("p95 %s at or above %s" % (_ms(summary.p95_us), _ms(p95_limit)))
    if jitter_limit is not None and summary.jitter_us is not None and summary.jitter_us >= jitter_limit:
        reasons.append("jitter %s at or above %s" % (_ms(summary.jitter_us), _ms(jitter_limit)))
    return StatusResult(status="degraded" if reasons else "up", reasons=reasons)