diff --git a/plugins/host_agent/metrics_query.py b/plugins/host_agent/metrics_query.py index 9b18843..1438443 100644 --- a/plugins/host_agent/metrics_query.py +++ b/plugins/host_agent/metrics_query.py @@ -7,10 +7,12 @@ defined") guard. """ from __future__ import annotations +from datetime import datetime, timedelta, timezone + from sqlalchemy import func, or_, select from steward.core.time_range import bucket_seconds -from steward.models.metrics import PluginMetric +from steward.models.metrics import PluginMetric, PluginMetricHourly SOURCE_MODULE = "host_agent" @@ -53,33 +55,58 @@ async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[st return out -async def _history_for_host(session, host_name: str, since) -> dict[str, list]: +async def _history_for_host(session, host_name: str, since, *, raw_days: int = 7) -> dict[str, list]: """{metric: [[epoch_ms, avg_value], …]} host-level series since `since`. - Buckets + averages in SQL (date_bin to ~120 buckets) so we return a readable - point count instead of shipping every raw sample to Python and downsampling - there — a 30d range was reading hundreds of thousands of rows per load. The - bucket width is epoch-aligned so the x axis is stable across refreshes. - Epoch-ms x values feed a linear chart axis (no Chart.js date adapter). + Retention rolls raw plugin_metrics older than `raw_days` into hourly averages + (plugin_metrics_hourly) and deletes the raw rows. So we read the rollup for the + part of the range older than that boundary and raw (bucket-averaged in SQL, + capped to ≤1h to match the rollup) for the recent part — never shipping raw + samples to Python. The two windows don't overlap, so appending hourly-then-raw + keeps each metric's series time-ordered. Epoch-ms x feeds a linear chart axis. """ - width_s = bucket_seconds(since, 120) - bucket = func.date_bin( - func.make_interval(0, 0, 0, 0, 0, 0, width_s), # width_s seconds + now = datetime.now(timezone.utc) + raw_cutoff = (now - timedelta(days=raw_days)).replace(minute=0, second=0, microsecond=0) + series: dict[str, list] = {m: [] for m in HISTORY_METRICS} + + # ── Older than the raw window: the hourly rollup ── + if since < raw_cutoff: + hourly = (await session.execute( + select(PluginMetricHourly.metric_name, PluginMetricHourly.bucket, + PluginMetricHourly.value_avg) + .where( + PluginMetricHourly.source_module == SOURCE_MODULE, + PluginMetricHourly.resource_name == host_name, + PluginMetricHourly.metric_name.in_(HISTORY_METRICS), + PluginMetricHourly.bucket >= since, + PluginMetricHourly.bucket < raw_cutoff, + ) + .order_by(PluginMetricHourly.bucket) + )).all() + for metric_name, bucket, avg in hourly: + series[metric_name].append([int(bucket.timestamp() * 1000), round(float(avg), 2)]) + + # ── Recent part: raw, bucket-averaged in SQL ── + raw_since = since if since >= raw_cutoff else raw_cutoff + width_s = bucket_seconds(raw_since, 120) + if since < raw_cutoff: + width_s = min(width_s, 3600) # ≤ 1h so it matches the hourly portion + rbucket = func.date_bin( + func.make_interval(0, 0, 0, 0, 0, 0, width_s), PluginMetric.recorded_at, func.to_timestamp(0), # epoch origin ).label("bucket") - rows = (await session.execute( - select(PluginMetric.metric_name, bucket, func.avg(PluginMetric.value)) + raw = (await session.execute( + select(PluginMetric.metric_name, rbucket, func.avg(PluginMetric.value)) .where( PluginMetric.source_module == SOURCE_MODULE, PluginMetric.resource_name == host_name, PluginMetric.metric_name.in_(HISTORY_METRICS), - PluginMetric.recorded_at >= since, + PluginMetric.recorded_at >= raw_since, ) - .group_by(PluginMetric.metric_name, bucket) - .order_by(bucket) + .group_by(PluginMetric.metric_name, rbucket) + .order_by(rbucket) )).all() - series: dict[str, list] = {m: [] for m in HISTORY_METRICS} - for metric_name, b, avg in rows: + for metric_name, b, avg in raw: series[metric_name].append([int(b.timestamp() * 1000), round(float(avg), 2)]) return series diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 14dcfe7..b63ae1a 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -14,7 +14,7 @@ from steward.models.users import UserRole from sqlalchemy import select, func, or_, and_ from datetime import timedelta -from steward.core.settings import public_base_url +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 from steward.models.metrics import PluginMetric @@ -637,7 +637,8 @@ async def host_detail_charts(host_id: str): select(Host).where(Host.id == host_id))).scalar_one_or_none() if host is None: return "", 404 - series = await _history_for_host(session, host.name, since) + raw_days = int(await get_setting(session, "metrics.retention.raw_days") or 7) + series = await _history_for_host(session, host.name, since, raw_days=raw_days) return await render_template("_host_charts.html", series=series, range_key=range_key) diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index 563a918..e1f9975 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -6,7 +6,6 @@ from typing import TYPE_CHECKING from sqlalchemy import delete from steward.models.monitors import MonitorResult -from steward.models.metrics import PluginMetric from steward.models.ansible import AnsibleRun if TYPE_CHECKING: @@ -26,7 +25,6 @@ async def run_cleanup(app: "Quart") -> None: async with session.begin(): for model, ts_col in [ (MonitorResult, MonitorResult.checked_at), - (PluginMetric, PluginMetric.recorded_at), (AnsibleRun, AnsibleRun.started_at), ]: result = await session.execute( @@ -35,9 +33,29 @@ async def run_cleanup(app: "Quart") -> None: if result.rowcount: logger.info(f"Pruned {result.rowcount} rows from {model.__tablename__}") + # plugin_metrics is NOT blanket-deleted here — it's rolled up to hourly + # then pruned, so multi-week host history stays cheap. + await _run_metrics_retention(session, now) await _run_docker_retention(session, now) +async def _run_metrics_retention(session, now: datetime) -> None: + """Roll up + prune plugin_metrics (raw → hourly → gone). Windows read fresh + from settings each run (rule 25 — UI change takes effect next cleanup, no + restart). get_setting's SELECT autobegins, so read inside the begin block.""" + from steward.core.metrics_retention import rollup_plugin_metrics + from steward.core.settings import get_setting + + async with session.begin(): + raw_days = int(await get_setting(session, "metrics.retention.raw_days") or 7) + rollup_days = int(await get_setting(session, "metrics.retention.rollup_days") or 90) + counts = await rollup_plugin_metrics( + session, raw_days=raw_days, rollup_days=rollup_days, now=now, + ) + if counts and any(counts.values()): + logger.info("Metrics retention: %s", counts) + + async def _run_docker_retention(session, now: datetime) -> None: """Drive the docker plugin's rollup + prune via its capability, if loaded. diff --git a/steward/core/metrics_retention.py b/steward/core/metrics_retention.py new file mode 100644 index 0000000..4f02e52 --- /dev/null +++ b/steward/core/metrics_retention.py @@ -0,0 +1,112 @@ +"""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 + + 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) + ) + + # ── 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, + } diff --git a/steward/core/settings.py b/steward/core/settings.py index 339dc75..76c6905 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -84,6 +84,10 @@ DEFAULTS: dict[str, Any] = { "docker.retention.metrics_raw_days": 7, "docker.retention.metrics_rollup_days": 90, "docker.retention.events_days": 30, + # Host/plugin metrics retention (plugin_metrics): keep a short raw window at + # the agent's ~30s cadence, then roll up to hourly averages kept much longer. + "metrics.retention.raw_days": 7, + "metrics.retention.rollup_days": 90, "plugins.index_url": "https://git.fabledsword.com/bvandeusen/Steward-plugins/raw/branch/main/index.yaml", # Default-enabled plugins. These are the generic, non-vendor-specific # bundled plugins (protocols/standards, not a single product) — useful on diff --git a/steward/migrations/versions/0024_plugin_metrics_hourly.py b/steward/migrations/versions/0024_plugin_metrics_hourly.py new file mode 100644 index 0000000..1ea9239 --- /dev/null +++ b/steward/migrations/versions/0024_plugin_metrics_hourly.py @@ -0,0 +1,43 @@ +"""Hourly rollup table for plugin_metrics + +Adds plugin_metrics_hourly — the coarse series that retention rolls raw +plugin_metrics into before pruning them, so multi-day/week host history stays +cheap to store. One row per (source_module, resource_name, metric_name, hour); +the unique constraint is the conflict target for the idempotent rollup upsert. + +Revision ID: 0024_plugin_metrics_hourly +Revises: 0023_plugin_metrics_indexes +Create Date: 2026-06-20 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0024_plugin_metrics_hourly" +down_revision: Union[str, None] = "0023_plugin_metrics_indexes" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "plugin_metrics_hourly", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("source_module", sa.String(length=64), nullable=False), + sa.Column("resource_name", sa.String(length=255), nullable=False), + sa.Column("metric_name", sa.String(length=128), nullable=False), + sa.Column("bucket", sa.DateTime(timezone=True), nullable=False), + sa.Column("value_avg", sa.Float(), nullable=False, server_default="0"), + sa.Column("value_max", sa.Float(), nullable=False, server_default="0"), + sa.Column("sample_count", sa.Integer(), nullable=False, server_default="0"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("source_module", "resource_name", "metric_name", "bucket", + name="uq_plugin_metrics_hourly_bucket"), + ) + op.create_index("ix_plugin_metrics_hourly_lookup", "plugin_metrics_hourly", + ["source_module", "resource_name", "metric_name", "bucket"]) + + +def downgrade() -> None: + op.drop_index("ix_plugin_metrics_hourly_lookup", table_name="plugin_metrics_hourly") + op.drop_table("plugin_metrics_hourly") diff --git a/steward/models/metrics.py b/steward/models/metrics.py index 979b68f..8a26996 100644 --- a/steward/models/metrics.py +++ b/steward/models/metrics.py @@ -1,7 +1,7 @@ from __future__ import annotations import uuid from datetime import datetime, timezone -from sqlalchemy import DateTime, Float, Index, String +from sqlalchemy import DateTime, Float, Index, Integer, String, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from .base import Base @@ -28,3 +28,30 @@ class PluginMetric(Base): Index("ix_plugin_metrics_module_resource_metric_recorded", "source_module", "resource_name", "metric_name", "recorded_at"), ) + + +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. + + One row per (source_module, resource_name, metric_name, hour bucket); the + unique constraint is the conflict target for the idempotent rollup upsert. + Charts read this for the part of a range older than the raw-retention window. + """ + __tablename__ = "plugin_metrics_hourly" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) + source_module: Mapped[str] = mapped_column(String(64), nullable=False) + resource_name: Mapped[str] = mapped_column(String(255), nullable=False) + metric_name: Mapped[str] = mapped_column(String(128), nullable=False) + bucket: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + value_avg: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) + value_max: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) + sample_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + __table_args__ = ( + UniqueConstraint("source_module", "resource_name", "metric_name", "bucket", + name="uq_plugin_metrics_hourly_bucket"), + Index("ix_plugin_metrics_hourly_lookup", + "source_module", "resource_name", "metric_name", "bucket"), + ) diff --git a/steward/settings/routes.py b/steward/settings/routes.py index 517d2e1..5b37672 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -133,6 +133,8 @@ _RETENTION_FIELDS = [ ("docker_metrics_raw_days", "docker.retention.metrics_raw_days"), ("docker_metrics_rollup_days", "docker.retention.metrics_rollup_days"), ("docker_events_days", "docker.retention.events_days"), + ("metrics_raw_days", "metrics.retention.raw_days"), + ("metrics_rollup_days", "metrics.retention.rollup_days"), ] diff --git a/steward/templates/settings/thresholds.html b/steward/templates/settings/thresholds.html index 204b049..db83c0c 100644 --- a/steward/templates/settings/thresholds.html +++ b/steward/templates/settings/thresholds.html @@ -70,6 +70,21 @@ "Keep container start/stop/die/health history this long.") }} +
+ Bounds how much host-agent metric history is stored (CPU, memory, disk, network, + temps, …). Raw per-sample points are kept for the raw window, then rolled up + into hourly averages kept for the rollup window. Host charts read raw for the + recent part of a range and hourly for older data. Applied by the hourly cleanup. +
+ + {{ days("Raw metrics", "metrics_raw_days", "metrics.retention.raw_days", + "Keep per-sample host metrics this long, then roll up to hourly averages.") }} + {{ days("Rolled-up metrics", "metrics_rollup_days", "metrics.retention.rollup_days", + "Keep the hourly-averaged series this long for multi-week history.") }} +