From 5d3c895542f4fc45a5f797e5da2705fe0fca0d37 Mon Sep 17 00:00:00 2001 From: zhaolei <12153398@qq.com> Date: Thu, 29 Oct 2026 17:02:37 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20policy=20=E5=8A=A0=E8=BD=BD=20bug=20+=20?= =?UTF-8?q?DB=20=E7=B4=A2=E5=BC=95=20+=20=E5=BD=92=E6=A1=A3=E8=84=9A?= =?UTF-8?q?=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【Bug 修复】policy.py _load_db_or_yaml 根因: select(PolicyVersion).order_by(desc).scalar_one_or_none() 表有多条历史版本 → Multiple rows 异常 → 每次启动都回退 YAML 修复: 改成 db.query(PolicyVersion).order_by(desc).first() 用 ORM Query API 取最新一条, 返回真正的 PolicyVersion 对象 验证: 容器启动后日志中不再出现 'DB 加载策略失败（回退 YAML）' 【索引】models.py ScanTask 加复合索引 ix_scan_tasks_channel_created ON (channel, created_at DESC) 解决 dashboard/tasks_page 高频查询的全表扫描问题 (已在运行中 DB 用 CREATE INDEX CONCURRENTLY 补上) 【归档】scripts/archive_old_tasks.py - scan_tasks 90天前 清 detail JSONB (省空间) - scan_tasks 365天前 移到 scan_tasks_archive 归档表 - idempotency_records 7天 TTL 清过期幂等缓存 - policy_versions 只留最新 5 个 - 支持 --dry-run (默认) + --apply (真正执行) - 分批 1000 条处理, 避免长事务锁表 - 归档表 scan_tasks_archive 已建 (含 4 个索引) 数据增长: 日均 523 条 → 年 ~19万条 ≈ 910 MB 当前 3660 条 ≈ 4.5 MB, 增长可控, 脚本可每周/每月 cron 跑一次 --- app/models.py | 5 + app/services/policy.py | 11 +- scripts/archive_old_tasks.py | 264 +++++++++++++++++++++++++++++++++++ 3 files changed, 276 insertions(+), 4 deletions(-) create mode 100644 scripts/archive_old_tasks.py diff --git a/app/models.py b/app/models.py index c4a18b4..7ca7b89 100644 --- a/app/models.py +++ b/app/models.py @@ -71,6 +71,11 @@ class ScanTask(Base): back_populates="task", cascade="all, delete-orphan", lazy="selectin" ) + __table_args__ = ( + # 控制台高频查询: WHERE channel = X ORDER BY created_at DESC LIMIT N + Index("ix_scan_tasks_channel_created", "channel", "created_at", postgresql_using="btree"), + ) + class ScanFinding(Base): __tablename__ = "scan_findings" diff --git a/app/services/policy.py b/app/services/policy.py index 4f587d4..b919baa 100644 --- a/app/services/policy.py +++ b/app/services/policy.py @@ -300,15 +300,18 @@ class PolicyStore: def _load_db_or_yaml(self) -> Policy: """优先 DB，回退 YAML。""" try: - from sqlalchemy import select from app.db import SessionLocal from app.models import PolicyVersion db = SessionLocal() try: - pv = db.execute( - select(PolicyVersion).order_by(PolicyVersion.created_at.desc()) - ).scalar_one_or_none() + # 用 ORM Query API — 直接返回 PolicyVersion 对象 + # 而不是 Row 对象（db.execute(select) 会返回 Row） + pv = ( + db.query(PolicyVersion) + .order_by(PolicyVersion.created_at.desc()) + .first() # 取最新一条，不管历史版本多少条 + ) if pv: policy = Policy( version=pv.version, diff --git a/scripts/archive_old_tasks.py b/scripts/archive_old_tasks.py new file mode 100644 index 0000000..422e263 --- /dev/null +++ b/scripts/archive_old_tasks.py @@ -0,0 +1,264 @@ +"""DB 归档 + 清理脚本。 + +职责: + 1. scan_tasks 90 天前: 清 detail JSONB (节省空间, 扫描详情留 90 天) + 2. scan_tasks 365 天前: 移到 scan_tasks_archive (含 ScanFinding) + 3. idempotency_records: 清 7 天前的 (幂等缓存过期重扫) + 4. policy_versions: 只保留最新 5 个版本 (历史版本清掉) + +用法: + docker compose exec api python3 scripts/archive_old_tasks.py # dry-run, 只报告不动数据 + docker compose exec api python3 scripts/archive_old_tasks.py --apply # 真正执行 + +建议: 每天 cron 跑一次 + +设计原则: + - 小批量提交 (1000 条一批), 避免长事务锁表 + - 先 dry-run 再 apply, 输出清晰 + - pg_repack / VACUUM ANALYZE 在脚本最后触发 (让 autovacuum 兜底也行) +""" +from __future__ import annotations + +import argparse +import logging +import sys +import os +from datetime import timedelta, datetime, timezone +from pathlib import Path + +sys.path.insert(0, "/app") +os.environ.setdefault("MESHSAFE_MASTER_KEY", "Dt0ktRs8SjEH8VxWZJDX125NV1qiII88HoiXZPhWers=") + +from app.db import SessionLocal # noqa: E402 +from app.models import ( # noqa: E402 + ScanTask, ScanFinding, IdempotencyRecord, PolicyVersion, +) +from sqlalchemy import select, func # noqa: E402 + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +log = logging.getLogger("archive") + +# ---- 配置 ---- +DETAIL_RETENTION_DAYS = 90 # 90 天前清 detail JSONB +ARCHIVE_RETENTION_DAYS = 365 # 365 天前移归档 +IDEM_RETENTION_DAYS = 7 # 幂等缓存 7 天 TTL +POLICY_KEEP_VERSIONS = 5 # policy_versions 只留最新 5 个 +BATCH_SIZE = 1000 # 分批处理 + + +def _cutoff(days: int) -> datetime: + return datetime.now(timezone.utc) - timedelta(days=days) + + +def archive_tasks(dry_run: bool) -> None: + """归档 365 天前的 scan_tasks 到 scan_tasks_archive。""" + cutoff = _cutoff(ARCHIVE_RETENTION_DAYS) + db = SessionLocal() + try: + total = db.execute( + select(func.count(ScanTask.id)).where(ScanTask.created_at < cutoff) + ).scalar() or 0 + + log.info(f"[归档] {ARCHIVE_RETENTION_DAYS} 天前的 scan_tasks: {total} 条") + + if total == 0: + log.info(" → 无需归档") + return + + if dry_run: + log.info(f" [dry-run] 将归档 {total} 条") + return + + # 分批 + moved = 0 + while True: + batch = db.execute( + select(ScanTask).where(ScanTask.created_at < cutoff).limit(BATCH_SIZE) + ).scalars().all() + if not batch: + break + + for task in batch: + try: + db.execute( + __import__("sqlalchemy").text(""" + INSERT INTO scan_tasks_archive + (id, channel, client_id, filename, declared_ext, size, + sha256, md5, real_type, status, mode, verdict, action, score, + reason, policy_version, has_detail, has_findings, + created_at, finished_at, duration_ms, archived_at) + VALUES + (:id, :channel, :client_id, :filename, :declared_ext, :size, + :sha256, :md5, :real_type, :status, :mode, :verdict, :action, :score, + :reason, :policy_version, :has_detail, :has_findings, + :created_at, :finished_at, :duration_ms, NOW()) + ON CONFLICT (id) DO NOTHING + """), + { + "id": task.id, + "channel": task.channel, + "client_id": task.client_id, + "filename": task.filename, + "declared_ext": task.declared_ext, + "size": task.size, + "sha256": task.sha256, + "md5": task.md5, + "real_type": task.real_type, + "status": task.status, + "mode": task.mode, + "verdict": task.verdict, + "action": task.action, + "score": task.score, + "reason": task.reason, + "policy_version": task.policy_version, + "has_detail": bool(task.detail and len(task.detail) > 0), + "has_findings": bool(task.findings), + "created_at": task.created_at, + "finished_at": task.finished_at, + "duration_ms": task.duration_ms, + }, + ) + # CASCADE 删 ScanFinding + db.delete(task) + moved += 1 + except Exception as e: + log.warning(f" task {task.id} 归档失败: {e}") + db.rollback() + continue + db.commit() + log.info(f" 已归档 {moved}/{total}") + + log.info(f" ✅ 归档完成: {moved} 条") + finally: + db.close() + + +def scrub_detail(dry_run: bool) -> None: + """90~365 天之间的任务清 detail JSONB (留摘要, 省空间)。""" + db_cutoff = _cutoff(DETAIL_RETENTION_DAYS) + arc_cutoff = _cutoff(ARCHIVE_RETENTION_DAYS) + db = SessionLocal() + try: + total = db.execute( + select(func.count(ScanTask.id)).where( + ScanTask.created_at < db_cutoff, + ScanTask.created_at >= arc_cutoff, + ScanTask.detail.isnot(None), + ) + ).scalar() or 0 + + log.info(f"[清 detail] {DETAIL_RETENTION_DAYS}~{ARCHIVE_RETENTION_DAYS} 天之间有 detail 的: {total} 条") + + if total == 0: + log.info(" → 无需清理") + return + + if dry_run: + log.info(f" [dry-run] 将清 {total} 条 detail") + return + + # 分批 UPDATE + updated = 0 + while True: + batch_ids = db.execute( + select(ScanTask.id).where( + ScanTask.created_at < db_cutoff, + ScanTask.created_at >= arc_cutoff, + ScanTask.detail.isnot(None), + ).limit(BATCH_SIZE) + ).scalars().all() + if not batch_ids: + break + db.execute( + __import__("sqlalchemy").update(ScanTask).where(ScanTask.id.in_(batch_ids)).values(detail=None) + ) + db.commit() + updated += len(batch_ids) + log.info(f" 已清 {updated}/{total}") + + log.info(f" ✅ detail 清理完成: {updated} 条") + finally: + db.close() + + +def clean_idempotency(dry_run: bool) -> None: + """清 7 天前的幂等缓存 (过期的可以重扫)。""" + cutoff = _cutoff(IDEM_RETENTION_DAYS) + db = SessionLocal() + try: + total = db.execute( + select(func.count(IdempotencyRecord.id)).where(IdempotencyRecord.created_at < cutoff) + ).scalar() or 0 + + log.info(f"[幂等清理] {IDEM_RETENTION_DAYS} 天前的 idempotency_records: {total} 条") + + if total == 0: + log.info(" → 无需清理") + return + + if dry_run: + log.info(f" [dry-run] 将清 {total} 条") + return + + db.execute( + __import__("sqlalchemy").delete(IdempotencyRecord).where(IdempotencyRecord.created_at < cutoff) + ) + db.commit() + log.info(f" ✅ 幂等清理完成: {total} 条") + finally: + db.close() + + +def trim_policy_versions(dry_run: bool) -> None: + """policy_versions 只留最新 N 个。""" + db = SessionLocal() + try: + all_pv = db.execute( + select(PolicyVersion).order_by(PolicyVersion.created_at.desc()) + ).scalars().all() + + keep = all_pv[:POLICY_KEEP_VERSIONS] + drop = all_pv[POLICY_KEEP_VERSIONS:] + + log.info(f"[policy] 共 {len(all_pv)} 条历史版本, 留 {len(keep)} 删 {len(drop)}") + + if not drop: + log.info(" → 无需清理") + return + + if dry_run: + for pv in drop: + log.info(f" [dry-run] 将删 {pv.version}") + return + + for pv in drop: + db.delete(pv) + db.commit() + log.info(f" ✅ policy_versions 清理完成: 删 {len(drop)} 条") + finally: + db.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description="MeshSafe DB 归档 + 清理") + parser.add_argument("--apply", action="store_true", help="真正执行 (默认 dry-run)") + args = parser.parse_args() + + dry = not args.apply + log.info(f"模式: {'DRY-RUN (仅报告)' if dry else 'APPLY (真正执行)'}") + log.info("") + + archive_tasks(dry) + log.info("") + scrub_detail(dry) + log.info("") + clean_idempotency(dry) + log.info("") + trim_policy_versions(dry) + + if dry: + log.info("\n💡 确认无误后加 --apply 参数正式执行") + + +if __name__ == "__main__": + main()