Compare commits

...

2 Commits

Author SHA1 Message Date
olieske 20740b3704 feat: TCP connect fallback for ping, flag high latency vs. ping
- If a peer filters ICMP, measure the TCP connect time to its peer port
  instead; such values are marked with a trailing "t"
- Show ⚡ between Latency and Ping when stellar-core's latency exceeds
  the ping/TCP RTT by more than LATENCY_PING_RATIO (3), hinting at a
  VPN exit / relay in front of the actual node
- Legend below the peer table, only for markers actually shown

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-23 08:40:31 +00:00
olieske 602592e304 feat: add ICMP ping column to peer table
- ping_batch() pings each peer's public IP in parallel from the host
  (system ping, 1 packet, 1 s timeout, private IPs skipped)
- New "Ping" column next to stellar-core's "Latency" in the -p/-g table;
  "—" when the peer filters ICMP
- Low ping with high latency hints at a VPN exit / relay in front of
  the actual node

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-23 08:21:36 +00:00
2 changed files with 111 additions and 3 deletions
+2
View File
@@ -49,6 +49,7 @@ Single self-contained script (`pi-stat`). Logic is split into helper functions a
- **PostgreSQL** — `psql` inside the container; used for peer DB counts and pubkey→IP mapping - **PostgreSQL** — `psql` inside the container; used for peer DB counts and pubkey→IP mapping
- **ZFS CLI** — `zpool`/`zfs` commands on the host for disk health (`z01pool`); if `zpool` is not found (`HAS_ZFS = False`), falls back to `shutil.disk_usage()` — works on Linux without ZFS, WSL2, and Windows - **ZFS CLI** — `zpool`/`zfs` commands on the host for disk health (`z01pool`); if `zpool` is not found (`HAS_ZFS = False`), falls back to `shutil.disk_usage()` — works on Linux without ZFS, WSL2, and Windows
- **ip-api.com** — batch geolocation (Country, City), cached in `_geo_cache`; only fetched with `-g` - **ip-api.com** — batch geolocation (Country, City), cached in `_geo_cache`; only fetched with `-g`
- **System `ping` / TCP connect** — per peer from the host (`ping_batch()`, parallel, private IPs skipped): one ICMP ping (1 s timeout); if the peer filters ICMP, falls back to a TCP connect on the peer port from its address (1 s timeout). Shown as `Ping` next to stellar-core's `Latency` in the `-p`/`-g` table; TCP values are marked with a trailing `t` (legend printed below the table), `—` if neither answers
### Output sections (in order) ### Output sections (in order)
@@ -75,6 +76,7 @@ The refresh deadline is anchored to the start of the cycle (`cycle_start`), not
- **Dynamic column widths** — all table columns sized at runtime from actual data length - **Dynamic column widths** — all table columns sized at runtime from actual data length
- **NAT detection** — inbound peers with private IPs are replaced with their DB-stored public IP (`db_peer_addr` from PostgreSQL) - **NAT detection** — inbound peers with private IPs are replaced with their DB-stored public IP (`db_peer_addr` from PostgreSQL)
- **Validator matching** — `load_validators()` parses `stellar-core.cfg` and indexes by IP, name, and public key; matched peers show a truncated key + `(Vn)` label - **Validator matching** — `load_validators()` parses `stellar-core.cfg` and indexes by IP, name, and public key; matched peers show a truncated key + `(Vn)` label
- **Ping vs. Latency** — `Latency` is stellar-core's round trip to the peer *process*, `Ping` is ICMP to the peer's IP; a low ping with high latency hints at a VPN exit / relay in front of the actual node. Peers with `Latency > LATENCY_PING_RATIO (3) × Ping` get a `⚡` between the two columns (2-column cell, since `⚡` is double-width in terminals); a legend line is printed below the table only when a `⚡` or `t` marker is present
- **Geo caching** — `_geo_cache` is a module-level dict; `geo_batch()` only fetches IPs not already cached - **Geo caching** — `_geo_cache` is a module-level dict; `geo_batch()` only fetches IPs not already cached
- **Flag parsing** — manual `sys.argv` parsing; `-g` implies `-p`; `-i` accepts an optional next argument as the refresh interval - **Flag parsing** — manual `sys.argv` parsing; `-g` implies `-p`; `-i` accepts an optional next argument as the refresh interval
- **Config precedence** — `config_refresh()` reads the config value; an explicit `-i SEC` argument overrides it - **Config precedence** — `config_refresh()` reads the config value; an explicit `-i SEC` argument overrides it
+109 -3
View File
@@ -18,6 +18,7 @@ import socket
import subprocess import subprocess
import sys import sys
import time import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone from datetime import datetime, timezone
from urllib.request import urlopen, Request from urllib.request import urlopen, Request
from urllib.error import URLError from urllib.error import URLError
@@ -42,6 +43,7 @@ HORIZON_URL = "http://127.0.0.1:8000" # queried from inside the contai
CFG_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stellar-core.cfg") CFG_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stellar-core.cfg")
DEFAULT_REFRESH = 5 DEFAULT_REFRESH = 5
REFRESH_WARN_RATIO = 0.8 # warn if a cycle takes >= 80% of the refresh interval REFRESH_WARN_RATIO = 0.8 # warn if a cycle takes >= 80% of the refresh interval
LATENCY_PING_RATIO = 3 # mark peers with ⚡ when core latency > 3× ping/TCP RTT
CONFIG_PATHS = [ CONFIG_PATHS = [
os.environ.get("PI_STAT_CONFIG"), os.environ.get("PI_STAT_CONFIG"),
os.path.join(os.path.expanduser("~"), ".config", "pi-stat", "config.json"), os.path.join(os.path.expanduser("~"), ".config", "pi-stat", "config.json"),
@@ -151,6 +153,83 @@ def geo_str(ip):
return ", ".join(p for p in geo if p) return ", ".join(p for p in geo if p)
return "" return ""
HAS_PING = shutil.which("ping") is not None
_PING_RE = re.compile(r"[=<]\s*([\d.,]+)\s*ms", re.IGNORECASE)
def ping_rtt(ip, timeout_s=1):
"""Single ICMP ping via the system 'ping'. Returns RTT in ms, or None (no reply / filtered)."""
if os.name == "nt":
cmd = ["ping", "-n", "1", "-w", str(timeout_s * 1000), ip]
else:
cmd = ["ping", "-c", "1", "-W", str(timeout_s), "-n", ip]
try:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_s + 2,
env={**os.environ, "LC_ALL": "C"})
except Exception:
return None
if r.returncode != 0:
return None
# Linux: "time=37.5 ms", Windows: "time=37ms" / "Zeit<1ms"
for line in r.stdout.splitlines():
if "ttl" in line.lower():
mt = _PING_RE.search(line)
if mt:
try:
return float(mt.group(1).replace(",", "."))
except ValueError:
pass
return None
def tcp_rtt(ip, port, timeout_s=1):
"""TCP connect time to ip:port in ms (≈ 1 RTT: SYN → SYN/ACK), or None if not connectable."""
t = time.perf_counter()
try:
with socket.create_connection((ip, port), timeout=timeout_s):
return (time.perf_counter() - t) * 1000
except OSError:
return None
def probe_rtt(ip, port):
"""ICMP ping first; if the peer filters ICMP, fall back to a TCP connect on its peer port.
Returns (rtt_ms, "icmp" | "tcp") or (None, None)."""
if HAS_PING:
v = ping_rtt(ip)
if v is not None:
return v, "icmp"
if port:
v = tcp_rtt(ip, port)
if v is not None:
return v, "tcp"
return None, None
def ping_batch(addrs):
"""Probe all public peer addresses ("ip:port") in parallel.
Returns {ip: (rtt_ms, kind)}; private/invalid IPs are skipped."""
targets = {}
for addr in addrs:
ip, _, port = addr.rpartition(":")
try:
if ipaddress.ip_address(ip).is_private:
continue
except ValueError:
continue
targets[ip] = int(port) if port.isdigit() else None
if not targets:
return {}
with ThreadPoolExecutor(max_workers=min(32, len(targets))) as ex:
return dict(zip(targets, ex.map(lambda t: probe_rtt(*t), targets.items())))
def fmt_ms(v):
"""Format a millisecond value: one decimal below 10 ms, integer above."""
if v is None:
return "—"
return f"{v:.1f}ms" if v < 10 else f"{v:.0f}ms"
def fmt_ping(res):
"""Ping cell: ICMP value plain, TCP-connect value marked with a trailing 't'."""
v, kind = res if res else (None, None)
return f"{fmt_ms(v)} {'t' if kind == 'tcp' else ' '}"
def load_config(): def load_config():
"""Read the first existing config file (JSON object). Returns {} if none is usable.""" """Read the first existing config file (JSON object). Returns {} if none is usable."""
for path in CONFIG_PATHS: for path in CONFIG_PATHS:
@@ -549,6 +628,12 @@ def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, e
_all_ips.append(node_ip_str) _all_ips.append(node_ip_str)
geo_batch([ip for ip in _all_ips if ip]) geo_batch([ip for ip in _all_ips if ip])
# --- Ping from this host to each peer's known IP ---
# Compared with stellar-core's "Latency", a low ping but high
# latency hints at a VPN exit / relay in front of the actual node.
# ICMP first, TCP connect to the peer port as fallback (marked 't').
ping_ms = ping_batch([p.get("address", "") for p in inbound + outbound])
def _is_private(ip): def _is_private(ip):
try: try:
return ipaddress.ip_address(ip).is_private return ipaddress.ip_address(ip).is_private
@@ -643,10 +728,12 @@ def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, e
ratio_total = max(ratio_total, len("RX/TX")) ratio_total = max(ratio_total, len("RX/TX"))
_traffic_hdr = f" {'RX/TX':>{ratio_total}}" _traffic_hdr = f" {'RX/TX':>{ratio_total}}"
_traffic_node = f" {fmt_ratio_cell(None, ratio_l_w, ratio_r_w):>{ratio_total}}" _traffic_node = f" {fmt_ratio_cell(None, ratio_l_w, ratio_r_w):>{ratio_total}}"
HDR = f" {hdr_num_prefix}{'NodeID':<{col_w}} {'Version':<{ver_w}}{_ip_hdr_part}{_geo_hdr_part} {'Latency':>8} {'Uptime':<8}{_traffic_hdr} Flow" HDR = f" {hdr_num_prefix}{'NodeID':<{col_w}} {'Version':<{ver_w}}{_ip_hdr_part}{_geo_hdr_part} {'Latency':>8} {'':2} {'Ping':>9} {'Uptime':<8}{_traffic_hdr} Flow"
NODE_ROW = f" {node_row_prefix}{node_id_str:<{col_w}} {node_ver_str:<{ver_w}}{_ip_node_part}{_geo_node_part} {'—':>8} {node_up_str:<8}{_traffic_node} {node_flow_str}" NODE_ROW = f" {node_row_prefix}{node_id_str:<{col_w}} {node_ver_str:<{ver_w}}{_ip_node_part}{_geo_node_part} {'—':>8} {'':2} {fmt_ping(None):>9} {node_up_str:<8}{_traffic_node} {node_flow_str}"
SEP = " " + "─" * (len(HDR) - 2) SEP = " " + "─" * (len(HDR) - 2)
lat_flagged = [] # peers marked ⚡ (for the legend)
def show(peers, is_out): def show(peers, is_out):
for p in sorted(peers, key=lambda p: p.get("elapsed", 0), reverse=True): for p in sorted(peers, key=lambda p: p.get("elapsed", 0), reverse=True):
raw_addr = p.get("address", "?").rsplit(":", 1)[0] raw_addr = p.get("address", "?").rsplit(":", 1)[0]
@@ -662,6 +749,16 @@ def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, e
ver = fmt_version(p.get("ver", "")) ver = fmt_version(p.get("ver", ""))
lat = p.get("latency", "?") lat = p.get("latency", "?")
lat_s = "n/a" if lat == 86400000 else f"{lat}ms" lat_s = "n/a" if lat == 86400000 else f"{lat}ms"
ping_res = ping_ms.get(raw_addr)
ping_s = fmt_ping(ping_res)
# ⚡ when stellar-core's latency exceeds the network RTT by more than
# LATENCY_PING_RATIO (possible VPN exit / relay in front of the node).
# "⚡" is 2 columns wide in the terminal, so the blank is 2 spaces.
rtt = ping_res[0] if ping_res else None
lat_flag = ("⚡" if isinstance(lat, (int, float)) and lat != 86400000
and rtt and lat > LATENCY_PING_RATIO * rtt else " ")
if lat_flag == "⚡":
lat_flagged.append(pid)
up = fmt_up(p.get("elapsed", 0)) up = fmt_up(p.get("elapsed", 0))
rx = p.get("byte_read", 0) or 0 rx = p.get("byte_read", 0) or 0
tx = p.get("byte_write", 0) or 0 tx = p.get("byte_write", 0) or 0
@@ -690,7 +787,7 @@ def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, e
else: else:
_parts = fmt_ratio_parts(rx, tx) _parts = fmt_ratio_parts(rx, tx)
_traffic_row = f" {fmt_ratio_cell(_parts, ratio_l_w, ratio_r_w):>{ratio_total}}" _traffic_row = f" {fmt_ratio_cell(_parts, ratio_l_w, ratio_r_w):>{ratio_total}}"
print(f" {row_prefix}{node_col:<{col_w}} {ver:<{ver_w}}{_ip_row_part}{_geo_row_part} {lat_s:>8} {up:<8}{_traffic_row} {flow}") print(f" {row_prefix}{node_col:<{col_w}} {ver:<{ver_w}}{_ip_row_part}{_geo_row_part} {lat_s:>8} {lat_flag} {ping_s:>9} {up:<8}{_traffic_row} {flow}")
print() print()
print(HDR) print(HDR)
@@ -704,6 +801,15 @@ def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, e
print() print()
print(" Outbound") print(" Outbound")
show(outbound, True) show(outbound, True)
legend = []
if lat_flagged:
legend.append(f"⚡ = Latency > {LATENCY_PING_RATIO}× Ping (possible VPN exit / relay)")
if any(kind == "tcp" for _, kind in ping_ms.values()):
legend.append("Ping: t = TCP connect to peer port (ICMP filtered)")
if legend:
print()
for line in legend:
print(f" {line}")
print() print()
return outbound_numbered, container return outbound_numbered, container