From 41fc8073dbdc8a06a73dfe6e8cc3852a9187443f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 17 Oct 2026 09:36:47 -0400 Subject: [PATCH] feat: readings kept through a storage stall; an agent's backlog arrives in order (#5995) On 2026-10-17 the NFS under Postgres stalled for ~15 minutes. Two things went wrong for the readings that would have explained it: - The SNMP poll stored before it polled again, so while the database hung it stopped polling. It now collects on schedule and hands each round to a Holding (steward/core/holding.py): one writer stores what is held, oldest first, each reading at the time it was collected. Bounded at two hours of polls, with a 30-minute deadline on a store; the device list read falls back to the last one read after 10 s. - "Mount not answering" (10 checks in a row) went pending on Docker02 and Docker03 and never fired: the agent sent each post's backlog with its newest, healthy sample, and that sample ended every post's run. Agent 1.22.0 queues the new sample behind the backlog, so Steward receives a host's samples in the order they were taken. Co-Authored-By: Claude Opus 5.5 --- docs/plugins/host-agent-design.md | 4 +- plugins/host_agent/agent.py | 22 +- plugins/snmp/scheduler.py | 206 +++++++++++++----- steward/core/holding.py | 99 +++++++++ tests/core/test_holding.py | 64 ++++++ .../host_agent/test_agent_build_payload.py | 14 +- tests/plugins/snmp/test_scheduler.py | 55 +++++ 7 files changed, 387 insertions(+), 77 deletions(-) create mode 100644 steward/core/holding.py create mode 100644 tests/core/test_holding.py create mode 100644 tests/plugins/snmp/test_scheduler.py diff --git a/docs/plugins/host-agent-design.md b/docs/plugins/host-agent-design.md index dcacb64..b5c4d83 100644 --- a/docs/plugins/host-agent-design.md +++ b/docs/plugins/host-agent-design.md @@ -327,7 +327,7 @@ The agent's self-reported hostname in the payload is advisory. The real identity ### Failure behavior - POST fails → push sample to ring buffer (max 20), exponential backoff `30s → 60s → 120s → 300s cap`, resets on first success. -- On next success → flush all buffered samples in a single batched POST. +- On next success → flush the buffer oldest first, five samples a POST (`MAX_SAMPLES_PER_POST`). The new sample queues behind the backlog, so Steward always receives a host's samples in the order they were taken (1.22.0, #5995): a rule counting checks in a row then sees a stall's run of breaching checks unbroken. Before, each post carried the newest sample too, and after the 2026-10-17 NFS stall its healthy reading ended every post's run, so "Mount not answering" (10 in a row) never fired. - Ring buffer full → drop oldest, keep newest. Logged at DEBUG (expected during extended outages). - Malformed payload rejection (400) → drop the sample (don't re-buffer, it'll just be rejected again). Indicates agent/server version skew. - Token rejection (401) → log ERROR with remediation hint, continue retry loop at backoff cadence. Admin fixes via UI + conf file edit; no restart needed. @@ -461,7 +461,7 @@ Content-Type: application/json ### Design points -- **`samples` is always a list**, even for a single sample. This is how the ring buffer drains after a failure — agent POSTs `[current, ...buffered]` in a single request. One-path server code. +- **`samples` is always a list**, even for a single sample. This is how the ring buffer drains after a failure — agent POSTs the oldest buffered samples, with the current one only once the backlog is through. One-path server code. - **`ts` is agent-reported, ISO-8601 UTC.** Server trusts it for `PluginMetric.recorded_at`. Host clocks drift but NTP is reliable enough for our purposes. - **If `|ts - now| > 5 minutes`**, server logs a WARN but still accepts. Don't reject, don't lose data over clock skew. - **`metadata` is sent on every POST**, not just on change. Server-side diff detects actual changes and only writes on change. Cost per POST is one dict — negligible. Benefit: server can cleanly detect agent restarts. diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index ce36d80..506c638 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -22,7 +22,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor, wait as wait_futures from datetime import datetime, timezone -AGENT_VERSION = "1.21.0" +AGENT_VERSION = "1.22.0" # Default path to the local Docker Engine socket. Overridable via the # `docker_socket` config key; collection is silently skipped if it's absent or @@ -1956,10 +1956,16 @@ def _drop_logs(sample: dict) -> dict: return sample -def split_backlog(buffered: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple: - """(the buffered samples to send with this one, the ones held for later): - the oldest limit-1, leaving room for the new sample.""" - return buffered[:limit - 1], buffered[limit - 1:] +def split_backlog(queued: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple: + """(the samples to send in this post, the ones held for later): the oldest + `limit` of the backlog with the new sample queued behind it. + + The new sample waits its turn. Sent ahead of the backlog it reached Steward + before older samples did, and a rule counting checks in a row saw each + post's run of stalled checks ended by the newest, healthy one: after the + 2026-10-17 NFS stall "Mount not answering" (10 in a row) never fired + (#5995).""" + return queued[:limit], queued[limit:] def next_backoff(current: int) -> int: @@ -2074,10 +2080,10 @@ def main_loop(conf_path: str) -> int: docker_logs_enabled=docker_logs_enabled, docker_log_exclude=docker_log_exclude) buffered = buffer.drain() - send, held = split_backlog(buffered) + send, held = split_backlog(buffered + [sample]) maint_events, maint_keys = maintenance_results(maint_seen) payload = build_payload( - samples=send + [sample], + samples=send, hostname=hostname, metadata=metadata, events=events + ([lost_contact_event(lost_since)] if lost_since else []) @@ -2098,7 +2104,7 @@ def main_loop(conf_path: str) -> int: rate_state["updates_due"] = True for s in held: buffer.push(s) - if send: + if buffered: _log("INFO", f"flushed {len(send)} buffered samples" + (f", {len(held)} still held" if held else "")) backoff = 0 diff --git a/plugins/snmp/scheduler.py b/plugins/snmp/scheduler.py index 6291e9e..fe349e1 100644 --- a/plugins/snmp/scheduler.py +++ b/plugins/snmp/scheduler.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio import logging +from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone +from types import SimpleNamespace from typing import TYPE_CHECKING from sqlalchemy import select @@ -47,15 +49,75 @@ def make_poll_task(app: "Quart") -> ScheduledTask: ) -async def _do_poll(app: "Quart", poll=None, walk=None) -> None: - """Poll every device at once, then write each one's outcome to its row - and its answers to the metric tables. A device whose vendor tables are - due is walked first, so a new disk is read in the same poll. `poll` and - `walk` stand in for poller.poll_device and poller.walk_tables in tests.""" - global RATES - from steward.core.alerts import MetricBatch +# Polls collected but not yet stored, written one store at a time (#5995): a +# stall of the storage under Postgres delays them instead of losing them. Two +# hours at the default one-minute poll. +HELD_POLLS = 120 +# How long the device list may take to read before the last one read stands +# in, so a stalled database doesn't stop the polling it is about to explain. +DEVICES_READ_S = 10.0 +# How long a poll waits for its own store before collecting on. +STORE_WAIT_S = 20.0 +# Each device as last read: plain copies, so a poll can go on without the +# database, and a walk's rows found meanwhile can be kept on them. +DEVICES: list | None = None + + +@dataclass +class Polled: + """One device's poll, collected and waiting to be stored.""" + at: datetime + device_id: str + name: str + failed: str | None = None # the poll raised: why, in words + error: str | None = None + scan: object = None # a TableScan when the tables were walked + answered: bool = False + missing: list = field(default_factory=list) + texts: dict = field(default_factory=dict) + values: list = field(default_factory=list) # (metric, value) + + +def _snapshot(device) -> SimpleNamespace: + from sqlalchemy import inspect as sa_inspect + return SimpleNamespace(**{a.key: getattr(device, a.key) + for a in sa_inspect(type(device)).column_attrs}) + + +async def _devices(app: "Quart") -> list: + """The devices to poll: read now, or as last read when the database + doesn't answer within DEVICES_READ_S.""" + global DEVICES from .models import SnmpDevice + + async def read() -> list: + async with app.db_sessionmaker() as session: + return [_snapshot(d) for d in (await session.execute(select(SnmpDevice))).scalars()] + + try: + DEVICES = await asyncio.wait_for(read(), DEVICES_READ_S) + except Exception as exc: + if DEVICES is None: + logger.warning("SNMP devices not read (%s); nothing to poll yet", exc or type(exc).__name__) + return [] + logger.warning("SNMP devices not read within %.0f s (%s); polling the %d last read", + DEVICES_READ_S, exc or type(exc).__name__, len(DEVICES)) + return DEVICES + + +async def _do_poll(app: "Quart", poll=None, walk=None) -> None: + """Poll every device at once, hold what each answered, and store it: the + device's outcome on its row, its answers in the metric tables, each at + the time it was collected. A device whose vendor tables are due is walked + first, so a new disk is read in the same poll. When the database doesn't + take the store within STORE_WAIT_S the poll returns and the next one + collects on; what is held is stored, oldest first, once it does (#5995). + `poll` and `walk` stand in for poller.poll_device and poller.walk_tables + in tests.""" + global RATES + from steward.core.holding import holding + from .poller import poll_device, walk_tables from .readings import state_of, tables_for from .service import RateKeeper, credentials, readings_for @@ -64,67 +126,91 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None: walk = walk or walk_tables if RATES is None: RATES = RateKeeper() + held = holding(app, "SNMP poll", store=lambda rounds: _store(app, rounds), limit=HELD_POLLS) - async with app.db_sessionmaker() as session: - devices = (await session.execute(select(SnmpDevice))).scalars().all() - if not devices: - return + devices = await _devices(app) + if devices: + started = datetime.now(timezone.utc) + scans, asked = {}, {} - started = datetime.now(timezone.utc) - scans, asked = {}, {} + async def one(d): + creds = credentials(d) + if tables_due(d, started): + scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or [])) + found = scans.get(d.id) + asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None) + return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]]) - async def one(d): - creds = credentials(d) - if tables_due(d, started): - scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or [])) - found = scans.get(d.id) - asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None) - return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]]) + results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True) - results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True) + now = datetime.now(timezone.utc) + polls = [] + for device, result in zip(devices, results): + if isinstance(result, BaseException): + logger.warning("SNMP poll of %s failed: %s", device.name, result) + polls.append(Polled(now, device.id, device.name, failed=f"SNMP poll failed: {result}")) + device.last_error = polls[-1].failed + continue + p = Polled(now, device.id, device.name, error=result.error, scan=scans.get(device.id), + answered=bool(result.readings or result.texts), missing=list(result.missing)) + if p.scan is not None and p.scan.error is None: + # Kept on the copy too, so a stall doesn't walk it again every poll. + device.tables, device.tables_at = p.scan.rows, now + device.last_error = result.error + missing = set(result.missing) + for reading in asked[device.id]: + value = result.readings.get(reading.key) + if reading.unit == "state": + # Its word is shown as answered; its level is stored, + # unless Steward doesn't know the word (level None). + raw = value if value is not None else result.texts.get(reading.key) + if raw is None: + continue + p.texts[reading.key], value = state_of(reading, raw) + elif value is None: + if reading.key in result.texts: # a word where a number was asked for + missing.add(reading.key) + continue + elif reading.counter: + value = RATES.rate(device.id, reading, value, now) + else: + value = value * reading.scale + reading.offset + if value is not None: + p.values.append((reading.key, value)) + p.missing = sorted(missing) + polls.append(p) + held.hold(polls) + await held.flush(STORE_WAIT_S) + + +async def _store(app: "Quart", rounds: list[list[Polled]]) -> None: + """Write held polls, oldest first, in one transaction.""" + from steward.core.alerts import MetricBatch + + from .models import SnmpDevice - now = datetime.now(timezone.utc) async with app.db_sessionmaker() as session: async with session.begin(): batch = MetricBatch() - for device, result in zip(devices, results): - row = await session.get(SnmpDevice, device.id) - if row is None: # deleted while it was being polled - continue - row.last_polled_at = now - if isinstance(result, BaseException): - logger.warning("SNMP poll of %s failed: %s", device.name, result) - row.last_error = f"SNMP poll failed: {result}" - continue - scan = scans.get(device.id) - if scan is not None and scan.error is None: - row.tables, row.tables_at = scan.rows, now - row.last_error = result.error - missing = set(result.missing) - if result.readings or result.texts: - row.last_ok_at = now - texts = dict(row.texts or {}) if result.error else {} - for reading in asked[device.id]: - value = result.readings.get(reading.key) - if reading.unit == "state": - # Its word is shown as answered; its level is stored, - # unless Steward doesn't know the word (level None). - raw = value if value is not None else result.texts.get(reading.key) - if raw is None: - continue - texts[reading.key], value = state_of(reading, raw) - elif value is None: - if reading.key in result.texts: # a word where a number was asked for - missing.add(reading.key) + for polls in rounds: + for p in polls: + row = await session.get(SnmpDevice, p.device_id) + if row is None: # deleted since it was polled continue - elif reading.counter: - value = RATES.rate(device.id, reading, value, now) - else: - value = value * reading.scale + reading.offset - if value is None: + row.last_polled_at = p.at + if p.failed: + row.last_error = p.failed continue - batch.add(source_module="snmp", resource_name=device.name, - metric_name=reading.key, value=value, recorded_at=now) - row.missing = sorted(missing) - row.texts = texts + if p.scan is not None and p.scan.error is None: + row.tables, row.tables_at = p.scan.rows, p.at + row.last_error = p.error + if p.answered: + row.last_ok_at = p.at + # A failed poll keeps the words it had; one that answered says them afresh. + texts = dict(row.texts or {}) if p.error else {} + texts.update(p.texts) + row.missing, row.texts = p.missing, texts + for key, value in p.values: + batch.add(source_module="snmp", resource_name=p.name, + metric_name=key, value=value, recorded_at=p.at) await batch.record(session) diff --git a/steward/core/holding.py b/steward/core/holding.py new file mode 100644 index 0000000..7ca9da0 --- /dev/null +++ b/steward/core/holding.py @@ -0,0 +1,99 @@ +"""Collected readings held in memory until the database takes them (#5995). + +Steward's Postgres sits on the NFS storage it watches. When that storage +stalls, every write waits, and a collector that stores what it found before +collecting again stops collecting for as long as the stall lasts: on +2026-10-17 the NAS readings went blind for the fifteen minutes they mattered. + +A collector built on a Holding collects on schedule whatever the database is +doing, and hands each round to the holding. One writer at a time stores what +is held, oldest first, each item at the time it was collected, so a stall +delays the readings instead of losing them, and alert rules count them in the +order they happened. + + held = holding(app, "SNMP poll", store=_store, limit=120) + held.hold(round) # one collected round + await held.flush(wait_s=20) # store it, or leave it to the writer + +The holding is bounded: past `limit` rounds the oldest is dropped, and that +is logged with how many have gone. A store has `deadline_s`: generous, past +any stall seen so far, so a connection that will never answer is given up +and the rounds stay held for a fresh one rather than waiting forever. +""" +from __future__ import annotations + +import asyncio +import logging +from collections import deque +from typing import Awaitable, Callable + +logger = logging.getLogger(__name__) + + +class Holding: + def __init__(self, name: str, store: Callable[[list], Awaitable[None]], limit: int, + deadline_s: float = 1800.0) -> None: + self.name = name + self.store = store + self.limit = limit + self.deadline_s = deadline_s + self._held: deque[tuple[int, object]] = deque() + self._seq = 0 + self._writer: asyncio.Task | None = None + self.dropped = 0 + + def __len__(self) -> int: + return len(self._held) + + def hold(self, item) -> None: + """Hold one collected round until it is stored.""" + self._seq += 1 + self._held.append((self._seq, item)) + while len(self._held) > self.limit: + self._held.popleft() + self.dropped += 1 + logger.warning("%s: the database hasn't taken a round in %d; dropped the oldest " + "(%d dropped so far)", self.name, self.limit, self.dropped) + + async def flush(self, wait_s: float) -> bool: + """Start the writer if it isn't running, and wait up to `wait_s` for + it. True when nothing is left held. A writer still waiting on the + database is left to finish: the next round joins what it stores next.""" + writer = self._writer + if writer is None or writer.done() or writer.get_loop() is not asyncio.get_running_loop(): + if not self._held: + return True + writer = self._writer = asyncio.create_task(self._write()) + done, _ = await asyncio.wait({writer}, timeout=wait_s) + if not done: + logger.warning("%s: the database hasn't taken %d held round(s) in %.0f s; " + "collecting on", self.name, len(self._held), wait_s) + return not self._held + + async def _write(self) -> None: + while self._held: + batch = list(self._held) + try: + await asyncio.wait_for(self.store([item for _, item in batch]), self.deadline_s) + except TimeoutError: + logger.error("%s: storing %d held round(s) took over %.0f s; given up, they stay " + "held (the database may still finish it)", self.name, len(batch), + self.deadline_s) + return + except Exception: + logger.exception("%s: storing %d held round(s) failed; they stay held", + self.name, len(batch)) + return + last = batch[-1][0] + while self._held and self._held[0][0] <= last: + self._held.popleft() + + +def holding(app, name: str, store: Callable[[list], Awaitable[None]], limit: int) -> Holding: + """The app's Holding for one collector, made on first use.""" + holdings = getattr(app, "_holdings", None) + if holdings is None: + holdings = app._holdings = {} + if name not in holdings: + holdings[name] = Holding(name, store, limit) + return holdings[name] diff --git a/tests/core/test_holding.py b/tests/core/test_holding.py new file mode 100644 index 0000000..013d50f --- /dev/null +++ b/tests/core/test_holding.py @@ -0,0 +1,64 @@ +"""Collected rounds held in memory until the database takes them (#5995): +stored oldest first, one writer at a time, bounded, and kept through a +store that fails or never answers.""" +import asyncio +from types import SimpleNamespace as NS + +from steward.core.holding import Holding, holding + + +def test_rounds_held_while_the_store_hangs_are_stored_in_order_once_it_answers(): + async def main(): + gate, stored = asyncio.Event(), [] + + async def store(rounds): + await gate.wait() + stored.append(list(rounds)) + + held = Holding("t", store, limit=10) + held.hold("a") + assert await held.flush(0.05) is False # the store hangs: the caller goes on + held.hold("b") + assert await held.flush(0.05) is False and len(held) == 2 + gate.set() + assert await held.flush(1.0) is True and len(held) == 0 + return stored + + assert asyncio.run(main()) == [["a"], ["b"]] # "b" waited for the store in flight + + +def test_past_its_limit_the_oldest_round_goes_and_is_counted(): + held = Holding("t", store=None, limit=2) + for r in ("a", "b", "c"): + held.hold(r) + assert len(held) == 2 and held.dropped == 1 and [i for _, i in held._held] == ["b", "c"] + + +def test_a_store_that_fails_or_overruns_its_deadline_keeps_the_rounds_held(): + async def main(): + calls = [] + + async def failing(rounds): + calls.append(list(rounds)) + raise RuntimeError("database refused") + + held = Holding("t", failing, limit=10) + held.hold("a") + assert await held.flush(1.0) is False and len(held) == 1 + + async def never(rounds): + await asyncio.sleep(3600) + + stuck = Holding("t", never, limit=10, deadline_s=0.05) + stuck.hold("a") + assert await stuck.flush(1.0) is False and len(stuck) == 1 # given up, not lost + return calls + + assert asyncio.run(main()) == [["a"]] + + +def test_an_app_has_one_holding_per_collector(): + app = NS() + first = holding(app, "SNMP poll", store=None, limit=5) + assert holding(app, "SNMP poll", store=None, limit=5) is first + assert holding(app, "UniFi poll", store=None, limit=5) is not first diff --git a/tests/plugins/host_agent/test_agent_build_payload.py b/tests/plugins/host_agent/test_agent_build_payload.py index e4a092a..8872eb9 100644 --- a/tests/plugins/host_agent/test_agent_build_payload.py +++ b/tests/plugins/host_agent/test_agent_build_payload.py @@ -120,11 +120,11 @@ def test_backoff_schedule_stays_inside_the_stale_window(): assert next_backoff(60) == 60 # capped below Steward's 180 s stale window (#5154) -def test_a_backlog_drains_a_few_samples_a_post_oldest_first(): +def test_a_backlog_drains_a_few_samples_a_post_oldest_first_and_the_new_one_waits_its_turn(): from plugins.host_agent.agent import MAX_SAMPLES_PER_POST, split_backlog - buffered = list(range(20)) - send, held = split_backlog(buffered) - assert send == [0, 1, 2, 3] and held == list(range(4, 20)) - assert len(send) + 1 == MAX_SAMPLES_PER_POST # room for the new sample - assert split_backlog([7]) == ([7], []) - assert split_backlog([]) == ([], []) + queued = list(range(20)) + ["new"] + send, held = split_backlog(queued) + assert send == [0, 1, 2, 3, 4] and len(send) == MAX_SAMPLES_PER_POST + assert held == list(range(5, 20)) + ["new"] # never ahead of an older sample + assert split_backlog(["new"]) == (["new"], []) + assert split_backlog([7, "new"]) == ([7, "new"], []) diff --git a/tests/plugins/snmp/test_scheduler.py b/tests/plugins/snmp/test_scheduler.py new file mode 100644 index 0000000..b51e4ec --- /dev/null +++ b/tests/plugins/snmp/test_scheduler.py @@ -0,0 +1,55 @@ +"""An SNMP poll collects on schedule while the database hangs, and what it +held is stored oldest first, each at its own time, once it answers (#5995).""" +import asyncio +from contextlib import asynccontextmanager +from types import SimpleNamespace as NS + +import plugins.snmp.scheduler as scheduler +import plugins.snmp.service as service +from plugins.snmp.poller import Credentials, PollResult + + +def _device(): + return NS(id="d1", name="nas", address="10.0.0.5", port=161, version="2c", presets=["system"], + interfaces="", custom=None, tables={}, tables_at=None, last_error=None, texts={}) + + +def test_polls_go_on_while_the_database_hangs_and_are_stored_in_order_after(monkeypatch): + async def main(): + stalled = asyncio.Event() + + @asynccontextmanager + async def hung_session(): + await stalled.wait() # the storage under Postgres has stalled + yield None + + app = NS(db_sessionmaker=hung_session) + gate, stored = asyncio.Event(), [] + + async def store(app_, rounds): + await gate.wait() + stored.extend(rounds) + + uptimes = iter([100.0, 200.0]) + + async def poll(address, port, creds, oids): + return PollResult(readings={"uptime": next(uptimes)}) + + monkeypatch.setattr(scheduler, "DEVICES", [_device()]) + monkeypatch.setattr(scheduler, "DEVICES_READ_S", 0.05) + monkeypatch.setattr(scheduler, "STORE_WAIT_S", 0.05) + monkeypatch.setattr(scheduler, "_store", store) + monkeypatch.setattr(service, "credentials", lambda d: Credentials(community="public")) + scheduler.RATES = None + + await scheduler._do_poll(app, poll=poll) # the device list is the last one read + await asyncio.sleep(0.01) + await scheduler._do_poll(app, poll=poll) # the first store hasn't returned: polled anyway + gate.set() + await app._holdings["SNMP poll"].flush(1.0) + return stored + + rounds = asyncio.run(main()) + assert [[(p.name, p.values) for p in r] for r in rounds] == [ + [("nas", [("uptime", 100.0)])], [("nas", [("uptime", 200.0)])]] + assert rounds[0][0].at < rounds[1][0].at # each stored at the time it was collected