yml2213
34d667cba0
作业是在调度线程里同步跑的：进程一重启（部署、崩溃、手工重启），那一轮就再没 机会写回结果，automation_runs 会永远停在 running，页面上一直显示「执行中」。 登录任务早有孤儿清理，运行记录原先漏了。 线上就是这么暴露的：部署时重建容器打断了正在跑的 ck_relogin 第 14 轮，重启清理 把 20 个在途账号标成了「重新登录中断（旧 Cookie 已保留）」（这部分是对的）， 但 run 14 本身永远卡在 running，detail 为空。 新增 cleanup_orphan_runs()，在启动清理链里调用：把所有 running 的运行记录收尾为 failed + finished_at + 「服务重启中断」。如实记为 failed 而不是 success —— 那轮 确实没跑完；被中断的账号由各自的孤儿清理负责标记并按退避重试。 新增回归测试：只有 running 的被收尾，已完成的不受影响。 （另一件事一并说明：上一次部署我报告「零中断」是验证过早 —— 检查时部署的 up -d 还没执行到。实际情况是 mysql 未被重建（达到了本次改动目的），但应用容器因镜像 变化被正常重建，因此打断了那轮重登。）
296 lines
9.9 KiB
Python
296 lines
9.9 KiB
Python
"""进程内自动化调度器。
|
||
|
||
背景：后台的日常动作原本全靠人工点（判断 CK 失效、重登、收集资料、记台账）。
|
||
自动化按「渐进上线」推进，因此：
|
||
|
||
- 调度器总开关 ``AUTOMATION_ENABLED`` **默认关闭**；
|
||
- 每个作业自己还有开关，**默认关闭**，在 ``/automation`` 页面打开；
|
||
- 作业状态全部落库（``automation_jobs`` / ``automation_runs``），进程重启不丢
|
||
调度进度，也不会因为重启而重复跑一轮。
|
||
|
||
部署是单容器 ``--workers 1``，所以一个进程内线程就够，不需要分布式锁。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
from loguru import logger
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ...database import SessionLocal
|
||
from ...models import AutomationJob, AutomationRun
|
||
|
||
# 调度轮询间隔，也是作业的运行节奏：每个 tick 跑一轮。
|
||
#
|
||
# 注意区分两个「间隔」：
|
||
# - **本常量**：作业多久跑一轮（5 分钟）；
|
||
# - **job.interval_seconds**：同一行数据多久重新检测一次（默认 6h），只用于选行 cutoff。
|
||
#
|
||
# 分开的原因：全库 12,881 条的 CK 若想「6 小时内都被查一遍」，需要约 36 条/分钟的
|
||
# 吞吐；按 5 分钟一轮、单轮 200 条算 = 40 条/分钟，刚好够。若把作业周期也设成 6h，
|
||
# 就变成 200 条/6h = 800 条/天，全库要 16 天才能过一轮，等于没巡检。
|
||
TICK_SECONDS = 300
|
||
|
||
TRUTHY = {"1", "true", "yes", "on"}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class JobSpec:
|
||
"""一个可调度的作业。``run`` 接收会话与作业行，返回本轮计数。"""
|
||
|
||
key: str
|
||
title: str
|
||
interval_seconds: int
|
||
batch_size: int
|
||
concurrency: int
|
||
run: Callable[[Session, AutomationJob], dict[str, Any]]
|
||
|
||
|
||
_REGISTRY: dict[str, JobSpec] = {}
|
||
_thread: threading.Thread | None = None
|
||
_stop_event = threading.Event()
|
||
_lock = threading.Lock()
|
||
|
||
|
||
def register_job(spec: JobSpec) -> None:
|
||
_REGISTRY[spec.key] = spec
|
||
|
||
|
||
def registered_jobs() -> dict[str, JobSpec]:
|
||
return dict(_REGISTRY)
|
||
|
||
|
||
def reset_registry() -> None:
|
||
"""清空作业注册表（测试隔离用；正常启动只调用 register_default_jobs）。"""
|
||
_REGISTRY.clear()
|
||
|
||
|
||
def automation_enabled() -> bool:
|
||
return os.getenv("AUTOMATION_ENABLED", "").strip().lower() in TRUTHY
|
||
|
||
|
||
def ensure_jobs(db: Session) -> None:
|
||
"""把注册表里的作业补进 ``automation_jobs``（幂等）。
|
||
|
||
只补缺失的行，**不覆盖已有配置** —— 运维在页面上改过的开关、间隔、批量
|
||
不能因为重启被重置回默认值。
|
||
"""
|
||
existing = {row.key for row in db.query(AutomationJob.key).all()}
|
||
for spec in _REGISTRY.values():
|
||
if spec.key in existing:
|
||
continue
|
||
db.add(
|
||
AutomationJob(
|
||
key=spec.key,
|
||
title=spec.title,
|
||
enabled=False,
|
||
interval_seconds=spec.interval_seconds,
|
||
batch_size=spec.batch_size,
|
||
concurrency=spec.concurrency,
|
||
config=None,
|
||
next_run_at=None,
|
||
last_status="",
|
||
last_message="",
|
||
updated_at=datetime.now(UTC),
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
|
||
def _as_utc(value: datetime | None) -> datetime | None:
|
||
"""DB 时间列存的是 naive UTC，读出来要补时区标记才能跟 now 比较。"""
|
||
if value is None:
|
||
return None
|
||
if value.tzinfo is None:
|
||
return value.replace(tzinfo=UTC)
|
||
return value.astimezone(UTC)
|
||
|
||
|
||
def _due(job: AutomationJob, now: datetime) -> bool:
|
||
"""到点判定：没排过时间视为到点（刚打开开关就会跑一轮）。"""
|
||
next_run_at = _as_utc(job.next_run_at)
|
||
return next_run_at is None or next_run_at <= now
|
||
|
||
|
||
def cleanup_orphan_runs(db: Session, message: str = "服务重启中断") -> int:
|
||
"""把上次进程留下的 ``running`` 运行记录收尾。
|
||
|
||
作业是在调度线程里同步跑的：进程一重启（部署、崩溃、手工重启），那轮作业就
|
||
再没机会写回结果，``automation_runs`` 会永远停在 ``running``，页面上一直显示
|
||
「执行中」。登录任务有自己的孤儿清理，运行记录原先漏了。
|
||
|
||
收尾为 ``failed`` 而不是 ``success``：那轮确实没跑完，如实记录；被中断的账号
|
||
由各自的孤儿清理负责标记，并会按退避重试。
|
||
"""
|
||
orphans = db.query(AutomationRun).filter(AutomationRun.status == "running").all()
|
||
for run in orphans:
|
||
run.status = "failed"
|
||
run.finished_at = datetime.now(UTC)
|
||
run.message = message
|
||
if orphans:
|
||
db.commit()
|
||
return len(orphans)
|
||
|
||
|
||
def run_due_jobs(db: Session, *, now: datetime | None = None) -> list[dict[str, Any]]:
|
||
"""跑一轮「到点的作业」，返回每轮结果。
|
||
|
||
抽成普通函数是为了可测：测试直接调用即可，不必启动线程或等 60 秒。
|
||
作业串行执行 —— 目前只有 CK 巡检，单轮约 20~40 秒，远小于 tick，串行
|
||
反而省掉了并发调度的复杂度。
|
||
"""
|
||
now = now or datetime.now(UTC)
|
||
outcomes: list[dict[str, Any]] = []
|
||
jobs = (
|
||
db.query(AutomationJob)
|
||
.filter(AutomationJob.enabled.is_(True))
|
||
.order_by(AutomationJob.key.asc())
|
||
.all()
|
||
)
|
||
for job in jobs:
|
||
spec = _REGISTRY.get(job.key)
|
||
if spec is None or not _due(job, now):
|
||
continue
|
||
outcomes.append(_run_one(db, job, spec))
|
||
|
||
# 到期但没注册实现的作业（例如降级回滚后残留的行）只留日志，不静默消失。
|
||
for job in jobs:
|
||
if job.key not in _REGISTRY:
|
||
logger.warning("自动化作业 {} 没有注册实现，已跳过", job.key)
|
||
return outcomes
|
||
|
||
|
||
def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]:
|
||
started = datetime.now(UTC)
|
||
run = AutomationRun(job_key=job.key, started_at=started, status="running")
|
||
db.add(run)
|
||
db.commit()
|
||
|
||
counters: dict[str, Any] = {}
|
||
status = "success"
|
||
message = ""
|
||
try:
|
||
counters = spec.run(db, job) or {}
|
||
message = str(counters.pop("message", "") or "")
|
||
except Exception as exc: # noqa: BLE001 作业边界要兜住任何异常，不能拖垮调度线程
|
||
db.rollback()
|
||
status = "failed"
|
||
message = f"{type(exc).__name__}: {exc}"[:500]
|
||
logger.exception("自动化作业 {} 执行失败", job.key)
|
||
|
||
finished = datetime.now(UTC)
|
||
run.finished_at = finished
|
||
run.status = status
|
||
run.message = message
|
||
run.scanned = int(counters.get("scanned", 0) or 0)
|
||
run.valid = int(counters.get("valid", 0) or 0)
|
||
run.invalid = int(counters.get("invalid", 0) or 0)
|
||
run.relogin_started = int(counters.get("relogin_started", 0) or 0)
|
||
run.relogin_success = int(counters.get("relogin_success", 0) or 0)
|
||
run.skipped = int(counters.get("skipped", 0) or 0)
|
||
run.detail = counters.get("detail") or None
|
||
|
||
job.last_run_at = finished
|
||
# 下一轮固定在一个 tick 之后；interval_seconds 是「同一行多久重检」，
|
||
# 由作业自己在选行时用，不参与调度。
|
||
job.next_run_at = finished + timedelta(seconds=TICK_SECONDS)
|
||
job.last_status = status
|
||
job.last_message = message
|
||
job.updated_at = finished
|
||
db.commit()
|
||
|
||
logger.info(
|
||
"自动化作业 {} 完成：status={} scanned={} valid={} invalid={} skipped={}",
|
||
job.key,
|
||
status,
|
||
run.scanned,
|
||
run.valid,
|
||
run.invalid,
|
||
run.skipped,
|
||
)
|
||
return {
|
||
"job_key": job.key,
|
||
"status": status,
|
||
"message": message,
|
||
"scanned": run.scanned,
|
||
"valid": run.valid,
|
||
"invalid": run.invalid,
|
||
"skipped": run.skipped,
|
||
}
|
||
|
||
|
||
def run_job_now(db: Session, key: str) -> dict[str, Any]:
|
||
"""立刻跑一个作业，**不看 enabled 开关**。
|
||
|
||
页面上的「立即跑一次」用这个：运维需要能在正式打开开关之前先试一轮。
|
||
"""
|
||
job = db.query(AutomationJob).filter(AutomationJob.key == key).first()
|
||
spec = _REGISTRY.get(key)
|
||
if job is None or spec is None:
|
||
raise KeyError(key)
|
||
return _run_one(db, job, spec)
|
||
|
||
|
||
def _loop() -> None:
|
||
db = SessionLocal()
|
||
try:
|
||
ensure_jobs(db)
|
||
except Exception: # noqa: BLE001 初始化失败不应让线程静默退出
|
||
logger.exception("自动化作业初始化失败")
|
||
finally:
|
||
db.close()
|
||
|
||
logger.info("自动化调度器已启动，tick={}s", TICK_SECONDS)
|
||
while not _stop_event.wait(TICK_SECONDS):
|
||
db = SessionLocal()
|
||
try:
|
||
run_due_jobs(db)
|
||
except Exception: # noqa: BLE001 单轮异常不能让调度线程死掉
|
||
db.rollback()
|
||
logger.exception("自动化调度轮询异常")
|
||
finally:
|
||
db.close()
|
||
logger.info("自动化调度器已停止")
|
||
|
||
|
||
def start_scheduler() -> bool:
|
||
"""启动调度线程。返回是否真的启动了（未开启总开关时返回 False）。"""
|
||
global _thread
|
||
if not automation_enabled():
|
||
logger.info("AUTOMATION_ENABLED 未开启，自动化调度器不启动")
|
||
return False
|
||
with _lock:
|
||
if _thread is not None and _thread.is_alive():
|
||
return False
|
||
_stop_event.clear()
|
||
_thread = threading.Thread(
|
||
target=_loop, name="automation-scheduler", daemon=True
|
||
)
|
||
_thread.start()
|
||
return True
|
||
|
||
|
||
def stop_scheduler() -> None:
|
||
global _thread
|
||
with _lock:
|
||
if _thread is None:
|
||
return
|
||
_stop_event.set()
|
||
_thread.join(timeout=5)
|
||
_thread = None
|
||
|
||
|
||
def scheduler_status() -> dict[str, Any]:
|
||
return {
|
||
"enabled": automation_enabled(),
|
||
"running": _thread is not None and _thread.is_alive(),
|
||
"tick_seconds": TICK_SECONDS,
|
||
"registered": sorted(_REGISTRY),
|
||
}
|