olieske
4e27efc3b9
- Optional JSON config ($PI_STAT_CONFIG, ~/.config/pi-stat/config.json, <script dir>/pi-stat.json) with refresh_interval; -i SEC overrides it - Refresh deadline anchored to cycle start; warn when a cycle takes >= 80% of the refresh interval or exceeds it - Switch the tty to cbreak once per session so keypresses made during a long cycle are not dropped - Ship pi-stat.json with the repo Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
946 lines
38 KiB
Python
Executable File
946 lines
38 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# =============================================================
|
||
# pi-stat
|
||
# Fast alternative to 'pi-node status'
|
||
# - Disk info directly via ZFS (no 'du', no long scan)
|
||
# - Peer data directly via stellar-core (no Horizon detour)
|
||
# =============================================================
|
||
|
||
import contextlib
|
||
import io
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import re
|
||
import select
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from urllib.request import urlopen, Request
|
||
from urllib.error import URLError
|
||
|
||
try:
|
||
import tty
|
||
import termios
|
||
HAS_TTY = True
|
||
except ImportError:
|
||
HAS_TTY = False
|
||
|
||
try:
|
||
import docker
|
||
except ImportError:
|
||
print("ERROR: 'docker' package missing. Install with: pip install docker")
|
||
sys.exit(1)
|
||
|
||
# --- Configuration ---
|
||
HAS_ZFS = shutil.which("zpool") is not None
|
||
CANDIDATE_CONTAINERS = ["mainnet", "testnet2", "testnet"]
|
||
HORIZON_URL = "http://localhost:31401"
|
||
CFG_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stellar-core.cfg")
|
||
DEFAULT_REFRESH = 5
|
||
REFRESH_WARN_RATIO = 0.8 # warn if a cycle takes >= 80% of the refresh interval
|
||
CONFIG_PATHS = [
|
||
os.environ.get("PI_STAT_CONFIG"),
|
||
os.path.join(os.path.expanduser("~"), ".config", "pi-stat", "config.json"),
|
||
os.path.join(os.path.dirname(os.path.realpath(__file__)), "pi-stat.json"),
|
||
]
|
||
|
||
# --- Colors / Icons ---
|
||
GREEN = "\033[0;32m"
|
||
YELLOW = "\033[1;33m"
|
||
RED = "\033[0;31m"
|
||
BOLD = "\033[1m"
|
||
RESET = "\033[0m"
|
||
|
||
def ok(msg): print(f" {GREEN}✅{RESET} {msg}")
|
||
def warn(msg): print(f" {YELLOW}⚠️{RESET} {msg}")
|
||
def err(msg): print(f" {RED}❌{RESET} {msg}")
|
||
def hdr(title):
|
||
print(f"\n{BOLD}{title}{RESET}")
|
||
print("=" * len(title))
|
||
|
||
# --- Helper functions ---
|
||
def fmt_up(s):
|
||
s = int(s)
|
||
if s < 60: return f"{s}s"
|
||
if s < 3600: return f"{s//60}m{s%60:02d}s"
|
||
h = s // 3600; m = (s % 3600) // 60
|
||
return f"{h}h{m:02d}m"
|
||
|
||
def fmt_version(ver_str):
|
||
"""Extract vMAJOR.MINOR from a stellar-core version string."""
|
||
m = re.search(r'(\d+\.\d+)', ver_str or "")
|
||
return f"v{m.group(1)}" if m else "—"
|
||
|
||
def _f1(v):
|
||
s = f"{v:.1f}"
|
||
return s[:-2] if s.endswith('.0') else s
|
||
|
||
def fmt_bytes(b):
|
||
if b >= 1024**3: return f"{b/1024**3:.2f}GiB"
|
||
if b >= 1024**2: return f"{_f1(b/1024**2)}MiB"
|
||
return f"{b/1024:.0f}KiB"
|
||
|
||
def fmt_rate(bps):
|
||
if bps >= 1e9: return f"{bps/1e9:.2f}Gbps"
|
||
if bps >= 1e6: return f"{_f1(bps/1e6)}Mbps"
|
||
if bps >= 1e3: return f"{bps/1e3:.0f}kbps"
|
||
return f"{bps:.0f}bps"
|
||
|
||
def fmt_ratio_parts(rx, tx):
|
||
"""Return (left, right) tuple for aligned-colon display, or None if no data."""
|
||
if rx == 0 and tx == 0:
|
||
return None
|
||
if tx == 0:
|
||
return ("∞", "1")
|
||
if rx == 0:
|
||
return ("1", "∞")
|
||
r = rx / tx
|
||
def _fv(v): return _f1(v) if v < 10 else f"{v:.0f}"
|
||
if r >= 1:
|
||
return (_fv(r), "1")
|
||
else:
|
||
return ("1", _fv(1/r))
|
||
|
||
def fmt_ratio_cell(parts, l_w, r_w):
|
||
"""Render a ratio cell with the colon at a fixed column position."""
|
||
if parts is None:
|
||
return f"{'—':^{l_w + 1 + r_w}}"
|
||
l, r = parts
|
||
return f"{l:>{l_w}}:{r:<{r_w}}"
|
||
|
||
_geo_cache = {}
|
||
|
||
def geo_batch(ips):
|
||
"""Batch-fetch geolocation for multiple IPs via ip-api.com. Updates _geo_cache."""
|
||
to_fetch = [ip for ip in ips if ip not in _geo_cache and ip and ip != "—"]
|
||
if not to_fetch:
|
||
return
|
||
try:
|
||
payload = json.dumps([{"query": ip} for ip in to_fetch[:100]]).encode()
|
||
req = Request(
|
||
"http://ip-api.com/batch?fields=status,query,countryCode,city",
|
||
data=payload,
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
with urlopen(req, timeout=5) as resp:
|
||
for item in json.loads(resp.read()):
|
||
ip = item.get("query")
|
||
if item.get("status") == "success":
|
||
_geo_cache[ip] = (item.get("countryCode", ""), item.get("city", ""))
|
||
else:
|
||
_geo_cache[ip] = None
|
||
except Exception:
|
||
for ip in to_fetch:
|
||
_geo_cache[ip] = None
|
||
|
||
def geo_label(ip):
|
||
"""Format IP with geolocation info."""
|
||
geo = _geo_cache.get(ip)
|
||
if geo:
|
||
return f"{ip} {', '.join(p for p in geo if p)}"
|
||
return ip
|
||
|
||
def geo_str(ip):
|
||
"""Return geo string (e.g. 'DE, Berlin') for an IP, or empty string if unavailable."""
|
||
geo = _geo_cache.get(ip)
|
||
if geo:
|
||
return ", ".join(p for p in geo if p)
|
||
return ""
|
||
|
||
def load_config():
|
||
"""Read the first existing config file (JSON object). Returns {} if none is usable."""
|
||
for path in CONFIG_PATHS:
|
||
if not path or not os.path.isfile(path):
|
||
continue
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
if isinstance(data, dict):
|
||
return data
|
||
print(f"WARN: Config {path}: kein JSON-Objekt – wird ignoriert", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"WARN: Config {path} nicht lesbar: {e}", file=sys.stderr)
|
||
return {}
|
||
return {}
|
||
|
||
|
||
def config_refresh(cfg):
|
||
"""refresh_interval from config, clamped to >= 1 s; falls back to DEFAULT_REFRESH."""
|
||
val = cfg.get("refresh_interval", DEFAULT_REFRESH)
|
||
try:
|
||
return max(1, int(val))
|
||
except (TypeError, ValueError):
|
||
print(f"WARN: refresh_interval '{val}' ungültig – benutze {DEFAULT_REFRESH}s", file=sys.stderr)
|
||
return DEFAULT_REFRESH
|
||
|
||
|
||
def load_validators(cfg_path):
|
||
"""Parse [[VALIDATORS]] sections, returning {ip_or_name: (index, public_key)}."""
|
||
if not os.path.exists(cfg_path):
|
||
print(f"ERROR: stellar-core.cfg not found at {cfg_path}")
|
||
sys.exit(1)
|
||
result = {}
|
||
idx = 0
|
||
current_ip = current_key = current_name = None
|
||
try:
|
||
with open(cfg_path) as f:
|
||
for line in f:
|
||
if re.match(r'^\s*\[\[VALIDATORS\]\]', line):
|
||
if current_key:
|
||
idx += 1
|
||
if current_ip: result[current_ip] = (idx, current_key)
|
||
if current_name: result[current_name] = (idx, current_key)
|
||
result[current_key] = (idx, current_key)
|
||
current_ip = current_key = current_name = None
|
||
m = re.match(r'^\s*ADDRESS\s*=\s*"([^":]+)', line)
|
||
if m: current_ip = m.group(1)
|
||
m = re.match(r'^\s*PUBLIC_KEY\s*=\s*"([^"]+)"', line)
|
||
if m: current_key = m.group(1)
|
||
m = re.match(r'^\s*NAME\s*=\s*"([^"]+)"', line)
|
||
if m: current_name = m.group(1)
|
||
if current_key:
|
||
idx += 1
|
||
if current_ip: result[current_ip] = (idx, current_key)
|
||
if current_name: result[current_name] = (idx, current_key)
|
||
result[current_key] = (idx, current_key)
|
||
except OSError:
|
||
pass
|
||
return result
|
||
|
||
def zfs_get(cmd):
|
||
r = subprocess.run(cmd, capture_output=True, text=True)
|
||
return r.stdout.strip() if r.returncode == 0 else None
|
||
|
||
def detect_container(client):
|
||
"""Return (container, name) for the first running candidate, or first found."""
|
||
found = None
|
||
for name in CANDIDATE_CONTAINERS:
|
||
try:
|
||
c = client.containers.get(name)
|
||
if c.status == "running":
|
||
return c, name
|
||
if found is None:
|
||
found = (c, name)
|
||
except docker.errors.NotFound:
|
||
pass
|
||
return found if found else (None, None)
|
||
|
||
def exec_json(container, command):
|
||
"""Executes a stellar-core http-command and returns the parsed JSON."""
|
||
try:
|
||
_, (stdout, _stderr) = container.exec_run(command, demux=True)
|
||
if not stdout:
|
||
return None
|
||
raw = stdout.decode()
|
||
idx = raw.find('{')
|
||
if idx >= 0:
|
||
obj, _ = json.JSONDecoder().raw_decode(raw, idx)
|
||
return obj
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, extended_traffic=False):
|
||
"""Collect data and render one complete status screen.
|
||
|
||
Returns (outbound_numbered, container) where outbound_numbered is a list of
|
||
dicts [{id, ip, label, location}] for the interactive drop workflow.
|
||
"""
|
||
# --- Pre-load data ---
|
||
container, container_name = detect_container(client)
|
||
if container_name is None:
|
||
container_name = CANDIDATE_CONTAINERS[0]
|
||
|
||
core_info = None
|
||
if container:
|
||
raw = exec_json(container, "stellar-core http-command info")
|
||
if raw:
|
||
core_info = raw.get("info")
|
||
|
||
hz_data = None
|
||
try:
|
||
with urlopen(HORIZON_URL + "/", timeout=5) as resp:
|
||
hz_data = json.loads(resp.read())
|
||
except Exception:
|
||
pass
|
||
|
||
# =============================================================
|
||
# 1. CONTAINER STATUS
|
||
# =============================================================
|
||
if not peers_extended:
|
||
hdr("🐳 Container Status")
|
||
|
||
if container is None:
|
||
err(f"Container '{container_name}' not found")
|
||
else:
|
||
if container.status == "running":
|
||
ok(f"Container: {container_name} (Running)")
|
||
else:
|
||
err(f"Container: {container_name} ({container.status})")
|
||
|
||
try:
|
||
s = container.stats(stream=False)
|
||
cpu_delta = (s["cpu_stats"]["cpu_usage"]["total_usage"]
|
||
- s["precpu_stats"]["cpu_usage"]["total_usage"])
|
||
sys_delta = (s["cpu_stats"]["system_cpu_usage"]
|
||
- s["precpu_stats"]["system_cpu_usage"])
|
||
num_cpus = (s["cpu_stats"].get("online_cpus")
|
||
or len(s["cpu_stats"]["cpu_usage"].get("percpu_usage", [1])))
|
||
cpu_pct = (cpu_delta / sys_delta) * num_cpus * 100.0 if sys_delta > 0 else 0.0
|
||
mem_use = s["memory_stats"]["usage"]
|
||
mem_lim = s["memory_stats"]["limit"]
|
||
mem_str = f"{fmt_bytes(mem_use)} / {fmt_bytes(mem_lim)}"
|
||
print(f" CPU: {cpu_pct:.2f}% | RAM: {mem_str}")
|
||
except Exception:
|
||
print(" Stats not available")
|
||
|
||
if hz_data:
|
||
print(f" Horizon: {hz_data.get('horizon_version', '?')}")
|
||
print(f" Core: {hz_data.get('core_version', '?')}")
|
||
|
||
# =============================================================
|
||
# 2. DISK
|
||
# =============================================================
|
||
if not HAS_ZFS:
|
||
hdr("💾 Disk")
|
||
try:
|
||
path = (os.path.splitdrive(os.path.abspath(__file__))[0] + "\\"
|
||
if sys.platform == "win32" else "/")
|
||
usage = shutil.disk_usage(path)
|
||
cap_pct = f"{usage.used / usage.total * 100:.0f}%"
|
||
ok(f"Path: {path}")
|
||
print(f" Size: {fmt_bytes(usage.total)} | Used: {fmt_bytes(usage.used)} ({cap_pct}) | Free: {fmt_bytes(usage.free)}")
|
||
except Exception as e:
|
||
err(f"Disk info not available: {e}")
|
||
else:
|
||
hdr("💾 Disk (ZFS: z01pool)")
|
||
zfs_line = zfs_get(["zpool", "list", "-H", "-o",
|
||
"health,size,alloc,free,cap,frag", "z01pool"])
|
||
if not zfs_line:
|
||
err("Pool z01pool not found (or missing permissions)")
|
||
else:
|
||
health, size, alloc, free, cap, frag = zfs_line.split("\t")
|
||
compress = zfs_get(["zfs", "get", "-H", "-o", "value", "compression", "z01pool"]) or "?"
|
||
ratio = zfs_get(["zfs", "get", "-H", "-o", "value", "compressratio", "z01pool"]) or "?"
|
||
mount = zfs_get(["zfs", "get", "-H", "-o", "value", "mountpoint", "z01pool"]) or "?"
|
||
|
||
if health == "ONLINE":
|
||
ok(f"Health: {health}")
|
||
else:
|
||
err(f"Health: {health}")
|
||
print(f" Size: {size} | Used: {alloc} ({cap}) | Free: {free}")
|
||
print(f" Fragmentation: {frag}")
|
||
print(f" Compression: {compress} (ratio: {ratio})")
|
||
print(f" Mountpoint: {mount}")
|
||
|
||
# =============================================================
|
||
# 3. PROTOCOL STATUS (stellar-core)
|
||
# =============================================================
|
||
hdr("⭐ Protocol Status")
|
||
|
||
if core_info is None:
|
||
err("stellar-core not reachable")
|
||
else:
|
||
q = core_info["quorum"]["qset"]
|
||
state = core_info["state"]
|
||
block = core_info["ledger"]["num"]
|
||
proto = core_info["protocol_version"]
|
||
ledger_proto = core_info["ledger"].get("version", "—")
|
||
age = core_info["ledger"]["age"]
|
||
started = core_info["startedOn"]
|
||
|
||
if state == "Synced!":
|
||
ok(f"State: {state}")
|
||
else:
|
||
warn(f"State: {state}")
|
||
print(f" Block: {block} (Ledger age: {age}s)")
|
||
print(f" Protocol: {proto} (Ledger: {ledger_proto})")
|
||
phase = q.get('phase', '—')
|
||
agree = q.get('agree', '—')
|
||
missing = q.get('missing', '—')
|
||
lag_ms = q.get('lag_ms', '—')
|
||
print(f" Quorum Phase: {phase} (agree: {agree}, missing: {missing}, lag: {lag_ms}ms)")
|
||
print(f" Node started: {started}")
|
||
|
||
# =============================================================
|
||
# 4. API STATUS (Horizon)
|
||
# =============================================================
|
||
hdr("🌅 API Status (Horizon)")
|
||
|
||
if hz_data is None:
|
||
err(f"Horizon API not reachable ({HORIZON_URL})")
|
||
else:
|
||
core_blk = hz_data.get("core_latest_ledger", 0)
|
||
hist_blk = hz_data.get("history_latest_ledger", 0)
|
||
ings_blk = hz_data.get("ingest_latest_ledger", 0)
|
||
lag = core_blk - ings_blk
|
||
closed_at = hz_data.get("history_latest_ledger_closed_at", "?")
|
||
|
||
if lag <= 2:
|
||
ok(f"Status: Synced (Ingest lag: {lag} ledger)")
|
||
else:
|
||
warn(f"Status: Lagging (Ingest {lag} ledger behind core)")
|
||
core_diff = (block - core_blk) if (core_info is not None and core_blk) else None
|
||
core_diff_str = f" (Δ {core_diff:+d} vs Core)" if core_diff is not None else ""
|
||
print(f" Core Latest Ledger: {core_blk}{core_diff_str}")
|
||
print(f" Ingest Latest Ledger: {ings_blk} (last closed: {closed_at})")
|
||
print(f" History Oldest: {hist_blk}")
|
||
|
||
# =============================================================
|
||
# 5. NODE METRICS (stellar-core)
|
||
# =============================================================
|
||
hdr("📊 Node Metrics")
|
||
|
||
metrics_raw = None
|
||
if container:
|
||
metrics_raw = exec_json(container, "stellar-core http-command metrics")
|
||
|
||
if metrics_raw is None:
|
||
err("Metrics not available")
|
||
else:
|
||
m = metrics_raw.get("metrics", {})
|
||
lag = m.get("scp.timing.first-to-self-externalize-lag", {})
|
||
def _ms(val): return f"{val:.1f}ms" if val is not None else "—"
|
||
print(f" SCP externalize lag: min {_ms(lag.get('min'))} mean {_ms(lag.get('mean'))} max {_ms(lag.get('max'))}")
|
||
print(f" Percentiles (ms): p50 {_ms(lag.get('median'))} p75 {_ms(lag.get('75%'))} p95 {_ms(lag.get('95%'))} p98 {_ms(lag.get('98%'))} p99 {_ms(lag.get('99%'))} p99.9 {_ms(lag.get('99.9%'))}")
|
||
|
||
# =============================================================
|
||
# 6. PEER CONNECTIONS (stellar-core)
|
||
# =============================================================
|
||
hdr("🌐 Peer Connections")
|
||
|
||
peers_raw = None
|
||
if container:
|
||
peers_raw = exec_json(container, "stellar-core http-command peers?compact=false")
|
||
|
||
outbound_numbered = [] # [{id, ip, label, location}] in display order, for interactive drop
|
||
|
||
if peers_raw is None:
|
||
err("Peer data not available (stellar-core not reachable)")
|
||
else:
|
||
ap = peers_raw.get("authenticated_peers", {})
|
||
inbound = ap.get("inbound", []) or []
|
||
outbound = ap.get("outbound", []) or []
|
||
pp = peers_raw.get("pending_peers", {})
|
||
p_in = pp.get("inbound", []) or []
|
||
p_out = pp.get("outbound", []) or []
|
||
|
||
in_str = f"Inbound: {len(inbound)}" + (f" (+{len(p_in)} pending)" if p_in else "")
|
||
out_str = f"Outbound: {len(outbound)}" + (f" (+{len(p_out)} pending)" if p_out else "")
|
||
(warn if p_in else ok)(in_str)
|
||
(warn if p_out else ok)(out_str)
|
||
|
||
if peers_extended:
|
||
# --- Fetch full node IDs for interactive drop ---
|
||
fullkey_map = {} # address -> full node ID
|
||
if interactive and container:
|
||
fk_raw = exec_json(
|
||
container,
|
||
["stellar-core", "http-command", "peers?fullkeys=true"]
|
||
)
|
||
if fk_raw:
|
||
fk_ap = fk_raw.get("authenticated_peers", {})
|
||
for peers_list in (fk_ap.get("outbound") or [], fk_ap.get("inbound") or []):
|
||
for p in peers_list:
|
||
addr = p.get("address", "").rsplit(":", 1)[0]
|
||
nid = p.get("id", "")
|
||
if addr and nid:
|
||
fullkey_map[addr] = nid
|
||
|
||
# --- Peers DB ---
|
||
db_counts = None
|
||
try:
|
||
exit_code, db_out = container.exec_run([
|
||
"psql", "-U", "stellar", "-d", "core",
|
||
"-t", "-A", "-F", "|",
|
||
"-c", "SELECT type, COUNT(*) FROM peers GROUP BY type;"
|
||
])
|
||
if exit_code == 0 and db_out:
|
||
db_counts = {0: 0, 1: 0, 2: 0}
|
||
for line in db_out.decode().strip().splitlines():
|
||
parts = line.split("|")
|
||
if len(parts) == 2:
|
||
try:
|
||
t = int(parts[0])
|
||
if t in db_counts:
|
||
db_counts[t] = int(parts[1])
|
||
except ValueError:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
print()
|
||
if db_counts is not None:
|
||
print(f" Peers DB: Inbound: {db_counts[0]} | Outbound: {db_counts[1]} | Preferred: {db_counts[2]}")
|
||
else:
|
||
print(f" Peers DB: not queryable (sqlite3 or path missing?)")
|
||
|
||
validators = load_validators(CFG_PATH)
|
||
|
||
def scp_direction(rx, tx):
|
||
if rx == 0 and tx == 0:
|
||
return "◆"
|
||
if tx == 0 or rx / max(tx, 1) > 1.5:
|
||
return "▼" # Receiving more from peer
|
||
if rx == 0 or tx / max(rx, 1) > 1.5:
|
||
return "▲" # Sending more to peer
|
||
return "◆"
|
||
|
||
# Node uptime
|
||
node_up_str = "—"
|
||
if core_info:
|
||
try:
|
||
dt = datetime.fromisoformat(core_info["startedOn"].replace("Z", "+00:00"))
|
||
node_elapsed = int((datetime.now(timezone.utc) - dt).total_seconds())
|
||
node_up_str = fmt_up(node_elapsed)
|
||
except Exception:
|
||
pass
|
||
|
||
# Node ID
|
||
node_id_str = core_info.get("node_id", "—") if core_info else "—"
|
||
if node_id_str == "—" and container:
|
||
try:
|
||
_, out = container.exec_run(
|
||
["grep", "-oP", r"(?<=nodeID=)G[A-Z0-9]{55}", "/tmp/stellar-core.log"]
|
||
)
|
||
if out:
|
||
node_id_str = out.decode().splitlines()[0].strip() or "—"
|
||
except Exception:
|
||
pass
|
||
|
||
# Node IP
|
||
node_ip_str = "—"
|
||
try:
|
||
node_ip_str = socket.gethostbyname(socket.gethostname())
|
||
except Exception:
|
||
pass
|
||
|
||
# Quorum phase for Flow column
|
||
node_flow_str = "—"
|
||
if core_info:
|
||
try:
|
||
phase = core_info["quorum"]["qset"]["phase"]
|
||
node_flow_str = "◆" if phase == "EXTERNALIZE" else phase[:4]
|
||
except Exception:
|
||
pass
|
||
|
||
# --- Geolocation batch lookup ---
|
||
if geo_enabled:
|
||
_all_ips = [p.get("address", "").rsplit(":", 1)[0] for p in inbound + outbound]
|
||
if node_ip_str != "—":
|
||
_all_ips.append(node_ip_str)
|
||
geo_batch([ip for ip in _all_ips if ip])
|
||
|
||
def _is_private(ip):
|
||
try:
|
||
return ipaddress.ip_address(ip).is_private
|
||
except Exception:
|
||
return False
|
||
|
||
# pid_w: use actual displayed length (validators are truncated to 5)
|
||
def _disp_len(p):
|
||
ra = p.get("address", "?").rsplit(":", 1)[0]
|
||
pid = p.get("id", "")
|
||
return 5 if (validators.get(ra) or validators.get(pid)) else len(pid)
|
||
pid_w = max((_disp_len(p) for p in inbound + outbound), default=5)
|
||
col_w = pid_w + 5 # pid + " (Vn)" or pid + spaces
|
||
|
||
# Node version: prefer core_info["build"], fall back to Horizon
|
||
node_ver_str = "—"
|
||
if core_info:
|
||
node_ver_str = fmt_version(core_info.get("build", ""))
|
||
if node_ver_str == "—" and hz_data:
|
||
node_ver_str = fmt_version(hz_data.get("core_version", ""))
|
||
|
||
# ver_w: dynamic width based on actual version strings
|
||
_all_ver_strs = [fmt_version(p.get("ver", "")) for p in inbound + outbound]
|
||
ver_w = max((len(s) for s in _all_ver_strs + [node_ver_str]), default=7)
|
||
ver_w = max(ver_w, len("Version"))
|
||
|
||
# addr_w: dynamic width based on actual IP strings (only when show_ip)
|
||
if show_ip:
|
||
_all_addr_strs = [
|
||
p.get("address", "").rsplit(":", 1)[0]
|
||
for p in inbound + outbound
|
||
]
|
||
if node_ip_str != "—":
|
||
_all_addr_strs.append(node_ip_str)
|
||
ip_col_hdr = "IP"
|
||
addr_w = max((len(s) for s in _all_addr_strs), default=15)
|
||
addr_w = max(addr_w, len(ip_col_hdr))
|
||
|
||
if geo_enabled:
|
||
_all_geo_strs = [
|
||
geo_str(p.get("address", "").rsplit(":", 1)[0])
|
||
for p in inbound + outbound
|
||
]
|
||
if node_ip_str != "—":
|
||
_all_geo_strs.append(geo_str(node_ip_str))
|
||
geo_col_hdr = "Location"
|
||
geo_w = max((len(s) for s in _all_geo_strs + [""]), default=8)
|
||
geo_w = max(geo_w, len(geo_col_hdr))
|
||
|
||
if show_ip:
|
||
node_addr_disp = node_ip_str if node_ip_str != "—" else "—"
|
||
if geo_enabled:
|
||
node_geo_disp = geo_str(node_ip_str) if node_ip_str != "—" else "—"
|
||
|
||
# In interactive mode outbound rows get a number prefix
|
||
num_col = interactive
|
||
if num_col:
|
||
hdr_num_prefix = " # "
|
||
node_row_prefix = " "
|
||
else:
|
||
hdr_num_prefix = ""
|
||
node_row_prefix = ""
|
||
|
||
_ip_hdr_part = f" {ip_col_hdr:<{addr_w}}" if show_ip else ""
|
||
_ip_node_part = f" {node_addr_disp:<{addr_w}}" if show_ip else ""
|
||
_geo_hdr_part = f" {geo_col_hdr:<{geo_w}}" if geo_enabled else ""
|
||
_geo_node_part = f" {node_geo_disp:<{geo_w}}" if geo_enabled else ""
|
||
# Summed rates across all peers for the node summary row
|
||
total_rx_rate = sum(
|
||
(p.get("byte_read", 0) or 0) * 8 / (p.get("elapsed", 0) or 1)
|
||
for p in inbound + outbound if (p.get("elapsed", 0) or 0) > 0
|
||
)
|
||
total_tx_rate = sum(
|
||
(p.get("byte_write", 0) or 0) * 8 / (p.get("elapsed", 0) or 1)
|
||
for p in inbound + outbound if (p.get("elapsed", 0) or 0) > 0
|
||
)
|
||
total_rx_str = fmt_rate(total_rx_rate) if total_rx_rate > 0 else "—"
|
||
total_tx_str = fmt_rate(total_tx_rate) if total_tx_rate > 0 else "—"
|
||
|
||
if extended_traffic:
|
||
_traffic_hdr = f" {'RX':>9} {'TX':>9} {'RX/s':>8} {'TX/s':>8}"
|
||
_traffic_node = f" {'—':>9} {'—':>9} {total_rx_str:>8} {total_tx_str:>8}"
|
||
else:
|
||
_all_ratio_parts = [
|
||
fmt_ratio_parts(p.get("byte_read", 0) or 0, p.get("byte_write", 0) or 0)
|
||
for p in inbound + outbound
|
||
]
|
||
_valid_parts = [p for p in _all_ratio_parts if p is not None]
|
||
ratio_l_w = max((len(p[0]) for p in _valid_parts), default=1)
|
||
ratio_r_w = max((len(p[1]) for p in _valid_parts), default=1)
|
||
ratio_total = ratio_l_w + 1 + ratio_r_w
|
||
ratio_total = max(ratio_total, len("RX/TX"))
|
||
_traffic_hdr = f" {'RX/TX':>{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"
|
||
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}"
|
||
SEP = " " + "─" * (len(HDR) - 2)
|
||
|
||
def show(peers, is_out):
|
||
for p in sorted(peers, key=lambda p: p.get("elapsed", 0), reverse=True):
|
||
raw_addr = p.get("address", "?").rsplit(":", 1)[0]
|
||
pid = p.get("id", "?")
|
||
display_addr = raw_addr
|
||
addr_disp = display_addr if show_ip else ""
|
||
geo_disp = geo_str(display_addr) if geo_enabled else ""
|
||
v_match = validators.get(display_addr) or validators.get(raw_addr) or validators.get(pid)
|
||
if v_match:
|
||
node_col = f"{v_match[1][:5]} (V{v_match[0]})"
|
||
else:
|
||
node_col = pid
|
||
ver = fmt_version(p.get("ver", ""))
|
||
lat = p.get("latency", "?")
|
||
lat_s = "n/a" if lat == 86400000 else f"{lat}ms"
|
||
up = fmt_up(p.get("elapsed", 0))
|
||
rx = p.get("byte_read", 0) or 0
|
||
tx = p.get("byte_write", 0) or 0
|
||
elapsed = p.get("elapsed", 0) or 0
|
||
flow = scp_direction(rx, tx)
|
||
rx_rate_s = fmt_rate(rx * 8 / elapsed) if elapsed > 0 else "—"
|
||
tx_rate_s = fmt_rate(tx * 8 / elapsed) if elapsed > 0 else "—"
|
||
|
||
if num_col and is_out:
|
||
n = len(outbound_numbered) + 1
|
||
full_id = (fullkey_map.get(display_addr)
|
||
or fullkey_map.get(raw_addr)
|
||
or pid)
|
||
location = geo_label(display_addr) if geo_enabled else display_addr
|
||
outbound_numbered.append({"id": full_id, "ip": display_addr, "label": node_col, "location": location})
|
||
row_prefix = f"{n:>2} "
|
||
elif num_col:
|
||
row_prefix = " "
|
||
else:
|
||
row_prefix = ""
|
||
|
||
_ip_row_part = f" {addr_disp:<{addr_w}}" if show_ip else ""
|
||
_geo_row_part = f" {geo_disp:<{geo_w}}" if geo_enabled else ""
|
||
if extended_traffic:
|
||
_traffic_row = f" {fmt_bytes(rx):>9} {fmt_bytes(tx):>9} {rx_rate_s:>8} {tx_rate_s:>8}"
|
||
else:
|
||
_parts = fmt_ratio_parts(rx, tx)
|
||
_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()
|
||
print(HDR)
|
||
print(NODE_ROW)
|
||
print(SEP)
|
||
if inbound:
|
||
print(" Inbound")
|
||
show(inbound, False)
|
||
if outbound:
|
||
if inbound:
|
||
print()
|
||
print(" Outbound")
|
||
show(outbound, True)
|
||
|
||
print()
|
||
return outbound_numbered, container
|
||
|
||
|
||
def do_drop_workflow(outbound_peers, container):
|
||
"""Interactive drop peer workflow."""
|
||
if not outbound_peers:
|
||
print(" Keine outbound peers verfügbar.")
|
||
time.sleep(2)
|
||
return
|
||
|
||
print(f"\n Peer-Nummer (1–{len(outbound_peers)}), ESC=Abbrechen: ", end="", flush=True)
|
||
old_settings = termios.tcgetattr(sys.stdin)
|
||
tty.setcbreak(sys.stdin.fileno())
|
||
try:
|
||
ch = os.read(sys.stdin.fileno(), 1).decode(errors="replace")
|
||
finally:
|
||
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
|
||
print(ch)
|
||
if ch == '\x1b' or not ch.isdigit():
|
||
return
|
||
try:
|
||
n = int(ch)
|
||
except ValueError:
|
||
return
|
||
if n < 1 or n > len(outbound_peers):
|
||
print(f" Ungültige Nummer (1-{len(outbound_peers)})")
|
||
time.sleep(2)
|
||
return
|
||
|
||
peer = outbound_peers[n - 1]
|
||
full_id = peer["id"]
|
||
ip = peer["ip"]
|
||
label = peer["label"]
|
||
location = peer.get("location", ip)
|
||
print(f" Drop peer {n} ({label[:8]}... / {location})? [ENTER] ja [ESC] abbrechen: ", end="", flush=True)
|
||
old_settings = termios.tcgetattr(sys.stdin)
|
||
tty.setcbreak(sys.stdin.fileno())
|
||
try:
|
||
ch = os.read(sys.stdin.fileno(), 1).decode(errors="replace")
|
||
finally:
|
||
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
|
||
print()
|
||
if ch in ('\r', '\n'):
|
||
result = exec_json(
|
||
container,
|
||
["stellar-core", "http-command", f"droppeer?node={full_id}"]
|
||
)
|
||
if result is None or result.get("status", "").upper() == "OK":
|
||
print(f" Peer {full_id[:8]}... ({ip}) getrennt.")
|
||
else:
|
||
print(f" Unerwartete Antwort: {result.get('status', result)}")
|
||
time.sleep(2)
|
||
|
||
|
||
def drop_all_workflow(outbound_peers, container):
|
||
"""Drop all outbound peers with a single confirmation."""
|
||
if not outbound_peers:
|
||
print(" Keine outbound peers verfügbar.")
|
||
time.sleep(2)
|
||
return
|
||
|
||
peer_list = " ".join(f"{p['label'][:8]}…/{p.get('location', p['ip'])}" for p in outbound_peers)
|
||
print(f"\n Drop all ({len(outbound_peers)}): {peer_list}")
|
||
print(f" [ENTER] ja [ESC] abbrechen: ", end="", flush=True)
|
||
old_settings = termios.tcgetattr(sys.stdin)
|
||
tty.setcbreak(sys.stdin.fileno())
|
||
try:
|
||
ch = os.read(sys.stdin.fileno(), 1).decode(errors="replace")
|
||
finally:
|
||
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
|
||
print()
|
||
if ch not in ('\r', '\n'):
|
||
return
|
||
|
||
ok_count = 0
|
||
for peer in outbound_peers:
|
||
result = exec_json(
|
||
container,
|
||
["stellar-core", "http-command", f"droppeer?node={peer['id']}"]
|
||
)
|
||
if result is None or result.get("status", "").upper() == "OK":
|
||
print(f" Getrennt: {peer['label'][:8]}... / {peer.get('location', peer['ip'])}")
|
||
ok_count += 1
|
||
else:
|
||
print(f" Fehlgeschlagen ({peer.get('location', peer['ip'])}): {result.get('status', result)}")
|
||
print(f"\n {ok_count}/{len(outbound_peers)} Verbindungen getrennt.")
|
||
time.sleep(3)
|
||
|
||
|
||
def interactive_loop(client, refresh_interval, peers_extended, geo_enabled, show_ip, extended_traffic=False):
|
||
if not HAS_TTY:
|
||
print("ERROR: Interaktiver Modus benötigt 'tty'/'termios'")
|
||
sys.exit(1)
|
||
|
||
outbound_peers = []
|
||
container = None
|
||
traffic_extended = extended_traffic # local, toggled with 'x'
|
||
|
||
old_settings = termios.tcgetattr(sys.stdin)
|
||
# cbreak wird EINMAL für die gesamte Session gesetzt (nicht erst nach jedem
|
||
# run_cycle()). Andernfalls bleibt das Terminal während der Ausführung im
|
||
# zeilengepufferten Modus des alten old_settings, und das anschließende
|
||
# tty.setcbreak(..., TCSAFLUSH) verwirft dabei sogar noch ungelesene
|
||
# Eingaben – eine während einer langsamen Ausführung gedrückte Taste wie
|
||
# 'q' würde so nie ankommen.
|
||
tty.setcbreak(sys.stdin.fileno())
|
||
try:
|
||
while True:
|
||
cycle_start = time.monotonic()
|
||
buf = io.StringIO()
|
||
with contextlib.redirect_stdout(buf):
|
||
outbound_peers, container = run_cycle(
|
||
client, peers_extended, geo_enabled, show_ip, interactive=True,
|
||
extended_traffic=traffic_extended
|
||
)
|
||
cycle_duration = time.monotonic() - cycle_start
|
||
if cycle_duration >= refresh_interval:
|
||
warn(f"Update dauerte {cycle_duration:.1f}s – "
|
||
f"länger als das Intervall ({refresh_interval}s). "
|
||
f"Intervall (refresh_interval) erhöhen.")
|
||
elif cycle_duration >= refresh_interval * REFRESH_WARN_RATIO:
|
||
pct = cycle_duration / refresh_interval * 100
|
||
warn(f"Update dauerte {cycle_duration:.1f}s "
|
||
f"({pct:.0f}% des {refresh_interval}s-Intervalls) – knapp.")
|
||
sys.stdout.write("\033[2J\033[H" + buf.getvalue())
|
||
sys.stdout.flush()
|
||
|
||
# Timer läuft ab Beginn der Ausführung, nicht erst ab deren Ende –
|
||
# so verlängert die Update-Laufzeit nicht zusätzlich das Intervall.
|
||
deadline = cycle_start + refresh_interval
|
||
last_secs = -1
|
||
while True:
|
||
remaining = max(0.0, deadline - time.monotonic())
|
||
secs = int(remaining) + (1 if remaining % 1 > 0 else 0)
|
||
if secs != last_secs:
|
||
x_label = "RX/TX" if traffic_extended else "detail"
|
||
text = (f" [d] Drop peer [a] Drop all [x] {x_label} [q] Beenden "
|
||
f"Refresh in {secs}s (Update: {cycle_duration:.1f}s) ")
|
||
sys.stdout.write("\r" + text)
|
||
sys.stdout.flush()
|
||
last_secs = secs
|
||
|
||
# Immer mindestens einmal (nicht-blockierend, wenn remaining
|
||
# bereits 0 ist) auf Eingabe prüfen – auch wenn die Deadline
|
||
# schon verstrichen ist, weil die Ausführung selbst länger als
|
||
# das Intervall dauerte. Sonst wird 'q' & Co. in diesem Fall
|
||
# nie abgefragt, da sofort in den nächsten Zyklus gesprungen würde.
|
||
timeout = min(remaining, 0.25) if remaining > 0 else 0
|
||
rlist, _, _ = select.select([sys.stdin], [], [], timeout)
|
||
if rlist:
|
||
ch = sys.stdin.read(1)
|
||
if ch in ('q', 'Q', '\x03', '\x04'):
|
||
print()
|
||
return
|
||
elif ch in ('d', 'D') and outbound_peers and container:
|
||
print()
|
||
do_drop_workflow(outbound_peers, container)
|
||
break # trigger immediate refresh
|
||
elif ch in ('a', 'A') and outbound_peers and container:
|
||
print()
|
||
drop_all_workflow(outbound_peers, container)
|
||
break # trigger immediate refresh
|
||
elif ch in ('x', 'X'):
|
||
traffic_extended = not traffic_extended
|
||
break # trigger immediate refresh
|
||
|
||
if remaining <= 0:
|
||
break
|
||
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
|
||
sys.stdout.write("\r\n")
|
||
sys.stdout.flush()
|
||
|
||
|
||
USAGE = """\
|
||
pi-stat — Stellar Core node status monitor
|
||
|
||
Usage:
|
||
./pi-stat.py [OPTIONS]
|
||
|
||
Options:
|
||
-p, --peers Extended peer table
|
||
-g, --geo Geolocation per peer (implies -p)
|
||
-s, --show-ip Show IP column (hidden by default)
|
||
-x, --extended Full RX/TX/RX-rate/TX-rate columns (default: ratio only)
|
||
-i [SEC], --interactive [SEC]
|
||
Auto-refresh mode (overrides the configured interval)
|
||
-?, --help Show this help and exit
|
||
|
||
Config file (first match wins, JSON):
|
||
$PI_STAT_CONFIG
|
||
~/.config/pi-stat/config.json
|
||
<script dir>/pi-stat.json
|
||
|
||
Keys:
|
||
refresh_interval Auto-refresh interval in seconds (default: 5)
|
||
|
||
Interactive keys:
|
||
d Drop one outbound peer
|
||
a Drop all outbound peers
|
||
q Quit
|
||
"""
|
||
|
||
|
||
def main():
|
||
# --- CLI options ---
|
||
args = sys.argv[1:]
|
||
|
||
if "-?" in args or "--help" in args:
|
||
print(USAGE, end="")
|
||
sys.exit(0)
|
||
|
||
peers_extended = "-p" in args or "--peers" in args
|
||
geo_enabled = "-g" in args or "--geo" in args
|
||
show_ip = "-s" in args or "--show-ip" in args
|
||
interactive = "-i" in args or "--interactive" in args
|
||
extended_traffic = "-x" in args or "--extended" in args
|
||
|
||
refresh_interval = config_refresh(load_config())
|
||
if interactive:
|
||
for flag in ("-i", "--interactive"):
|
||
if flag in args:
|
||
idx = args.index(flag)
|
||
if idx + 1 < len(args):
|
||
try:
|
||
refresh_interval = max(1, int(args[idx + 1]))
|
||
except ValueError:
|
||
pass
|
||
|
||
if geo_enabled:
|
||
peers_extended = True
|
||
|
||
# --- Docker client ---
|
||
try:
|
||
client = docker.from_env()
|
||
except Exception as e:
|
||
print(f"ERROR: Docker not reachable: {e}")
|
||
sys.exit(1)
|
||
|
||
if interactive:
|
||
interactive_loop(client, refresh_interval, peers_extended, geo_enabled, show_ip, extended_traffic)
|
||
else:
|
||
run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, extended_traffic=extended_traffic)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|