From fe87c7b34330950eec69e33176596797c119a41a Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 17 Apr 2026 14:19:26 +0800 Subject: [PATCH] debug --- .../app/services/cluster_runtime_service.py | 44 +++++++++- .../app/services/runtime_status_service.py | 86 ++++++++++++++++++- domain-web/src/views/runtime/RuntimeView.vue | 61 +++++++++++++ 3 files changed, 188 insertions(+), 3 deletions(-) diff --git a/domain-api/app/services/cluster_runtime_service.py b/domain-api/app/services/cluster_runtime_service.py index 0917722..9a5f6ec 100644 --- a/domain-api/app/services/cluster_runtime_service.py +++ b/domain-api/app/services/cluster_runtime_service.py @@ -191,17 +191,44 @@ def cleanup_imported_runtime_nodes(*, region: str, role: str, keep_node_code: st def register_local_control_heartbeat() -> None: + from app.services.detect_job_service import get_active_detect_job_summary + from app.services.worker_control_service import detect_worker_runtime + + worker_runtime = detect_worker_runtime() + worker_online = bool(worker_runtime.get("running", False)) + active_job = get_active_detect_job_summary(event_limit=5) or {} + node_stats = list(active_job.get("node_stats") or []) + local_bucket = next( + (item for item in node_stats if str(item.get("node_code") or "").strip() == settings.node_code), + {}, + ) + items_total = int(local_bucket.get("items_total", 0) or 0) + items_claimed = int(local_bucket.get("items_claimed", 0) or 0) + items_running = int(local_bucket.get("items_running", 0) or 0) + items_completed = int(local_bucket.get("items_completed", 0) or 0) + current_load = max(items_running, items_claimed, 0) + detect_participating = bool(worker_online and (items_total > 0 or current_load > 0)) + node_status = "busy" if current_load > 0 else "online" register_node_heartbeat( node_code=settings.node_code, region=settings.node_region, role=settings.node_role, - status="online", - current_load=0, + status=node_status, + current_load=current_load, metadata={ "service": "domain-api", "api_host": settings.api_host, "api_port": settings.api_port, "worker_mode": settings.worker_mode, + "worker_online": worker_online, + "worker_process_count": int(worker_runtime.get("process_count", 0) or 0), + "detect_participating": detect_participating, + "active_job_code": str(active_job.get("job_code") or ""), + "active_job_status": str(active_job.get("status") or ""), + "job_items_total": items_total, + "job_items_claimed": items_claimed, + "job_items_running": items_running, + "job_items_completed": items_completed, "updated_at": datetime.now().isoformat(timespec="seconds"), }, ) @@ -264,6 +291,7 @@ def get_cluster_snapshot() -> dict: stale_nodes: list[str] = [] offline_nodes: list[str] = [] online_worker_nodes = 0 + dedicated_online_worker_nodes = 0 online_control_nodes = 0 for node in nodes: @@ -271,6 +299,9 @@ def get_cluster_snapshot() -> dict: node_role = str(node.get("role") or "unknown") node_region = str(node.get("region") or "unknown") node_code = str(node.get("node_code") or "") + metadata = node.get("metadata") or {} + node_current_load = int(node.get("current_load", 0) or 0) + effective_worker = False status_counts[node_status] = status_counts.get(node_status, 0) + 1 role_counts[node_role] = role_counts.get(node_role, 0) + 1 @@ -283,9 +314,17 @@ def get_cluster_snapshot() -> dict: if node_status == "offline": offline_nodes.append(node_code) if node_role == "worker" and node_status in {"online", "busy"}: + dedicated_online_worker_nodes += 1 + effective_worker = True + elif node_role == "control" and node_status in {"online", "busy"}: + if bool(metadata.get("worker_online", False)) or bool(metadata.get("detect_participating", False)) or node_current_load > 0: + effective_worker = True + if effective_worker: online_worker_nodes += 1 if node_role == "control" and node_status in {"online", "busy"}: online_control_nodes += 1 + node["is_effective_worker"] = effective_worker + node["detect_participating"] = bool(metadata.get("detect_participating", False) or node_current_load > 0) return { "nodes": nodes, @@ -300,6 +339,7 @@ def get_cluster_snapshot() -> dict: "stale_nodes": stale_nodes, "offline_nodes": offline_nodes, "online_worker_nodes": online_worker_nodes, + "dedicated_online_worker_nodes": dedicated_online_worker_nodes, "online_control_nodes": online_control_nodes, }, } diff --git a/domain-api/app/services/runtime_status_service.py b/domain-api/app/services/runtime_status_service.py index 17180a9..3cffebd 100644 --- a/domain-api/app/services/runtime_status_service.py +++ b/domain-api/app/services/runtime_status_service.py @@ -64,7 +64,7 @@ def _build_multi_region_readiness( mainland_worker_nodes = [ node for node in nodes if str(node.get("region") or "") == "mainland" - and str(node.get("role") or "") == "worker" + and bool(node.get("is_effective_worker", False)) and str(node.get("status") or "") in {"online", "busy"} ] @@ -153,6 +153,85 @@ def _build_multi_region_readiness( } +def _build_participating_detect_nodes(*, cluster_snapshot: dict, detect_snapshot: dict, worker_runtime: dict) -> list[dict]: + cluster_nodes = list(cluster_snapshot.get("nodes") or []) + active_job = detect_snapshot.get("active_job") or {} + queue_nodes = list((detect_snapshot.get("queue_health") or {}).get("nodes") or []) + + cluster_map = { + str(item.get("node_code") or "").strip(): item + for item in cluster_nodes + if str(item.get("node_code") or "").strip() + } + queue_map = { + str(item.get("node_code") or "").strip(): item + for item in queue_nodes + if str(item.get("node_code") or "").strip() + } + + merged: list[dict] = [] + seen: set[str] = set() + for item in list(active_job.get("node_stats") or []): + node_code = str(item.get("node_code") or "").strip() + if not node_code or node_code == "unassigned": + continue + seen.add(node_code) + cluster_node = cluster_map.get(node_code, {}) + queue_node = queue_map.get(node_code, {}) + merged.append( + { + "node_code": node_code, + "role": str(cluster_node.get("role") or "worker"), + "region": str(cluster_node.get("region") or settings.node_region), + "status": str(cluster_node.get("status") or "unknown"), + "is_effective_worker": bool(cluster_node.get("is_effective_worker", False) or str(cluster_node.get("role") or "") == "worker"), + "detect_participating": True, + "current_load": int(cluster_node.get("current_load", 0) or 0), + "items_total": int(item.get("items_total", 0) or 0), + "items_pending": int(item.get("items_pending", 0) or 0), + "items_claimed": int(item.get("items_claimed", 0) or 0), + "items_running": int(item.get("items_running", 0) or 0), + "items_completed": int(item.get("items_completed", 0) or 0), + "items_failed": int(item.get("items_failed", 0) or 0), + "processed_recent": int(queue_node.get("processed_recent", 0) or 0), + "processed_per_minute": float(queue_node.get("processed_per_minute", 0) or 0), + } + ) + + if worker_runtime.get("running", False) and settings.node_code not in seen: + cluster_node = cluster_map.get(settings.node_code, {}) + if cluster_node: + merged.append( + { + "node_code": settings.node_code, + "role": str(cluster_node.get("role") or settings.node_role), + "region": str(cluster_node.get("region") or settings.node_region), + "status": str(cluster_node.get("status") or "online"), + "is_effective_worker": True, + "detect_participating": True, + "current_load": int(cluster_node.get("current_load", 0) or 0), + "items_total": 0, + "items_pending": 0, + "items_claimed": 0, + "items_running": 0, + "items_completed": 0, + "items_failed": 0, + "processed_recent": 0, + "processed_per_minute": 0, + } + ) + + return sorted( + merged, + key=lambda item: ( + -int(item.get("items_running", 0) or 0), + -int(item.get("items_claimed", 0) or 0), + -int(item.get("processed_recent", 0) or 0), + str(item.get("node_code") or ""), + ), + ) + + def get_runtime_status() -> dict: runtime_settings = get_runtime_settings() worker_runtime = detect_worker_runtime() @@ -201,6 +280,11 @@ def get_runtime_status() -> dict: "queue_health": queue_health, "capacity_plan": capacity_plan, } + detect_payload["participating_nodes"] = _build_participating_detect_nodes( + cluster_snapshot=cluster_snapshot, + detect_snapshot=detect_payload, + worker_runtime=worker_runtime, + ) append_runtime_projection_if_changed(detect=detect_payload, cluster=cluster_snapshot) sync_summary = get_sync_summary(record_limit=5) readiness = _build_multi_region_readiness( diff --git a/domain-web/src/views/runtime/RuntimeView.vue b/domain-web/src/views/runtime/RuntimeView.vue index e56fc63..82cacab 100644 --- a/domain-web/src/views/runtime/RuntimeView.vue +++ b/domain-web/src/views/runtime/RuntimeView.vue @@ -307,6 +307,50 @@ +
+
+

当前参与检测节点

+ 直接展示哪些节点正在领任务、执行任务或产生近窗吞吐，controller 兼跑检测也会计入。 +
+ + + + + + + + + + + + + + + + + + +
+

动作留痕

@@ -334,6 +378,7 @@
在线控制面 {{ runtime.cluster?.summary?.online_control_nodes || 0 }} 在线 Worker {{ runtime.cluster?.summary?.online_worker_nodes || 0 }} + 独立 Worker {{ runtime.cluster?.summary?.dedicated_online_worker_nodes || 0 }} 忙碌 {{ runtime.cluster?.summary?.status_counts?.busy || 0 }} 失活 {{ runtime.cluster?.summary?.status_counts?.stale || 0 }} 离线 {{ runtime.cluster?.summary?.status_counts?.offline || 0 }} @@ -350,6 +395,13 @@ + + +