chore(netdiag): add standalone UDP path diagnostics
This commit is contained in:
justme committed
1 parent
be97e24693
commit
f6e477fb10
1 file changed
+235
@@ -0,0 +1,235 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Small dependency-free UDP path test with sequence, loss, and RTT reporting.
|
||||
|
||||
Run a bounded echo receiver on the remote host:
|
||||
python3 scripts/netdiag/udp_path.py server --bind 0.0.0.0 --port 443 --duration 45
|
||||
|
||||
Then run the client from the source host:
|
||||
python3 scripts/netdiag/udp_path.py client --target 203.0.113.10 --port 443 \
|
||||
--count 251 --rate 31 --size 64 --timeout 3 --socket-mode sendto
|
||||
|
||||
The server echoes each valid datagram unchanged. The client uses its sequence and
|
||||
monotonic send timestamp to report lost replies and round-trip latency. It uses
|
||||
no TCP control channel, so TCP service occupancy on the same port is irrelevant.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import math
|
||||
import secrets
|
||||
import select
|
||||
import socket
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
from collections import Counter
|
||||
|
||||
|
||||
HEADER = struct.Struct("!8sIQ") # random run id, sequence, monotonic ns
|
||||
MAX_DATAGRAM = 65_507
|
||||
|
||||
|
||||
def percentile(values: list[float], p: float) -> float | None:
|
||||
if not values:
|
||||
return None
|
||||
ordered = sorted(values)
|
||||
index = max(0, math.ceil(p * len(ordered)) - 1)
|
||||
return ordered[index]
|
||||
|
||||
|
||||
def format_ms(value: float | None) -> str:
|
||||
return "n/a" if value is None else f"{value:.3f} ms"
|
||||
|
||||
|
||||
def is_probe_packet(data: bytes) -> bool:
|
||||
return len(data) >= HEADER.size
|
||||
|
||||
|
||||
def run_server(args: argparse.Namespace) -> int:
|
||||
sock = socket.socket(socket.AF_INET6 if ":" in args.bind else socket.AF_INET, socket.SOCK_DGRAM)
|
||||
if sock.family == socket.AF_INET6:
|
||||
sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
|
||||
try:
|
||||
sock.bind((args.bind, args.port))
|
||||
except OSError as exc:
|
||||
print(f"bind failed on {args.bind}:{args.port}: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
print(f"listening udp://{args.bind}:{args.port} for {args.duration:g}s", flush=True)
|
||||
deadline = time.monotonic() + args.duration if args.duration > 0 else None
|
||||
received = echoed = malformed = duplicates = 0
|
||||
sequences: set[tuple[str, bytes, int]] = set()
|
||||
sources: Counter[str] = Counter()
|
||||
next_report = time.monotonic() + 1
|
||||
sock.settimeout(0.25)
|
||||
try:
|
||||
while deadline is None or time.monotonic() < deadline:
|
||||
try:
|
||||
data, address = sock.recvfrom(MAX_DATAGRAM)
|
||||
except socket.timeout:
|
||||
continue
|
||||
received += 1
|
||||
source = f"{address[0]}:{address[1]}"
|
||||
sources[source] += 1
|
||||
if not is_probe_packet(data):
|
||||
malformed += 1
|
||||
continue
|
||||
run_id, sequence, _ = HEADER.unpack_from(data)
|
||||
key = (source, run_id, sequence)
|
||||
if key in sequences:
|
||||
duplicates += 1
|
||||
else:
|
||||
sequences.add(key)
|
||||
try:
|
||||
sock.sendto(data, address)
|
||||
echoed += 1
|
||||
except OSError as exc:
|
||||
print(f"reply failed to {source}: {exc}", file=sys.stderr, flush=True)
|
||||
if time.monotonic() >= next_report:
|
||||
print(f"server progress received={received} unique={len(sequences)} echoed={echoed}", flush=True)
|
||||
next_report += 1
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
source_summary = ", ".join(f"{source}={count}" for source, count in sources.most_common()) or "none"
|
||||
print(
|
||||
f"server received={received} unique={len(sequences)} echoed={echoed} "
|
||||
f"malformed={malformed} duplicate_seq={duplicates} sources=[{source_summary}]",
|
||||
flush=True,
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def receive_replies(
|
||||
sock: socket.socket,
|
||||
pending: dict[int, int],
|
||||
rtts_ms: list[float],
|
||||
duplicates: list[int],
|
||||
wrong_source: list[int],
|
||||
receive_errors: list[str],
|
||||
run_id: bytes,
|
||||
target: tuple[str, int],
|
||||
wait_seconds: float,
|
||||
) -> None:
|
||||
deadline = time.monotonic() + wait_seconds
|
||||
while time.monotonic() < deadline:
|
||||
remaining = deadline - time.monotonic()
|
||||
readable, _, _ = select.select([sock], [], [], max(0.0, min(remaining, 0.05)))
|
||||
if not readable:
|
||||
continue
|
||||
try:
|
||||
data, address = sock.recvfrom(MAX_DATAGRAM)
|
||||
except OSError as exc:
|
||||
if not receive_errors:
|
||||
receive_errors.append(f"{type(exc).__name__}: {exc}")
|
||||
continue
|
||||
if address[1] != target[1] or address[0] not in (target[0], "::ffff:" + target[0]):
|
||||
wrong_source[0] += 1
|
||||
continue
|
||||
if not is_probe_packet(data):
|
||||
continue
|
||||
reply_run_id, sequence, sent_ns = HEADER.unpack_from(data)
|
||||
if reply_run_id != run_id:
|
||||
wrong_source[0] += 1
|
||||
continue
|
||||
if sequence not in pending:
|
||||
duplicates[0] += 1
|
||||
continue
|
||||
pending.pop(sequence)
|
||||
rtts_ms.append((time.monotonic_ns() - sent_ns) / 1_000_000)
|
||||
|
||||
|
||||
def run_client(args: argparse.Namespace) -> int:
|
||||
if args.size < HEADER.size or args.size > MAX_DATAGRAM:
|
||||
print(f"--size must be between {HEADER.size} and {MAX_DATAGRAM}", file=sys.stderr)
|
||||
return 2
|
||||
try:
|
||||
addresses = socket.getaddrinfo(args.target, args.port, type=socket.SOCK_DGRAM)
|
||||
except OSError as exc:
|
||||
print(f"resolve failed for {args.target}: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
family, socktype, proto, _, target = addresses[0]
|
||||
sock = socket.socket(family, socktype, proto)
|
||||
pending: dict[int, int] = {}
|
||||
rtts_ms: list[float] = []
|
||||
duplicates = [0]
|
||||
wrong_source = [0]
|
||||
receive_errors: list[str] = []
|
||||
send_errors = 0
|
||||
run_id = secrets.token_bytes(8)
|
||||
local_endpoint = None
|
||||
payload_padding = bytes(args.size - HEADER.size)
|
||||
|
||||
try:
|
||||
if args.socket_mode == "connected":
|
||||
sock.connect(target)
|
||||
start = time.monotonic()
|
||||
for sequence in range(args.count):
|
||||
send_at = start + sequence / args.rate
|
||||
while True:
|
||||
remaining = send_at - time.monotonic()
|
||||
if remaining <= 0:
|
||||
break
|
||||
receive_replies(sock, pending, rtts_ms, duplicates, wrong_source, receive_errors, run_id, target, max(0.0, min(remaining, 0.05)))
|
||||
|
||||
sent_ns = time.monotonic_ns()
|
||||
packet = HEADER.pack(run_id, sequence, sent_ns) + payload_padding
|
||||
try:
|
||||
if args.socket_mode == "connected":
|
||||
sock.send(packet)
|
||||
else:
|
||||
sock.sendto(packet, target)
|
||||
if local_endpoint is None:
|
||||
local_endpoint = sock.getsockname()
|
||||
pending[sequence] = sent_ns
|
||||
except OSError as exc:
|
||||
send_errors += 1
|
||||
print(f"send failed seq={sequence}: {exc}", file=sys.stderr, flush=True)
|
||||
|
||||
receive_replies(sock, pending, rtts_ms, duplicates, wrong_source, receive_errors, run_id, target, args.timeout)
|
||||
finally:
|
||||
sock.close()
|
||||
|
||||
sent = args.count - send_errors
|
||||
received = len(rtts_ms)
|
||||
loss = 0.0 if sent == 0 else 100.0 * (sent - received) / sent
|
||||
elapsed = args.count / args.rate
|
||||
print(f"target={args.target}:{args.port} local={local_endpoint} socket_mode={args.socket_mode} size={args.size} rate={args.rate:g} pps duration≈{elapsed:.2f}s run_id={run_id.hex()}")
|
||||
print(f"sent={sent} replies={received} timeout_loss={loss:.2f}% send_errors={send_errors} duplicates={duplicates[0]} wrong_source={wrong_source[0]} receive_errors={receive_errors}")
|
||||
print(f"rtt p50={format_ms(percentile(rtts_ms, 0.50))} p95={format_ms(percentile(rtts_ms, 0.95))} max={format_ms(max(rtts_ms) if rtts_ms else None)}")
|
||||
return 0 if received > 0 else 1
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
modes = parser.add_subparsers(dest="mode", required=True)
|
||||
server = modes.add_parser("server", help="run a bounded or continuous UDP echo receiver")
|
||||
server.add_argument("--bind", default="0.0.0.0")
|
||||
server.add_argument("--port", type=int, required=True)
|
||||
server.add_argument("--duration", type=float, default=45, help="seconds; 0 runs until Ctrl-C")
|
||||
server.set_defaults(run=run_server)
|
||||
client = modes.add_parser("client", help="send sequenced UDP probes and measure replies")
|
||||
client.add_argument("--target", required=True)
|
||||
client.add_argument("--port", type=int, required=True)
|
||||
client.add_argument("--count", type=int, default=251)
|
||||
client.add_argument("--rate", type=float, default=31, help="datagrams per second")
|
||||
client.add_argument("--size", type=int, default=64, help="UDP payload size in bytes")
|
||||
client.add_argument("--timeout", type=float, default=3, help="reply drain timeout in seconds")
|
||||
client.add_argument(
|
||||
"--socket-mode",
|
||||
choices=("sendto", "connected"),
|
||||
default="sendto",
|
||||
help="send with sendto() or a connected UDP socket; compare these when diagnosing VM networking",
|
||||
)
|
||||
client.set_defaults(run=run_client)
|
||||
return parser
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
arguments = build_parser().parse_args()
|
||||
if getattr(arguments, "count", 1) < 1 or getattr(arguments, "rate", 1) <= 0:
|
||||
raise SystemExit("--count must be positive and --rate must be greater than zero")
|
||||
raise SystemExit(arguments.run(arguments))
|
||||
Reference in new issue
Block a user