From 1836e2a3cc65fdcd06459d67d2efe14aeabd4c15 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 13 May 2026 12:07:14 +0100 Subject: [PATCH] Fix mixed quiet-hours notification replay queueing --- internal/notifications/notifications.go | 149 +++++++++++++------ internal/notifications/notifications_test.go | 100 +++++++++++++ 2 files changed, 202 insertions(+), 47 deletions(-) diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index a0ea9874d..c2d75cb46 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -73,6 +73,11 @@ type notificationDeliveryJob struct { AppriseConfig *AppriseConfig } +type notificationQueueBucket struct { + job notificationDeliveryJob + nextRetryAt *time.Time +} + // createSecureWebhookClient creates an HTTP client with security controls func (n *NotificationManager) createSecureWebhookClient(timeout time.Duration) *http.Client { return n.createSecureWebhookClientWithTLS(timeout, false) @@ -342,35 +347,50 @@ func annotateResolvedMetadata(alert *alerts.Alert, resolvedAt time.Time) { alert.Metadata[metadataResolvedAt] = resolvedAt.Format(time.RFC3339) } +func quietHoursReplayAtForAlert(alert *alerts.Alert, now time.Time) *time.Time { + if alert == nil || alert.Metadata == nil { + return nil + } + raw, ok := alert.Metadata[alerts.MetadataQuietHoursReplayAt] + if !ok { + return nil + } + + var parsed time.Time + switch value := raw.(type) { + case string: + ts, err := time.Parse(time.RFC3339, strings.TrimSpace(value)) + if err != nil { + return nil + } + parsed = ts + case time.Time: + parsed = value + case float64: + if value <= 0 { + return nil + } + parsed = time.Unix(int64(value), 0) + default: + return nil + } + + if !parsed.After(now) { + return nil + } + parsed = parsed.UTC() + return &parsed +} + func quietHoursReplayAtForAlerts(alertList []*alerts.Alert, now time.Time) *time.Time { var replayAt time.Time for _, alert := range alertList { - if alert == nil || alert.Metadata == nil { + alertReplayAt := quietHoursReplayAtForAlert(alert, now) + if alertReplayAt == nil { continue } - raw, ok := alert.Metadata[alerts.MetadataQuietHoursReplayAt] - if !ok { - continue - } - - var parsed time.Time - switch value := raw.(type) { - case string: - ts, err := time.Parse(time.RFC3339, strings.TrimSpace(value)) - if err != nil { - continue - } - parsed = ts - case time.Time: - parsed = value - case float64: - if value <= 0 { - continue - } - parsed = time.Unix(int64(value), 0) - } - if parsed.After(now) && (replayAt.IsZero() || parsed.After(replayAt)) { - replayAt = parsed + if replayAt.IsZero() || alertReplayAt.After(replayAt) { + replayAt = *alertReplayAt } } @@ -381,6 +401,39 @@ func quietHoursReplayAtForAlerts(alertList []*alerts.Alert, now time.Time) *time return &replayAt } +func notificationQueueBucketsForJob(job notificationDeliveryJob, now time.Time) []notificationQueueBucket { + if len(job.Alerts) == 0 { + return []notificationQueueBucket{{job: job}} + } + + buckets := make([]notificationQueueBucket, 0, len(job.Alerts)) + bucketIndexes := make(map[string]int, len(job.Alerts)) + for _, alert := range job.Alerts { + nextRetryAt := quietHoursReplayAtForAlert(alert, now) + key := "immediate" + if nextRetryAt != nil { + key = nextRetryAt.Format(time.RFC3339Nano) + } + + index, exists := bucketIndexes[key] + if !exists { + bucketJob := job + bucketJob.Alerts = nil + bucket := notificationQueueBucket{job: bucketJob} + if nextRetryAt != nil { + replayAt := *nextRetryAt + bucket.nextRetryAt = &replayAt + } + index = len(buckets) + bucketIndexes[key] = index + buckets = append(buckets, bucket) + } + buckets[index].job.Alerts = append(buckets[index].job.Alerts, alert) + } + + return buckets +} + // NormalizeAppriseConfig cleans and normalizes Apprise configuration values. func NormalizeAppriseConfig(cfg AppriseConfig) AppriseConfig { normalized := cfg @@ -1182,31 +1235,33 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue, continue } - notif := &QueuedNotification{ - Type: queueTypeForNotificationDeliveryJob(job), - Alerts: job.Alerts, - Config: configJSON, - MaxAttempts: 3, - NextRetryAt: quietHoursReplayAtForAlerts(job.Alerts, time.Now()), - } - if err := queue.Enqueue(notif); err != nil { - anyFailed = true - n.logNotificationJobError(job, err, "failed to enqueue notification - falling back to direct send") - n.dispatchNotificationJobAsync(job, "failed to send notification after queue enqueue failure") - continue - } + for _, bucket := range notificationQueueBucketsForJob(job, time.Now()) { + notif := &QueuedNotification{ + Type: queueTypeForNotificationDeliveryJob(bucket.job), + Alerts: bucket.job.Alerts, + Config: configJSON, + MaxAttempts: 3, + NextRetryAt: bucket.nextRetryAt, + } + if err := queue.Enqueue(notif); 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") + continue + } - logger := log.Debug(). - Str("type", job.Type). - Str("event", string(job.Event)). - Int("alertCount", len(job.Alerts)) - if notif.NextRetryAt != nil { - logger = logger.Time("nextRetryAt", *notif.NextRetryAt) + logger := log.Debug(). + Str("type", bucket.job.Type). + Str("event", string(bucket.job.Event)). + Int("alertCount", len(bucket.job.Alerts)) + if notif.NextRetryAt != nil { + logger = logger.Time("nextRetryAt", *notif.NextRetryAt) + } + if bucket.job.WebhookConfig != nil { + logger = logger.Str("webhookName", bucket.job.WebhookConfig.Name) + } + logger.Msg("enqueued notification delivery job") } - if job.WebhookConfig != nil { - logger = logger.Str("webhookName", job.WebhookConfig.Name) - } - logger.Msg("enqueued notification delivery job") } if anyFailed && len(jobs) > 0 && jobs[0].Event == eventResolved { diff --git a/internal/notifications/notifications_test.go b/internal/notifications/notifications_test.go index 57eb130db..5beffd925 100644 --- a/internal/notifications/notifications_test.go +++ b/internal/notifications/notifications_test.go @@ -2,6 +2,7 @@ package notifications import ( "context" + "database/sql" "encoding/json" "io" "net/http" @@ -3087,6 +3088,105 @@ func TestProcessQueuedNotification_QuietHoursReplayUsesQueueRetryTime(t *testing } } +func TestEnqueueNotificationJobs_MixedQuietHoursGroupKeepsImmediateAlertsReady(t *testing.T) { + queue, err := NewNotificationQueue(t.TempDir()) + if err != nil { + t.Fatalf("failed to create queue: %v", err) + } + defer func() { _ = queue.Stop() }() + + replayAt := time.Now().Add(time.Hour).UTC() + immediate := &alerts.Alert{ + ID: "immediate-critical", + Type: "cpu", + Level: alerts.AlertLevelCritical, + ResourceName: "node-1", + Message: "critical alert should send now", + StartTime: time.Now().Add(-time.Minute), + } + deferred := &alerts.Alert{ + ID: "quiet-hours-warning", + Type: "memory", + Level: alerts.AlertLevelWarning, + ResourceName: "node-2", + Message: "warning alert should wait", + StartTime: time.Now().Add(-time.Minute), + Metadata: map[string]interface{}{ + alerts.MetadataQuietHoursSuppressed: true, + alerts.MetadataQuietHoursSuppressionReason: "non-critical", + alerts.MetadataQuietHoursReplayAt: replayAt.Format(time.RFC3339), + }, + } + jobs := buildNotificationDeliveryJobs( + EmailConfig{Enabled: true, SMTPHost: "smtp.example.com", SMTPPort: 587, From: "alerts@example.com", To: []string{"ops@example.com"}}, + nil, + AppriseConfig{}, + []*alerts.Alert{immediate, deferred}, + eventAlert, + time.Time{}, + ) + + nm := &NotificationManager{} + if failed := nm.enqueueNotificationJobs(queue, jobs); failed { + t.Fatal("did not expect mixed quiet-hours enqueue to fall back") + } + + pending, err := queue.GetPending(10) + if err != nil { + t.Fatalf("GetPending failed: %v", err) + } + if len(pending) != 1 { + t.Fatalf("expected one immediately pending notification, got %d", len(pending)) + } + if pending[0].NextRetryAt != nil { + t.Fatalf("immediate notification next retry = %v, want nil", pending[0].NextRetryAt) + } + if len(pending[0].Alerts) != 1 || pending[0].Alerts[0].ID != immediate.ID { + t.Fatalf("pending alerts = %#v, want only %q", pending[0].Alerts, immediate.ID) + } + + rows, err := queue.db.Query(`SELECT alerts, next_retry_at FROM notification_queue WHERE type = ? ORDER BY next_retry_at IS NOT NULL, next_retry_at`, "email") + if err != nil { + t.Fatalf("failed to read queued notifications: %v", err) + } + defer rows.Close() + + type queuedRow struct { + alerts []*alerts.Alert + nextRetryAt sql.NullInt64 + } + var queued []queuedRow + for rows.Next() { + var row queuedRow + var alertsJSON string + if err := rows.Scan(&alertsJSON, &row.nextRetryAt); err != nil { + t.Fatalf("failed to scan queued notification: %v", err) + } + if err := json.Unmarshal([]byte(alertsJSON), &row.alerts); err != nil { + t.Fatalf("failed to unmarshal queued alerts: %v", err) + } + queued = append(queued, row) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate queued notifications: %v", err) + } + if len(queued) != 2 { + t.Fatalf("queued rows = %d, want 2", len(queued)) + } + if queued[0].nextRetryAt.Valid { + t.Fatalf("immediate row next_retry_at = %d, want NULL", queued[0].nextRetryAt.Int64) + } + if len(queued[0].alerts) != 1 || queued[0].alerts[0].ID != immediate.ID { + t.Fatalf("immediate row alerts = %#v, want only %q", queued[0].alerts, immediate.ID) + } + if !queued[1].nextRetryAt.Valid || queued[1].nextRetryAt.Int64 != replayAt.Unix() { + t.Fatalf("deferred row next_retry_at = %#v, want %d", queued[1].nextRetryAt, replayAt.Unix()) + } + if len(queued[1].alerts) != 1 || queued[1].alerts[0].ID != deferred.ID { + t.Fatalf("deferred row alerts = %#v, want only %q", queued[1].alerts, deferred.ID) + } +} + func TestGetQueue_NilQueue(t *testing.T) { nm := &NotificationManager{}