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 <noreply@anthropic.com>
This commit is contained in:
+45
-13
@@ -3,10 +3,17 @@
|
|||||||
A failed check says why in `error_msg` (#5430): a down monitor that can't
|
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
|
say whether nothing answered or something refused leaves the operator to
|
||||||
guess.
|
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
|
from __future__ import annotations
|
||||||
import asyncio
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
|
import re
|
||||||
import socket
|
import socket
|
||||||
import time
|
import time
|
||||||
|
|
||||||
@@ -31,22 +38,46 @@ def tcp_failure(exc: BaseException, address: str, port: int) -> str:
|
|||||||
return str(exc) or type(exc).__name__
|
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:
|
async def tcp_check(address: str, port: int) -> dict:
|
||||||
"""TCP connect probe. Returns {is_up, response_ms, error_msg}."""
|
"""TCP connect probe. Returns {is_up, response_ms, error_msg}."""
|
||||||
start = time.monotonic()
|
|
||||||
try:
|
try:
|
||||||
reader, writer = await asyncio.wait_for(
|
ms = await asyncio.wait_for(asyncio.to_thread(_connect_ms, address, port),
|
||||||
asyncio.open_connection(address, port),
|
timeout=TCP_TIMEOUT * 2) # resolving, then the connect's own 5 s
|
||||||
timeout=TCP_TIMEOUT,
|
return {"is_up": True, "response_ms": round(ms, 2), "error_msg": None}
|
||||||
)
|
|
||||||
writer.close()
|
|
||||||
await writer.wait_closed()
|
|
||||||
return {"is_up": True, "response_ms": round((time.monotonic() - start) * 1000, 2),
|
|
||||||
"error_msg": None}
|
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
return {"is_up": False, "response_ms": None, "error_msg": tcp_failure(exc, address, port)}
|
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:
|
def ping_failure(returncode: int, stderr: str) -> str:
|
||||||
"""Why `ping -c 1` failed: exit 1 is no reply; anything else is ping's
|
"""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."""
|
own complaint (an unknown name, no route), first line of it."""
|
||||||
@@ -67,7 +98,7 @@ async def icmp_check(address: str) -> dict:
|
|||||||
try:
|
try:
|
||||||
proc = await asyncio.create_subprocess_exec(
|
proc = await asyncio.create_subprocess_exec(
|
||||||
"ping", "-c", "1", "-W", str(PING_WAIT_S), address,
|
"ping", "-c", "1", "-W", str(PING_WAIT_S), address,
|
||||||
stdout=asyncio.subprocess.DEVNULL,
|
stdout=asyncio.subprocess.PIPE,
|
||||||
stderr=asyncio.subprocess.PIPE,
|
stderr=asyncio.subprocess.PIPE,
|
||||||
)
|
)
|
||||||
except Exception as exc:
|
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']}"
|
result["error_msg"] = f"ping unavailable, TCP check instead: {result['error_msg']}"
|
||||||
return result
|
return result
|
||||||
try:
|
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:
|
except asyncio.TimeoutError:
|
||||||
proc.kill()
|
proc.kill()
|
||||||
try:
|
try:
|
||||||
@@ -87,7 +118,8 @@ async def icmp_check(address: str) -> dict:
|
|||||||
return {"is_up": False, "response_ms": None,
|
return {"is_up": False, "response_ms": None,
|
||||||
"error_msg": f"No reply within {PING_TIMEOUT:g} s"}
|
"error_msg": f"No reply within {PING_TIMEOUT:g} s"}
|
||||||
if proc.returncode == 0:
|
if proc.returncode == 0:
|
||||||
return {"is_up": True, "response_ms": round((time.monotonic() - start) * 1000, 2),
|
rtt = ping_rtt(stdout)
|
||||||
"error_msg": None}
|
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,
|
return {"is_up": False, "response_ms": None,
|
||||||
"error_msg": ping_failure(proc.returncode, (stderr or b"").decode(errors="replace"))}
|
"error_msg": ping_failure(proc.returncode, (stderr or b"").decode(errors="replace"))}
|
||||||
|
|||||||
@@ -17,19 +17,19 @@ from steward.monitors import ping
|
|||||||
(RuntimeError("odd"), "odd"),
|
(RuntimeError("odd"), "odd"),
|
||||||
])
|
])
|
||||||
def test_a_tcp_failure_says_why(monkeypatch, exc, want):
|
def test_a_tcp_failure_says_why(monkeypatch, exc, want):
|
||||||
async def boom(*a, **kw):
|
def boom(*a, **kw):
|
||||||
raise exc
|
raise exc
|
||||||
monkeypatch.setattr(ping.asyncio, "open_connection", boom)
|
monkeypatch.setattr(ping, "_connect_ms", boom)
|
||||||
got = asyncio.run(ping.tcp_check("nas.lan", 22))
|
got = asyncio.run(ping.tcp_check("nas.lan", 22))
|
||||||
assert got == {"is_up": False, "response_ms": None, "error_msg": want}
|
assert got == {"is_up": False, "response_ms": None, "error_msg": want}
|
||||||
|
|
||||||
|
|
||||||
class _Proc:
|
class _Proc:
|
||||||
def __init__(self, returncode, stderr=b""):
|
def __init__(self, returncode, stderr=b"", stdout=b""):
|
||||||
self.returncode, self._stderr = returncode, stderr
|
self.returncode, self._stderr, self._stdout = returncode, stderr, stdout
|
||||||
|
|
||||||
async def communicate(self):
|
async def communicate(self):
|
||||||
return b"", self._stderr
|
return self._stdout, self._stderr
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("code,stderr,want", [
|
@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
|
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):
|
def test_without_ping_the_tcp_stand_in_says_so(monkeypatch):
|
||||||
async def no_ping(*a, **kw):
|
async def no_ping(*a, **kw):
|
||||||
raise FileNotFoundError("ping")
|
raise FileNotFoundError("ping")
|
||||||
|
|
||||||
async def refused(*a, **kw):
|
def refused(*a, **kw):
|
||||||
raise ConnectionRefusedError(111, "Connection refused")
|
raise ConnectionRefusedError(111, "Connection refused")
|
||||||
monkeypatch.setattr(ping.asyncio, "create_subprocess_exec", no_ping)
|
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"))
|
got = asyncio.run(ping.icmp_check("nas.lan"))
|
||||||
assert got["error_msg"] == "ping unavailable, TCP check instead: Connection refused on port 80"
|
assert got["error_msg"] == "ping unavailable, TCP check instead: Connection refused on port 80"
|
||||||
|
|||||||
Reference in New Issue
Block a user