yml2213
df62898d22
线上现象：同一页面上「CK 自动巡检」的最近运行是 21:45:34，而「下次运行」显示 21:50:34 且已经过去了 —— 巡检明明只要 19 秒，却总是等在那里。 根因是把心跳周期直接当成了作业周期：TICK_SECONDS=300 既是「循环多久醒一次」， 又被写进 next_run_at = 作业结束 + 300s。两者同频会系统性跳轮： 心跳 t=0 ck_health 19 秒跑完 → next = 5:19 心跳 t=5:00 差 19 秒没到点 → 整轮空转，什么都不做 心跳 t=10:00 才真正跑到 于是作业实际是每 10 分钟一轮，而不是设计的 5 分钟；只要作业耗时不是刚好 0， 就会稳定地"差一点点"被跳过。两个作业共享同一个心跳时，还会互相把对方的到期 时刻推后。 改动： - 拆成两个常量：TICK_SECONDS=30（心跳，只做「有没有到点的作业」检查，几乎不 耗资源）、JOB_INTERVAL_SECONDS=300（作业周期，跑完一轮后隔多久再跑）。 - next_run_at 改用 JOB_INTERVAL_SECONDS。 - /api/automation/jobs 的 scheduler 同时暴露两个值；启动日志一并打印。 - 页面：「单账号重检间隔」加 tooltip 说明它不决定调度频率（它只控制同一行多久 重检/重试），避免和调度周期混淆。 回归测试（3 条）：心跳必须比作业周期密至少 4 倍；周期未到时连跳多次心跳都不跑； 把 next_run_at 拨到过去后下一次心跳必须跑到。 全量 304 通过。
309 lines
11 KiB
Python
309 lines
11 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 天才能过一轮，等于没巡检。
|
||
# 心跳间隔：只用来「看一眼有没有到点的作业」，几乎不耗资源，所以设得比作业
|
||
# 周期密得多。作业周期由各自的 next_run_at（= 自己结束时刻 + JOB_INTERVAL）决定。
|
||
#
|
||
# 这里曾经把心跳周期直接当作业周期用（心跳 300s + next_run_at = 结束 + 300s），
|
||
# 两者同频会导致系统性跳轮：作业 19 秒跑完 → next = +5 分钟，而下一次心跳正好
|
||
# 在 5 分钟整，常常差几十秒没到点，于是整轮空转，实际变成每 10 分钟才跑一次。
|
||
# 表现就是「检测明明只要 19 秒，却要排队等很久」。
|
||
TICK_SECONDS = 30
|
||
|
||
# 作业自身的运行周期：跑完一轮后隔多久再跑下一轮。
|
||
JOB_INTERVAL_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=JOB_INTERVAL_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(
|
||
"自动化调度器已启动：心跳 {}s，作业周期 {}s", TICK_SECONDS, JOB_INTERVAL_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,
|
||
"job_interval_seconds": JOB_INTERVAL_SECONDS,
|
||
"registered": sorted(_REGISTRY),
|
||
}
|