391 lines
19 KiB
Python
391 lines
19 KiB
Python
"""Windows ping spike monitor. Python 3.8+, standard library only."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import deque
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime
|
|
import ipaddress
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import socket
|
|
import statistics
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
|
|
RTT = re.compile(r"([<=]?)\s*(\d+(?:[.,]\d+)?)\s*ms\b", re.I)
|
|
IP = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
|
|
|
|
|
|
def timestamp():
|
|
return datetime.now().astimezone().isoformat(timespec="milliseconds")
|
|
|
|
|
|
def run_command(command, timeout, stop, capture_path=None):
|
|
"""Capture output without blocking shutdown or filling a pipe."""
|
|
if capture_path is not None:
|
|
# Send stdout/stderr straight to disk, preserving the complete original
|
|
# output even if this monitor is interrupted before tracing finishes.
|
|
with capture_path.open("wb", buffering=0) as capture:
|
|
with subprocess.Popen(command, stdout=capture, stderr=subprocess.STDOUT,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) as process:
|
|
deadline = time.monotonic() + timeout
|
|
try:
|
|
while process.poll() is None:
|
|
if stop.wait(0.2) or time.monotonic() >= deadline:
|
|
process.kill()
|
|
process.wait()
|
|
return -1, capture_path.read_bytes().decode(errors="replace")
|
|
except BaseException:
|
|
process.kill()
|
|
process.wait()
|
|
raise
|
|
return process.returncode, capture_path.read_bytes().decode(errors="replace")
|
|
with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) as process:
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
try:
|
|
output, _ = process.communicate(timeout=0.2)
|
|
return process.returncode, output.decode(errors="replace")
|
|
except subprocess.TimeoutExpired:
|
|
if stop.is_set() or time.monotonic() >= deadline:
|
|
process.kill()
|
|
output, _ = process.communicate()
|
|
return -1, output.decode(errors="replace") + "\n[Cancelled or command timed out]\n"
|
|
|
|
|
|
def parse_ping(output):
|
|
# TTL identifies an echo reply, excluding summaries and unreachable messages.
|
|
for line in output.splitlines():
|
|
if re.search(r"TTL\s*=\s*\d+", line, re.I):
|
|
match = RTT.search(line)
|
|
if match:
|
|
return 0.5 if match[1] == "<" else float(match[2].replace(",", "."))
|
|
raise ValueError("Echo reply received, but its latency could not be parsed: " + line)
|
|
return None
|
|
|
|
|
|
class Detector:
|
|
def __init__(self, threshold=100, multiplier=3, increase=40, window=60,
|
|
required=3, period=60, consecutive=2, start=None):
|
|
self.threshold = threshold
|
|
self.multiplier = multiplier
|
|
self.increase = increase
|
|
self.baseline = deque(maxlen=window)
|
|
self.required = required
|
|
self.period = period
|
|
self.consecutive = consecutive
|
|
self.period_start = time.monotonic() if start is None else start
|
|
self.period_bad = 0
|
|
self.streak = 0
|
|
|
|
def observe(self, latency, now=None):
|
|
now = time.monotonic() if now is None else now
|
|
elapsed_periods = int((now - self.period_start) // self.period)
|
|
completed_bad = None
|
|
trigger = False
|
|
if elapsed_periods:
|
|
completed_bad = self.period_bad
|
|
self.streak = self.streak + 1 if completed_bad >= self.required else 0
|
|
# Unobserved periods (e.g. computer sleep) break the streak.
|
|
if elapsed_periods > 1:
|
|
self.streak = 0
|
|
trigger = self.streak >= self.consecutive
|
|
self.period_start += elapsed_periods * self.period
|
|
self.period_bad = 0
|
|
baseline = statistics.median(self.baseline) if self.baseline else None
|
|
relative = (baseline is not None and len(self.baseline) >= 10
|
|
and latency is not None
|
|
and latency >= max(baseline * self.multiplier, baseline + self.increase))
|
|
bad = latency is None or latency >= self.threshold or relative
|
|
self.period_bad += int(bad)
|
|
# Freeze the healthy baseline during incidents so persistent lag stays visible.
|
|
if latency is not None and not bad:
|
|
self.baseline.append(latency)
|
|
return {"bad": bool(bad), "baseline_ms": baseline,
|
|
"trigger": trigger, "period_bad": self.period_bad,
|
|
"completed_period_bad": completed_bad, "consecutive_bad_periods": self.streak}
|
|
|
|
|
|
def parse_trace(output):
|
|
hops = []
|
|
for line in output.splitlines():
|
|
number = re.match(r"^\s*(\d+)\s+", line)
|
|
if not number:
|
|
continue
|
|
remainder = line[number.end():]
|
|
matches = list(RTT.finditer(remainder))
|
|
times = [0.5 if m[1] == "<" else float(m[2].replace(",", ".")) for m in matches]
|
|
addresses = IP.findall(remainder)
|
|
hostname_match = re.search(r"([^\s\[\]]+)\s+\[(?:\d{1,3}\.){3}\d{1,3}\]", remainder)
|
|
hops.append({"hop": int(number[1]), "ip": addresses[-1] if addresses else None,
|
|
"hostname": hostname_match[1] if hostname_match else None,
|
|
"rtt_ms": times, "median_ms": statistics.median(times) if times else None,
|
|
"timeouts": remainder.count("*")})
|
|
return hops
|
|
|
|
|
|
def sustained_jump(hops, destination, jump_ms):
|
|
"""Return structured evidence only when a jump persists to the destination."""
|
|
responding = [h for h in hops if h["median_ms"] is not None]
|
|
if not responding or responding[-1]["ip"] != destination:
|
|
return None
|
|
for index, hop in enumerate(responding):
|
|
previous = responding[index - 1] if index else None
|
|
floor = previous["median_ms"] if previous else 0
|
|
downstream = responding[index + 1:]
|
|
if (hop["median_ms"] - floor >= jump_ms and downstream
|
|
and all(h["median_ms"] - floor >= jump_ms for h in downstream)):
|
|
return {"previous": previous, "hop": hop, "floor_ms": floor,
|
|
"destination_ms": responding[-1]["median_ms"]}
|
|
return None
|
|
|
|
|
|
def analyze_trace(hops, destination, jump_ms):
|
|
responding = [h for h in hops if h["median_ms"] is not None]
|
|
if not responding:
|
|
return "Inconclusive: no hops returned timed replies."
|
|
if responding[-1]["ip"] != destination:
|
|
return "Inconclusive: destination did not return timed replies; cannot locate the end-to-end delay."
|
|
evidence = sustained_jump(hops, destination, jump_ms)
|
|
if evidence:
|
|
previous, hop, floor = evidence["previous"], evidence["hop"], evidence["floor_ms"]
|
|
start = f"hop {previous['hop']} ({previous['ip']})" if previous else "this computer"
|
|
return (f"Possible delay begins between {start} and hop {hop['hop']} ({hop['ip']}); "
|
|
f"RTT rises from {floor:.1f} to {hop['median_ms']:.1f} ms and stays elevated "
|
|
"through the destination. This is a clue, not proof of a faulty router.")
|
|
if len(responding) >= 2 and responding[-1]["median_ms"] - responding[-2]["median_ms"] >= jump_ms:
|
|
return "Delay appears only at the destination/final segment; the destination may also deprioritize ICMP."
|
|
return ("No sustained downstream latency jump found. Isolated slow or silent hops do not identify "
|
|
"a forwarding problem; the spike may have ended before tracing reached it.")
|
|
|
|
|
|
class Log:
|
|
def __init__(self, directory):
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
self.directory = directory
|
|
self.file = (directory / "events.jsonl").open("a", encoding="utf-8")
|
|
self.lock = threading.Lock()
|
|
|
|
def write(self, event, message=None, **fields):
|
|
with self.lock:
|
|
now = timestamp()
|
|
self.file.write(json.dumps({"time": now, "event": event, **fields}) + "\n")
|
|
self.file.flush()
|
|
if message:
|
|
print(f"[{now}] {message}", flush=True)
|
|
|
|
def ping(self, target, latency, state):
|
|
label = "NO REPLY" if latency is None else f"{latency:.1f} ms"
|
|
message = f"{target}: {label}"
|
|
if state["bad"]:
|
|
self.write("ping", message + " [SPIKE/LOSS]",
|
|
target=target, latency_ms=latency, **state)
|
|
else:
|
|
with self.lock:
|
|
print(f"[{timestamp()}] {message}", flush=True)
|
|
|
|
|
|
def trace(target, args, log, stop, baseline=None, incident_id=None, attempt=1):
|
|
name = datetime.now().strftime("trace_%Y%m%d_%H%M%S_%f")
|
|
command = ["tracert", "-4", "-h", str(args.max_hops), "-w", str(args.trace_timeout), target]
|
|
raw_path = log.directory / (name + ".pending.txt")
|
|
log.write("trace_start", f"Running tracert to {target} (attempt {attempt}); keeping output only if a spike is captured.",
|
|
command=command, incident_id=incident_id, attempt=attempt)
|
|
try:
|
|
code, output = run_command(command, args.max_hops * 3 * args.trace_timeout / 1000 + 30,
|
|
stop, capture_path=raw_path)
|
|
except BaseException:
|
|
raw_path.unlink(missing_ok=True)
|
|
raise
|
|
hops = parse_trace(output)
|
|
assessment = analyze_trace(hops, target, args.hop_jump)
|
|
evidence = sustained_jump(hops, target, args.hop_jump)
|
|
spike_limit = args.threshold
|
|
if baseline is not None:
|
|
spike_limit = min(spike_limit, max(baseline * args.multiplier, baseline + args.increase))
|
|
# Preserve transient samples even when medians hide them or later hops recover.
|
|
# Capturing a slow diagnostic reply is separate from locating a forwarding fault.
|
|
spike_probes = [{"hop": hop["hop"], "ip": hop["ip"], "hostname": hop["hostname"],
|
|
"rtt_ms": rtt}
|
|
for hop in hops for rtt in hop["rtt_ms"] if rtt >= spike_limit]
|
|
captured = bool(spike_probes)
|
|
peak_ms = max((rtt for hop in hops for rtt in hop["rtt_ms"]), default=None)
|
|
if captured and not evidence:
|
|
assessment = "Individual slow trace probe captured; this alone does not locate a faulty hop. " + assessment
|
|
if code == -1:
|
|
assessment = "Trace cancelled or timed out. " + assessment
|
|
if captured:
|
|
saved_path = log.directory / (name + ".txt")
|
|
raw_path.replace(saved_path)
|
|
else:
|
|
raw_path.unlink(missing_ok=True)
|
|
saved_path = None
|
|
report = {"time": timestamp(), "target": target, "returncode": code, "command": command,
|
|
"assessment": assessment, "hops": hops, "raw_file": str(saved_path) if saved_path else None,
|
|
"spike_captured": captured, "spike_limit_ms": spike_limit,
|
|
"spike_probes": spike_probes, "spike_probe_count": len(spike_probes),
|
|
"peak_probe_ms": peak_ms, "sustained_jump_found": evidence is not None,
|
|
"incident_id": incident_id, "attempt": attempt}
|
|
if captured:
|
|
(log.directory / (name + ".json")).write_text(json.dumps(report, indent=2), encoding="utf-8")
|
|
# Keep only the short result in the append-only event log for discarded attempts.
|
|
summary = {key: value for key, value in report.items() if key not in ("hops", "spike_probes")}
|
|
suffix = f" Saved: {saved_path}" if captured else " Output discarded (no captured spike)."
|
|
log.write("trace_complete", assessment + suffix, **summary)
|
|
return report
|
|
|
|
|
|
def trace_incident(target, args, log, stop, baseline):
|
|
incident_id = datetime.now().strftime("incident_%Y%m%d_%H%M%S_%f")
|
|
attempt = 0
|
|
log.write("incident_start", "Incident detected: repeating tracert until a spike is captured.",
|
|
incident_id=incident_id, baseline_ms=baseline)
|
|
while not stop.is_set():
|
|
attempt += 1
|
|
report = trace(target, args, log, stop, baseline, incident_id, attempt)
|
|
if report["spike_captured"]:
|
|
log.write("incident_captured", f"Spike captured on trace {attempt}. Returning to ping monitoring.",
|
|
incident_id=incident_id, attempts=attempt, raw_file=report["raw_file"])
|
|
return
|
|
if stop.is_set():
|
|
break
|
|
log.write("trace_retry", f"No trace probe crossed the spike threshold; next trace in {args.trace_retry:g}s.",
|
|
incident_id=incident_id, attempt=attempt)
|
|
if stop.wait(args.trace_retry):
|
|
break
|
|
log.write("incident_cancelled", "Trace capture stopped before a spike was captured.",
|
|
incident_id=incident_id, attempts=attempt)
|
|
|
|
|
|
def positive_float(value):
|
|
result = float(value)
|
|
if not 0 < result < float("inf"):
|
|
raise argparse.ArgumentTypeError("must be a finite number greater than zero")
|
|
return result
|
|
|
|
|
|
def positive_int(value):
|
|
result = int(value)
|
|
if result <= 0:
|
|
raise argparse.ArgumentTypeError("must be greater than zero")
|
|
return result
|
|
|
|
|
|
def arguments():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--target", default="1.1.1.1", help="IPv4 address or hostname (default: 1.1.1.1)")
|
|
parser.add_argument("--interval", type=positive_float, default=1, help="seconds between probes")
|
|
parser.add_argument("--threshold", type=positive_float, default=100, help="absolute spike threshold in ms")
|
|
parser.add_argument("--multiplier", type=positive_float, default=3, help="relative baseline multiplier")
|
|
parser.add_argument("--increase", type=positive_float, default=40, help="minimum relative increase in ms")
|
|
parser.add_argument("--required", type=positive_int, default=3, help="bad probes needed per period (default: 3)")
|
|
parser.add_argument("--period", type=positive_float, default=60, help="period duration in seconds (default: 60)")
|
|
parser.add_argument("--consecutive", type=positive_int, default=2, help="consecutive bad periods needed (default: 2)")
|
|
parser.add_argument("--baseline-window", type=positive_int, default=60)
|
|
parser.add_argument("--cooldown", type=positive_float, default=120, help="seconds after capture before another incident can start")
|
|
parser.add_argument("--trace-retry", type=positive_float, default=2, help="seconds between incident trace attempts (default: 2)")
|
|
parser.add_argument("--ping-timeout", type=positive_int, default=1000, help="ping timeout in ms")
|
|
parser.add_argument("--trace-timeout", type=positive_int, default=500, help="per-hop probe timeout in ms")
|
|
parser.add_argument("--max-hops", type=positive_int, default=30)
|
|
parser.add_argument("--hop-jump", type=positive_float, default=30, help="sustained hop RTT increase in ms")
|
|
parser.add_argument("--log-dir", type=Path, default=Path(__file__).resolve().parent / "logs")
|
|
parser.add_argument("--count", type=positive_int, help="stop after this many probes (default: continuous)")
|
|
parser.add_argument("--trace-now", action="store_true", help="also run a diagnostic immediately")
|
|
args = parser.parse_args()
|
|
if args.max_hops > 255:
|
|
parser.error("--max-hops cannot exceed 255")
|
|
if args.baseline_window < 10:
|
|
parser.error("--baseline-window must be at least 10")
|
|
return args
|
|
|
|
|
|
def main():
|
|
args = arguments()
|
|
if sys.platform != "win32":
|
|
print("This program requires Windows (ping.exe and tracert.exe).", file=sys.stderr)
|
|
return 1
|
|
try:
|
|
target = socket.gethostbyname(args.target)
|
|
ipaddress.IPv4Address(target)
|
|
except (OSError, ValueError) as error:
|
|
print(f"Cannot resolve target: {error}", file=sys.stderr)
|
|
return 1
|
|
stop = threading.Event()
|
|
log = Log(args.log_dir.resolve())
|
|
detector = Detector(args.threshold, args.multiplier, args.increase,
|
|
args.baseline_window, args.required, args.period, args.consecutive)
|
|
executor = ThreadPoolExecutor(max_workers=1)
|
|
pending = None
|
|
next_incident = float("-inf")
|
|
pending_incident = False
|
|
samples = 0
|
|
failed = False
|
|
try:
|
|
log.write("start", f"Monitoring {args.target} ({target}). Ctrl+C stops. Logs: {log.directory}",
|
|
target=target, settings={k: str(v) if isinstance(v, Path) else v for k, v in vars(args).items()})
|
|
log.write("trigger_rule", f"Trace after {args.required}+ spikes/losses per {args.period:g}s "
|
|
f"period for {args.consecutive} consecutive periods.")
|
|
if args.trace_now:
|
|
pending = executor.submit(trace, target, args, log, stop)
|
|
while not stop.is_set() and (args.count is None or samples < args.count):
|
|
started = time.monotonic()
|
|
if pending is not None and pending.done():
|
|
pending.result()
|
|
pending = None
|
|
if pending_incident:
|
|
next_incident = time.monotonic() + args.cooldown
|
|
pending_incident = False
|
|
code, output = run_command(["ping", "-4", "-n", "1", "-w", str(args.ping_timeout), target],
|
|
args.ping_timeout / 1000 + 5, stop)
|
|
if stop.is_set():
|
|
break
|
|
latency = parse_ping(output)
|
|
if code == 0 and latency is None:
|
|
raise ValueError("ping reported success but no echo reply could be parsed: " + output)
|
|
state = detector.observe(latency)
|
|
if state["completed_period_bad"] is not None:
|
|
log.write("period_complete", f"Period: {state['completed_period_bad']} spikes/losses; "
|
|
f"consecutive qualifying periods: {state['consecutive_bad_periods']}.",
|
|
bad_count=state["completed_period_bad"],
|
|
streak=state["consecutive_bad_periods"])
|
|
samples += 1
|
|
log.ping(target, latency, state)
|
|
if state["trigger"] and pending is None and time.monotonic() >= next_incident:
|
|
baseline = state["baseline_ms"] if len(detector.baseline) >= 10 else None
|
|
pending_incident = True
|
|
pending = executor.submit(trace_incident, target, args, log, stop, baseline)
|
|
if args.count is None or samples < args.count:
|
|
stop.wait(max(0, args.interval - (time.monotonic() - started)))
|
|
if pending:
|
|
if pending_incident:
|
|
stop.set() # --count remains a bounded monitoring run.
|
|
log.write("waiting", "Finishing diagnostic; Ctrl+C cancels.")
|
|
while not pending.done():
|
|
time.sleep(0.2)
|
|
pending.result()
|
|
except KeyboardInterrupt:
|
|
print("\nStopping...", flush=True)
|
|
except Exception as error:
|
|
failed = True
|
|
log.write("error", f"Error: {error}", error=str(error))
|
|
finally:
|
|
stop.set()
|
|
# Python 3.8 has no shutdown(cancel_futures=...). There is at most
|
|
# one submitted trace; cancel it if queued, or let stop end it if running.
|
|
if pending is not None:
|
|
pending.cancel()
|
|
executor.shutdown(wait=True)
|
|
log.write("stop", "Monitor stopped.", samples=samples)
|
|
log.file.close()
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|