bvandeusenandClaude Opus 5.5
2df8431de9
The same Holding as the SNMP poll, for the rest of Steward's own collectors: - Monitors: the list is read within 10 s or the last one read stands in; a start the database doesn't take is kept in memory, so a monitor isn't checked twice; each result is held and stored at the time it was checked. A stall shows on the monitors as the checks it saw, not a gap. - UniFi and Traefik: each poll's writes are held (hold_write) and run oldest first in one transaction; every reading is stamped with its scrape. - A held round the database refuses (as opposed to can't be reached) is stored alone and let go, so it can't hold up every round after it. The Prometheus scrape is left as it is: its one target, Fabled Archives, keeps its own database on the same NFS. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
193 lines
6.6 KiB
Python
193 lines
6.6 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 == []
|
|
|
|
|
|
class _HungSession(_Session):
|
|
"""The storage under Postgres has stalled: nothing answers until `gate`."""
|
|
|
|
def __init__(self, store, gate):
|
|
super().__init__(store)
|
|
self.gate = gate
|
|
|
|
async def __aenter__(self):
|
|
await self.gate.wait()
|
|
return self
|
|
|
|
|
|
async def test_checks_go_on_while_the_database_hangs_and_are_stored_at_their_own_time(monkeypatch):
|
|
async def fake_run(m):
|
|
return dict(sched._BLANK, is_up=False, error_msg="502 Bad Gateway")
|
|
|
|
monkeypatch.setattr(sched, "run_monitor", fake_run)
|
|
monkeypatch.setattr(sched, "MetricBatch", _Batch)
|
|
monkeypatch.setattr(sched, "MONITORS_READ_S", 0.05)
|
|
monkeypatch.setattr(sched, "STORE_WAIT_S", 0.05)
|
|
store, gate = _Store([_m("app")]), asyncio.Event()
|
|
app = NS(config={"MONITORS_POLL_INTERVAL": 60}, db_sessionmaker=lambda: _HungSession(store, gate))
|
|
sched._memory(app).monitors = list(store.monitors) # as last read, before the stall
|
|
|
|
await sched.run_due_monitors(app) # the list and the stamp time out
|
|
await _settle(app) # the result is held, the monitor freed
|
|
assert store.added == [] and sched._state(app)[0] == set()
|
|
await sched.run_due_monitors(app)
|
|
await _settle(app)
|
|
assert store.added == [] # its start was kept in memory: not due again
|
|
|
|
checked_by = datetime.now(timezone.utc)
|
|
gate.set() # the database answers
|
|
await app._holdings["Monitor checks"].flush(1.0)
|
|
[res] = store.added
|
|
assert res.is_up is False and res.checked_at < checked_by # stored at the time it was checked
|