from __future__ import annotations from copy import deepcopy from uuid import uuid4 from app.core.config import settings from app.services.ops_execution_mode_service import ( build_execution_mode_options, decorate_execution_mode_fields, execution_mode_label, ) from app.services.ops_agent_service import list_managed_nodes_with_agent_state, list_ops_job_events_for_jobs from app.services.ops_job_service import cancel_ops_job, create_ops_job_batch, list_ops_jobs from app.services.ops_template_service import build_ops_template_payload, get_ops_action_template _NODE_AGENT_SERVICE_NAME = "domaincheck-node-agent" _PLAYBOOK_RUN_ACTIVE_STATUSES = {"queued", "dispatching", "running", "awaiting_approval"} _PLAYBOOK_RUN_TERMINAL_PROBLEM_STATUSES = {"failed", "blocked", "cancelled"} def _normalize_focus_ref(focus_ref: object) -> dict: if not isinstance(focus_ref, dict): return {} normalized: dict[str, object] = {} for raw_key, raw_value in focus_ref.items(): key = str(raw_key or "").strip() if not key: continue if isinstance(raw_value, str): value = raw_value.strip() if value: normalized[key] = value continue if isinstance(raw_value, bool): normalized[key] = raw_value continue if isinstance(raw_value, (int, float)): if raw_value: normalized[key] = raw_value continue if raw_value is not None: normalized[key] = raw_value return normalized def _normalize_node_codes(raw_node_codes: object) -> list[str]: normalized: list[str] = [] seen: set[str] = set() for item in list(raw_node_codes or []): node_code = str(item or "").strip() if not node_code or node_code in seen: continue seen.add(node_code) normalized.append(node_code) return normalized def _build_playbook_run_code() -> str: return f"pbr-{uuid4().hex[:10]}" def _slim_playbook_job(job: dict | None = None) -> dict: normalized_job = dict(job or {}) status = str(normalized_job.get("status") or "") return { "id": int(normalized_job.get("id") or 0), "job_code": str(normalized_job.get("job_code") or ""), "action": str(normalized_job.get("action") or ""), "status": status, "status_label": _playbook_run_status_label(status), "target_node_code": str(normalized_job.get("target_node_code") or ""), "created_at": str(normalized_job.get("created_at") or ""), "started_at": str(normalized_job.get("started_at") or ""), "finished_at": str(normalized_job.get("finished_at") or ""), "updated_at": str(normalized_job.get("updated_at") or ""), "error_message": str(normalized_job.get("error_message") or ""), } def _increment_status(status_counts: dict[str, int], status: str) -> None: normalized_status = str(status or "").strip() or "unknown" status_counts[normalized_status] = int(status_counts.get(normalized_status, 0) or 0) + 1 def _latest_job_time(job: dict) -> str: return ( str(job.get("updated_at") or "").strip() or str(job.get("finished_at") or "").strip() or str(job.get("started_at") or "").strip() or str(job.get("created_at") or "").strip() ) def _derive_run_status(*, status_counts: dict[str, int], jobs_total: int) -> str: if jobs_total <= 0: return "queued" if sum(int(status_counts.get(status, 0) or 0) for status in _PLAYBOOK_RUN_TERMINAL_PROBLEM_STATUSES) > 0: return "attention" if sum(int(status_counts.get(status, 0) or 0) for status in _PLAYBOOK_RUN_ACTIVE_STATUSES) > 0: return "running" if int(status_counts.get("success", 0) or 0) >= jobs_total: return "success" return "queued" def _terminal_problem_label(status_counts: dict[str, int]) -> str: if int(status_counts.get("failed", 0) or 0) > 0: return "失败" if int(status_counts.get("blocked", 0) or 0) > 0: return "阻断" if int(status_counts.get("cancelled", 0) or 0) > 0: return "已取消" return "" def _active_status_label(status_counts: dict[str, int]) -> str: if int(status_counts.get("running", 0) or 0) > 0: return "执行中" if int(status_counts.get("dispatching", 0) or 0) > 0: return "待派发" if int(status_counts.get("awaiting_approval", 0) or 0) > 0: return "待审批" if int(status_counts.get("queued", 0) or 0) > 0: return "排队中" return "" def _playbook_run_status_label(status: object) -> str: normalized_status = str(status or "").strip() mapping = { "queued": "已创建", "running": "收口中", "success": "成功", "attention": "待关注", } return mapping.get(normalized_status, normalized_status or "未知") def _build_playbook_run_focus_ref( run: dict | None = None, *, step_key: str = "", step_title: str = "", event_key: str = "", node_code: str = "", ) -> dict: normalized_run = dict(run or {}) return { "kind": "playbook_run", "run_code": str(normalized_run.get("run_code") or "").strip(), "playbook_key": str(normalized_run.get("playbook_key") or "").strip(), "group_key": str(normalized_run.get("group_key") or "").strip(), "focus_step_key": str(step_key or normalized_run.get("focus_step_key") or "").strip(), "focus_step_title": str(step_title or normalized_run.get("focus_step_title") or "").strip(), "event_key": str(event_key or "").strip(), "node_code": str(node_code or "").strip(), } def _build_playbook_step_summary_text(step: dict | None = None) -> str: normalized_step = dict(step or {}) title = str(normalized_step.get("title") or normalized_step.get("step_key") or "步骤").strip() status = str(normalized_step.get("status") or "").strip() status_counts = dict(normalized_step.get("status_counts") or {}) if status == "attention": problem_label = _terminal_problem_label(status_counts) or "异常" return f"{title} 当前出现{problem_label}，建议先查看该步骤事件流。" if status == "running": active_label = _active_status_label(status_counts) or "处理中" return f"{title} 当前处于{active_label}，可继续观察该步骤事件。" if status == "success": return f"{title} 已完成收口。" if status == "queued": return f"{title} 已创建，等待后续任务推进。" return f"{title} 当前状态为 {_playbook_run_status_label(status)}。" def _build_playbook_run_summary_text(run: dict | None = None) -> str: normalized_run = dict(run or {}) focus_summary = str(normalized_run.get("focus_summary") or "").strip() if focus_summary: return focus_summary title = str(normalized_run.get("playbook_title") or normalized_run.get("playbook_key") or "编排").strip() status = str(normalized_run.get("status") or "").strip() if status == "success": return f"{title} 已全部收口。" if status == "attention": return f"{title} 当前存在需要优先处理的异常步骤。" if status == "running": return f"{title} 仍在执行中，可继续查看步骤与事件流。" if status == "queued": return f"{title} 已创建，等待执行推进。" return f"{title} 当前状态为 {_playbook_run_status_label(status)}。" def _build_bootstrap_target_state_counts(target_node_codes: list[str], managed_node_map: dict[str, dict]) -> dict[str, int]: counts = { "total": 0, "online": 0, "pending_bootstrap": 0, "runtime_only": 0, "stale": 0, "unmanaged": 0, "unknown": 0, } normalized_codes = _normalize_node_codes(target_node_codes) counts["total"] = len(normalized_codes) for node_code in normalized_codes: node = dict(managed_node_map.get(node_code) or {}) if not node: counts["unknown"] += 1 continue state = str(node.get("agent_state") or "").strip() if state in {"online", "online_busy"}: counts["online"] += 1 continue if state in {"pending_bootstrap", "runtime_only", "stale"}: counts[state] += 1 continue if bool(node.get("is_managed", False)): counts["unknown"] += 1 else: counts["unmanaged"] += 1 return counts def _format_bootstrap_state_parts(state_counts: dict[str, int]) -> list[str]: parts: list[str] = [] if int(state_counts.get("online", 0) or 0) > 0: parts.append(f"已在线 {int(state_counts.get('online', 0) or 0)}") if int(state_counts.get("pending_bootstrap", 0) or 0) > 0: parts.append(f"待回连 {int(state_counts.get('pending_bootstrap', 0) or 0)}") if int(state_counts.get("runtime_only", 0) or 0) > 0: parts.append(f"仅 runtime 在线 {int(state_counts.get('runtime_only', 0) or 0)}") if int(state_counts.get("stale", 0) or 0) > 0: parts.append(f"心跳过期 {int(state_counts.get('stale', 0) or 0)}") if int(state_counts.get("unmanaged", 0) or 0) > 0: parts.append(f"未纳管 {int(state_counts.get('unmanaged', 0) or 0)}") if int(state_counts.get("unknown", 0) or 0) > 0: parts.append(f"待确认 {int(state_counts.get('unknown', 0) or 0)}") return parts def _enrich_onboarding_bootstrap_run(run: dict, managed_node_map: dict[str, dict]) -> None: normalized_run = dict(run or {}) target_node_codes = _normalize_node_codes(normalized_run.get("target_node_codes") or []) if not target_node_codes: return state_counts = _build_bootstrap_target_state_counts(target_node_codes, managed_node_map) state_parts = _format_bootstrap_state_parts(state_counts) total = int(state_counts.get("total", 0) or 0) run["bootstrap_state_counts"] = state_counts run_status = str(run.get("status") or "").strip() if run_status == "attention": summary = str(run.get("focus_summary") or "").strip() or "接入工单编排当前出现异常。" run["focus_summary"] = f"{summary} 可直接进入任务详情查看 bootstrap 结果或错误回执。" return if run_status == "running": run["focus_level"] = "warning" run["focus_summary"] = ( f"正在为 {total} 个节点生成接入工单，完成后可直接复制 bootstrap 脚本、env 和一键落地命令。" ) return if run_status == "queued": run["focus_level"] = "info" run["focus_summary"] = "接入工单编排已创建，等待控制面生成 bootstrap 方案。" return if run_status != "success": return if total == 1: node_code = target_node_codes[0] node = dict(managed_node_map.get(node_code) or {}) node_state = str(node.get("agent_state") or "").strip() if node_state in {"online", "online_busy"}: run["focus_level"] = "success" run["focus_summary"] = "接入工单已生成，目标节点已完成 Agent 接入，可继续执行接管后验收。" return if node_state == "pending_bootstrap": run["focus_level"] = "warning" run["focus_summary"] = "接入工单已生成，节点待执行 bootstrap 脚本并回连 Agent。" return if node_state == "runtime_only": run["focus_level"] = "warning" run["focus_summary"] = "接入工单已生成，但当前只有 runtime 心跳，Node Agent 还未真正接管。" return if node_state == "stale": run["focus_level"] = "warning" run["focus_summary"] = "接入工单已生成，但 Agent 心跳已过期，建议先看任务详情和 Agent 日志。" return run["focus_level"] = "success" if int(state_counts.get("online", 0) or 0) >= total else "warning" if int(state_counts.get("online", 0) or 0) >= total: run["focus_summary"] = f"已为 {total} 个节点生成接入工单，目标节点已全部完成 Agent 接入，可继续执行接管后验收。" return state_text = " / ".join(state_parts) if state_parts else "状态待确认" run["focus_summary"] = ( f"已为 {total} 个节点生成接入工单，当前状态：{state_text}。" " 下一步可进入任务详情复制 bootstrap 脚本，并在节点执行后再做接管验收。" ) def _build_playbook_job_metadata( *, playbook: dict, preview: dict, step: dict, run_code: str, requested_by: str, execution_mode: str, ) -> dict: return { "playbook": { "run_code": run_code, "key": str(playbook.get("key") or "").strip(), "title": str(playbook.get("title") or "").strip(), "group_key": str(playbook.get("group_key") or "").strip(), "group_title": str(playbook.get("group_title") or "").strip(), "requested_by": requested_by, "execution_mode": execution_mode, "execution_mode_label": execution_mode_label(execution_mode), "step_key": str(step.get("step_key") or "").strip(), "step_title": str(step.get("title") or "").strip(), "step_order": int(step.get("order") or 0), "step_count": int(preview.get("step_count") or 0), "expected_jobs_total": int(preview.get("expected_jobs_total") or 0), "target_nodes_total": int(preview.get("target_nodes_total") or 0), "stop_on_failure": bool(playbook.get("stop_on_failure", True)), "template_key": str(step.get("template_key") or "").strip(), } } def _list_playbook_run_jobs(run_code: str, *, scan_limit: int = 1000) -> list[dict]: normalized_run_code = str(run_code or "").strip() if not normalized_run_code: return [] safe_scan_limit = min(max(int(scan_limit or 1000), 1), 2000) jobs = list_ops_jobs(limit=safe_scan_limit, compact=False) return [ job for job in jobs if str(((job.get("metadata") or {}).get("playbook") or {}).get("run_code") or "").strip() == normalized_run_code ] def _group_playbook_runs_from_jobs(jobs: list[dict], *, limit: int = 12, run_code_filter: str = "") -> dict: safe_limit = min(max(int(limit or 12), 1), 200) normalized_filter = str(run_code_filter or "").strip() grouped_runs: dict[str, dict] = {} for job in jobs: playbook_meta = dict((job.get("metadata") or {}).get("playbook") or {}) run_code = str(playbook_meta.get("run_code") or "").strip() if not run_code: continue if normalized_filter and run_code != normalized_filter: continue step_key = str(playbook_meta.get("step_key") or "").strip() step_order = int(playbook_meta.get("step_order") or 0) step_title = str(playbook_meta.get("step_title") or step_key or "").strip() template_key = str(playbook_meta.get("template_key") or "").strip() job_status = str(job.get("status") or "").strip() target_node_code = str(job.get("target_node_code") or "").strip() latest_time = _latest_job_time(job) run = grouped_runs.get(run_code) if not run: run = { "run_code": run_code, "playbook_key": str(playbook_meta.get("key") or "").strip(), "playbook_title": str(playbook_meta.get("title") or "").strip(), "group_key": str(playbook_meta.get("group_key") or "").strip(), "group_title": str(playbook_meta.get("group_title") or "").strip(), "requested_by": str(playbook_meta.get("requested_by") or job.get("requested_by") or "").strip(), "execution_mode": str(playbook_meta.get("execution_mode") or job.get("execution_mode") or "").strip(), "execution_mode_label": ( str(playbook_meta.get("execution_mode_label") or "").strip() or execution_mode_label(str(playbook_meta.get("execution_mode") or job.get("execution_mode") or "").strip()) ), "stop_on_failure": bool(playbook_meta.get("stop_on_failure", True)), "step_count": int(playbook_meta.get("step_count") or 0), "expected_jobs_total": int(playbook_meta.get("expected_jobs_total") or 0), "target_nodes_total": int(playbook_meta.get("target_nodes_total") or 0), "target_node_codes": [], "created_at": str(job.get("created_at") or ""), "updated_at": latest_time, "status_counts": {}, "jobs_total": 0, "success_jobs_total": 0, "terminal_jobs_total": 0, "steps": [], "latest_job": {}, "_target_node_set": set(), "_step_map": {}, } grouped_runs[run_code] = run run["jobs_total"] = int(run.get("jobs_total", 0) or 0) + 1 _increment_status(run["status_counts"], job_status) if job_status == "success": run["success_jobs_total"] = int(run.get("success_jobs_total", 0) or 0) + 1 if job_status == "success" or job_status in _PLAYBOOK_RUN_TERMINAL_PROBLEM_STATUSES: run["terminal_jobs_total"] = int(run.get("terminal_jobs_total", 0) or 0) + 1 if target_node_code and target_node_code not in run["_target_node_set"]: run["_target_node_set"].add(target_node_code) run["target_node_codes"].append(target_node_code) created_at = str(job.get("created_at") or "").strip() if created_at and (not str(run.get("created_at") or "").strip() or created_at < str(run.get("created_at") or "").strip()): run["created_at"] = created_at if latest_time and latest_time >= str(run.get("updated_at") or "").strip(): run["updated_at"] = latest_time run["latest_job"] = _slim_playbook_job(job) step_map = run["_step_map"] step_row = step_map.get(step_key) if not step_row: step_row = { "step_key": step_key, "title": step_title, "template_key": template_key, "order": step_order, "jobs_total": 0, "nodes_total": 0, "status_counts": {}, "terminal_jobs_total": 0, "success_jobs_total": 0, "latest_job": {}, "_node_set": set(), } step_map[step_key] = step_row run["steps"].append(step_row) step_row["jobs_total"] = int(step_row.get("jobs_total", 0) or 0) + 1 _increment_status(step_row["status_counts"], job_status) if job_status == "success": step_row["success_jobs_total"] = int(step_row.get("success_jobs_total", 0) or 0) + 1 if job_status == "success" or job_status in _PLAYBOOK_RUN_TERMINAL_PROBLEM_STATUSES: step_row["terminal_jobs_total"] = int(step_row.get("terminal_jobs_total", 0) or 0) + 1 if target_node_code and target_node_code not in step_row["_node_set"]: step_row["_node_set"].add(target_node_code) step_row["nodes_total"] = len(step_row["_node_set"]) current_latest = _latest_job_time(step_row.get("latest_job") or {}) if latest_time and latest_time >= current_latest: step_row["latest_job"] = _slim_playbook_job(job) runs = list(grouped_runs.values()) managed_node_map: dict[str, dict] = {} try: managed_nodes_payload = list_managed_nodes_with_agent_state() managed_node_map = { str(item.get("node_code") or "").strip(): dict(item or {}) for item in list(managed_nodes_payload.get("nodes") or []) if str(item.get("node_code") or "").strip() } except Exception: managed_node_map = {} for run in runs: run["target_nodes_total"] = int(run.get("target_nodes_total") or len(run.get("target_node_codes") or [])) run["status"] = _derive_run_status( status_counts=dict(run.get("status_counts") or {}), jobs_total=int(run.get("jobs_total", 0) or 0), ) run["completion_percent"] = round( (int(run.get("terminal_jobs_total", 0) or 0) / max(int(run.get("jobs_total", 0) or 0), 1)) * 100, 1, ) for step in list(run.get("steps") or []): step["status"] = _derive_run_status( status_counts=dict(step.get("status_counts") or {}), jobs_total=int(step.get("jobs_total", 0) or 0), ) step["status_label"] = _playbook_run_status_label(step.get("status")) step["completion_percent"] = round( (int(step.get("terminal_jobs_total", 0) or 0) / max(int(step.get("jobs_total", 0) or 0), 1)) * 100, 1, ) step["target_node_codes"] = sorted(str(code or "").strip() for code in step.get("_node_set") or [] if str(code or "").strip()) step["summary"] = _build_playbook_step_summary_text(step) step["summary_text"] = str(step.get("summary") or "") step["focus_ref"] = _build_playbook_run_focus_ref( run, step_key=str(step.get("step_key") or ""), step_title=str(step.get("title") or step.get("step_key") or ""), ) step.pop("_node_set", None) run["steps"] = sorted( list(run.get("steps") or []), key=lambda item: (int(item.get("order") or 0), str(item.get("step_key") or "")), ) problem_steps: list[dict] = [] active_steps: list[dict] = [] for step in list(run.get("steps") or []): step_status_counts = dict(step.get("status_counts") or {}) if str(step.get("status") or "") == "attention": problem_steps.append( { "step_key": str(step.get("step_key") or ""), "title": str(step.get("title") or step.get("step_key") or ""), "status": str(step.get("status") or ""), "reason_label": _terminal_problem_label(step_status_counts), "latest_job": dict(step.get("latest_job") or {}), } ) elif str(step.get("status") or "") in {"running", "queued"}: active_steps.append( { "step_key": str(step.get("step_key") or ""), "title": str(step.get("title") or step.get("step_key") or ""), "status": str(step.get("status") or ""), "reason_label": _active_status_label(step_status_counts), "latest_job": dict(step.get("latest_job") or {}), } ) run["problem_steps"] = problem_steps run["active_steps"] = active_steps if problem_steps: first_problem = problem_steps[0] run["focus_level"] = "danger" run["focus_step_key"] = str(first_problem.get("step_key") or "") run["focus_step_title"] = str(first_problem.get("title") or "") run["focus_summary"] = ( f"{str(first_problem.get('title') or first_problem.get('step_key') or '步骤').strip()} 当前出现" f"{str(first_problem.get('reason_label') or '异常').strip()}，建议先看事件，再决定是否整轮重跑。" ) elif active_steps: first_active = active_steps[0] run["focus_level"] = "warning" run["focus_step_key"] = str(first_active.get("step_key") or "") run["focus_step_title"] = str(first_active.get("title") or "") run["focus_summary"] = ( f"{str(first_active.get('title') or first_active.get('step_key') or '步骤').strip()} 当前处于" f"{str(first_active.get('reason_label') or '处理中').strip()}，建议先看事件和节点现场输出。" ) elif str(run.get("status") or "") == "success": run["focus_level"] = "success" run["focus_step_key"] = "" run["focus_step_title"] = "" run["focus_summary"] = "当前整轮编排已全部收口。" else: run["focus_level"] = "info" run["focus_step_key"] = "" run["focus_step_title"] = "" run["focus_summary"] = "当前编排已创建，等待节点继续执行。" if str(run.get("playbook_key") or "").strip() == "onboarding.bootstrap": _enrich_onboarding_bootstrap_run(run, managed_node_map) run["status_label"] = _playbook_run_status_label(run.get("status")) run["steps_total"] = len(list(run.get("steps") or [])) run["steps_success"] = sum( 1 for step in list(run.get("steps") or []) if str(step.get("status") or "").strip() == "success" ) run["steps_running"] = sum( 1 for step in list(run.get("steps") or []) if str(step.get("status") or "").strip() in {"running", "queued"} ) run["steps_problem"] = sum( 1 for step in list(run.get("steps") or []) if str(step.get("status") or "").strip() == "attention" ) run["steps_terminal"] = sum( 1 for step in list(run.get("steps") or []) if str(step.get("status") or "").strip() in {"success", "attention"} ) run["summary"] = _build_playbook_run_summary_text(run) run["summary_text"] = str(run.get("summary") or "") run["focus_ref"] = _build_playbook_run_focus_ref(run) run.pop("_target_node_set", None) run.pop("_step_map", None) runs.sort( key=lambda item: ( str(item.get("updated_at") or ""), str(item.get("created_at") or ""), str(item.get("run_code") or ""), ), reverse=True, ) return {"runs": runs[:safe_limit]} def _playbook_catalog() -> list[dict]: return [ { "key": "onboarding.bootstrap", "group_key": "onboarding", "group_title": "接管验收", "title": "生成节点接入工单", "description": "在控制面为目标节点生成一轮可追踪的 Node Agent 接入工单，产出 env、脚本和一键落地命令。", "target_scope": "node-set", "default_execution_mode": "control-plane", "execution_modes": ["control-plane"], "default_auto_approve": True, "stop_on_failure": True, "steps": [ { "step_key": "bootstrap", "title": "生成 Node Agent 接入工单", "template_key": "node.bootstrap", "payload": {}, } ], }, { "key": "onboarding.acceptance", "group_key": "onboarding", "group_title": "接管验收", "title": "接管后验收", "description": "按 健康快照 -> Node Agent 状态/日志 -> Worker 状态/日志 的顺序，对新接管节点做一轮标准接管验收。", "target_scope": "node-set", "default_execution_mode": "remote-agent", "execution_modes": ["remote-agent", "ssh"], "default_auto_approve": True, "stop_on_failure": True, "steps": [ { "step_key": "health", "title": "采集健康快照", "template_key": "health.snapshot", "payload": {}, }, { "step_key": "node_agent_status", "title": "采集 Node Agent 状态", "template_key": "service.status", "payload": { "service_name": _NODE_AGENT_SERVICE_NAME, }, }, { "step_key": "node_agent_logs", "title": "收集 Node Agent 日志", "template_key": "logs.collect", "payload": { "service_name": _NODE_AGENT_SERVICE_NAME, "lines": 120, }, }, { "step_key": "worker_status", "title": "采集 Worker 状态", "template_key": "service.status", "payload": { "service_name": settings.worker_service_name, }, }, { "step_key": "worker_logs", "title": "收集 Worker 日志", "template_key": "logs.collect", "payload": { "service_name": settings.worker_service_name, "lines": 120, }, }, ], }, { "key": "inspection.standard", "group_key": "diagnostics", "group_title": "标准巡检", "title": "标准巡检", "description": "按 健康快照 -> Worker 日志 -> 诊断包 的顺序，对目标节点做一次标准联调巡检。", "target_scope": "node-set", "default_execution_mode": "remote-agent", "execution_modes": ["remote-agent", "ssh"], "default_auto_approve": True, "stop_on_failure": True, "steps": [ { "step_key": "health", "title": "采集健康快照", "template_key": "health.snapshot", "payload": {}, }, { "step_key": "worker_logs", "title": "收集 Worker 日志", "template_key": "logs.collect", "payload": { "service_name": settings.worker_service_name, "lines": 120, }, }, { "step_key": "diagnostics", "title": "收集诊断包", "template_key": "diagnostics.collect", "payload": { "lines": 200, }, }, ], }, { "key": "scene.logs.key", "group_key": "scene", "group_title": "执行现场", "title": "关键现场日志", "description": "只回收 Worker 的关键日志样本，适合先看现场而不放大量级。", "target_scope": "node-set", "default_execution_mode": "remote-agent", "execution_modes": ["remote-agent", "ssh"], "default_auto_approve": True, "stop_on_failure": True, "steps": [ { "step_key": "worker_logs", "title": "收集 Worker 关键日志", "template_key": "logs.collect", "payload": { "service_name": settings.worker_service_name, "lines": 120, }, } ], }, { "key": "scene.logs.full", "group_key": "scene", "group_title": "执行现场", "title": "全量现场取证", "description": "回收较大日志样本并补一份诊断包，适合深入排查远端执行面。", "target_scope": "node-set", "default_execution_mode": "remote-agent", "execution_modes": ["remote-agent", "ssh"], "default_auto_approve": True, "stop_on_failure": True, "steps": [ { "step_key": "worker_logs", "title": "收集 Worker 全量日志", "template_key": "logs.collect", "payload": { "service_name": settings.worker_service_name, "lines": 300, }, }, { "step_key": "diagnostics", "title": "收集诊断包", "template_key": "diagnostics.collect", "payload": { "lines": 300, }, }, ], }, { "key": "scene.diagnostics", "group_key": "scene", "group_title": "执行现场", "title": "现场诊断包", "description": "直接回收诊断包，适合对参与节点做一次较轻量的远端取证。", "target_scope": "node-set", "default_execution_mode": "remote-agent", "execution_modes": ["remote-agent", "ssh"], "default_auto_approve": True, "stop_on_failure": True, "steps": [ { "step_key": "diagnostics", "title": "收集诊断包", "template_key": "diagnostics.collect", "payload": { "lines": 200, }, } ], }, ] def get_ops_playbooks() -> dict: playbooks = [decorate_execution_mode_fields(item) for item in deepcopy(_playbook_catalog())] return { "items": playbooks, "summary": { "total": len(playbooks), "groups": sorted({str(item.get("group_key") or "").strip() for item in playbooks if str(item.get("group_key") or "").strip()}), }, } def get_ops_playbook(playbook_key: str) -> dict: normalized_key = str(playbook_key or "").strip() if not normalized_key: return {} for item in _playbook_catalog(): if str(item.get("key") or "").strip() == normalized_key: return decorate_execution_mode_fields(deepcopy(item)) return {} def preview_ops_playbook(payload: dict | None = None) -> tuple[bool, str, dict]: source_payload = dict(payload or {}) playbook_key = str(source_payload.get("playbook_key") or "").strip() if not playbook_key: return False, "playbook_key 不能为空", {} playbook = get_ops_playbook(playbook_key) if not playbook: return False, "playbook 不存在", {} node_codes = _normalize_node_codes(source_payload.get("node_codes") or []) if not node_codes: return False, "该 playbook 需要 node_codes", {} requested_by = str(source_payload.get("requested_by") or "api").strip() or "api" supported_execution_modes = [ str(item or "").strip() for item in list(playbook.get("execution_modes") or []) if str(item or "").strip() ] if not supported_execution_modes: supported_execution_modes = [str(playbook.get("default_execution_mode") or "remote-agent").strip() or "remote-agent"] execution_mode = str(source_payload.get("execution_mode") or playbook.get("default_execution_mode") or "remote-agent").strip() or "remote-agent" if execution_mode not in supported_execution_modes: return False, f"该 playbook 仅支持 {' / '.join(supported_execution_modes)}", { "playbook": { "key": playbook_key, "title": str(playbook.get("title") or "").strip(), "execution_modes": supported_execution_modes, "default_execution_mode": str(playbook.get("default_execution_mode") or "").strip(), "default_execution_mode_label": str(playbook.get("default_execution_mode_label") or "").strip(), "execution_mode_options": build_execution_mode_options( supported_execution_modes, str(playbook.get("default_execution_mode") or "").strip() or "remote-agent", ), } } auto_approve = bool(source_payload.get("auto_approve", playbook.get("default_auto_approve", True))) steps_preview: list[dict] = [] template_keys: list[str] = [] for index, step in enumerate(list(playbook.get("steps") or []), start=1): template_key = str(step.get("template_key") or "").strip() template = get_ops_action_template(template_key) if not template: return False, f"playbook step 缺少动作模板: {template_key}", {} normalize_ok, normalize_message, normalized_payload = build_ops_template_payload(template_key, step.get("payload") or {}) if not normalize_ok: return False, normalize_message, {} template_keys.append(template_key) steps_preview.append( { "order": index, "step_key": str(step.get("step_key") or f"step-{index}").strip() or f"step-{index}", "title": str(step.get("title") or template.get("title") or template_key).strip() or template_key, "template_key": template_key, "action": str(template.get("action") or "").strip(), "payload": normalized_payload, "execution_mode": execution_mode, "execution_mode_label": execution_mode_label(execution_mode), "auto_approve": auto_approve, "target_nodes_total": len(node_codes), "expected_jobs": len(node_codes), } ) preview = { "playbook": { "key": playbook_key, "title": str(playbook.get("title") or "").strip(), "group_key": str(playbook.get("group_key") or "").strip(), "group_title": str(playbook.get("group_title") or "").strip(), "description": str(playbook.get("description") or "").strip(), "stop_on_failure": bool(playbook.get("stop_on_failure", True)), "default_execution_mode": str(playbook.get("default_execution_mode") or "").strip(), "default_execution_mode_label": str(playbook.get("default_execution_mode_label") or "").strip(), "execution_modes": supported_execution_modes, "execution_mode_options": build_execution_mode_options( supported_execution_modes, str(playbook.get("default_execution_mode") or "").strip() or "remote-agent", ), }, "requested_by": requested_by, "execution_mode": execution_mode, "execution_mode_label": execution_mode_label(execution_mode), "auto_approve": auto_approve, "target_nodes_total": len(node_codes), "target_node_codes": node_codes, "step_count": len(steps_preview), "expected_jobs_total": len(node_codes) * len(steps_preview), "template_keys": template_keys, "steps": steps_preview, } return True, "playbook 预览生成成功", preview def get_recent_ops_playbook_runs( *, limit: int = 12, scan_limit: int = 360, status: str = "", group_key: str = "", query: str = "", ) -> dict: safe_limit = min(max(int(limit or 12), 1), 50) safe_scan_limit = min(max(int(scan_limit or max(safe_limit * 30, 120)), safe_limit), 1000) normalized_status = str(status or "").strip() normalized_group_key = str(group_key or "").strip() normalized_query = str(query or "").strip().lower() jobs = list_ops_jobs(limit=safe_scan_limit, compact=False) grouped = _group_playbook_runs_from_jobs(jobs, limit=safe_scan_limit) all_runs = list(grouped.get("runs") or []) available_status_counts: dict[str, int] = {} available_group_counts: dict[str, int] = {} for run in all_runs: run_status = str(run.get("status") or "").strip() or "queued" available_status_counts[run_status] = int(available_status_counts.get(run_status, 0) or 0) + 1 run_group_key = str(run.get("group_key") or "").strip() if run_group_key: available_group_counts[run_group_key] = int(available_group_counts.get(run_group_key, 0) or 0) + 1 filtered_runs: list[dict] = [] for run in all_runs: run_status = str(run.get("status") or "").strip() run_group_key = str(run.get("group_key") or "").strip() haystack = " ".join( [ str(run.get("run_code") or ""), str(run.get("playbook_key") or ""), str(run.get("playbook_title") or ""), str(run.get("requested_by") or ""), " ".join(list(run.get("target_node_codes") or [])), ] ).lower() if normalized_status and run_status != normalized_status: continue if normalized_group_key and run_group_key != normalized_group_key: continue if normalized_query and normalized_query not in haystack: continue filtered_runs.append(run) runs = filtered_runs[:safe_limit] summary = { "total": len(runs), "filtered_total": len(filtered_runs), "unfiltered_total": len(all_runs), "success": sum(1 for item in runs if str(item.get("status") or "") == "success"), "running": sum(1 for item in runs if str(item.get("status") or "") == "running"), "attention": sum(1 for item in runs if str(item.get("status") or "") == "attention"), "queued": sum(1 for item in runs if str(item.get("status") or "") == "queued"), "scanned_jobs": len(jobs), "available_status_counts": available_status_counts, "available_group_counts": available_group_counts, "filters": { "status": normalized_status, "group_key": normalized_group_key, "query": normalized_query, }, } return { "runs": runs, "summary": summary, } def get_ops_playbook_run(run_code: str, *, scan_limit: int = 1000) -> dict: normalized_run_code = str(run_code or "").strip() if not normalized_run_code: return {} matched_jobs = _list_playbook_run_jobs(normalized_run_code, scan_limit=scan_limit) grouped = _group_playbook_runs_from_jobs(matched_jobs, limit=1, run_code_filter=normalized_run_code) return next(iter(grouped.get("runs") or []), {}) def list_ops_playbook_run_events( run_code: str, *, limit: int = 80, step_key: str = "", node_code: str = "", scan_limit: int = 1000, ) -> dict: normalized_run_code = str(run_code or "").strip() normalized_step_key = str(step_key or "").strip() normalized_node_code = str(node_code or "").strip() safe_limit = min(max(int(limit or 80), 1), 300) matched_jobs = _list_playbook_run_jobs(normalized_run_code, scan_limit=scan_limit) if not matched_jobs: return { "run_code": normalized_run_code, "events": [], "summary": { "total": 0, "returned_total": 0, "available_step_counts": {}, "available_node_counts": {}, "level_counts": {}, "event_type_counts": {}, "job_status_counts": {}, "latest_at": "", "filters": { "step_key": normalized_step_key, "node_code": normalized_node_code, "limit": safe_limit, }, }, } playbook_run = _group_playbook_runs_from_jobs(matched_jobs, limit=1, run_code_filter=normalized_run_code) run_payload = next(iter(playbook_run.get("runs") or []), {}) job_map = {int(job.get("id") or 0): dict(job or {}) for job in matched_jobs if int(job.get("id") or 0) > 0} raw_events = list_ops_job_events_for_jobs(list(job_map.keys()), limit=min(max(safe_limit * 4, 120), 800)) available_step_counts: dict[str, int] = {} available_node_counts: dict[str, int] = {} level_counts: dict[str, int] = {} event_type_counts: dict[str, int] = {} job_status_counts: dict[str, int] = {} events: list[dict] = [] latest_at = "" for event in raw_events: job = job_map.get(int(event.get("job_id") or 0), {}) playbook_meta = dict((job.get("metadata") or {}).get("playbook") or {}) event_step_key = str(playbook_meta.get("step_key") or "").strip() event_step_title = str(playbook_meta.get("step_title") or event_step_key or "").strip() event_node_code = str(event.get("node_code") or job.get("target_node_code") or "").strip() if event_step_key: available_step_counts[event_step_key] = int(available_step_counts.get(event_step_key, 0) or 0) + 1 if event_node_code: available_node_counts[event_node_code] = int(available_node_counts.get(event_node_code, 0) or 0) + 1 event_level = str(event.get("level") or "").strip() or "info" level_counts[event_level] = int(level_counts.get(event_level, 0) or 0) + 1 event_type = str(event.get("event_type") or "").strip() or "event" event_type_counts[event_type] = int(event_type_counts.get(event_type, 0) or 0) + 1 job_status = str(job.get("status") or "").strip() or "queued" job_status_counts[job_status] = int(job_status_counts.get(job_status, 0) or 0) + 1 event_created_at = str(event.get("created_at") or "").strip() if event_created_at and event_created_at > latest_at: latest_at = event_created_at if normalized_step_key and event_step_key != normalized_step_key: continue if normalized_node_code and event_node_code != normalized_node_code: continue event_summary_text = str(event.get("summary_text") or event.get("summary") or event.get("message") or "").strip() event_source_focus_ref = _normalize_focus_ref(event.get("focus_ref")) events.append( { **event, "run_code": normalized_run_code, "job_code": str(job.get("job_code") or ""), "job_status": job_status, "job_status_label": _playbook_run_status_label(job_status), "action": str(job.get("action") or ""), "target_node_code": str(job.get("target_node_code") or ""), "step_key": event_step_key, "step_title": event_step_title, "summary_text": event_summary_text, "source_focus_ref": event_source_focus_ref, "focus_ref": _build_playbook_run_focus_ref( run_payload, step_key=event_step_key, step_title=event_step_title, event_key=str(event.get("event_key") or "").strip(), node_code=event_node_code, ), } ) return { "run_code": normalized_run_code, "playbook_run": run_payload, "events": events[:safe_limit], "summary": { "total": len(events), "returned_total": len(events), "available_step_counts": available_step_counts, "available_node_counts": available_node_counts, "level_counts": level_counts, "event_type_counts": event_type_counts, "job_status_counts": job_status_counts, "latest_at": latest_at, "filters": { "step_key": normalized_step_key, "node_code": normalized_node_code, "limit": safe_limit, }, }, } def rerun_ops_playbook_run(run_code: str, *, requested_by: str = "api") -> tuple[bool, str, dict]: playbook_run = get_ops_playbook_run(run_code) if not playbook_run: return False, "playbook run 不存在", {} playbook_key = str(playbook_run.get("playbook_key") or "").strip() node_codes = _normalize_node_codes(playbook_run.get("target_node_codes") or []) if not playbook_key or not node_codes: return False, "当前 playbook run 缺少重跑所需上下文", {"playbook_run": playbook_run} return execute_ops_playbook( { "playbook_key": playbook_key, "node_codes": node_codes, "execution_mode": str(playbook_run.get("execution_mode") or "remote-agent").strip() or "remote-agent", "auto_approve": True, "requested_by": f"{str(requested_by or 'api').strip() or 'api'}/rerun/{playbook_key}", } ) def cancel_ops_playbook_run(run_code: str, *, cancelled_by: str = "api", reason: str = "") -> tuple[bool, str, dict]: playbook_run = get_ops_playbook_run(run_code) if not playbook_run: return False, "playbook run 不存在", {} normalized_run_code = str(run_code or "").strip() matched_jobs = _list_playbook_run_jobs(normalized_run_code, scan_limit=2000) if not matched_jobs: return False, "当前 playbook run 没有关联任务", {"playbook_run": playbook_run} cancellable_jobs = [ job for job in matched_jobs if str(job.get("status") or "").strip() not in {"success", "failed", "cancelled"} ] if not cancellable_jobs: return False, "当前 playbook run 已全部收口，无需取消", { "playbook_run": playbook_run, "jobs_total": len(matched_jobs), "cancelled_count": 0, } cancelled_count = 0 failed_count = 0 results: list[dict] = [] for job in cancellable_jobs: job_id = int(job.get("id") or 0) ok, message, data = cancel_ops_job( job_id, cancelled_by=str(cancelled_by or "api").strip() or "api", reason=reason, ) if ok: cancelled_count += 1 else: failed_count += 1 results.append( { "job_id": job_id, "job_code": str(job.get("job_code") or ""), "ok": ok, "message": message, "status": str(((data or {}).get("job") or {}).get("status") or job.get("status") or ""), } ) latest_run = get_ops_playbook_run(run_code) return failed_count <= 0, ( f"已取消 {cancelled_count} 条编排子任务" if failed_count <= 0 else f"编排取消部分完成，成功 {cancelled_count}，失败 {failed_count}" ), { "playbook_run": latest_run or playbook_run, "jobs_total": len(matched_jobs), "cancellable_jobs_total": len(cancellable_jobs), "cancelled_count": cancelled_count, "failed_count": failed_count, "results": results, } def execute_ops_playbook(payload: dict | None = None) -> tuple[bool, str, dict]: preview_ok, preview_message, preview = preview_ops_playbook(payload) if not preview_ok: return False, preview_message, {} playbook = dict(preview.get("playbook") or {}) requested_by = str(preview.get("requested_by") or "api").strip() or "api" node_codes = _normalize_node_codes(preview.get("target_node_codes") or []) execution_mode = str(preview.get("execution_mode") or "remote-agent").strip() or "remote-agent" auto_approve = bool(preview.get("auto_approve", True)) stop_on_failure = bool(playbook.get("stop_on_failure", True)) run_code = _build_playbook_run_code() results: list[dict] = [] successful_steps = 0 failed_steps = 0 for step in list(preview.get("steps") or []): template_key = str(step.get("template_key") or "").strip() step_key = str(step.get("step_key") or template_key).strip() or template_key ok, message, data = create_ops_job_batch( { "template_key": template_key, "target_node_codes": node_codes, "execution_mode": execution_mode, "auto_approve": auto_approve, "requested_by": f"{requested_by}/playbook/{playbook.get('key') or ''}/{step_key}", "payload": dict(step.get("payload") or {}), "metadata": _build_playbook_job_metadata( playbook=playbook, preview=preview, step=step, run_code=run_code, requested_by=requested_by, execution_mode=execution_mode, ), } ) results.append( { "run_code": run_code, "step_key": step_key, "template_key": template_key, "title": str(step.get("title") or "").strip(), "ok": ok, "message": message, "data": data or {}, } ) if ok: successful_steps += 1 continue failed_steps += 1 if stop_on_failure: playbook_runs = get_recent_ops_playbook_runs(limit=20, scan_limit=max(len(node_codes) * max(len(results), 1) * 4, 120)) playbook_run = next( (item for item in list(playbook_runs.get("runs") or []) if str(item.get("run_code") or "").strip() == run_code), {}, ) return False, message, { "playbook": playbook, "mode": "ops-playbook", "run_code": run_code, "requested_by": requested_by, "execution_mode": execution_mode, "target_node_codes": node_codes, "successful_steps": successful_steps, "failed_steps": failed_steps, "results": results, "preview": preview, "playbook_run": playbook_run, } playbook_runs = get_recent_ops_playbook_runs(limit=20, scan_limit=max(len(node_codes) * max(len(results), 1) * 4, 120)) playbook_run = next( (item for item in list(playbook_runs.get("runs") or []) if str(item.get("run_code") or "").strip() == run_code), {}, ) return failed_steps <= 0, ( f"playbook {playbook.get('title') or playbook.get('key') or ''} 已创建 {successful_steps} 个步骤，回执 {run_code}" if failed_steps <= 0 else f"playbook 已部分创建，成功 {successful_steps}，失败 {failed_steps}，回执 {run_code}" ), { "playbook": playbook, "mode": "ops-playbook", "run_code": run_code, "requested_by": requested_by, "execution_mode": execution_mode, "target_node_codes": node_codes, "successful_steps": successful_steps, "failed_steps": failed_steps, "results": results, "preview": preview, "playbook_run": playbook_run, }