from __future__ import annotations import json import socket import threading import urllib.error import urllib.request from datetime import datetime, timedelta from app.core.config import settings from app.core.db import db_read_retry, get_db _DEBUG_SCHEMA_SQL = """ CREATE TABLE IF NOT EXISTS detect_debug_events ( id BIGSERIAL PRIMARY KEY, source_region VARCHAR(32) NOT NULL DEFAULT '', node_code VARCHAR(64) NOT NULL DEFAULT '', service VARCHAR(64) NOT NULL DEFAULT '', event_type VARCHAR(64) NOT NULL DEFAULT '', level VARCHAR(16) NOT NULL DEFAULT 'info', message TEXT NOT NULL DEFAULT '', payload_json JSONB, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_detect_debug_events_created ON detect_debug_events(created_at DESC); """ _DEBUG_SCHEMA_READY = False _DEBUG_SCHEMA_LOCK = threading.Lock() def _format_time(value: datetime | None) -> str: return value.isoformat(sep=" ", timespec="seconds") if value else "" def _parse_time(value: str | None) -> datetime | None: text = str(value or "").strip() if not text: return None normalized = text.replace("T", " ") for pattern in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"): try: return datetime.strptime(normalized, pattern) except ValueError: continue return None def _normalize_region(value: str | None, fallback: str) -> str: text = str(value or "").strip() if not text or text == "unknown": return str(fallback or "unknown").strip() or "unknown" return text def _safe_json_dumps(payload: object) -> str: try: return json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str) except Exception: return json.dumps({"_serialization_error": "payload_not_json_serializable"}, ensure_ascii=False, sort_keys=True) def _normalize_message(value: str | None, fallback: str = "") -> str: text = str(value or "").strip() if not text: text = fallback return text[:2000] def _response_code_ok(value: object) -> bool: return value in (0, "0", None, "") def _debug_ingest_url(base_url: str) -> str: text = str(base_url or "").strip().rstrip("/") if not text: return "" if text.endswith("/api/v1"): return f"{text}/runtime/debug-ingest" if text.endswith("/api/v1/runtime"): return f"{text}/debug-ingest" return f"{text}/api/v1/runtime/debug-ingest" def ensure_debug_event_schema() -> None: global _DEBUG_SCHEMA_READY if _DEBUG_SCHEMA_READY: return with _DEBUG_SCHEMA_LOCK: if _DEBUG_SCHEMA_READY: return with get_db() as conn: conn.autocommit = False with conn.cursor() as cur: cur.execute(_DEBUG_SCHEMA_SQL) conn.commit() _DEBUG_SCHEMA_READY = True def append_debug_event( *, service: str, event_type: str, message: str, level: str = "info", payload: dict | None = None, source_region: str | None = None, node_code: str | None = None, ) -> int: ensure_debug_event_schema() normalized_region = _normalize_region(source_region, settings.node_region) normalized_node_code = str(node_code or settings.node_code).strip() or settings.node_code normalized_service = str(service or "").strip() or "unknown" normalized_event_type = str(event_type or "").strip() or "event" normalized_level = str(level or "info").strip() or "info" normalized_message = _normalize_message(message, fallback=f"{normalized_service}/{normalized_event_type}") payload_text = _safe_json_dumps(payload or {}) with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT id FROM detect_debug_events WHERE source_region = %s AND node_code = %s AND service = %s AND event_type = %s AND level = %s AND message = %s AND COALESCE(payload_json, '{}'::jsonb) = %s::jsonb AND created_at >= CURRENT_TIMESTAMP - interval '60 seconds' ORDER BY created_at DESC, id DESC LIMIT 1 """, ( normalized_region, normalized_node_code, normalized_service, normalized_event_type, normalized_level, normalized_message, payload_text, ), ) existing = cur.fetchone() if existing: return int(existing[0]) cur.execute( """ INSERT INTO detect_debug_events ( source_region, node_code, service, event_type, level, message, payload_json, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s::jsonb, CURRENT_TIMESTAMP) RETURNING id """, ( normalized_region, normalized_node_code, normalized_service, normalized_event_type, normalized_level, normalized_message, payload_text, ), ) record_id = int(cur.fetchone()[0]) conn.commit() return record_id def _load_debug_event_record(record_id: int) -> dict | None: with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT id, source_region, node_code, service, event_type, level, message, payload_json, created_at FROM detect_debug_events WHERE id = %s LIMIT 1 """, (int(record_id),), ) row = cur.fetchone() if not row: return None return { "id": int(row[0]), "source_region": str(row[1] or ""), "node_code": str(row[2] or ""), "service": str(row[3] or ""), "event_type": str(row[4] or ""), "level": str(row[5] or "info"), "message": str(row[6] or ""), "payload": row[7] if isinstance(row[7], dict) else {}, "created_at": _format_time(row[8]), } def _normalize_worker_log_event(debug_event: dict) -> dict | None: payload = dict(debug_event.get("payload") or {}) message = _normalize_message(debug_event.get("message"), fallback="worker_log") domain = str(payload.get("domain") or "").strip().lower() status = str(payload.get("status") or "").strip().lower() if not domain and ":" in message: domain = message.rsplit(":", 1)[-1].strip().lower() event_type = "" if status == "completed": event_type = "domain_completed" elif status == "failed": event_type = "domain_failed" elif status == "blacklisted": event_type = "domain_blacklisted" elif "开始检测域名" in message: event_type = "domain_started" elif "域名检测完成" in message: event_type = "domain_completed" payload.setdefault("status", "completed") elif "域名已命中黑名单" in message or "命中黑名单" in message: event_type = "domain_blacklisted" payload.setdefault("status", "blacklisted") elif "域名检测失败" in message: event_type = "domain_failed" payload.setdefault("status", "failed") if not event_type or not domain: return None payload.setdefault("domain", domain) payload["imported_from_debug_event"] = True payload["debug_event_record_id"] = int(debug_event.get("id") or 0) payload["debug_event_source_region"] = str(debug_event.get("source_region") or "") return { "node_code": str(debug_event.get("node_code") or "").strip(), "event_type": event_type, "level": str(debug_event.get("level") or "info"), "message": message, "payload": payload, "created_at": str(debug_event.get("created_at") or "").strip(), } def _normalize_debug_event_job_identity(payload: dict | None) -> dict: normalized_payload = dict(payload or {}) if isinstance(payload, dict) else {} nested_job = normalized_payload.get("job") if isinstance(normalized_payload.get("job"), dict) else {} raw_job_id = normalized_payload.get("job_id") if raw_job_id in (None, "", 0, "0"): raw_job_id = normalized_payload.get("target_job_id") if raw_job_id in (None, "", 0, "0"): raw_job_id = nested_job.get("job_id") try: job_id = int(raw_job_id or 0) except Exception: job_id = 0 job_code = str( normalized_payload.get("job_code") or normalized_payload.get("target_job_code") or nested_job.get("job_code") or "" ).strip() cycle_token = str(normalized_payload.get("cycle_token") or nested_job.get("cycle_token") or "").strip() return { "job_id": job_id, "job_code": job_code, "cycle_token": cycle_token, "has_identity": bool(job_id > 0 or job_code), } def _load_detect_job_summary_by_job_code(job_code: str, *, event_limit: int = 1) -> dict | None: normalized_job_code = str(job_code or "").strip() if not normalized_job_code: return None from app.services.detect_job_service import get_detect_job_summary with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT id FROM detect_jobs WHERE job_code = %s ORDER BY id DESC LIMIT 1 """, (normalized_job_code,), ) row = cur.fetchone() if not row: return None return get_detect_job_summary(int(row[0]), event_limit=event_limit) def _resolve_target_job_for_debug_event(event_payload: dict | None) -> tuple[dict | None, str]: from app.services.detect_job_service import get_active_detect_job_summary, get_detect_job_summary identity = _normalize_debug_event_job_identity(event_payload) if not identity["has_identity"]: return None, "missing_job_identity" payload_job_id = int(identity["job_id"] or 0) payload_job_code = str(identity["job_code"] or "").strip() payload_cycle_token = str(identity["cycle_token"] or "").strip() active_job = get_active_detect_job_summary(event_limit=1) or {} active_job_id = int(active_job.get("job_id") or 0) active_job_code = str(active_job.get("job_code") or active_job.get("runtime_job_code") or "").strip() target_job: dict | None = None if payload_job_id > 0: if active_job_id == payload_job_id: target_job = active_job else: target_job = get_detect_job_summary(payload_job_id, event_limit=1) elif payload_job_code: if active_job_code and active_job_code == payload_job_code: target_job = active_job else: target_job = _load_detect_job_summary_by_job_code(payload_job_code, event_limit=1) if not target_job: return None, "job_not_found" target_job_id = int(target_job.get("job_id") or 0) target_job_code = str(target_job.get("job_code") or target_job.get("runtime_job_code") or "").strip() target_cycle_token = str(target_job.get("current_cycle_token") or "").strip() if payload_job_id > 0 and target_job_id > 0 and target_job_id != payload_job_id: return None, "job_mismatch" if payload_job_code and target_job_code and target_job_code != payload_job_code: return None, "job_mismatch" if payload_cycle_token and target_cycle_token and payload_cycle_token != target_cycle_token: return None, "cycle_mismatch" return target_job, "matched" def _ingest_worker_log_into_active_job(debug_event: dict) -> dict: if str(debug_event.get("event_type") or "").strip() != "worker_log": return {"imported": False, "reason": "not_worker_log"} normalized_event = _normalize_worker_log_event(debug_event) if not normalized_event: return {"imported": False, "reason": "not_domain_progress_event"} from app.services.sync_push_service import ( _apply_detect_result_event_to_domain, _apply_detect_result_event_to_job_item, ) target_job, resolve_reason = _resolve_target_job_for_debug_event(normalized_event.get("payload")) if not target_job: return {"imported": False, "reason": resolve_reason} target_job_id = int(target_job.get("job_id") or 0) if target_job_id <= 0: return {"imported": False, "reason": "no_active_job"} debug_event_record_id = int(debug_event.get("id") or 0) updated_job_items = 0 with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT id FROM detect_run_events WHERE job_id = %s AND (payload_json->>'debug_event_record_id') = %s ORDER BY id DESC LIMIT 1 """, (target_job_id, str(debug_event_record_id)), ) existing = cur.fetchone() if existing: return { "imported": False, "reason": "deduplicated", "target_job_id": target_job_id, "detect_run_event_id": int(existing[0]), } created_at = _parse_time(normalized_event.get("created_at")) payload_json = _safe_json_dumps(normalized_event.get("payload") or {}) if created_at: cur.execute( """ INSERT INTO detect_run_events ( job_id, node_code, event_type, level, message, payload_json, created_at ) VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s) RETURNING id """, ( target_job_id, normalized_event["node_code"], normalized_event["event_type"], normalized_event["level"], normalized_event["message"], payload_json, created_at, ), ) else: cur.execute( """ INSERT INTO detect_run_events ( job_id, node_code, event_type, level, message, payload_json ) VALUES (%s, %s, %s, %s, %s, %s::jsonb) RETURNING id """, ( target_job_id, normalized_event["node_code"], normalized_event["event_type"], normalized_event["level"], normalized_event["message"], payload_json, ), ) detect_run_event_id = int((cur.fetchone() or [0])[0] or 0) _apply_detect_result_event_to_domain(cur, normalized_event) updated_job_items = _apply_detect_result_event_to_job_item( cur, target_job_id=target_job_id, event=normalized_event, ) conn.commit() return { "imported": True, "reason": "imported", "target_job_id": target_job_id, "target_job_code": str(target_job.get("job_code") or ""), "detect_run_event_id": detect_run_event_id, "updated_job_items": updated_job_items, "event_type": normalized_event["event_type"], "domain": str((normalized_event.get("payload") or {}).get("domain") or ""), } @db_read_retry() def list_debug_events( limit: int = 50, *, service: str | None = None, event_type: str | None = None, source_region: str | None = None, node_code: str | None = None, level: str | None = None, before_id: int | None = None, after_id: int | None = None, created_after: str | None = None, ) -> dict: ensure_debug_event_schema() safe_limit = max(1, min(int(limit or 50), 500)) conditions: list[str] = [] params: list[object] = [] if str(service or "").strip(): conditions.append("service = %s") params.append(str(service).strip()) if str(event_type or "").strip(): conditions.append("event_type = %s") params.append(str(event_type).strip()) if str(source_region or "").strip(): conditions.append("source_region = %s") params.append(str(source_region).strip()) if str(node_code or "").strip(): conditions.append("node_code = %s") params.append(str(node_code).strip()) if str(level or "").strip(): conditions.append("level = %s") params.append(str(level).strip()) if before_id is not None and int(before_id or 0) > 0: conditions.append("id < %s") params.append(int(before_id)) if after_id is not None and int(after_id or 0) > 0: conditions.append("id > %s") params.append(int(after_id)) parsed_created_after = _parse_time(created_after) if parsed_created_after: conditions.append("created_at >= %s") params.append(parsed_created_after) where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else "" with get_db() as conn: with conn.cursor() as cur: cur.execute( f""" SELECT id, source_region, node_code, service, event_type, level, message, payload_json, created_at FROM detect_debug_events {where_clause} ORDER BY created_at DESC, id DESC LIMIT %s """, (*params, safe_limit + 1), ) rows = cur.fetchall() has_more = len(rows) > safe_limit rows = rows[:safe_limit] records = [ { "id": row[0], "source_region": row[1], "node_code": row[2], "service": row[3], "event_type": row[4], "level": row[5], "message": row[6], "payload": row[7] if isinstance(row[7], dict) else {}, "created_at": _format_time(row[8]), } for row in rows ] next_before_id = records[-1]["id"] if has_more and records else None latest_id = records[0]["id"] if records else None return { "records": records, "has_more": has_more, "next_before_id": next_before_id, "latest_id": latest_id, } @db_read_retry() def get_debug_event_overview(*, window_minutes: int = 10, source_region: str | None = None) -> dict: ensure_debug_event_schema() safe_window = max(1, min(int(window_minutes or 10), 180)) conditions = ["created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval"] params: list[object] = [safe_window] if str(source_region or "").strip(): conditions.append("source_region = %s") params.append(str(source_region).strip()) where_clause = f"WHERE {' AND '.join(conditions)}" with get_db() as conn: with conn.cursor() as cur: cur.execute( f""" SELECT level, count(*) FROM detect_debug_events {where_clause} GROUP BY level """, tuple(params), ) level_counts = {str(level or "info"): int(count) for level, count in cur.fetchall()} cur.execute( f""" SELECT service, event_type, level, count(*) AS total FROM detect_debug_events {where_clause} GROUP BY service, event_type, level ORDER BY total DESC, service ASC, event_type ASC LIMIT 12 """, tuple(params), ) hot_events = [ { "service": str(row[0] or ""), "event_type": str(row[1] or ""), "level": str(row[2] or "info"), "count": int(row[3] or 0), } for row in cur.fetchall() ] cur.execute( f""" SELECT CASE WHEN event_type LIKE 'task_pull%%' THEN 'task_pull' WHEN event_type LIKE 'detect_result_projection_sync%%' THEN 'detect_result_sync' WHEN event_type LIKE 'runtime_projection_sync%%' THEN 'runtime_projection_sync' WHEN event_type LIKE 'sync_push%%' THEN 'sync_push' WHEN event_type LIKE 'runtime_action%%' THEN 'runtime_action' ELSE 'other' END AS bucket, count(*) AS total, count(*) FILTER (WHERE level = 'warning') AS warnings, count(*) FILTER (WHERE level = 'error') AS errors FROM detect_debug_events {where_clause} GROUP BY bucket ORDER BY total DESC, bucket ASC """, tuple(params), ) trend_buckets = [ { "bucket": str(row[0] or "other"), "total": int(row[1] or 0), "warnings": int(row[2] or 0), "errors": int(row[3] or 0), } for row in cur.fetchall() ] cur.execute( f""" SELECT id, source_region, node_code, service, event_type, level, message, payload_json, created_at FROM detect_debug_events {where_clause} AND level IN ('warning', 'error') ORDER BY created_at DESC, id DESC LIMIT 8 """, tuple(params), ) recent_issues = [ { "id": row[0], "source_region": row[1], "node_code": row[2], "service": row[3], "event_type": row[4], "level": row[5], "message": row[6], "payload": row[7] if isinstance(row[7], dict) else {}, "created_at": _format_time(row[8]), } for row in cur.fetchall() ] cur.execute( f""" SELECT source_region, service, event_type, level, message, count(*) AS total, min(created_at) AS first_seen_at, max(created_at) AS last_seen_at FROM detect_debug_events {where_clause} AND level IN ('warning', 'error') GROUP BY source_region, service, event_type, level, message ORDER BY total DESC, last_seen_at DESC LIMIT 12 """, tuple(params), ) issue_groups = [ { "source_region": str(row[0] or ""), "service": str(row[1] or ""), "event_type": str(row[2] or ""), "level": str(row[3] or "info"), "message": str(row[4] or ""), "count": int(row[5] or 0), "first_seen_at": _format_time(row[6]), "last_seen_at": _format_time(row[7]), "risk_score": int(row[5] or 0) * (3 if str(row[3] or "info") == "error" else 2), "risk_label": "high" if (int(row[5] or 0) >= 3 and str(row[3] or "info") == "error") else ("medium" if int(row[5] or 0) >= 3 or str(row[3] or "info") == "error" else "low"), } for row in cur.fetchall() ] total = sum(level_counts.values()) return { "window_minutes": safe_window, "source_region": str(source_region or "").strip(), "total": total, "level_counts": { "info": int(level_counts.get("info", 0)), "warning": int(level_counts.get("warning", 0)), "error": int(level_counts.get("error", 0)), }, "trend_buckets": trend_buckets, "hot_events": hot_events, "recent_issues": recent_issues, "issue_groups": issue_groups, } def get_debug_diagnosis(*, window_minutes: int = 10, source_region: str | None = None, sync_summary: dict | None = None) -> dict: overview = get_debug_event_overview(window_minutes=window_minutes, source_region=source_region) summary = sync_summary or {} latest_record = summary.get("latest_record") or {} latest_result_batch = summary.get("latest_detect_result_batch") or {} detect_result_batches = summary.get("detect_result_batches") or {} projected = int((detect_result_batches.get("state_counts") or {}).get("projected", 0) or 0) bucket_map = { str(item.get("bucket") or "other"): item for item in (overview.get("trend_buckets") or []) if isinstance(item, dict) } task_pull = bucket_map.get("task_pull", {}) sync_push = bucket_map.get("sync_push", {}) result_sync = bucket_map.get("detect_result_sync", {}) runtime_action = bucket_map.get("runtime_action", {}) def _build( *, status: str, title: str, summary_text: str, suspect: str, evidence: str, action: str, next_step: str, confidence: str, suggested_checks: list[str], recommended_action_api: str, ) -> dict: return { "status": status, "title": title, "summary": summary_text, "suspect": suspect, "evidence": evidence, "action": action, "next_step": next_step, "confidence": confidence, "suggested_checks": suggested_checks, "recommended_action_api": recommended_action_api, "window_minutes": int(overview.get("window_minutes", window_minutes) or window_minutes), "source_region": str(source_region or "").strip(), } if int(result_sync.get("errors", 0) or 0) > 0 or str(latest_result_batch.get("latest_push_status") or "").strip().lower() == "failed": return _build( status="danger", title="结果同步失败", summary_text="任务执行可能仍在继续，但检测结果跨地域回传已经出现明确失败。", suspect="sync-agent 到海外 ingest / 目标接收面", evidence=f"detect_result_sync errors={int(result_sync.get('errors', 0) or 0)}；latest_push_status={str(latest_result_batch.get('latest_push_status') or '-')}", action="优先检查结果推送链路", next_step="先查看 latest_push_error，再核对海外端 /runtime/sync-ingest 接收和网络连通性。", confidence="high", suggested_checks=[ "curl -s http://127.0.0.1:8100/api/v1/runtime/sync-summary", "journalctl -u domaincheck-sync-agent -n 50 --no-pager", "curl -s https://api-domain.aakkx.top/api/v1/runtime/readiness", ], recommended_action_api="push_sync", ) if projected > 0 or int(result_sync.get("warnings", 0) or 0) > 0: return _build( status="warning", title="结果同步滞留", summary_text="结果投影已经生成，但尚未稳定进入 delivered/synced。", suspect="sync-agent 推送窗口 / 远端 ack / 目标侧消费速度", evidence=f"projected={projected}；detect_result_sync warnings={int(result_sync.get('warnings', 0) or 0)}", action="优先检查 sync-agent 与远端确认", next_step="结合结果同步摘要中的最近 Push/Ingress 时间，确认是卡在推送前还是目标接收后。", confidence="medium", suggested_checks=[ "curl -s http://127.0.0.1:8100/api/v1/runtime/sync-summary", "curl -s -X POST http://127.0.0.1:8100/api/v1/runtime/actions/push_sync", "journalctl -u domaincheck-sync-agent -n 50 --no-pager", ], recommended_action_api="push_sync", ) if int(task_pull.get("errors", 0) or 0) > 0 or int(task_pull.get("warnings", 0) or 0) > 0: return _build( status="warning", title="任务拉取链路异常", summary_text="控制面最近从海外拉取待检测任务时出现失败或半成功。", suspect="task-export / task-ack / controller 入库", evidence=f"task_pull warnings={int(task_pull.get('warnings', 0) or 0)} errors={int(task_pull.get('errors', 0) or 0)}", action="优先检查任务拉取与确认", next_step="筛选 task_pull_success/partial/failed，确认是 export、ack 还是本地 ingest 失败。", confidence="high" if int(task_pull.get("errors", 0) or 0) > 0 else "medium", suggested_checks=[ "curl -s -X POST http://127.0.0.1:8100/api/v1/runtime/actions/pull_tasks", "curl -s http://127.0.0.1:8100/api/v1/runtime/debug-overview", "journalctl -u domaincheck-api -n 80 --no-pager", ], recommended_action_api="pull_tasks", ) if int(runtime_action.get("errors", 0) or 0) > 0 or int(runtime_action.get("warnings", 0) or 0) > 0: return _build( status="warning", title="控制动作链路波动", summary_text="运行时动作请求已发出，但 controller 对 worker/api/sync-agent 的控制链路近期不稳定。", suspect="systemctl 权限 / 本机环境 / 控制信道", evidence=f"runtime_action warnings={int(runtime_action.get('warnings', 0) or 0)} errors={int(runtime_action.get('errors', 0) or 0)}", action="优先检查控制动作执行侧", next_step="对比 runtime_action_requested/runtime_action_finished，确认是否请求到了但执行失败。", confidence="medium", suggested_checks=[ "curl -s http://127.0.0.1:8100/api/v1/runtime/status", "systemctl status domaincheck-worker --no-pager -l", "journalctl -u domaincheck-api -n 80 --no-pager", ], recommended_action_api="", ) if str(latest_record.get("status") or "").strip().lower() == "failed" or int(sync_push.get("errors", 0) or 0) > 0: return _build( status="warning", title="运行态同步异常", summary_text="控制面心跳/运行态同步最近出现失败，可能影响海外端看到的集群状态。", suspect="runtime_projection 推送 / 海外 runtime ingest", evidence=f"latest_record_status={str(latest_record.get('status') or '-')}；sync_push errors={int(sync_push.get('errors', 0) or 0)}", action="优先检查运行态同步", next_step="筛选 runtime_projection_sync_failed，检查海外 /runtime/sync-ingest 是否正常接收。", confidence="medium", suggested_checks=[ "curl -s http://127.0.0.1:8100/api/v1/runtime/readiness", "curl -s http://127.0.0.1:8100/api/v1/runtime/sync-summary", "journalctl -u domaincheck-sync-agent -n 50 --no-pager", ], recommended_action_api="push_sync", ) return _build( status="ok", title="当前链路整体稳定", summary_text="任务拉取、运行态同步、结果同步和控制动作近窗内都没有明显异常。", suspect="暂无明确阻塞段", evidence=( f"task_pull={int(task_pull.get('total', 0) or 0)} " f"sync_push={int(sync_push.get('total', 0) or 0)} " f"detect_result_sync={int(result_sync.get('total', 0) or 0)}" ), action="继续观察即可", next_step="如果要排更细问题，优先看结果同步摘要和最近问题列表。", confidence="medium", suggested_checks=[ "curl -s http://127.0.0.1:8100/api/v1/runtime/readiness", "curl -s http://127.0.0.1:8100/api/v1/runtime/debug-overview", ], recommended_action_api="", ) def get_debug_handoff_report( *, window_minutes: int = 10, source_region: str | None = None, sync_summary: dict | None = None, readiness: dict | None = None, ) -> dict: summary = sync_summary or {} overview = get_debug_event_overview(window_minutes=window_minutes, source_region=source_region) diagnosis = get_debug_diagnosis(window_minutes=window_minutes, source_region=source_region, sync_summary=summary) readiness_payload = readiness or {} latest_record = summary.get("latest_record") or {} latest_result_batch = summary.get("latest_detect_result_batch") or {} level_counts = overview.get("level_counts") or {} trend_buckets = overview.get("trend_buckets") or [] recent_issues = overview.get("recent_issues") or [] issue_groups = overview.get("issue_groups") or [] lines = [ f"诊断结论: {diagnosis.get('title', '-')}", f"状态: {diagnosis.get('status', '-')} | 置信度: {diagnosis.get('confidence', '-')}", f"摘要: {diagnosis.get('summary', '-')}", f"优先怀疑: {diagnosis.get('suspect', '-')}", f"证据: {diagnosis.get('evidence', '-')}", f"建议动作: {diagnosis.get('action', '-')}", f"下一步: {diagnosis.get('next_step', '-')}", f"推荐 API 动作: {diagnosis.get('recommended_action_api', '-') or '-'}", f"集群就绪: {readiness_payload.get('status', '-')}/{readiness_payload.get('summary', '-')}", "", f"近 {overview.get('window_minutes', window_minutes)} 分钟事件: total={overview.get('total', 0)} info={level_counts.get('info', 0)} warning={level_counts.get('warning', 0)} error={level_counts.get('error', 0)}", "趋势桶: " + (" | ".join( f"{item.get('bucket', 'other')} total={item.get('total', 0)} warn={item.get('warnings', 0)} err={item.get('errors', 0)}" for item in trend_buckets ) or "-"), f"最新同步记录: {latest_record.get('sync_type', '-')}/{latest_record.get('status', '-')}/{latest_record.get('created_at', '-')}", f"最新结果批次: {latest_result_batch.get('job_code', '-')}/{latest_result_batch.get('sync_state', '-')}/push={latest_result_batch.get('latest_push_status', '-')}/ingest={latest_result_batch.get('latest_ingest_status', '-')}", "最近问题:", ] if recent_issues: for issue in recent_issues[:5]: lines.append( f"- [{issue.get('created_at', '-')}] {issue.get('service', '-')}/{issue.get('event_type', '-')} {issue.get('level', '-')} {issue.get('message', '-')}" ) else: lines.append("- 近窗内没有 warning/error") checks = diagnosis.get("suggested_checks") or [] lines.append("建议命令:") if checks: for cmd in checks: lines.append(f"- {cmd}") else: lines.append("- 暂无") compact_lines = [ f"[{diagnosis.get('status', '-')}] {diagnosis.get('title', '-')}", f"摘要: {diagnosis.get('summary', '-')}", f"怀疑点: {diagnosis.get('suspect', '-')}", f"动作: {diagnosis.get('action', '-')} | API: {diagnosis.get('recommended_action_api', '-') or '-'}", f"证据: {diagnosis.get('evidence', '-')}", f"就绪: {readiness_payload.get('status', '-')}/{readiness_payload.get('summary', '-')}", ] if issue_groups: compact_lines.append("重复问题:") for issue in issue_groups[:3]: compact_lines.append( f"- {issue.get('service', '-')}/{issue.get('event_type', '-')} " f"{issue.get('level', '-')} x{issue.get('count', 0)}: {issue.get('message', '-')}" ) else: compact_lines.append("重复问题: 近窗内没有 warning/error 聚集") compact_lines.append("建议命令:") if checks: for cmd in checks[:3]: compact_lines.append(f"- {cmd}") else: compact_lines.append("- 暂无") failure_handoff = { "generated_from": "runtime/debug-handoff", "source_region": str(source_region or "").strip(), "window_minutes": overview.get("window_minutes", window_minutes), "diagnosis": diagnosis, "readiness": readiness_payload, "latest_record": latest_record, "latest_detect_result_batch": latest_result_batch, "issue_groups": issue_groups[:8], "recent_issues": recent_issues[:8], "suggested_checks": checks, } return { "window_minutes": overview.get("window_minutes", window_minutes), "source_region": str(source_region or "").strip(), "diagnosis": diagnosis, "overview": overview, "issue_groups": issue_groups, "readiness": readiness_payload, "sync_summary": { "latest_record": latest_record, "latest_detect_result_batch": latest_result_batch, }, "failure_handoff": failure_handoff, "compact_report_text": "\n".join(compact_lines), "report_text": "\n".join(lines), } def ingest_debug_event(payload: dict, *, shared_token: str | None = None) -> tuple[bool, str, dict]: configured_token = str(settings.sync_shared_token or "").strip() incoming_token = str(shared_token or "").strip() if not configured_token: return False, "调试事件共享 token 未配置，拒绝远端写入", {"configuration_required": True} if incoming_token != configured_token: return False, "调试事件 token 校验失败", {} record_id = append_debug_event( source_region=str(payload.get("source_region") or settings.node_region), node_code=str(payload.get("node_code") or ""), service=str(payload.get("service") or "unknown"), event_type=str(payload.get("event_type") or "event"), level=str(payload.get("level") or "info"), message=_normalize_message(payload.get("message"), fallback="remote debug event"), payload=payload.get("payload") or {}, ) debug_event = _load_debug_event_record(record_id) or { "id": int(record_id), "source_region": str(payload.get("source_region") or settings.node_region), "node_code": str(payload.get("node_code") or ""), "service": str(payload.get("service") or "unknown"), "event_type": str(payload.get("event_type") or "event"), "level": str(payload.get("level") or "info"), "message": _normalize_message(payload.get("message"), fallback="remote debug event"), "payload": payload.get("payload") or {}, "created_at": "", } job_import = {} if str(debug_event.get("service") or "").strip() == "worker-event": try: job_import = _ingest_worker_log_into_active_job(debug_event) except Exception as exc: job_import = {"imported": False, "reason": f"job_import_failed: {exc}"} return True, "调试事件接收成功", {"record_id": record_id, "job_import": job_import} def push_debug_event( *, service: str, event_type: str, message: str, level: str = "info", payload: dict | None = None, ) -> tuple[bool, str, dict]: append_debug_event( service=service, event_type=event_type, message=message, level=level, payload=payload, ) if not settings.sync_push_enabled: return False, "未启用调试事件远端回传", {"local_only": True} ingest_url = _debug_ingest_url(settings.sync_target_api_base_url) if not ingest_url: return False, "未配置调试事件远端地址", {"local_only": True} request_payload = { "source_region": settings.node_region, "node_code": settings.node_code, "hostname": socket.gethostname(), "service": str(service or "").strip() or "unknown", "event_type": str(event_type or "").strip() or "event", "level": str(level or "info").strip() or "info", "message": _normalize_message(message, fallback=f"{service}/{event_type}"), "payload": payload or {}, } request = urllib.request.Request( ingest_url, data=_safe_json_dumps(request_payload).encode("utf-8"), headers={ "Content-Type": "application/json", **({"X-Domaincheck-Sync-Token": settings.sync_shared_token} if settings.sync_shared_token else {}), }, method="POST", ) try: with urllib.request.urlopen(request, timeout=10) as response: raw = response.read().decode("utf-8") response_data = json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: error_body = exc.read().decode("utf-8", errors="ignore") if hasattr(exc, "read") else "" return False, f"调试事件远端回传失败: HTTP {getattr(exc, 'code', 500)}", { "http_status": getattr(exc, "code", 500), "response_text": error_body[:500], } except json.JSONDecodeError as exc: return False, f"调试事件远端回传失败: 响应不是合法 JSON ({exc})", {} except Exception as exc: return False, f"调试事件远端回传失败: {exc}", {} if not _response_code_ok(response_data.get("code")): return False, str(response_data.get("message") or "调试事件远端回传失败"), {"response": response_data} return True, "调试事件已回传海外", {"response": response_data}