FabledSteward/steward/core/metric_series.py

"""Reads over the generic metric store (plugin_metrics + its hourly rollup).

Any source that records into PluginMetric — host_agent, traefik, unifi, snmp,
the monitor types, Steward's self-monitor — is readable through these without
importing the plugin that wrote it.
"""
from __future__ import annotations

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

# Raw rows older than this are rolled into plugin_metrics_hourly and deleted
# (see core/metrics_retention.py), so a range reaching past it reads the rollup.
RAW_DAYS = 7

# Loose index scans: step through the distinct leading values of the
# (source_module, resource_name, …) index one probe at a time instead of
# scanning every sample. Postgres has no native skip scan before 18.
_SOURCES_SQL = text("""
    WITH RECURSIVE s(v) AS (
        SELECT min(source_module) FROM plugin_metrics
        UNION ALL
        SELECT (SELECT min(source_module) FROM plugin_metrics WHERE source_module > s.v)
        FROM s WHERE s.v IS NOT NULL
    )
    SELECT v FROM s WHERE v IS NOT NULL
""")

_RESOURCES_SQL = text("""
    WITH RECURSIVE r(v) AS (
        SELECT min(resource_name) FROM plugin_metrics WHERE source_module = :src
        UNION ALL
        SELECT (SELECT min(resource_name) FROM plugin_metrics
                WHERE source_module = :src AND resource_name > r.v)
        FROM r WHERE r.v IS NOT NULL
    )
    SELECT v FROM r WHERE v IS NOT NULL
""")


async def sources(db) -> list[str]:
    """Every source_module with raw samples on record."""
    return [v for (v,) in await db.execute(_SOURCES_SQL)]


async def resources(db, source_module: str) -> list[str]:
    """Every resource a source has raw samples for, sorted."""
    return [v for (v,) in await db.execute(_RESOURCES_SQL, {"src": source_module})]


async def latest_values(db, source_module: str, resource: str, *,
                        include_children: bool = False) -> list[PluginMetric]:
    """The newest sample of each metric for a resource, sorted by (resource, metric).

    include_children also returns the resource's sub-resources, named
    "<resource>:<part>" (a host's mounts, interfaces, sensors).
    """
    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())
    )).scalars())


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.

    The part of the range older than the raw window comes from the hourly
    rollup; the rest is bucket-averaged in SQL (buckets no wider than an hour
    when both are present, so the series is evenly grained across the seam).
    """
    now = datetime.now(timezone.utc)
    raw_cutoff = (now - timedelta(days=RAW_DAYS)).replace(minute=0, second=0, microsecond=0)
    out: list[tuple[datetime, float]] = []

    if since < raw_cutoff:
        hourly = await db.execute(
            select(PluginMetricHourly.bucket, PluginMetricHourly.value_avg)
            .where(
                PluginMetricHourly.source_module == source_module,
                PluginMetricHourly.resource_name == resource,
                PluginMetricHourly.metric_name == metric,
                PluginMetricHourly.bucket >= since,
                PluginMetricHourly.bucket < raw_cutoff,
            )
            .order_by(PluginMetricHourly.bucket)
        )
        out.extend((b, float(v)) for b, v in hourly)

    raw_since = max(since, raw_cutoff)
    width_s = bucket_seconds(raw_since, target)
    if since < raw_cutoff:
        width_s = min(width_s, 3600)
    bucket = func.date_bin(
        func.make_interval(0, 0, 0, 0, 0, 0, width_s),
        PluginMetric.recorded_at,
        func.to_timestamp(0),
    ).label("bucket")
    raw = await db.execute(
        select(bucket, func.avg(PluginMetric.value))
        .where(
            PluginMetric.source_module == source_module,
            PluginMetric.resource_name == resource,
            PluginMetric.metric_name == metric,
            PluginMetric.recorded_at >= raw_since,
        )
        .group_by(bucket)
        .order_by(bucket)
    )
    out.extend((b, float(v)) for b, v in raw)
    return out