From edea8482b57cc50add9b5adc00efd5571865c376 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 18:31:07 -0400 Subject: [PATCH] fix(monitors): run each monitor on its own interval, from its scheduled start (#5151) 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 --- steward/app.py | 6 +- steward/monitors/scheduler.py | 96 ++++++++++++----- tests/core/test_monitor_schedule.py | 154 ++++++++++++++++++++++++++++ 3 files changed, 226 insertions(+), 30 deletions(-) create mode 100644 tests/core/test_monitor_schedule.py diff --git a/steward/app.py b/steward/app.py index f9c7789..1a26c6a 100644 --- a/steward/app.py +++ b/steward/app.py @@ -367,7 +367,7 @@ def create_app( def _register_core_tasks(app: Quart) -> None: from .core.scheduler import ScheduledTask - poll_interval = app.config.get("MONITORS_POLL_INTERVAL", 60) + from .monitors.scheduler import DISPATCH_TICK_SECONDS as MONITOR_DISPATCH_TICK cleanup_interval = 3600 # hourly async def run_monitor_checks(): @@ -418,10 +418,12 @@ def _register_core_tasks(app: Quart) -> None: interval_seconds=60, run_on_startup=False, ), + # Ticks often and only starts checks: each monitor decides whether it + # is due from its own interval (steward/monitors/scheduler.py). ScheduledTask( name="monitor_check", coro_factory=run_monitor_checks, - interval_seconds=poll_interval, + interval_seconds=MONITOR_DISPATCH_TICK, run_on_startup=True, ), ScheduledTask( diff --git a/steward/monitors/scheduler.py b/steward/monitors/scheduler.py index 39ed50f..5c8cd0e 100644 --- a/steward/monitors/scheduler.py +++ b/steward/monitors/scheduler.py @@ -1,52 +1,92 @@ -"""Single scheduled task that runs every due Monitor of any type.""" +"""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 +from sqlalchemy import select, update from steward.core.alerts import MetricBatch from steward.models.monitors import Monitor, MonitorResult -from .runner import run_monitor +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: - """Check every enabled Monitor whose interval has elapsed, persist results. + """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), mirroring how the scheduler fires - on the global cadence but each monitor decides if it is actually due. - """ + 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() - due: list[Monitor] = [] - for m in monitors: - interval = m.check_interval_seconds or global_interval - if m.last_checked_at is None or (now - m.last_checked_at).total_seconds() >= interval: - due.append(m) - if not due: - return + 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) - pairs = await asyncio.gather( - *[run_monitor(m) for m in due], return_exceptions=True - ) - check_time = datetime.now(timezone.utc) - async with app.db_sessionmaker() as session: - async with session.begin(): - for monitor, res in zip(due, pairs): - if isinstance(res, Exception): - logger.error("Monitor %r check errored: %s", monitor.name, res) - continue +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, @@ -58,10 +98,6 @@ async def run_due_monitors(app) -> None: tls_expires_at=res["tls_expires_at"], error_msg=res["error_msg"], )) - db_monitor = await session.get(Monitor, monitor.id) - if db_monitor: - db_monitor.last_checked_at = check_time - # Alert pipeline — keyed by monitor type (source) + name. batch = MetricBatch() batch.add( @@ -76,3 +112,7 @@ async def run_due_monitors(app) -> None: value=res["response_ms"], ) await batch.record(session) + except Exception: + logger.exception("Monitor %r check errored", monitor.name) + finally: + in_flight.discard(monitor.id) diff --git a/tests/core/test_monitor_schedule.py b/tests/core/test_monitor_schedule.py new file mode 100644 index 0000000..94dffc6 --- /dev/null +++ b/tests/core/test_monitor_schedule.py @@ -0,0 +1,154 @@ +"""Monitors run on their own interval, not the dispatcher's: due is measured +from the scheduled start, each check runs on its own, one still running is not +started twice, and a check that never finishes is recorded down (#5151).""" +import asyncio +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +import pytest + +from steward.monitors import scheduler as sched +from steward.monitors.scheduler import DISPATCH_TICK_SECONDS, is_due + +NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc) + + +def _m(mid="m1", interval=0, checked_ago=None, name=None): + return NS(id=mid, name=name or mid, type="http", enabled=True, check_interval_seconds=interval, + last_checked_at=None if checked_ago is None else NOW - timedelta(seconds=checked_ago)) + + +@pytest.mark.parametrize("monitor, due", [ + (_m(), True), # never checked + (_m(checked_ago=60), True), # exactly one interval (global 60) + (_m(checked_ago=60 - DISPATCH_TICK_SECONDS / 2), True), # within half a tick: not pushed a tick late + (_m(checked_ago=55), False), + (_m(interval=30, checked_ago=30), True), # its own interval, not the global one + (_m(interval=30, checked_ago=20), False), + (_m(interval=300, checked_ago=120), False), + (_m(checked_ago=86400), True), # long overdue: due, and runs once +]) +def test_is_due(monitor, due): + assert is_due(monitor, NOW, 60) is due + + +class _Store: + def __init__(self, monitors): + self.monitors, self.stamped, self.added = monitors, [], [] + + +class _Session: + def __init__(self, store): + self.store = store + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + def begin(self): + return self + + async def execute(self, stmt): + if getattr(stmt, "is_update", False): + # Like the real stamp: every monitor in this fake was due. + self.store.stamped.append(stmt) + for m in self.store.monitors: + m.last_checked_at = datetime.now(timezone.utc) + return NS() + return NS(scalars=lambda: iter(self.store.monitors)) + + async def commit(self): + pass + + def add(self, obj): + self.store.added.append(obj) + + +class _Batch: + def add(self, **kw): + pass + + async def record(self, session): + pass + + +def _app(store): + return NS(config={"MONITORS_POLL_INTERVAL": 60}, db_sessionmaker=lambda: _Session(store)) + + +async def _settle(app): + _, tasks = sched._state(app) + if tasks: + await asyncio.gather(*list(tasks)) + + +async def test_a_slow_check_holds_up_nobody_and_is_not_started_twice(monkeypatch): + release = asyncio.Event() + started = [] + + async def fake_run(m): + started.append(m.id) + if m.id == "slow": + await release.wait() + return dict(sched._BLANK, is_up=True) + + monkeypatch.setattr(sched, "run_monitor", fake_run) + monkeypatch.setattr(sched, "MetricBatch", _Batch) + store = _Store([_m("slow"), _m("fast")]) + app = _app(store) + + await sched.run_due_monitors(app) + await asyncio.sleep(0) + await asyncio.sleep(0) + assert sorted(started) == ["fast", "slow"] + assert [r.monitor_id for r in store.added] == ["fast"] # recorded while slow still runs + + await sched.run_due_monitors(app) # next tick: slow is in flight + await asyncio.sleep(0) + assert started.count("slow") == 1 + + release.set() + await _settle(app) + assert sorted(r.monitor_id for r in store.added) == ["fast", "slow"] + assert sched._state(app)[0] == set() + assert len(store.stamped) == 1 # stamped once, when both were due + + +async def test_a_check_that_raises_frees_its_monitor(monkeypatch): + async def boom(m): + raise RuntimeError("probe broke") + + monkeypatch.setattr(sched, "run_monitor", boom) + monkeypatch.setattr(sched, "MetricBatch", _Batch) + store = _Store([_m("m1")]) + app = _app(store) + await sched.run_due_monitors(app) + await _settle(app) + assert store.added == [] and sched._state(app)[0] == set() + + +async def test_a_check_that_never_finishes_is_recorded_down(monkeypatch): + async def hang(m): + await asyncio.sleep(3600) + + monkeypatch.setattr(sched, "run_monitor", hang) + monkeypatch.setattr(sched, "MetricBatch", _Batch) + monkeypatch.setattr(sched, "CHECK_DEADLINE_SECONDS", 0.01) + store = _Store([_m("m1")]) + app = _app(store) + await sched.run_due_monitors(app) + await _settle(app) + [res] = store.added + assert res.is_up is False and "did not finish" in res.error_msg + assert sched._state(app)[0] == set() + + +async def test_nothing_due_stamps_nothing(monkeypatch): + m = _m("m1") + m.last_checked_at = datetime.now(timezone.utc) - timedelta(seconds=10) + store = _Store([m]) + app = _app(store) + await sched.run_due_monitors(app) + assert store.stamped == [] and store.added == []