diff --git a/docs/core/alerting.md b/docs/core/alerting.md index b348b2f..45add95 100644 --- a/docs/core/alerting.md +++ b/docs/core/alerting.md @@ -8,7 +8,7 @@ Alert rules evaluate every metric that flows through `record_metric()`. There is When any monitor or plugin calls `record_metric()`: -1. The metric value is written to the `plugin_metrics` table +1. The metric value is written to the `plugin_metrics` table, and `plugin_metric_latest` moves to it if it is the newest sample of that metric 2. All enabled `AlertRule` rows matching `(source_module, resource_name, metric_name)` are loaded 3. Each matching rule is evaluated against the new value 4. If a state transition occurs, an `AlertEvent` row is written @@ -136,3 +136,4 @@ Defined in `steward/models/alerts.py`: - **`alert_states`** — one row per rule, tracks current state and consecutive failure count - **`alert_events`** — append-only log of all state transitions and notification outcomes - **`plugin_metrics`** — all metric values written by any monitor or plugin +- **`plugin_metric_latest`** — the newest sample of each (source, resource, metric). Pages that show "what it is now" read this instead of searching history; retention drops a row once its sample leaves the raw window diff --git a/docs/reference/code-map.md b/docs/reference/code-map.md index 395c3db..3777fe6 100644 --- a/docs/reference/code-map.md +++ b/docs/reference/code-map.md @@ -89,6 +89,7 @@ Quick reference for where key functions, models, and entry points live in the co | `AlertState` | `steward/models/alerts.py` | `alert_states` | | `AlertEvent` | `steward/models/alerts.py` | `alert_events` | | `PluginMetric` | `steward/models/metrics.py` | `plugin_metrics` | +| `PluginMetricLatest` | `steward/models/metrics.py` | `plugin_metric_latest` | | `AnsibleRun` | `steward/models/ansible.py` | `ansible_runs` | | `User` | `steward/models/users.py` | `users` | | `AppSetting` | `steward/models/settings.py` | `app_settings` | diff --git a/plugins/host_agent/metrics_query.py b/plugins/host_agent/metrics_query.py index 1438443..9aa067c 100644 --- a/plugins/host_agent/metrics_query.py +++ b/plugins/host_agent/metrics_query.py @@ -9,7 +9,7 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone -from sqlalchemy import func, or_, select +from sqlalchemy import func, select from steward.core.time_range import bucket_seconds from steward.models.metrics import PluginMetric, PluginMetricHourly @@ -27,30 +27,14 @@ HISTORY_METRICS = ( async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[str, float]]: """{resource_name: {metric: value}} — latest sample per (resource, metric) for - a host + its sub-resources. - - DISTINCT ON picks the newest row per group in one index-ordered pass over - ix_plugin_metrics_module_resource_metric_recorded, instead of a GROUP-BY-max - subquery self-joined back to the table (two passes over the whole history). + a host + its sub-resources, from plugin_metric_latest (#5208). A host whose + agent has gone quiet keeps its last values until they age out of the raw + window; the caller decides staleness from the registration's last_seen_at. """ - rows = (await session.execute( - select(PluginMetric) - .where( - PluginMetric.source_module == SOURCE_MODULE, - or_( - PluginMetric.resource_name == host_name, - PluginMetric.resource_name.like(host_name + ":%"), - ), - ) - .distinct(PluginMetric.resource_name, PluginMetric.metric_name) - .order_by( - PluginMetric.resource_name, - PluginMetric.metric_name, - PluginMetric.recorded_at.desc(), - ) - )).scalars().all() + from steward.core.metric_series import latest_values + out: dict[str, dict[str, float]] = {} - for r in rows: + for r in await latest_values(session, SOURCE_MODULE, host_name, include_children=True): out.setdefault(r.resource_name, {})[r.metric_name] = r.value return out diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 5ee0734..20a90a7 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -13,10 +13,11 @@ import secrets from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for, session from steward.auth.middleware import public_route, require_role from steward.models.users import UserRole -from sqlalchemy import select, func, or_, and_ +from sqlalchemy import select, or_, and_ from datetime import timedelta from steward.core.formatting import worst_reading +from steward.core.metric_series import latest_by_resource, points as metric_points from steward.core.settings import get_setting, public_base_url from steward.core.time_range import parse_range, RANGE_OPTIONS from steward.models.hosts import Host @@ -490,37 +491,21 @@ async def _fleet_rows(session) -> list[dict]: )).scalars().all() } if host_ids else {} - subq = ( - select( - PluginMetric.resource_name, - PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("max_ts"), - ) - .where(PluginMetric.source_module == SOURCE_MODULE) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - ).subquery() - latest_rows = (await session.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) & - (PluginMetric.metric_name == subq.c.metric_name) & - (PluginMetric.recorded_at == subq.c.max_ts), - ).where(PluginMetric.source_module == SOURCE_MODULE) - )).scalars().all() - + host_names = [h.name for h in hosts.values()] + latest_by = await latest_by_resource( + session, SOURCE_MODULE, host_names + [n + ":/" for n in host_names]) latest: dict[str, dict[str, float]] = {} root_disk: dict[str, float] = {} # host_name → root (/) disk % - for row in latest_rows: - if row.resource_name.endswith(":/") and row.metric_name == "disk_used_pct": - root_disk[row.resource_name[:-2]] = row.value - if ":" in row.resource_name: + for resource, metrics in latest_by.items(): + if resource.endswith(":/"): + if "disk_used_pct" in metrics: + root_disk[resource[:-2]] = metrics["disk_used_pct"].value continue - latest.setdefault(row.resource_name, {})[row.metric_name] = row.value + latest[resource] = {m: r.value for m, r in metrics.items()} # Per-host recent series for the inline sparklines (cpu/mem/load host-level, # disk from the root mount), so each metric shows its trend beside the value — # the host-panel at-a-glance, on the fleet widget. 1h window keeps it cheap. - host_names = [h.name for h in hosts.values()] spark_series: dict[str, dict[str, list]] = { n: {"cpu": [], "mem": [], "disk": [], "load": []} for n in host_names } @@ -789,37 +774,25 @@ async def host_vitals(host_id: str): HostAgentRegistration.host_id == host_id))).scalar_one_or_none() latest = await _latest_metrics_for_host(db, host.name) if reg else {} - # Recent series for the at-a-glance sparklines (cpu/mem/load host-level, - # disk from the root mount sub-resource). + # The last 6 h for the at-a-glance sparklines (cpu/mem/load host-level, + # disk from the root mount), averaged into ~120 points in SQL, so the + # line spans the whole window and only what is drawn is fetched. series_raw: dict[str, list[float]] = {"cpu": [], "mem": [], "disk": [], "load": []} if reg: since = datetime.now(timezone.utc) - timedelta(hours=6) - spark_rows = (await db.execute( - select(PluginMetric).where( - PluginMetric.source_module == SOURCE_MODULE, - PluginMetric.recorded_at >= since, - or_( - and_(PluginMetric.resource_name == host.name, - PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct", "load_1m"))), - and_(PluginMetric.resource_name == host.name + ":/", - PluginMetric.metric_name == "disk_used_pct"), - ), - ).order_by(PluginMetric.recorded_at) - )).scalars().all() - for r in spark_rows: - if r.resource_name == host.name: - key = {"cpu_pct": "cpu", "mem_used_pct": "mem", "load_1m": "load"}.get(r.metric_name) - if key: - series_raw[key].append(r.value) - else: - series_raw["disk"].append(r.value) + for key, resource, metric in (("cpu", host.name, "cpu_pct"), + ("mem", host.name, "mem_used_pct"), + ("load", host.name, "load_1m"), + ("disk", host.name + ":/", "disk_used_pct")): + series_raw[key] = [v for _t, v in await metric_points( + db, SOURCE_MODULE, resource, metric, since, target=120)] from steward.core.status import sparkline_svg hostlvl = latest.get(host.name, {}) # At-a-glance disk = the root filesystem (what people actually care about), # not the "worst" mount which is opaque. Worst stays on the full-metrics page. disk_root = (latest.get(host.name + ":/", {}) or {}).get("disk_used_pct") - sparks = {k: sparkline_svg(v[-120:]) for k, v in series_raw.items()} + sparks = {k: sparkline_svg(v) for k, v in series_raw.items()} # Normalize load by core count so it's comparable across hardware (load/core # as %). Cores derived from the per-core CPU sub-resources already collected. cores = sum(1 for k in latest if k.startswith(host.name + ":core")) diff --git a/plugins/snmp/routes.py b/plugins/snmp/routes.py index 6157338..41d2823 100644 --- a/plugins/snmp/routes.py +++ b/plugins/snmp/routes.py @@ -3,10 +3,11 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from quart import Blueprint, current_app, render_template, request -from sqlalchemy import func, select +from sqlalchemy import select from steward.auth.middleware import require_role from steward.models.users import UserRole +from steward.core.metric_series import latest_by_resource from steward.models.metrics import PluginMetric snmp_bp = Blueprint("snmp", __name__, template_folder="templates") @@ -14,35 +15,8 @@ snmp_bp = Blueprint("snmp", __name__, template_folder="templates") async def _latest_readings(db, device_names: list[str]) -> dict[str, dict[str, float]]: """Return {device_name: {label: latest_value}} for all configured devices.""" - if not device_names: - return {} - - # Latest value per (resource_name, metric_name) pair - subq = ( - select( - PluginMetric.resource_name, - PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("latest_at"), - ) - .where(PluginMetric.source_module == "snmp") - .where(PluginMetric.resource_name.in_(device_names)) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - .subquery() - ) - result = await db.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) - & (PluginMetric.metric_name == subq.c.metric_name) - & (PluginMetric.recorded_at == subq.c.latest_at), - ).where(PluginMetric.source_module == "snmp") - ) - rows = result.scalars().all() - - out: dict[str, dict[str, float]] = {} - for row in rows: - out.setdefault(row.resource_name, {})[row.metric_name] = row.value - return out + latest = await latest_by_resource(db, "snmp", device_names) + return {name: {m: r.value for m, r in metrics.items()} for name, metrics in latest.items()} async def _history(db, device_name: str, hours: int = 24) -> dict[str, list]: diff --git a/steward/core/alerts.py b/steward/core/alerts.py index c40f753..db0c452 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -18,7 +18,7 @@ from steward.models.alerts import ( MaintenanceWindow, ) from steward.core.formatting import human_duration -from steward.models.metrics import PluginMetric +from steward.models.metrics import PluginMetric, PluginMetricLatest if TYPE_CHECKING: from quart import Quart @@ -155,6 +155,34 @@ def _after_commit(session: AsyncSession, fn, *args) -> None: info[_AFTER_COMMIT].append((fn, args)) +LATEST_CHUNK = 1000 + + +async def _keep_latest(session: AsyncSession, rows: list[PluginMetric]) -> None: + """Move each key's row in plugin_metric_latest to the newest sample in + `rows`. A row only moves forward, so a backlog arriving after newer samples + leaves it alone. Keys go in sorted order, so two batches sharing keys lock + them in the same order and cannot deadlock.""" + from sqlalchemy.dialects.postgresql import insert as pg_insert + + newest: dict[tuple[str, str, str], PluginMetric] = {} + for row in rows: + key = (row.source_module, row.resource_name, row.metric_name) + if key not in newest or row.recorded_at > newest[key].recorded_at: + newest[key] = row + values = [{"source_module": k[0], "resource_name": k[1], "metric_name": k[2], + "value": newest[k].value, "recorded_at": newest[k].recorded_at} + for k in sorted(newest)] + # Five parameters a row; chunks keep a statement well under asyncpg's 32767. + for i in range(0, len(values), LATEST_CHUNK): + stmt = pg_insert(PluginMetricLatest).values(values[i:i + LATEST_CHUNK]) + await session.execute(stmt.on_conflict_do_update( + index_elements=["source_module", "resource_name", "metric_name"], + set_={"value": stmt.excluded.value, "recorded_at": stmt.excluded.recorded_at}, + where=PluginMetricLatest.recorded_at <= stmt.excluded.recorded_at, + )) + + async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases=None) -> None: """Write metric rows and evaluate the alert rules that watch them. @@ -182,6 +210,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases row.recorded_at = now session.add_all(rows) await session.flush() + await _keep_latest(session, rows) mark("flush") by_key: dict[tuple[str, str, str], list[PluginMetric]] = {} diff --git a/steward/core/metric_series.py b/steward/core/metric_series.py index 256beeb..c177bfc 100644 --- a/steward/core/metric_series.py +++ b/steward/core/metric_series.py @@ -11,7 +11,7 @@ from datetime import datetime, timedelta, timezone from sqlalchemy import func, or_, select, text from steward.core.time_range import bucket_seconds -from steward.models.metrics import PluginMetric, PluginMetricHourly +from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest # Loose index scans: step through the distinct leading values of the # (source_module, resource_name, …) index one probe at a time instead of @@ -48,26 +48,49 @@ async def resources(db, source_module: str) -> list[str]: return [v for (v,) in await db.execute(_RESOURCES_SQL, {"src": source_module})] +def _resource_match(resource: str, include_children: bool): + match = PluginMetricLatest.resource_name == resource + if include_children: + like = resource.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + ":%" + match = or_(match, PluginMetricLatest.resource_name.like(like, escape="\\")) + return match + + async def latest_values(db, source_module: str, resource: str, *, - include_children: bool = False) -> list[PluginMetric]: + include_children: bool = False) -> list[PluginMetricLatest]: """The newest sample of each metric for a resource, sorted by (resource, metric). include_children also returns the resource's sub-resources, named - ":" (a host's mounts, interfaces, sensors). + ":" (a host's mounts, interfaces, sensors). Read from + plugin_metric_latest, so the cost is the resource's metric count, not its + history. """ - match = PluginMetric.resource_name == resource - if include_children: - like = resource.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + ":%" - match = or_(match, PluginMetric.resource_name.like(like, escape="\\")) return list((await db.execute( - select(PluginMetric) - .where(PluginMetric.source_module == source_module, match) - .distinct(PluginMetric.resource_name, PluginMetric.metric_name) - .order_by(PluginMetric.resource_name, PluginMetric.metric_name, - PluginMetric.recorded_at.desc()) + select(PluginMetricLatest) + .where(PluginMetricLatest.source_module == source_module, + _resource_match(resource, include_children)) + .order_by(PluginMetricLatest.resource_name, PluginMetricLatest.metric_name) )).scalars()) +async def latest_by_resource(db, source_module: str, resources, metrics=None, + ) -> dict[str, dict[str, PluginMetricLatest]]: + """{resource: {metric: newest sample}} for several resources of a source, + optionally only the named metrics. A resource with no samples is absent.""" + resources = list(resources) + if not resources: + return {} + q = select(PluginMetricLatest).where( + PluginMetricLatest.source_module == source_module, + PluginMetricLatest.resource_name.in_(resources)) + if metrics is not None: + q = q.where(PluginMetricLatest.metric_name.in_(list(metrics))) + out: dict[str, dict[str, PluginMetricLatest]] = {} + for r in (await db.execute(q)).scalars(): + out.setdefault(r.resource_name, {})[r.metric_name] = r + return out + + async def points(db, source_module: str, resource: str, metric: str, since: datetime, *, target: int = 120) -> list[tuple[datetime, float]]: """(time, average) points for one metric since `since`, oldest first. diff --git a/steward/core/metrics_retention.py b/steward/core/metrics_retention.py index 4f02e52..8be4085 100644 --- a/steward/core/metrics_retention.py +++ b/steward/core/metrics_retention.py @@ -49,7 +49,7 @@ async def rollup_plugin_metrics( from sqlalchemy import delete, func, select from sqlalchemy.dialects.postgresql import insert as pg_insert - from steward.models.metrics import PluginMetric, PluginMetricHourly + from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest if now is None: now = datetime.now(timezone.utc) @@ -97,6 +97,11 @@ async def rollup_plugin_metrics( await session.execute( delete(PluginMetric).where(PluginMetric.recorded_at < raw_cutoff) ) + # A latest value goes with its raw history: a resource that stopped + # reporting a raw window ago is no longer "now" anywhere. + latest_pruned = (await session.execute( + delete(PluginMetricLatest).where(PluginMetricLatest.recorded_at < raw_cutoff) + )).rowcount or 0 # ── 2. Prune rolled-up rows beyond the rollup window ── rollup_cutoff = now - timedelta(days=rollup_days) @@ -109,4 +114,5 @@ async def rollup_plugin_metrics( "buckets_rolled": rolled, "raw_rows_rolled": rolled_rows, "rollup_pruned": rollup_pruned, + "latest_pruned": latest_pruned, } diff --git a/steward/core/renames.py b/steward/core/renames.py index 84ed589..52a19c0 100644 --- a/steward/core/renames.py +++ b/steward/core/renames.py @@ -11,7 +11,7 @@ from sqlalchemy import and_, delete, func, or_, update from sqlalchemy.ext.asyncio import AsyncSession from steward.models.alerts import AlertRule, MaintenanceWindow -from steward.models.metrics import PluginMetric, PluginMetricHourly +from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest # What a host's name keys, and how each source names a part of the host: # host_agent "web01:/var", "web01:core3"; docker "web01/nginx". @@ -47,7 +47,7 @@ async def carry_rename(session: AsyncSession, sources, old: str, new: str) -> di if not old or old == new: return {} sources = tuple(sources) - for model in (PluginMetric, PluginMetricHourly): + for model in (PluginMetric, PluginMetricHourly, PluginMetricLatest): await session.execute(delete(model).where( _matches(model.source_module, model.resource_name, sources, new))) moved = {} @@ -58,6 +58,8 @@ async def carry_rename(session: AsyncSession, sources, old: str, new: str) -> di ("metric_rows", PluginMetric, PluginMetric.source_module, PluginMetric.resource_name), ("hourly_rows", PluginMetricHourly, PluginMetricHourly.source_module, PluginMetricHourly.resource_name), + ("latest_rows", PluginMetricLatest, PluginMetricLatest.source_module, + PluginMetricLatest.resource_name), ): result = await session.execute( update(model).where(_matches(src, res, sources, old)) diff --git a/steward/hosts/service.py b/steward/hosts/service.py index b6917e5..f8a0344 100644 --- a/steward/hosts/service.py +++ b/steward/hosts/service.py @@ -144,92 +144,50 @@ async def host_monitors(db, host_id: str) -> list[dict]: async def agent_metrics_by_host(db, host_names: list[str]) -> dict[str, dict]: - """Latest agent cpu/mem per host name from the generic PluginMetric table. + """Latest agent cpu/mem per host name, from plugin_metric_latest. - Read directly (PluginMetric is a core model) so the Hosts list can show an + Core-safe (no host_agent import), so the Hosts list can show an at-a-glance agent column without importing the host_agent plugin. """ - from steward.models.metrics import PluginMetric - if not host_names: - return {} - wanted = ("cpu_pct", "mem_used_pct") - subq = ( - select( - PluginMetric.resource_name, PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("m"), - ) - .where( - PluginMetric.source_module == "host_agent", - PluginMetric.metric_name.in_(wanted), - PluginMetric.resource_name.in_(host_names), - ) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - ).subquery() - rows = (await db.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) - & (PluginMetric.metric_name == subq.c.metric_name) - & (PluginMetric.recorded_at == subq.c.m), - ) - )).scalars().all() - out: dict[str, dict] = {} - for r in rows: - out.setdefault(r.resource_name, {})[r.metric_name] = r.value - return out + from steward.core.metric_series import latest_by_resource + latest = await latest_by_resource(db, "host_agent", host_names, ("cpu_pct", "mem_used_pct")) + return {name: {m: r.value for m, r in metrics.items()} for name, metrics in latest.items()} async def agent_overview_by_host( db, host_names: list[str], stale_after_seconds: int = 180, ) -> dict[str, dict]: - """Per-host agent glance (cpu/mem/disk + freshness) from PluginMetric. + """Per-host agent glance (cpu/mem/disk + freshness) from plugin_metric_latest. {host_name: {cpu_pct, mem_used_pct, disk_used_pct_worst, disk_root, last_at, fresh}} for each host with any agent samples. Core-safe (no host_agent import). Freshness compares the latest host-level sample against the host_agent plugin's stale window (stale_after_seconds; - callers pass agent_stale_after_seconds(app.config)). + callers pass agent_stale_after_seconds(app.config)). A host whose agent has + gone quiet keeps its last values with fresh=False (#5208). """ - from steward.models.metrics import PluginMetric + from steward.core.metric_series import latest_by_resource if not host_names: return {} now = datetime.now(timezone.utc) stale_after = int(stale_after_seconds) - resources = list(host_names) + [h + ":/" for h in host_names] - subq = ( - select( - PluginMetric.resource_name, PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("m"), - ) - .where( - PluginMetric.source_module == "host_agent", - PluginMetric.resource_name.in_(resources), - PluginMetric.metric_name.in_(AGENT_GLANCE_METRICS), - ) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - ).subquery() - rows = (await db.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) - & (PluginMetric.metric_name == subq.c.metric_name) - & (PluginMetric.recorded_at == subq.c.m), - ).where(PluginMetric.source_module == "host_agent") - )).scalars().all() + latest = await latest_by_resource( + db, "host_agent", list(host_names) + [h + ":/" for h in host_names], + set(AGENT_GLANCE_METRICS) | {"disk_used_pct"}) out: dict[str, dict] = {} - for r in rows: - if r.resource_name.endswith(":/") and r.metric_name == "disk_used_pct": - out.setdefault(r.resource_name[:-2], {})["disk_root"] = r.value - elif r.resource_name in host_names: - d = out.setdefault(r.resource_name, {}) - d[r.metric_name] = r.value - if d.get("_ts") is None or r.recorded_at > d["_ts"]: - d["_ts"] = r.recorded_at - for d in out.values(): - ts = d.pop("_ts", None) + for name in host_names: + own = {m: r for m, r in latest.get(name, {}).items() if m in AGENT_GLANCE_METRICS} + root = latest.get(name + ":/", {}).get("disk_used_pct") + if not own and root is None: + continue + d = {m: r.value for m, r in own.items()} + if root is not None: + d["disk_root"] = root.value + ts = max((r.recorded_at for r in own.values()), default=None) d["last_at"] = ts d["fresh"] = bool(ts and (now - ts).total_seconds() <= stale_after) + out[name] = d return out diff --git a/steward/migrations/versions/0030_plugin_metric_latest.py b/steward/migrations/versions/0030_plugin_metric_latest.py new file mode 100644 index 0000000..b689b1f --- /dev/null +++ b/steward/migrations/versions/0030_plugin_metric_latest.py @@ -0,0 +1,55 @@ +"""The newest sample of each metric, kept beside plugin_metrics (#5208) + +Revision ID: 0030_plugin_metric_latest +Revises: 0029_alert_no_data +Create Date: 2026-10-07 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0030_plugin_metric_latest" +down_revision: Union[str, None] = "0029_alert_no_data" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "plugin_metric_latest", + sa.Column("source_module", sa.String(64), primary_key=True), + sa.Column("resource_name", sa.String(255), primary_key=True), + sa.Column("metric_name", sa.String(128), primary_key=True), + sa.Column("value", sa.Float, nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + ) + # Seed from history without a full scan: step through the distinct keys of + # the (source, resource, metric, recorded_at) index one probe at a time, + # then take each key's newest row with a backward probe of the same index. + op.execute(""" + WITH RECURSIVE k AS ( + (SELECT source_module, resource_name, metric_name FROM plugin_metrics + ORDER BY source_module, resource_name, metric_name LIMIT 1) + UNION ALL + SELECT n.source_module, n.resource_name, n.metric_name + FROM k, LATERAL ( + SELECT p.source_module, p.resource_name, p.metric_name FROM plugin_metrics p + WHERE (p.source_module, p.resource_name, p.metric_name) + > (k.source_module, k.resource_name, k.metric_name) + ORDER BY p.source_module, p.resource_name, p.metric_name LIMIT 1 + ) n + ) + INSERT INTO plugin_metric_latest + (source_module, resource_name, metric_name, value, recorded_at) + SELECT k.source_module, k.resource_name, k.metric_name, l.value, l.recorded_at + FROM k, LATERAL ( + SELECT p.value, p.recorded_at FROM plugin_metrics p + WHERE p.source_module = k.source_module AND p.resource_name = k.resource_name + AND p.metric_name = k.metric_name + ORDER BY p.recorded_at DESC LIMIT 1 + ) l + """) + + +def downgrade() -> None: + op.drop_table("plugin_metric_latest") diff --git a/steward/models/metrics.py b/steward/models/metrics.py index 0cd35c6..93f68b2 100644 --- a/steward/models/metrics.py +++ b/steward/models/metrics.py @@ -30,6 +30,25 @@ class PluginMetric(Base): ) +class PluginMetricLatest(Base): + """The newest sample of each (source, resource, metric), kept beside + plugin_metrics by record_metrics so "what is it now" never scans history. + + Postgres has no skip scan, so a newest-per-key query over plugin_metrics + reads every raw sample in the window: on a seven-day window that took a + dashboard widget a minute (#5208). This table is one row per key. Retention + drops a row once its sample has aged out of the raw window, so a resource + that stopped reporting disappears when its raw history does. + """ + __tablename__ = "plugin_metric_latest" + + source_module: Mapped[str] = mapped_column(String(64), primary_key=True) + resource_name: Mapped[str] = mapped_column(String(255), primary_key=True) + metric_name: Mapped[str] = mapped_column(String(128), primary_key=True) + value: Mapped[float] = mapped_column(Float, nullable=False) + recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + class PluginMetricHourly(Base): """Hourly rollup of plugin_metrics — the coarse series that retention rolls raw samples into before pruning them, so multi-day/week history stays cheap. diff --git a/steward/static/dash-cells.js b/steward/static/dash-cells.js new file mode 100644 index 0000000..fe6d200 --- /dev/null +++ b/steward/static/dash-cells.js @@ -0,0 +1,90 @@ +/* Dashboard widget cells: say when a cell is loading, when it could not load, + and how fresh what it shows is (#5208). + + Each .dash-cell-body is an htmx target with a request deadline + (hx-request timeout), so every fetch ends in a swap or one of the error + events below. A cell that polls keeps polling through htmx's own `every` + trigger; a cell that loads once is retried here, re-armed from the failure + handler of each attempt, so a failed retry arms the next one. */ +(function () { + var RETRY_SECONDS = 30; + + function hhmm(d) { + return d.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); + } + + function asofOf(cell) { + var box = cell.closest('.grid-stack-item-content, .dash-cell'); + return box ? box.querySelector('.widget-asof') : null; + } + + function isCell(el) { + return el && el.classList && el.classList.contains('dash-cell-body'); + } + + function scheduleRetry(cell) { + clearTimeout(cell._retry); + if (cell.dataset.poll === '1') return; // its own poll is the retry + cell._retry = setTimeout(function () { + htmx.trigger(cell, 'retry'); + }, RETRY_SECONDS * 1000); + } + + function failed(cell, why) { + var asof = asofOf(cell); + var when = cell.dataset.loadedAt ? new Date(+cell.dataset.loadedAt) : null; + if (when) { + // Keep what it showed; say it is no longer current. + if (asof) { + asof.textContent = "couldn't refresh · as of " + hhmm(when); + asof.classList.add('widget-asof-failed'); + asof.title = why; + } + } else { + var retryLine = cell.dataset.poll === '1' + ? 'Trying again with the next refresh.' + : 'Trying again in ' + RETRY_SECONDS + 's.'; + cell.innerHTML = ''; + var box = document.createElement('div'); + box.className = 'cell-state cell-state-failed'; + var line = document.createElement('div'); + line.textContent = "Couldn't load this widget (" + why + '). ' + retryLine; + var btn = document.createElement('button'); + btn.type = 'button'; + btn.className = 'btn btn-sm'; + btn.textContent = 'Retry now'; + btn.addEventListener('click', function () { + clearTimeout(cell._retry); + htmx.trigger(cell, 'retry'); + }); + box.appendChild(line); + box.appendChild(btn); + cell.appendChild(box); + } + scheduleRetry(cell); + } + + document.body.addEventListener('htmx:afterSwap', function (e) { + var cell = e.detail.target; + if (!isCell(cell)) return; + clearTimeout(cell._retry); + var now = new Date(); + cell.dataset.loadedAt = String(now.getTime()); + var asof = asofOf(cell); + if (asof) { + asof.textContent = 'as of ' + hhmm(now); + asof.classList.remove('widget-asof-failed'); + asof.title = 'Loaded ' + now.toLocaleString(); + } + }); + + document.body.addEventListener('htmx:responseError', function (e) { + if (isCell(e.detail.target)) failed(e.detail.target, 'HTTP ' + e.detail.xhr.status); + }); + document.body.addEventListener('htmx:sendError', function (e) { + if (isCell(e.detail.target)) failed(e.detail.target, 'no connection'); + }); + document.body.addEventListener('htmx:timeout', function (e) { + if (isCell(e.detail.target)) failed(e.detail.target, 'timed out'); + }); +})(); diff --git a/steward/static/steward.css b/steward/static/steward.css index 12b220d..e49fecb 100644 --- a/steward/static/steward.css +++ b/steward/static/steward.css @@ -371,6 +371,12 @@ body.dash-editing .widget-drawer { transform:translateY(0); } .widget-header { display:flex; justify-content:space-between; align-items:center; margin-bottom:0.75rem; } .widget-title { font-family:var(--fs-font-body); font-size:var(--fs-size-tiny); font-weight:500; color:var(--text-muted); text-transform:uppercase; letter-spacing:var(--fs-tracking-tiny); } .widget-link { font-size:0.8rem; color:var(--text-muted); } +/* A widget cell's own state (#5208): loading, couldn't load, and how fresh + the shown data is. Set by static/dash-cells.js. */ +.widget-asof { font-size:var(--fs-size-tiny); color:var(--text-dim); white-space:nowrap; } +.widget-asof-failed { color:var(--yellow); } +.cell-state { font-size:0.85rem; color:var(--text-muted); padding:0.25rem 0; } +.cell-state-failed { display:flex; flex-direction:column; align-items:flex-start; gap:0.5rem; color:var(--yellow); } .widget-link:hover { color:var(--text); } /* Time range selector */ diff --git a/steward/templates/dashboard/_grid_item.html b/steward/templates/dashboard/_grid_item.html index e5562b3..430e019 100644 --- a/steward/templates/dashboard/_grid_item.html +++ b/steward/templates/dashboard/_grid_item.html @@ -11,6 +11,7 @@ {{ item.title }} + Details → {% if can_edit %}