"""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_load_is_read_against_the_hosts_cores(app): """#5251: cores are counted from the per-core CPU sub-resources, and load per core is load_1m as a % of them, the figure "load" thresholds grade.""" from steward.core.alerts import record_metrics from steward.hosts.service import agent_overview_by_host now = datetime.now(timezone.utc) host = "cores-" + uuid.uuid4().hex[:8] async def go(): async with app.db_sessionmaker() as s: async with s.begin(): await record_metrics(s, [ _sample(host, "load_1m", 3.0, now), _sample(host + ":core0", "cpu_pct", 80.0, now), _sample(host + ":core1", "cpu_pct", 90.0, now), ]) return await agent_overview_by_host(s, [host], 180) h = asyncio.run(go())[host] assert (h["cores"], h["load_per_core"]) == (2, 150) 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])}