feat(norms): hosts, containers and monitor response times are graded against their own normal (#5742)

Host CPU, memory and load per core, container CPU and memory, and every
monitor's response time learn a normal per hour (milestone 513, System 182).
A reading past its fixed cutoff but within its normal reads healthy and says
what is usual at this hour; one well out of it keeps its grade and is a
metric anomaly. Disk, a monitor down and an agent offline keep their cutoffs.

- Each normal's floor, the smallest departure that counts, is a setting:
  Settings -> Thresholds -> Normals (norms.floor.*). Defaults: host CPU 10
  points, host memory 5, load per core 10, container CPU 10 points of one
  core, container memory 5, monitor response 100 ms; Traefik's 2-point 4xx
  and 50 ms p95 floors move there from constants. learn_normal takes a
  floor function (floor_setting) read at each pass.
- learn_normal(parts=False): a host learns, its cores don't.
- Ingest records load_pct (load_1m / cores x 100), the figure pages grade.
- Shared grading: anomalies.against / graded, worst_reading(normals=),
  hosts' host_grades, docker usage(thresholds=, norms=), monitors'
  latency_grade. Applied to host vitals, the Hosts widget, MCP overview,
  Docker usage, and the monitor list, widget and page.
- The monitor and container pages list their anomalies.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/core/anomalies.py

@@ -51,6 +51,33 @@ def departs(level: float, norm: Norm, floor: float) -> bool:
     return level > norm.usual_hold * DEPART and level - norm.usual_hold >= floor
 
 
+def against(value: float, norm: Norm | None, floor: float, unit: str) -> tuple[bool, str]:
+    """(departs, what is usual) for a page's reading past its fixed cutoff
+    (#5741, #5742). Without a normal it departs, and the cutoff stands; within
+    one it is usual for that resource, and the page says so. `unit` is "%"
+    (one decimal) or " ms" (whole)."""
+    if norm is None:
+        return True, ""
+    usual = f"{norm.usual_hold:.1f}{unit}" if unit == "%" else f"{norm.usual_hold:.0f}{unit}"
+    return departs(value, norm, floor), f"usual {usual} at this hour"
+
+
+def graded(value: float | None, kind: str, thresholds: dict, norm: Norm | None,
+           floor: float, unit: str) -> tuple[str, str | None]:
+    """(status class, why) for a behaviour reading on a page (#5742): graded
+    against Settings → Thresholds, except that one past its cutoff but within
+    its own normal for the hour is "ok", and `why` says what is usual. Only
+    for behaviour (CPU, memory, load, latency); a failure keeps its cutoff
+    whatever its history (System 182)."""
+    from steward.core.formatting import status_class
+    cls = status_class(value, kind, thresholds)
+    if cls in ("warn", "crit") and norm is not None:
+        out, said = against(value, norm, floor, unit)
+        if not out:
+            return "ok", said
+    return cls, None
+
+
 # Each learned key reporting in the window, with the window's average. Keys
 # come from plugin_metric_latest (one row each), and each average is a range
 # read of the series index, so the pass never scans the raw table.
@@ -64,18 +91,22 @@ CROSS JOIN LATERAL (
       AND p.recorded_at <= :now
 ) v
 WHERE l.source_module || '/' || l.metric_name = ANY(:keys)
+  AND (strpos(l.resource_name, ':') = 0 OR NOT l.source_module || '/' || l.metric_name = ANY(:whole))
   AND l.recorded_at >= :since AND v.level IS NOT NULL
-""").bindparams(bindparam("keys", type_=ARRAY(String)))
+""").bindparams(bindparam("keys", type_=ARRAY(String)), bindparam("whole", type_=ARRAY(String)))
 
 
 async def levels_now(session, now: datetime) -> dict[tuple[str, str, str], float]:
     """{(source, resource, metric): its average over the last WINDOW}, for
-    every learned metric and every metric one is gated on."""
+    every learned metric and every metric one is gated on. A metric learned
+    without its parts reads its whole resources only."""
     keys = sorted({f"{s}/{m}" for s, m in LEARNED}
                   | {f"{k.source_module}/{k.gate[0]}" for k in LEARNED.values() if k.gate})
     if not keys:
         return {}
-    rows = await session.execute(_LEVELS, {"keys": keys, "since": now - WINDOW, "now": now})
+    whole = sorted(f"{k.source_module}/{k.metric_name}" for k in LEARNED.values() if not k.parts)
+    rows = await session.execute(_LEVELS, {"keys": keys, "whole": whole,
+                                           "since": now - WINDOW, "now": now})
     return {(s, r, m): float(level) for s, r, m, level in rows}
 
 
@@ -98,7 +129,7 @@ async def detect(session, now: datetime | None = None) -> dict:
     for key, level in levels.items():
         norm = norms.get(key)
         learned = LEARNED.get((key[0], key[2]))
-        out = norm is not None and learned is not None and departs(level, norm, learned.floor)
+        out = norm is not None and learned is not None and departs(level, norm, learned.least_departure)
         current = open_now.pop(key, None)
         if current is not None:
             span, anomaly = current

steward/core/formatting.py

@@ -233,16 +233,23 @@ def required(value: Any, what: str) -> Any:
 _SEVERITY = {"crit": 2, "warn": 1}
 
 
-def worst_reading(readings, thresholds: dict):
+def worst_reading(readings, thresholds: dict, normals: dict | None = None):
     """The reading furthest past its cutoff, as (label, value, grade), or None
     when every reading is within its cutoffs or missing.
 
     `readings` is (label, value, kind) triples, graded like status_class; the
     first of equally bad readings wins, so list them in the order a reader
-    should hear about them."""
+    should hear about them. `normals` ({kind: (Norm, floor)}) are the
+    resource's own normals for this hour: a reading past its cutoff but
+    within its normal is not past it (#5742)."""
+    from steward.core.anomalies import graded
+    normals = normals or {}
     worst, rank = None, 0
     for label, value, kind in readings:
-        grade = status_class(value, kind, thresholds)
+        if kind in normals:
+            grade = graded(value, kind, thresholds, *normals[kind], "%")[0]
+        else:
+            grade = status_class(value, kind, thresholds)
         if _SEVERITY.get(grade, 0) > rank:
             worst, rank = (label, value, grade), _SEVERITY[grade]
     return worst
@@ -327,6 +334,7 @@ METRICS: dict[str, tuple[str, str]] = {
     "load_1m": ("Load (1 min)", "runnable processes, averaged over a minute"),
     "load_5m": ("Load (5 min)", "runnable processes, averaged over 5 minutes"),
     "load_15m": ("Load (15 min)", "runnable processes, averaged over 15 minutes"),
+    "load_pct": ("Load per core", "1-minute load as a % of the host's cores; 100% keeps every core busy"),
     "uptime_secs": ("Uptime", "time since the host booted"),
     "seconds_since_seen": ("Time since the agent reported", "over 3 minutes means the agent is offline"),
     "clock_off_secs": ("Clock off by", "how far the host's clock is from Steward's, either way"),

steward/core/norms.py

@@ -35,7 +35,7 @@ from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import ARRAY, DateTime, String, bindparam, delete, select, text, tuple_
+from sqlalchemy import ARRAY, Boolean, DateTime, String, bindparam, delete, select, text, tuple_
 
 from steward.models.norms import MetricNorm
 
@@ -51,31 +51,45 @@ USUAL = 0.9             # "usual" is this percentile across past days: one odd d
 class Learned:
     """A metric that learns a normal. `floor` is the smallest departure,
     in the metric's own unit, worth calling an anomaly (#5740), so that
-    0.1% → 0.3% of a share or 2 ms → 4 ms of latency is never news.
+    0.1% → 0.3% of a share or 2 ms → 4 ms of latency is never news. It may
+    be a function, read at each pass, for a floor that is a setting (#5742).
 
     `gate` is (metric, least): the resource is judged only while that metric
     of its own averages at least `least` over the window, as a router's
     error share means nothing over a handful of requests (#5741). `least`
     may be a function, read at each pass, for a minimum that is a setting. `since`:
     history before it is not learned from, for a metric whose meaning
-    changed then."""
+    changed then. `parts=False` learns only whole resources, not their parts
+    ("<host>:core3"): a host's CPU has a normal, each of its cores doesn't."""
     source_module: str
     metric_name: str
-    floor: float
+    floor: float | Callable[[], float]
     gate: tuple[str, float | Callable[[], float]] | None = None
     since: datetime | None = None
+    parts: bool = True
+
+    @property
+    def least_departure(self) -> float:
+        return float(self.floor() if callable(self.floor) else self.floor)
 
 
 LEARNED: dict[tuple[str, str], Learned] = {}
 
 
-def learn_normal(source_module: str, metric_name: str, floor: float, *,
+def learn_normal(source_module: str, metric_name: str, floor: float | Callable[[], float], *,
                  gate: tuple[str, float | Callable[[], float]] | None = None,
-                 since: datetime | None = None) -> None:
+                 since: datetime | None = None, parts: bool = True) -> None:
     """Have Steward learn this metric's normal for every resource reporting
     it. A plugin calls this from setup(), as it calls prune_by_age."""
-    LEARNED[(source_module, metric_name)] = Learned(source_module, metric_name, float(floor),
-                                                    gate, since)
+    LEARNED[(source_module, metric_name)] = Learned(
+        source_module, metric_name, floor if callable(floor) else float(floor), gate, since, parts)
+
+
+def floor_setting(app, name: str) -> Callable[[], float]:
+    """A floor read from Settings → Thresholds (norms.floor.<name>) at each
+    pass, so a saved change applies at once."""
+    from steward.core.settings import floor_of
+    return lambda: floor_of(app.config.get("THRESHOLDS", {}), name)
 
 
 @dataclass(frozen=True)
@@ -103,20 +117,23 @@ def slot_of(at: datetime, kind: str) -> int:
 # own slot and its neighbours'.
 _REBUILD = text("""
 WITH learned AS (
-    SELECT key, since FROM unnest(CAST(:keys AS text[]), CAST(:starts AS timestamptz[])) AS k(key, since)
+    SELECT key, since, parts
+    FROM unnest(CAST(:keys AS text[]), CAST(:starts AS timestamptz[]), CAST(:parts AS boolean[]))
+         AS k(key, since, parts)
 ),
 all_hours AS (
     SELECT h.source_module AS s, h.resource_name AS r, h.metric_name AS m, h.bucket AS b,
            COALESCE(h.value_p95, h.value_avg) AS hold, h.value_max AS peak
     FROM plugin_metrics_hourly h
     JOIN learned k ON k.key = h.source_module || '/' || h.metric_name
-    WHERE h.bucket >= k.since
+    WHERE h.bucket >= k.since AND (k.parts OR strpos(h.resource_name, ':') = 0)
     UNION ALL
     SELECT p.source_module, p.resource_name, p.metric_name, date_trunc('hour', p.recorded_at),
            percentile_cont(0.95) WITHIN GROUP (ORDER BY p.value), max(p.value)
     FROM plugin_metrics p
     JOIN learned k ON k.key = p.source_module || '/' || p.metric_name
     WHERE p.recorded_at >= k.since AND p.recorded_at < :this_hour
+      AND (k.parts OR strpos(p.resource_name, ':') = 0)
     GROUP BY 1, 2, 3, 4
 ),
 hours AS (
@@ -157,7 +174,8 @@ SELECT s, r, m, (slot + n.off + slots) % slots, kind,
 FROM slotted CROSS JOIN generate_series(-CAST(:neighbours AS int), CAST(:neighbours AS int)) AS n(off)
 GROUP BY s, r, m, (slot + n.off + slots) % slots, kind, days
 """).bindparams(bindparam("keys", type_=ARRAY(String)),
-               bindparam("starts", type_=ARRAY(DateTime(timezone=True))))
+               bindparam("starts", type_=ARRAY(DateTime(timezone=True))),
+               bindparam("parts", type_=ARRAY(Boolean)))
 
 
 async def rebuild_norms(session, now: datetime | None = None) -> int:
@@ -172,6 +190,7 @@ async def rebuild_norms(session, now: datetime | None = None) -> int:
         "keys": [f"{s}/{m}" for s, m in LEARNED],
         # Each key learns from WEEK_WINDOW back, or from its own `since`.
         "starts": [max(week_since, k.since) if k.since else week_since for k in LEARNED.values()],
+        "parts": [k.parts for k in LEARNED.values()],
         "day_since": now - DAY_WINDOW,
         "this_hour": now.replace(minute=0, second=0, microsecond=0),
         "week_ready": now - timedelta(days=WEEK_MIN_DAYS),

steward/core/settings.py

@@ -100,6 +100,19 @@ DEFAULTS: dict[str, Any] = {
     # A router's 4xx share over fewer requests than this in 15 minutes isn't
     # graded, on the page or for anomalies: one 404 of three is 33% (#5741).
     "traefik.threshold.min_requests": 30,
+    # The smallest departure from a metric's normal worth calling an anomaly
+    # (#5742), on top of being a quarter over its usual hold: a share's floor
+    # is in percentage points, a response time's in ms. 5 points is the
+    # baseline; bursty figures (CPU, load) get 10 so an ordinary build or
+    # backup minute isn't news.
+    "norms.floor.host_cpu_pct": 10,
+    "norms.floor.host_mem_pct": 5,
+    "norms.floor.host_load_pct": 10,
+    "norms.floor.container_cpu_pct": 10,
+    "norms.floor.container_mem_pct": 5,
+    "norms.floor.monitor_ms": 100,
+    "norms.floor.router_4xx_pct": 2,
+    "norms.floor.router_p95_ms": 50,
     # Degraded/critical cutoffs for metric coloring (warn=amber, crit=red).
     # cpu/mem/disk/load are percentages (load is load-per-core %); temp in °C.
     # uptime is "higher is better" so its warn/crit are floors. Ping latency
@@ -540,9 +553,23 @@ def to_thresholds_cfg(settings: dict[str, Any]) -> dict:
         # Fewer requests than this in 15 minutes and a router's 4xx share isn't graded.
         "router_requests": {"warn": g("traefik.threshold.min_requests", 30),
                             "crit": g("traefik.threshold.min_requests", 30), "dir": "low"},
+        # Not a cutoff: each normal's floor (norms.floor.*), read by floor_of().
+        "floors": {name: float(g(f"norms.floor.{name}", DEFAULTS[f"norms.floor.{name}"]))
+                   for name in NORM_FLOORS},
     }
 
 
+NORM_FLOORS = ("host_cpu_pct", "host_mem_pct", "host_load_pct", "container_cpu_pct",
+               "container_mem_pct", "monitor_ms", "router_4xx_pct", "router_p95_ms")
+
+
+def floor_of(thresholds: dict, name: str) -> float:
+    """A normal's floor (one of NORM_FLOORS) from to_thresholds_cfg's output:
+    the smallest departure from it worth calling an anomaly (#5742)."""
+    got = (thresholds.get("floors") or {}).get(name)
+    return float(got if got is not None else DEFAULTS[f"norms.floor.{name}"])
+
+
 _STATUS_STYLE = {"warn": "color:var(--yellow);font-weight:500;",
                  "crit": "color:var(--red);font-weight:500;"}
 
@@ -557,6 +584,12 @@ def threshold_style_for(value: Any, kind: str, thresholds: dict) -> str:
     return _STATUS_STYLE.get(status_class(value, kind, thresholds), "")
 
 
+def grade_style(grade: str | None) -> str:
+    """threshold_style_for's style for a grade already worked out ("warn" /
+    "crit"); empty for anything else."""
+    return _STATUS_STYLE.get(grade or "", "")
+
+
 def to_plugins_cfg(settings: dict[str, Any]) -> dict:
     """Assemble {plugin_name: {...config}} from all plugin.* keys."""
     result = {}