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) + } + } +}