1264 lines
54 KiB
Python
1264 lines
54 KiB
Python
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,
|
||
}
|