From e0d7cec5716b9a413c0a5ad1649edde49ee2cb45 Mon Sep 17 00:00:00 2001 From: Lars Laskowski Date: Tue, 18 Aug 2026 04:55:15 +0000 Subject: [PATCH] Isolate webhook delivery per destination and skip futile retries Two failure modes made webhook delivery degrade far worse than it needed to when a single endpoint went bad: - deliver treated every non-2xx alike, so a permanent 400/404/410 burned the full retry/backoff budget re-POSTing an identical body that could never be accepted. - A single serial worker drained the shared queue, so that wasted budget (and any slow endpoint) head-of-line-blocked every other webhook and every queued event. Give each webhook its own bounded queue, worker goroutine, and rate-limit state, so one dead destination can only delay and drop its own events. The severity filter moves to Notify, where it is applied inline while fanning out; it is pure and cheap, so it costs the collector nothing. Classify delivery errors: transport failures and 5xx stay retryable, as do 408 and 429 which explicitly invite a retry, while every other 4xx is a permanent rejection that ends the attempt immediately. Closes #63 --- docs/ARCHITECTURE.md | 29 +++-- internal/alert/notify.go | 234 +++++++++++++++++++++------------- internal/alert/notify_test.go | 149 ++++++++++++++++++++++ 3 files changed, 316 insertions(+), 96 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ee1a1f6..957299c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -176,17 +176,30 @@ so the server-reported alert state always agrees with what the dashboard visuall `alert.Notifier` POSTs alert transition events to zero or more configured HTTP webhooks (Slack, Discord, Home Assistant, ntfy, or any endpoint accepting a JSON/templated body). -Delivery is fully decoupled from collection: `Notify` only enqueues events onto a bounded -channel (`defaultNotifyQueueSize` = 256) and returns immediately; a single background -worker goroutine (`Start`/`dispatch`) drains the queue and performs the actual (retrying, -potentially slow) HTTP calls. This guarantees a hung or failing webhook endpoint can never -stall the collector's fast tick — if the queue fills up (a persistent backlog of slow -deliveries), further events are dropped with a logged warning rather than blocking. +Delivery is fully decoupled from collection: `Notify` only enqueues events and returns +immediately; background worker goroutines (`Start`/`dispatch`) drain the queues and +perform the actual (retrying, potentially slow) HTTP calls. This guarantees a hung or +failing webhook endpoint can never stall the collector's fast tick — if a queue fills up +(a persistent backlog of slow deliveries), further events are dropped with a logged +warning rather than blocking. + +**Each webhook gets its own bounded queue (`defaultNotifyQueueSize` = 256) and its own +worker** (`webhookWorker`). Delivery to one destination is therefore never held up by +another: a webhook pointing at a dead host can only delay — and only ever drop — its own +events. `Notify` applies the severity filter inline (it is pure and cheap) and fans the +event out onto the queue of every webhook it reaches. Per-webhook behavior: a severity filter (`min_level`), an optional `text/template`-rendered body (defaulting to a fixed JSON payload when unset), retry with -exponential backoff up to `notify_max_retries`, and a per-`(url, metric, resource)` -rate limiter (`notify_min_interval_seconds`) that coalesces repeated firings. `cleared` +exponential backoff up to `notify_max_retries`, and a per-`(metric, resource)` +rate limiter (`notify_min_interval_seconds`) that coalesces repeated firings. + +**Only retryable failures consume the retry budget.** Transport errors (DNS, refused +connections, timeouts) and 5xx responses are transient, as are `408 Request Timeout` and +`429 Too Many Requests`, which explicitly invite a retry. Every other 4xx is a permanent +rejection of that exact request — a revoked webhook URL, a malformed payload — so +`deliver` gives up after the first attempt instead of re-POSTing an identical body that +cannot succeed. `cleared` events deliberately bypass the rate limiter — a recovery signal must always reach a state-based consumer (e.g. a Home Assistant `binary_sensor`) so it can never get stuck reporting an alert that has actually cleared; only repeated *firings* are coalesced. diff --git a/internal/alert/notify.go b/internal/alert/notify.go index 513cdea..e481305 100644 --- a/internal/alert/notify.go +++ b/internal/alert/notify.go @@ -3,10 +3,15 @@ package alert // notify.go delivers alert transition events to configured HTTP webhooks. // // Delivery is fully decoupled from metric collection: Notify only enqueues -// events onto a bounded channel and returns immediately, and a single -// background worker drains the queue and performs the (potentially slow, -// retrying) HTTP POSTs. This guarantees that a hung or failing webhook can -// never block the collector's fast tick. +// events onto bounded channels and returns immediately, and background workers +// drain those queues and perform the (potentially slow, retrying) HTTP POSTs. +// This guarantees that a hung or failing webhook can never block the +// collector's fast tick. +// +// Each webhook gets its own queue and worker, so a dead or slow endpoint can +// only delay its own deliveries — it can never head-of-line-block the other +// webhooks. Attempts that fail with a permanent client error (most 4xx) are +// not retried at all, since re-POSTing the same body cannot change the answer. // // A single generic webhook — a URL plus an optional Go text/template body — // is enough to target Slack, Discord, Home Assistant, ntfy, and similar @@ -16,6 +21,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "log/slog" "net/http" @@ -47,18 +53,27 @@ type webhook struct { // with NewNotifier, call Start once to launch its worker, and feed it events // via Notify. It is safe for concurrent use. type Notifier struct { - webhooks []webhook + workers []*webhookWorker client *http.Client maxRetries int backoff time.Duration minInterval time.Duration log *slog.Logger + wg sync.WaitGroup +} + +// webhookWorker owns everything that belongs to exactly one destination: its +// resolved config, its own bounded queue, and its rate-limiting state. Giving +// every webhook a private queue and goroutine is what keeps one unreachable +// endpoint from delaying deliveries to the healthy ones. +type webhookWorker struct { + wh webhook queue chan Event - wg sync.WaitGroup - // lastSent tracks the last delivery time per webhook URL for rate - // limiting. Only the worker goroutine touches it, so it needs no lock. + // lastSent tracks the last delivery time per (metric, resource) for rate + // limiting. Only this webhook's own goroutine touches it, so it needs no + // lock. lastSent map[string]time.Time } @@ -74,7 +89,7 @@ func NewNotifier(cfg config.Alerts, log *slog.Logger) (*Notifier, error) { log = slog.Default() } - webhooks := make([]webhook, 0, len(cfg.Webhooks)) + workers := make([]*webhookWorker, 0, len(cfg.Webhooks)) for i, w := range cfg.Webhooks { wh := webhook{ url: w.URL, @@ -95,18 +110,20 @@ func NewNotifier(cfg config.Alerts, log *slog.Logger) (*Notifier, error) { } wh.tmpl = tmpl } - webhooks = append(webhooks, wh) + workers = append(workers, &webhookWorker{ + wh: wh, + queue: make(chan Event, defaultNotifyQueueSize), + lastSent: make(map[string]time.Time), + }) } return &Notifier{ - webhooks: webhooks, + workers: workers, client: &http.Client{}, maxRetries: cfg.NotifyMaxRetries, backoff: time.Duration(cfg.NotifyRetryBackoffSeconds * float64(time.Second)), minInterval: time.Duration(cfg.NotifyMinIntervalSeconds * float64(time.Second)), log: log, - queue: make(chan Event, defaultNotifyQueueSize), - lastSent: make(map[string]time.Time), }, nil } @@ -121,117 +138,134 @@ func parseMinLevel(s string) Level { } } -// Start launches the background delivery worker. It returns immediately; the -// worker runs until ctx is canceled, at which point in-flight retries stop -// promptly and any queued-but-undelivered events are dropped. Call Start at -// most once. +// Start launches one background delivery worker per configured webhook. It +// returns immediately; the workers run until ctx is canceled, at which point +// in-flight retries stop promptly and any queued-but-undelivered events are +// dropped. Call Start at most once. func (n *Notifier) Start(ctx context.Context) { - n.wg.Add(1) - go func() { - defer n.wg.Done() - for { - select { - case <-ctx.Done(): - return - case ev := <-n.queue: - n.dispatch(ctx, ev) + for _, w := range n.workers { + n.wg.Add(1) + go func(w *webhookWorker) { + defer n.wg.Done() + for { + select { + case <-ctx.Done(): + return + case ev := <-w.queue: + n.dispatch(ctx, w, ev) + } } - } - }() + }(w) + } } -// Stop waits for the delivery worker to exit. It must be called only after the -// context passed to Start has been canceled, otherwise it blocks until that -// happens; any events still queued at cancellation are dropped, since delivery -// is best-effort. It joins the worker goroutine so a caller can be sure no -// delivery is still in flight once Stop returns. +// Stop waits for the delivery workers to exit. It must be called only after +// the context passed to Start has been canceled, otherwise it blocks until +// that happens; any events still queued at cancellation are dropped, since +// delivery is best-effort. It joins the worker goroutines so a caller can be +// sure no delivery is still in flight once Stop returns. func (n *Notifier) Stop() { n.wg.Wait() } -// Notify enqueues events for asynchronous delivery. It never blocks: if the -// queue is full (a backlog of slow deliveries), the event is dropped with a -// warning rather than stalling the collector. +// Notify fans each event out onto the queue of every webhook it matches, +// applying the (cheap, stateless) per-webhook severity filter inline. It never +// blocks: if a webhook's queue is full (a backlog of slow deliveries to that +// endpoint), the event is dropped for that webhook with a warning rather than +// stalling the collector — and only that webhook is affected. func (n *Notifier) Notify(events []Event) { for _, ev := range events { - select { - case n.queue <- ev: - default: - n.log.Warn("alert notification queue full, dropping event", - "metric", ev.Metric, "resource", ev.Resource, "kind", ev.Kind) + for _, w := range n.workers { + if !eventReaches(ev, w.wh.minLevel) { + continue + } + select { + case w.queue <- ev: + default: + n.log.Warn("alert notification queue full, dropping event", + "url", w.wh.url, "metric", ev.Metric, "resource", ev.Resource, "kind", ev.Kind) + } } } } -// dispatch delivers one event to every webhook it matches, applying the -// per-webhook severity filter and rate limit. -func (n *Notifier) dispatch(ctx context.Context, ev Event) { - for _, wh := range n.webhooks { - if !eventReaches(ev, wh.minLevel) { - continue - } - // cleared events bypass the rate limiter: a recovery signal must - // always be delivered so a state-based consumer (e.g. a Home Assistant - // binary_sensor) can never get stuck reporting an alert that has - // actually cleared. The limiter only coalesces repeated firings. - if ev.Kind != KindCleared && n.rateLimited(wh.url, ev) { - n.log.Warn("alert notification rate-limited, dropping event", - "url", wh.url, "metric", ev.Metric, "resource", ev.Resource, "kind", ev.Kind) - continue - } - body, err := renderBody(wh, ev) - if err != nil { - n.log.Error("alert notification render failed", "url", wh.url, "error", err) - continue - } - if n.deliver(ctx, wh, body) { - n.recordSent(wh.url, ev) - } +// dispatch delivers one event to one webhook, applying that webhook's rate +// limit. It runs on the webhook's own worker goroutine. +func (n *Notifier) dispatch(ctx context.Context, w *webhookWorker, ev Event) { + // cleared events bypass the rate limiter: a recovery signal must always be + // delivered so a state-based consumer (e.g. a Home Assistant + // binary_sensor) can never get stuck reporting an alert that has actually + // cleared. The limiter only coalesces repeated firings. + if ev.Kind != KindCleared && n.rateLimited(w, ev) { + n.log.Warn("alert notification rate-limited, dropping event", + "url", w.wh.url, "metric", ev.Metric, "resource", ev.Resource, "kind", ev.Kind) + return + } + body, err := renderBody(w.wh, ev) + if err != nil { + n.log.Error("alert notification render failed", "url", w.wh.url, "error", err) + return + } + if n.deliver(ctx, w.wh, body) { + n.recordSent(w, ev) } } -// rateLimited reports whether delivering ev to url should be suppressed -// because the previous successful delivery of the same metric to the same -// URL was too recent. The rate limit is keyed per (url, metric, resource) so -// a fast-flapping metric can't flood a webhook, while a distinct metric -// alerting in the same tick is still delivered. Only firing events reach -// here (cleared events bypass the limiter in dispatch), so it purely -// coalesces repeated escalations. The event timestamp (not wall clock) drives -// the decision so it is deterministic and testable. -func (n *Notifier) rateLimited(url string, ev Event) bool { +// rateLimited reports whether delivering ev to w should be suppressed because +// the previous successful delivery of the same metric to the same webhook was +// too recent. The state lives on the worker and is keyed per +// (metric, resource) so a fast-flapping metric can't flood a webhook, while a +// distinct metric alerting in the same tick is still delivered. Only firing +// events reach here (cleared events bypass the limiter in dispatch), so it +// purely coalesces repeated escalations. The event timestamp (not wall clock) +// drives the decision so it is deterministic and testable. +func (n *Notifier) rateLimited(w *webhookWorker, ev Event) bool { if n.minInterval <= 0 { return false } - key := url + "\x00" + ev.Metric + "\x00" + ev.Resource - last, ok := n.lastSent[key] + last, ok := w.lastSent[rateLimitKey(ev)] return ok && ev.At.Sub(last) < n.minInterval } -// recordSent stamps the last-delivery time for (url, metric, resource) after -// a successful delivery, so the rate limiter counts deliveries rather than +// recordSent stamps the last-delivery time for (metric, resource) after a +// successful delivery, so the rate limiter counts deliveries rather than // attempts: a failed delivery must not suppress the next firing. -func (n *Notifier) recordSent(url string, ev Event) { +func (n *Notifier) recordSent(w *webhookWorker, ev Event) { if n.minInterval <= 0 { return } - key := url + "\x00" + ev.Metric + "\x00" + ev.Resource - n.lastSent[key] = ev.At + w.lastSent[rateLimitKey(ev)] = ev.At +} + +// rateLimitKey identifies the alert stream an event belongs to within one +// webhook. The NUL separator keeps metric and resource unambiguous. +func rateLimitKey(ev Event) string { + return ev.Metric + "\x00" + ev.Resource } // deliver POSTs body to a webhook, retrying with exponential backoff on -// failure until it succeeds, exhausts maxRetries, or ctx is canceled. It -// reports whether delivery ultimately succeeded, and always returns without -// panicking so a dead endpoint can't crash the worker. +// failure until it succeeds, exhausts maxRetries, hits a permanent error, or +// ctx is canceled. It reports whether delivery ultimately succeeded, and +// always returns without panicking so a dead endpoint can't crash the worker. func (n *Notifier) deliver(ctx context.Context, wh webhook, body []byte) bool { backoff := n.backoff for attempt := 0; ; attempt++ { - if err := n.post(ctx, wh, body); err == nil { + err := n.post(ctx, wh, body) + switch { + case err == nil: return true - } else if attempt >= n.maxRetries { + case !retryable(err): + // A rejected request (bad URL, bad payload, revoked webhook) will + // be rejected again identically, so burning the retry budget only + // delays this webhook's remaining events for no benefit. + n.log.Error("alert notification rejected, not retrying", + "url", wh.url, "attempts", attempt+1, "error", err) + return false + case attempt >= n.maxRetries: n.log.Error("alert notification giving up after retries", "url", wh.url, "attempts", attempt+1, "error", err) return false - } else { + default: n.log.Warn("alert notification delivery failed, will retry", "url", wh.url, "attempt", attempt+1, "error", err) } @@ -242,6 +276,30 @@ func (n *Notifier) deliver(ctx context.Context, wh webhook, body []byte) bool { } } +// statusError is a non-2xx webhook response. +type statusError struct{ code int } + +func (e *statusError) Error() string { return fmt.Sprintf("webhook returned status %d", e.code) } + +// retryable reports whether a failed delivery attempt is worth repeating. +// Transport failures (DNS, refused connections, timeouts) are transient by +// nature, so they always are. HTTP status errors only are when the server +// might answer differently for an identical request: 5xx (server-side +// trouble), plus 408 Request Timeout and 429 Too Many Requests, which +// explicitly invite a retry. Every other 4xx is a permanent rejection of this +// request. +func retryable(err error) bool { + var se *statusError + if !errors.As(err, &se) { + return true + } + switch se.code { + case http.StatusRequestTimeout, http.StatusTooManyRequests: + return true + } + return se.code < 400 || se.code >= 500 +} + // post performs a single delivery attempt, returning an error for a transport // failure or a non-2xx response. func (n *Notifier) post(ctx context.Context, wh webhook, body []byte) error { @@ -260,7 +318,7 @@ func (n *Notifier) post(ctx context.Context, wh webhook, body []byte) error { } defer func() { _ = resp.Body.Close() }() if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("webhook returned status %d", resp.StatusCode) + return &statusError{code: resp.StatusCode} } return nil } diff --git a/internal/alert/notify_test.go b/internal/alert/notify_test.go index 1fdb462..421362a 100644 --- a/internal/alert/notify_test.go +++ b/internal/alert/notify_test.go @@ -3,6 +3,8 @@ package alert import ( "context" "encoding/json" + "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -431,3 +433,150 @@ func TestEventReaches(t *testing.T) { }) } } + +// A permanent client error (e.g. 404 from a revoked webhook URL) must not be +// retried: re-POSTing the identical body cannot change the answer, and the +// retry budget would only delay this webhook's remaining events +// (regression test for #63). +func TestNotifier_DoesNotRetryPermanentClientError(t *testing.T) { + var attempts int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&attempts, 1) + w.WriteHeader(http.StatusNotFound) + })) + defer srv.Close() + + n, err := NewNotifier(config.Alerts{ + Webhooks: []config.Webhook{{URL: srv.URL}}, + NotifyMaxRetries: 3, + NotifyRetryBackoffSeconds: 0.001, + }, nil) + if err != nil { + t.Fatalf("NewNotifier: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + n.Start(ctx) + + n.Notify([]Event{firedEvent(time.Now())}) + + // Give the worker ample time to run any retries it might (wrongly) attempt. + time.Sleep(100 * time.Millisecond) + if got := atomic.LoadInt32(&attempts); got != 1 { + t.Fatalf("expected exactly 1 attempt for a 404, got %d", got) + } +} + +// 429 Too Many Requests explicitly invites a retry, so unlike other 4xx it +// must still consume the retry budget. +func TestNotifier_RetriesRetryableClientError(t *testing.T) { + var attempts int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&attempts, 1) + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + + n, err := NewNotifier(config.Alerts{ + Webhooks: []config.Webhook{{URL: srv.URL}}, + NotifyMaxRetries: 2, + NotifyRetryBackoffSeconds: 0.001, + }, nil) + if err != nil { + t.Fatalf("NewNotifier: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + n.Start(ctx) + + n.Notify([]Event{firedEvent(time.Now())}) + + deadline := time.After(2 * time.Second) + for atomic.LoadInt32(&attempts) < 3 { + select { + case <-deadline: + t.Fatalf("expected 3 attempts (1 + 2 retries) for a 429, got %d", atomic.LoadInt32(&attempts)) + case <-time.After(time.Millisecond): + } + } +} + +func TestRetryable(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"transport error", errors.New("connection refused"), true}, + {"400 bad request", &statusError{code: http.StatusBadRequest}, false}, + {"404 not found", &statusError{code: http.StatusNotFound}, false}, + {"410 gone", &statusError{code: http.StatusGone}, false}, + {"408 request timeout", &statusError{code: http.StatusRequestTimeout}, true}, + {"429 too many requests", &statusError{code: http.StatusTooManyRequests}, true}, + {"500 internal server error", &statusError{code: http.StatusInternalServerError}, true}, + {"503 service unavailable", &statusError{code: http.StatusServiceUnavailable}, true}, + {"304 not modified", &statusError{code: http.StatusNotModified}, true}, + {"wrapped 403", fmt.Errorf("post failed: %w", &statusError{code: http.StatusForbidden}), false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := retryable(c.err); got != c.want { + t.Fatalf("retryable(%v) = %v, want %v", c.err, got, c.want) + } + }) + } +} + +// A hung webhook must not delay deliveries to the other configured webhooks: +// each destination drains its own queue on its own goroutine, so a healthy +// endpoint gets its event while a dead one is still blocked +// (regression test for #63). +func TestNotifier_SlowWebhookDoesNotBlockOthers(t *testing.T) { + release := make(chan struct{}) + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-release + w.WriteHeader(http.StatusOK) + })) + defer slow.Close() + defer close(release) + + fast, ch := newCapturingServer(t, http.StatusOK) + + n, err := NewNotifier(config.Alerts{ + Webhooks: []config.Webhook{{URL: slow.URL}, {URL: fast.URL}}, + }, nil) + if err != nil { + t.Fatalf("NewNotifier: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + n.Start(ctx) + + n.Notify([]Event{firedEvent(time.Now())}) + + // The fast webhook must be delivered while the slow one is still hanging. + waitForRequest(t, ch) +} + +// Rate-limit state is per webhook, so a delivery to one destination must not +// suppress the same event on another. +func TestNotifier_RateLimitStateIsPerWebhook(t *testing.T) { + first, ch1 := newCapturingServer(t, http.StatusOK) + second, ch2 := newCapturingServer(t, http.StatusOK) + + n, err := NewNotifier(config.Alerts{ + Webhooks: []config.Webhook{{URL: first.URL}, {URL: second.URL}}, + NotifyMinIntervalSeconds: 60, + }, nil) + if err != nil { + t.Fatalf("NewNotifier: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + n.Start(ctx) + + n.Notify([]Event{firedEvent(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))}) + + waitForRequest(t, ch1) + waitForRequest(t, ch2) +}