464 lines
13 KiB
Python
464 lines
13 KiB
Python
"""Alerting: failing loudly once, not thirty-six thousand times."""
|
|
|
|
import asyncio
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow.alerts import (
|
|
ALERTING_EVENTS,
|
|
FLAP_THRESHOLD,
|
|
LEDGER_RETAIN_S,
|
|
RATE_LIMIT,
|
|
Alert,
|
|
AlertManager,
|
|
AlertsConfig,
|
|
Channel,
|
|
Rule,
|
|
describe,
|
|
)
|
|
from fluksio.flow.events import EventBus
|
|
|
|
|
|
class Ledger:
|
|
"""What the database holds, without one."""
|
|
|
|
def __init__(self) -> None:
|
|
self.rows: list[tuple[str, float]] = []
|
|
|
|
def load(self, since: float) -> list[tuple[str, float]]:
|
|
return [row for row in self.rows if row[1] >= since]
|
|
|
|
def add(self, key: str, ts: float, keep_since: float) -> None:
|
|
self.rows = [row for row in self.rows if row[1] >= keep_since]
|
|
self.rows.append((key, ts))
|
|
|
|
|
|
class Clock:
|
|
"""A hand-wound clock, so cooldowns take no real time."""
|
|
|
|
def __init__(self) -> None:
|
|
self.now = 0.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self.now += seconds
|
|
|
|
|
|
def manager(clock: Clock) -> tuple[AlertManager, list[Alert]]:
|
|
sent: list[Alert] = []
|
|
config = AlertsConfig(
|
|
channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
|
|
rules=[Rule(events=[], channels=["phone"], cooldown_s=900)],
|
|
)
|
|
# No grace: these tests are about what is sent, not how long a connection
|
|
# has to stay down first.
|
|
alerts = AlertManager(EventBus(), config=config, now=clock, grace_s=0)
|
|
|
|
async def capture(channel, alert):
|
|
sent.append(alert)
|
|
return True
|
|
|
|
alerts.send = capture # type: ignore[method-assign]
|
|
return alerts, sent
|
|
|
|
|
|
def error_event(node: str = "heating.pump") -> dict:
|
|
return {"type": "node_error", "flow": "heating", "node": node, "error": "boom"}
|
|
|
|
|
|
def test_a_failure_reaches_the_channel():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
asyncio.run(alerts.handle(error_event()))
|
|
|
|
assert [a.title for a in sent] == ["heating.pump failed"]
|
|
assert sent[0].body == "boom"
|
|
|
|
|
|
def test_ordinary_traffic_is_not_an_alert():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
asyncio.run(alerts.handle({"type": "message_value", "name": "heating.temp"}))
|
|
asyncio.run(alerts.handle({"type": "node_executed", "node": "heating.pump"}))
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_the_same_failure_repeating_is_one_alert():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
async def scenario():
|
|
for _ in range(50):
|
|
await alerts.handle(error_event())
|
|
clock.advance(1)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_the_cooldown_ends_and_says_what_was_missed():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
async def scenario():
|
|
await alerts.handle(error_event())
|
|
for _ in range(4):
|
|
clock.advance(10)
|
|
await alerts.handle(error_event())
|
|
clock.advance(1000)
|
|
await alerts.handle(error_event())
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == 2
|
|
assert "4 more since the last alert" in sent[1].body
|
|
|
|
|
|
def test_different_nodes_alert_separately():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
async def scenario():
|
|
await alerts.handle(error_event("heating.pump"))
|
|
await alerts.handle(error_event("heating.valve"))
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == 2
|
|
|
|
|
|
def test_a_flapping_connection_goes_quiet():
|
|
"""A device dropping every few seconds is one story, not one alert each."""
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
async def scenario():
|
|
for _ in range(FLAP_THRESHOLD * 2 + 6):
|
|
await alerts.handle(
|
|
{"type": "node_health", "node": "heating.pump", "health": "down"}
|
|
)
|
|
clock.advance(5)
|
|
await alerts.handle(
|
|
{"type": "node_health", "node": "heating.pump", "health": "ok"}
|
|
)
|
|
clock.advance(5)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
# The first drop is worth knowing about; the rest is noise.
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_a_storm_is_capped():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
async def scenario():
|
|
for i in range(RATE_LIMIT + 20):
|
|
await alerts.handle(error_event(f"heating.n{i}"))
|
|
clock.advance(1)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == RATE_LIMIT
|
|
|
|
|
|
def test_a_cooldown_survives_a_restart():
|
|
"""A node that stays down was announced again by every restart."""
|
|
clock, ledger = Clock(), Ledger()
|
|
|
|
async def scenario() -> list[Alert]:
|
|
first, sent_before = manager(clock)
|
|
first._ledger = ledger
|
|
await first.handle(error_event())
|
|
clock.advance(120)
|
|
|
|
second, sent_after = manager(clock)
|
|
second._ledger = ledger
|
|
await second.restore()
|
|
await second.handle(error_event())
|
|
assert len(sent_before) == 1
|
|
clock.advance(1000)
|
|
await second.handle(error_event())
|
|
return sent_after
|
|
|
|
sent_after = asyncio.run(scenario())
|
|
|
|
# Silent inside the cooldown, and speaking again once it has passed.
|
|
assert len(sent_after) == 1
|
|
|
|
|
|
def test_the_ceiling_per_hour_survives_a_restart():
|
|
clock, ledger = Clock(), Ledger()
|
|
clock.now = 10_000.0
|
|
ledger.rows = [
|
|
(f"node_error:heating.n{i}", clock.now - 60) for i in range(RATE_LIMIT)
|
|
]
|
|
alerts, sent = manager(clock)
|
|
alerts._ledger = ledger
|
|
|
|
async def scenario():
|
|
await alerts.restore()
|
|
await alerts.handle(error_event("heating.new"))
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_a_clock_set_back_does_not_hold_a_key_silent():
|
|
clock, ledger = Clock(), Ledger()
|
|
clock.now = 100.0
|
|
ledger.rows = [("node_error:heating.pump", 50_000.0)]
|
|
alerts, sent = manager(clock)
|
|
alerts._ledger = ledger
|
|
|
|
async def scenario():
|
|
await alerts.restore()
|
|
clock.advance(1000)
|
|
await alerts.handle(error_event())
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_sent_alerts_are_written_to_the_ledger_and_old_ones_dropped():
|
|
clock, ledger = Clock(), Ledger()
|
|
clock.now = LEDGER_RETAIN_S * 2
|
|
ledger.rows = [("node_error:heating.old", 1.0)]
|
|
alerts, _sent = manager(clock)
|
|
alerts._ledger = ledger
|
|
|
|
asyncio.run(alerts.handle(error_event()))
|
|
|
|
assert ledger.rows == [("node_error:heating.pump", clock.now)]
|
|
|
|
|
|
def test_a_ledger_that_fails_does_not_stop_the_alert():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
class Broken:
|
|
def load(self, since):
|
|
raise OSError("disk")
|
|
|
|
def add(self, key, ts, keep_since):
|
|
raise OSError("disk")
|
|
|
|
alerts._ledger = Broken() # type: ignore[assignment]
|
|
|
|
async def scenario():
|
|
await alerts.restore()
|
|
await alerts.handle(error_event())
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_an_alert_no_channel_took_does_not_spend_its_cooldown():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
delivered = iter([False, True])
|
|
|
|
async def flaky(channel, alert):
|
|
sent.append(alert)
|
|
return next(delivered)
|
|
|
|
alerts.send = flaky # type: ignore[method-assign]
|
|
|
|
async def scenario():
|
|
await alerts.handle(error_event())
|
|
clock.advance(10)
|
|
await alerts.handle(error_event())
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(sent) == 2
|
|
|
|
|
|
def test_an_alert_held_back_by_the_ceiling_does_not_spend_its_cooldown():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
|
|
async def scenario():
|
|
for i in range(RATE_LIMIT):
|
|
await alerts.handle(error_event(f"heating.n{i}"))
|
|
await alerts.handle(error_event("heating.late"))
|
|
# An hour on, the ceiling has cleared and the same fault is still due.
|
|
clock.advance(3601)
|
|
await alerts.handle(error_event("heating.late"))
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert [a.node for a in sent][-1] == "heating.late"
|
|
assert "1 more since the last alert" in sent[-1].body
|
|
|
|
|
|
def down(node: str = "heating.mqtt") -> dict:
|
|
return {"type": "node_health", "node": node, "health": "down", "detail": "refused"}
|
|
|
|
|
|
def up(node: str = "heating.mqtt") -> dict:
|
|
return {"type": "node_health", "node": node, "health": "ok"}
|
|
|
|
|
|
def with_grace(grace_s: float) -> tuple[AlertManager, list[Alert]]:
|
|
alerts, sent = manager(Clock())
|
|
alerts.grace_s = grace_s
|
|
return alerts, sent
|
|
|
|
|
|
def test_a_connection_that_comes_back_within_the_grace_alerts_nobody():
|
|
alerts, sent = with_grace(0.05)
|
|
|
|
async def scenario():
|
|
await alerts.handle(down())
|
|
await alerts.handle(up())
|
|
await asyncio.sleep(0.1)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_a_connection_that_stays_down_is_told_after_the_grace():
|
|
alerts, sent = with_grace(0.05)
|
|
|
|
async def scenario():
|
|
await alerts.handle(down())
|
|
assert sent == []
|
|
await asyncio.sleep(0.1)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert [a.title for a in sent] == ["heating.mqtt lost its connection"]
|
|
|
|
|
|
def test_a_pending_alert_is_dropped_when_the_manager_stops():
|
|
alerts, sent = with_grace(0.05)
|
|
|
|
async def scenario():
|
|
await alerts.handle(down())
|
|
runner = asyncio.create_task(alerts.run())
|
|
await asyncio.sleep(0.01)
|
|
runner.cancel()
|
|
await asyncio.gather(runner, return_exceptions=True)
|
|
await asyncio.sleep(0.1)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_nothing_is_sent_without_a_rule():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
alerts.config = AlertsConfig(
|
|
channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
|
|
rules=[Rule(events=["queue_unavailable"], channels=["phone"])],
|
|
)
|
|
|
|
asyncio.run(alerts.handle(error_event()))
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_alerting_can_be_switched_off():
|
|
clock = Clock()
|
|
alerts, sent = manager(clock)
|
|
alerts.config = alerts.config.model_copy(update={"enabled": False})
|
|
|
|
asyncio.run(alerts.handle(error_event()))
|
|
|
|
assert sent == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("event", "expected"),
|
|
[
|
|
(
|
|
{"type": "flow_quarantined", "flow": "heating"},
|
|
"Flow 'heating' was quarantined",
|
|
),
|
|
(
|
|
{"type": "queue_unavailable", "error": "gone"},
|
|
"The work queue is unreachable",
|
|
),
|
|
({"type": "engine_degraded", "reason": "lag"}, "The engine is struggling"),
|
|
# Every run ends. Only a failed one is worth a phone buzzing.
|
|
(
|
|
{"type": "run_finished", "flow": "nightly", "run": 7, "status": "error"},
|
|
"A run of 'nightly' failed",
|
|
),
|
|
({"type": "run_finished", "flow": "nightly", "status": "ok"}, None),
|
|
({"type": "run_finished", "flow": "nightly", "status": "cancelled"}, None),
|
|
({"type": "node_health", "health": "ok"}, None),
|
|
# The engine publishes `health`, not `status`: reading the wrong key
|
|
# meant a device dropping never alerted anyone.
|
|
(
|
|
{"type": "node_health", "node": "heating.pump", "health": "down"},
|
|
"heating.pump lost its connection",
|
|
),
|
|
],
|
|
)
|
|
def test_every_alerting_event_reads_as_a_sentence(event, expected):
|
|
alert = describe(event)
|
|
assert (alert.title if alert else None) == expected
|
|
|
|
|
|
def test_a_dashboard_channel_publishes_the_alert_as_a_record():
|
|
published: list[tuple[str, dict]] = []
|
|
config = AlertsConfig(
|
|
channels=[
|
|
Channel(name="panel", kind="dashboard", config={"message": "house.notice"})
|
|
],
|
|
rules=[Rule(events=[], channels=["panel"])],
|
|
)
|
|
alerts = AlertManager(EventBus(), config=config, now=Clock())
|
|
alerts.publish = lambda name, value: published.append((name, value))
|
|
|
|
asyncio.run(alerts.handle(error_event()))
|
|
|
|
assert len(published) == 1
|
|
name, record = published[0]
|
|
assert name == "house.notice"
|
|
# Flat named scalars, which is what a notification widget binds.
|
|
assert record["title"] == "heating.pump failed"
|
|
assert record["severity"] == "error"
|
|
assert all(isinstance(v, str) for v in record.values())
|
|
|
|
|
|
def test_a_dashboard_channel_without_a_message_says_so():
|
|
alerts = AlertManager(EventBus(), now=Clock())
|
|
alerts.publish = lambda name, value: None
|
|
channel = Channel(name="panel", kind="dashboard")
|
|
|
|
with pytest.raises(ValueError):
|
|
asyncio.run(
|
|
alerts.send(channel, Alert(title="t", body="b"), raise_on_error=True)
|
|
)
|
|
|
|
|
|
ALERTS_ROUTE = Path(__file__).parents[3] / "frontend/src/routes/_layout/alerts.tsx"
|
|
|
|
|
|
@pytest.mark.skipif(not ALERTS_ROUTE.exists(), reason="no frontend in this checkout")
|
|
def test_the_alerts_screen_offers_every_alerting_event():
|
|
"""The chooser lists the events by hand, so it can drift out of this set.
|
|
|
|
Only the ids have to agree — the labels beside them are UI copy.
|
|
"""
|
|
block = re.search(r"const EVENTS[^=]*= \[(.*?)\n\]", ALERTS_ROUTE.read_text(), re.S)
|
|
assert block, "the EVENTS list moved — this test needs following"
|
|
assert set(re.findall(r'\["(\w+)"', block.group(1))) == ALERTING_EVENTS
|