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>
155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
"""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 == []
|