add jitter matrix (p95/p99/cv) to ping stats

src/netsleuth/models.py

@@ -265,6 +265,9 @@ class PingResult:
     max_ms: float | None = None
     mdev_ms: float | None = None
     jitter_ms: float | None = None
+    p95_ms: float | None = None
+    p99_ms: float | None = None
+    cv: float | None = None
     samples: list[float | None] = field(default_factory=list)
 
 

src/netsleuth/probes/latency.py

@@ -6,7 +6,7 @@ import time
 
 from netsleuth.models import Capabilities, PingResult
 from netsleuth.netinfo import choose_latency_backend
-from netsleuth.stats import rtt_stats
+from netsleuth.stats import jitter_matrix, rtt_stats
 
 
 def summarize_ping(
@@ -17,6 +17,7 @@ def summarize_ping(
     samples: list[float | None],
 ) -> PingResult:
     s = rtt_stats(samples)
+    j = jitter_matrix(samples)
     return PingResult(
         label=label,
         host=host,
@@ -30,6 +31,9 @@ def summarize_ping(
         max_ms=s.max_ms,
         mdev_ms=s.mdev_ms,
         jitter_ms=s.jitter_ms,
+        p95_ms=j.p95_ms,
+        p99_ms=j.p99_ms,
+        cv=j.cv,
         samples=samples,
     )
 
@@ -88,7 +92,11 @@ async def ping_host(
     timeout: float,
     backend: str,
     source_ip: str | None = None,
+    semaphore: asyncio.Semaphore | None = None,
 ) -> PingResult:
+    if semaphore is not None:
+        async with semaphore:
+            return await ping_host(host, label, count, interval, timeout, backend, source_ip)
     resolved = await _resolve(host)
     if backend in ("icmp_win", "icmp_dgram", "icmp_raw"):
         try:
@@ -111,10 +119,14 @@ async def ping_fanout(
     interval: float,
     timeout: float,
     source_ip: str | None = None,
+    semaphore: asyncio.Semaphore | None = None,
 ) -> list[PingResult]:
     backend = choose_latency_backend(caps)
     return list(
         await asyncio.gather(
-            *(ping_host(host, label, count, interval, timeout, backend, source_ip) for label, host in hosts)
+            *(
+                ping_host(host, label, count, interval, timeout, backend, source_ip, semaphore)
+                for label, host in hosts
+            )
         )
     )

src/netsleuth/stats.py

@@ -51,3 +51,29 @@ def percentile(values: list[float], p: float) -> float:
     low = int(rank)
     high = min(low + 1, len(ordered) - 1)
     return ordered[low] + (ordered[high] - ordered[low]) * (rank - low)
+
+
+class JitterMatrix(NamedTuple):
+    p50_ms: float | None
+    p95_ms: float | None
+    p99_ms: float | None
+    stdev_ms: float | None
+    cv: float | None
+    outlier_pct: float
+
+
+def jitter_matrix(samples: list[float | None]) -> JitterMatrix:
+    got = [s for s in samples if s is not None]
+    if not got:
+        return JitterMatrix(None, None, None, None, None, 0.0)
+    p50 = round(percentile(got, 50.0), 3)
+    p95 = round(percentile(got, 95.0), 3)
+    p99 = round(percentile(got, 99.0), 3)
+    if len(got) == 1:
+        return JitterMatrix(p50, p95, p99, 0.0, 0.0, 0.0)
+    mean = sum(got) / len(got)
+    stdev = (sum((v - mean) ** 2 for v in got) / len(got)) ** 0.5
+    cv = 0.0 if mean == 0 else stdev / mean
+    outliers = sum(1 for v in got if v > p95)
+    outlier_pct = round(100.0 * outliers / len(got), 3)
+    return JitterMatrix(p50, p95, p99, round(stdev, 3), round(cv, 3), outlier_pct)

tests/test_prober_stats.py

@@ -62,3 +62,25 @@ def test_summarize_of_zero_probes_is_a_well_formed_empty_result():
 def test_the_method_tag_is_preserved_verbatim():
     for method in ("icmp_win", "icmp_dgram", "icmp_raw", "tcp", "cfL4"):
         assert summarize_ping("h", "x", None, method, [1.0]).method == method
+
+
+def test_summarize_fills_the_jitter_matrix_fields():
+    result = summarize_ping("h", "1.1.1.1", "1.1.1.1", "icmp_dgram", [10.0, 12.0, 14.0, 16.0, 40.0])
+    assert result.p95_ms is not None
+    assert result.p99_ms is not None
+    assert result.cv is not None
+    assert result.p99_ms >= result.p95_ms
+
+
+def test_summarize_of_an_all_timeout_run_has_no_jitter_matrix():
+    result = summarize_ping("h", "example.test", None, "icmp_win", [None, None, None])
+    assert result.p95_ms is None
+    assert result.p99_ms is None
+    assert result.cv is None
+
+
+def test_summarize_of_a_single_sample_has_zero_cv():
+    result = summarize_ping("h", "1.1.1.1", "1.1.1.1", "cfL4", [9.0])
+    assert result.p95_ms == 9.0
+    assert result.p99_ms == 9.0
+    assert result.cv == 0.0

tests/test_stats.py

@@ -4,7 +4,7 @@ import math
 
 import pytest
 
-from netsleuth.stats import percentile, rtt_stats
+from netsleuth.stats import jitter_matrix, percentile, rtt_stats
 
 
 def test_all_samples_present():
@@ -78,3 +78,67 @@ def test_percentile_interpolates_between_ranks():
 def test_percentile_handles_single_and_empty_inputs():
     assert percentile([7.0], 90.0) == 7.0
     assert percentile([], 90.0) == 0.0
+
+
+def test_jitter_matrix_percentiles_match_the_shared_percentile_helper():
+    samples = [10.0, 12.0, 14.0, 16.0, 40.0]
+    m = jitter_matrix(samples)
+    assert m.p50_ms == pytest.approx(percentile(samples, 50.0), abs=1e-3)
+    assert m.p95_ms == pytest.approx(percentile(samples, 95.0), abs=1e-3)
+    assert m.p99_ms == pytest.approx(percentile(samples, 99.0), abs=1e-3)
+
+
+def test_cv_is_stdev_over_mean():
+    # mean = 15, population stdev = sqrt(((5)^2+(3)^2+(1)^2+(1)^2+(3)^2+(5)^2)/6) = sqrt(70/6)
+    samples = [10.0, 12.0, 14.0, 16.0, 18.0, 20.0]
+    m = jitter_matrix(samples)
+    mean = sum(samples) / len(samples)
+    expected_stdev = (sum((v - mean) ** 2 for v in samples) / len(samples)) ** 0.5
+    assert m.stdev_ms == pytest.approx(expected_stdev, abs=1e-3)
+    assert m.cv == pytest.approx(expected_stdev / mean, abs=1e-3)
+
+
+def test_cv_of_a_constant_series_is_zero():
+    m = jitter_matrix([10.0, 10.0, 10.0])
+    assert m.stdev_ms == 0.0
+    assert m.cv == 0.0
+
+
+def test_cv_of_a_single_sample_is_zero_not_a_division_by_zero():
+    m = jitter_matrix([13.5])
+    assert m.p50_ms == m.p95_ms == m.p99_ms == 13.5
+    assert m.stdev_ms == 0.0
+    assert m.cv == 0.0
+    assert m.outlier_pct == 0.0
+
+
+def test_jitter_matrix_of_all_timeouts_is_all_none():
+    m = jitter_matrix([None, None, None])
+    assert m.p50_ms is None
+    assert m.p95_ms is None
+    assert m.p99_ms is None
+    assert m.stdev_ms is None
+    assert m.cv is None
+    assert m.outlier_pct == 0.0
+
+
+def test_jitter_matrix_empty_input_is_not_a_division_by_zero():
+    m = jitter_matrix([])
+    assert m.p50_ms is None
+    assert m.outlier_pct == 0.0
+
+
+def test_outlier_share_counts_samples_above_p95():
+    # Ten samples clustered at 10ms, one spike at 500ms: the spike sits above p95.
+    samples = [10.0] * 10 + [500.0]
+    m = jitter_matrix(samples)
+    above = sum(1 for v in samples if v > m.p95_ms)
+    assert m.outlier_pct == pytest.approx(100.0 * above / len(samples), abs=1e-3)
+    assert above >= 1
+
+
+def test_every_jitter_field_is_finite_so_json_serialization_cannot_break():
+    for samples in ([], [None], [1.0], [1.0, None, 3.0], [1.0] * 10 + [500.0]):
+        m = jitter_matrix(samples)
+        for value in (m.p50_ms, m.p95_ms, m.p99_ms, m.stdev_ms, m.cv, m.outlier_pct):
+            assert value is None or math.isfinite(value)