diff --git a/steward/app.py b/steward/app.py index e069f69..3341651 100644 --- a/steward/app.py +++ b/steward/app.py @@ -268,6 +268,10 @@ def _register_core_tasks(app: Quart) -> None: from .core.reports import check_and_send await check_and_send(app) + async def run_self_monitor(): + from .core.self_monitor import record_self_metrics + await record_self_metrics(app) + app._task_registry.extend([ ScheduledTask( name="weekly_report_check", @@ -275,6 +279,15 @@ def _register_core_tasks(app: Quart) -> None: interval_seconds=3600, run_on_startup=False, ), + # Watch our own fd usage so a descriptor leak surfaces as a metric/alert + # instead of a silent Errno 24 lockup. Cheap; plugin_metrics roll up + # hourly so the 60s cadence is storage-bounded. + ScheduledTask( + name="self_monitor", + coro_factory=run_self_monitor, + interval_seconds=60, + run_on_startup=True, + ), ScheduledTask( name="monitor_check", coro_factory=run_monitor_checks, diff --git a/steward/core/scheduler.py b/steward/core/scheduler.py index 101e016..87bd026 100644 --- a/steward/core/scheduler.py +++ b/steward/core/scheduler.py @@ -1,7 +1,7 @@ from __future__ import annotations import asyncio import logging -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Callable, Coroutine logger = logging.getLogger(__name__) @@ -15,28 +15,65 @@ class ScheduledTask: run_on_startup: bool = False +@dataclass +class _DueTracker: + """Decides which scheduled tasks are due, with a self-overlap guard. + + Pure (no asyncio, no clock of its own — `now` is passed in) so the + scheduling policy is unit-testable without timing races. A task whose prior + run is still in flight is NOT re-fired: overlapping poll runs stack up open + connections/subprocesses and amplify any per-poll resource use — the same + failure mode behind the fd-leak lockups. The skipped task is retried on the + next tick once it completes (its last_run isn't advanced while skipped). + """ + last_run: dict[str, float] = field(default_factory=dict) + in_flight: set[str] = field(default_factory=set) + + def due(self, tasks: list[ScheduledTask], now: float) -> list[ScheduledTask]: + ready: list[ScheduledTask] = [] + for task in tasks: + if now - self.last_run.get(task.name, 0) < task.interval_seconds: + continue + if task.name in self.in_flight: + logger.warning( + "Scheduled task %r still running — skipping this tick", + task.name) + continue + ready.append(task) + return ready + + def mark_started(self, task: ScheduledTask, now: float) -> None: + self.in_flight.add(task.name) + self.last_run[task.name] = now + + def mark_done(self, name: str) -> None: + self.in_flight.discard(name) + + async def start_scheduler(tasks: list[ScheduledTask]) -> None: """Run scheduled tasks in a loop. Call with asyncio.create_task().""" - last_run: dict[str, float] = {} + tracker = _DueTracker() + + def _spawn(task: ScheduledTask, now: float) -> None: + tracker.mark_started(task, now) + asyncio.create_task(_run_task(task, tracker)) for task in tasks: if task.run_on_startup: logger.info(f"Startup task: {task.name}") - asyncio.create_task(_run_task(task)) - last_run[task.name] = asyncio.get_event_loop().time() + _spawn(task, asyncio.get_event_loop().time()) while True: now = asyncio.get_event_loop().time() - for task in tasks: - last = last_run.get(task.name, 0) - if now - last >= task.interval_seconds: - asyncio.create_task(_run_task(task)) - last_run[task.name] = now + for task in tracker.due(tasks, now): + _spawn(task, now) await asyncio.sleep(1) -async def _run_task(task: ScheduledTask) -> None: +async def _run_task(task: ScheduledTask, tracker: _DueTracker) -> None: try: await task.coro_factory() except Exception: logger.exception(f"Scheduled task {task.name!r} raised an exception") + finally: + tracker.mark_done(task.name) diff --git a/steward/core/self_monitor.py b/steward/core/self_monitor.py new file mode 100644 index 0000000..cc0eadd --- /dev/null +++ b/steward/core/self_monitor.py @@ -0,0 +1,116 @@ +"""Self-monitoring: record Steward's own open file-descriptor usage. + +A leaked socket/file handle in a poll loop (see the SNMP and UniFi fd-leak +issues) used to fail silently until the process hit its fd ceiling and +`socket.accept()` started raising `OSError: [Errno 24] Too many open files` — +taking the whole app down with no early warning. + +This turns that failure mode into an observable signal. Steward monitors other +things; it should monitor itself. We record two metrics each tick under +`source_module="steward"`: + + • ``open_fds`` — raw count of open descriptors + • ``open_fds_pct`` — that count as a percentage of the soft RLIMIT_NOFILE + +Both flow through the normal alert pipeline, so the operator can attach an alert +rule to either via the existing alert-rules UI. As a zero-config floor we also +log a WARNING once usage crosses ``FD_WARN_PCT`` — a leak becomes visible even +before any rule is set up. +""" +from __future__ import annotations +import logging +import os + +logger = logging.getLogger(__name__) + +# Stdlib-only fd counting via /proc keeps this dependency-free. resource is +# POSIX-only but always present on the Linux runtime image; guard anyway so +# imports never explode on a dev machine. +try: + import resource +except ImportError: # pragma: no cover - non-POSIX + resource = None # type: ignore[assignment] + +# Warn (without needing a configured alert rule) once we're using this fraction +# of the soft fd limit. 80% leaves headroom to act before accepts start failing. +FD_WARN_PCT = 80.0 + +_warned = False # de-dupe the WARNING so a sustained leak doesn't spam the log + + +def count_open_fds() -> int | None: + """Number of open file descriptors for this process, or None if unknown. + + Reads ``/proc/self/fd`` (Linux). Returns None where /proc isn't available + (e.g. a macOS dev box) so callers degrade to a no-op rather than guessing. + """ + try: + return len(os.listdir("/proc/self/fd")) + except OSError: + return None + + +def fd_soft_limit() -> int | None: + """Soft RLIMIT_NOFILE for this process, or None if it can't be read. + + None when the limit is unknown or 'unlimited' (RLIM_INFINITY) — a percentage + against an unbounded ceiling is meaningless, so we skip the pct metric then. + """ + if resource is None: + return None + try: + soft, _hard = resource.getrlimit(resource.RLIMIT_NOFILE) + except (ValueError, OSError): + return None + if soft <= 0 or soft == resource.RLIM_INFINITY: + return None + return soft + + +async def record_self_metrics(app) -> None: + """Record open-fd usage as Steward's own metrics; warn past the floor. + + No-op (logged at debug) when fd accounting isn't available on this platform, + so it's safe to schedule unconditionally. + """ + global _warned + + fds = count_open_fds() + if fds is None: + logger.debug("self_monitor: /proc/self/fd unavailable — skipping fd metrics") + return + + from .alerts import record_metric + + soft = fd_soft_limit() + pct = (fds / soft * 100.0) if soft else None + + async with app.db_sessionmaker() as session: + async with session.begin(): + await record_metric( + session=session, + source_module="steward", + resource_name="process", + metric_name="open_fds", + value=float(fds), + ) + if pct is not None: + await record_metric( + session=session, + source_module="steward", + resource_name="process", + metric_name="open_fds_pct", + value=pct, + ) + + if pct is not None and pct >= FD_WARN_PCT: + if not _warned: + logger.warning( + "Open file descriptors at %.0f%% of the soft limit (%d/%d) — " + "possible descriptor leak; the app will stop accepting " + "connections if this reaches 100%%.", pct, fds, soft) + _warned = True + else: + _warned = False # recovered — re-arm the warning for the next breach + + logger.debug("self_monitor: open_fds=%d soft_limit=%s pct=%s", fds, soft, pct) diff --git a/tests/core/test_fd_stability.py b/tests/core/test_fd_stability.py new file mode 100644 index 0000000..30d1841 --- /dev/null +++ b/tests/core/test_fd_stability.py @@ -0,0 +1,46 @@ +"""Real-fd regression canary. + +Drives a real socket code path (the TCP reachability probe) in a tight loop and +asserts the process's open-fd count doesn't grow. If someone reintroduces a +socket leak in the probe path — the class of bug behind the Errno 24 lockups — +this fails in CI instead of in production. Uses real descriptors, not fakes, so +the assertion has teeth. +""" +import asyncio +import os + +import pytest + +from steward.monitors.ping import tcp_check + +_ITERATIONS = 100 + + +def _fd_count() -> int | None: + try: + return len(os.listdir("/proc/self/fd")) + except OSError: + return None + + +async def _hammer_tcp_check() -> None: + # 127.0.0.1:1 has no listener → connection refused immediately. Each call must + # fully release its socket; a leak would add ~one fd per iteration. + for _ in range(_ITERATIONS): + await tcp_check("127.0.0.1", 1) + + +def test_tcp_check_does_not_leak_fds(): + if _fd_count() is None: + pytest.skip("/proc/self/fd unavailable on this platform") + + asyncio.run(_hammer_tcp_check()) # warm up (lazy imports, caches) + before = _fd_count() + asyncio.run(_hammer_tcp_check()) + after = _fd_count() + + # Small slack for interpreter-internal fds; a genuine leak over 100 iterations + # would be far larger than this. + assert after - before <= 5, ( + f"open fds grew {before}->{after} over {_ITERATIONS} tcp_check calls " + "— possible socket leak") diff --git a/tests/core/test_scheduler_overlap.py b/tests/core/test_scheduler_overlap.py new file mode 100644 index 0000000..23e00a3 --- /dev/null +++ b/tests/core/test_scheduler_overlap.py @@ -0,0 +1,58 @@ +"""Unit tests for the scheduler's poll-overlap guard (_DueTracker). + +The tracker is pure (clock passed in) so we can assert the overlap policy +deterministically: a task whose prior run is still in flight is never re-fired, +and a skipped task isn't penalised — it runs on the next tick once it finishes. +This is the rail that stops a hung poll from stacking overlapping runs (which +would amplify any per-poll resource/fd use). +""" +from steward.core.scheduler import ScheduledTask, _DueTracker + + +def _task(name: str, interval: int) -> ScheduledTask: + return ScheduledTask(name=name, coro_factory=lambda: None, interval_seconds=interval) + + +def test_due_only_after_interval_elapses(): + t = _task("a", 60) + tr = _DueTracker() + assert tr.due([t], now=59) == [] # 59 < 60 + assert tr.due([t], now=60) == [t] # interval elapsed + + +def test_in_flight_task_is_skipped_even_when_due(): + t = _task("a", 0) # due every tick + tr = _DueTracker() + tr.mark_started(t, now=0) + assert tr.due([t], now=100) == [] # still running → skipped + tr.mark_done("a") + assert tr.due([t], now=100) == [t] # finished → eligible again + + +def test_skip_does_not_advance_last_run_so_it_retries(): + t = _task("a", 10) + tr = _DueTracker() + tr.mark_started(t, now=0) + assert tr.due([t], now=100) == [] # due but in flight + assert tr.last_run["a"] == 0 # not advanced by the skip + tr.mark_done("a") + assert tr.due([t], now=100) == [t] # retried promptly after completion + + +def test_mark_started_advances_last_run_and_marks_in_flight(): + t = _task("a", 10) + tr = _DueTracker() + tr.mark_started(t, now=50) + assert tr.last_run["a"] == 50 + assert tr.in_flight == {"a"} + assert tr.due([t], now=55) == [] # 5 < 10 (not yet due) + assert tr.due([t], now=61) == [] # due by interval, but still in flight + tr.mark_done("a") + assert tr.due([t], now=61) == [t] + + +def test_independent_tasks_do_not_block_each_other(): + a, b = _task("a", 0), _task("b", 0) + tr = _DueTracker() + tr.mark_started(a, now=0) # a hangs + assert tr.due([a, b], now=10) == [b] # b still fires diff --git a/tests/core/test_self_monitor.py b/tests/core/test_self_monitor.py new file mode 100644 index 0000000..715b3fa --- /dev/null +++ b/tests/core/test_self_monitor.py @@ -0,0 +1,115 @@ +"""Unit tests for the self-fd watchdog (no DB, no network). + +Exercises the fd accounting, the percentage/warning floor, and that the +recorder degrades to a no-op where fd accounting isn't available. The DB session +and the alert pipeline are faked — we only assert that the right metrics get +handed to record_metric. +""" +import asyncio +import logging + +import steward.core.alerts as alerts +import steward.core.self_monitor as sm + + +# ── fd accounting ──────────────────────────────────────────────────────────── + +def test_count_open_fds_sane_or_none(): + n = sm.count_open_fds() + # On the Linux CI image /proc exists; stdin/stdout/stderr are always open. + assert n is None or n >= 3 + + +def test_fd_soft_limit_positive_or_none(): + soft = sm.fd_soft_limit() + assert soft is None or soft > 0 + + +# ── fake session / app plumbing ────────────────────────────────────────────── + +class _Ctx: + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + +class _FakeSession(_Ctx): + def begin(self): + return _Ctx() + + +class _FakeApp: + def db_sessionmaker(self): + return _FakeSession() + + +def _patch_recorder(monkeypatch): + recorded: list[tuple] = [] + + async def fake_record_metric(session, source_module, resource_name, metric_name, value): + recorded.append((source_module, resource_name, metric_name, value)) + + monkeypatch.setattr(alerts, "record_metric", fake_record_metric) + return recorded + + +# ── record_self_metrics ────────────────────────────────────────────────────── + +def test_records_both_fd_metrics(monkeypatch): + recorded = _patch_recorder(monkeypatch) + monkeypatch.setattr(sm, "count_open_fds", lambda: 50) + monkeypatch.setattr(sm, "fd_soft_limit", lambda: 1000) + + asyncio.run(sm.record_self_metrics(_FakeApp())) + + assert ("steward", "process", "open_fds", 50.0) in recorded + assert ("steward", "process", "open_fds_pct", 5.0) in recorded + + +def test_no_pct_metric_when_limit_unknown(monkeypatch): + recorded = _patch_recorder(monkeypatch) + monkeypatch.setattr(sm, "count_open_fds", lambda: 50) + monkeypatch.setattr(sm, "fd_soft_limit", lambda: None) + + asyncio.run(sm.record_self_metrics(_FakeApp())) + + metric_names = [m for (_s, _r, m, _v) in recorded] + assert "open_fds" in metric_names + assert "open_fds_pct" not in metric_names # meaningless without a ceiling + + +def test_warns_past_floor(monkeypatch, caplog): + _patch_recorder(monkeypatch) + monkeypatch.setattr(sm, "count_open_fds", lambda: 900) + monkeypatch.setattr(sm, "fd_soft_limit", lambda: 1000) # 90% ≥ FD_WARN_PCT + monkeypatch.setattr(sm, "_warned", False) + + with caplog.at_level(logging.WARNING, logger=sm.logger.name): + asyncio.run(sm.record_self_metrics(_FakeApp())) + + assert any("soft limit" in r.message for r in caplog.records) + + +def test_no_warning_below_floor(monkeypatch, caplog): + _patch_recorder(monkeypatch) + monkeypatch.setattr(sm, "count_open_fds", lambda: 100) + monkeypatch.setattr(sm, "fd_soft_limit", lambda: 1000) # 10% + monkeypatch.setattr(sm, "_warned", False) + + with caplog.at_level(logging.WARNING, logger=sm.logger.name): + asyncio.run(sm.record_self_metrics(_FakeApp())) + + assert not any("soft limit" in r.message for r in caplog.records) + + +def test_noop_when_fd_count_unavailable(monkeypatch): + # /proc absent → must return before touching the DB. + monkeypatch.setattr(sm, "count_open_fds", lambda: None) + + class _Boom: + def db_sessionmaker(self): + raise AssertionError("must not open a session when fds are unknown") + + asyncio.run(sm.record_self_metrics(_Boom())) # no raise