#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import shutil import sys import time import urllib.error import urllib.request from datetime import datetime DEFAULT_API_BASE = "http://127.0.0.1:8100/api/v1" def _now_text() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _safe_int(value: object, default: int = 0) -> int: try: return int(value or 0) except Exception: return int(default) def _safe_float(value: object, default: float = 0.0) -> float: try: return float(value or 0.0) except Exception: return float(default) def _safe_text(value: object) -> str: return str(value or "").strip() def _fetch_api(api_base: str, path: str) -> dict: url = f"{api_base.rstrip('/')}/{path.lstrip('/')}" try: request = urllib.request.Request(url, headers={"User-Agent": "domaincheck-runtime-observer/0.1"}) with urllib.request.urlopen(request, timeout=8) as response: payload = json.loads(response.read().decode("utf-8", errors="ignore")) if isinstance(payload, dict): data = payload.get("data") if isinstance(data, dict): return data return payload return {"_error": f"unexpected payload type: {type(payload).__name__}", "_url": url} except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="ignore") return {"_error": f"http {exc.code}", "_body": body[:1000], "_url": url} except Exception as exc: return {"_error": str(exc), "_url": url} def _observe_payload(api_base: str) -> dict: runtime_status = _fetch_api(api_base, "runtime/status") direct_active_job = _fetch_api(api_base, "detect/job/active") runtime_active_job = dict(((runtime_status.get("detect") or {}).get("active_job") or {})) readiness = _fetch_api(api_base, "runtime/readiness") sync_summary = _fetch_api(api_base, "runtime/sync-summary") active_job = _merge_active_job_payloads(direct_active_job, runtime_active_job) return { "runtime_status": runtime_status, "active_job": active_job, "direct_active_job": direct_active_job, "runtime_active_job": runtime_active_job, "readiness": readiness, "sync_summary": sync_summary, } def _real_node_count(active_job: dict) -> int: count = 0 for item in list((active_job or {}).get("node_stats") or []): node_code = _safe_text((item or {}).get("node_code")) if node_code and node_code != "unassigned": count += 1 return count def _merge_active_job_payloads(direct_active_job: dict, runtime_active_job: dict) -> dict: direct_active_job = dict(direct_active_job or {}) runtime_active_job = dict(runtime_active_job or {}) if not runtime_active_job: return direct_active_job if not direct_active_job: return runtime_active_job merged = dict(direct_active_job) same_job = ( _safe_int(direct_active_job.get("job_id")) > 0 and _safe_int(direct_active_job.get("job_id")) == _safe_int(runtime_active_job.get("job_id")) ) or ( _safe_text(direct_active_job.get("job_code")) and _safe_text(direct_active_job.get("job_code")) == _safe_text(runtime_active_job.get("job_code")) ) if not same_job: return direct_active_job preferred_runtime_keys = { "node_stats", "distributed_node_stats", "display_items_claimed", "display_items_running", "display_current_load", "display_active_threads", "display_max_threads", "display_items_completed", "display_items_failed", "display_active_node_codes", "processed_recent", "processed_per_minute", "completed_recent", "failed_recent", "blacklisted_recent", "runtime_snapshot_job_id", "runtime_snapshot_job_code", "runtime_snapshot_queue", } runtime_is_richer = _real_node_count(runtime_active_job) >= _real_node_count(direct_active_job) for key in preferred_runtime_keys: if key in runtime_active_job and runtime_is_richer: merged[key] = runtime_active_job.get(key) return merged def _active_node_rows(active_job: dict) -> list[dict]: rows: list[dict] = [] for raw_item in list(active_job.get("node_stats") or []): item = dict(raw_item or {}) node_code = _safe_text(item.get("node_code")) if not node_code or node_code == "unassigned": continue current_load = max( _safe_int(item.get("current_load")), _safe_int(item.get("active_threads")), _safe_int(item.get("display_running")), _safe_int(item.get("items_running")), ) processed_recent = _safe_int(item.get("processed_recent")) items_claimed = _safe_int(item.get("items_claimed")) status = _safe_text(item.get("status")) if current_load <= 0 and processed_recent <= 0 and items_claimed <= 0 and status not in {"busy", "online"}: continue rows.append( { "node_code": node_code, "status": status or "-", "region": _safe_text(item.get("region")) or "-", "role": _safe_text(item.get("role")) or "-", "current_load": current_load, "display_running": _safe_int(item.get("display_running")), "active_threads": _safe_int(item.get("active_threads")), "max_threads": _safe_int(item.get("max_threads")), "items_claimed": items_claimed, "items_running": _safe_int(item.get("items_running")), "processed_recent": processed_recent, "processed_per_minute": _safe_float(item.get("processed_per_minute")), "completed_recent": _safe_int(item.get("completed_recent")), "failed_recent": _safe_int(item.get("failed_recent")), "blacklisted_recent": _safe_int(item.get("blacklisted_recent")), "last_heartbeat_at": _safe_text(item.get("last_heartbeat_at")), } ) rows.sort( key=lambda item: ( -int(item.get("current_load", 0)), -int(item.get("processed_recent", 0)), -int(item.get("items_claimed", 0)), str(item.get("node_code") or ""), ) ) return rows def _step_rows(active_job: dict) -> list[dict]: rows: list[dict] = [] for raw_item in list(active_job.get("step_stats") or []): item = dict(raw_item or {}) rows.append( { "step_code": _safe_text(item.get("step_code")), "step_name": _safe_text(item.get("step_name")), "pending": _safe_int(item.get("items_pending")), "running": _safe_int(item.get("items_running")), "completed": _safe_int(item.get("items_completed")), "failed": _safe_int(item.get("items_failed")), "blacklisted": _safe_int(item.get("items_blacklisted")), } ) rows.sort(key=lambda item: (-int(item["pending"]), -int(item["running"]), item["step_name"])) return rows def _summary_numbers(payload: dict) -> dict: runtime_status = dict(payload.get("runtime_status") or {}) active_job = dict(payload.get("active_job") or {}) active_nodes = _active_node_rows(active_job) active_processes = len(active_nodes) active_threads = max( _safe_int(active_job.get("display_active_threads")), _safe_int(active_job.get("display_items_running")), sum(max(_safe_int(item.get("current_load")), _safe_int(item.get("display_running"))) for item in active_nodes), ) max_threads = max( _safe_int(active_job.get("display_max_threads")), sum(_safe_int(item.get("max_threads")) for item in active_nodes if _safe_int(item.get("max_threads")) > 0), ) processed_recent = _safe_int(active_job.get("processed_recent")) failed_recent = _safe_int(active_job.get("failed_recent")) blacklisted_recent = _safe_int(active_job.get("blacklisted_recent")) items_pending = _safe_int(active_job.get("items_pending")) items_claimed = _safe_int(active_job.get("items_claimed")) items_running = _safe_int(active_job.get("items_running")) items_completed = _safe_int(active_job.get("items_completed")) items_failed = _safe_int(active_job.get("items_failed")) items_blacklisted = _safe_int(active_job.get("items_blacklisted")) return { "job_id": _safe_int(active_job.get("job_id")), "job_code": _safe_text(active_job.get("job_code")), "job_status": _safe_text(active_job.get("status")) or "-", "progress_percent": _safe_float(active_job.get("progress_percent")), "active_processes": active_processes, "active_threads": active_threads, "max_threads": max_threads, "processed_recent": processed_recent, "processed_per_minute": _safe_float(active_job.get("processed_per_minute")), "completed_recent": _safe_int(active_job.get("completed_recent")), "failed_recent": failed_recent, "blacklisted_recent": blacklisted_recent, "items_pending": items_pending, "items_claimed": items_claimed, "items_running": items_running, "items_completed": items_completed, "items_failed": items_failed, "items_blacklisted": items_blacklisted, "online_worker_nodes": _safe_int(((runtime_status.get("cluster_summary") or {}).get("online_worker_nodes"))), "online_control_nodes": _safe_int(((runtime_status.get("cluster_summary") or {}).get("online_control_nodes"))), } def _diagnose_state(summary: dict) -> tuple[str, str]: items_pending = int(summary["items_pending"]) active_processes = int(summary["active_processes"]) active_threads = int(summary["active_threads"]) processed_recent = int(summary["processed_recent"]) items_running = int(summary["items_running"]) if active_processes > 0 and active_threads > 0 and processed_recent > 0: return "真跑中", "已经有真实执行面和最近吞吐，不是只剩显示残影。" if active_processes > 0 and active_threads > 0 and items_running > 0: return "在跑但偏慢", "有执行面，也有运行中任务，但最近吞吐还没完全拉起来。" if items_pending > 0 and active_processes == 0 and active_threads == 0: return "没跑起来", "队列还有积压，但当前没看到真实执行面在消化。" if items_pending > 0 and processed_recent == 0: return "疑似残影", "队列还有积压，但最近没有看到处理增量，需要继续查执行链。" return "观察中", "当前口径没有明确异常，但还需要继续看最近吞吐是否持续变化。" def _line(char: str = "-") -> str: width = max(60, min(120, shutil.get_terminal_size((100, 20)).columns)) return char * width def _render_summary(payload: dict, *, top_nodes: int) -> str: runtime_status = dict(payload.get("runtime_status") or {}) active_job = dict(payload.get("active_job") or {}) readiness = dict(payload.get("readiness") or {}) sync_summary = dict(payload.get("sync_summary") or {}) summary = _summary_numbers(payload) state_label, state_reason = _diagnose_state(summary) active_nodes = _active_node_rows(active_job) step_rows = _step_rows(active_job) lines: list[str] = [] lines.append(_line("=")) lines.append(f"domainCheck 运行观察面板 {_now_text()}") lines.append(_line("=")) lines.append(f"状态判断: {state_label}") lines.append(f"判断理由: {state_reason}") lines.append( f"当前任务: job_id={summary['job_id']} job_code={summary['job_code'] or '-'} status={summary['job_status']} 进度={summary['progress_percent']:.2f}%" ) lines.append( f"就绪状态: { _safe_text(readiness.get('status')) or '-' } 摘要: { _safe_text(readiness.get('summary')) or '-' }" ) lines.append( f"集群在线: mainland worker={summary['online_worker_nodes']} mainland controller={summary['online_control_nodes']}" ) lines.append(_line()) lines.append("一眼先看这 4 组数:") lines.append( f"任务积压: pending={summary['items_pending']} claimed={summary['items_claimed']} running={summary['items_running']}" ) lines.append( f"结果产出: completed={summary['items_completed']} failed={summary['items_failed']} blacklisted={summary['items_blacklisted']}" ) lines.append( f"执行面: active_processes={summary['active_processes']} active_threads={summary['active_threads']} max_threads={summary['max_threads']}" ) lines.append( f"最近吞吐: processed_recent={summary['processed_recent']} per_minute={summary['processed_per_minute']:.2f} failed_recent={summary['failed_recent']} blacklisted_recent={summary['blacklisted_recent']}" ) detect_runtime = dict(((runtime_status.get("detect") or {}).get("active_job") or {})) if detect_runtime: lines.append(_line()) lines.append( "运行口径提示: " f"display_running={_safe_int(detect_runtime.get('display_items_running'))} " f"display_threads={_safe_int(detect_runtime.get('display_active_threads'))} " f"display_max={_safe_int(detect_runtime.get('display_max_threads'))}" ) lines.append(_line()) lines.append("当前真正有动作的节点:") if not active_nodes: lines.append("- 暂时没看到有真实负载或最近吞吐的节点。") else: for item in active_nodes[: max(1, int(top_nodes or 8))]: lines.append( "- " f"{item['node_code']} " f"load={item['current_load']} " f"running={item['display_running']} " f"max={item['max_threads']} " f"processed_recent={item['processed_recent']} " f"failed_recent={item['failed_recent']} " f"blacklisted_recent={item['blacklisted_recent']} " f"status={item['status']}" ) lines.append(_line()) lines.append("步骤分布:") for item in step_rows: lines.append( "- " f"{item['step_name'] or item['step_code']} " f"pending={item['pending']} running={item['running']} " f"completed={item['completed']} failed={item['failed']} blacklisted={item['blacklisted']}" ) detect_batches = dict((sync_summary.get("detect_result_batches") or {})) if sync_summary: lines.append(_line()) lines.append( "结果回传: " f"enabled={_safe_text(sync_summary.get('enabled')) or '-'} " f"pending_batches={_safe_int(detect_batches.get('pending'))} " f"pushing_batches={_safe_int(detect_batches.get('pushing'))} " f"failed_batches={_safe_int(detect_batches.get('failed'))} " f"synced_batches={_safe_int(detect_batches.get('synced'))}" ) warnings = list(readiness.get("warnings") or []) info = list(readiness.get("info") or []) if warnings or info: lines.append(_line()) lines.append("当前提示:") for text in warnings[:5]: lines.append(f"- warning: {_safe_text(text)}") for text in info[:3]: lines.append(f"- info: {_safe_text(text)}") lines.append(_line("=")) lines.append("建议:") lines.append("- 先盯 `任务积压 / 执行面 / 最近吞吐`，这三组一起动，才算真跑。") lines.append("- `日志` 只拿来辅助定位，不要拿日志多少判断是不是在跑。") lines.append("- 如果 `pending` 很高，但 `active_threads` 和 `processed_recent` 都接近 0，就是没跑起来。") lines.append("- 如果 `failed_recent` 很高而 `completed_recent`、`blacklisted_recent` 很低，说明更像外部步骤超时，不是黑名单在推进。") return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description="Render a human-friendly runtime observation panel.") parser.add_argument("--api-base", default=os.getenv("DOMAINCHECK_API_BASE", DEFAULT_API_BASE)) parser.add_argument("--watch", type=int, default=0, help="refresh interval in seconds; 0 means run once") parser.add_argument("--top-nodes", type=int, default=10) parser.add_argument("--json", action="store_true", help="print raw normalized payload instead of panel") args = parser.parse_args() while True: payload = _observe_payload(args.api_base) if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2, default=str)) else: if int(args.watch or 0) > 0: print("\033[2J\033[H", end="") print(_render_summary(payload, top_nodes=max(1, int(args.top_nodes or 10)))) if int(args.watch or 0) <= 0: return 0 time.sleep(max(2, int(args.watch))) if __name__ == "__main__": raise SystemExit(main())