omarchy/plugins/io.github.x3me.nexthop/nexthopd/probes.py

"""Persistent ping and TCP-handshake probes, and the rolling windows they feed.

One long-lived `ping` process per target rather than one process per sample.
At two samples a second, spawning a process each time would mean 172,800
forks a day inside a laptop's idle budget; `ping -i` already does the timing
for us, and `-O` makes it say so out loud when a packet goes missing.
"""

import collections
import re
import shutil
import socket
import statistics
import subprocess
import threading
import time
from collections import deque


def nearest_rank(ordered, p: float):
    """Nearest-rank percentile of a NON-EMPTY sorted sequence.

    One implementation shared by the probe stats, the per-app socket stats
    and the speed baseline, so the three cannot drift apart — they read the
    same figure off the same rule. The index formula already collapses to
    element 0 for a single-element input, so no length special-case.
    """
    i = min(len(ordered) - 1, max(0, int(round(p * (len(ordered) - 1)))))
    return ordered[i]

# [1787562260.703963] 64 bytes from 10.10.0.1: icmp_seq=1 ttl=64 time=9.13 ms
RE_REPLY = re.compile(r"^\[(\d+\.\d+)\].*icmp_seq=(\d+).*time=([\d.]+)\s*ms")
# [1787562369.690501] no answer yet for icmp_seq=1
RE_PENDING = re.compile(r"^\[(\d+\.\d+)\]\s+no answer yet for icmp_seq=(\d+)")
# [...] From 10.10.0.147 icmp_seq=1 Destination Host Unreachable
RE_UNREACH = re.compile(r"^\[(\d+\.\d+)\].*icmp_seq=(\d+).*(?:Unreachable|unreachable)")


# How much recent history judges an instrument, and the fewest samples that
# can carry a judgement at all.
#
# Shared, deliberately, by the two places that ask "what has this instrument
# been doing lately": the bench, which ranks instruments over a window, and
# TcpProbe, which needs its own recent p50 to tell a retransmit from a slow
# path. They live here rather than on the bench because `instruments` imports
# this module and not the other way round, and an alias in the direction the
# imports already run is the only one Python will take.
#
# The point of aliasing rather than repeating the number: two literals plus a
# test catches drift on the next test run, an alias makes the drift
# impossible. `test_the_window_matches_what_the_bench_ranks_on` is kept even
# though it now passes by construction — it catches someone replacing an
# alias with a literal, which is the drift it was written against.
RECENT_WINDOW_S = 300.0
RECENT_MIN_SAMPLES = 8


class Series:
    """A rolling window of (timestamp, rtt_ms or None) for one target.

    None means the probe went out and nothing came back. Keeping losses in
    the same series as the replies is what lets a single pass compute both
    latency and loss over any sub-window.
    """

    def __init__(self, window_s: float = 1830.0):
        self.window_s = window_s
        self._samples = deque()
        self._lock = threading.Lock()

    def add(self, t: float, rtt, loaded: bool = False):
        """Record one probe result, tagged with whether the link was busy.

        The tag is what makes bufferbloat visible: the same connection can
        answer in 15 ms while idle and 300 ms while a download runs, and a
        score built only on the idle number calls that line excellent right
        up until someone uses it.
        """
        with self._lock:
            self._samples.append((t, rtt, bool(loaded)))
            cutoff = t - self.window_s
            while self._samples and self._samples[0][0] < cutoff:
                self._samples.popleft()

    def since(self, seconds: float):
        cutoff = time.time() - seconds
        with self._lock:
            return [s for s in self._samples if s[0] >= cutoff]

    def all(self):
        with self._lock:
            return list(self._samples)

    @staticmethod
    def split_by_load(samples):
        """(idle, loaded) — probes taken while the link was quiet vs busy.

        Samples are indexed rather than unpacked throughout, so a caller
        holding older two-element samples still reads as idle instead of
        raising.
        """
        idle = [s for s in samples if not (len(s) > 2 and s[2])]
        loaded = [s for s in samples if len(s) > 2 and s[2]]
        return idle, loaded

    @staticmethod
    def stats(samples) -> dict:
        """Latency percentiles, jitter and loss over the samples given.

        Jitter is mean absolute difference between consecutive replies
        (RFC 3550's IPDV), not standard deviation: a connection that
        alternates 10/40/10/40 ms feels far worse than one that drifts
        smoothly across the same range, and only IPDV says so.
        """
        total = len(samples)
        if total == 0:
            return {"count": 0, "loss": None, "p50": None, "p75": None,
                    "p95": None, "jitter": None, "last": None, "max": None}

        rtts = [s[1] for s in samples if s[1] is not None]
        lost = total - len(rtts)
        loss = lost / total

        if not rtts:
            return {"count": total, "loss": loss, "p50": None, "p75": None,
                    "p95": None, "jitter": None, "last": None, "max": None}

        ordered = sorted(rtts)

        deltas = [abs(rtts[i] - rtts[i - 1]) for i in range(1, len(rtts))]
        last = next((x[1] for x in reversed(samples) if x[1] is not None), None)

        return {
            "count": total,
            "loss": loss,
            "p50": round(statistics.median(ordered), 2),
            "p75": round(nearest_rank(ordered, 0.75), 2),
            "p95": round(nearest_rank(ordered, 0.95), 2),
            "max": round(ordered[-1], 2),
            "jitter": round(statistics.fmean(deltas), 2) if deltas else 0.0,
            "last": round(last, 2) if last is not None else None,
        }


class PingProbe(threading.Thread):
    """Runs one `ping` forever, restarting it if it dies, feeding a Series.

    A probe never raises into the daemon: if `ping` is missing, the target
    stops resolving, or the interface goes away, the thread backs off and
    keeps trying while the series simply records losses.
    """


    def __init__(self, target: str, series: Series, interval_ms: int = 500,
                 name: str = "", loaded_fn=None):
        super().__init__(name=f"probe-{name or target}", daemon=True)
        self.target = target
        self.series = series
        # Asked at the moment a sample lands, so each probe is tagged with
        # the link state it actually experienced rather than whatever the
        # link was doing when the window is later read.
        self.loaded_fn = loaded_fn
        self.interval = max(0.2, interval_ms / 1000.0)
        self._stop = threading.Event()
        self._proc = None
        # seq -> timestamp first seen unanswered, drained by _expire()
        self._pending = {}
        # seq -> when it was charged as lost. A packet the grace period gave
        # up on can still be reported afterwards — the gateway's Destination
        # Host Unreachable for it arrives later than the grace, in the real
        # recording by 0.35 s — and without this it would be charged twice.
        # Held for one further grace period, which is as long as a late report
        # can be believed to belong to that packet.
        self._charged = {}

    def _loaded(self) -> bool:
        try:
            return bool(self.loaded_fn()) if self.loaded_fn else False
        except Exception:
            return False        # a probe never raises into the daemon

    def stop(self):
        self._stop.set()
        proc = self._proc
        if proc and proc.poll() is None:
            try:
                proc.terminate()
            except OSError:
                pass

    def set_interval(self, seconds: float):
        """Change cadence in place — a benched instrument idles, a seated
        one probes at full rate, without tearing the thread down. `ping`
        takes its interval on the command line, so the running process is
        retired and the run loop respawns it with the new one."""
        seconds = max(0.2, float(seconds))
        if abs(seconds - self.interval) < 1e-9:
            return
        self.interval = seconds
        proc = self._proc
        if proc and proc.poll() is None:
            try:
                proc.terminate()
            except OSError:
                pass

    def _reset_tracking(self):
        """Forget both maps together.

        `ping` numbers from 1 again on every respawn, so a seq remembered past
        the process that produced it would suppress a real loss on the next
        one — turning a guard against overcharging into an undercount, which
        is the same defect facing the other way.
        """
        self._pending.clear()
        self._charged.clear()

    def _expire(self, now: float):
        """A packet still unanswered after the grace period is a lost packet.

        `ping -O` reports "no answer yet" as soon as it sends the next probe,
        but a slow reply can still land, so a pending seq is only counted as
        lost once it is too old to come back.
        """
        grace = self.interval * 2.5 + 1.0
        for seq, t in list(self._pending.items()):
            if now - t > grace:
                del self._pending[seq]
                self._charged[seq] = now
                self.series.add(t, None, self._loaded())
        # Bounded by the same clock that fills it: a seq stops being
        # remembered once no report about it could still arrive.
        for seq, t in list(self._charged.items()):
            if now - t > grace:
                del self._charged[seq]

    def run(self):
        backoff = 1.0
        while not self._stop.is_set():
            if not shutil.which("ping") or not self.target:
                time.sleep(min(backoff, 30.0))
                backoff = min(backoff * 2, 30.0)
                continue
            try:
                self._run_once()
                backoff = 1.0
            except Exception:
                # Never let a parse or spawn failure take the daemon with it.
                time.sleep(min(backoff, 30.0))
                backoff = min(backoff * 2, 30.0)

    def _run_once(self):
        cmd = ["ping", "-n", "-O", "-D", "-i", f"{self.interval:g}",
               "-W", "1", self.target]
        self._reset_tracking()
        self._proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
            text=True, bufsize=1,
        )
        try:
            for line in self._proc.stdout:
                if self._stop.is_set():
                    break
                self._consume(line)
            # ping exited: whatever was outstanding never arrived.
            for seq, t in self._pending.items():
                self.series.add(t, None, self._loaded())
            self._reset_tracking()
        finally:
            proc, self._proc = self._proc, None
            if proc:
                try:
                    proc.terminate()
                    proc.wait(timeout=2)
                except subprocess.TimeoutExpired:
                    # Would not go quietly: do not leave it running.
                    try:
                        proc.kill()
                        proc.wait(timeout=2)
                    except (OSError, subprocess.TimeoutExpired):
                        pass
                except OSError:
                    pass

    def _consume(self, line: str):
        m = RE_REPLY.match(line)
        if m:
            t, seq, rtt = float(m.group(1)), int(m.group(2)), float(m.group(3))
            self._pending.pop(seq, None)
            # A reply this late cannot un-lose the packet — the window it
            # belonged to has already been read — and recording the RTT as
            # well would put two samples on the wire's one packet.
            if seq not in self._charged:
                self.series.add(t, rtt, self._loaded())
            self._expire(t)
            return

        m = RE_UNREACH.match(line)
        if m:
            t, seq = float(m.group(1)), int(m.group(2))
            self._pending.pop(seq, None)
            if seq not in self._charged:
                self.series.add(t, None, self._loaded())
            self._expire(t)
            return

        m = RE_PENDING.match(line)
        if m:
            t, seq = float(m.group(1)), int(m.group(2))
            # `ping -O` repeats "no answer yet" for the same seq, so one that
            # has already been charged must not be put back on the pending
            # list to be charged a second time.
            if seq not in self._charged:
                self._pending.setdefault(seq, t)
            self._expire(t)


class TcpProbe(threading.Thread):
    """Connect-time RTT to the anchor's TLS port, feeding a Series.

    ICMP measures what routers choose to answer, and they answer it fast:
    many devices handle it in hardware, in an ASIC or via XDP, while real
    traffic waits in the user-space path behind the queues that actually
    hold it up. Anything on the way can also reply on the destination's
    behalf, because there is nothing in ICMP to prove otherwise.

    A TCP handshake cannot be shortcut that way. The SYN has to reach a
    listener that completes it, over port 443 where the user's own traffic
    goes, so its round trip is the one applications experience. One
    connection per sample, opened and closed — no payload, no TLS, nothing
    kept.

    Since 0.2.0 these are seated instruments in the bench (instruments.py):
    the two best of four feed the scored internet leg, so a TCP series moves
    the score whenever it holds a seat. The anchor's ICMP figure is still
    recorded beside it per minute (`lag_icmp`) so the switch stays auditable.
    """

    CONNECT_TIMEOUT_S = 2.0
    # Linux and macOS both start TCP's retransmit timer at one second, so a
    # handshake that comes back at or past this did not measure a slow path:
    # its SYN was dropped and the kernel sent another. The number is the
    # kernel's constant, not the network's round trip, and folding it into a
    # latency percentile reports the line as slow when what happened is that
    # a packet was lost.
    #
    # The connect timeout was already drawing this line, in the wrong place
    # and for the wrong reason: a handshake needing TWO retransmits waits
    # 1 s + 2 s, exceeds CONNECT_TIMEOUT_S and is recorded as loss, while one
    # needing a single retransmit returns at ~1 s and was recorded as a round
    # trip. The same event, accounted two opposite ways, with the boundary
    # wherever the timeout happened to fall.
    # One initial RTO, with slop for timer granularity and scheduling. The
    # first retransmit fires at 1000 ms on Linux, macOS and Windows alike.
    RETRANSMIT_MARGIN_MS = 900.0
    # The baseline is this instrument's own recent p50 over the shared window
    # above — aliases, not copies, so "recent" cannot come to mean two things.
    RETRANSMIT_WINDOW_S = RECENT_WINDOW_S
    RETRANSMIT_MIN_SAMPLES = RECENT_MIN_SAMPLES

    def __init__(self, target: str, series: Series, interval_s: float = 1.0,
                 name: str = "", loaded_fn=None, port: int = 443):
        super().__init__(name=f"tcp-{name or target}", daemon=True)
        self.target = target
        self.port = port
        self.series = series
        self.interval = max(0.25, interval_s)
        self.loaded_fn = loaded_fn
        self._stop = threading.Event()
        self.ever_connected = False
        # The recent round trips this probe has actually seen, for the
        # comparison above. Bounded, and its own — the series it feeds is
        # merged with other instruments and cannot answer "what does THIS
        # path usually do". Held as (when, rtt) so the window is a duration
        # rather than a count, which is what makes it survive a cadence
        # change: a benched instrument probes at a fraction of the rate.
        self._recent = collections.deque(maxlen=1024)
        # Samples this probe declined to call latency, so the reclassification
        # can be seen rather than inferred from a loss rate. A dropped SYN and
        # a slow line are different faults with different owners.
        self.retransmits = 0
        self.unclassified = 0

    def stop(self):
        self._stop.set()

    def set_interval(self, seconds: float):
        """Picked up on the next cycle; nothing to tear down here."""
        self.interval = max(0.25, float(seconds))

    def _loaded(self) -> bool:
        try:
            return bool(self.loaded_fn()) if self.loaded_fn else False
        except Exception:
            return False

    def _once(self):
        started = time.time()
        t0 = time.monotonic()
        try:
            sock = socket.create_connection((self.target, self.port),
                                            timeout=self.CONNECT_TIMEOUT_S)
        except (OSError, ValueError):
            self.series.add(started, None, self._loaded())
            return
        rtt = (time.monotonic() - t0) * 1000.0
        try:
            sock.close()
        except OSError:
            pass
        # The handshake completed, so the target is reachable, whatever the
        # kernel had to do to get there.
        self.ever_connected = True
        verdict = self._classify(started, rtt)
        if verdict == "retransmit":
            # Loss on new connections, which is what it is. Recorded the same
            # way a refused or timed-out connect already is, so it charges the
            # loss term and Reliability rather than the latency percentiles.
            self.retransmits += 1
            self.series.add(started, None, self._loaded())
            return
        if verdict == "unknown":
            # Past the floor before this probe has a baseline to judge it
            # against. It is either a retransmit or a genuinely slow path and
            # nothing here can tell which, so it is not recorded as either —
            # inventing a loss and publishing a suspect latency are both
            # claims, and the honest move is to make neither.
            #
            # It still feeds the baseline, and that is not an oversight. A
            # link whose real round trip is past the floor — p50 1200 ms, say
            # — has every sample land here, so a deque that only accepted
            # classified samples would never reach its minimum, the baseline
            # would never form, and the instrument would stay unclassified
            # for ever: nothing recorded, count never growing, `penalty()`
            # returning None, and the bench able neither to seat it nor to
            # call it dead. A silent unrankable instrument, invisible because
            # it is not failing, merely absent.
            self.unclassified += 1
            self._recent.append((started, rtt))
            return
        self._recent.append((started, rtt))
        self.series.add(started, round(rtt, 2), self._loaded())

    def _baseline_ms(self, now: float):
        """This instrument's own recent p50, or None while it has too few."""
        cutoff = now - self.RETRANSMIT_WINDOW_S
        recent = [rtt for t, rtt in self._recent if t >= cutoff]
        if len(recent) < self.RETRANSMIT_MIN_SAMPLES:
            return None
        return statistics.median(recent)

    def _classify(self, now: float, rtt_ms: float) -> str:
        """"reply", "retransmit" or "unknown" for a handshake that completed.

        A connect rescued by a retransmitted SYN is a lost packet, not a slow
        path, and the threshold has to be relative or it mislabels distance as
        loss: one full RTO ABOVE what this instrument usually sees. A satellite
        link whose p50 is 600 ms gets a threshold of 1500, so a 1045 ms sample
        there stays the measurement it is.
        """
        if rtt_ms < self.RETRANSMIT_MARGIN_MS:
            return "reply"
        baseline = self._baseline_ms(now)
        if baseline is None:
            return "unknown"
        # Note what does NOT reach the baseline once one exists: a sample this
        # returns "retransmit" for. Feeding those back would raise the
        # threshold on the instrument's own retransmits and the rule would
        # quietly stop firing exactly where it is needed most.
        return "retransmit" if rtt_ms >= baseline + self.RETRANSMIT_MARGIN_MS \
            else "reply"

    def run(self):
        while not self._stop.is_set():
            if not self.target:
                self._stop.wait(5.0)
                continue
            t0 = time.monotonic()
            try:
                self._once()
            except Exception:
                # Never let a socket or DNS failure take the daemon with it.
                pass
            self._stop.wait(max(0.0, self.interval - (time.monotonic() - t0)))