Nexus Agent
09d4613e46
366 台批量执行 results JSON 超 TEXT 64KB 导致 flush 失败、状态卡在 running。 Co-authored-by: Cursor <cursoragent@cursor.com>
561 lines
18 KiB
Python
561 lines
18 KiB
Python
"""Script execution live state in Redis (2h TTL); MySQL writes are batched only (no per-event flush)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Optional
|
|
|
|
from server.domain.models import ScriptExecution
|
|
from server.domain.repositories import ScriptExecutionRepository
|
|
from server.infrastructure.redis.client import get_redis
|
|
|
|
logger = logging.getLogger("nexus.script_execution_store")
|
|
|
|
REDIS_KEY_PREFIX = "script_exec:"
|
|
REDIS_FLUSH_QUEUE = "script_exec:flush_queue"
|
|
REDIS_TTL_SECONDS = 7200 # 2 hours — live execution state max lifetime
|
|
FLUSH_INTERVAL_SECONDS = 60 # background batch flush Redis → MySQL
|
|
FLUSH_BATCH_LIMIT = 50 # max execution records written per flush cycle
|
|
STUCK_NO_PROGRESS_SECONDS = 3600 # auto mark stuck if running with no update for 1h
|
|
# MySQL TEXT ≈64KB — large batch runs overflow; persist uses MEDIUMTEXT + trimmed I/O.
|
|
PERSIST_STDIO_MAX_CHARS = 512
|
|
MYSQL_RESULTS_SOFT_LIMIT_BYTES = 3_500_000
|
|
|
|
TERMINAL_STATUSES = frozenset({"completed", "failed", "partial", "cancelled"})
|
|
|
|
_FAILED_SERVER_STATUSES = frozenset({
|
|
"failed",
|
|
"error",
|
|
"skipped",
|
|
"timeout",
|
|
"connection_error",
|
|
"unknown",
|
|
"cancelled",
|
|
})
|
|
|
|
|
|
def _is_server_exec_success(entry: Any) -> bool:
|
|
"""True when a server finished successfully (counts toward progress numerator)."""
|
|
if not isinstance(entry, dict) or not entry:
|
|
return False
|
|
phase = str(entry.get("phase") or "").strip()
|
|
if phase in ("pending", "cancelled"):
|
|
return False
|
|
status = str(entry.get("status") or "").strip().lower()
|
|
if status in _FAILED_SERVER_STATUSES:
|
|
return False
|
|
exit_code = entry.get("exit_code")
|
|
if exit_code is not None and exit_code != 0:
|
|
return False
|
|
if exit_code == 0:
|
|
return True
|
|
if status in ("success", "completed", "done"):
|
|
return True
|
|
if phase == "done":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _is_server_exec_done(entry: Any) -> bool:
|
|
"""True when a server has a non-pending result (counts toward done/remaining)."""
|
|
if not isinstance(entry, dict) or not entry:
|
|
return False
|
|
phase = str(entry.get("phase") or "").strip()
|
|
if phase == "pending":
|
|
return False
|
|
return True
|
|
|
|
|
|
def execution_progress_stats(
|
|
results: dict[str, Any],
|
|
server_ids: Optional[list] = None,
|
|
) -> dict[str, Any]:
|
|
"""Counts for UI / WebSocket: done, remaining, success, failed, total, progress."""
|
|
ids = server_ids
|
|
if ids is None:
|
|
ids = [int(k) for k in results if k != "_meta" and str(k).isdigit()]
|
|
total = len(ids)
|
|
done = 0
|
|
success = 0
|
|
failed_done = 0
|
|
for sid in ids:
|
|
entry = results.get(str(sid))
|
|
if not _is_server_exec_done(entry):
|
|
continue
|
|
done += 1
|
|
if _is_server_exec_success(entry):
|
|
success += 1
|
|
else:
|
|
failed_done += 1
|
|
return {
|
|
"total": total,
|
|
"done": done,
|
|
"remaining": max(0, total - done),
|
|
"success": success,
|
|
"failed": failed_done,
|
|
"progress": execution_progress_summary(results, ids),
|
|
}
|
|
|
|
|
|
def execution_progress_summary(
|
|
results: dict[str, Any],
|
|
server_ids: Optional[list] = None,
|
|
) -> str:
|
|
"""Return 'success_count/total' for UI (failures and pending excluded from numerator)."""
|
|
ids = server_ids
|
|
if ids is None:
|
|
ids = [int(k) for k in results if k != "_meta" and str(k).isdigit()]
|
|
total = len(ids)
|
|
if not total:
|
|
return "—"
|
|
success = sum(
|
|
1
|
|
for sid in ids
|
|
if _is_server_exec_success(results.get(str(sid)))
|
|
)
|
|
return f"{success}/{total}"
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _parse_server_ids_json(raw: Optional[str]) -> list:
|
|
try:
|
|
parsed = json.loads(raw or "[]")
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return parsed if isinstance(parsed, list) else []
|
|
|
|
|
|
def _redis_key(execution_id: int) -> str:
|
|
return f"{REDIS_KEY_PREFIX}{execution_id}"
|
|
|
|
|
|
def unpack_results_blob(raw: Optional[str]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
if not raw:
|
|
return {}, []
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return {}, []
|
|
if not isinstance(data, dict):
|
|
return {}, []
|
|
meta = data.pop("_meta", None) or {}
|
|
events = meta.get("events", []) if isinstance(meta, dict) else []
|
|
return data, events if isinstance(events, list) else []
|
|
|
|
|
|
def pack_results_blob(results: dict[str, Any], events: list[dict[str, Any]]) -> str:
|
|
payload = dict(results)
|
|
payload["_meta"] = {
|
|
"events": events,
|
|
"updated_at": _now_iso(),
|
|
}
|
|
return json.dumps(payload, ensure_ascii=False)
|
|
|
|
|
|
def _truncate_text(text: str, max_chars: int) -> tuple[str, bool]:
|
|
if len(text) <= max_chars:
|
|
return text, False
|
|
suffix = "\n…(truncated)"
|
|
keep = max(0, max_chars - len(suffix))
|
|
return text[:keep] + suffix, True
|
|
|
|
|
|
def compact_results_for_mysql_persist(results: dict[str, Any]) -> dict[str, Any]:
|
|
"""Trim per-server stdout/stderr before MySQL flush (Redis keeps full live state)."""
|
|
out: dict[str, Any] = {}
|
|
for sid, entry in results.items():
|
|
if sid == "_meta" or not isinstance(entry, dict):
|
|
continue
|
|
row = {k: v for k, v in entry.items() if k not in ("stdout", "stderr")}
|
|
for field in ("stdout", "stderr"):
|
|
raw = entry.get(field)
|
|
if raw is None:
|
|
continue
|
|
truncated, was = _truncate_text(str(raw), PERSIST_STDIO_MAX_CHARS)
|
|
row[field] = truncated
|
|
if was:
|
|
row[f"{field}_truncated"] = True
|
|
out[str(sid)] = row
|
|
return out
|
|
|
|
|
|
def pack_mysql_results_blob(
|
|
results: dict[str, Any],
|
|
events: list[dict[str, Any]],
|
|
) -> str:
|
|
"""Compact I/O then pack; shrink further if JSON still exceeds soft byte limit."""
|
|
compact = compact_results_for_mysql_persist(results)
|
|
blob = pack_results_blob(compact, events)
|
|
if len(blob.encode("utf-8")) <= MYSQL_RESULTS_SOFT_LIMIT_BYTES:
|
|
return blob
|
|
|
|
slimmer: dict[str, Any] = {}
|
|
for sid, entry in compact.items():
|
|
row = dict(entry)
|
|
row.pop("stdout", None)
|
|
if "stderr" in row:
|
|
row["stderr"], _ = _truncate_text(str(row["stderr"]), 256)
|
|
slimmer[sid] = row
|
|
blob = pack_results_blob(slimmer, events)
|
|
if len(blob.encode("utf-8")) <= MYSQL_RESULTS_SOFT_LIMIT_BYTES:
|
|
return blob
|
|
|
|
summary: dict[str, Any] = {}
|
|
for sid, entry in compact.items():
|
|
stderr, _ = _truncate_text(str(entry.get("stderr") or ""), 128)
|
|
summary[sid] = {
|
|
"status": entry.get("status"),
|
|
"exit_code": entry.get("exit_code"),
|
|
"phase": entry.get("phase"),
|
|
"stderr": stderr,
|
|
"persist_summary_only": True,
|
|
}
|
|
return pack_results_blob(summary, events)
|
|
|
|
|
|
async def load_live(execution_id: int) -> Optional[dict[str, Any]]:
|
|
redis = get_redis()
|
|
raw = await redis.get(_redis_key(execution_id))
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
async def save_live(live: dict[str, Any]) -> None:
|
|
redis = get_redis()
|
|
eid = int(live["id"])
|
|
live["updated_at"] = _now_iso()
|
|
await redis.set(_redis_key(eid), json.dumps(live, ensure_ascii=False), ex=REDIS_TTL_SECONDS)
|
|
await redis.sadd(REDIS_FLUSH_QUEUE, str(eid))
|
|
|
|
|
|
async def init_live_from_execution(
|
|
execution: ScriptExecution,
|
|
*,
|
|
long_task: bool = False,
|
|
results: Optional[dict[str, Any]] = None,
|
|
) -> dict[str, Any]:
|
|
server_ids = _parse_server_ids_json(execution.server_ids)
|
|
|
|
live = {
|
|
"id": execution.id,
|
|
"script_id": execution.script_id,
|
|
"command": execution.command,
|
|
"server_ids": server_ids,
|
|
"status": execution.status or "running",
|
|
"results": results or {},
|
|
"events": [
|
|
{
|
|
"at": _now_iso(),
|
|
"action": "started",
|
|
"detail": f"开始执行 {len(server_ids)} 台服务器"
|
|
+ (" (长任务)" if long_task else ""),
|
|
"operator": execution.operator,
|
|
}
|
|
],
|
|
"operator": execution.operator,
|
|
"long_task": long_task,
|
|
"started_at": str(execution.started_at) if execution.started_at else _now_iso(),
|
|
"completed_at": None,
|
|
}
|
|
if getattr(execution, "credential_id", None) is not None:
|
|
live["credential_id"] = execution.credential_id
|
|
await save_live(live)
|
|
return live
|
|
|
|
|
|
async def append_event(
|
|
live: dict[str, Any],
|
|
action: str,
|
|
detail: str,
|
|
operator: Optional[str] = None,
|
|
) -> None:
|
|
live.setdefault("events", []).append(
|
|
{
|
|
"at": _now_iso(),
|
|
"action": action,
|
|
"detail": detail,
|
|
"operator": operator or live.get("operator"),
|
|
}
|
|
)
|
|
|
|
|
|
async def flush_to_mysql(
|
|
execution_repo: ScriptExecutionRepository,
|
|
live: dict[str, Any],
|
|
) -> Optional[ScriptExecution]:
|
|
"""Persist current live snapshot to MySQL (results + event timeline in _meta)."""
|
|
eid = int(live["id"])
|
|
status = live.get("status", "running")
|
|
results = live.get("results") or {}
|
|
events = live.get("events") or []
|
|
blob = pack_mysql_results_blob(results, events)
|
|
|
|
execution = await execution_repo.update_status(eid, status, blob)
|
|
if not execution:
|
|
logger.warning("Flush skipped: script execution %s not found in MySQL", eid)
|
|
redis = get_redis()
|
|
await redis.srem(REDIS_FLUSH_QUEUE, str(eid))
|
|
return None
|
|
|
|
if status in TERMINAL_STATUSES and execution:
|
|
execution.completed_at = datetime.now(timezone.utc)
|
|
# update_status already sets completed_at when not running — good
|
|
|
|
redis = get_redis()
|
|
await redis.srem(REDIS_FLUSH_QUEUE, str(eid))
|
|
logger.debug("Flushed script execution %s to MySQL (status=%s)", eid, status)
|
|
return execution
|
|
|
|
|
|
async def apply_update(
|
|
execution_repo: ScriptExecutionRepository,
|
|
execution_id: int,
|
|
*,
|
|
status: str,
|
|
results: dict[str, Any],
|
|
event_action: str,
|
|
event_detail: str,
|
|
operator: str,
|
|
terminal: bool = False,
|
|
long_task: Optional[bool] = None,
|
|
) -> dict[str, Any]:
|
|
"""Update Redis only; MySQL is flushed in batches by script_execution_flush (60s)."""
|
|
live = await load_live(execution_id)
|
|
if not live:
|
|
execution = await execution_repo.get_by_id(execution_id)
|
|
if not execution:
|
|
raise ValueError(f"Execution {execution_id} not found")
|
|
res, ev = unpack_results_blob(execution.results)
|
|
live = {
|
|
"id": execution_id,
|
|
"script_id": execution.script_id,
|
|
"command": execution.command,
|
|
"server_ids": _parse_server_ids_json(execution.server_ids),
|
|
"status": execution.status,
|
|
"results": res,
|
|
"events": ev,
|
|
"operator": execution.operator,
|
|
"long_task": (execution.command or "").startswith("[long_task] "),
|
|
"started_at": str(execution.started_at) if execution.started_at else _now_iso(),
|
|
"completed_at": str(execution.completed_at) if execution.completed_at else None,
|
|
}
|
|
|
|
live["status"] = status
|
|
live["results"] = results
|
|
if long_task is not None:
|
|
live["long_task"] = long_task
|
|
if terminal or status in TERMINAL_STATUSES:
|
|
live["completed_at"] = _now_iso()
|
|
|
|
await append_event(live, event_action, event_detail, operator)
|
|
await save_live(live)
|
|
return live
|
|
|
|
|
|
async def get_execution_view(
|
|
execution_repo: ScriptExecutionRepository,
|
|
execution_id: int,
|
|
) -> Optional[dict[str, Any]]:
|
|
"""Prefer Redis live view; fall back to MySQL."""
|
|
live = await load_live(execution_id)
|
|
if live:
|
|
results = live.get("results") or {}
|
|
return {
|
|
"id": live["id"],
|
|
"script_id": live.get("script_id"),
|
|
"command": live.get("command"),
|
|
"server_ids": json.dumps(live.get("server_ids", [])),
|
|
"status": live.get("status"),
|
|
"results": results,
|
|
"events": live.get("events") or [],
|
|
"operator": live.get("operator"),
|
|
"started_at": live.get("started_at"),
|
|
"completed_at": live.get("completed_at"),
|
|
"long_task": live.get("long_task", False),
|
|
"progress": execution_progress_summary(results, live.get("server_ids")),
|
|
"credential_id": live.get("credential_id"),
|
|
"label": live.get("label"),
|
|
"source": "redis",
|
|
}
|
|
|
|
execution = await execution_repo.get_by_id(execution_id)
|
|
if not execution:
|
|
return None
|
|
results, events = unpack_results_blob(execution.results)
|
|
server_ids = _parse_server_ids_json(execution.server_ids)
|
|
return {
|
|
"id": execution.id,
|
|
"script_id": execution.script_id,
|
|
"command": execution.command,
|
|
"server_ids": execution.server_ids,
|
|
"status": execution.status,
|
|
"results": results,
|
|
"events": events,
|
|
"operator": execution.operator,
|
|
"started_at": str(execution.started_at) if execution.started_at else None,
|
|
"completed_at": str(execution.completed_at) if execution.completed_at else None,
|
|
"long_task": (execution.command or "").startswith("[long_task] "),
|
|
"progress": execution_progress_summary(results, server_ids),
|
|
"credential_id": getattr(execution, "credential_id", None),
|
|
"source": "mysql",
|
|
}
|
|
|
|
|
|
async def patch_live_server_result(
|
|
execution_id: int,
|
|
server_id: int,
|
|
entry: dict[str, Any],
|
|
) -> None:
|
|
"""Merge one server result into Redis live state (for stop while short-task running)."""
|
|
live = await load_live(execution_id)
|
|
if not live:
|
|
return
|
|
merged = dict(entry)
|
|
if merged.get("phase") == "pending" and "started_at" not in merged:
|
|
merged["started_at"] = _now_iso()
|
|
results = dict(live.get("results") or {})
|
|
results[str(server_id)] = {**(results.get(str(server_id)) or {}), **merged}
|
|
live["results"] = results
|
|
await save_live(live)
|
|
|
|
|
|
def _parse_iso(ts: str) -> Optional[datetime]:
|
|
try:
|
|
return datetime.fromisoformat(ts)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _pending_stuck_server_ids(results: dict[str, Any], now: datetime) -> list[str]:
|
|
stuck: list[str] = []
|
|
for sid, entry in results.items():
|
|
if sid == "_meta" or not isinstance(entry, dict):
|
|
continue
|
|
if entry.get("phase") != "pending" or entry.get("auto_stuck"):
|
|
continue
|
|
started = _parse_iso(entry.get("started_at", ""))
|
|
if not started:
|
|
continue
|
|
if (now - started).total_seconds() >= STUCK_NO_PROGRESS_SECONDS:
|
|
stuck.append(str(sid))
|
|
return stuck
|
|
|
|
|
|
async def detect_stuck_executions(
|
|
execution_repo: ScriptExecutionRepository,
|
|
audit_fn,
|
|
) -> int:
|
|
"""Auto-mark executions with no Redis update for STUCK_NO_PROGRESS_SECONDS."""
|
|
redis = get_redis()
|
|
marked = 0
|
|
now = datetime.now(timezone.utc)
|
|
cursor = 0
|
|
while True:
|
|
cursor, keys = await redis.scan(cursor, match=f"{REDIS_KEY_PREFIX}*", count=50)
|
|
for key in keys:
|
|
try:
|
|
eid = int(key.replace(REDIS_KEY_PREFIX, ""))
|
|
except ValueError:
|
|
continue
|
|
live = await load_live(eid)
|
|
if not live:
|
|
continue
|
|
if live.get("status") not in ("running", "partial"):
|
|
continue
|
|
results = dict(live.get("results") or {})
|
|
events = live.get("events") or []
|
|
|
|
stuck_sids = _pending_stuck_server_ids(results, now)
|
|
if stuck_sids:
|
|
for sid in stuck_sids:
|
|
row = dict(results.get(sid) or {})
|
|
row["auto_stuck"] = True
|
|
row["stuck_at"] = _now_iso()
|
|
results[sid] = row
|
|
await apply_update(
|
|
execution_repo,
|
|
eid,
|
|
status=live.get("status", "running"),
|
|
results=results,
|
|
event_action="server_auto_stuck",
|
|
event_detail=(
|
|
f"服务器 {', '.join('#' + s for s in stuck_sids)} "
|
|
f"pending 超过 {STUCK_NO_PROGRESS_SECONDS // 3600} 小时"
|
|
),
|
|
operator="system",
|
|
)
|
|
await audit_fn(
|
|
"server_auto_stuck",
|
|
"script_execution",
|
|
eid,
|
|
live.get("operator", "system"),
|
|
)
|
|
marked += 1
|
|
continue
|
|
|
|
if live.get("status") != "running":
|
|
continue
|
|
if any(e.get("action") in ("stuck_marked", "auto_stuck") for e in events[-3:]):
|
|
continue
|
|
updated = _parse_iso(live.get("updated_at", ""))
|
|
if not updated or (now - updated).total_seconds() < STUCK_NO_PROGRESS_SECONDS:
|
|
continue
|
|
await apply_update(
|
|
execution_repo,
|
|
eid,
|
|
status="running",
|
|
results=results,
|
|
event_action="auto_stuck",
|
|
event_detail=(
|
|
f"超过 {STUCK_NO_PROGRESS_SECONDS // 3600} 小时无状态更新，系统自动标记可能卡住"
|
|
),
|
|
operator="system",
|
|
)
|
|
await audit_fn(
|
|
"auto_stuck",
|
|
"script_execution",
|
|
eid,
|
|
live.get("operator", "system"),
|
|
)
|
|
marked += 1
|
|
if cursor == 0:
|
|
break
|
|
return marked
|
|
|
|
|
|
async def flush_pending_batch(execution_repo: ScriptExecutionRepository) -> int:
|
|
"""Batch-write queued execution snapshots from Redis to MySQL (cap per cycle)."""
|
|
redis = get_redis()
|
|
try:
|
|
ids = await redis.smembers(REDIS_FLUSH_QUEUE)
|
|
except Exception as e:
|
|
logger.error("Redis smembers script_exec flush queue failed: %s", e)
|
|
return 0
|
|
|
|
id_list = list(ids or [])[:FLUSH_BATCH_LIMIT]
|
|
flushed = 0
|
|
for raw_id in id_list:
|
|
try:
|
|
eid = int(raw_id)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
live = await load_live(eid)
|
|
if not live:
|
|
await redis.srem(REDIS_FLUSH_QUEUE, raw_id)
|
|
continue
|
|
try:
|
|
await flush_to_mysql(execution_repo, live)
|
|
flushed += 1
|
|
except Exception as e:
|
|
logger.error("Flush script execution %s failed: %s", eid, e)
|
|
return flushed
|