bvandeusenandClaude Opus 5.5
d5601575d8
#5473 "Opening an alert rule takes up to 15 s". With the rule pickers now reading plugin_metric_latest, nine integration tests failed on rule_fields: "host_agent has no resource 'box-…'". They seeded a resource by adding a raw PluginMetric row, which no production path does: host-agent ingest and record_metric both go through record_metrics, which keeps the latest-value table. The seeds now go through record_metrics too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
186 lines
7.4 KiB
Python
186 lines
7.4 KiB
Python
"""Integration (#5063): against real Postgres, notifications and actions start
|
|
only after the evaluating transaction commits; a maintenance window mutes them
|
|
without stopping the state from moving, and an alert still firing when the
|
|
window ends notifies then; an alert action is held to what its source lists."""
|
|
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)
|
|
|
|
|
|
async def _rule(app, tag, **extra):
|
|
from steward.alerts import service as alert_svc
|
|
from steward.core.alerts import record_metrics
|
|
from steward.models.metrics import PluginMetric
|
|
from steward.models.users import User, UserRole
|
|
res = f"box-{tag}"
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
user = User(username=f"dp-{tag}", email=f"dp-{tag}@test", password_hash="x",
|
|
role=UserRole.admin)
|
|
s.add(user)
|
|
await record_metrics(s, [PluginMetric(source_module="host_agent", resource_name=res,
|
|
metric_name="cpu_pct", value=1.0)])
|
|
await s.flush()
|
|
fields = await alert_svc.rule_fields(
|
|
s, name=f"hot {tag}", source_module="host_agent", resource_name=res,
|
|
metric_name="cpu_pct", operator=">", threshold=90)
|
|
rule = await alert_svc.add_rule(s, fields, created_by=user.id, **extra)
|
|
return rule.id, res, user.id
|
|
|
|
|
|
async def _events(app, rule_id):
|
|
from sqlalchemy import select
|
|
from steward.models.alerts import AlertEvent
|
|
async with app.db_sessionmaker() as s:
|
|
rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == rule_id)
|
|
.order_by(AlertEvent.transitioned_at))).scalars().all()
|
|
return [(e.from_state, e.to_state, e.note, e.ansible_run_id) for e in rows]
|
|
|
|
|
|
async def _settle():
|
|
for _ in range(5):
|
|
await asyncio.sleep(0)
|
|
|
|
|
|
def test_a_rolled_back_evaluation_notifies_nobody(app):
|
|
from steward.core import alerts
|
|
|
|
sent = []
|
|
|
|
async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
|
|
sent.append((notif_type, value))
|
|
|
|
async def _go():
|
|
rule_id, res, _ = await _rule(app, uuid.uuid4().hex[:8])
|
|
|
|
class Boom(Exception):
|
|
pass
|
|
|
|
with pytest.raises(Boom):
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
|
|
raise Boom
|
|
await _settle()
|
|
assert sent == [] and await _events(app, rule_id) == []
|
|
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await alerts.record_metric(s, "host_agent", res, "cpu_pct", 96.0)
|
|
await _settle()
|
|
assert sent == [] # not before the commit
|
|
await _settle()
|
|
assert sent == [("firing", 96.0)]
|
|
|
|
with patch.object(alerts, "_dispatch_notification", fake_dispatch):
|
|
asyncio.run(_go())
|
|
|
|
|
|
def test_maintenance_mutes_but_the_state_moves_and_notifies_when_it_ends(app):
|
|
from sqlalchemy import select, update
|
|
from steward.core import alerts
|
|
from steward.models.alerts import AlertState, AlertStateEnum, MaintenanceWindow
|
|
|
|
sent = []
|
|
|
|
async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
|
|
sent.append((notif_type, value))
|
|
|
|
async def _go():
|
|
tag = uuid.uuid4().hex[:8]
|
|
rule_id, res, user_id = await _rule(app, tag)
|
|
now = datetime.now(timezone.utc)
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
window = MaintenanceWindow(
|
|
name=f"patching {tag}", start_at=now - timedelta(minutes=1),
|
|
end_at=now + timedelta(hours=1), scope_source="host_agent",
|
|
scope_resource=res, created_by=user_id)
|
|
s.add(window)
|
|
window_id = window.id
|
|
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
|
|
await _settle()
|
|
async with app.db_sessionmaker() as s:
|
|
st = (await s.execute(select(AlertState).where(
|
|
AlertState.rule_id == rule_id))).scalar_one()
|
|
assert st.state == AlertStateEnum.firing # evaluated inside the window
|
|
[(frm, to, note, _)] = await _events(app, rule_id)
|
|
assert (frm, to, note) == ("inactive", "firing",
|
|
f"notification muted: maintenance window patching {tag}")
|
|
assert sent == []
|
|
|
|
await alerts.notify_after_maintenance(app) # window still open: held
|
|
await _settle()
|
|
assert sent == []
|
|
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await s.execute(update(MaintenanceWindow).where(MaintenanceWindow.id == window_id)
|
|
.values(end_at=datetime.now(timezone.utc)))
|
|
await alerts.notify_after_maintenance(app)
|
|
await alerts.notify_after_maintenance(app) # released once
|
|
await _settle()
|
|
assert sent == [("firing", 95.0)]
|
|
[(_, _, note, _)] = await _events(app, rule_id)
|
|
assert note.endswith("; notified when the window ended")
|
|
|
|
with patch.object(alerts, "_dispatch_notification", fake_dispatch):
|
|
asyncio.run(_go())
|
|
|
|
|
|
def test_an_alert_action_outside_its_source_is_refused(app, tmp_path):
|
|
from steward.core import alerts
|
|
from steward.models.ansible import AnsibleRun
|
|
from sqlalchemy import func, select
|
|
|
|
(tmp_path / "fix.yml").write_text("- hosts: all\n tasks: []\n")
|
|
(tmp_path / "inventory.ini").write_text("localhost\n")
|
|
app.config["ANSIBLE"] = {"sources": [{"name": "t", "type": "local", "path": str(tmp_path)}]}
|
|
|
|
async def quiet(*args, **kwargs):
|
|
pass
|
|
|
|
async def _go():
|
|
tag = uuid.uuid4().hex[:8]
|
|
rule_id, res, _ = await _rule(
|
|
app, tag, ansible_action={"source": "t", "playbook": "../../etc/evil.yml",
|
|
"inventory": "inventory.ini"})
|
|
async with app.db_sessionmaker() as s:
|
|
before = (await s.execute(select(func.count()).select_from(AnsibleRun))).scalar_one()
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
|
|
loop = asyncio.get_running_loop()
|
|
deadline = loop.time() + 30
|
|
while alerts._background:
|
|
assert loop.time() < deadline, "alert tasks did not finish"
|
|
await asyncio.wait(set(alerts._background), timeout=deadline - loop.time())
|
|
[(frm, to, note, run_id)] = await _events(app, rule_id)
|
|
assert (frm, to, run_id) == ("inactive", "firing", None)
|
|
assert note == "Ansible action not started: Source 't' has no playbook '../../etc/evil.yml'"
|
|
async with app.db_sessionmaker() as s:
|
|
after = (await s.execute(select(func.count()).select_from(AnsibleRun))).scalar_one()
|
|
assert after == before
|
|
|
|
with patch.object(alerts, "_dispatch_notification", quiet):
|
|
asyncio.run(_go())
|