Tell someone when the engine breaks
Everything that goes wrong already travelled the event bus, but the only subscriber was the editor's websocket — so a flow quarantined at three in the morning was invisible until someone opened the browser. An alert manager now watches the same bus and forwards failures to ntfy, email or a webhook. Most of what it does is decline to send: the same node failing every second is one alert with a count of what followed, a connection flapping up and down is muted until it settles, and nothing gets past ten notifications an hour. Verified against a live instance — six identical failures produced one alert carrying the real traceback message. Channels and rules are configured through the API, with a test send so a channel can be proven before anything depends on it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY
This commit is contained in:
root
+11
-1
@@ -1,6 +1,15 @@
|
||||
from fastapi import APIRouter
|
||||
|
||||
from app.api.routes import flows, login, oauth, private, secrets, users, utils
|
||||
from app.api.routes import (
|
||||
alerts,
|
||||
flows,
|
||||
login,
|
||||
oauth,
|
||||
private,
|
||||
secrets,
|
||||
users,
|
||||
utils,
|
||||
)
|
||||
|
||||
api_router = APIRouter()
|
||||
api_router.include_router(login.router)
|
||||
@@ -9,6 +18,7 @@ api_router.include_router(utils.router)
|
||||
api_router.include_router(flows.router)
|
||||
api_router.include_router(flows.ws_router)
|
||||
api_router.include_router(secrets.router)
|
||||
api_router.include_router(alerts.router)
|
||||
# Always mounted so the generated SDK stays the same shape; the endpoints
|
||||
# themselves refuse to work unless MCP is switched on.
|
||||
api_router.include_router(oauth.router)
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
"""Alerting configuration: which failures go where."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
|
||||
from app.api.deps import FlowControllerDep, get_current_user
|
||||
from app.core.config import settings
|
||||
from app.flow.alerts import Alert, AlertsConfig
|
||||
from app.models import Message
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/alerts", tags=["alerts"], dependencies=[Depends(get_current_user)]
|
||||
)
|
||||
|
||||
|
||||
def _path() -> Path:
|
||||
return settings.ALERTS_FILE
|
||||
|
||||
|
||||
def read_config() -> AlertsConfig:
|
||||
"""The stored configuration, or an empty one. Blocking."""
|
||||
path = _path()
|
||||
if not path.exists():
|
||||
return AlertsConfig()
|
||||
try:
|
||||
return AlertsConfig.model_validate_json(path.read_text())
|
||||
except Exception:
|
||||
# A hand-edited file that no longer parses must not stop the engine.
|
||||
return AlertsConfig()
|
||||
|
||||
|
||||
def write_config(config: AlertsConfig) -> None:
|
||||
"""Blocking."""
|
||||
path = _path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(config.model_dump_json(indent=2))
|
||||
|
||||
|
||||
@router.get("/config", response_model=AlertsConfig)
|
||||
async def read_alerts_config() -> Any:
|
||||
"""What the engine alerts on, and where it sends it."""
|
||||
return await run_in_threadpool(read_config)
|
||||
|
||||
|
||||
@router.put("/config", response_model=AlertsConfig)
|
||||
async def save_alerts_config(body: AlertsConfig, controller: FlowControllerDep) -> Any:
|
||||
"""Replace the configuration. Takes effect immediately."""
|
||||
known = {channel.name for channel in body.channels}
|
||||
for rule in body.rules:
|
||||
missing = [name for name in rule.channels if name not in known]
|
||||
if missing:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail=f"No channel named {', '.join(repr(m) for m in missing)}",
|
||||
)
|
||||
|
||||
await run_in_threadpool(write_config, body)
|
||||
if controller.alerts is not None:
|
||||
controller.alerts.config = body
|
||||
return body
|
||||
|
||||
|
||||
@router.post("/test/{channel_name}", response_model=Message)
|
||||
async def test_channel(channel_name: str, controller: FlowControllerDep) -> Any:
|
||||
"""Send one alert, to prove the channel works before relying on it."""
|
||||
if controller.alerts is None:
|
||||
raise HTTPException(status_code=503, detail="Alerting is not running")
|
||||
|
||||
config = controller.alerts.config
|
||||
channel = next((c for c in config.channels if c.name == channel_name), None)
|
||||
if channel is None:
|
||||
raise HTTPException(status_code=404, detail=f"No channel '{channel_name}'")
|
||||
|
||||
await controller.alerts.send(
|
||||
channel,
|
||||
Alert(
|
||||
title="Fluksio test alert",
|
||||
body="If you are reading this, the channel works.",
|
||||
severity="warning",
|
||||
),
|
||||
)
|
||||
return Message(message=f"Sent a test alert through '{channel_name}'")
|
||||
@@ -41,6 +41,9 @@ class Settings(BaseSettings):
|
||||
# Flows live on disk as a git repository; secrets stay outside it.
|
||||
FLOWS_DIR: Path = Path("flow-data/flows")
|
||||
SECRETS_FILE: Path = Path("flow-data/secrets.enc")
|
||||
# Which failures reach which channel. Beside the flows, not in them:
|
||||
# alerting is the deployment's concern, not any one flow's.
|
||||
ALERTS_FILE: Path = Path("flow-data/alerts.json")
|
||||
# The MCP endpoint, and the OAuth server agents authenticate against. Off
|
||||
# until someone asks for it: it opens client registration to the network.
|
||||
MCP_ENABLED: bool = False
|
||||
|
||||
@@ -0,0 +1,331 @@
|
||||
"""Failing loudly: engine failures reach a person, not just a log line.
|
||||
|
||||
Everything that goes wrong already travels the event bus — a node raising, a
|
||||
connection dropping, a flow being quarantined, the queue going away. Until now
|
||||
the only subscriber was the editor's websocket, so a failure at three in the
|
||||
morning was invisible.
|
||||
|
||||
The alert manager subscribes to the same bus and forwards what matters to a
|
||||
channel the operator configured. What it mostly does is *not* send: the same
|
||||
node failing every second is one alert, not thirty-six thousand.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from collections import deque
|
||||
from typing import Any, Literal
|
||||
|
||||
import httpx
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.flow.events import EventBus
|
||||
from app.flow.secrets import resolve_params
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# What is worth waking someone for. Everything else on the bus is traffic.
|
||||
ALERTING_EVENTS = {
|
||||
"node_error",
|
||||
"node_health",
|
||||
"flow_quarantined",
|
||||
"task_crashed",
|
||||
"engine_degraded",
|
||||
"cascade_dropped",
|
||||
"queue_unavailable",
|
||||
}
|
||||
|
||||
# The same fault repeating is the same alert.
|
||||
DEFAULT_COOLDOWN_S = 900.0
|
||||
# A connection flapping is one story, not one alert per transition.
|
||||
FLAP_WINDOW_S = 600.0
|
||||
FLAP_THRESHOLD = 3
|
||||
# However bad it gets, this is the most anyone is told per hour.
|
||||
RATE_LIMIT = 10
|
||||
RATE_WINDOW_S = 3600.0
|
||||
|
||||
|
||||
class Channel(BaseModel):
|
||||
"""Somewhere to send an alert."""
|
||||
|
||||
name: str
|
||||
kind: Literal["ntfy", "smtp", "webhook"]
|
||||
enabled: bool = True
|
||||
# ntfy: server + topic + optional token. smtp: to. webhook: url.
|
||||
config: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
|
||||
class Rule(BaseModel):
|
||||
"""Which events go to which channels."""
|
||||
|
||||
events: list[str] = Field(default_factory=list)
|
||||
channels: list[str] = Field(default_factory=list)
|
||||
cooldown_s: float = DEFAULT_COOLDOWN_S
|
||||
|
||||
def matches(self, event_type: str) -> bool:
|
||||
# No list means every alerting event.
|
||||
return not self.events or event_type in self.events
|
||||
|
||||
|
||||
class AlertsConfig(BaseModel):
|
||||
"""The whole alerting setup, as stored and as the API sees it."""
|
||||
|
||||
enabled: bool = True
|
||||
channels: list[Channel] = Field(default_factory=list)
|
||||
rules: list[Rule] = Field(default_factory=list)
|
||||
|
||||
|
||||
class Alert(BaseModel):
|
||||
"""What a channel is asked to deliver."""
|
||||
|
||||
title: str
|
||||
body: str
|
||||
severity: Literal["warning", "error"] = "error"
|
||||
flow: str = ""
|
||||
node: str = ""
|
||||
|
||||
|
||||
def describe(event: dict[str, Any]) -> Alert | None:
|
||||
"""Turn an engine event into something worth reading, or nothing."""
|
||||
kind = event.get("type")
|
||||
flow = str(event.get("flow") or "")
|
||||
node = str(event.get("node") or "")
|
||||
error = str(event.get("error") or event.get("reason") or "")
|
||||
|
||||
if kind == "node_error":
|
||||
return Alert(
|
||||
title=f"{node or 'A node'} failed",
|
||||
body=error or "The node raised while running.",
|
||||
flow=flow,
|
||||
node=node,
|
||||
)
|
||||
if kind == "node_health":
|
||||
if event.get("status") != "down":
|
||||
return None
|
||||
return Alert(
|
||||
title=f"{node or 'A node'} lost its connection",
|
||||
body=str(event.get("detail") or "Reported itself down."),
|
||||
severity="warning",
|
||||
flow=flow,
|
||||
node=node,
|
||||
)
|
||||
if kind == "flow_quarantined":
|
||||
return Alert(
|
||||
title=f"Flow '{flow}' was quarantined",
|
||||
body=(
|
||||
f"It kept crashing, so the engine stopped restarting it. {error}"
|
||||
).strip(),
|
||||
flow=flow,
|
||||
)
|
||||
if kind == "task_crashed":
|
||||
return Alert(
|
||||
title=f"{event.get('task') or 'A background task'} crashed",
|
||||
body=f"{error} Restarting it.",
|
||||
severity="warning",
|
||||
flow=flow,
|
||||
)
|
||||
if kind == "engine_degraded":
|
||||
return Alert(title="The engine is struggling", body=error or "Degraded.")
|
||||
if kind == "cascade_dropped":
|
||||
return Alert(
|
||||
title="Work was given up on",
|
||||
body=(
|
||||
f"An item for {node or 'a node'} came back "
|
||||
f"{event.get('deliveries')} times and was set aside."
|
||||
),
|
||||
flow=flow,
|
||||
node=node,
|
||||
)
|
||||
if kind == "queue_unavailable":
|
||||
return Alert(title="The work queue is unreachable", body=error or "")
|
||||
return None
|
||||
|
||||
|
||||
def dedup_key(event: dict[str, Any]) -> str:
|
||||
"""What counts as "the same alert again"."""
|
||||
return f"{event.get('type')}:{event.get('node') or event.get('flow') or ''}"
|
||||
|
||||
|
||||
class AlertManager:
|
||||
"""Watches the event bus and tells someone when it matters."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
events: EventBus,
|
||||
config: AlertsConfig | None = None,
|
||||
now: Any = time.monotonic,
|
||||
) -> None:
|
||||
self._events = events
|
||||
self.config = config or AlertsConfig()
|
||||
self._now = now
|
||||
self._last_sent: dict[str, float] = {}
|
||||
self._suppressed: dict[str, int] = {}
|
||||
self._health_flips: dict[str, deque[float]] = {}
|
||||
self._flapping: dict[str, float] = {}
|
||||
self._recent: deque[float] = deque()
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# The loop
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
async def run(self) -> None:
|
||||
"""Consume the bus until cancelled."""
|
||||
async with self._events.subscribe() as queue:
|
||||
while True:
|
||||
event = await queue.get()
|
||||
try:
|
||||
await self.handle(event)
|
||||
except Exception:
|
||||
logger.exception("Alerting failed for %s", event.get("type"))
|
||||
|
||||
async def handle(self, event: dict[str, Any]) -> None:
|
||||
if not self.config.enabled:
|
||||
return
|
||||
kind = str(event.get("type") or "")
|
||||
if kind not in ALERTING_EVENTS:
|
||||
return
|
||||
if self._flaps(event):
|
||||
return
|
||||
|
||||
alert = describe(event)
|
||||
if alert is None:
|
||||
return
|
||||
|
||||
rules = [r for r in self.config.rules if r.matches(kind)]
|
||||
if not rules:
|
||||
return
|
||||
|
||||
key = dedup_key(event)
|
||||
cooldown = min(r.cooldown_s for r in rules)
|
||||
if not self._due(key, cooldown):
|
||||
return
|
||||
if not self._within_rate_limit():
|
||||
return
|
||||
|
||||
held = self._suppressed.pop(key, 0)
|
||||
if held:
|
||||
alert = alert.model_copy(
|
||||
update={"body": f"{alert.body} ({held} more since the last alert.)"}
|
||||
)
|
||||
|
||||
names = {name for rule in rules for name in rule.channels}
|
||||
for channel in self.config.channels:
|
||||
if channel.enabled and channel.name in names:
|
||||
await self.send(channel, alert)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Deciding whether to speak
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def _due(self, key: str, cooldown: float) -> bool:
|
||||
now = self._now()
|
||||
last = self._last_sent.get(key)
|
||||
if last is not None and now - last < cooldown:
|
||||
self._suppressed[key] = self._suppressed.get(key, 0) + 1
|
||||
return False
|
||||
self._last_sent[key] = now
|
||||
return True
|
||||
|
||||
def _within_rate_limit(self) -> bool:
|
||||
now = self._now()
|
||||
while self._recent and self._recent[0] < now - RATE_WINDOW_S:
|
||||
self._recent.popleft()
|
||||
if len(self._recent) >= RATE_LIMIT:
|
||||
logger.warning("Alert rate limit reached; holding back")
|
||||
return False
|
||||
self._recent.append(now)
|
||||
return True
|
||||
|
||||
def _flaps(self, event: dict[str, Any]) -> bool:
|
||||
"""Is this health event part of a connection flapping up and down?
|
||||
|
||||
A device dropping every ten seconds should produce one alert, then
|
||||
silence until it settles.
|
||||
"""
|
||||
if event.get("type") != "node_health":
|
||||
return False
|
||||
key = str(event.get("node") or "")
|
||||
now = self._now()
|
||||
|
||||
muted_until = self._flapping.get(key)
|
||||
if muted_until is not None:
|
||||
if now < muted_until:
|
||||
# Still flapping — push the window out and stay quiet.
|
||||
if event.get("status") == "down":
|
||||
self._flapping[key] = now + FLAP_WINDOW_S
|
||||
return True
|
||||
del self._flapping[key]
|
||||
|
||||
flips = self._health_flips.setdefault(key, deque())
|
||||
flips.append(now)
|
||||
while flips and flips[0] < now - FLAP_WINDOW_S:
|
||||
flips.popleft()
|
||||
if len(flips) >= FLAP_THRESHOLD * 2:
|
||||
self._flapping[key] = now + FLAP_WINDOW_S
|
||||
logger.info("Node '%s' is flapping; muting its health alerts", key)
|
||||
return False
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Delivery
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
async def send(self, channel: Channel, alert: Alert) -> None:
|
||||
try:
|
||||
config = resolve_params(channel.config)
|
||||
except Exception as exc:
|
||||
logger.error("Channel '%s' has unusable settings: %s", channel.name, exc)
|
||||
return
|
||||
|
||||
try:
|
||||
if channel.kind == "ntfy":
|
||||
await self._send_ntfy(config, alert)
|
||||
elif channel.kind == "webhook":
|
||||
await self._send_webhook(config, alert)
|
||||
else:
|
||||
await self._send_email(config, alert)
|
||||
except Exception as exc:
|
||||
logger.error("Could not alert through '%s': %s", channel.name, exc)
|
||||
|
||||
async def _send_ntfy(self, config: dict[str, Any], alert: Alert) -> None:
|
||||
server = str(config.get("server") or "https://ntfy.sh").rstrip("/")
|
||||
topic = config.get("topic")
|
||||
if not topic:
|
||||
raise ValueError("ntfy needs a topic")
|
||||
headers = {
|
||||
"Title": alert.title,
|
||||
"Priority": "high" if alert.severity == "error" else "default",
|
||||
"Tags": "warning" if alert.severity == "warning" else "rotating_light",
|
||||
}
|
||||
if config.get("token"):
|
||||
headers["Authorization"] = f"Bearer {config['token']}"
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
response = await client.post(
|
||||
f"{server}/{topic}", content=alert.body.encode(), headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
async def _send_webhook(self, config: dict[str, Any], alert: Alert) -> None:
|
||||
url = config.get("url")
|
||||
if not url:
|
||||
raise ValueError("a webhook channel needs a url")
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
response = await client.post(str(url), json=alert.model_dump())
|
||||
response.raise_for_status()
|
||||
|
||||
async def _send_email(self, config: dict[str, Any], alert: Alert) -> None:
|
||||
from app.core.config import settings
|
||||
from app.utils import send_email
|
||||
|
||||
recipient = config.get("to")
|
||||
if not recipient:
|
||||
raise ValueError("an smtp channel needs a recipient")
|
||||
if not settings.emails_enabled:
|
||||
raise RuntimeError("no SMTP configuration")
|
||||
await asyncio.to_thread(
|
||||
send_email,
|
||||
email_to=str(recipient),
|
||||
subject=alert.title,
|
||||
html_content=f"<p>{alert.body}</p>",
|
||||
)
|
||||
@@ -22,6 +22,7 @@ from typing import Any, cast
|
||||
from fastapi import FastAPI
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
|
||||
from app.flow.alerts import AlertManager
|
||||
from app.flow.events import EventBus
|
||||
from app.flow.executor import ExecutionService
|
||||
from app.flow.messages import MessageSpec, qualify
|
||||
@@ -199,6 +200,7 @@ class FlowController:
|
||||
max_workers: int | None = None,
|
||||
fastapi_app: FastAPI | None = None,
|
||||
execution: ExecutionService | None = None,
|
||||
alerts: AlertManager | None = None,
|
||||
) -> None:
|
||||
self.store = store
|
||||
self.state = state if state is not None else MemoryState()
|
||||
@@ -207,6 +209,7 @@ class FlowController:
|
||||
self.app = fastapi_app
|
||||
# Without one, every trigger runs inline where it was raised.
|
||||
self.execution = execution
|
||||
self.alerts = alerts
|
||||
|
||||
self.pipeline: Pipeline | None = None
|
||||
self.loaded: dict[str, LoadedNode] = {}
|
||||
|
||||
@@ -10,9 +10,11 @@ from fastapi.routing import APIRoute
|
||||
from starlette.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.api.main import api_router
|
||||
from app.api.routes.alerts import read_config as read_alerts_config
|
||||
from app.core import security
|
||||
from app.core.config import settings
|
||||
from app.flow import logs
|
||||
from app.flow.alerts import AlertManager
|
||||
from app.flow.controller import FlowController
|
||||
from app.flow.events import event_bus
|
||||
from app.flow.executor import ExecutionService
|
||||
@@ -65,6 +67,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
# Connectors register their node types before any flow is built with them.
|
||||
load_plugins()
|
||||
|
||||
alerts = AlertManager(event_bus, config=read_alerts_config())
|
||||
execution = ExecutionService(
|
||||
queue=_work_queue(),
|
||||
max_workers=settings.FLOW_MAX_WORKERS,
|
||||
@@ -77,11 +80,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
max_workers=settings.FLOW_MAX_WORKERS,
|
||||
fastapi_app=app,
|
||||
execution=execution,
|
||||
alerts=alerts,
|
||||
)
|
||||
app.state.flow_controller = controller
|
||||
watchdog = LoopWatchdog(event_bus)
|
||||
app.state.watchdog = watchdog
|
||||
watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
|
||||
alerts_task = asyncio.create_task(alerts.run(), name="alert-manager")
|
||||
await controller.start()
|
||||
try:
|
||||
# A mounted sub-app gets no lifespan of its own, so the MCP session
|
||||
@@ -90,6 +95,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
yield
|
||||
finally:
|
||||
watchdog_task.cancel()
|
||||
alerts_task.cancel()
|
||||
await controller.stop()
|
||||
close_shared_client()
|
||||
if settings.MCP_ENABLED:
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
"""Alerting: failing loudly once, not thirty-six thousand times."""
|
||||
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from app.flow.alerts import (
|
||||
FLAP_THRESHOLD,
|
||||
RATE_LIMIT,
|
||||
Alert,
|
||||
AlertManager,
|
||||
AlertsConfig,
|
||||
Channel,
|
||||
Rule,
|
||||
describe,
|
||||
)
|
||||
from app.flow.events import EventBus
|
||||
|
||||
|
||||
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)],
|
||||
)
|
||||
alerts = AlertManager(EventBus(), config=config, now=clock)
|
||||
|
||||
async def capture(channel, alert):
|
||||
sent.append(alert)
|
||||
|
||||
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", "status": "down"}
|
||||
)
|
||||
clock.advance(5)
|
||||
await alerts.handle(
|
||||
{"type": "node_health", "node": "heating.pump", "status": "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_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"),
|
||||
({"type": "node_health", "status": "ok"}, None),
|
||||
],
|
||||
)
|
||||
def test_every_alerting_event_reads_as_a_sentence(event, expected):
|
||||
alert = describe(event)
|
||||
assert (alert.title if alert else None) == expected
|
||||
Reference in New Issue
Block a user