from typing import List from .net_latency_summary import LatencySummary 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)