dekun
108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
"""进程内定时任务：离线巡检、随机任务下发、每日清算。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from apscheduler.triggers.interval import IntervalTrigger
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.config import settings
|
|
from app.database import session_scope
|
|
from app.jobs.daily_yield import run_daily_yield_cron
|
|
from app.services.heartbeat import mark_stale_nodes
|
|
from app.services.locks import acquire_lock
|
|
from app.services.tasks import dispatch_random_tasks
|
|
|
|
logger = logging.getLogger(__name__)
|
|
UTC = ZoneInfo("UTC")
|
|
_scheduler: BackgroundScheduler | None = None
|
|
|
|
|
|
def start_scheduler() -> None:
|
|
global _scheduler
|
|
if settings.disable_scheduler:
|
|
logger.info("scheduler disabled")
|
|
return
|
|
if _scheduler is not None and _scheduler.running:
|
|
return
|
|
|
|
scheduler = BackgroundScheduler(timezone=UTC, job_defaults={"coalesce": True, "max_instances": 1})
|
|
scheduler.add_job(
|
|
_guard(_offline_job),
|
|
IntervalTrigger(minutes=settings.offline_check_minutes, timezone=UTC),
|
|
id="offline_check",
|
|
replace_existing=True,
|
|
misfire_grace_time=120,
|
|
)
|
|
scheduler.add_job(
|
|
_guard(_dispatch_job),
|
|
IntervalTrigger(minutes=settings.task_dispatch_minutes, timezone=UTC),
|
|
id="task_dispatch",
|
|
replace_existing=True,
|
|
misfire_grace_time=120,
|
|
)
|
|
yield_zone = ZoneInfo(settings.yield_timezone)
|
|
scheduler.add_job(
|
|
_guard(_yield_job),
|
|
CronTrigger(hour=settings.yield_hour, minute=settings.yield_minute, timezone=yield_zone),
|
|
id="daily_yield",
|
|
replace_existing=True,
|
|
misfire_grace_time=3600,
|
|
)
|
|
scheduler.start()
|
|
_scheduler = scheduler
|
|
logger.info(
|
|
"scheduler started; daily yield at %02d:%02d %s",
|
|
settings.yield_hour,
|
|
settings.yield_minute,
|
|
settings.yield_timezone,
|
|
)
|
|
|
|
|
|
def stop_scheduler() -> None:
|
|
global _scheduler
|
|
if _scheduler is not None and _scheduler.running:
|
|
_scheduler.shutdown(wait=False)
|
|
_scheduler = None
|
|
|
|
|
|
def _guard(func):
|
|
def wrapper() -> None:
|
|
try:
|
|
func()
|
|
except Exception:
|
|
logger.exception("scheduled job failed: %s", func.__name__)
|
|
|
|
return wrapper
|
|
|
|
|
|
def _offline_job() -> None:
|
|
ttl = timedelta(seconds=max(30, settings.offline_check_minutes * 60 - 30))
|
|
if not acquire_lock("offline_check", ttl):
|
|
return
|
|
with session_scope() as session:
|
|
changed = mark_stale_nodes(session)
|
|
if changed:
|
|
logger.info("marked offline: %s", ",".join(changed))
|
|
|
|
|
|
def _dispatch_job() -> None:
|
|
ttl = timedelta(seconds=max(30, settings.task_dispatch_minutes * 60 - 30))
|
|
if not acquire_lock("task_dispatch", ttl):
|
|
return
|
|
with session_scope() as session:
|
|
tasks = dispatch_random_tasks(session)
|
|
if tasks:
|
|
logger.info("dispatched %s tasks", len(tasks))
|
|
|
|
|
|
def _yield_job() -> None:
|
|
if not acquire_lock("daily_yield", timedelta(hours=2)):
|
|
logger.info("daily yield skipped, lock held")
|
|
return
|
|
run_daily_yield_cron()
|