bvandeusenandClaude Opus 5.5
edea8482b5
The dispatcher ticked once per MONITORS_POLL_INTERVAL and stamped last_checked_at when the whole batch finished. With a 60 s tick and a 60 s monitor the next tick saw 59.x s elapsed and skipped, so the monitor ran every 120 s; a 30 s monitor could never run faster than the tick. - The dispatcher ticks every 5 s (DISPATCH_TICK_SECONDS) and reads the monitors and intervals fresh each tick, so a changed interval applies at the next one. - Due is measured from the scheduled start, stamped before the check runs, with half a tick of slack so tick jitter never costs a whole tick. A long-overdue monitor runs once. - Each check runs in its own task and records its own result, so a slow probe holds up nobody; a monitor still being checked is not started again, and it is freed in a finally. - A check past CHECK_DEADLINE_SECONDS (120) is recorded down with "Check did not finish within 120 s" rather than holding its monitor forever. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
119 lines
4.9 KiB
Python
119 lines
4.9 KiB
Python
"""Dispatch every due Monitor of any type, each check in its own task.
|
|
|
|
The dispatcher ticks every DISPATCH_TICK_SECONDS. Each tick it reads the
|
|
enabled monitors and their intervals fresh (a changed interval applies at the
|
|
next tick), stamps the due ones with the time they were due to start, and
|
|
starts each check on its own. A slow check (a 10 s timeout) holds up nobody
|
|
else, and a monitor still being checked is not started twice.
|
|
"""
|
|
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select, update
|
|
|
|
from steward.core.alerts import MetricBatch
|
|
from steward.models.monitors import Monitor, MonitorResult
|
|
from .runner import _BLANK, run_monitor
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# How often the dispatcher looks for due monitors. A monitor runs within one
|
|
# tick of its interval, so this is the cadence's worst-case lateness.
|
|
DISPATCH_TICK_SECONDS = 5
|
|
# A check past this is recorded down rather than left holding its monitor:
|
|
# far beyond any probe's own timeout, so it only ever catches a hang.
|
|
CHECK_DEADLINE_SECONDS = 120
|
|
|
|
|
|
def is_due(monitor, now: datetime, global_interval: int) -> bool:
|
|
"""Whether a monitor's interval has elapsed since its last scheduled start.
|
|
|
|
Half a tick of slack, so a check stamped at t isn't missed at t+interval
|
|
by a few milliseconds of tick jitter and pushed back a whole tick."""
|
|
if monitor.last_checked_at is None:
|
|
return True
|
|
interval = monitor.check_interval_seconds or global_interval
|
|
elapsed = (now - monitor.last_checked_at).total_seconds()
|
|
return elapsed >= interval - DISPATCH_TICK_SECONDS / 2
|
|
|
|
|
|
def _state(app) -> tuple[set, set]:
|
|
"""(monitor ids being checked, their tasks) for this app."""
|
|
if not hasattr(app, "_monitor_checks"):
|
|
app._monitor_checks = (set(), set())
|
|
return app._monitor_checks
|
|
|
|
|
|
async def run_due_monitors(app) -> None:
|
|
"""Start a check for every enabled Monitor that is due and not already
|
|
being checked. Returns once they are started, not finished.
|
|
|
|
Per-monitor `check_interval_seconds` overrides the global
|
|
MONITORS_POLL_INTERVAL (0 = use global)."""
|
|
global_interval = int(app.config.get("MONITORS_POLL_INTERVAL", 60))
|
|
now = datetime.now(timezone.utc)
|
|
in_flight, tasks = _state(app)
|
|
|
|
async with app.db_sessionmaker() as session:
|
|
monitors = list((await session.execute(
|
|
select(Monitor).where(Monitor.enabled.is_(True))
|
|
)).scalars())
|
|
due = [m for m in monitors if m.id not in in_flight and is_due(m, now, global_interval)]
|
|
if not due:
|
|
return
|
|
# Stamp the scheduled start before checking: the next run is due one
|
|
# interval from now, however long this check takes.
|
|
await session.execute(
|
|
update(Monitor).where(Monitor.id.in_([m.id for m in due])).values(last_checked_at=now))
|
|
await session.commit()
|
|
|
|
for m in due:
|
|
in_flight.add(m.id)
|
|
task = asyncio.create_task(_check_and_record(app, m))
|
|
tasks.add(task)
|
|
task.add_done_callback(tasks.discard)
|
|
|
|
|
|
async def _check_and_record(app, monitor: Monitor) -> None:
|
|
"""Run one monitor's check and store its result and metrics."""
|
|
in_flight, _ = _state(app)
|
|
try:
|
|
try:
|
|
res = await asyncio.wait_for(run_monitor(monitor), CHECK_DEADLINE_SECONDS)
|
|
except asyncio.TimeoutError:
|
|
res = dict(_BLANK, error_msg=f"Check did not finish within {CHECK_DEADLINE_SECONDS} s")
|
|
check_time = datetime.now(timezone.utc)
|
|
async with app.db_sessionmaker() as session:
|
|
async with session.begin():
|
|
session.add(MonitorResult(
|
|
monitor_id=monitor.id,
|
|
checked_at=check_time,
|
|
is_up=res["is_up"],
|
|
response_ms=res["response_ms"],
|
|
status_code=res["status_code"],
|
|
resolved_ip=res["resolved_ip"],
|
|
content_matched=res["content_matched"],
|
|
tls_expires_at=res["tls_expires_at"],
|
|
error_msg=res["error_msg"],
|
|
))
|
|
# Alert pipeline — keyed by monitor type (source) + name.
|
|
batch = MetricBatch()
|
|
batch.add(
|
|
source_module=monitor.type,
|
|
resource_name=monitor.name, metric_name="is_up",
|
|
value=1.0 if res["is_up"] else 0.0,
|
|
)
|
|
if res["response_ms"] is not None:
|
|
batch.add(
|
|
source_module=monitor.type,
|
|
resource_name=monitor.name, metric_name="response_ms",
|
|
value=res["response_ms"],
|
|
)
|
|
await batch.record(session)
|
|
except Exception:
|
|
logger.exception("Monitor %r check errored", monitor.name)
|
|
finally:
|
|
in_flight.discard(monitor.id)
|