1891 lines
72 KiB
Python
1891 lines
72 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import threading
|
||
from datetime import datetime
|
||
from uuid import uuid4
|
||
|
||
from psycopg2 import errors
|
||
|
||
from app.core.config import settings
|
||
from app.core.db import get_db
|
||
from app.services.ops_execution_capability_service import (
|
||
supports_control_plane_action,
|
||
supports_local_runtime_action,
|
||
)
|
||
from app.services.ops_execution_mode_service import execution_mode_label
|
||
from app.services.ops_runtime_executor_service import (
|
||
execute_local_support_action,
|
||
execute_ssh_action,
|
||
supports_ssh_execution,
|
||
)
|
||
from app.services.runtime_control_service import runtime_action
|
||
from app.services.runtime_status_service import get_runtime_status
|
||
|
||
|
||
_OPS_SCHEMA_SQL = """
|
||
CREATE TABLE IF NOT EXISTS ops_managed_nodes (
|
||
node_code VARCHAR(64) PRIMARY KEY,
|
||
region VARCHAR(32) NOT NULL DEFAULT 'unknown',
|
||
role VARCHAR(32) NOT NULL DEFAULT 'worker',
|
||
title VARCHAR(128) NOT NULL DEFAULT '',
|
||
ssh_host VARCHAR(255) NOT NULL DEFAULT '',
|
||
ssh_port INTEGER NOT NULL DEFAULT 22,
|
||
ssh_user VARCHAR(64) NOT NULL DEFAULT '',
|
||
auth_mode VARCHAR(32) NOT NULL DEFAULT 'key',
|
||
deploy_channel VARCHAR(64) NOT NULL DEFAULT 'stable',
|
||
is_enabled BOOLEAN NOT NULL DEFAULT TRUE,
|
||
metadata_json JSONB,
|
||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
last_seen_at TIMESTAMP,
|
||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS ops_managed_node_secrets (
|
||
node_code VARCHAR(64) PRIMARY KEY REFERENCES ops_managed_nodes(node_code) ON DELETE CASCADE,
|
||
ssh_password TEXT NOT NULL DEFAULT '',
|
||
ssh_private_key TEXT NOT NULL DEFAULT '',
|
||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS ops_jobs (
|
||
id BIGSERIAL PRIMARY KEY,
|
||
job_code VARCHAR(64) NOT NULL UNIQUE,
|
||
action VARCHAR(128) NOT NULL DEFAULT '',
|
||
target_type VARCHAR(32) NOT NULL DEFAULT 'node',
|
||
target_node_code VARCHAR(64) NOT NULL DEFAULT '',
|
||
status VARCHAR(32) NOT NULL DEFAULT 'queued',
|
||
execution_mode VARCHAR(32) NOT NULL DEFAULT 'remote-agent',
|
||
requested_by VARCHAR(64) NOT NULL DEFAULT 'api',
|
||
payload_json JSONB,
|
||
metadata_json JSONB,
|
||
result_json JSONB,
|
||
error_message TEXT NOT NULL DEFAULT '',
|
||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
started_at TIMESTAMP,
|
||
finished_at TIMESTAMP,
|
||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_ops_jobs_status_created
|
||
ON ops_jobs(status, created_at DESC);
|
||
|
||
CREATE TABLE IF NOT EXISTS ops_job_steps (
|
||
id BIGSERIAL PRIMARY KEY,
|
||
job_id BIGINT NOT NULL REFERENCES ops_jobs(id) ON DELETE CASCADE,
|
||
step_key VARCHAR(64) NOT NULL DEFAULT '',
|
||
title VARCHAR(128) NOT NULL DEFAULT '',
|
||
node_code VARCHAR(64) NOT NULL DEFAULT '',
|
||
status VARCHAR(32) NOT NULL DEFAULT 'queued',
|
||
stdout_text TEXT NOT NULL DEFAULT '',
|
||
stderr_text TEXT NOT NULL DEFAULT '',
|
||
result_json JSONB,
|
||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
started_at TIMESTAMP,
|
||
finished_at TIMESTAMP,
|
||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_ops_job_steps_job_created
|
||
ON ops_job_steps(job_id, created_at ASC);
|
||
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS risk_level VARCHAR(16) NOT NULL DEFAULT 'medium';
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS approval_required BOOLEAN NOT NULL DEFAULT FALSE;
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS approval_status VARCHAR(32) NOT NULL DEFAULT 'not_required';
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS approved_by VARCHAR(64) NOT NULL DEFAULT '';
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS approved_at TIMESTAMP;
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS blocked_reason TEXT NOT NULL DEFAULT '';
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS cancellation_reason TEXT NOT NULL DEFAULT '';
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS dispatched_at TIMESTAMP;
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS target_selector_json JSONB;
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS policy_json JSONB;
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS rollout_id BIGINT;
|
||
ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS metadata_json JSONB;
|
||
"""
|
||
|
||
_LOCAL_RUNTIME_ACTIONS = {
|
||
"runtime.start_worker": "start_worker",
|
||
"runtime.stop_worker": "stop_worker",
|
||
"runtime.start_detection": "start_detection",
|
||
"runtime.stop_detection": "stop_detection",
|
||
"runtime.restart_api": "restart_api",
|
||
"runtime.start_sync_agent": "start_sync_agent",
|
||
"runtime.stop_sync_agent": "stop_sync_agent",
|
||
"runtime.push_sync": "push_sync",
|
||
"runtime.pull_tasks": "pull_tasks",
|
||
}
|
||
|
||
_OPS_SCHEMA_LOCK = threading.Lock()
|
||
_OPS_SCHEMA_READY = False
|
||
_OPS_SCHEMA_ADVISORY_LOCK_KEY = 90421801
|
||
_OPS_REQUIRED_TABLES = (
|
||
"ops_managed_nodes",
|
||
"ops_managed_node_secrets",
|
||
"ops_jobs",
|
||
"ops_job_steps",
|
||
)
|
||
_OPS_REQUIRED_COLUMNS = {
|
||
"ops_jobs": (
|
||
"risk_level",
|
||
"approval_required",
|
||
"approval_status",
|
||
"approved_by",
|
||
"approved_at",
|
||
"blocked_reason",
|
||
"cancellation_reason",
|
||
"dispatched_at",
|
||
"target_selector_json",
|
||
"policy_json",
|
||
"rollout_id",
|
||
),
|
||
"ops_job_steps": ("stdout_text", "stderr_text", "result_json"),
|
||
}
|
||
|
||
|
||
def ensure_ops_schema() -> None:
|
||
global _OPS_SCHEMA_READY
|
||
if _OPS_SCHEMA_READY:
|
||
return
|
||
with _OPS_SCHEMA_LOCK:
|
||
if _OPS_SCHEMA_READY:
|
||
return
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
if _ops_schema_basics_present(cur):
|
||
_OPS_SCHEMA_READY = True
|
||
return
|
||
conn.autocommit = False
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT pg_advisory_xact_lock(%s)", (_OPS_SCHEMA_ADVISORY_LOCK_KEY,))
|
||
cur.execute(_OPS_SCHEMA_SQL)
|
||
conn.commit()
|
||
except Exception as exc:
|
||
recoverable = isinstance(exc, (errors.DeadlockDetected, errors.LockNotAvailable))
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
if not recoverable:
|
||
raise
|
||
with conn.cursor() as cur:
|
||
if not _ops_schema_basics_present(cur):
|
||
raise
|
||
_OPS_SCHEMA_READY = True
|
||
|
||
|
||
def _ops_schema_basics_present(cur) -> bool:
|
||
for table_name in _OPS_REQUIRED_TABLES:
|
||
cur.execute("SELECT to_regclass(%s)", (f"public.{table_name}",))
|
||
row = cur.fetchone()
|
||
if not row or not row[0]:
|
||
return False
|
||
|
||
for table_name, required_columns in _OPS_REQUIRED_COLUMNS.items():
|
||
cur.execute(
|
||
"""
|
||
SELECT column_name
|
||
FROM information_schema.columns
|
||
WHERE table_schema = 'public' AND table_name = %s
|
||
""",
|
||
(table_name,),
|
||
)
|
||
existing_columns = {str(row[0] or "").strip() for row in list(cur.fetchall() or [])}
|
||
if not set(required_columns).issubset(existing_columns):
|
||
return False
|
||
return True
|
||
|
||
|
||
def _decode_json(value: object) -> dict:
|
||
if isinstance(value, dict):
|
||
return value
|
||
if value in (None, ""):
|
||
return {}
|
||
try:
|
||
return json.loads(value)
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _ts(value: object) -> str:
|
||
if isinstance(value, datetime):
|
||
return value.isoformat(sep=" ", timespec="seconds")
|
||
return ""
|
||
|
||
|
||
def _serialize_step_row(row: tuple) -> dict:
|
||
return {
|
||
"id": int(row[0]),
|
||
"step_key": str(row[1] or ""),
|
||
"title": str(row[2] or ""),
|
||
"node_code": str(row[3] or ""),
|
||
"status": str(row[4] or ""),
|
||
"stdout": str(row[5] or ""),
|
||
"stderr": str(row[6] or ""),
|
||
"result": _decode_json(row[7]),
|
||
"created_at": _ts(row[8]),
|
||
"started_at": _ts(row[9]),
|
||
"finished_at": _ts(row[10]),
|
||
"updated_at": _ts(row[11]),
|
||
}
|
||
|
||
|
||
def _job_status_label(status: object) -> str:
|
||
normalized_status = str(status or "").strip()
|
||
mapping = {
|
||
"queued": "排队中",
|
||
"awaiting_approval": "待审批",
|
||
"blocked": "阻断",
|
||
"dispatching": "派发中",
|
||
"running": "执行中",
|
||
"success": "成功",
|
||
"failed": "失败",
|
||
"cancelled": "已取消",
|
||
"partially_succeeded": "部分成功",
|
||
"completed_with_issues": "带问题完成",
|
||
}
|
||
return mapping.get(normalized_status, normalized_status or "未知")
|
||
|
||
|
||
def _job_approval_status_label(approval_status: object, *, approval_required: bool = False) -> str:
|
||
normalized_status = str(approval_status or "").strip()
|
||
if not normalized_status or normalized_status == "not_required":
|
||
return "待审批" if bool(approval_required) else "无需审批"
|
||
mapping = {
|
||
"approved": "已审批",
|
||
"blocked": "已阻断",
|
||
"cancelled": "已取消",
|
||
}
|
||
return mapping.get(normalized_status, normalized_status)
|
||
|
||
|
||
def _build_job_summary_text(job: dict) -> str:
|
||
normalized_job = dict(job or {})
|
||
status = str(normalized_job.get("status") or "").strip()
|
||
target_node_code = str(normalized_job.get("target_node_code") or "").strip()
|
||
execution_mode_label_text = str(normalized_job.get("execution_mode_label") or "").strip()
|
||
requested_by = str(normalized_job.get("requested_by") or "").strip()
|
||
error_message = str(normalized_job.get("error_message") or "").strip()
|
||
blocked_reason = str(normalized_job.get("blocked_reason") or "").strip()
|
||
cancellation_reason = str(normalized_job.get("cancellation_reason") or "").strip()
|
||
|
||
if cancellation_reason and status == "cancelled":
|
||
return cancellation_reason
|
||
if blocked_reason and status == "blocked":
|
||
return blocked_reason
|
||
if error_message and status in {"failed", "blocked", "cancelled", "completed_with_issues"}:
|
||
return error_message
|
||
if status == "awaiting_approval":
|
||
return "任务已创建，等待审批后继续派发。"
|
||
if status == "running":
|
||
if target_node_code:
|
||
return f"目标节点 {target_node_code} 正在执行该任务。"
|
||
return "任务正在执行中。"
|
||
if status == "success":
|
||
if target_node_code:
|
||
return f"目标节点 {target_node_code} 已完成该任务。"
|
||
return "任务已执行完成。"
|
||
|
||
parts: list[str] = []
|
||
if target_node_code:
|
||
parts.append(f"目标节点 {target_node_code}")
|
||
if requested_by:
|
||
parts.append(f"发起 {requested_by}")
|
||
if execution_mode_label_text:
|
||
parts.append(f"方式 {execution_mode_label_text}")
|
||
if parts:
|
||
return " / ".join(parts)
|
||
return f"当前状态 {_job_status_label(status)}。"
|
||
|
||
|
||
def _build_job_step_counters(steps: list[dict] | None = None) -> dict:
|
||
normalized_steps = list(steps or [])
|
||
total = len(normalized_steps)
|
||
running = sum(1 for item in normalized_steps if str(item.get("status") or "").strip() in {"running", "dispatching"})
|
||
success = sum(1 for item in normalized_steps if str(item.get("status") or "").strip() == "success")
|
||
failed = sum(1 for item in normalized_steps if str(item.get("status") or "").strip() in {"failed", "blocked", "cancelled"})
|
||
terminal = sum(
|
||
1
|
||
for item in normalized_steps
|
||
if str(item.get("status") or "").strip() in {"success", "failed", "blocked", "cancelled", "partially_succeeded"}
|
||
)
|
||
return {
|
||
"steps_total": total,
|
||
"steps_running": running,
|
||
"steps_success": success,
|
||
"steps_failed": failed,
|
||
"steps_terminal": terminal,
|
||
}
|
||
|
||
|
||
def _serialize_job_row(row: tuple, *, steps: list[dict] | None = None) -> dict:
|
||
normalized_execution_mode = str(row[6] or "")
|
||
normalized_steps = list(steps or [])
|
||
approval_required = bool(row[17])
|
||
serialized = {
|
||
"id": int(row[0]),
|
||
"job_code": str(row[1] or ""),
|
||
"action": str(row[2] or ""),
|
||
"target_type": str(row[3] or ""),
|
||
"target_node_code": str(row[4] or ""),
|
||
"status": str(row[5] or ""),
|
||
"status_label": _job_status_label(row[5]),
|
||
"execution_mode": normalized_execution_mode,
|
||
"execution_mode_label": execution_mode_label(normalized_execution_mode),
|
||
"requested_by": str(row[7] or ""),
|
||
"payload": _decode_json(row[8]),
|
||
"metadata": _decode_json(row[9]),
|
||
"result": _decode_json(row[10]),
|
||
"error_message": str(row[11] or ""),
|
||
"created_at": _ts(row[12]),
|
||
"started_at": _ts(row[13]),
|
||
"finished_at": _ts(row[14]),
|
||
"updated_at": _ts(row[15]),
|
||
"risk_level": str(row[16] or ""),
|
||
"approval_required": approval_required,
|
||
"approval_status": str(row[18] or ""),
|
||
"approval_status_label": _job_approval_status_label(row[18], approval_required=approval_required),
|
||
"approved_by": str(row[19] or ""),
|
||
"approved_at": _ts(row[20]),
|
||
"blocked_reason": str(row[21] or ""),
|
||
"cancellation_reason": str(row[22] or ""),
|
||
"dispatched_at": _ts(row[23]),
|
||
"target_selector": _decode_json(row[24]),
|
||
"policy": _decode_json(row[25]),
|
||
"rollout_id": int(row[26]) if row[26] else 0,
|
||
"steps": normalized_steps,
|
||
"steps_loaded": steps is not None,
|
||
"is_compact": False,
|
||
"target_node_codes": [str(row[4] or "")] if str(row[4] or "").strip() else [],
|
||
"focus_ref": {
|
||
"kind": "ops_job",
|
||
"job_id": int(row[0]),
|
||
"job_code": str(row[1] or ""),
|
||
"action": str(row[2] or ""),
|
||
"target_node_code": str(row[4] or ""),
|
||
},
|
||
}
|
||
serialized.update(_build_job_step_counters(normalized_steps))
|
||
serialized["summary"] = _build_job_summary_text(serialized)
|
||
serialized["summary_text"] = str(serialized.get("summary") or "")
|
||
return serialized
|
||
|
||
|
||
def _serialize_node_row(row: tuple) -> dict:
|
||
return {
|
||
"node_code": str(row[0] or ""),
|
||
"region": str(row[1] or ""),
|
||
"role": str(row[2] or ""),
|
||
"title": str(row[3] or ""),
|
||
"ssh_host": str(row[4] or ""),
|
||
"ssh_port": int(row[5] or 22),
|
||
"ssh_user": str(row[6] or ""),
|
||
"auth_mode": str(row[7] or ""),
|
||
"deploy_channel": str(row[8] or ""),
|
||
"is_enabled": bool(row[9]),
|
||
"metadata": _decode_json(row[10]),
|
||
"created_at": _ts(row[11]),
|
||
"last_seen_at": _ts(row[12]),
|
||
"updated_at": _ts(row[13]),
|
||
}
|
||
|
||
|
||
def _load_node_secret_flags(node_codes: list[str]) -> dict[str, dict]:
|
||
normalized_codes = [str(item or "").strip() for item in node_codes if str(item or "").strip()]
|
||
if not normalized_codes:
|
||
return {}
|
||
result: dict[str, dict] = {}
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT node_code, ssh_password, ssh_private_key
|
||
FROM ops_managed_node_secrets
|
||
WHERE node_code = ANY(%s)
|
||
""",
|
||
(normalized_codes,),
|
||
)
|
||
rows = cur.fetchall()
|
||
for row in rows:
|
||
node_code = str(row[0] or "").strip()
|
||
result[node_code] = {
|
||
"ssh_password_configured": bool(str(row[1] or "").strip()),
|
||
"ssh_private_key_configured": bool(str(row[2] or "").strip()),
|
||
}
|
||
return result
|
||
|
||
|
||
def _parse_ssh_entry(raw_value: object) -> dict:
|
||
raw = str(raw_value or "").strip()
|
||
if not raw:
|
||
return {}
|
||
parts = raw.split(maxsplit=2)
|
||
if len(parts) < 2:
|
||
return {}
|
||
host_port = str(parts[0] or "").strip()
|
||
ssh_user = str(parts[1] or "").strip()
|
||
secret = str(parts[2] or "").strip() if len(parts) >= 3 else ""
|
||
ssh_host = host_port
|
||
ssh_port = 22
|
||
if ":" in host_port:
|
||
host_candidate, port_candidate = host_port.rsplit(":", 1)
|
||
if host_candidate and port_candidate.isdigit():
|
||
ssh_host = host_candidate
|
||
ssh_port = max(int(port_candidate), 1)
|
||
if secret.startswith("<") and secret.endswith(">") and len(secret) >= 2:
|
||
secret = secret[1:-1].strip()
|
||
payload = {
|
||
"ssh_host": ssh_host,
|
||
"ssh_port": ssh_port,
|
||
"ssh_user": ssh_user,
|
||
}
|
||
if secret:
|
||
payload["auth_mode"] = "password"
|
||
payload["ssh_password"] = secret
|
||
return payload
|
||
|
||
|
||
def _upsert_managed_node_secret(
|
||
*,
|
||
node_code: str,
|
||
ssh_password: str = "",
|
||
ssh_private_key: str = "",
|
||
clear_ssh_password: bool = False,
|
||
clear_ssh_private_key: bool = False,
|
||
) -> None:
|
||
normalized_node_code = str(node_code or "").strip()
|
||
if not normalized_node_code:
|
||
return
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO ops_managed_node_secrets (
|
||
node_code, ssh_password, ssh_private_key, updated_at
|
||
) VALUES (%s, %s, %s, CURRENT_TIMESTAMP)
|
||
ON CONFLICT (node_code) DO UPDATE SET
|
||
ssh_password = CASE
|
||
WHEN %s THEN ''
|
||
WHEN %s <> '' THEN %s
|
||
ELSE ops_managed_node_secrets.ssh_password
|
||
END,
|
||
ssh_private_key = CASE
|
||
WHEN %s THEN ''
|
||
WHEN %s <> '' THEN %s
|
||
ELSE ops_managed_node_secrets.ssh_private_key
|
||
END,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
""",
|
||
(
|
||
normalized_node_code,
|
||
"" if clear_ssh_password else ssh_password,
|
||
"" if clear_ssh_private_key else ssh_private_key,
|
||
clear_ssh_password,
|
||
ssh_password,
|
||
ssh_password,
|
||
clear_ssh_private_key,
|
||
ssh_private_key,
|
||
ssh_private_key,
|
||
),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _pick_text_value(payload: dict, key: str, fallback: str = "", *, default: str = "") -> str:
|
||
if key in payload:
|
||
normalized = str(payload.get(key) or "").strip()
|
||
if normalized:
|
||
return normalized
|
||
normalized_fallback = str(fallback or "").strip()
|
||
if normalized_fallback:
|
||
return normalized_fallback
|
||
return str(default or "").strip()
|
||
|
||
|
||
def _pick_int_value(payload: dict, key: str, fallback: int, *, default: int = 0, minimum: int = 0) -> int:
|
||
if key in payload and payload.get(key) not in (None, ""):
|
||
try:
|
||
return max(int(payload.get(key) or default), minimum)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
return max(int(fallback or default), minimum)
|
||
except Exception:
|
||
return max(int(default), minimum)
|
||
|
||
|
||
def _compact_value(value: object, *, depth: int = 0) -> object:
|
||
if isinstance(value, str):
|
||
return value if len(value) <= 240 else f"{value[:240]}..."
|
||
if isinstance(value, (int, float, bool)) or value is None:
|
||
return value
|
||
if depth >= 2:
|
||
if isinstance(value, list):
|
||
return f"<list:{len(value)}>"
|
||
if isinstance(value, dict):
|
||
return f"<dict:{len(value)}>"
|
||
return str(value)
|
||
if isinstance(value, list):
|
||
limited = [_compact_value(item, depth=depth + 1) for item in value[:6]]
|
||
if len(value) > 6:
|
||
limited.append(f"...(+{len(value) - 6} more)")
|
||
return limited
|
||
if isinstance(value, dict):
|
||
compacted: dict[str, object] = {}
|
||
for index, (key, item) in enumerate(value.items()):
|
||
if index >= 8:
|
||
compacted["__truncated__"] = f"+{len(value) - 8} more keys"
|
||
break
|
||
compacted[str(key)] = _compact_value(item, depth=depth + 1)
|
||
return compacted
|
||
return str(value)
|
||
|
||
|
||
def _compact_job(job: dict) -> dict:
|
||
compacted = dict(job or {})
|
||
compacted["payload"] = _compact_value(job.get("payload") or {})
|
||
compacted["metadata"] = _compact_value(job.get("metadata") or {})
|
||
compacted["result"] = _compact_value(job.get("result") or {})
|
||
compacted["steps"] = []
|
||
compacted["is_compact"] = True
|
||
compacted["result_summary"] = _compact_value(job.get("result") or {})
|
||
return compacted
|
||
|
||
|
||
def _build_job_code() -> str:
|
||
stamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
||
return f"ops-{stamp}-{uuid4().hex[:6]}"
|
||
|
||
|
||
def _normalize_requested_by(value: object) -> str:
|
||
normalized = str(value or "api").strip() or "api"
|
||
return normalized[:64]
|
||
|
||
|
||
def upsert_managed_node(payload: dict) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
node_code = str(payload.get("node_code") or "").strip()
|
||
if not node_code:
|
||
return False, "node_code 不能为空", {}
|
||
parsed_ssh_entry = _parse_ssh_entry(payload.get("ssh_entry"))
|
||
merged_payload = {
|
||
**dict(payload or {}),
|
||
**{key: value for key, value in parsed_ssh_entry.items() if value not in ("", None)},
|
||
}
|
||
ssh_password = str(merged_payload.get("ssh_password") or "").strip()
|
||
ssh_private_key = str(merged_payload.get("ssh_private_key") or "")
|
||
clear_ssh_password = bool(merged_payload.get("clear_ssh_password", False))
|
||
clear_ssh_private_key = bool(merged_payload.get("clear_ssh_private_key", False))
|
||
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
node_code, region, role, title, ssh_host, ssh_port, ssh_user, auth_mode, deploy_channel, is_enabled, metadata_json, created_at, last_seen_at, updated_at
|
||
FROM ops_managed_nodes
|
||
WHERE node_code = %s
|
||
LIMIT 1
|
||
""",
|
||
(node_code,),
|
||
)
|
||
existing_row = cur.fetchone()
|
||
existing_node = _serialize_node_row(existing_row) if existing_row else {}
|
||
existing_metadata = dict(existing_node.get("metadata") or {})
|
||
incoming_metadata = dict(merged_payload.get("metadata") or {})
|
||
|
||
region = _pick_text_value(merged_payload, "region", str(existing_node.get("region") or ""), default="unknown") or "unknown"
|
||
role = _pick_text_value(merged_payload, "role", str(existing_node.get("role") or ""), default="worker") or "worker"
|
||
title = _pick_text_value(merged_payload, "title", str(existing_node.get("title") or ""), default=node_code) or node_code
|
||
ssh_host = _pick_text_value(merged_payload, "ssh_host", str(existing_node.get("ssh_host") or ""))
|
||
ssh_port = _pick_int_value(merged_payload, "ssh_port", int(existing_node.get("ssh_port") or 22), default=22, minimum=1)
|
||
ssh_user = _pick_text_value(merged_payload, "ssh_user", str(existing_node.get("ssh_user") or ""))
|
||
auth_mode = _pick_text_value(merged_payload, "auth_mode", str(existing_node.get("auth_mode") or ""), default="key") or "key"
|
||
deploy_channel = _pick_text_value(
|
||
merged_payload,
|
||
"deploy_channel",
|
||
str(existing_node.get("deploy_channel") or ""),
|
||
default="stable",
|
||
) or "stable"
|
||
is_enabled = bool(merged_payload["is_enabled"]) if "is_enabled" in merged_payload else bool(existing_node.get("is_enabled", True))
|
||
metadata = {
|
||
**existing_metadata,
|
||
**incoming_metadata,
|
||
}
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO ops_managed_nodes (
|
||
node_code, region, role, title, ssh_host, ssh_port, ssh_user, auth_mode, deploy_channel, is_enabled, metadata_json, updated_at
|
||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP)
|
||
ON CONFLICT (node_code) DO UPDATE SET
|
||
region = EXCLUDED.region,
|
||
role = EXCLUDED.role,
|
||
title = EXCLUDED.title,
|
||
ssh_host = EXCLUDED.ssh_host,
|
||
ssh_port = EXCLUDED.ssh_port,
|
||
ssh_user = EXCLUDED.ssh_user,
|
||
auth_mode = EXCLUDED.auth_mode,
|
||
deploy_channel = EXCLUDED.deploy_channel,
|
||
is_enabled = EXCLUDED.is_enabled,
|
||
metadata_json = EXCLUDED.metadata_json,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
RETURNING
|
||
node_code, region, role, title, ssh_host, ssh_port, ssh_user, auth_mode, deploy_channel, is_enabled, metadata_json, created_at, last_seen_at, updated_at
|
||
""",
|
||
(
|
||
node_code,
|
||
region,
|
||
role,
|
||
title,
|
||
ssh_host,
|
||
ssh_port,
|
||
ssh_user,
|
||
auth_mode,
|
||
deploy_channel,
|
||
is_enabled,
|
||
json.dumps(metadata, ensure_ascii=False),
|
||
),
|
||
)
|
||
row = cur.fetchone()
|
||
conn.commit()
|
||
_upsert_managed_node_secret(
|
||
node_code=node_code,
|
||
ssh_password=ssh_password,
|
||
ssh_private_key=ssh_private_key,
|
||
clear_ssh_password=clear_ssh_password,
|
||
clear_ssh_private_key=clear_ssh_private_key,
|
||
)
|
||
node = _serialize_node_row(row)
|
||
node.update(_load_node_secret_flags([node_code]).get(node_code, {}))
|
||
return True, "托管节点已保存", {"node": node}
|
||
|
||
|
||
def list_managed_nodes() -> list[dict]:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
node_code, region, role, title, ssh_host, ssh_port, ssh_user, auth_mode, deploy_channel, is_enabled, metadata_json, created_at, last_seen_at, updated_at
|
||
FROM ops_managed_nodes
|
||
ORDER BY region ASC, role ASC, node_code ASC
|
||
"""
|
||
)
|
||
rows = cur.fetchall()
|
||
items = [_serialize_node_row(row) for row in rows]
|
||
secret_flags = _load_node_secret_flags([str(item.get("node_code") or "") for item in items])
|
||
for item in items:
|
||
item.update(secret_flags.get(str(item.get("node_code") or "").strip(), {}))
|
||
return items
|
||
|
||
|
||
def sync_managed_nodes_from_cluster(*, dry_run: bool = False) -> dict:
|
||
from app.services.cluster_runtime_service import get_cluster_snapshot
|
||
|
||
snapshot = get_cluster_snapshot()
|
||
cluster_nodes = list(snapshot.get("nodes") or [])
|
||
synced_nodes: list[dict] = []
|
||
|
||
for cluster_node in cluster_nodes:
|
||
node_code = str(cluster_node.get("node_code") or "").strip()
|
||
if not node_code:
|
||
continue
|
||
payload = {
|
||
"node_code": node_code,
|
||
"region": str(cluster_node.get("region") or "unknown").strip() or "unknown",
|
||
"role": str(cluster_node.get("role") or "worker").strip() or "worker",
|
||
"title": str(cluster_node.get("hostname") or node_code).strip() or node_code,
|
||
"ssh_host": (
|
||
str(cluster_node.get("ip") or "").strip()
|
||
if str(cluster_node.get("ip") or "").strip() not in {"", "127.0.0.1", "::1", "localhost"}
|
||
else ""
|
||
),
|
||
"metadata": {
|
||
"imported_from_cluster": True,
|
||
"cluster_status": str(cluster_node.get("status") or ""),
|
||
"is_effective_worker": bool(cluster_node.get("is_effective_worker", False)),
|
||
"detect_participating": bool(cluster_node.get("detect_participating", False)),
|
||
"last_heartbeat_at": str(cluster_node.get("last_heartbeat_at") or ""),
|
||
"source_metadata": cluster_node.get("metadata") or {},
|
||
},
|
||
}
|
||
synced_nodes.append(payload)
|
||
if not dry_run:
|
||
upsert_managed_node(payload)
|
||
|
||
return {
|
||
"dry_run": dry_run,
|
||
"cluster_nodes_total": len(cluster_nodes),
|
||
"synced_count": len(synced_nodes),
|
||
"nodes": synced_nodes,
|
||
}
|
||
|
||
|
||
def _create_step(cur, *, job_id: int, step_key: str, title: str, node_code: str, status: str) -> int:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO ops_job_steps (
|
||
job_id, step_key, title, node_code, status, created_at, updated_at
|
||
) VALUES (%s, %s, %s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||
RETURNING id
|
||
""",
|
||
(job_id, step_key, title, node_code, status),
|
||
)
|
||
return int(cur.fetchone()[0])
|
||
|
||
|
||
def _update_step(
|
||
cur,
|
||
*,
|
||
step_id: int,
|
||
status: str,
|
||
stdout_text: str = "",
|
||
stderr_text: str = "",
|
||
result: dict | None = None,
|
||
started: bool = False,
|
||
finished: bool = False,
|
||
) -> None:
|
||
started_sql = "started_at = COALESCE(started_at, CURRENT_TIMESTAMP)," if started else ""
|
||
finished_sql = "finished_at = CURRENT_TIMESTAMP," if finished else ""
|
||
cur.execute(
|
||
f"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = %s,
|
||
stdout_text = %s,
|
||
stderr_text = %s,
|
||
result_json = %s,
|
||
{started_sql}
|
||
{finished_sql}
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
status,
|
||
stdout_text,
|
||
stderr_text,
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
step_id,
|
||
),
|
||
)
|
||
|
||
|
||
def _update_job(
|
||
cur,
|
||
*,
|
||
job_id: int,
|
||
status: str,
|
||
result: dict | None = None,
|
||
error_message: str = "",
|
||
started: bool = False,
|
||
finished: bool = False,
|
||
) -> None:
|
||
started_sql = "started_at = COALESCE(started_at, CURRENT_TIMESTAMP)," if started else ""
|
||
finished_sql = "finished_at = CURRENT_TIMESTAMP," if finished else ""
|
||
cur.execute(
|
||
f"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = %s,
|
||
result_json = %s,
|
||
error_message = %s,
|
||
{started_sql}
|
||
{finished_sql}
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
status,
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
error_message,
|
||
job_id,
|
||
),
|
||
)
|
||
|
||
|
||
def _get_job_row(cur, job_id: int) -> tuple | None:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
id, job_code, action, target_type, target_node_code, status, execution_mode, requested_by,
|
||
payload_json, metadata_json, result_json, error_message, created_at, started_at, finished_at, updated_at,
|
||
risk_level, approval_required, approval_status, approved_by, approved_at, blocked_reason,
|
||
cancellation_reason, dispatched_at, target_selector_json, policy_json, rollout_id
|
||
FROM ops_jobs
|
||
WHERE id = %s
|
||
""",
|
||
(job_id,),
|
||
)
|
||
return cur.fetchone()
|
||
|
||
|
||
def _get_steps(cur, job_id: int) -> list[dict]:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
id, step_key, title, node_code, status, stdout_text, stderr_text, result_json, created_at, started_at, finished_at, updated_at
|
||
FROM ops_job_steps
|
||
WHERE job_id = %s
|
||
ORDER BY id ASC
|
||
""",
|
||
(job_id,),
|
||
)
|
||
return [_serialize_step_row(row) for row in cur.fetchall()]
|
||
|
||
|
||
def _execute_local_job(action: str, payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
if action == "health.snapshot":
|
||
payload = get_runtime_status()
|
||
return True, "运行时快照采集成功", {"runtime": payload}
|
||
|
||
runtime_action_name = _LOCAL_RUNTIME_ACTIONS.get(action)
|
||
if runtime_action_name:
|
||
ok, message, data = runtime_action(runtime_action_name)
|
||
return ok, message, data
|
||
|
||
if supports_ssh_execution(action):
|
||
return execute_local_support_action(action, payload)
|
||
|
||
return False, f"当前未实现本机即时执行动作: {action}", {}
|
||
|
||
|
||
def _resolve_bootstrap_target_defaults(target_node_code: str) -> dict:
|
||
normalized_node_code = str(target_node_code or "").strip()
|
||
if not normalized_node_code:
|
||
return {"node_region": "mainland", "node_role": "worker"}
|
||
|
||
managed_node = next(
|
||
(item for item in list_managed_nodes() if str(item.get("node_code") or "").strip() == normalized_node_code),
|
||
{},
|
||
)
|
||
if managed_node:
|
||
return {
|
||
"node_region": str(managed_node.get("region") or "mainland").strip() or "mainland",
|
||
"node_role": str(managed_node.get("role") or "worker").strip() or "worker",
|
||
}
|
||
|
||
from app.services.cluster_runtime_service import get_cluster_snapshot
|
||
|
||
cluster = get_cluster_snapshot()
|
||
cluster_node = next(
|
||
(item for item in list(cluster.get("nodes") or []) if str(item.get("node_code") or "").strip() == normalized_node_code),
|
||
{},
|
||
)
|
||
return {
|
||
"node_region": str(cluster_node.get("region") or "mainland").strip() or "mainland",
|
||
"node_role": str(cluster_node.get("role") or "worker").strip() or "worker",
|
||
}
|
||
|
||
|
||
def _execute_control_plane_job(
|
||
action: str,
|
||
*,
|
||
job_id: int = 0,
|
||
target_node_code: str,
|
||
payload: dict | None = None,
|
||
requested_by: str = "api",
|
||
metadata: dict | None = None,
|
||
) -> tuple[bool, str, dict]:
|
||
normalized_payload = dict(payload or {})
|
||
normalized_metadata = dict(metadata or {})
|
||
|
||
if action == "node.bootstrap":
|
||
target_defaults = _resolve_bootstrap_target_defaults(target_node_code)
|
||
node_region = str(normalized_payload.get("node_region") or target_defaults.get("node_region") or "mainland").strip() or "mainland"
|
||
node_role = str(normalized_payload.get("node_role") or target_defaults.get("node_role") or "worker").strip() or "worker"
|
||
try:
|
||
expires_in_hours = int(normalized_payload.get("expires_in_hours") or 72)
|
||
except Exception:
|
||
expires_in_hours = 72
|
||
from app.services.ops_agent_service import build_node_agent_bootstrap_plan
|
||
|
||
ok, message, data = build_node_agent_bootstrap_plan(
|
||
node_code=str(target_node_code or "").strip(),
|
||
node_region=node_region,
|
||
node_role=node_role,
|
||
issued_by=str(requested_by or "api").strip() or "api",
|
||
expires_in_hours=expires_in_hours,
|
||
control_plane_base_url=str(normalized_payload.get("control_plane_base_url") or "").strip(),
|
||
root_dir=str(normalized_payload.get("root_dir") or "/opt/domaincheck").strip() or "/opt/domaincheck",
|
||
metadata={
|
||
**normalized_metadata,
|
||
"issued_from": "ops-job/control-plane",
|
||
"target_node_code": str(target_node_code or "").strip(),
|
||
"node_region": node_region,
|
||
"node_role": node_role,
|
||
},
|
||
)
|
||
if ok:
|
||
return True, message, {
|
||
**data,
|
||
"action": action,
|
||
"target_node_code": str(target_node_code or "").strip(),
|
||
"node_region": node_region,
|
||
"node_role": node_role,
|
||
"execution_mode": "control-plane",
|
||
}
|
||
return False, message, data
|
||
|
||
if action == "migration.execute":
|
||
from app.services.ops_migration_service import execute_ops_migration_job
|
||
|
||
return execute_ops_migration_job(
|
||
job_id=int(job_id or 0),
|
||
target_node_code=str(target_node_code or "").strip(),
|
||
payload=dict(normalized_payload or {}),
|
||
requested_by=str(requested_by or "api").strip() or "api",
|
||
metadata=dict(normalized_metadata or {}),
|
||
)
|
||
|
||
return False, f"当前未实现控制面执行动作: {action}", {}
|
||
|
||
|
||
def _execute_local_job_record(job_id: int) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
conn.rollback()
|
||
return False, "任务不存在", {}
|
||
job = _serialize_job_row(row)
|
||
if str(job.get("execution_mode") or "") != "local-runtime":
|
||
conn.rollback()
|
||
return False, "当前任务不是本机即时执行任务", {"job": job}
|
||
if str(job.get("target_node_code") or "") != settings.node_code:
|
||
conn.rollback()
|
||
return False, "当前任务目标节点不是本机", {"job": job}
|
||
if str(job.get("status") or "") != "queued":
|
||
conn.rollback()
|
||
return False, f"当前任务状态为 {job.get('status') or ''}，不可执行", {"job": job}
|
||
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||
dispatched_at = COALESCE(dispatched_at, CURRENT_TIMESTAMP),
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
conn.commit()
|
||
|
||
ok, message, result = _execute_local_job(str(job.get("action") or ""), dict(job.get("payload") or {}))
|
||
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = %s,
|
||
stdout_text = %s,
|
||
stderr_text = %s,
|
||
result_json = %s,
|
||
finished_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(
|
||
"success" if ok else "failed",
|
||
message if ok else "",
|
||
"" if ok else message,
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
int(job_id),
|
||
),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = %s,
|
||
result_json = %s,
|
||
error_message = %s,
|
||
finished_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
"success" if ok else "failed",
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
"" if ok else message,
|
||
int(job_id),
|
||
),
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_release_service import refresh_release_rollout_for_job
|
||
|
||
refresh_release_rollout_for_job(int(job_id))
|
||
return ok, message, {"job": get_ops_job(int(job_id))}
|
||
|
||
|
||
def _execute_control_plane_job_record(job_id: int) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
conn.rollback()
|
||
return False, "任务不存在", {}
|
||
job = _serialize_job_row(row)
|
||
if str(job.get("execution_mode") or "") != "control-plane":
|
||
conn.rollback()
|
||
return False, "当前任务不是控制面执行任务", {"job": job}
|
||
if str(job.get("status") or "") != "queued":
|
||
conn.rollback()
|
||
return False, f"当前任务状态为 {job.get('status') or ''}，不可执行", {"job": job}
|
||
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||
dispatched_at = COALESCE(dispatched_at, CURRENT_TIMESTAMP),
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
conn.commit()
|
||
|
||
ok, message, result = _execute_control_plane_job(
|
||
str(job.get("action") or ""),
|
||
job_id=int(job_id),
|
||
target_node_code=str(job.get("target_node_code") or ""),
|
||
payload=dict(job.get("payload") or {}),
|
||
requested_by=str(job.get("requested_by") or "api"),
|
||
metadata=dict(job.get("metadata") or {}),
|
||
)
|
||
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = %s,
|
||
stdout_text = %s,
|
||
stderr_text = %s,
|
||
result_json = %s,
|
||
finished_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(
|
||
"success" if ok else "failed",
|
||
message if ok else "",
|
||
"" if ok else message,
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
int(job_id),
|
||
),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = %s,
|
||
result_json = %s,
|
||
error_message = %s,
|
||
finished_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
"success" if ok else "failed",
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
"" if ok else message,
|
||
int(job_id),
|
||
),
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=str(job.get("target_node_code") or ""),
|
||
event_type="job_executed_on_control_plane",
|
||
level="info" if ok else "error",
|
||
message=f"任务已在控制面执行: {str(job.get('action') or '')}",
|
||
payload=result,
|
||
)
|
||
from app.services.ops_release_service import refresh_release_rollout_for_job
|
||
|
||
refresh_release_rollout_for_job(int(job_id))
|
||
return ok, message, {"job": get_ops_job(int(job_id))}
|
||
|
||
|
||
def _execute_ssh_job_record(job_id: int) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
conn.rollback()
|
||
return False, "任务不存在", {}
|
||
job = _serialize_job_row(row)
|
||
if str(job.get("execution_mode") or "") != "ssh":
|
||
conn.rollback()
|
||
return False, "当前任务不是 SSH 执行任务", {"job": job}
|
||
if str(job.get("status") or "") != "queued":
|
||
conn.rollback()
|
||
return False, f"当前任务状态为 {job.get('status') or ''}，不可执行", {"job": job}
|
||
|
||
target_node_code = str(job.get("target_node_code") or "").strip()
|
||
managed_node = next(
|
||
(item for item in list_managed_nodes() if str(item.get("node_code") or "").strip() == target_node_code),
|
||
{},
|
||
)
|
||
if not managed_node:
|
||
conn.rollback()
|
||
return False, "目标节点未纳入托管清单，无法执行 SSH 任务", {"job": job}
|
||
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||
dispatched_at = COALESCE(dispatched_at, CURRENT_TIMESTAMP),
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
conn.commit()
|
||
|
||
ok, message, result = execute_ssh_action(
|
||
managed_node,
|
||
str(job.get("action") or ""),
|
||
dict(job.get("payload") or {}),
|
||
)
|
||
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = %s,
|
||
stdout_text = %s,
|
||
stderr_text = %s,
|
||
result_json = %s,
|
||
finished_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(
|
||
"success" if ok else "failed",
|
||
message if ok else "",
|
||
"" if ok else message,
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
int(job_id),
|
||
),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = %s,
|
||
result_json = %s,
|
||
error_message = %s,
|
||
finished_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
"success" if ok else "failed",
|
||
json.dumps(result or {}, ensure_ascii=False),
|
||
"" if ok else message,
|
||
int(job_id),
|
||
),
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=str(job.get("target_node_code") or ""),
|
||
event_type="job_executed_over_ssh",
|
||
level="info" if ok else "error",
|
||
message=f"任务已通过 SSH 执行: {str(job.get('action') or '')}",
|
||
payload=result,
|
||
)
|
||
from app.services.ops_release_service import refresh_release_rollout_for_job
|
||
|
||
refresh_release_rollout_for_job(int(job_id))
|
||
return ok, message, {"job": get_ops_job(int(job_id))}
|
||
|
||
|
||
def create_ops_job(payload: dict) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
action = str(payload.get("action") or "").strip()
|
||
if not action:
|
||
return False, "action 不能为空", {}
|
||
|
||
target_type = str(payload.get("target_type") or "node").strip() or "node"
|
||
target_node_code = str(payload.get("target_node_code") or settings.node_code).strip() or settings.node_code
|
||
target_selector = payload.get("target_selector") or {}
|
||
requested_by = _normalize_requested_by(payload.get("requested_by"))
|
||
input_payload = payload.get("payload") or {}
|
||
metadata = payload.get("metadata") or {}
|
||
auto_approve = bool(payload.get("auto_approve", False))
|
||
rollout_id = int(payload.get("rollout_id") or 0) or None
|
||
local_action = target_node_code == settings.node_code and supports_local_runtime_action(action)
|
||
control_plane_action = supports_control_plane_action(action)
|
||
requested_execution_mode = (
|
||
"local-runtime"
|
||
if local_action
|
||
else ("control-plane" if control_plane_action else str(payload.get("execution_mode") or "remote-agent").strip() or "remote-agent")
|
||
)
|
||
|
||
from app.services.ops_policy_service import preview_ops_job_policy
|
||
|
||
policy_preview = preview_ops_job_policy(
|
||
{
|
||
"action": action,
|
||
"target_type": target_type,
|
||
"target_node_code": target_node_code,
|
||
"target_selector": target_selector,
|
||
"execution_mode": requested_execution_mode,
|
||
"payload": input_payload,
|
||
}
|
||
)
|
||
risk_level = str(policy_preview.get("risk_level") or "medium")
|
||
approval_required = bool(policy_preview.get("approval_required", False))
|
||
blocked = bool(policy_preview.get("blocked", False))
|
||
blocked_reason = "；".join(list(policy_preview.get("blocking_reasons") or []))
|
||
|
||
execution_mode = requested_execution_mode
|
||
run_now = (
|
||
bool(payload.get("run_now", True))
|
||
if (local_action or control_plane_action or execution_mode == "ssh")
|
||
and not blocked
|
||
and not (approval_required and not auto_approve)
|
||
else False
|
||
)
|
||
job_code = _build_job_code()
|
||
initial_status = "blocked" if blocked else ("awaiting_approval" if approval_required and not auto_approve else "queued")
|
||
approval_status = "blocked" if blocked else ("approved" if (not approval_required or auto_approve) else "pending")
|
||
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO ops_jobs (
|
||
job_code, action, target_type, target_node_code, status, execution_mode, requested_by, payload_json, metadata_json,
|
||
created_at, updated_at, risk_level, approval_required, approval_status, approved_by, approved_at,
|
||
blocked_reason, target_selector_json, policy_json, rollout_id
|
||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""",
|
||
(
|
||
job_code,
|
||
action,
|
||
target_type,
|
||
target_node_code,
|
||
initial_status,
|
||
execution_mode,
|
||
requested_by,
|
||
json.dumps(input_payload, ensure_ascii=False),
|
||
json.dumps(metadata, ensure_ascii=False),
|
||
risk_level,
|
||
approval_required,
|
||
approval_status,
|
||
requested_by if approval_required and auto_approve else "",
|
||
datetime.now() if approval_required and auto_approve else None,
|
||
blocked_reason,
|
||
json.dumps(target_selector, ensure_ascii=False),
|
||
json.dumps(policy_preview, ensure_ascii=False),
|
||
rollout_id,
|
||
),
|
||
)
|
||
job_id = int(cur.fetchone()[0])
|
||
step_id = _create_step(
|
||
cur,
|
||
job_id=job_id,
|
||
step_key="dispatch",
|
||
title="调度动作",
|
||
node_code=target_node_code,
|
||
status=initial_status,
|
||
)
|
||
|
||
if blocked:
|
||
policy_result = {
|
||
"dispatch_state": "blocked",
|
||
"policy": policy_preview,
|
||
"hint": "请先解决阻断条件，再重新创建或重新排队任务。",
|
||
}
|
||
_update_job(cur, job_id=job_id, status="blocked", result=policy_result, error_message=blocked_reason)
|
||
_update_step(
|
||
cur,
|
||
step_id=step_id,
|
||
status="blocked",
|
||
stderr_text=blocked_reason,
|
||
result=policy_result,
|
||
finished=True,
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=job_id,
|
||
node_code=target_node_code,
|
||
event_type="job_blocked",
|
||
level="warning",
|
||
message=f"任务已被策略阻断: {action}",
|
||
payload={"policy": policy_preview},
|
||
)
|
||
return False, blocked_reason or "任务被策略阻断", {"job": get_ops_job(job_id), "executed_immediately": False}
|
||
|
||
if approval_required and not auto_approve:
|
||
approval_result = {
|
||
"dispatch_state": "awaiting_approval",
|
||
"policy": policy_preview,
|
||
"hint": "该任务需要审批后才能进入执行队列。",
|
||
}
|
||
_update_job(cur, job_id=job_id, status="awaiting_approval", result=approval_result)
|
||
_update_step(
|
||
cur,
|
||
step_id=step_id,
|
||
status="awaiting_approval",
|
||
result=approval_result,
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=job_id,
|
||
node_code=target_node_code,
|
||
event_type="job_awaiting_approval",
|
||
level="info",
|
||
message=f"任务等待审批: {action}",
|
||
payload={"policy": policy_preview},
|
||
)
|
||
return True, "运维任务已创建，等待审批", {"job": get_ops_job(job_id), "executed_immediately": False}
|
||
|
||
if run_now:
|
||
_update_job(cur, job_id=job_id, status="running", started=True)
|
||
_update_step(cur, step_id=step_id, status="running", started=True)
|
||
if execution_mode == "control-plane":
|
||
ok, message, result = _execute_control_plane_job(
|
||
action,
|
||
job_id=job_id,
|
||
target_node_code=target_node_code,
|
||
payload=dict(input_payload or {}),
|
||
requested_by=requested_by,
|
||
metadata=dict(metadata or {}),
|
||
)
|
||
elif execution_mode == "ssh":
|
||
managed_node = next(
|
||
(item for item in list_managed_nodes() if str(item.get("node_code") or "").strip() == target_node_code),
|
||
{},
|
||
)
|
||
if not managed_node:
|
||
ok, message, result = (
|
||
False,
|
||
"目标节点未纳入托管清单，无法执行 SSH 任务",
|
||
{"target_node_code": target_node_code, "executor": "ssh"},
|
||
)
|
||
else:
|
||
ok, message, result = execute_ssh_action(
|
||
managed_node,
|
||
action,
|
||
dict(input_payload or {}),
|
||
)
|
||
else:
|
||
ok, message, result = _execute_local_job(action, dict(input_payload or {}))
|
||
_update_step(
|
||
cur,
|
||
step_id=step_id,
|
||
status="success" if ok else "failed",
|
||
stdout_text=message if ok else "",
|
||
stderr_text="" if ok else message,
|
||
result=result,
|
||
finished=True,
|
||
)
|
||
_update_job(
|
||
cur,
|
||
job_id=job_id,
|
||
status="success" if ok else "failed",
|
||
result=result,
|
||
error_message="" if ok else message,
|
||
started=True,
|
||
finished=True,
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=job_id,
|
||
node_code=target_node_code,
|
||
event_type=(
|
||
"job_executed_on_control_plane"
|
||
if execution_mode == "control-plane"
|
||
else ("job_executed_over_ssh" if execution_mode == "ssh" else "job_executed_locally")
|
||
),
|
||
level="info" if ok else "error",
|
||
message=(
|
||
f"任务已在控制面即时执行: {action}"
|
||
if execution_mode == "control-plane"
|
||
else (f"任务已通过 SSH 即时执行: {action}" if execution_mode == "ssh" else f"任务已在本机即时执行: {action}")
|
||
),
|
||
payload=result,
|
||
)
|
||
return ok, message, {"job": get_ops_job(job_id), "executed_immediately": True}
|
||
|
||
queued_result = {
|
||
"dispatch_state": "queued",
|
||
"waiting_for": (
|
||
"node-agent"
|
||
if execution_mode == "remote-agent"
|
||
else ("ssh-executor" if execution_mode == "ssh" else execution_mode)
|
||
),
|
||
"hint": (
|
||
"后续可由海外控制面的控制面执行器生成接入工单。"
|
||
if execution_mode == "control-plane"
|
||
else (
|
||
"该任务将由海外控制面的 SSH 执行器直接连到目标节点执行。"
|
||
if execution_mode == "ssh"
|
||
else "后续可由海外控制面的 node-agent / SSH 执行器消费该任务。"
|
||
)
|
||
),
|
||
"policy": policy_preview,
|
||
}
|
||
_update_job(cur, job_id=job_id, status="queued", result=queued_result)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=job_id,
|
||
node_code=target_node_code,
|
||
event_type="job_created",
|
||
level="info",
|
||
message=f"任务已创建并进入队列: {action}",
|
||
payload={"execution_mode": execution_mode, "policy": policy_preview},
|
||
)
|
||
return True, "运维任务已创建，等待执行器接管", {"job": get_ops_job(job_id), "executed_immediately": False}
|
||
|
||
|
||
def create_ops_job_batch(payload: dict) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
template_key = str(payload.get("template_key") or "").strip()
|
||
target_node_codes = []
|
||
seen_node_codes: set[str] = set()
|
||
for item in list(payload.get("target_node_codes") or []):
|
||
node_code = str(item or "").strip()
|
||
if not node_code or node_code in seen_node_codes:
|
||
continue
|
||
seen_node_codes.add(node_code)
|
||
target_node_codes.append(node_code)
|
||
if not target_node_codes:
|
||
return False, "target_node_codes 不能为空", {}
|
||
|
||
requested_by = _normalize_requested_by(payload.get("requested_by"))
|
||
input_payload = payload.get("payload") or {}
|
||
metadata = payload.get("metadata") or {}
|
||
execution_mode = str(payload.get("execution_mode") or "").strip()
|
||
auto_approve = payload.get("auto_approve")
|
||
|
||
from app.services.ops_template_service import build_ops_template_payload, get_ops_action_template
|
||
|
||
template = get_ops_action_template(template_key)
|
||
if not template:
|
||
return False, "动作模板不存在", {}
|
||
|
||
normalize_ok, normalize_message, normalized_action_payload = build_ops_template_payload(template_key, input_payload)
|
||
if not normalize_ok:
|
||
return False, normalize_message, {}
|
||
|
||
action = str(template.get("action") or "").strip()
|
||
execution_mode = execution_mode or str(template.get("default_execution_mode") or "remote-agent").strip() or "remote-agent"
|
||
if auto_approve is None:
|
||
auto_approve = bool(template.get("default_auto_approve", False))
|
||
else:
|
||
auto_approve = bool(auto_approve)
|
||
|
||
created_jobs: list[dict] = []
|
||
item_results: list[dict] = []
|
||
status_counts: dict[str, int] = {}
|
||
created_count = 0
|
||
blocked_count = 0
|
||
awaiting_approval_count = 0
|
||
failed_count = 0
|
||
|
||
for node_code in target_node_codes:
|
||
ok, message, data = create_ops_job(
|
||
{
|
||
"action": action,
|
||
"target_type": "node",
|
||
"target_node_code": node_code,
|
||
"requested_by": requested_by,
|
||
"execution_mode": execution_mode,
|
||
"auto_approve": auto_approve,
|
||
"payload": normalized_action_payload,
|
||
"metadata": metadata,
|
||
}
|
||
)
|
||
job = (data.get("job") or {}) if isinstance(data, dict) else {}
|
||
compact_job = _compact_job(job) if job else {}
|
||
job_status = str(job.get("status") or ("failed" if not ok else "unknown")).strip() or "unknown"
|
||
status_counts[job_status] = int(status_counts.get(job_status, 0) or 0) + 1
|
||
if compact_job:
|
||
created_jobs.append(compact_job)
|
||
created_count += 1
|
||
if job_status == "blocked":
|
||
blocked_count += 1
|
||
elif job_status == "awaiting_approval":
|
||
awaiting_approval_count += 1
|
||
elif not ok and not job:
|
||
failed_count += 1
|
||
item_results.append(
|
||
{
|
||
"node_code": node_code,
|
||
"ok": ok,
|
||
"message": message,
|
||
"status": job_status,
|
||
"job": compact_job,
|
||
}
|
||
)
|
||
|
||
if created_count <= 0:
|
||
return False, "未能创建任何运维任务", {
|
||
"template": template,
|
||
"results": item_results,
|
||
"status_counts": status_counts,
|
||
}
|
||
|
||
summary = {
|
||
"template_key": template_key,
|
||
"action": action,
|
||
"requested_by": requested_by,
|
||
"execution_mode": execution_mode,
|
||
"auto_approve": auto_approve,
|
||
"target_nodes_total": len(target_node_codes),
|
||
"created_count": created_count,
|
||
"blocked_count": blocked_count,
|
||
"awaiting_approval_count": awaiting_approval_count,
|
||
"failed_count": failed_count,
|
||
"status_counts": status_counts,
|
||
}
|
||
message = (
|
||
f"已创建 {created_count} 个运维任务"
|
||
f"；待审批 {awaiting_approval_count}"
|
||
f"；阻断 {blocked_count}"
|
||
f"；失败 {failed_count}"
|
||
)
|
||
return True, message, {
|
||
"template": template,
|
||
"summary": summary,
|
||
"metadata": _compact_value(metadata),
|
||
"jobs": created_jobs,
|
||
"results": item_results,
|
||
}
|
||
|
||
|
||
def list_ops_jobs(limit: int = 20, *, rollout_id: int | None = None, compact: bool = True) -> list[dict]:
|
||
ensure_ops_schema()
|
||
limited = min(max(int(limit or 20), 1), 10000)
|
||
conditions: list[str] = []
|
||
params: list[object] = []
|
||
if rollout_id is not None and int(rollout_id or 0) > 0:
|
||
conditions.append("rollout_id = %s")
|
||
params.append(int(rollout_id))
|
||
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, job_code, action, target_type, target_node_code, status, execution_mode, requested_by,
|
||
payload_json, metadata_json, result_json, error_message, created_at, started_at, finished_at, updated_at,
|
||
risk_level, approval_required, approval_status, approved_by, approved_at, blocked_reason,
|
||
cancellation_reason, dispatched_at, target_selector_json, policy_json, rollout_id
|
||
FROM ops_jobs
|
||
{where_clause}
|
||
ORDER BY id DESC
|
||
LIMIT %s
|
||
""",
|
||
(*params, limited),
|
||
)
|
||
rows = cur.fetchall()
|
||
jobs = [_serialize_job_row(row) for row in rows]
|
||
if compact:
|
||
return [_compact_job(job) for job in jobs]
|
||
return jobs
|
||
|
||
|
||
def get_ops_job(job_id: int) -> dict:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
return {}
|
||
steps = _get_steps(cur, int(job_id))
|
||
return _serialize_job_row(row, steps=steps)
|
||
|
||
|
||
def get_ops_job_summary() -> dict:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT status, count(*)
|
||
FROM ops_jobs
|
||
GROUP BY status
|
||
"""
|
||
)
|
||
rows = cur.fetchall()
|
||
status_counts = {str(row[0] or ""): int(row[1] or 0) for row in rows}
|
||
return {
|
||
"status_counts": status_counts,
|
||
"queued": int(status_counts.get("queued", 0) or 0),
|
||
"running": int(status_counts.get("running", 0) or 0),
|
||
"awaiting_approval": int(status_counts.get("awaiting_approval", 0) or 0),
|
||
"blocked": int(status_counts.get("blocked", 0) or 0),
|
||
"cancelled": int(status_counts.get("cancelled", 0) or 0),
|
||
"success": int(status_counts.get("success", 0) or 0),
|
||
"failed": int(status_counts.get("failed", 0) or 0),
|
||
"total": sum(status_counts.values()),
|
||
}
|
||
|
||
|
||
def approve_ops_job(job_id: int, *, approved_by: str = "api") -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
normalized_approved_by = str(approved_by or "api").strip() or "api"
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
conn.rollback()
|
||
return False, "任务不存在", {}
|
||
job = _serialize_job_row(row)
|
||
if str(job.get("status") or "") != "awaiting_approval":
|
||
conn.rollback()
|
||
return False, "当前任务不处于待审批状态", {"job": job}
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = 'queued',
|
||
approval_status = 'approved',
|
||
approved_by = %s,
|
||
approved_at = CURRENT_TIMESTAMP,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(normalized_approved_by, int(job_id)),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = 'queued',
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE job_id = %s
|
||
""",
|
||
(int(job_id),),
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=str((get_ops_job(int(job_id)) or {}).get("target_node_code") or ""),
|
||
event_type="job_approved",
|
||
level="info",
|
||
message=f"任务审批通过: {job_id}",
|
||
payload={"approved_by": normalized_approved_by},
|
||
)
|
||
from app.services.ops_release_service import refresh_release_rollout_for_job
|
||
|
||
refresh_release_rollout_for_job(int(job_id))
|
||
return True, "任务已审批通过并进入队列", {"job": get_ops_job(int(job_id))}
|
||
|
||
|
||
def cancel_ops_job(job_id: int, *, cancelled_by: str = "api", reason: str = "") -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
normalized_cancelled_by = str(cancelled_by or "api").strip() or "api"
|
||
normalized_reason = str(reason or "").strip()
|
||
with get_db() as conn:
|
||
conn.autocommit = False
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
conn.rollback()
|
||
return False, "任务不存在", {}
|
||
job = _serialize_job_row(row)
|
||
current_status = str(job.get("status") or "")
|
||
if current_status in {"success", "failed", "cancelled"}:
|
||
conn.rollback()
|
||
return False, f"当前任务状态为 {current_status}，不可取消", {"job": job}
|
||
result = dict(job.get("result") or {})
|
||
result["cancelled_by"] = normalized_cancelled_by
|
||
if normalized_reason:
|
||
result["cancel_reason"] = normalized_reason
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_jobs
|
||
SET
|
||
status = 'cancelled',
|
||
cancellation_reason = %s,
|
||
result_json = %s,
|
||
updated_at = CURRENT_TIMESTAMP,
|
||
finished_at = CASE WHEN finished_at IS NULL THEN CURRENT_TIMESTAMP ELSE finished_at END
|
||
WHERE id = %s
|
||
""",
|
||
(
|
||
normalized_reason,
|
||
json.dumps(result, ensure_ascii=False),
|
||
int(job_id),
|
||
),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
UPDATE ops_job_steps
|
||
SET
|
||
status = CASE WHEN status IN ('success', 'failed', 'cancelled') THEN status ELSE 'cancelled' END,
|
||
stderr_text = CASE WHEN %s <> '' THEN %s ELSE stderr_text END,
|
||
updated_at = CURRENT_TIMESTAMP,
|
||
finished_at = CASE WHEN finished_at IS NULL THEN CURRENT_TIMESTAMP ELSE finished_at END
|
||
WHERE job_id = %s
|
||
""",
|
||
(
|
||
normalized_reason,
|
||
normalized_reason,
|
||
int(job_id),
|
||
),
|
||
)
|
||
conn.commit()
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=str((get_ops_job(int(job_id)) or {}).get("target_node_code") or ""),
|
||
event_type="job_cancelled",
|
||
level="warning",
|
||
message=f"任务已取消: {job_id}",
|
||
payload={"cancelled_by": normalized_cancelled_by, "reason": normalized_reason},
|
||
)
|
||
from app.services.ops_release_service import refresh_release_rollout_for_job
|
||
|
||
refresh_release_rollout_for_job(int(job_id))
|
||
return True, "任务已取消", {"job": get_ops_job(int(job_id))}
|
||
|
||
|
||
def dispatch_ops_job(job_id: int) -> tuple[bool, str, dict]:
|
||
ensure_ops_schema()
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
row = _get_job_row(cur, int(job_id))
|
||
if not row:
|
||
return False, "任务不存在", {}
|
||
job = _serialize_job_row(row)
|
||
|
||
status = str(job.get("status") or "")
|
||
if status != "queued":
|
||
return False, f"当前任务状态为 {status}，不可派发", {"job": job}
|
||
|
||
target_node_code = str(job.get("target_node_code") or "")
|
||
execution_mode = str(job.get("execution_mode") or "")
|
||
if execution_mode == "control-plane":
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=target_node_code,
|
||
event_type="job_dispatch_requested",
|
||
level="info",
|
||
message=f"请求控制面执行任务: {job_id}",
|
||
payload={"action": str(job.get('action') or ''), "execution_mode": execution_mode},
|
||
)
|
||
ok, message, data = _execute_control_plane_job_record(int(job_id))
|
||
return ok, f"已按控制面执行派发：{message}", data
|
||
|
||
if execution_mode == "local-runtime" and target_node_code == settings.node_code:
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=target_node_code,
|
||
event_type="job_dispatch_requested",
|
||
level="info",
|
||
message=f"请求本机即时执行任务: {job_id}",
|
||
payload={"action": str(job.get('action') or '')},
|
||
)
|
||
ok, message, data = _execute_local_job_record(int(job_id))
|
||
return ok, f"已按本机即时执行派发：{message}", data
|
||
|
||
if execution_mode == "ssh":
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=target_node_code,
|
||
event_type="job_dispatch_requested",
|
||
level="info",
|
||
message=f"请求 SSH 执行器执行任务: {job_id}",
|
||
payload={"action": str(job.get('action') or ''), "execution_mode": execution_mode},
|
||
)
|
||
ok, message, data = _execute_ssh_job_record(int(job_id))
|
||
return ok, f"已按 SSH 执行派发：{message}", data
|
||
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=target_node_code,
|
||
event_type="job_dispatch_requested",
|
||
level="info",
|
||
message=f"任务等待 node-agent 拉取: {job_id}",
|
||
payload={"execution_mode": execution_mode},
|
||
)
|
||
return True, "任务保持 queued，等待对应 node-agent 拉取", {"job": job}
|