From db61213f396ccf3e617a92da2298496fb06b2997 Mon Sep 17 00:00:00 2001 From: olieske Date: Wed, 24 Jun 2026 12:00:01 +0000 Subject: [PATCH] initial commit --- .gitignore | 34 ++ CLAUDE.md | 59 ++++ pi-stat | 884 +++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + stellar-core.cfg | 63 ++++ 5 files changed, 1041 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100755 pi-stat create mode 100644 requirements.txt create mode 100644 stellar-core.cfg diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bdd26c8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,34 @@ +# Virtual Environment +venv/ +.venv/ +env/ + +# Python Cache +__pycache__/ +*.pyc +*.pyo +*.pyd + +# Distribution / Build +dist/ +build/ +*.egg-info/ + +# Lokale Konfiguration & Secrets +*.env +.env +config.local.* +secrets.* + +# IDE +.vscode/ +.idea/ +*.swp + +# Claude Code +.claude + +# Logs & temporäre Dateien +*.log +*.tmp +*~ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..0d835ad --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,59 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## What This Is + +**pi-stat** is a fast status monitoring tool for Stellar Core nodes on the Pi Network. It provides quick system diagnostics as an alternative to the slow `pi-node status` command. + +## Running + +```bash +./pi-stat.py # Standard report +./pi-stat.py -p / --peers # Extended peer table +./pi-stat.py -g / --geo # With geolocation (implies -p) +./pi-stat.py -s / --show-ip # Show IP column (hidden by default) +./pi-stat.py -x / --extended # Show full RX/TX/RX-rate/TX-rate columns (default: ratio only) +./pi-stat.py -i [SEC] / --interactive [SEC] # Interactive auto-refresh (default: 10s) +``` + +No build step. Requires `pip install docker` and a running Docker daemon. Runs on Linux, WSL2, and Windows. + +## Architecture + +Single self-contained script (`pi-stat.py`). Logic is split into helper functions and a `run_cycle()` function that collects data and renders one full screen, called either once (normal mode) or in a loop (interactive mode). + +### Data sources + +- **Docker SDK** — container CPU/RAM stats; all commands run inside the auto-detected container (`mainnet` → `testnet2` → `testnet`, first running one) via `exec_run` +- **Stellar Core HTTP API** — `exec_json()` runs `stellar-core http-command ` inside the container +- **Horizon API** — `http://localhost:31401` +- **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 +- **ip-api.com** — batch geolocation (Country, City), cached in `_geo_cache`; only fetched with `-g` + +### Output sections (in order) + +1. Container stats — Docker CPU/RAM, Horizon + Core versions (skipped in `-p`/`-g` mode) +2. Disk — ZFS (`z01pool`) if `zpool` is available, otherwise `shutil.disk_usage()` (skipped in `-p`/`-g` mode) +3. Protocol status — Stellar Core state, ledger, quorum phase +4. API status — Horizon ingest lag +5. Peer connections — compact counts, or full table with `-p`/`-g` + +### Interactive mode (`-i`) + +`interactive_loop()` renders via `run_cycle()` into a `StringIO` buffer, writes it atomically to avoid flicker, then waits for keypresses using `select` in `setcbreak` mode: +- `d` — drop a single outbound peer (`do_drop_workflow`) +- `a` — drop all outbound peers (`drop_all_workflow`) +- `q` / Ctrl+C — clean exit with terminal state restore + +Outbound peers are numbered in the table. Full node IDs are fetched via `peers?fullkeys=true` for the `droppeer` API call. + +### Key patterns + +- **Graceful degradation** — every external call is wrapped in try-except; missing data renders as `—` +- **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) +- **Validator matching** — `load_validators()` parses `stellar-core.cfg` and indexes by IP, name, and public key; matched peers show a truncated key + `(Vn)` label +- **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 diff --git a/pi-stat b/pi-stat new file mode 100755 index 0000000..05c4224 --- /dev/null +++ b/pi-stat @@ -0,0 +1,884 @@ +#!/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") + +# --- 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_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) + try: + while True: + 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 + ) + sys.stdout.write("\033[2J\033[H" + buf.getvalue()) + sys.stdout.flush() + + deadline = time.monotonic() + refresh_interval + tty.setcbreak(sys.stdin.fileno()) + try: + 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 Refresh in {secs}s " + sys.stdout.write("\r" + text) + sys.stdout.flush() + last_secs = secs + + if remaining <= 0: + break + + rlist, _, _ = select.select([sys.stdin], [], [], min(remaining, 0.25)) + if rlist: + ch = sys.stdin.read(1) + if ch in ('q', 'Q', '\x03', '\x04'): + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) + print() + return + elif ch in ('d', 'D') and outbound_peers and container: + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) + print() + do_drop_workflow(outbound_peers, container) + break # trigger immediate refresh + elif ch in ('a', 'A') and outbound_peers and container: + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) + 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 + finally: + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) + + 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 (default interval: 10 s) + -?, --help Show this help and exit + +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 = 10 + 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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c7ccf34 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +docker>=7.1.0 diff --git a/stellar-core.cfg b/stellar-core.cfg new file mode 100644 index 0000000..38ea64c --- /dev/null +++ b/stellar-core.cfg @@ -0,0 +1,63 @@ +# Mainnet - Watcher node (default configuration) +# 0.0.2 + +PUBLIC_HTTP_PORT = false +PEER_PORT = 31402 + +COMMANDS = ["ll?level=info"] +LOG_FILE_PATH = "/tmp/stellar-core.log" +BUCKET_DIR_PATH = "/opt/stellar/core/buckets" + +NETWORK_PASSPHRASE = "Pi Network" +KNOWN_CURSORS = ["HORIZON"] + +DATABASE = "postgresql://dbname=core host=localhost user=stellar password=7fc0dea171825fd05f0b64a5af63b24f" +CATCHUP_COMPLETE=false +CATCHUP_RECENT=256 + +PREFERRED_PEER_KEYS = [ + "$validator1", + "$validator2", + "$validator3" +] + +PREFERRED_PEERS_ONLY = false + +NODE_SEED = "SCN4DRF2NST53Z74PI56AFFXUQ2ORZ6WSQUDDESKFIVBIADRYKDCASZL" +NODE_IS_VALIDATOR = false + +UNSAFE_QUORUM = false +FAILURE_SAFETY = -1 +QUORUM_INTERSECTION_CHECKER = true + +NODE_HOME_DOMAIN = "pi-core-team" + +[[HOME_DOMAINS]] +HOME_DOMAIN = "pi-core-team" +QUALITY="MEDIUM" + +[[VALIDATORS]] +NAME="validator1" +HOME_DOMAIN="pi-core-team" +PUBLIC_KEY="GDLA3KOMXKNHU5AGQB2R5VRBYCKM4X37NQVHTWRIQXNCOFOW4MZZSDTL" +ADDRESS="34.95.11.164:31402" +HISTORY="curl -sf https://history.mainnet.minepi.com/{0} -o {1}" + +[[VALIDATORS]] +NAME="validator2" +HOME_DOMAIN="pi-core-team" +PUBLIC_KEY="GDGGD5XGBWLYS4XYJSUCX2CQ7ZS2JKIQTDOMPMPUU6LX74AINMRH2LMI" +ADDRESS="34.88.93.19:31402" +HISTORY="curl -sf https://history.mainnet.minepi.com/{0} -o {1}" + +[[VALIDATORS]] +NAME="validator3" +HOME_DOMAIN="pi-core-team" +PUBLIC_KEY="GAXTE5AV5OCOEGJE4OPVMN5UJHAKDQCSVMXI4P7SZIGQXVO7LP4JKX4M" +ADDRESS="34.64.252.77:31402" +HISTORY="curl -sf https://history.mainnet.minepi.com/{0} -o {1}" + +[HISTORY.local] +get="cp /opt/stellar/history/local/{0} {1}" +put="cp {0} /opt/stellar/history/local/{1}" +mkdir="mkdir -p /opt/stellar/history/local/{0}"