bvandeusenandClaude Opus 5.5
b8f73270e7
The dashboard's host widgets took 18-60 s. Every "latest reading" query (hosts overview, host-agent fleet, host page, vitals, SNMP, MCP get_host) searched plugin_metrics for the newest row per metric. Postgres has no skip scan, so each one read every raw sample in the 7-day window. - plugin_metric_latest: one row per (source, resource, metric). record_metrics upserts it in the same transaction, moving forward only and in sorted key order. Migration 0030 seeds it by stepping the existing index. - Retention drops a latest row once its sample leaves the raw window, so a resource that stopped reporting disappears when its history does, as before. Renames carry it along. - A quiet host keeps its last values, marked stale. A time-bounded query would have shown it as missing. - Vitals sparklines span the 6 h they claim, averaged in SQL to ~120 points, instead of fetching 6 h raw and drawing the last 120 samples. - Dashboard and share-view cells: "Loading…" until the first answer; a 20 s deadline per fetch; "Couldn't load (why)" with Retry now and an automatic retry for cells that don't poll; "as of HH:MM" in the header, and "couldn't refresh · as of" when a refresh fails over shown data. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
"""Bound plugin_metrics growth: roll old raw samples up to hourly, prune the rest.
|
||
|
||
plugin_metrics grows by (sources × resources × sample cadence) — host agents push
|
||
host-level + per-core/mount/iface sub-resources every ~30s, so a fleet accrues
|
||
millions of rows. We keep raw samples for a short window, aggregate everything
|
||
older into hourly averages (plugin_metrics_hourly) and delete the raw rows, then
|
||
prune hourly beyond a longer window. Charts read raw for the recent part of a
|
||
range and hourly for the older part.
|
||
|
||
Driven by the core cleanup task (steward.core.cleanup). Runs inside the caller's
|
||
open transaction; never opens or commits its own.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta
|
||
|
||
|
||
def _hour_floor(dt: datetime) -> datetime:
|
||
"""Truncate a datetime down to the start of its hour (drops min/sec/µs)."""
|
||
return dt.replace(minute=0, second=0, microsecond=0)
|
||
|
||
|
||
def _rollup_cutoff(now: datetime, raw_days: int) -> datetime:
|
||
"""Hour-aligned boundary below which raw metrics get rolled up + deleted.
|
||
|
||
Aligning to the hour means we only roll up *whole* elapsed hours — a bucket
|
||
is never split across the keep/roll boundary, so a re-run can't produce a
|
||
partial-then-complete duplicate for the same hour.
|
||
"""
|
||
return _hour_floor(now - timedelta(days=raw_days))
|
||
|
||
|
||
async def rollup_plugin_metrics(
|
||
session,
|
||
*,
|
||
raw_days: int,
|
||
rollup_days: int,
|
||
now: datetime | None = None,
|
||
) -> dict:
|
||
"""Roll up + prune plugin_metrics. Returns a counts dict for logging.
|
||
|
||
1. Aggregate plugin_metrics older than the (hour-aligned) raw window into
|
||
plugin_metrics_hourly (avg/max per source/resource/metric/hour), upserting
|
||
so a re-run is idempotent, then delete those raw rows.
|
||
2. Prune rolled-up rows older than the rollup window.
|
||
"""
|
||
from datetime import timezone
|
||
|
||
from sqlalchemy import delete, func, select
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
|
||
from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest
|
||
|
||
if now is None:
|
||
now = datetime.now(timezone.utc)
|
||
|
||
rolled = rolled_rows = rollup_pruned = 0
|
||
|
||
# ── 1. Roll up raw metrics older than the raw window into hourly buckets ──
|
||
raw_cutoff = _rollup_cutoff(now, raw_days)
|
||
hour = func.date_trunc("hour", PluginMetric.recorded_at)
|
||
agg = (
|
||
select(
|
||
PluginMetric.source_module,
|
||
PluginMetric.resource_name,
|
||
PluginMetric.metric_name,
|
||
hour.label("bucket"),
|
||
func.avg(PluginMetric.value).label("value_avg"),
|
||
func.max(PluginMetric.value).label("value_max"),
|
||
func.count().label("sample_count"),
|
||
)
|
||
.where(PluginMetric.recorded_at < raw_cutoff)
|
||
.group_by(
|
||
PluginMetric.source_module, PluginMetric.resource_name,
|
||
PluginMetric.metric_name, hour,
|
||
)
|
||
)
|
||
for r in (await session.execute(agg)).all():
|
||
avg_v = float(r.value_avg or 0.0)
|
||
max_v = float(r.value_max or 0.0)
|
||
cnt = int(r.sample_count or 0)
|
||
await session.execute(
|
||
pg_insert(PluginMetricHourly)
|
||
.values(
|
||
source_module=r.source_module, resource_name=r.resource_name,
|
||
metric_name=r.metric_name, bucket=r.bucket,
|
||
value_avg=avg_v, value_max=max_v, sample_count=cnt,
|
||
)
|
||
.on_conflict_do_update(
|
||
constraint="uq_plugin_metrics_hourly_bucket",
|
||
set_={"value_avg": avg_v, "value_max": max_v, "sample_count": cnt},
|
||
)
|
||
)
|
||
rolled += 1
|
||
rolled_rows += cnt
|
||
if rolled:
|
||
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)
|
||
res = await session.execute(
|
||
delete(PluginMetricHourly).where(PluginMetricHourly.bucket < rollup_cutoff)
|
||
)
|
||
rollup_pruned = res.rowcount or 0
|
||
|
||
return {
|
||
"buckets_rolled": rolled,
|
||
"raw_rows_rolled": rolled_rows,
|
||
"rollup_pruned": rollup_pruned,
|
||
"latest_pruned": latest_pruned,
|
||
}
|