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)