From 9fbc4e7b3a8d5ddff9886852c700d43b9b6322c7 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Tue, 22 Sep 2026 16:52:29 +0100 Subject: [PATCH] fix(notifications): persist and group resolved alert deliveries Prevent simultaneous recoveries from bypassing the configured grouping window. Preserve destination receipts and pending restart recovery, and keep PagerDuty incident keys separate. Include the notification contract and regression coverage with the runtime change. Change-source: pulse-maintainer --- .../v6/internal/subsystems/notifications.md | 44 +++- internal/notifications/notifications.go | 107 +++++++- internal/notifications/notifications_test.go | 1 + internal/notifications/queue.go | 9 +- internal/notifications/resolved_grouping.go | 83 ++++++ .../resolved_grouping_contract_test.go | 245 ++++++++++++++++++ 6 files changed, 475 insertions(+), 14 deletions(-) create mode 100644 internal/notifications/resolved_grouping.go create mode 100644 internal/notifications/resolved_grouping_contract_test.go diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 0aaf7680ac..22156b30f7 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -43,7 +43,9 @@ Unknown and absent persisted targets preserve historical all-destination behavior, and destination tag filters still apply after target selection. Grouping is an explicit runtime policy: `grouping.enabled=false` or a zero window delivers each alert independently, and disabling grouping flushes any -pending alerts as individual deliveries. Grouped provider payloads must retain +in-memory pending firing alerts as individual deliveries. Already admitted +persistent recovery groups retain their original deadline across a policy edit. +Grouped provider payloads must retain every alert, while live ntfy firing deliveries and webhook tests share the same severity-derived title, priority, and tags. Email rendering treats `patrol_finding` as a finding rather than a numeric @@ -139,6 +141,46 @@ stable opaque routing identities and must not expose credentials. ## Current State +### Persistent resolved-alert grouping + +`SendResolvedAlert` applies the configured grouping window to recoveries as +well as firings. Event types are never mixed. With the persistent queue available, +`internal/notifications/resolved_grouping.go` persists the first recovery's +window and coalesces only untouched, not-yet-due pending jobs with identical +notification type, destination configuration and quiet-hours schedule. Queue +scheduling metadata identifies these groups; it is not the provider HTTP method. +Deadlines round up to queue timestamp precision, never shorten the grouping +window, and do not slide with each added occurrence. Retry batches are immutable. +No schema or driver modification is needed. Pending groups survive manager +shutdown/reopen. The queue-unavailable fallback uses a separate in-memory buffer +and does not promise restart durability. + +Admission retains destination-specific successful firing receipts. A destination +that never received an occurrence, or an unannounced alert, must not receive its +recovery. Group extension deduplicates by alert ID and occurrence start time, +retains each resolution timestamp and regenerates the group's operational links. +Delivery consumes only receipts for the occurrences in the delivered job. +Grouping disabled or a zero window keeps new recoveries individual; changing +that policy does not rewrite already-admitted persistent jobs. + +Generic payloads and built-in/custom recovery summary data retain all grouped +alerts and the count, rather than rendering only the first alert. PagerDuty's +built-in Events v2 resolve path remains one deduplication key per request; +it is explicitly excluded from coalescing to avoid consuming other incidents' +receipts. Existing rate limits, retry budgets, destination security checks and +firing grouping are unchanged. + +Verification: `resolved_grouping_contract_test.go` establishes ordinary firing +receipts then resolves fifteen alerts in one process, after manager restart, +and across restart with pending recoveries. It checks unseen-destination and +unannounced-alert suppression, per-occurrence resolution metadata, eleven +recovery renderers, individual PagerDuty keys, and the disabled-grouping burst's +rate-limit/dead-letter consequence. The HTTP adapter is in-process; queue and +receipt persistence are real. Focused and race runs establish this synthetic +boundary, not a full notification-suite result, installed monitor recovery, +provider acceptance, reporter confirmation or release qualification. The existing +singleton delivery test explicitly disables grouping. + ### Webhook retry delay conversion `parseRetryAfterBackoff` bounds parsed integer seconds before multiplying by diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 726b5106a3..8bfb445a17 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -245,6 +245,8 @@ type NotificationManager struct { groupingEnabled bool pendingAlerts []*alerts.Alert groupTimer *time.Timer + pendingResolved []*alerts.Alert + resolvedGroupTimer *time.Timer groupByNode bool publicURL string // Full URL to access Pulse groupByGuest bool @@ -887,6 +889,9 @@ func (n *NotificationManager) SetNotifyOnResolve(enabled bool) { n.mu.Lock() was := n.notifyOnResolve n.notifyOnResolve = enabled + if !enabled { + n.takePendingResolvedLocked() + } n.mu.Unlock() if was != enabled { @@ -949,12 +954,14 @@ func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNod n.groupByGuest = byGuest var pending []*alerts.Alert + var resolved []*alerts.Alert var emailConfig EmailConfig var webhooks []WebhookConfig var appriseConfig AppriseConfig var initialTarget notificationDeliveryTarget var queue *NotificationQueue if !enabled || seconds == 0 { + resolved = n.takePendingResolvedLocked() pending = append(pending, n.pendingAlerts...) n.pendingAlerts = n.pendingAlerts[:0] if n.groupTimer != nil { @@ -969,6 +976,9 @@ func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNod } n.mu.Unlock() + for _, alert := range resolved { + n.dispatchResolvedAlerts([]*alerts.Alert{alert}) + } for _, alert := range pending { n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, []*alerts.Alert{alert}, initialTarget, nil, queue) } @@ -1063,6 +1073,7 @@ func (n *NotificationManager) SetEnabled(enabled bool) { changed = n.enabled != enabled n.enabled = enabled if !enabled { + n.takePendingResolvedLocked() for i := range n.pendingAlerts { n.pendingAlerts[i] = nil } @@ -1408,7 +1419,10 @@ func (n *NotificationManager) filterResolvedJobsByDeliveryReceipt(jobs []notific return filtered } -// SendResolvedAlert delivers notifications for a resolved alert immediately. +// SendResolvedAlert applies the same grouping window to recoveries as firings. +// The two event types have separate buffers; a recovery can never be rendered +// as a firing alert. Normal operation persists the grouping window in the +// delivery queue; the in-memory buffer is only the queue-unavailable fallback. func (n *NotificationManager) SendResolvedAlert(resolved *alerts.ResolvedAlert) { if resolved == nil || resolved.Alert == nil { return @@ -1426,6 +1440,52 @@ func (n *NotificationManager) SendResolvedAlert(resolved *alerts.ResolvedAlert) } annotateResolvedMetadata(alertCopy, resolvedAt) + n.mu.Lock() + if !n.enabled || !n.notifyOnResolve { + n.mu.Unlock() + return + } + if n.queue == nil && n.groupingEnabled && n.groupWindow > 0 { + // Repeated reconciliation of one occurrence must not duplicate batch rows. + for _, pending := range n.pendingResolved { + if pending.ID == alertCopy.ID && pending.StartTime.Equal(alertCopy.StartTime) { + n.mu.Unlock() + return + } + } + n.pendingResolved = append(n.pendingResolved, alertCopy) + if n.resolvedGroupTimer == nil { + n.resolvedGroupTimer = time.AfterFunc(n.groupWindow, n.sendGroupedResolvedAlerts) + } + n.mu.Unlock() + return + } + n.mu.Unlock() + n.dispatchResolvedAlerts([]*alerts.Alert{alertCopy}) +} + +// takePendingResolvedLocked drains only the recovery buffer. Call with mu held. +func (n *NotificationManager) takePendingResolvedLocked() []*alerts.Alert { + pending := n.pendingResolved + n.pendingResolved = nil + if n.resolvedGroupTimer != nil { + n.resolvedGroupTimer.Stop() + n.resolvedGroupTimer = nil + } + return pending +} + +func (n *NotificationManager) sendGroupedResolvedAlerts() { + n.mu.Lock() + pending := n.takePendingResolvedLocked() + n.mu.Unlock() + n.dispatchResolvedAlerts(pending) +} + +func (n *NotificationManager) dispatchResolvedAlerts(alertList []*alerts.Alert) { + if len(alertList) == 0 { + return + } n.mu.RLock() enabled := n.enabled && n.notifyOnResolve emailConfig := copyEmailConfig(n.emailConfig) @@ -1433,18 +1493,13 @@ func (n *NotificationManager) SendResolvedAlert(resolved *alerts.ResolvedAlert) appriseConfig := copyAppriseConfig(n.appriseConfig) queue := n.queue n.mu.RUnlock() - if !enabled { - log.Debug(). - Str("alertID", alertCopy.ID). - Msg("resolved notifications disabled, skipping") return } - - alertsToSend := []*alerts.Alert{alertCopy} - jobs := buildNotificationDeliveryJobs(emailConfig, webhooks, appriseConfig, alertsToSend, eventResolved, resolvedAt) + jobs := buildNotificationDeliveryJobs(emailConfig, webhooks, appriseConfig, alertList, eventResolved, resolvedTimeFromAlerts(alertList)) + // Filter each destination independently at dispatch, not merely when the + // alert enters the window. Never announce a recovery for an undelivered firing. jobs = n.filterResolvedJobsByDeliveryReceipt(jobs) - if queue != nil { n.enqueueNotificationJobs(queue, jobs) } else { @@ -1628,6 +1683,14 @@ func buildNotificationDeliveryJobsForSelection( routedAlerts := routeNotificationAlerts(alertsToSend, webhook.TagFilter, webhook.TagMode, webhook.MinimumSeverity, event) if len(routedAlerts) > 0 { webhookCopy := webhook + // PagerDuty Events v2 resolves one dedup_key, not a list. + // Never consume other occurrences' receipts for that request. + if event == eventResolved && webhook.Service == "pagerduty" && strings.TrimSpace(webhook.Template) == "" { + for _, alert := range routedAlerts { + jobs = append(jobs, notificationDeliveryJob{Type: "webhook", Event: event, Alerts: []*alerts.Alert{alert}, ResolvedAt: resolvedAt, WebhookConfig: &webhookCopy}) + } + continue + } jobs = append(jobs, notificationDeliveryJob{ Type: "webhook", Event: event, @@ -1753,6 +1816,12 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue, return false } + n.mu.RLock() + groupWindow := n.groupWindow + if !n.groupingEnabled { + groupWindow = 0 + } + n.mu.RUnlock() anyFailed := false for _, job := range jobs { configJSON, err := configJSONForNotificationDeliveryJob(job) @@ -1770,7 +1839,13 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue, MaxAttempts: 3, NextRetryAt: bucket.nextRetryAt, } - if err := queue.Enqueue(notif); err != nil { + var enqueueErr error + if job.Event == eventResolved && groupWindow > 0 && !(job.WebhookConfig != nil && job.WebhookConfig.Service == "pagerduty" && strings.TrimSpace(job.WebhookConfig.Template) == "") { + enqueueErr = queue.enqueueResolvedGroup(notif, groupWindow) + } else { + enqueueErr = queue.Enqueue(notif) + } + if err := enqueueErr; err != nil { anyFailed = true n.logNotificationJobError(bucket.job, err, "failed to enqueue notification - falling back to direct send") n.dispatchNotificationJobAsync(bucket.job, "failed to send notification after queue enqueue failure") @@ -2685,6 +2760,17 @@ func (n *NotificationManager) sendResolvedWebhook(webhook WebhookConfig, alertLi data.ResolvedAtISO = resolvedAt.Format(time.RFC3339) data.Duration = formatWebhookDuration(resolvedAt.Sub(alert.StartTime)) data.Message = fmt.Sprintf("%s on %s is now healthy", alert.ResourceName, alert.Node) + data.AlertCount = len(alertList) + data.Alerts = alertList + if len(alertList) > 1 { + names := make([]string, 0, len(alertList)) + for _, a := range alertList { + if a != nil { + names = append(names, fmt.Sprintf("%s on %s", a.ResourceName, a.Node)) + } + } + data.Message = fmt.Sprintf("%d alerts resolved: %s", len(alertList), strings.Join(names, "; ")) + } var err error webhook, data, err = n.prepareWebhookDeliveryContext(webhook, data) @@ -4020,6 +4106,7 @@ func (n *NotificationManager) Stop() { n.stopOnce.Do(func() { n.mu.Lock() n.enabled = false + n.takePendingResolvedLocked() queue := n.queue cleanupDone := n.cleanupDone client := n.webhookClient diff --git a/internal/notifications/notifications_test.go b/internal/notifications/notifications_test.go index b0f0d47e7c..5f6517959f 100644 --- a/internal/notifications/notifications_test.go +++ b/internal/notifications/notifications_test.go @@ -4044,6 +4044,7 @@ func TestSendResolvedAlert(t *testing.T) { nm.AddWebhook(webhook) nm.SetNotifyOnResolve(true) + nm.SetGroupingConfig(false, 0, false, false) alert := &alerts.Alert{ ID: "test-alert", diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index bbc76f7a60..3141e7a3c3 100644 --- a/internal/notifications/queue.go +++ b/internal/notifications/queue.go @@ -705,6 +705,12 @@ func normalizeNotificationLinks( // Enqueue adds a notification to the queue func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error { + nq.mu.Lock() + defer nq.mu.Unlock() + return nq.enqueueLocked(notif) +} + +func (nq *NotificationQueue) enqueueLocked(notif *QueuedNotification) error { if notif == nil { return fmt.Errorf("notification cannot be nil") } @@ -780,9 +786,6 @@ func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error { return fmt.Errorf("failed to marshal operational links: %w", err) } - nq.mu.Lock() - defer nq.mu.Unlock() - query := ` INSERT INTO notification_queue (id, type, method, status, alerts, operational_links, config, attempts, max_attempts, created_at, next_retry_at) diff --git a/internal/notifications/resolved_grouping.go b/internal/notifications/resolved_grouping.go new file mode 100644 index 0000000000..e8b2058724 --- /dev/null +++ b/internal/notifications/resolved_grouping.go @@ -0,0 +1,83 @@ +package notifications + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" +) + +// enqueueResolvedGroup persists the window on the first resolution and appends +// subsequent occurrences to that not-yet-due delivery. Only untouched pending +// groups with identical destination configuration and quiet-hours schedule can +// merge. Attempts/retries are immutable batches. The queue mutex also serializes +// claims, so no delivery can be claimed while its payload is being extended. +// Method is queue scheduling metadata, not the HTTP method in Config. +func (nq *NotificationQueue) enqueueResolvedGroup(notif *QueuedNotification, window time.Duration) error { + if notif == nil || window <= 0 { + return nq.Enqueue(notif) + } + nq.mu.Lock() + defer nq.mu.Unlock() + now := time.Now() + method := "resolved-group" + if notif.NextRetryAt != nil { + method += ":" + notif.NextRetryAt.UTC().Format(time.RFC3339Nano) + } + var id string + var raw []byte + err := nq.db.QueryRow(`SELECT id, alerts FROM notification_queue + WHERE type = ? AND config = ? AND method = ? AND status = 'pending' + AND attempts = 0 AND next_retry_at > ? AND created_at >= ? + ORDER BY created_at, id LIMIT 1`, notif.Type, string(notif.Config), method, now.Unix(), now.Add(-window).Unix()).Scan(&id, &raw) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("read pending resolved group: %w", err) + } + if errors.Is(err, sql.ErrNoRows) { + // Queue deadlines have second precision: round up rather than sending early. + deadline := now.Add(window).Truncate(time.Second).Add(time.Second) + if notif.NextRetryAt == nil || notif.NextRetryAt.Before(deadline) { + notif.NextRetryAt = &deadline + } + notif.Method = method + return nq.enqueueLocked(notif) + } + var batch []*alerts.Alert + if err := json.Unmarshal(raw, &batch); err != nil { + return fmt.Errorf("decode pending resolved group: %w", err) + } + for _, incoming := range notif.Alerts { + if incoming == nil { + continue + } + duplicate := false + for _, existing := range batch { + if existing != nil && existing.ID == incoming.ID && existing.StartTime.Equal(incoming.StartTime) { + duplicate = true + break + } + } + if !duplicate { + batch = append(batch, incoming) + } + } + links, err := normalizeNotificationLinks(id, notif.DestinationID, notificationLinksForAlerts(batch, notif.DestinationID), notificationDeliveryStateForQueueStatus(QueueStatusPending), nil, nil) + if err != nil { + return err + } + encoded, err := json.Marshal(batch) + if err != nil { + return err + } + encodedLinks, err := json.Marshal(links) + if err != nil { + return err + } + if _, err := nq.db.Exec(`UPDATE notification_queue SET alerts = ?, operational_links = ? WHERE id = ?`, string(encoded), string(encodedLinks), id); err != nil { + return fmt.Errorf("extend pending resolved group: %w", err) + } + return nil +} diff --git a/internal/notifications/resolved_grouping_contract_test.go b/internal/notifications/resolved_grouping_contract_test.go new file mode 100644 index 0000000000..3fd37bd36d --- /dev/null +++ b/internal/notifications/resolved_grouping_contract_test.go @@ -0,0 +1,245 @@ +package notifications + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" +) + +func waitResolvedContract(t *testing.T, predicate func() bool) { + t.Helper() + deadline := time.Now().Add(12 * time.Second) + for !predicate() { + if time.Now().After(deadline) { + t.Fatal("notification condition not reached") + } + time.Sleep(5 * time.Millisecond) + } +} + +// Ordinary firing establishes real receipts. Restart reloads those receipts; +// no receipt or delivery job is manufactured by this test. +func TestResolvedGroupingOrdinaryRestart(t *testing.T) { + for _, stage := range []string{"same_process", "after_firing", "pending_recovery"} { + t.Run(stage, func(t *testing.T) { + type payload struct { + Event string `json:"event"` + Alerts []*alerts.Alert `json:"alerts"` + } + received := make(chan payload, 64) + server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var p payload + if err := json.NewDecoder(r.Body).Decode(&p); err != nil { + t.Error(err) + } + received <- p + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + dir := t.TempDir() + hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL, Enabled: true} + open := func() *NotificationManager { + m := NewNotificationManagerWithDataDir("", dir) + m.webhookClient = server.Client() + if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + m.AddWebhook(hook) + m.SetNotifyOnResolve(true) + m.SetGroupingConfig(true, 1, false, false) + t.Cleanup(m.Stop) + return m + } + m := open() + batch := make([]*alerts.Alert, 15) + for i := range batch { + batch[i] = &alerts.Alert{ID: fmt.Sprintf("burst-%02d", i), ResourceName: fmt.Sprintf("guest-%02d", i), Type: "cpu", Level: alerts.AlertLevelWarning, StartTime: time.Now().Add(-time.Minute)} + m.SendAlert(batch[i]) + } + job := notificationDeliveryJob{Type: "webhook", Event: eventResolved, Alerts: batch, WebhookConfig: &hook} + waitResolvedContract(t, func() bool { + jobs := m.filterResolvedJobsByDeliveryReceipt([]notificationDeliveryJob{job}) + return len(jobs) == 1 && len(jobs[0].Alerts) == 15 + }) + firing := <-received + if len(firing.Alerts) != 15 { + t.Fatalf("firing batch = %d", len(firing.Alerts)) + } + if stage == "after_firing" { + m.Stop() + m = open() + } + // A newly configured destination must not receive recoveries for firings + // delivered only to ops, even when sharing its endpoint. + unseenHook := hook + unseenHook.ID = "never-delivered" + m.AddWebhook(unseenHook) + resolvedAt := time.Now().Truncate(time.Second) + for i, a := range batch { + if m.CancelAlert(a.ID) { + t.Fatal("delivered occurrence cancelled as unannounced") + } + m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: a, ResolvedTime: resolvedAt.Add(time.Duration(i) * time.Second)}) + } + m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: &alerts.Alert{ID: "unannounced", ResourceName: "unannounced", StartTime: time.Now()}, ResolvedTime: resolvedAt}) + if stage == "pending_recovery" { + m.Stop() + m = open() + m.AddWebhook(unseenHook) + } + select { + case p := <-received: + t.Fatalf("resolved delivery bypassed grouping window: event=%s alerts=%d", p.Event, len(p.Alerts)) + case <-time.After(150 * time.Millisecond): + } + var recovery payload + select { + case recovery = <-received: + case <-time.After(5 * time.Second): + t.Fatal("no grouped recovery") + } + if recovery.Event != "resolved" || len(recovery.Alerts) != 15 { + t.Fatalf("recovery event=%s alerts=%d", recovery.Event, len(recovery.Alerts)) + } + for i, a := range recovery.Alerts { + if a.ID != batch[i].ID || a.Metadata[metadataResolvedAt] != resolvedAt.Add(time.Duration(i)*time.Second).Format(time.RFC3339) { + t.Fatalf("lost occurrence/resolution metadata: %+v", a) + } + } + waitResolvedContract(t, func() bool { return len(m.filterResolvedJobsByDeliveryReceipt([]notificationDeliveryJob{job})) == 0 }) + stats, err := m.GetQueue().GetQueueStats() + if err != nil { + t.Fatal(err) + } + if stats["dlq"] != 0 || stats["failed"] != 0 { + t.Fatalf("unexpected delivery failures: %v", stats) + } + m.Stop() + select { + case p := <-received: + t.Fatalf("extra recovery: %+v", p) + default: + } + }) + } +} + +func TestResolvedGroupedWebhookNamesEveryAlert(t *testing.T) { + for _, service := range []string{"generic", "discord", "slack", "telegram", "teams", "teams-adaptive", "pushover", "gotify", "ntfy", "mattermost", "custom"} { + t.Run(service, func(t *testing.T) { + var body string + server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + body = string(b) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + m := NewNotificationManagerWithDataDir("", t.TempDir()) + defer m.Stop() + m.webhookClient = server.Client() + if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL + "?chat_id=1234", Enabled: true, Service: service} + if service == "custom" { + hook.Service = "generic" + hook.Template = `{"message":"{{.Message}}","count":{{.AlertCount}}}` + } + batch := contractGroupedAlerts() + if err := m.sendResolvedWebhook(hook, batch, time.Now()); err != nil { + t.Fatal(err) + } + for _, a := range batch { + if !strings.Contains(body, a.ResourceName) { + t.Errorf("%s missing %s: %s", service, a.ResourceName, body) + } + } + }) + } +} + +// Exercise HTTP payloads without a listener or external network. Persistent +// queue operations remain real; this adapter replaces only provider transport. +type resolvedContractTransport struct{ handler http.Handler } + +func (tr resolvedContractTransport) RoundTrip(r *http.Request) (*http.Response, error) { + recorder := httptest.NewRecorder() + tr.handler.ServeHTTP(recorder, r) + return recorder.Result(), nil +} + +type resolvedContractServer struct { + URL string + client *http.Client +} + +func newResolvedContractServer(handler http.Handler) *resolvedContractServer { + return &resolvedContractServer{URL: "http://127.0.0.1:12345", client: &http.Client{Transport: resolvedContractTransport{handler}}} +} +func (s *resolvedContractServer) Client() *http.Client { return s.client } +func (s *resolvedContractServer) Close() {} + +// Disabled grouping remains immediate. With the same 15-alert workload this +// deliberately hits the unchanged internal 10/minute destination limit; grouped +// recovery above sends one request instead and creates no dead letters. +func TestResolvedGroupingDisabledBurstRetainsRateLimit(t *testing.T) { + server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) + m := NewNotificationManagerWithDataDir("", t.TempDir()) + defer m.Stop() + m.webhookClient = server.Client() + if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL, Enabled: true} + m.AddWebhook(hook) + m.SetNotifyOnResolve(true) + m.SetGroupingConfig(true, 1, false, false) + batch := make([]*alerts.Alert, 15) + for i := range batch { + batch[i] = &alerts.Alert{ID: fmt.Sprintf("limit-%d", i), ResourceName: fmt.Sprintf("limit-%d", i), StartTime: time.Now().Add(-time.Minute), Level: alerts.AlertLevelWarning} + m.SendAlert(batch[i]) + } + waitResolvedContract(t, func() bool { stats, err := m.GetQueue().GetQueueStats(); return err == nil && stats["sent"] == 1 }) + m.SetGroupingConfig(false, 1, false, false) + for _, a := range batch { + m.CancelAlert(a.ID) + m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: a, ResolvedTime: time.Now()}) + } + waitResolvedContract(t, func() bool { + stats, err := m.GetQueue().GetQueueStats() + return err == nil && stats["dlq"] == 6 && stats["sent"] == 10 + }) + dlq, err := m.GetQueue().GetDLQ(20) + if err != nil { + t.Fatal(err) + } + if len(dlq) != 6 { + t.Fatalf("dead letters=%d", len(dlq)) + } + for _, job := range dlq { + if job.Attempts != 3 || (job.LastError == nil || !strings.Contains(*job.LastError, "rate limit exceeded")) { + t.Errorf("wrong retry outcome: %+v", job) + } + } +} + +func TestResolvedGroupingPagerDutyKeepsIndividualKeys(t *testing.T) { + hook := WebhookConfig{ID: "pd", Service: "pagerduty", Enabled: true} + batch := contractGroupedAlerts() + jobs := buildNotificationDeliveryJobs(EmailConfig{}, []WebhookConfig{hook}, AppriseConfig{}, batch, eventResolved, time.Now()) + if len(jobs) != len(batch) { + t.Fatalf("PagerDuty jobs=%d, want %d independent dedup keys", len(jobs), len(batch)) + } + for i, job := range jobs { + if len(job.Alerts) != 1 || job.Alerts[0].ID != batch[i].ID { + t.Fatalf("wrong PagerDuty occurrence: %+v", job) + } + } +}