diff --git a/docker-compose.yml b/docker-compose.yml index 51745cc..498f085 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -133,6 +133,29 @@ services: start_period: 15s retries: 3 + # ---------- Cron (定时任务: DB 归档 + VACUUM) ---------- + # 纯 Python 实现, 零额外依赖, 和 api 共享 meshsafe:latest 镜像 + # 任务: + # 每天 03:00 archive_old_tasks.py --apply (清 90 天 detail / 365 天归档 / 幂等 TTL) + # 每周一 03:10 VACUUM ANALYZE (维护表统计信息) + cron: + image: meshsafe:latest + container_name: meshsafe-cron + restart: unless-stopped + networks: [meshsafe_net] + depends_on: + postgres: + condition: service_healthy + environment: + - TZ=Asia/Shanghai + - MESHSAFE_ENV=prod + - MESHSAFE_DB_URL=postgresql+psycopg2://meshsafe:${MESHSAFE_DB_PASSWORD:-meshsafe}@postgres:5432/meshsafe + - MESHSAFE_MASTER_KEY=${MESHSAFE_MASTER_KEY:?MASTER_KEY_REQUIRED} + volumes: + - ./data/logs:/var/log/meshsafe # 归档日志共享目录 + - ./scripts:/app/scripts:ro # archive_old_tasks.py + cron_entrypoint.py + entrypoint: ["python3", "/app/scripts/cron_entrypoint.py"] + # ---------- Nginx(反向代理,内网 HTTP 明文)---------- # 外部端口 44380 → 内部 80 → api:8443 # 无 TLS,外层防护靠光闸 diff --git a/scripts/cron_entrypoint.py b/scripts/cron_entrypoint.py new file mode 100644 index 0000000..92850cc --- /dev/null +++ b/scripts/cron_entrypoint.py @@ -0,0 +1,121 @@ +"""MeshSafe 定时任务调度器 — 零依赖,Docker 容器前台常驻。 + +和 api 共享同一个镜像,容器 entrypoint 换成本脚本即可。 +用法: docker run meshsafe:latest python3 scripts/cron_entrypoint.py + +支持的任务: + 每天 03:00 归档 + 清理 DB (archive_old_tasks.py --apply) + 每周一 03:10 VACUUM ANALYZE (维护表统计信息) +""" +from __future__ import annotations + +import logging +import os +import subprocess +import sys +import time +from datetime import datetime, timedelta + +sys.path.insert(0, "/app") + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", +) +log = logging.getLogger("scheduler") + + +def run(cmd: list[str], label: str) -> None: + """跑一个命令,捕获输出并记日志。""" + log.info(f"[{label}] 开始执行: {' '.join(cmd)}") + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=1800) + if proc.stdout: + for line in proc.stdout.strip().splitlines(): + log.info(f"[{label}] {line}") + if proc.stderr: + for line in proc.stderr.strip().splitlines(): + log.warning(f"[{label}] {line}") + if proc.returncode == 0: + log.info(f"[{label}] ✅ 完成") + else: + log.error(f"[{label}] ❌ 返回码 {proc.returncode}") + except subprocess.TimeoutExpired: + log.error(f"[{label}] ❌ 超时 (30min)") + except Exception as e: + log.error(f"[{label}] ❌ 异常: {e}") + + +def run_vacuum() -> None: + """VACUUM ANALYZE — 维护表统计信息。""" + try: + from app.db import engine + from sqlalchemy import text + + tables = [ + "scan_tasks", "scan_tasks_archive", + "scan_findings", "idempotency_records", + "policy_versions", "audit_logs", + ] + with engine.begin() as conn: + for tbl in tables: + log.info(f"[VACUUM] {tbl} ...") + conn.execute(text(f"VACUUM ANALYZE {tbl}")) + log.info("[VACUUM] ✅ 全部完成") + except Exception as e: + log.error(f"[VACUUM] ❌ 异常: {e}") + + +def seconds_until(target_hour: int, target_minute: int, weekday: int | None = None) -> float: + """计算到下次目标时间的秒数。 + + weekday: 0=周一...6=周日, None=每天 + """ + now = datetime.now() + candidate = now.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) + + if weekday is not None: + # 找到本周目标日 + days_ahead = weekday - now.weekday() + if days_ahead < 0 or (days_ahead == 0 and now >= candidate): + days_ahead += 7 + candidate += timedelta(days=days_ahead) + else: + if candidate <= now: + candidate += timedelta(days=1) + + return (candidate - now).total_seconds() + + +def main() -> None: + log.info("=" * 60) + log.info("MeshSafe 定时任务调度器启动") + log.info(f"PID={os.getpid()} TZ=Asia/Shanghai") + log.info("任务:") + log.info(" - 每天 03:00 archive_old_tasks.py --apply") + log.info(" - 每周一 03:10 VACUUM ANALYZE") + log.info("=" * 60) + + while True: + # 计算两个任务下次执行时间,取更早的那个 + secs_archive = seconds_until(3, 0, weekday=None) # 每天 03:00 + secs_vacuum = seconds_until(3, 10, weekday=0) # 周一 03:10 + + if secs_archive <= secs_vacuum: + log.info(f"下一个任务: 归档 → {secs_archive/3600:.1f}h 后") + time.sleep(secs_archive) + run( + [sys.executable, "/app/scripts/archive_old_tasks.py", "--apply"], + "ARCHIVE", + ) + # 跑完后清下一次时间 (避免在 03:00 这一刻两个任务重叠) + continue + else: + log.info(f"下一个任务: VACUUM → {secs_vacuum/3600:.1f}h 后") + time.sleep(secs_vacuum) + run_vacuum() + continue + + +if __name__ == "__main__": + main()