FabledSteward/tests/integration/test_alert_no_data.py

"""Integration (#5062): against real Postgres, a rule whose series stops is
marked no_data once, notified once, and cleared by the next sample; a firing
alert that goes silent stays firing."""
from __future__ import annotations

import asyncio
import os
import uuid
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

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 test_silence_is_marked_once_and_cleared_by_data(app):
    from sqlalchemy import select, update
    from steward.alerts import service as alert_svc
    from steward.core import alerts
    from steward.core.watch import watch
    from steward.models.alerts import AlertEvent, AlertRule, AlertState, AlertStateEnum
    from steward.models.metrics import PluginMetric
    from steward.models.users import User, UserRole

    sent: list[tuple[str, str, str | None]] = []

    async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
        sent.append((rule.name, notif_type, note))

    async def _go():
        tag = uuid.uuid4().hex[:8]
        res = f"quiet-{tag}"
        now = datetime.now(timezone.utc)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"nd-{tag}", email=f"nd-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                s.add(user)
                await alerts.record_metrics(s, [PluginMetric(
                    source_module="host_agent", resource_name=res, metric_name="cpu_pct",
                    value=1.0, recorded_at=now - timedelta(hours=1))])
                await s.flush()
                ids = {}
                for key, notify in (("loud", True), ("hushed", False)):
                    fields = await alert_svc.rule_fields(
                        s, name=f"{key} {tag}", source_module="host_agent", resource_name=res,
                        metric_name="cpu_pct", operator=">", threshold=90,
                        notify_no_data=notify)
                    ids[key] = (await alert_svc.add_rule(s, fields, created_by=user.id)).id
                # Every minute, from 20 to 18 minutes ago; then nothing.
                batch = alerts.MetricBatch()
                for ago in (20, 19, 18):
                    batch.add(source_module="host_agent", resource_name=res, metric_name="cpu_pct",
                              value=10.0, recorded_at=now - timedelta(minutes=ago))
                await batch.record(s)

        async def state(key):
            async with app.db_sessionmaker() as s:
                return (await s.execute(select(AlertState).where(
                    AlertState.rule_id == ids[key]))).scalar_one()

        async def events(key):
            async with app.db_sessionmaker() as s:
                rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == ids[key])
                                        .order_by(AlertEvent.transitioned_at))).scalars().all()
            return [(e.from_state, e.to_state, e.note) for e in rows]

        st = await state("loud")
        assert st.sample_interval_s == pytest.approx(60)
        assert st.last_sample_at == now - timedelta(minutes=18)

        # Steward has been watching all along (#5441 covers a fresh start).
        w = watch(app, "alert_silence")
        w.since, w.last = now - timedelta(days=1), datetime.now(timezone.utc)
        await alerts.check_silence(app)
        await alerts.check_silence(app)  # already marked: nothing more
        await asyncio.sleep(0)
        st = await state("loud")
        assert st.state == AlertStateEnum.inactive
        assert st.no_data_since == now - timedelta(minutes=18) + timedelta(seconds=300)
        [(frm, to, note)] = await events("loud")
        assert (frm, to) == ("inactive", "no_data")
        assert note == "no sample for 18m (usually every 1m)"
        assert len(await events("hushed")) == 1
        assert sent == [(f"loud {tag}", "no_data", note)]  # the hushed rule notifies nobody

        async with app.db_sessionmaker() as s:
            active = {r.id: alert_svc.alert_label(st_) for st_, r in await alert_svc.active_alerts(s)}
        assert active[ids["loud"]] == active[ids["hushed"]] == "no_data"

        # Data resumes, breaching: the mark clears and the rule fires.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
        await asyncio.sleep(0)
        st = await state("loud")
        assert st.no_data_since is None and st.state == AlertStateEnum.firing
        # The resume and the firing it leads to are one event, silent to
        # firing: two events at one instant used to sort either way (#5442).
        first, back = await events("loud")
        assert first[:2] == ("inactive", "no_data")
        assert back[:2] == ("no_data", "firing") and back[2].startswith("data resumed after ")
        assert [t for n, t, _ in sent if n == f"loud {tag}"] == ["no_data", "firing"]
        # The hushed rule says nothing of silence, but its firing is still news.
        assert [t for n, t, _ in sent if n == f"hushed {tag}"] == ["firing"]

        # A firing alert that goes silent stays firing, marked.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(update(AlertState).where(AlertState.rule_id == ids["loud"])
                                .values(last_sample_at=now - timedelta(hours=2)))
                await s.execute(update(AlertRule).where(AlertRule.id == ids["hushed"])
                                .values(enabled=False))
        await alerts.check_silence(app)
        st = await state("loud")
        assert st.state == AlertStateEnum.firing and st.no_data_since is not None
        assert (await events("loud"))[-1][:2] == ("firing", "no_data")
        assert len(await events("hushed")) == 2  # disabled: not watched

    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
        asyncio.run(_go())


def test_a_series_quiet_only_while_steward_was_down_is_not_silent(app):
    """#5441: Steward comes back to a series last sampled 18 minutes ago. Its
    outage is not the series' silence, so nothing is marked; once Steward has
    watched for three of the series' gaps with nothing arriving, it is."""
    from sqlalchemy import select
    from steward.alerts import service as alert_svc
    from steward.core import alerts
    from steward.core.watch import watch
    from steward.models.alerts import AlertEvent, AlertState
    from steward.models.metrics import PluginMetric
    from steward.models.users import User, UserRole

    sent = []

    async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
        sent.append((rule.id, notif_type))

    async def _go():
        tag = uuid.uuid4().hex[:8]
        res = f"back-{tag}"
        now = datetime.now(timezone.utc)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"bk-{tag}", email=f"bk-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                s.add(user)
                await alerts.record_metrics(s, [PluginMetric(
                    source_module="host_agent", resource_name=res, metric_name="cpu_pct",
                    value=1.0, recorded_at=now - timedelta(minutes=18))])
                await s.flush()
                fields = await alert_svc.rule_fields(
                    s, name=f"back {tag}", source_module="host_agent", resource_name=res,
                    metric_name="cpu_pct", operator=">", threshold=90, notify_no_data=True)
                rule_id = (await alert_svc.add_rule(s, fields, created_by=user.id)).id
                await s.flush()
                st = (await s.execute(select(AlertState).where(AlertState.rule_id == rule_id))).scalar_one()
                st.last_sample_at, st.sample_interval_s = now - timedelta(minutes=18), 60.0

        async def marks():
            async with app.db_sessionmaker() as s:
                st = (await s.execute(select(AlertState).where(AlertState.rule_id == rule_id))).scalar_one()
                n = len((await s.execute(select(AlertEvent).where(AlertEvent.rule_id == rule_id))).all())
            return st.no_data_since, n

        await alerts.check_silence(app)        # the first run since Steward started
        await asyncio.sleep(0)
        assert await marks() == (None, 0) and (rule_id, "no_data") not in sent

        # Six quiet minutes later (as if), Steward has watched long enough.
        w = watch(app, "alert_silence")
        w.since = now - timedelta(minutes=6)
        await alerts.check_silence(app)
        await asyncio.sleep(0)
        assert await marks() == (w.since + timedelta(seconds=300), 1)
        assert sent.count((rule_id, "no_data")) == 1

    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
        asyncio.run(_go())