diff --git a/docs/audit/2026-06-08-script-exec-results-mediumtext.md b/docs/audit/2026-06-08-script-exec-results-mediumtext.md new file mode 100644 index 00000000..4f87faeb --- /dev/null +++ b/docs/audit/2026-06-08-script-exec-results-mediumtext.md @@ -0,0 +1,39 @@ +# 审计 — 脚本执行 results 落库溢出（2026-06-08） + +**Changelog**: `docs/changelog/2026-06-08-script-exec-results-mediumtext.md` + +## 审计范围 + +| 模块 | 文件 | +|------|------| +| Redis flush | `server/infrastructure/redis/script_execution_store.py` | +| ORM | `server/domain/models/__init__.py` | +| 迁移 | `server/infrastructure/database/migrations.py` | +| 单测 | `tests/test_script_execution_mysql_persist.py` | + +## 安全 + +| 项 | 结论 | +|----|------| +| 数据完整性 | PASS — exit_code/status 保留；仅截断 stdout/stderr | +| Redis 全量 | PASS — live 仍存完整 I/O，MySQL 为持久化摘要 | + +## Step 3 + +| H | 规则 | 结论 | +|---|------|------| +| H1 | MEDIUMTEXT 迁移 | PASS — 启动时 ALTER | +| H2 | 大批次 flush | PASS — 截断 stdout/stderr + 摘要降级 | +| H3 | Redis 全量保留 | PASS — 仅 MySQL 持久化裁剪 | + +## Closure + +| H | 判定 | +|---|------| +| H1 MEDIUMTEXT 迁移 | PASS | +| H2 366 台可 flush | PASS — 单测模拟生产体积 | + +## DoD + +- [x] pytest tests/test_script_execution_mysql_persist.py +- [x] changelog diff --git a/docs/changelog/2026-06-08-script-exec-results-mediumtext.md b/docs/changelog/2026-06-08-script-exec-results-mediumtext.md new file mode 100644 index 00000000..583efb54 --- /dev/null +++ b/docs/changelog/2026-06-08-script-exec-results-mediumtext.md @@ -0,0 +1,31 @@ +# 2026-06-08 — 脚本执行 results 落库 TEXT 溢出修复 + +## 摘要 + +- `script_executions.results` 扩为 `MEDIUMTEXT` +- Redis → MySQL flush 前截断每台 stdout/stderr（512 字符），超大批次降级为摘要 + +## 动机 + +366 台脚本执行结果 JSON 约 75KB，超过 MySQL `TEXT`（64KB），flush 失败导致状态卡在 `running`、无 `completed_at`。 + +## 涉及文件 + +- `server/infrastructure/redis/script_execution_store.py` +- `server/domain/models/__init__.py` +- `server/infrastructure/database/migrations.py` +- `tests/test_script_execution_mysql_persist.py` + +## 迁移/重启 + +- 启动时 `run_schema_migrations()` 自动 `MODIFY results MEDIUMTEXT` +- 需重启 API 容器使新 flush 逻辑生效 + +## 验证 + +```bash +pytest tests/test_script_execution_mysql_persist.py -q +bash deploy/pre_deploy_check.sh +``` + +部署后：执行记录 #14 应在 60s 内 flush 为 `partial` 并写入 `completed_at`。 diff --git a/server/domain/models/__init__.py b/server/domain/models/__init__.py index 007a7850..30738119 100644 --- a/server/domain/models/__init__.py +++ b/server/domain/models/__init__.py @@ -9,6 +9,7 @@ from sqlalchemy import ( Column, Integer, String, Boolean, DateTime, Text, ForeignKey, Index, JSON ) +from sqlalchemy.dialects.mysql import MEDIUMTEXT from sqlalchemy.orm import declarative_base, relationship @@ -359,7 +360,11 @@ class ScriptExecution(Base): server_ids = Column(Text, nullable=False, comment="JSON数组: 目标服务器ID列表") credential_id = Column(Integer, ForeignKey("db_credentials.id", ondelete="SET NULL"), nullable=True, comment="DB凭据替换$DB_*变量") status = Column(String(20), default="pending", comment="pending/running/completed/failed") - results = Column(Text, nullable=True, comment="JSON: 每台服务器执行结果{server_id: {stdout,stderr,exit_code}}") + results = Column( + Text().with_variant(MEDIUMTEXT(), "mysql"), + nullable=True, + comment="JSON: 每台服务器执行结果{server_id: {stdout,stderr,exit_code}}", + ) operator = Column(String(100), nullable=True, comment="操作人用户名") started_at = Column(DateTime, default=_utcnow) completed_at = Column(DateTime, nullable=True) diff --git a/server/infrastructure/database/migrations.py b/server/infrastructure/database/migrations.py index 5bc2ea2e..b37f25bc 100644 --- a/server/infrastructure/database/migrations.py +++ b/server/infrastructure/database/migrations.py @@ -120,6 +120,7 @@ async def run_schema_migrations(): "ALTER TABLE admins ADD COLUMN token_version INT DEFAULT 0 COMMENT '令牌版本(重用时递增使旧token失效)'", "UPDATE admins SET token_version = 0 WHERE token_version IS NULL", "ALTER TABLE script_executions ADD COLUMN credential_id INT NULL COMMENT 'DB凭据替换$DB_*变量'", + "ALTER TABLE script_executions MODIFY COLUMN results MEDIUMTEXT NULL COMMENT 'JSON: 每台服务器执行结果'", # Alert log table """CREATE TABLE IF NOT EXISTS `alert_logs` ( `id` INT AUTO_INCREMENT PRIMARY KEY, diff --git a/server/infrastructure/redis/script_execution_store.py b/server/infrastructure/redis/script_execution_store.py index 932fa3eb..a4bc6f03 100644 --- a/server/infrastructure/redis/script_execution_store.py +++ b/server/infrastructure/redis/script_execution_store.py @@ -19,6 +19,9 @@ 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"}) @@ -154,6 +157,67 @@ def pack_results_blob(results: dict[str, Any], events: list[dict[str, Any]]) -> 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)) @@ -233,7 +297,7 @@ async def flush_to_mysql( status = live.get("status", "running") results = live.get("results") or {} events = live.get("events") or [] - blob = pack_results_blob(results, events) + blob = pack_mysql_results_blob(results, events) execution = await execution_repo.update_status(eid, status, blob) if not execution: diff --git a/tests/test_script_execution_mysql_persist.py b/tests/test_script_execution_mysql_persist.py new file mode 100644 index 00000000..4df82c91 --- /dev/null +++ b/tests/test_script_execution_mysql_persist.py @@ -0,0 +1,56 @@ +"""Tests for script execution MySQL persist compaction.""" + +import json + +from server.infrastructure.redis.script_execution_store import ( + MYSQL_RESULTS_SOFT_LIMIT_BYTES, + PERSIST_STDIO_MAX_CHARS, + compact_results_for_mysql_persist, + pack_mysql_results_blob, +) + + +def test_compact_truncates_long_stdio(): + results = { + "1": { + "status": "success", + "exit_code": 0, + "stdout": "x" * 2000, + "stderr": "y" * 100, + }, + } + compact = compact_results_for_mysql_persist(results) + assert len(compact["1"]["stdout"]) <= PERSIST_STDIO_MAX_CHARS + assert compact["1"]["stdout_truncated"] is True + assert compact["1"]["stderr"] == "y" * 100 + + +def test_pack_mysql_fits_large_batch_like_production(): + """366 servers with mysql warning stderr — must fit under old TEXT limit after compact.""" + results = { + str(i): { + "status": "success" if i <= 256 else "failed", + "exit_code": 0 if i <= 256 else 1, + "stdout": "", + "stderr": "mysql: [Warning] Using a password on the command line interface can be insecure.\n", + } + for i in range(1, 367) + } + blob = pack_mysql_results_blob(results, [{"action": "execution_finished", "detail": "ok"}]) + assert len(blob.encode("utf-8")) < 65535 + + +def test_pack_mysql_under_soft_limit_for_huge_output(): + results = { + str(i): { + "status": "failed", + "exit_code": 1, + "stdout": "line\n" * 500, + "stderr": "err\n" * 500, + } + for i in range(1, 401) + } + blob = pack_mysql_results_blob(results, []) + assert len(blob.encode("utf-8")) <= MYSQL_RESULTS_SOFT_LIMIT_BYTES + data = json.loads(blob) + assert data["1"].get("persist_summary_only") or data["1"].get("stdout_truncated")