diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index 36403c4..c64a4e2 100644 --- a/plugins/docker/__init__.py +++ b/plugins/docker/__init__.py @@ -62,3 +62,10 @@ async def mcp_overview(db) -> list[dict]: def get_nav() -> list[dict]: # Surfaces in the sidebar's Infrastructure group (fleet-wide container view). return [{"label": "Docker", "href": "/plugins/docker/"}] + + +def get_alert_metrics() -> dict[str, list[str]]: + """What ingest.py records per container (resource "/"): + restart_count is cumulative (alert on crash-looping); is_healthy is 1.0 / + 0.0 from the container's HEALTHCHECK, only for containers that define one.""" + return {"docker": ["cpu_pct", "mem_pct", "restart_count", "is_healthy"]} diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index a0f2b18..2c4643f 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -31,3 +31,13 @@ def register_mcp_tools(register) -> None: async def mcp_overview(db) -> list[dict]: from .mcp_tools import mcp_overview as findings return await findings(db) + + +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 ":").""" + return {"host_agent": [ + "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/snmp/__init__.py b/plugins/snmp/__init__.py index 0222aee..e43d288 100644 --- a/plugins/snmp/__init__.py +++ b/plugins/snmp/__init__.py @@ -26,3 +26,9 @@ def get_blueprint(): def get_nav() -> list[dict]: # Surfaces in the sidebar's Infrastructure group (SNMP device readings). return [{"label": "SNMP", "href": "/plugins/snmp/"}] + + +def get_alert_metrics() -> dict[str, list[str]]: + """SNMP metric names are the OID labels each device is configured with, so + the list is empty here and the rule form offers the labels recorded so far.""" + return {"snmp": []} diff --git a/plugins/traefik/__init__.py b/plugins/traefik/__init__.py index 19f2982..5a85a3e 100644 --- a/plugins/traefik/__init__.py +++ b/plugins/traefik/__init__.py @@ -33,3 +33,11 @@ def get_blueprint(): def get_nav() -> list[dict]: # Surfaces in the sidebar's Infrastructure group (Traefik services view). return [{"label": "Traefik", "href": "/plugins/traefik/"}] + + +def get_alert_metrics() -> dict[str, list[str]]: + """What scheduler.py records for alert rules: per router (resource = the + router name) and cert_expiry_days per certificate (resource = its CN).""" + return {"traefik": ["request_rate", "error_rate_4xx_pct", "error_rate_5xx_pct", + "latency_p50_ms", "latency_p95_ms", "latency_p99_ms", + "response_bytes_rate", "cert_expiry_days"]} diff --git a/plugins/unifi/__init__.py b/plugins/unifi/__init__.py index 5dbedbb..e00a334 100644 --- a/plugins/unifi/__init__.py +++ b/plugins/unifi/__init__.py @@ -31,3 +31,9 @@ def get_blueprint(): def get_nav() -> list[dict]: # Surfaces in the sidebar's Infrastructure group (UniFi network overview). return [{"label": "UniFi", "href": "/plugins/unifi/"}] + + +def get_alert_metrics() -> dict[str, list[str]]: + """What scheduler.py records: the WAN (resource "wan") and the client count + (resource "clients").""" + return {"unifi": ["is_up", "latency_ms", "total_clients"]} diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py index 7c2fa62..0425986 100644 --- a/steward/alerts/routes.py +++ b/steward/alerts/routes.py @@ -122,17 +122,17 @@ async def rule_fields_api(): @alerts_bp.get("/rules/new") @require_role(UserRole.operator) async def new_rule(): - first_source = alert_svc.SOURCE_MODULES[0] + first_source = alert_svc.source_modules()[0] async with current_app.db_sessionmaker() as db: resources = await alert_svc.resources_for(db, first_source) return await render_template( "alerts/rules_form.html", rule=None, operators=_OPERATORS, - source_modules=alert_svc.SOURCE_MODULES, + source_modules=alert_svc.source_modules(), initial_source=first_source, resources=resources, - metrics=alert_svc.METRIC_CATALOG.get(first_source, []), + metrics=alert_svc.metric_catalog().get(first_source, []), ansible_sources=_ansible_source_names(), ) @@ -150,10 +150,10 @@ async def edit_rule(rule_id: str): "alerts/rules_form.html", rule=rule, operators=_OPERATORS, - source_modules=alert_svc.SOURCE_MODULES, + source_modules=alert_svc.source_modules(), initial_source=rule.source_module, resources=resources, - metrics=alert_svc.METRIC_CATALOG.get(rule.source_module, []), + metrics=alert_svc.metric_catalog().get(rule.source_module, []), ansible_sources=_ansible_source_names(), ) @@ -256,7 +256,7 @@ async def list_maintenance(): "alerts/maintenance.html", windows=windows, now=now, - source_modules=alert_svc.SOURCE_MODULES, + source_modules=alert_svc.source_modules(), ) @@ -266,7 +266,7 @@ async def new_maintenance(): return await render_template( "alerts/maintenance_form.html", window=None, - source_modules=alert_svc.SOURCE_MODULES, + source_modules=alert_svc.source_modules(), ) diff --git a/steward/alerts/service.py b/steward/alerts/service.py index 9a29774..5553897 100644 --- a/steward/alerts/service.py +++ b/steward/alerts/service.py @@ -6,6 +6,7 @@ pipeline) call the same code and refuse the same input for the same reasons. """ from __future__ import annotations +import logging from datetime import datetime, timezone from sqlalchemy import select @@ -16,32 +17,48 @@ from steward.models.alerts import ( from steward.models.metrics import PluginMetric from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor -# Static catalog: source_module → available metric names. -# Unified monitors emit metrics under their TYPE (icmp/tcp/dns/http), each with -# is_up (1.0 up / 0.0 down) and response_ms (where the probe times it). -METRIC_CATALOG: dict[str, list[str]] = { +logger = logging.getLogger(__name__) + +# What core itself records. Unified monitors emit metrics under their TYPE +# (icmp/tcp/dns/http), each with is_up (1.0 up / 0.0 down) and response_ms +# (where the probe times it); "steward" is the app watching itself. +CORE_METRICS: dict[str, list[str]] = { "icmp": ["is_up", "response_ms"], "tcp": ["is_up", "response_ms"], "dns": ["is_up", "response_ms"], "http": ["is_up", "response_ms"], - "traefik": ["request_rate", "error_rate", "latency_p50_ms", "latency_p95_ms", - "latency_p99_ms", "response_bytes_rate", "cert_expiry_days"], - "unifi": ["is_up", "latency_ms", "total_clients"], - # restart_count = cumulative restarts (alert on crash-looping); is_healthy = - # 1.0 healthy / 0.0 unhealthy from the container HEALTHCHECK (alert on <1). - "docker": ["cpu_pct", "mem_pct", "restart_count", "is_healthy"], - "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": [], + "steward": ["open_fds", "open_fds_pct"], } -# Ordered list for the source module picker. -SOURCE_MODULES = list(METRIC_CATALOG.keys()) + +def metric_catalog() -> dict[str, list[str]]: + """Every source an alert rule can watch, and its metric names. + + Core's own sources, then each loaded plugin's get_alert_metrics() export + ({source: [metric names]}), so the catalog is written by whatever emits + the metrics and a disabled plugin's sources drop out. An empty list means + the names are dynamic (snmp: the OID labels recorded so far). + """ + from steward.core.plugin_manager import get_loaded_plugin_modules + catalog = {src: list(names) for src, names in CORE_METRICS.items()} + for name, module in get_loaded_plugin_modules().items(): + hook = getattr(module, "get_alert_metrics", None) + if hook is None: + continue + try: + contributed = hook() or {} + except Exception: + logger.exception("Plugin %r: get_alert_metrics() raised; its metrics are skipped", name) + continue + for src, names in contributed.items(): + catalog[str(src)] = [str(n) for n in names] + return catalog + + +def source_modules() -> list[str]: + """The sources in picker order: core's first, then plugins'.""" + return list(metric_catalog()) + # States that need someone's attention: pending (breaching, not yet for long # enough), firing, and acknowledged-but-still-breaching. @@ -92,7 +109,7 @@ async def resources_for(db, source_module: str) -> list[str]: async def metrics_for(db, source_module: str) -> list[str]: """The metric names a rule on this source can watch: the static catalog, or for dynamic sources (snmp) the names recorded so far.""" - metrics = METRIC_CATALOG.get(source_module, []) + metrics = metric_catalog().get(source_module, []) if source_module and not metrics: result = await db.execute( select(PluginMetric.metric_name) @@ -123,8 +140,9 @@ async def rule_fields(db, *, name: str, source_module: str, resource_name: str, if not name: raise ValueError("an alert rule needs a name.") source_module = (source_module or "").strip() - if source_module not in METRIC_CATALOG: - raise ValueError(f"source_module {source_module!r} is not one of {', '.join(SOURCE_MODULES)}.") + catalog = metric_catalog() + if source_module not in catalog: + raise ValueError(f"source_module {source_module!r} is not one of {', '.join(catalog)}.") resource_name = (resource_name or "").strip() resources = await resources_for(db, source_module) if resource_name not in resources: @@ -227,8 +245,9 @@ def window_fields(*, name: str, start_at: datetime, end_at: datetime, raise ValueError("the window must end after it starts.") scope_source = (scope_source or "").strip() or None scope_resource = (scope_resource or "").strip() or None - if scope_source is not None and scope_source not in METRIC_CATALOG: - raise ValueError(f"source {scope_source!r} is not one of {', '.join(SOURCE_MODULES)}.") + catalog = metric_catalog() + if scope_source is not None and scope_source not in catalog: + raise ValueError(f"source {scope_source!r} is not one of {', '.join(catalog)}.") if scope_source is None: scope_resource = None # can't have resource without source return {"name": name[:128], "start_at": start_at, "end_at": end_at, diff --git a/steward/mcp/tools/alerts.py b/steward/mcp/tools/alerts.py index d0c5823..829af6c 100644 --- a/steward/mcp/tools/alerts.py +++ b/steward/mcp/tools/alerts.py @@ -204,13 +204,13 @@ async def get_alert_rule_options(source_module: str = "") -> dict: resource: is_up (1 up, 0 down) and response_ms. """ async with db_session() as db: + sources = alert_svc.source_modules() if not source_module: - return {"sources": {src: await alert_svc.metrics_for(db, src) - for src in alert_svc.SOURCE_MODULES}, + return {"sources": {src: await alert_svc.metrics_for(db, src) for src in sources}, "operators": [o.value for o in AlertOperator]} - if source_module not in alert_svc.METRIC_CATALOG: + if source_module not in sources: raise ValueError(f"source_module {source_module!r} is not one of " - f"{', '.join(alert_svc.SOURCE_MODULES)}.") + f"{', '.join(sources)}.") return {"source_module": source_module, "metrics": await alert_svc.metrics_for(db, source_module), "resources": await alert_svc.resources_for(db, source_module), diff --git a/tests/core/test_metric_catalog.py b/tests/core/test_metric_catalog.py new file mode 100644 index 0000000..c3185c2 --- /dev/null +++ b/tests/core/test_metric_catalog.py @@ -0,0 +1,97 @@ +"""The alert catalog names what is actually recorded: every metric a writer +hands to the alert pipeline (MetricBatch.add / record_metric with a literal +source and name) is one the catalog offers for that source (#5061).""" +import ast +import importlib +import pathlib + +import steward +from steward.alerts.service import CORE_METRICS + +_REPO = pathlib.Path(steward.__file__).parent.parent +_PLUGINS = ("traefik", "unifi", "docker", "host_agent", "snmp") + + +def catalog() -> dict[str, set[str]]: + cat = {src: set(names) for src, names in CORE_METRICS.items()} + for name in _PLUGINS: + for src, names in importlib.import_module(f"plugins.{name}").get_alert_metrics().items(): + cat[src] = set(names) + return cat + + +def _const(node): + return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else None + + +def _loop_names(func: ast.AST, var: str) -> set[str]: + """Literal first elements of `for var, _ in [("a", x), ("b", y)]` loops.""" + names: set[str] = set() + for node in ast.walk(func): + if not (isinstance(node, ast.For) and isinstance(node.iter, (ast.List, ast.Tuple))): + continue + target = node.target + first = target.elts[0] if isinstance(target, ast.Tuple) and target.elts else target + if isinstance(first, ast.Name) and first.id == var: + for elt in node.iter.elts: + if isinstance(elt, ast.Tuple) and elt.elts and _const(elt.elts[0]): + names.add(elt.elts[0].value) + return names + + +def recorded(source: str) -> set[tuple[str, str]]: + """(source, metric) pairs the code hands to the alert pipeline.""" + tree = ast.parse(source) + pairs: set[tuple[str, str]] = set() + for func in ast.walk(tree): + if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for call in ast.walk(func): + if not isinstance(call, ast.Call): + continue + callee = call.func + is_writer = (isinstance(callee, ast.Attribute) and callee.attr == "add" + and isinstance(callee.value, ast.Name) and callee.value.id == "batch") \ + 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} + src = _const(kw.get("source_module")) + metric = kw.get("metric_name") + if src is None or metric is None: + continue + if _const(metric): + pairs.add((src, metric.value)) + elif isinstance(metric, ast.Name): + pairs |= {(src, m) for m in _loop_names(func, metric.id)} + return pairs + + +def _writers() -> list[pathlib.Path]: + return sorted(p for root in ("steward", "plugins") for p in (_REPO / root).rglob("*.py") + if "migrations" not in p.parts and "batch.add(" in p.read_text()) + + +def test_the_catalog_guard_can_fail(): + code = ''' +async def poll(session): + batch = MetricBatch() + 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) +''' + pairs = recorded(code) + assert pairs == {("traefik", "error_rate"), ("traefik", "request_rate"), ("traefik", "made_up")} + cat = catalog() + assert {p for p in pairs if p[1] not in cat.get(p[0], set())} == { + ("traefik", "error_rate"), ("traefik", "made_up")} + + +def test_every_recorded_metric_is_in_the_catalog(): + files = _writers() + assert len(files) >= 6, files # docker, traefik, unifi, snmp, monitors, self_monitor + 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 + 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/mcp/test_monitoring_writes.py b/tests/mcp/test_monitoring_writes.py index 8fe9916..408eecc 100644 --- a/tests/mcp/test_monitoring_writes.py +++ b/tests/mcp/test_monitoring_writes.py @@ -26,7 +26,14 @@ ADMIN_TOOLS = {"delete_monitor", "delete_alert_rule"} @pytest.fixture def known(monkeypatch): - """A source with one resource and its catalog metrics, without a database.""" + """A source with one resource and its catalog metrics, without a database: + the host_agent plugin is loaded (its get_alert_metrics hook writes the + catalog) and has reported one host.""" + import plugins.host_agent + from steward.core import plugin_manager + monkeypatch.setattr(plugin_manager, "get_loaded_plugin_modules", + lambda: {"host_agent": plugins.host_agent}) + async def resources(db, src): return ["nas"] if src == "host_agent" else [] monkeypatch.setattr(alert_svc, "resources_for", resources)