diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index 2c4643f..2008328 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -35,9 +35,10 @@ async def mcp_overview(db) -> list[dict]: def get_alert_metrics() -> dict[str, list[str]]: """Host metrics ingest records for alert rules (resource = the host name; - disk_used_pct per mount, resource ":").""" + disk_used_pct per mount, resource ":"). seconds_since_seen + comes from the scheduler, not the agent: "agent offline" is > 180.""" return {"host_agent": [ - "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes", + "seconds_since_seen", "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes", "disk_used_pct_worst", "disk_used_pct", "load_1m", "load_5m", "load_15m", "uptime_secs", ]} diff --git a/plugins/host_agent/scheduler.py b/plugins/host_agent/scheduler.py index 85056c1..09e84ec 100644 --- a/plugins/host_agent/scheduler.py +++ b/plugins/host_agent/scheduler.py @@ -1,11 +1,12 @@ # plugins/host_agent/scheduler.py from __future__ import annotations import logging -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from typing import Iterable from sqlalchemy import select +from steward.core.alerts import MetricBatch from steward.core.scheduler import ScheduledTask from steward.models.hosts import Host from .models import HostAgentRegistration @@ -13,66 +14,52 @@ from .models import HostAgentRegistration logger = logging.getLogger(__name__) -def _filter_stale( - regs: Iterable, - *, - now: datetime, - stale_after_seconds: int, -) -> list: - """Pure staleness filter: returns the subset with last_seen_at strictly - older than (now - stale_after_seconds). Rows with last_seen_at=None are - never stale (they are unregistered-in-practice).""" - cutoff = now - timedelta(seconds=stale_after_seconds) - return [ - r for r in regs - if r.last_seen_at is not None and r.last_seen_at < cutoff - ] +def seconds_since_seen(regs: Iterable, host_names: dict[str, str], + now: datetime) -> list[tuple[str, float]]: + """(host name, seconds since its agent last reported) per registration. + + An agent that has never reported counts from when it was registered. A + registration whose host is gone is skipped. + """ + out = [] + for r in regs: + name = host_names.get(r.host_id) + seen = r.last_seen_at or r.created_at + if name is None or seen is None: + continue + out.append((name, max(0.0, (now - seen).total_seconds()))) + return out -async def find_stale_registrations(app, stale_after_seconds: int = 180) -> list[dict]: +async def record_seconds_since_seen(app) -> None: + """Record host_agent//seconds_since_seen for every registered agent, + so "agent offline" is an ordinary alert rule (seconds_since_seen > 180).""" + now = datetime.now(timezone.utc) async with app.db_sessionmaker() as session: - all_regs = (await session.execute( - select(HostAgentRegistration) - )).scalars().all() - stale_rows = _filter_stale( - all_regs, - now=datetime.now(timezone.utc), - stale_after_seconds=stale_after_seconds, - ) - if not stale_rows: - return [] - hosts = { - h.id: h for h in (await session.execute( - select(Host).where(Host.id.in_([r.host_id for r in stale_rows])) - )).scalars().all() - } - return [ - { - "host_id": r.host_id, - "host_name": hosts[r.host_id].name if r.host_id in hosts else "?", - "last_seen_at": r.last_seen_at, - } - for r in stale_rows - ] + async with session.begin(): + regs = (await session.execute(select(HostAgentRegistration))).scalars().all() + if not regs: + return + host_names = dict((await session.execute( + select(Host.id, Host.name).where(Host.id.in_([r.host_id for r in regs])) + )).all()) + batch = MetricBatch() + for name, seconds in seconds_since_seen(regs, host_names, now): + batch.add(source_module="host_agent", resource_name=name, + metric_name="seconds_since_seen", value=seconds, recorded_at=now) + await batch.record(session) def make_task(app) -> ScheduledTask: - async def _check_stale(): + async def _record(): try: - stale = await find_stale_registrations(app) + await record_seconds_since_seen(app) except Exception: - logger.exception("host_agent stale check failed") - return - if stale: - logger.info( - "host_agent: %d stale agent(s): %s", - len(stale), - [s["host_name"] for s in stale], - ) + logger.exception("host_agent: recording seconds_since_seen failed") return ScheduledTask( - name="host_agent_stale_check", - coro_factory=_check_stale, + name="host_agent_seconds_since_seen", + coro_factory=_record, interval_seconds=60, run_on_startup=False, ) diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py index 0425986..50053b1 100644 --- a/steward/alerts/routes.py +++ b/steward/alerts/routes.py @@ -24,6 +24,7 @@ def _rule_form(form) -> dict: "threshold": form.get("threshold"), "consecutive_failures_required": form.get("consecutive_failures_required"), "consecutive_ok_required": form.get("consecutive_ok_required"), + "notify_no_data": "notify_no_data" in form, } diff --git a/steward/alerts/service.py b/steward/alerts/service.py index 5553897..c7464cc 100644 --- a/steward/alerts/service.py +++ b/steward/alerts/service.py @@ -9,7 +9,7 @@ from __future__ import annotations import logging from datetime import datetime, timezone -from sqlalchemy import select +from sqlalchemy import or_, select from steward.models.alerts import ( AlertOperator, AlertRule, AlertState, AlertStateEnum, MaintenanceWindow, @@ -66,15 +66,25 @@ ACTIVE_STATES = (AlertStateEnum.firing, AlertStateEnum.acknowledged, AlertStateE async def active_alerts(db) -> list: - """(AlertState, AlertRule) rows not yet resolved, latest transition first.""" + """(AlertState, AlertRule) rows that need attention, latest transition + first: not yet resolved, or with a series that has gone silent (no_data).""" return list((await db.execute( select(AlertState, AlertRule) .join(AlertRule, AlertState.rule_id == AlertRule.id) - .where(AlertState.state.in_(ACTIVE_STATES)) + .where(or_(AlertState.state.in_(ACTIVE_STATES), AlertState.no_data_since.is_not(None))) .order_by(AlertState.last_transition_at.desc()) )).all()) +def alert_label(state: AlertState) -> str: + """What an alert is, in one word: its state, or "no_data" for a quiet rule + whose series has gone silent (a firing alert that goes silent stays + "firing"; check no_data_since for the mark).""" + if state.no_data_since is not None and state.state == AlertStateEnum.inactive: + return "no_data" + return state.state.value + + async def list_rules(db) -> list[AlertRule]: return list((await db.execute(select(AlertRule).order_by(AlertRule.name))).scalars()) @@ -129,7 +139,7 @@ def _choices(values: list[str], limit: int = 20) -> str: async def rule_fields(db, *, name: str, source_module: str, resource_name: str, metric_name: str, operator: str, threshold, consecutive_failures_required=1, - consecutive_ok_required=1) -> dict: + consecutive_ok_required=1, notify_no_data=True) -> dict: """Validated column values for a new or edited alert rule. The same choices the rule form offers: a known source, a resource that @@ -179,7 +189,7 @@ async def rule_fields(db, *, name: str, source_module: str, resource_name: str, return {"name": name[:128], "source_module": source_module, "resource_name": resource_name, "metric_name": metric_name, "operator": op, "threshold": threshold, "consecutive_failures_required": consecutive, - "consecutive_ok_required": ok_checks} + "consecutive_ok_required": ok_checks, "notify_no_data": bool(notify_no_data)} # Changing any of these changes what the rule is watching, so whatever state @@ -189,17 +199,21 @@ WATCHED_FIELDS = ("source_module", "resource_name", "metric_name", "operator", " async def apply_rule_change(db, rule: AlertRule, fields: dict, *, enabled: bool | None = None) -> None: """Apply validated fields (and optionally enabled) to a rule, in the - caller's transaction. A change to what it watches, or disabling it, puts - its alert back to inactive (core.alerts.reset_rule_state).""" + caller's transaction. A change to what it watches, or disabling or + re-enabling it, puts its alert back to inactive and starts its silence + watch over (core.alerts.reset_rule_state).""" from steward.core.alerts import reset_rule_state changed = [k for k in WATCHED_FIELDS if k in fields and getattr(rule, k) != fields[k]] for k, v in fields.items(): setattr(rule, k, v) disabling = enabled is False and rule.enabled + enabling = enabled is True and not rule.enabled if enabled is not None: rule.enabled = bool(enabled) if disabling: await reset_rule_state(db, rule, "rule disabled") + elif enabling: + await reset_rule_state(db, rule, "rule enabled") elif changed: await reset_rule_state(db, rule, "rule changed: " + ", ".join(changed)) diff --git a/steward/app.py b/steward/app.py index ec96e66..21ce721 100644 --- a/steward/app.py +++ b/steward/app.py @@ -344,6 +344,10 @@ def _register_core_tasks(app: Quart) -> None: from .core.cleanup import run_cleanup as _cleanup await _cleanup(app) + async def run_silence_check(): + from .core.alerts import check_silence + await check_silence(app) + async def run_report_check(): from .core.reports import check_and_send await check_and_send(app) @@ -368,6 +372,13 @@ def _register_core_tasks(app: Quart) -> None: interval_seconds=60, run_on_startup=True, ), + # Rules whose series has gone quiet get a visible no_data mark. + ScheduledTask( + name="alert_silence_check", + coro_factory=run_silence_check, + interval_seconds=60, + run_on_startup=False, + ), ScheduledTask( name="monitor_check", coro_factory=run_monitor_checks, diff --git a/steward/core/alerts.py b/steward/core/alerts.py index be67cb5..39b3ecd 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -3,7 +3,7 @@ import asyncio import logging import uuid from dataclasses import dataclass -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING from sqlalchemy import select @@ -141,14 +141,20 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non )).scalars()} for rule in rules: - 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("No alert_state row for rule %s, skipping", rule.id) continue samples = sorted(by_key[(rule.source_module, rule.resource_name, rule.metric_name)], key=lambda r: r.recorded_at) + muted = any(_in_window(w, rule.source_module, rule.resource_name) for w in windows) + resumed = _track_samples(session, rule, alert_state, samples, now) + if resumed and rule.notify_no_data and not muted and _app is not None: + event_id, value, note = resumed + asyncio.create_task( + _dispatch_notification(_app, "data_resumed", rule, value, event_id, note)) + if muted: + continue # suppressed by maintenance window for sample in samples: notifications = await _evaluate_rule(session, rule, alert_state, sample.value, sample.recorded_at) @@ -162,6 +168,78 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non _run_ansible_action(_app, rule, sample.value, event_id)) +# A series is silent once it has gone this long without a sample, or three of +# its usual gaps if that is longer. Until its cadence is known (fewer than two +# samples seen), it gets an hour: a monitor may check only every ten minutes. +NO_DATA_FLOOR_S = 300.0 +NO_DATA_GAPS = 3 +NO_DATA_UNKNOWN_S = 3600.0 + + +def next_interval(interval: float | None, gap: float) -> float: + """A series' usual gap between samples, updated with one more gap. + + A smoothed average, so one late sample barely moves it. A gap longer than + the silence threshold counts only as that threshold: an outage is not the + series' cadence. A real change of cadence still gets learned, a step at a + time (each step at most 1.5x). + """ + gap = max(0.0, gap) + if interval is None or interval <= 0: + return gap + return 0.75 * interval + 0.25 * min(gap, NO_DATA_GAPS * interval) + + +def no_data_after(interval: float | None) -> float: + """Seconds without a sample before a series counts as silent.""" + if not interval: + return NO_DATA_UNKNOWN_S + return max(NO_DATA_FLOOR_S, NO_DATA_GAPS * interval) + + +def silent_since(last_sample_at: datetime | None, interval: float | None, + watching_since: datetime, now: datetime) -> datetime | None: + """When a series went silent, or None while it is still reporting. + + A series that has never had a sample counts from when the rule started + watching it (created, changed or re-enabled). + """ + last = last_sample_at or watching_since + limit = timedelta(seconds=no_data_after(interval)) + return last + limit if now - last > limit else None + + +def _duration(seconds: float) -> str: + seconds = int(seconds) + if seconds < 120: + return f"{seconds}s" + if seconds < 7200: + return f"{seconds // 60}m" + return f"{seconds // 3600}h{(seconds % 3600) // 60:02d}m" + + +def _track_samples(session: AsyncSession, rule: AlertRule, state: AlertState, + samples: list[PluginMetric], now: datetime + ) -> tuple[str, float, str] | None: + """Note a rule's new samples (oldest first) for silence tracking. When the + series had been marked no_data, clear it and record a resume event, + returned as (event_id, value, note) for the caller to notify.""" + for sample in samples: + if state.last_sample_at is not None: + if sample.recorded_at <= state.last_sample_at: + continue # late or duplicate: not a gap + gap = (sample.recorded_at - state.last_sample_at).total_seconds() + state.sample_interval_s = next_interval(state.sample_interval_s, gap) + state.last_sample_at = sample.recorded_at + if state.no_data_since is None or not samples: + return None + note = f"data resumed after {_duration((now - state.no_data_since).total_seconds())} silent" + value = samples[-1].value + event_id = _add_event(session, rule.id, "no_data", state.state.value, value, now, note=note) + state.no_data_since = None + return event_id, value, note + + def _is_breached(value: float, operator: AlertOperator, threshold: float) -> bool: op = operator.value if op == ">": @@ -247,7 +325,8 @@ async def _evaluate_rule( async def reset_rule_state(session: AsyncSession, rule: AlertRule, note: str) -> None: """Put a rule's alert back to inactive because the rule itself changed - (what it watches, its threshold, or it was disabled). An alert that was + (what it watches, its threshold, or it was disabled or re-enabled), and + start its silence watch over. An alert that was firing or acknowledged gets a resolved event carrying `note`; nobody is notified, since nothing about the watched resource changed.""" state = (await session.execute( @@ -257,8 +336,12 @@ async def reset_rule_state(session: AsyncSession, rule: AlertRule, note: str) -> now = datetime.now(timezone.utc) if state.state in (AlertStateEnum.firing, AlertStateEnum.acknowledged): _add_event(session, rule.id, state.state.value, "resolved", None, now, note=note) - if state.state != AlertStateEnum.inactive: - state.last_transition_at = now + # The watch starts over now: silence is measured from here, against the + # cadence of whatever the rule watches from now on. + state.last_transition_at = now + state.last_sample_at = None + state.sample_interval_s = None + state.no_data_since = None state.state = AlertStateEnum.inactive state.consecutive_failure_count = 0 state.consecutive_ok_count = 0 @@ -294,8 +377,9 @@ async def _dispatch_notification( app: "Quart", notif_type: str, rule: AlertRule, - value: float, + value: float | None, event_id: str, + note: str | None = None, ) -> None: """Run after the transaction commits. Sends notifications and updates the event row.""" from steward.core.notifications import dispatch_notifications @@ -308,6 +392,7 @@ async def _dispatch_notification( "threshold": rule.threshold, "resource": rule.resource_name, "source_module": rule.source_module, + "note": note, "timestamp": datetime.now(timezone.utc).isoformat(), } @@ -395,3 +480,51 @@ async def _run_ansible_action( event.ansible_run_id = run_id await executor.start_run(app, run_id, playbook, inventory, source["path"], params) + + +async def check_silence(app: "Quart") -> None: + """Mark the enabled rules whose series has gone silent (no_data). + + Run every minute by the core scheduler. A rule is silent once its series + has had no sample for no_data_after(its usual gap); one that never had a + sample counts from when the rule started watching. The mark sits beside + the state, so a firing alert that goes silent stays firing. Each silence + is recorded once, as an event, and notified once (rule.notify_no_data, + outside maintenance windows), after the transaction commits. + """ + now = datetime.now(timezone.utc) + notify: list[tuple[AlertRule, str, str]] = [] + async with app.db_sessionmaker() as session: + async with session.begin(): + rows = (await session.execute( + select(AlertState, AlertRule) + .join(AlertRule, AlertState.rule_id == AlertRule.id) + .where(AlertRule.enabled.is_(True), AlertState.no_data_since.is_(None)) + )).all() + if not rows: + return + windows = (await session.execute( + select(MaintenanceWindow).where( + MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now) + )).scalars().all() + for state, rule in rows: + watching_since = max(rule.created_at, state.last_transition_at) + since = silent_since(state.last_sample_at, state.sample_interval_s, + watching_since, now) + if since is None: + continue + state.no_data_since = since + if state.last_sample_at is None: + note = (f"no sample yet, {_duration((now - watching_since).total_seconds())}" + " after the rule started watching") + else: + note = f"no sample for {_duration((now - state.last_sample_at).total_seconds())}" + if state.sample_interval_s: + note += f" (usually every {_duration(state.sample_interval_s)})" + event_id = _add_event(session, rule.id, state.state.value, "no_data", None, + now, note=note) + if rule.notify_no_data and not any( + _in_window(w, rule.source_module, rule.resource_name) for w in windows): + notify.append((rule, event_id, note)) + for rule, event_id, note in notify: + asyncio.create_task(_dispatch_notification(app, "no_data", rule, None, event_id, note)) diff --git a/steward/core/notifications.py b/steward/core/notifications.py index c98b020..fed6999 100644 --- a/steward/core/notifications.py +++ b/steward/core/notifications.py @@ -36,13 +36,15 @@ async def _send_email(cfg: dict, alert: dict) -> dict: msg["Subject"] = f"[Steward] {alert['state']} — {alert['rule_name']}" msg["From"] = cfg.get("username", "steward@localhost") msg["To"] = ", ".join(cfg["recipients"]) + value = "no data" if alert.get("value") is None else alert["value"] msg.set_content( f"Alert: {alert['state']}\n" f"Rule: {alert['rule_name']}\n" f"Resource: {alert['resource']}\n" - f"Metric: {alert['metric']} = {alert['value']}\n" + f"Metric: {alert['metric']} = {value}\n" f"Threshold: {alert['threshold']}\n" - f"Time: {alert['timestamp']}\n" + + (f"Note: {alert['note']}\n" if alert.get("note") else "") + + f"Time: {alert['timestamp']}\n" ) loop = asyncio.get_running_loop() await loop.run_in_executor(None, _smtp_send_sync, cfg, msg) diff --git a/steward/core/reports.py b/steward/core/reports.py index e65e793..fddd63b 100644 --- a/steward/core/reports.py +++ b/steward/core/reports.py @@ -55,7 +55,7 @@ async def build_report_data(app) -> dict: active_rows = await alert_svc.active_alerts(db) active_alerts = [ { - "state": state.state.value.upper(), + "state": alert_svc.alert_label(state).upper(), "name": rule.name, "resource": rule.resource_name, "metric": rule.metric_name, diff --git a/steward/mcp/tools/alerts.py b/steward/mcp/tools/alerts.py index 829af6c..97cb967 100644 --- a/steward/mcp/tools/alerts.py +++ b/steward/mcp/tools/alerts.py @@ -41,24 +41,30 @@ async def _usernames(db, ids) -> dict[str, str]: async def list_alerts() -> dict: - """Alerts that need attention now: firing, pending, or acknowledged but still breaching. + """Alerts that need attention now: firing, pending, acknowledged but still + breaching, or gone silent (no_data). pending: the condition holds but not yet for the rule's required number of consecutive checks. firing: it has. acknowledged: someone has seen it and - it still holds. Each carries the rule's condition, the resource and metric - it watches, and `since` (its latest state change). Firing, unacknowledged + it still holds. no_data: the metric the rule watches has stopped reporting + (no sample for three of its usual gaps, at least 5 minutes); a firing + alert that goes silent stays "firing" with `no_data_since` set. Each + carries the rule's condition, the resource and metric it watches, `since` + (its latest state change) and `last_sample_at`. Firing, unacknowledged alerts come first. An alert that cleared is not here; get_alert_history shows the transitions. """ async with db_session() as db: active = await alert_svc.active_alerts(db) names = await _usernames(db, [s.acknowledged_by for s, _ in active]) - order = {"firing": 0, "pending": 1, "acknowledged": 2} + order = {"firing": 0, "no_data": 1, "pending": 2, "acknowledged": 3} rows = [ { "rule_id": rule.id, "rule_name": rule.name, - "state": state.state.value, + "state": alert_svc.alert_label(state), + "no_data_since": iso(state.no_data_since), + "last_sample_at": iso(state.last_sample_at), "source_module": rule.source_module, "resource": rule.resource_name, "metric": rule.metric_name, @@ -70,7 +76,7 @@ async def list_alerts() -> dict: } for state, rule in active ] - rows.sort(key=lambda r: (order.get(r["state"], 3), -(r["age_s"] or 0))) + rows.sort(key=lambda r: (order.get(r["state"], 4), -(r["age_s"] or 0))) return {"alerts": rows, "total": len(rows)} @@ -135,7 +141,9 @@ async def list_alert_rules(enabled_only: bool = False, limit: int = 200) -> dict `state` is the rule's current state: "inactive" (condition not met), "pending", "firing", "acknowledged", "resolved" (cleared since it last - fired), or null if it has never been evaluated. `ansible_action` is the + fired), "no_data" (inactive and its metric has stopped reporting), or null + if it has no state row. `no_data_since` is set while its metric is silent, + whatever the state. `ansible_action` is the playbook the rule runs when it fires, if any (`check: true` = dry run). """ limit = clamp_limit(limit, 200, 500) @@ -153,8 +161,9 @@ async def list_alert_rules(enabled_only: bool = False, limit: int = 200) -> dict "resource": r.resource_name, "metric": r.metric_name, "condition": _condition(r), - "state": states[r.id].state.value if r.id in states else None, + "state": alert_svc.alert_label(states[r.id]) if r.id in states else None, "state_since": iso(states[r.id].last_transition_at) if r.id in states else None, + "no_data_since": iso(states[r.id].no_data_since) if r.id in states else None, "ansible_action": _action(r), } for r in rules @@ -224,12 +233,14 @@ async def _rule(db, rule: str) -> AlertRule: def _rule_row(r: AlertRule) -> dict: return {"rule_id": r.id, "rule_name": r.name, "enabled": r.enabled, "source_module": r.source_module, "resource": r.resource_name, - "metric": r.metric_name, "condition": _condition(r), "ansible_action": _action(r)} + "metric": r.metric_name, "condition": _condition(r), + "notify_no_data": r.notify_no_data, "ansible_action": _action(r)} async def create_alert_rule(name: str, source_module: str, resource: str, metric: str, operator: str, threshold: float, consecutive_checks: int = 1, - clear_after_checks: int = 1, enabled: bool = True) -> dict: + clear_after_checks: int = 1, notify_no_data: bool = True, + enabled: bool = True) -> dict: """Add an alert rule: notify when `metric` of `resource` crosses `threshold`. source_module, resource, metric: get_alert_rule_options lists the valid @@ -240,6 +251,9 @@ async def create_alert_rule(name: str, source_module: str, resource: str, metric clear_after_checks: how many checks in a row must be OK before a firing or acknowledged alert resolves (default 1). An acknowledgement holds until then, however long the breach lasts. + notify_no_data: also notify when the metric stops reporting and when it + resumes (default true). The silence shows in list_alerts either way. + "Agent offline" is source host_agent, metric seconds_since_seen, "> 180". Notifications go to the channels set in Steward's Settings. The playbook a rule runs when it fires is set separately (set_alert_action, admin). @@ -251,7 +265,7 @@ async def create_alert_rule(name: str, source_module: str, resource: str, metric db, name=name, source_module=source_module, resource_name=resource, metric_name=metric, operator=operator, threshold=threshold, consecutive_failures_required=consecutive_checks, - consecutive_ok_required=clear_after_checks) + consecutive_ok_required=clear_after_checks, notify_no_data=notify_no_data) r = await alert_svc.add_rule(db, fields, created_by=caller.user_id, enabled=enabled) await audit_mcp("alert_rule.created", "alert_rule", r.name, {"rule_id": r.id, "source": r.source_module, "resource": r.resource_name, @@ -264,14 +278,16 @@ async def update_alert_rule(rule: str, name: str | None = None, source_module: s operator: str | None = None, threshold: float | None = None, consecutive_checks: int | None = None, clear_after_checks: int | None = None, + notify_no_data: bool | None = None, enabled: bool | None = None) -> dict: """Change an alert rule. Only the arguments you pass change; the rest stay. rule: its name or id (list_alert_rules). Pass enabled=false to silence it without deleting it. The changed rule is validated as create_alert_rule validates a new one. Changing what it watches (source, resource, metric, - operator, threshold) or disabling it puts its alert back to inactive; a - firing alert records a resolved event saying why. Recorded in the audit log. + operator, threshold), or disabling or re-enabling it, puts its alert back + to inactive; a firing alert records a resolved event saying why. Recorded + in the audit log. """ async with db_session() as db, db.begin(): r = await _rule(db, rule) @@ -286,12 +302,14 @@ async def update_alert_rule(rule: str, name: str | None = None, source_module: s consecutive_failures_required=(r.consecutive_failures_required if consecutive_checks is None else consecutive_checks), consecutive_ok_required=(r.consecutive_ok_required - if clear_after_checks is None else clear_after_checks)) + if clear_after_checks is None else clear_after_checks), + notify_no_data=r.notify_no_data if notify_no_data is None else notify_no_data) await alert_svc.apply_rule_change(db, r, fields, enabled=enabled) changed = sorted(k for k, v in {"name": name, "source_module": source_module, "resource": resource, "metric": metric, "operator": operator, "threshold": threshold, "consecutive_checks": consecutive_checks, "clear_after_checks": clear_after_checks, + "notify_no_data": notify_no_data, "enabled": enabled}.items() if v is not None) await audit_mcp("alert_rule.updated", "alert_rule", r.name, {"rule_id": r.id, "changed": changed}) return _rule_row(r) diff --git a/steward/mcp/tools/hosts.py b/steward/mcp/tools/hosts.py index e4bc580..d3bd5ff 100644 --- a/steward/mcp/tools/hosts.py +++ b/steward/mcp/tools/hosts.py @@ -183,7 +183,7 @@ async def get_host(host: str, parts_limit: int = 20) -> dict: prefix = h.name + ":" alerts = [ { - "rule_id": rule.id, "rule_name": rule.name, "state": state.state.value, + "rule_id": rule.id, "rule_name": rule.name, "state": alert_svc.alert_label(state), "resource": rule.resource_name, "metric": rule.metric_name, "condition": f"{rule.metric_name} {rule.operator.value} {rule.threshold:g}", "since": iso(state.last_transition_at), "age_s": age_s(state.last_transition_at), diff --git a/steward/mcp/tools/overview.py b/steward/mcp/tools/overview.py index 1724cfd..827c11b 100644 --- a/steward/mcp/tools/overview.py +++ b/steward/mcp/tools/overview.py @@ -77,13 +77,18 @@ def alert_findings(active) -> list[dict]: sev, what = "critical", "is firing" elif s == "pending": sev, what = "warning", "is pending (breaching, not yet for long enough to fire)" - else: + elif s == "acknowledged": sev, what = "warning", "is acknowledged but still breaching" + else: + s, sev, what = "no_data", "warning", "has stopped reporting" + if state.no_data_since is not None and s != "no_data": + what += f" and has stopped reporting (no data since {iso(state.no_data_since)})" out.append(finding( sev, f"alert_{s}", rule.name, f"{rule.metric_name} {rule.operator.value} {rule.threshold:g} on " f"{rule.source_module}/{rule.resource_name} {what}", - since=state.last_transition_at, ids={"rule_id": rule.id}, + since=state.no_data_since if s == "no_data" else state.last_transition_at, + ids={"rule_id": rule.id}, next=f'get_alert_history(rule="{rule.id}")')) return out diff --git a/steward/migrations/versions/0029_alert_no_data.py b/steward/migrations/versions/0029_alert_no_data.py new file mode 100644 index 0000000..39fd1c8 --- /dev/null +++ b/steward/migrations/versions/0029_alert_no_data.py @@ -0,0 +1,49 @@ +"""Alerts notice silence: per-rule sample tracking and a no-data mark + +Revision ID: 0029_alert_no_data +Revises: 0028_alert_resolve_and_reset +Create Date: 2026-10-06 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0029_alert_no_data" +down_revision: Union[str, None] = "0028_alert_resolve_and_reset" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("alert_rules", sa.Column("notify_no_data", sa.Boolean, + nullable=False, server_default=sa.true())) + op.add_column("alert_states", sa.Column("last_sample_at", sa.DateTime(timezone=True), + nullable=True)) + op.add_column("alert_states", sa.Column("sample_interval_s", sa.Float, nullable=True)) + op.add_column("alert_states", sa.Column("no_data_since", sa.DateTime(timezone=True), + nullable=True)) + # Seed each existing rule's last sample and usual gap from the last day of + # its series, so the first silence check after the upgrade judges every + # rule against what it has actually been receiving. + op.execute(""" + UPDATE alert_states s + SET last_sample_at = k.last_at, sample_interval_s = k.gap + FROM ( + SELECT r.id AS rule_id, max(m.recorded_at) AS last_at, + EXTRACT(EPOCH FROM max(m.recorded_at) - min(m.recorded_at)) + / NULLIF(count(*) - 1, 0) AS gap + FROM alert_rules r + JOIN plugin_metrics m + ON m.source_module = r.source_module AND m.resource_name = r.resource_name + AND m.metric_name = r.metric_name AND m.recorded_at > now() - interval '1 day' + GROUP BY r.id + ) k + WHERE s.rule_id = k.rule_id + """) + + +def downgrade() -> None: + op.drop_column("alert_states", "no_data_since") + op.drop_column("alert_states", "sample_interval_s") + op.drop_column("alert_states", "last_sample_at") + op.drop_column("alert_rules", "notify_no_data") diff --git a/steward/models/alerts.py b/steward/models/alerts.py index d41111b..8141259 100644 --- a/steward/models/alerts.py +++ b/steward/models/alerts.py @@ -45,6 +45,9 @@ class AlertRule(Base): consecutive_ok_required: Mapped[int] = mapped_column(Integer, nullable=False, default=1, server_default="1") enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + # Notify when the watched series goes silent (no_data) and when it resumes. + notify_no_data: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, + server_default="true") # Optional admin-only action: run an Ansible playbook on transition-to-firing. # Shape: {source, playbook, inventory, extra_vars, limit, tags, check}. NULL = none. ansible_action: Mapped[dict | None] = mapped_column(JSON, nullable=True) @@ -74,6 +77,13 @@ class AlertState(Base): String(36), ForeignKey("users.id", ondelete="SET NULL"), nullable=True ) acknowledged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + # Silence tracking (core.alerts.check_silence): when the watched series last + # had a sample, its usual gap between samples, and since when it has had + # none for too long. no_data_since is a mark beside the state, not a state: + # a firing alert that goes silent stays firing. + last_sample_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + sample_interval_s: Mapped[float | None] = mapped_column(Float, nullable=True) + no_data_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) class MaintenanceWindow(Base): diff --git a/steward/templates/alerts/list.html b/steward/templates/alerts/list.html index a739cac..a69a64a 100644 --- a/steward/templates/alerts/list.html +++ b/steward/templates/alerts/list.html @@ -24,9 +24,16 @@ FIRING {% elif state.state.value == 'acknowledged' %} ACK - {% else %} + {% elif state.state.value == 'pending' %} PENDING {% endif %} + {% if state.no_data_since %} + NO DATA +
+ since {{ state.no_data_since.strftime("%Y-%m-%d %H:%M") }} UTC +
+ {% endif %} {{ rule.name }} {{ rule.resource_name }} diff --git a/steward/templates/alerts/rules_form.html b/steward/templates/alerts/rules_form.html index d20f83c..b9a3022 100644 --- a/steward/templates/alerts/rules_form.html +++ b/steward/templates/alerts/rules_form.html @@ -99,6 +99,18 @@ within the threshold. An acknowledgement holds until then.

+
+ +

+ Silence is always shown as NO DATA on the Alerts page: no sample for three of the + metric's usual gaps (at least 5 minutes). This also sends a notification when it starts + and when data resumes. A firing alert that goes silent stays firing.

+
+ {# ── Metric reference ────────────────────────────────────────────────── #}
@@ -113,7 +125,8 @@ "dns": [("is_up", "1.0 = resolved (and matches expected IP), 0.0 = failed"), ("response_ms", "resolution time in milliseconds")], "traefik": [("request_rate", "requests/sec for this router"), - ("error_rate", "fraction of 5xx responses (0–1)"), + ("error_rate_4xx_pct", "% of responses that are 4xx"), + ("error_rate_5xx_pct", "% of responses that are 5xx"), ("latency_p50_ms", "median response time ms"), ("latency_p95_ms", "p95 response time ms"), ("latency_p99_ms", "p99 response time ms"), @@ -126,6 +139,11 @@ ("mem_pct", "container memory usage %")], "http": [("is_up", "1.0 = check passed, 0.0 = failed"), ("response_ms", "HTTP response time in ms")], + "host_agent": [("seconds_since_seen", "seconds since the agent last reported; > 180 = offline"), + ("cpu_pct", "host CPU usage %"), + ("mem_used_pct", "host memory usage %"), + ("disk_used_pct_worst", "fullest disk, % used"), + ("disk_used_pct", "one mount, % used (resource host:mount)")], } %} {% for src, pairs in descs.items() %}
@@ -192,6 +210,20 @@
{% endif %} + {% if not rule %} + + {% endif %} +
Cancel diff --git a/tests/core/test_alert_no_data.py b/tests/core/test_alert_no_data.py new file mode 100644 index 0000000..148c99e --- /dev/null +++ b/tests/core/test_alert_no_data.py @@ -0,0 +1,98 @@ +"""Silence is alertable (#5062): a series' usual gap is learned from its +samples, an outage does not become its cadence, and a series counts as silent +after three usual gaps (at least five minutes; an hour while its cadence is +unknown).""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +import pytest + +from steward.core.alerts import ( + NO_DATA_FLOOR_S, NO_DATA_UNKNOWN_S, _track_samples, next_interval, no_data_after, silent_since, +) +from steward.models.alerts import AlertStateEnum + +NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc) + + +@pytest.mark.parametrize("interval,gap,want", [ + (None, 60, 60), # the first gap is the cadence + (0.0, 30, 30), + (60, 60, 60), # steady + (60, 80, 65), # a late sample barely moves it + (60, 3600, 90), # an outage counts as at most three gaps + (60, -5, 45), # never a negative gap +]) +def test_next_interval(interval, gap, want): + assert next_interval(interval, gap) == pytest.approx(want) + + +def test_a_new_cadence_is_learned_in_steps(): + interval, steps = 60.0, 0 + while no_data_after(interval) < 600: # samples now arrive every 10 minutes + interval, steps = next_interval(interval, 600), steps + 1 + assert steps == 3 + + +@pytest.mark.parametrize("interval,want", [ + (None, NO_DATA_UNKNOWN_S), (0.0, NO_DATA_UNKNOWN_S), + (30, NO_DATA_FLOOR_S), (100, 300), (600, 1800), +]) +def test_no_data_after(interval, want): + assert no_data_after(interval) == want + + +@pytest.mark.parametrize("last_ago,interval,watch_ago,silent_after", [ + (60, 60, 9999, None), # reporting + (300, 60, 9999, None), # exactly at the limit is not yet silent + (301, 60, 9999, 300), # silent since last + 300s + (2000, 600, 9999, 1800), # slow series: three of its gaps + (None, None, 3000, None), # never sampled, watched under an hour + (None, None, 4000, 3600), # never sampled, watched over an hour +]) +def test_silent_since(last_ago, interval, watch_ago, silent_after): + last = NOW - timedelta(seconds=last_ago) if last_ago is not None else None + watch = NOW - timedelta(seconds=watch_ago) + got = silent_since(last, interval, watch, NOW) + if silent_after is None: + assert got is None + else: + assert got == (last or watch) + timedelta(seconds=silent_after) + + +class _Session: + def __init__(self): + self.added = [] + + def add(self, row): + self.added.append(row) + + +def _state(**kw): + base = dict(state=AlertStateEnum.firing, last_sample_at=None, sample_interval_s=None, + no_data_since=None) + return NS(**{**base, **kw}) + + +def _samples(*ago_values): + return [NS(recorded_at=NOW - timedelta(seconds=a), value=v) for a, v in ago_values] + + +def test_tracking_learns_the_gap_and_skips_late_samples(): + st, session = _state(), _Session() + assert _track_samples(session, NS(id="r"), st, _samples((180, 1), (120, 1), (60, 1)), NOW) is None + assert (st.last_sample_at, st.sample_interval_s) == (NOW - timedelta(seconds=60), 60) + _track_samples(session, NS(id="r"), st, _samples((90, 1)), NOW) # older than the last + assert (st.last_sample_at, st.sample_interval_s) == (NOW - timedelta(seconds=60), 60) + assert session.added == [] + + +def test_a_sample_after_silence_clears_it_and_records_the_resume(): + st, session = _state(last_sample_at=NOW - timedelta(minutes=20), sample_interval_s=60, + no_data_since=NOW - timedelta(minutes=15)), _Session() + event_id, value, note = _track_samples(session, NS(id="r"), st, _samples((0, 42.0)), NOW) + assert st.no_data_since is None and st.state == AlertStateEnum.firing # still firing + assert st.sample_interval_s == 90 # the outage counted as three gaps, at most + assert value == 42.0 and note == "data resumed after 15m silent" + [event] = session.added + assert (event.id, event.from_state, event.to_state) == (event_id, "no_data", "firing") diff --git a/tests/core/test_metric_catalog.py b/tests/core/test_metric_catalog.py index c3185c2..a618847 100644 --- a/tests/core/test_metric_catalog.py +++ b/tests/core/test_metric_catalog.py @@ -55,7 +55,9 @@ def recorded(source: str) -> set[tuple[str, str]]: or (isinstance(callee, ast.Name) and callee.id == "record_metric") if not is_writer: continue - kw = {k.arg: k.value for k in call.keywords} + # Positional or keyword: (source_module, resource_name, metric_name, value). + kw = dict(zip(("source_module", "resource_name", "metric_name"), call.args)) + kw.update({k.arg: k.value for k in call.keywords}) src = _const(kw.get("source_module")) metric = kw.get("metric_name") if src is None or metric is None: @@ -79,19 +81,22 @@ async def poll(session): batch.add(source_module="traefik", resource_name="r", metric_name="error_rate", value=1) for metric_name, value in [("request_rate", 1), ("made_up", 2)]: batch.add(source_module="traefik", resource_name="r", metric_name=metric_name, value=value) + batch.add("traefik", "r", "by_position", 3) ''' pairs = recorded(code) - assert pairs == {("traefik", "error_rate"), ("traefik", "request_rate"), ("traefik", "made_up")} + assert pairs == {("traefik", "error_rate"), ("traefik", "request_rate"), ("traefik", "made_up"), + ("traefik", "by_position")} cat = catalog() assert {p for p in pairs if p[1] not in cat.get(p[0], set())} == { - ("traefik", "error_rate"), ("traefik", "made_up")} + ("traefik", "error_rate"), ("traefik", "made_up"), ("traefik", "by_position")} def test_every_recorded_metric_is_in_the_catalog(): files = _writers() - assert len(files) >= 6, files # docker, traefik, unifi, snmp, monitors, self_monitor + assert len(files) >= 7, files # docker, traefik, unifi, snmp, monitors, self_monitor, host_agent pairs = {p for f in files for p in recorded(f.read_text())} assert ("traefik", "error_rate_5xx_pct") in pairs and ("docker", "is_healthy") in pairs + assert ("host_agent", "seconds_since_seen") in pairs cat = catalog() missing = sorted(p for p in pairs if p[1] not in cat.get(p[0], set())) assert not missing, f"recorded but not in the alert catalog: {missing}" diff --git a/tests/integration/test_alert_no_data.py b/tests/integration/test_alert_no_data.py new file mode 100644 index 0000000..26ec19e --- /dev/null +++ b/tests/integration/test_alert_no_data.py @@ -0,0 +1,125 @@ +"""Integration (#5062): against real Postgres, a rule whose series stops is +marked no_data once, notified once, and cleared by the next sample; a firing +alert that goes silent stays firing.""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +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_silence_is_marked_once_and_cleared_by_data(app): + from sqlalchemy import select, update + from steward.alerts import service as alert_svc + from steward.core import alerts + from steward.models.alerts import AlertEvent, AlertRule, AlertState, AlertStateEnum + from steward.models.metrics import PluginMetric + from steward.models.users import User, UserRole + + sent: list[tuple[str, str, str | None]] = [] + + async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None): + sent.append((rule.name, notif_type, note)) + + async def _go(): + tag = uuid.uuid4().hex[:8] + res = f"quiet-{tag}" + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"nd-{tag}", email=f"nd-{tag}@test", password_hash="x", + role=UserRole.admin) + s.add(user) + s.add(PluginMetric(source_module="host_agent", resource_name=res, + metric_name="cpu_pct", value=1.0, + recorded_at=now - timedelta(hours=1))) + await s.flush() + ids = {} + for key, notify in (("loud", True), ("hushed", False)): + fields = await alert_svc.rule_fields( + s, name=f"{key} {tag}", source_module="host_agent", resource_name=res, + metric_name="cpu_pct", operator=">", threshold=90, + notify_no_data=notify) + ids[key] = (await alert_svc.add_rule(s, fields, created_by=user.id)).id + # Every minute, from 20 to 18 minutes ago; then nothing. + batch = alerts.MetricBatch() + for ago in (20, 19, 18): + batch.add(source_module="host_agent", resource_name=res, metric_name="cpu_pct", + value=10.0, recorded_at=now - timedelta(minutes=ago)) + await batch.record(s) + + async def state(key): + async with app.db_sessionmaker() as s: + return (await s.execute(select(AlertState).where( + AlertState.rule_id == ids[key]))).scalar_one() + + async def events(key): + async with app.db_sessionmaker() as s: + rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == ids[key]) + .order_by(AlertEvent.transitioned_at))).scalars().all() + return [(e.from_state, e.to_state, e.note) for e in rows] + + st = await state("loud") + assert st.sample_interval_s == pytest.approx(60) + assert st.last_sample_at == now - timedelta(minutes=18) + + await alerts.check_silence(app) + await alerts.check_silence(app) # already marked: nothing more + await asyncio.sleep(0) + st = await state("loud") + assert st.state == AlertStateEnum.inactive + assert st.no_data_since == now - timedelta(minutes=18) + timedelta(seconds=300) + [(frm, to, note)] = await events("loud") + assert (frm, to) == ("inactive", "no_data") + assert note == "no sample for 18m (usually every 60s)" + assert len(await events("hushed")) == 1 + assert sent == [(f"loud {tag}", "no_data", note)] # the hushed rule notifies nobody + + async with app.db_sessionmaker() as s: + active = {r.id: alert_svc.alert_label(st_) for st_, r in await alert_svc.active_alerts(s)} + assert active[ids["loud"]] == active[ids["hushed"]] == "no_data" + + # Data resumes, breaching: the mark clears and the rule fires. + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0) + await asyncio.sleep(0) + st = await state("loud") + assert st.no_data_since is None and st.state == AlertStateEnum.firing + first, *same_instant = await events("loud") + assert first[:2] == ("inactive", "no_data") + # The resume and the firing it leads to share the sample's timestamp. + assert sorted(e[:2] for e in same_instant) == [("inactive", "firing"), + ("no_data", "inactive")] + assert any((e[2] or "").startswith("data resumed after ") for e in same_instant) + assert [t for _, t, _ in sent if t != "firing"] == ["no_data", "data_resumed"] + + # A firing alert that goes silent stays firing, marked. + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(AlertState).where(AlertState.rule_id == ids["loud"]) + .values(last_sample_at=now - timedelta(hours=2))) + await s.execute(update(AlertRule).where(AlertRule.id == ids["hushed"]) + .values(enabled=False)) + await alerts.check_silence(app) + st = await state("loud") + assert st.state == AlertStateEnum.firing and st.no_data_since is not None + assert (await events("loud"))[-1][:2] == ("firing", "no_data") + assert len(await events("hushed")) == 3 # disabled: not watched + + with patch.object(alerts, "_dispatch_notification", fake_dispatch): + asyncio.run(_go()) diff --git a/tests/integration/test_alert_states.py b/tests/integration/test_alert_states.py index 04e062b..82cbd21 100644 --- a/tests/integration/test_alert_states.py +++ b/tests/integration/test_alert_states.py @@ -104,4 +104,13 @@ def test_acknowledge_holds_clear_after_and_reset(app): assert (await state()).state == AlertStateEnum.inactive assert (await events())[-1] == ("firing", "resolved", "rule disabled") + # Re-enabling starts the watch over: silence is measured from now. + async with app.db_sessionmaker() as s: + async with s.begin(): + r = await s.get(AlertRule, rule_id) + await alert_svc.apply_rule_change(s, r, {}, enabled=True) + st = await state() + assert (st.state, st.last_sample_at, st.no_data_since) == (AlertStateEnum.inactive, None, None) + assert (await events())[-1] == ("firing", "resolved", "rule disabled") + asyncio.run(_go()) diff --git a/tests/mcp/test_tools.py b/tests/mcp/test_tools.py index 6db2ae5..66b688e 100644 --- a/tests/mcp/test_tools.py +++ b/tests/mcp/test_tools.py @@ -143,13 +143,30 @@ def test_monitor_findings(): def test_alert_findings_severity_by_state(): rule = NS(id="r1", name="cpu", metric_name="cpu_pct", operator=AlertOperator.gt, threshold=90.0, source_module="host_agent", resource_name="nas") - rows = [(NS(state=NS(value=s), last_transition_at=NOW), rule) + rows = [(NS(state=NS(value=s), last_transition_at=NOW, no_data_since=None), rule) for s in ("firing", "pending", "acknowledged")] sev = {f["kind"]: f["severity"] for f in overview.alert_findings(rows)} assert sev == {"alert_firing": "critical", "alert_pending": "warning", "alert_acknowledged": "warning"} +def test_alert_findings_say_when_a_metric_went_silent(): + rule = NS(id="r1", name="cpu", metric_name="cpu_pct", operator=AlertOperator.gt, + threshold=90.0, source_module="host_agent", resource_name="nas") + silent = NOW - timedelta(minutes=10) + quiet, firing = overview.alert_findings([ + (NS(state=NS(value="inactive"), last_transition_at=NOW - timedelta(days=1), + no_data_since=silent), rule), + (NS(state=NS(value="firing"), last_transition_at=NOW - timedelta(hours=1), + no_data_since=silent), rule), + ]) + assert (quiet["kind"], quiet["severity"]) == ("alert_no_data", "warning") + assert quiet["since"] == "2026-10-06T09:50:00Z" and "stopped reporting" in quiet["summary"] + # A firing alert that goes silent stays a firing finding, marked. + assert (firing["kind"], firing["severity"]) == ("alert_firing", "critical") + assert "no data since 2026-10-06T09:50:00Z" in firing["summary"] + + THRESHOLDS = {"cpu": {"warn": 80, "crit": 90}, "mem": {"warn": 80, "crit": 90}, "disk": {"warn": 80, "crit": 90}} diff --git a/tests/plugins/host_agent/test_scheduler.py b/tests/plugins/host_agent/test_scheduler.py index 49be93a..a1b910b 100644 --- a/tests/plugins/host_agent/test_scheduler.py +++ b/tests/plugins/host_agent/test_scheduler.py @@ -1,48 +1,41 @@ -"""Pure-function tests for host_agent staleness filter.""" +"""Pure-function tests for the host_agent seconds_since_seen reconciler.""" from dataclasses import dataclass from datetime import datetime, timedelta, timezone -from plugins.host_agent.scheduler import _filter_stale +from plugins.host_agent.scheduler import seconds_since_seen @dataclass class FakeReg: host_id: str last_seen_at: datetime | None + created_at: datetime | None NOW = datetime(2026, 4, 15, 12, 0, 0, tzinfo=timezone.utc) +NAMES = {"h1": "nas", "h2": "web", "h3": "db"} -def test_filter_stale_returns_only_old_rows(): - regs = [ - FakeReg("fresh", NOW - timedelta(seconds=30)), - FakeReg("stale", NOW - timedelta(seconds=600)), - FakeReg("never", None), - ] - stale = _filter_stale(regs, now=NOW, stale_after_seconds=180) - ids = {r.host_id for r in stale} - assert ids == {"stale"} +def test_seconds_since_each_agent_last_reported(): + regs = [FakeReg("h1", NOW - timedelta(seconds=30), NOW - timedelta(days=9)), + FakeReg("h2", NOW - timedelta(minutes=10), NOW - timedelta(days=9))] + assert seconds_since_seen(regs, NAMES, NOW) == [("nas", 30.0), ("web", 600.0)] -def test_filter_stale_empty_input(): - assert _filter_stale([], now=NOW, stale_after_seconds=180) == [] +def test_an_agent_that_never_reported_counts_from_registration(): + regs = [FakeReg("h3", None, NOW - timedelta(minutes=5))] + assert seconds_since_seen(regs, NAMES, NOW) == [("db", 300.0)] -def test_filter_stale_all_fresh(): - regs = [ - FakeReg("a", NOW - timedelta(seconds=10)), - FakeReg("b", NOW - timedelta(seconds=30)), - ] - assert _filter_stale(regs, now=NOW, stale_after_seconds=60) == [] +def test_a_registration_without_its_host_is_skipped(): + regs = [FakeReg("gone", NOW, NOW)] + assert seconds_since_seen(regs, NAMES, NOW) == [] -def test_filter_stale_boundary_exact_cutoff_is_not_stale(): - # A row exactly at the cutoff should not be flagged (strict less-than). - regs = [FakeReg("edge", NOW - timedelta(seconds=180))] - assert _filter_stale(regs, now=NOW, stale_after_seconds=180) == [] +def test_a_clock_ahead_of_ours_reads_as_just_seen(): + regs = [FakeReg("h1", NOW + timedelta(seconds=5), NOW)] + assert seconds_since_seen(regs, NAMES, NOW) == [("nas", 0.0)] -def test_filter_stale_skips_none_last_seen(): - regs = [FakeReg("never", None)] - assert _filter_stale(regs, now=NOW, stale_after_seconds=1) == [] +def test_empty_input(): + assert seconds_since_seen([], NAMES, NOW) == []