FabledSteward/tests/integration/test_metric_latest.py

"""Integration: plugin_metric_latest, the newest sample per metric (#5208).

record_metrics keeps it beside plugin_metrics so "what is it now" reads one row
per metric instead of every raw sample in the window. Requires
STEWARD_DATABASE_URL.
"""
from __future__ import annotations

import asyncio
import os
import uuid
from datetime import datetime, timedelta, timezone

import pytest

pytestmark = pytest.mark.integration


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


def _sample(resource, metric, value, at):
    from steward.models.metrics import PluginMetric
    return PluginMetric(source_module="host_agent", resource_name=resource,
                        metric_name=metric, value=value, recorded_at=at)


async def _latest(s, resource):
    from sqlalchemy import select
    from steward.models.metrics import PluginMetricLatest
    rows = (await s.execute(select(PluginMetricLatest).where(
        PluginMetricLatest.source_module == "host_agent",
        PluginMetricLatest.resource_name == resource))).scalars().all()
    return {r.metric_name: (r.value, r.recorded_at) for r in rows}


def test_the_newest_sample_wins_and_a_late_backlog_does_not_move_it_back(app):
    from steward.core.alerts import record_metrics

    now = datetime.now(timezone.utc).replace(microsecond=0)
    host = "latest-" + uuid.uuid4().hex[:8]

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                # One batch holding two samples of a metric: the newer is kept.
                await record_metrics(s, [
                    _sample(host, "cpu_pct", 10.0, now - timedelta(minutes=2)),
                    _sample(host, "cpu_pct", 20.0, now - timedelta(minutes=1)),
                    _sample(host, "mem_used_pct", 40.0, now - timedelta(minutes=1)),
                ])
            async with s.begin():
                after_first = await _latest(s, host)
            async with s.begin():
                # An agent catching up posts older samples after newer ones.
                await record_metrics(s, [
                    _sample(host, "cpu_pct", 99.0, now - timedelta(minutes=5)),
                    _sample(host, "mem_used_pct", 41.0, now),
                ])
            return after_first, await _latest(s, host)

    first, second = asyncio.run(go())
    assert first["cpu_pct"] == (20.0, now - timedelta(minutes=1))
    assert second["cpu_pct"] == (20.0, now - timedelta(minutes=1))    # backlog ignored
    assert second["mem_used_pct"] == (41.0, now)                      # newer moves it


def test_a_quiet_host_keeps_its_last_values_and_reads_stale_not_missing(app):
    from steward.core.alerts import record_metrics
    from steward.hosts.service import agent_overview_by_host

    now = datetime.now(timezone.utc)
    quiet, live = ("quiet-" + uuid.uuid4().hex[:8]), ("live-" + uuid.uuid4().hex[:8])
    two_hours_ago = now - timedelta(hours=2)

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await record_metrics(s, [
                    _sample(quiet, "cpu_pct", 12.0, two_hours_ago),
                    _sample(quiet, "mem_used_pct", 34.0, two_hours_ago),
                    _sample(quiet + ":/", "disk_used_pct", 56.0, two_hours_ago),
                    _sample(live, "cpu_pct", 7.0, now - timedelta(seconds=20)),
                ])
            return await agent_overview_by_host(s, [quiet, live, "never-" + quiet], 180)

    overview = asyncio.run(go())
    q = overview[quiet]
    assert (q["cpu_pct"], q["mem_used_pct"], q["disk_root"]) == (12.0, 34.0, 56.0)
    assert q["fresh"] is False
    assert abs((q["last_at"] - two_hours_ago).total_seconds()) < 1
    assert overview[live]["fresh"] is True
    assert "never-" + quiet not in overview          # no samples at all: absent


def test_retention_drops_a_latest_value_with_its_raw_history(app):
    from steward.core.alerts import record_metrics
    from steward.core.metrics_retention import rollup_plugin_metrics

    now = datetime.now(timezone.utc)
    gone, here = ("gone-" + uuid.uuid4().hex[:8]), ("here-" + uuid.uuid4().hex[:8])

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await record_metrics(s, [
                    _sample(gone, "cpu_pct", 1.0, now - timedelta(days=10)),
                    _sample(here, "cpu_pct", 2.0, now - timedelta(hours=1)),
                ])
            async with s.begin():
                counts = await rollup_plugin_metrics(s, raw_days=7, rollup_days=90, now=now)
            return counts, await _latest(s, gone), await _latest(s, here)

    counts, gone_latest, here_latest = asyncio.run(go())
    assert counts["latest_pruned"] >= 1
    assert gone_latest == {}
    assert here_latest["cpu_pct"][0] == 2.0


def test_a_renamed_host_carries_its_latest_values(app):
    from steward.core.alerts import record_metrics
    from steward.core.renames import HOST_SOURCES, carry_rename

    now = datetime.now(timezone.utc)
    old, new = ("old-" + uuid.uuid4().hex[:8]), ("new-" + uuid.uuid4().hex[:8])

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await record_metrics(s, [
                    _sample(old, "cpu_pct", 5.0, now),
                    _sample(old + ":/", "disk_used_pct", 60.0, now),
                ])
            async with s.begin():
                await carry_rename(s, HOST_SOURCES, old, new)
            return (await _latest(s, old), await _latest(s, new),
                    await _latest(s, new + ":/"))

    old_latest, new_latest, new_root = asyncio.run(go())
    assert old_latest == {}
    assert new_latest["cpu_pct"][0] == 5.0
    assert new_root["disk_used_pct"][0] == 60.0


def test_the_backfill_fills_a_quiet_resource_once_and_leaves_live_ones_alone(app):
    import logging

    from steward.core.alerts import record_metrics
    from steward.core.metric_series import backfill_latest

    now = datetime.now(timezone.utc)
    quiet, live = ("bf-quiet-" + uuid.uuid4().hex[:8]), ("bf-live-" + uuid.uuid4().hex[:8])
    went_quiet = now - timedelta(hours=3)

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                # History from before the table existed: rows with no latest.
                s.add_all([
                    _sample(quiet, "cpu_pct", 10.0, went_quiet - timedelta(minutes=30)),
                    _sample(quiet, "cpu_pct", 11.0, went_quiet),
                    _sample(quiet, "mem_used_pct", 50.0, went_quiet - timedelta(minutes=10)),
                    _sample(quiet, "load_1m", 9.9, went_quiet - timedelta(hours=2)),  # outside the hour
                ])
            async with s.begin():
                await record_metrics(s, [_sample(live, "cpu_pct", 70.0, now)])
                s.add(_sample(live, "cpu_pct", 1.0, now - timedelta(minutes=5)))
        first = await backfill_latest(app.db_sessionmaker, logging.getLogger("test"))
        second = await backfill_latest(app.db_sessionmaker, logging.getLogger("test"))
        async with app.db_sessionmaker() as s:
            return first, second, await _latest(s, quiet), await _latest(s, live)

    first, second, quiet_latest, live_latest = asyncio.run(go())
    assert first >= 2 and second == 0
    assert quiet_latest["cpu_pct"][0] == 11.0
    assert quiet_latest["mem_used_pct"][0] == 50.0
    assert "load_1m" not in quiet_latest
    assert live_latest == {"cpu_pct": (70.0, live_latest["cpu_pct"][1])}