diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 15ef48a..74fd1f4 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -266,7 +266,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, log batches (m79) — appended one row per line when present, independent of whether this sample also carried container metrics. """ - from steward.core.alerts import record_metric + from steward.core.alerts import MetricBatch from .models import DockerContainer, DockerEvent, DockerMetric if swarm is not None: @@ -313,6 +313,7 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, )) # Current state + alerts from the newest snapshot only. + batch = MetricBatch() for c in latest_containers: name = c.get("name") if not name: @@ -350,19 +351,19 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, # on different hosts don't collide in the metric/alert namespace. resource = f"{host.name}/{name}" if c.get("status") == "running" and c.get("cpu_pct") is not None: - await record_metric( - session=session, source_module="docker", + batch.add( + source_module="docker", resource_name=resource, metric_name="cpu_pct", value=c["cpu_pct"], ) if c.get("mem_pct") is not None: - await record_metric( - session=session, source_module="docker", + batch.add( + source_module="docker", resource_name=resource, metric_name="mem_pct", value=c["mem_pct"], ) # Restart count is alertable regardless of state (crash-looping matters # most while the container is down/restarting). - await record_metric( - session=session, source_module="docker", + batch.add( + source_module="docker", resource_name=resource, metric_name="restart_count", value=float(c.get("restart_count", 0) or 0), ) @@ -371,12 +372,14 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, # container). health = c.get("health") if health in ("healthy", "unhealthy", "starting"): - await record_metric( - session=session, source_module="docker", + batch.add( + source_module="docker", resource_name=resource, metric_name="is_healthy", value=1.0 if health == "healthy" else 0.0, ) + await batch.record(session) + # Reap containers that vanished from the host's latest listing. Without this, # removed one-shot containers (CI job runners, buildkit builders, codex jobs) # accumulate forever as permanent "stopped" rows and inflate every count. diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 118f075..4e6e720 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -201,6 +201,7 @@ async def ingest(): return _error(401, "invalid_token") accepted = 0 + metric_rows: list[PluginMetric] = [] latest_ts: datetime | None = None docker_snapshots: list[tuple[datetime, list]] = [] docker_log_batches: list[tuple[datetime, list]] = [] @@ -213,9 +214,7 @@ async def ingest(): recorded_at = _parse_ts(sample["ts"]) except (KeyError, ValueError): continue - metrics = _expand_sample_to_metrics(sample, host.name, recorded_at) - for m in metrics: - session.add(m) + metric_rows.extend(_expand_sample_to_metrics(sample, host.name, recorded_at)) docker = sample.get("docker") if isinstance(docker, list) and docker: docker_snapshots.append((recorded_at, docker)) @@ -243,6 +242,10 @@ async def ingest(): await session.rollback() return _error(400, "malformed_payload", "no valid samples") + # Through the alert pipeline, so rules on host metrics can fire. + from steward.core.alerts import record_metrics + await record_metrics(session, metric_rows) + # Hand host-scoped container data to the docker plugin if it's enabled # (opportunistic synergy via the capability registry — no hard import, # no-op when docker is disabled). A failure here must never sink the diff --git a/plugins/snmp/scheduler.py b/plugins/snmp/scheduler.py index e0d4df7..b9b1c79 100644 --- a/plugins/snmp/scheduler.py +++ b/plugins/snmp/scheduler.py @@ -27,7 +27,7 @@ def make_poll_task(app: "Quart") -> ScheduledTask: async def _do_poll(app: "Quart") -> None: from .poller import poll_device - from steward.core.alerts import record_metric + from steward.core.alerts import MetricBatch devices: list[dict] = app.config["PLUGINS"]["snmp"].get("devices", []) if not devices: @@ -35,6 +35,7 @@ async def _do_poll(app: "Quart") -> None: async with app.db_sessionmaker() as session: async with session.begin(): + batch = MetricBatch() for device in devices: if not isinstance(device, dict): continue @@ -55,8 +56,7 @@ async def _do_poll(app: "Quart") -> None: continue for label, value in readings.items(): - await record_metric( - session=session, + batch.add( source_module="snmp", resource_name=name, metric_name=label, @@ -67,3 +67,4 @@ async def _do_poll(app: "Quart") -> None: logger.debug("SNMP polled %s (%s): %d OID(s)", name, host, len(readings)) else: logger.debug("SNMP polled %s (%s): no readings", name, host) + await batch.record(session) diff --git a/plugins/traefik/scheduler.py b/plugins/traefik/scheduler.py index 678e788..6bbc9e3 100644 --- a/plugins/traefik/scheduler.py +++ b/plugins/traefik/scheduler.py @@ -48,7 +48,7 @@ async def _do_scrape(app: "Quart") -> None: extract_certs, fetch_metrics, ) - from steward.core.alerts import record_metric + from steward.core.alerts import MetricBatch metrics_url: str = app.config["PLUGINS"]["traefik"]["metrics_url"] @@ -71,6 +71,7 @@ async def _do_scrape(app: "Quart") -> None: async with app.db_sessionmaker() as session: async with session.begin(): + batch = MetricBatch() # ── Per-service metrics ─────────────────────────────────────────── for router_name, metrics in router_metrics.items(): row = TraefikMetric( @@ -95,8 +96,7 @@ async def _do_scrape(app: "Quart") -> None: ("latency_p99_ms", metrics["latency_p99_ms"]), ("response_bytes_rate", metrics["response_bytes_rate"]), ]: - await record_metric( - session=session, + batch.add( source_module="traefik", resource_name=router_name, metric_name=metric_name, @@ -125,14 +125,15 @@ async def _do_scrape(app: "Quart") -> None: # Emit cert_expiry_days for alert rules days_remaining = (cert["not_after"] - scraped_at).total_seconds() / 86400.0 - await record_metric( - session=session, + batch.add( source_module="traefik", resource_name=cert["cn"] or cert["serial"], metric_name="cert_expiry_days", value=max(0.0, days_remaining), ) + await batch.record(session) + logger.debug( "Traefik scrape: %d router(s), %d cert(s)", len(router_metrics), diff --git a/plugins/unifi/scheduler.py b/plugins/unifi/scheduler.py index cfdeab6..d3b101b 100644 --- a/plugins/unifi/scheduler.py +++ b/plugins/unifi/scheduler.py @@ -54,7 +54,7 @@ async def _do_poll(app: "Quart") -> None: from .client import UnifiClient from .models import UnifiClientSnapshot, UnifiDevice, UnifiWanStat - from steward.core.alerts import record_metric + from steward.core.alerts import MetricBatch cfg = app.config["PLUGINS"]["unifi"] @@ -171,28 +171,27 @@ async def _do_poll(app: "Quart") -> None: )) # ── Alert metrics ───────────────────────────────────────────────── + batch = MetricBatch() if wan_stat.latency_ms is not None: - await record_metric( - session=session, + batch.add( source_module="unifi", resource_name="wan", metric_name="latency_ms", value=wan_stat.latency_ms, ) - await record_metric( - session=session, + batch.add( source_module="unifi", resource_name="wan", metric_name="is_up", value=1.0 if wan_stat.is_up else 0.0, ) - await record_metric( - session=session, + batch.add( source_module="unifi", resource_name="clients", metric_name="total_clients", value=float(len(clients)), ) + await batch.record(session) logger.debug( "UniFi poll: wan=%s, clients=%d, devices=%d", diff --git a/steward/alerts/service.py b/steward/alerts/service.py index 567c3fd..9a29774 100644 --- a/steward/alerts/service.py +++ b/steward/alerts/service.py @@ -33,6 +33,8 @@ METRIC_CATALOG: dict[str, list[str]] = { "host_agent": [ "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes", "disk_used_pct_worst", "load_1m", "load_5m", "load_15m", "uptime_secs", + # Per mount: the resource is ":". + "disk_used_pct", ], # snmp metrics are user-defined OID labels — populated dynamically from plugin_metrics "snmp": [], diff --git a/steward/core/alerts.py b/steward/core/alerts.py index 5f5c0f0..be67cb5 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from typing import TYPE_CHECKING -from sqlalchemy import and_, or_, select +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from steward.models.alerts import ( @@ -57,82 +57,109 @@ async def record_metric( metric_name: str, value: float, ) -> None: - """Write a metric row and evaluate alert rules inline. + """One metric sample: record_metrics with a single row.""" + await record_metrics(session, [PluginMetric( + source_module=source_module, resource_name=resource_name, + metric_name=metric_name, value=value)]) - Must be called inside an active transaction: - async with session.begin(): - await record_metric(session, ...) - Notification I/O is deferred via asyncio.create_task() so no network - calls occur while the transaction is open. +class MetricBatch: + """Gathers one transaction's samples so they are recorded, and their alert + rules evaluated, in a single record_metrics call: + + batch = MetricBatch() + batch.add(source_module="unifi", resource_name="wan", metric_name="is_up", value=1.0) + await batch.record(session) """ - now = datetime.now(timezone.utc) - # Write metric row - metric = PluginMetric( - source_module=source_module, - resource_name=resource_name, - metric_name=metric_name, - value=value, - recorded_at=now, - ) - session.add(metric) + def __init__(self) -> None: + self.rows: list[PluginMetric] = [] + + def add(self, source_module: str, resource_name: str, metric_name: str, value: float, + recorded_at: datetime | None = None) -> None: + self.rows.append(PluginMetric(source_module=source_module, resource_name=resource_name, + metric_name=metric_name, value=float(value), + recorded_at=recorded_at)) + + async def record(self, session: AsyncSession) -> None: + rows, self.rows = self.rows, [] + await record_metrics(session, rows) + + +def _in_window(w: MaintenanceWindow, source_module: str, resource_name: str) -> bool: + if w.scope_source is None: + return True + return w.scope_source == source_module and w.scope_resource in (None, resource_name) + + +async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> None: + """Write metric rows and evaluate the alert rules that watch them. + + Every ingest path comes through here, so a rule on any metric Steward + records can fire. Must be called inside an active transaction: + async with session.begin(): + await record_metrics(session, rows) + + Maintenance windows, rules and their states are read once per batch, not + once per row. Each rule sees its key's samples oldest first, so a backlog + (an agent catching up after an outage) counts consecutive checks as they + happened. Notification I/O is deferred via asyncio.create_task() so no + network calls occur while the transaction is open. + """ + if not rows: + return + now = datetime.now(timezone.utc) + for row in rows: + if row.recorded_at is None: + row.recorded_at = now + session.add_all(rows) await session.flush() - # Check active maintenance windows — skip rule evaluation if suppressed - mw_result = await session.execute( - select(MaintenanceWindow).where( - MaintenanceWindow.start_at <= now, - MaintenanceWindow.end_at >= now, - or_( - MaintenanceWindow.scope_source.is_(None), - and_( - MaintenanceWindow.scope_source == source_module, - or_( - MaintenanceWindow.scope_resource.is_(None), - MaintenanceWindow.scope_resource == resource_name, - ), - ), - ), - ).limit(1) - ) - if mw_result.scalar_one_or_none() is not None: - return # suppressed by maintenance window + by_key: dict[tuple[str, str, str], list[PluginMetric]] = {} + for row in rows: + by_key.setdefault((row.source_module, row.resource_name, row.metric_name), []).append(row) + sources = {k[0] for k in by_key} + metric_names = {k[2] for k in by_key} - # Load matching enabled rules - result = await session.execute( + rules = (await session.execute( select(AlertRule).where( - AlertRule.source_module == source_module, - AlertRule.resource_name == resource_name, - AlertRule.metric_name == metric_name, + AlertRule.source_module.in_(sources), + AlertRule.metric_name.in_(metric_names), AlertRule.enabled.is_(True), ) - ) - rules = result.scalars().all() + )).scalars().all() + rules = [r for r in rules if (r.source_module, r.resource_name, r.metric_name) in by_key] + if not rules: + return + + windows = (await session.execute( + select(MaintenanceWindow).where( + MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now) + )).scalars().all() + states = {st.rule_id: st for st in (await session.execute( + select(AlertState).where(AlertState.rule_id.in_([r.id for r in rules])) + )).scalars()} for rule in rules: - state_result = await session.execute( - select(AlertState).where(AlertState.rule_id == rule.id) - ) - alert_state = state_result.scalar_one_or_none() + if any(_in_window(w, rule.source_module, rule.resource_name) for w in windows): + continue # suppressed by maintenance window + alert_state = states.get(rule.id) if alert_state is None: - logger.warning(f"No alert_state row for rule {rule.id}, skipping") + logger.warning("No alert_state row for rule %s, skipping", rule.id) continue - - notifications = await _evaluate_rule(session, rule, alert_state, value, now) - - for notif_type, event_id in notifications: - if _app is not None: - asyncio.create_task( - _dispatch_notification( - _app, notif_type, rule, value, event_id - ) - ) - # Admin-configured Ansible action runs only on transition-to-firing. - if notif_type == "firing" and rule.ansible_action: + samples = sorted(by_key[(rule.source_module, rule.resource_name, rule.metric_name)], + key=lambda r: r.recorded_at) + for sample in samples: + notifications = await _evaluate_rule(session, rule, alert_state, sample.value, + sample.recorded_at) + for notif_type, event_id in notifications: + if _app is not None: asyncio.create_task( - _run_ansible_action(_app, rule, value, event_id) - ) + _dispatch_notification(_app, notif_type, rule, sample.value, event_id)) + # Admin-configured Ansible action runs only on transition-to-firing. + if notif_type == "firing" and rule.ansible_action: + asyncio.create_task( + _run_ansible_action(_app, rule, sample.value, event_id)) def _is_breached(value: float, operator: AlertOperator, threshold: float) -> bool: diff --git a/steward/core/self_monitor.py b/steward/core/self_monitor.py index cc0eadd..2f76487 100644 --- a/steward/core/self_monitor.py +++ b/steward/core/self_monitor.py @@ -80,28 +80,28 @@ async def record_self_metrics(app) -> None: logger.debug("self_monitor: /proc/self/fd unavailable — skipping fd metrics") return - from .alerts import record_metric + from .alerts import MetricBatch 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, + batch = MetricBatch() + batch.add( source_module="steward", resource_name="process", metric_name="open_fds", value=float(fds), ) if pct is not None: - await record_metric( - session=session, + batch.add( source_module="steward", resource_name="process", metric_name="open_fds_pct", value=pct, ) + await batch.record(session) if pct is not None and pct >= FD_WARN_PCT: if not _warned: diff --git a/steward/monitors/scheduler.py b/steward/monitors/scheduler.py index f5fd1b7..39ed50f 100644 --- a/steward/monitors/scheduler.py +++ b/steward/monitors/scheduler.py @@ -6,7 +6,7 @@ from datetime import datetime, timezone from sqlalchemy import select -from steward.core.alerts import record_metric +from steward.core.alerts import MetricBatch from steward.models.monitors import Monitor, MonitorResult from .runner import run_monitor @@ -63,14 +63,16 @@ async def run_due_monitors(app) -> None: db_monitor.last_checked_at = check_time # Alert pipeline — keyed by monitor type (source) + name. - await record_metric( - session=session, source_module=monitor.type, + batch = MetricBatch() + batch.add( + source_module=monitor.type, resource_name=monitor.name, metric_name="is_up", value=1.0 if res["is_up"] else 0.0, ) if res["response_ms"] is not None: - await record_metric( - session=session, source_module=monitor.type, + batch.add( + source_module=monitor.type, resource_name=monitor.name, metric_name="response_ms", value=res["response_ms"], ) + await batch.record(session) diff --git a/tests/core/test_self_monitor.py b/tests/core/test_self_monitor.py index 715b3fa..cff4316 100644 --- a/tests/core/test_self_monitor.py +++ b/tests/core/test_self_monitor.py @@ -3,7 +3,7 @@ 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. +handed to record_metrics. """ import asyncio import logging @@ -48,10 +48,10 @@ class _FakeApp: 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)) + async def fake_record_metrics(session, rows): + recorded.extend((r.source_module, r.resource_name, r.metric_name, r.value) for r in rows) - monkeypatch.setattr(alerts, "record_metric", fake_record_metric) + monkeypatch.setattr(alerts, "record_metrics", fake_record_metrics) return recorded diff --git a/tests/integration/test_host_agent_alerts.py b/tests/integration/test_host_agent_alerts.py new file mode 100644 index 0000000..50075a1 --- /dev/null +++ b/tests/integration/test_host_agent_alerts.py @@ -0,0 +1,75 @@ +"""Integration: a host-agent ingest POST evaluates alert rules on the host's +metrics (#5060). Before, ingest wrote rows straight to plugin_metrics and no +host_agent rule could ever fire.""" +from __future__ import annotations + +import asyncio +import hashlib +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +def test_ingest_fires_a_host_rule_from_a_backlog(app): + from sqlalchemy import select + from plugins.host_agent.models import HostAgentRegistration + from steward.alerts import service as alert_svc + from steward.models.alerts import AlertEvent, AlertState, AlertStateEnum + from steward.models.hosts import Host + from steward.models.metrics import PluginMetric + from steward.models.users import User, UserRole + + async def _go(): + tag = uuid.uuid4().hex[:8] + hostname, token = f"agent-{tag}", f"tok-{tag}-{uuid.uuid4().hex}" + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"ha-{tag}", email=f"ha-{tag}@test", password_hash="x", + role=UserRole.admin) + host = Host(name=hostname, address="10.9.9.9") + s.add_all([user, host]) + await s.flush() + s.add(HostAgentRegistration( + host_id=host.id, token_hash=hashlib.sha256(token.encode()).hexdigest())) + s.add(PluginMetric(source_module="host_agent", resource_name=hostname, + metric_name="cpu_pct", value=5.0)) + await s.flush() + fields = await alert_svc.rule_fields( + s, name=f"cpu {tag}", source_module="host_agent", resource_name=hostname, + metric_name="cpu_pct", operator=">", threshold=90, + consecutive_failures_required=2) + rule = await alert_svc.add_rule(s, fields, created_by=user.id) + rule_id = rule.id + + now = datetime.now(timezone.utc) + # Two buffered samples in one POST, newest first: evaluated oldest + # first, they make the two breaching checks the rule needs. + samples = [{"ts": (now - timedelta(seconds=s)).isoformat(), "cpu_pct": v} + for s, v in ((0, 97.0), (30, 95.0))] + client = app.test_client() + resp = await client.post("/plugins/host_agent/ingest", json={"samples": samples}, + headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 200, await resp.get_data(as_text=True) + + async with app.db_sessionmaker() as s: + st = (await s.execute(select(AlertState).where( + AlertState.rule_id == rule_id))).scalar_one() + events = (await s.execute(select(AlertEvent).where( + AlertEvent.rule_id == rule_id))).scalars().all() + assert st.state == AlertStateEnum.firing + assert [(e.from_state, e.to_state, e.metric_value) for e in events] == [ + ("pending", "firing", 97.0)] + + asyncio.run(_go())