From 4639ac6e52e0b1ea87161fa832093c7fb2f08fa1 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 19 Oct 2026 15:38:17 -0400 Subject: [PATCH] fix(monitors): a ping or TCP check times the network, not Steward (#6105) response_ms was the whole `ping -c 1` process, or an open_connection awaited on the event loop: about 15 ms on a LAN whose round trip is under 1, and every host read slow together whenever Steward was busy. A ping now reports the round trip ping itself printed (the elapsed time only if it printed none); a TCP connect is timed in a thread, the name resolved first and not counted. Co-Authored-By: Claude Opus 5.5 --- steward/monitors/ping.py | 58 ++++++++++++++++++++++++++-------- tests/core/test_ping_checks.py | 46 +++++++++++++++++++++++---- 2 files changed, 84 insertions(+), 20 deletions(-) diff --git a/steward/monitors/ping.py b/steward/monitors/ping.py index 275331b..39c873e 100644 --- a/steward/monitors/ping.py +++ b/steward/monitors/ping.py @@ -3,10 +3,17 @@ A failed check says why in `error_msg` (#5430): a down monitor that can't say whether nothing answered or something refused leaves the operator to guess. + +`response_ms` is the network's own time, not Steward's (#6105): ping's +reported round trip, and a TCP connect timed in a thread of its own. Timed +from the event loop it carried starting a process and whatever else the loop +was busy with: about 15 ms on a LAN whose round trip is under 1, and every +host "slow" together whenever Steward was. """ from __future__ import annotations import asyncio import logging +import re import socket import time @@ -31,22 +38,46 @@ def tcp_failure(exc: BaseException, address: str, port: int) -> str: return str(exc) or type(exc).__name__ +def _connect_ms(address: str, port: int) -> float: + """Milliseconds a TCP connect to address:port takes, the name resolved + first and not counted; each address tried in turn, as a client would. + Blocking: run it in a thread.""" + last: OSError | None = None + for family, kind, proto, _, sockaddr in socket.getaddrinfo(address, port, type=socket.SOCK_STREAM): + with socket.socket(family, kind, proto) as sock: + sock.settimeout(TCP_TIMEOUT) + start = time.perf_counter() + try: + sock.connect(sockaddr) + except OSError as exc: + last = exc + continue + return (time.perf_counter() - start) * 1000 + raise last or OSError(f"no address for {address}") + + async def tcp_check(address: str, port: int) -> dict: """TCP connect probe. Returns {is_up, response_ms, error_msg}.""" - start = time.monotonic() try: - reader, writer = await asyncio.wait_for( - asyncio.open_connection(address, port), - timeout=TCP_TIMEOUT, - ) - writer.close() - await writer.wait_closed() - return {"is_up": True, "response_ms": round((time.monotonic() - start) * 1000, 2), - "error_msg": None} + ms = await asyncio.wait_for(asyncio.to_thread(_connect_ms, address, port), + timeout=TCP_TIMEOUT * 2) # resolving, then the connect's own 5 s + return {"is_up": True, "response_ms": round(ms, 2), "error_msg": None} except Exception as exc: return {"is_up": False, "response_ms": None, "error_msg": tcp_failure(exc, address, port)} +_RTT = re.compile(rb"time[=<]([0-9.]+) ?ms") + + +def ping_rtt(stdout: bytes) -> float | None: + """The round trip ping reports ("time=0.412 ms", or "time<1 ms"), or None.""" + m = _RTT.search(stdout or b"") + try: + return float(m.group(1)) if m else None + except ValueError: + return None + + def ping_failure(returncode: int, stderr: str) -> str: """Why `ping -c 1` failed: exit 1 is no reply; anything else is ping's own complaint (an unknown name, no route), first line of it.""" @@ -67,7 +98,7 @@ async def icmp_check(address: str) -> dict: try: proc = await asyncio.create_subprocess_exec( "ping", "-c", "1", "-W", str(PING_WAIT_S), address, - stdout=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) except Exception as exc: @@ -77,7 +108,7 @@ async def icmp_check(address: str) -> dict: result["error_msg"] = f"ping unavailable, TCP check instead: {result['error_msg']}" return result try: - _, stderr = await asyncio.wait_for(proc.communicate(), timeout=PING_TIMEOUT) + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=PING_TIMEOUT) except asyncio.TimeoutError: proc.kill() try: @@ -87,7 +118,8 @@ async def icmp_check(address: str) -> dict: return {"is_up": False, "response_ms": None, "error_msg": f"No reply within {PING_TIMEOUT:g} s"} if proc.returncode == 0: - return {"is_up": True, "response_ms": round((time.monotonic() - start) * 1000, 2), - "error_msg": None} + rtt = ping_rtt(stdout) + ms = rtt if rtt is not None else (time.monotonic() - start) * 1000 # a ping that doesn't say + return {"is_up": True, "response_ms": round(ms, 2), "error_msg": None} return {"is_up": False, "response_ms": None, "error_msg": ping_failure(proc.returncode, (stderr or b"").decode(errors="replace"))} diff --git a/tests/core/test_ping_checks.py b/tests/core/test_ping_checks.py index 7bd8f49..aa262bb 100644 --- a/tests/core/test_ping_checks.py +++ b/tests/core/test_ping_checks.py @@ -17,19 +17,19 @@ from steward.monitors import ping (RuntimeError("odd"), "odd"), ]) def test_a_tcp_failure_says_why(monkeypatch, exc, want): - async def boom(*a, **kw): + def boom(*a, **kw): raise exc - monkeypatch.setattr(ping.asyncio, "open_connection", boom) + monkeypatch.setattr(ping, "_connect_ms", boom) got = asyncio.run(ping.tcp_check("nas.lan", 22)) assert got == {"is_up": False, "response_ms": None, "error_msg": want} class _Proc: - def __init__(self, returncode, stderr=b""): - self.returncode, self._stderr = returncode, stderr + def __init__(self, returncode, stderr=b"", stdout=b""): + self.returncode, self._stderr, self._stdout = returncode, stderr, stdout async def communicate(self): - return b"", self._stderr + return self._stdout, self._stderr @pytest.mark.parametrize("code,stderr,want", [ @@ -53,13 +53,45 @@ def test_a_reply_is_up_with_no_error(monkeypatch): assert got["is_up"] is True and got["error_msg"] is None +def test_a_reply_s_time_is_the_round_trip_ping_reports_not_the_process(monkeypatch): + out = (b"PING nas.lan (192.168.0.7) 56(84) bytes of data.\n" + b"64 bytes from 192.168.0.7: icmp_seq=1 ttl=64 time=0.412 ms\n") + + async def spawn(*a, **kw): + await asyncio.sleep(0.05) # a slow process start counts for nothing + return _Proc(0, stdout=out) + monkeypatch.setattr(ping.asyncio, "create_subprocess_exec", spawn) + assert asyncio.run(ping.icmp_check("nas.lan"))["response_ms"] == 0.41 + + +@pytest.mark.parametrize("out,want", [ + (b"64 bytes from 10.0.0.1: icmp_seq=1 ttl=64 time=12.3 ms", 12.3), + (b"64 bytes from 10.0.0.1: icmp_seq=1 ttl=64 time<1 ms", 1.0), + (b"64 bytes from 10.0.0.1: icmp_seq=1 ttl=64 time=0.05ms", 0.05), + (b"", None), +]) +def test_ping_s_round_trip_is_read_from_its_line(out, want): + assert ping.ping_rtt(out) == want + + +def test_a_tcp_connect_is_timed_apart_from_the_event_loop(monkeypatch): + import threading + + def connect(address, port): + assert threading.current_thread() is not threading.main_thread() + return 0.8 + + monkeypatch.setattr(ping, "_connect_ms", connect) + assert asyncio.run(ping.tcp_check("nas.lan", 22)) == {"is_up": True, "response_ms": 0.8, "error_msg": None} + + def test_without_ping_the_tcp_stand_in_says_so(monkeypatch): async def no_ping(*a, **kw): raise FileNotFoundError("ping") - async def refused(*a, **kw): + def refused(*a, **kw): raise ConnectionRefusedError(111, "Connection refused") monkeypatch.setattr(ping.asyncio, "create_subprocess_exec", no_ping) - monkeypatch.setattr(ping.asyncio, "open_connection", refused) + monkeypatch.setattr(ping, "_connect_ms", refused) got = asyncio.run(ping.icmp_check("nas.lan")) assert got["error_msg"] == "ping unavailable, TCP check instead: Connection refused on port 80"