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, PluginMetricLatest

# 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})]


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[PluginMetricLatest]:
    """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). Read from
    plugin_metric_latest, so the cost is the resource's metric count, not its
    history.
    """
    return list((await db.execute(
        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.

    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).
    The raw window is metrics.retention.raw_days, the setting the rollup in
    core/metrics_retention.py reads, so the seam is where the rollup put it.
    """
    from steward.core.settings import get_setting

    raw_days = int(await get_setting(db, "metrics.retention.raw_days") or 7)
    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