From 37a8f4a6ffeea75c8e9ab52873887e4cdaa50fdf Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 5 Aug 2026 18:50:50 +0100 Subject: [PATCH] Fix alert notification delivery correctness Fixes #1681 Fixes #1682 Fixes #1683 Contract-Neutral: Notification grouping initialization and alert-config propagation do not alter the broadly referenced agent-lifecycle or storage-recovery contracts; primary alerts, notifications, API, and monitoring contracts and regression proofs are updated. --- .../v6/internal/subsystems/alerts.md | 9 + .../v6/internal/subsystems/api-contracts.md | 4 + .../v6/internal/subsystems/monitoring.md | 3 + .../v6/internal/subsystems/notifications.md | 11 + internal/alerts/alerts_test.go | 58 ++++ internal/alerts/config_runtime.go | 29 +- internal/api/alerts.go | 5 +- internal/monitoring/monitor.go | 5 +- internal/notifications/notifications.go | 253 +++++++++++------- .../notifications_additional_test.go | 86 ++++++ internal/notifications/notifications_test.go | 78 ++++++ internal/notifications/webhook_enhanced.go | 53 +--- 12 files changed, 445 insertions(+), 149 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index d94a956e1..96d6354fd 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -24,6 +24,11 @@ target. The selected initial target owns firing, grouped, and recovery delivery, while every escalation level retains its own independently normalized target. Saving the alert configuration must update the live notification manager as well as persistence so delivery does not differ before and after restart. +Configuration-change reevaluation may resolve metric-backed alerts against +their updated thresholds and may apply explicit resource-disable policies, but +it must not treat provider-owned incidents as missing thresholds. Unrelated +configuration saves preserve those incidents and their acknowledgement state +until their provider evaluator supplies recovery evidence. Docker and Podman container CPU thresholds evaluate host-capacity-normalized CPU percent, not Docker's runtime-native per-core percent. Alert metadata may carry the raw per-core value and reporting host CPU count for evidence, but the @@ -508,6 +513,10 @@ resource type metadata before the legacy node fallback. Host-agent Ceph pool alerts may carry no-colon resource ids with `Instance == Node`, but when metadata or resource type says storage they must keep using storage threshold resolution and source-alias overrides instead of node defaults. +That reevaluation boundary distinguishes metric-backed alert types from +provider-owned incidents. A nil metric threshold may resolve only a known +threshold-backed metric; it is not recovery evidence for resource incidents, +and must not erase their acknowledgement state during a schedule-only save. Browser metric severity colors are also alert-backed. Workloads, Infrastructure, Storage, and the platform-page tables (Docker hosts and diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 03306997b..c1de53031 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -29,6 +29,10 @@ The accepted values are `all`, `email`, `webhook`, and `apprise`; the backend normalizes aliases and invalid values before responding, persisting, and updating the live notification runtime. Escalation levels use the same value vocabulary but remain independent of the initial target. +The same live update applies all four `schedule.grouping` fields atomically. +When grouping is disabled, pending and subsequent alerts are delivered +individually; the API boundary must not reduce that setting to the window or +grouping-key fields while silently ignoring `enabled`. `DELETE /api/ai/patrol/suppressions/finding_{findingID}` is the canonical reopen path for a dismissed Patrol finding. It removes the finding-backed suppression row, preserves the operator note, clears dismissal state in both diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 81baee4ed..bbcb4e8f5 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -24,6 +24,9 @@ initial-delivery target to the tenant notification manager. This is runtime wiring only: monitoring does not choose destinations or own notification policy, and live API saves must apply the same setting without requiring a monitor restart. +Monitor construction also applies the persisted grouping enabled flag, window, +and node/guest keys as one notification-manager policy, so restart behavior is +identical to a live alert-configuration save. Monitoring also owns the distinction between Proxmox VM power state and QEMU guest-agent reachability: fresh or never-healthy VMs with an enabled but unavailable guest agent stay `not-running`, while only VMs with recent healthy diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 56cb185b7..5aac4bd0c 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -23,6 +23,11 @@ Escalation delivery remains independently targetable per level, allowing an Apprise/ntfy first notification to escalate through email, or the reverse. 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 +every alert, while live ntfy firing deliveries and webhook tests share the same +severity-derived title, priority, and tags. ## Canonical Files @@ -258,6 +263,12 @@ cooldown is disabled or still active; scheduled escalation delivery is the explicit exception and must route through the dedicated escalation send path so the alert schedule, not transport cooldown, controls escalation cadence and channel targeting. +The grouping timer is also notification-owned delivery state. Live alert +configuration must apply enabled, window, node, and guest grouping fields as +one policy update. Turning grouping off must stop the timer and deliver every +already-pending alert separately; service templates must be rendered only +after the grouped summary contains every alert, so provider-specific payloads +cannot silently collapse to the first alert. `internal/api/notifications.go` and `frontend-modern/src/api/notifications.ts` are shared boundaries with diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index b2f56357d..6ef5d58bc 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -4568,6 +4568,64 @@ func TestReevaluateClearsDockerContainerAlertWhenOverrideDisabled(t *testing.T) } } +func TestUpdateConfigPreservesAcknowledgedProviderIncident(t *testing.T) { + m := newTestManager(t) + + resourceID := "truenas:system-1:app-container:plex" + alertID := resourceID + "::resource-incident" + ackTime := time.Now().Add(-time.Minute).UTC() + alert := &Alert{ + ID: alertID, + Type: "resource-incident", + Level: AlertLevelCritical, + ResourceID: resourceID, + ResourceName: "plex", + Node: "truenas-1", + Instance: "TrueNAS", + Message: "Container permissions need attention", + StartTime: time.Now().Add(-10 * time.Minute), + LastSeen: time.Now(), + Acknowledged: true, + AckTime: &ackTime, + AckUser: "operator", + Metadata: map[string]interface{}{ + "resourceType": "app-container", + "provider": "truenas", + }, + } + + resolved := make(chan string, 1) + m.SetResolvedCallback(func(id string) { + resolved <- id + }) + + m.mu.Lock() + m.setActiveAlertNoLock(alertID, alert) + m.mu.Unlock() + + config := m.GetConfig() + config.Schedule.Cooldown++ + m.UpdateConfig(config) + + m.mu.RLock() + _, retained := testLookupActiveAlert(t, m, alertID) + retainedAlert := m.activeAlerts[alertID] + m.mu.RUnlock() + + if !retained { + t.Fatalf("provider incident was resolved by an unrelated config save") + } + if retainedAlert == nil || !retainedAlert.Acknowledged || retainedAlert.AckUser != "operator" || retainedAlert.AckTime == nil || !retainedAlert.AckTime.Equal(ackTime) { + t.Fatalf("acknowledgement state was not preserved: %#v", retainedAlert) + } + + select { + case id := <-resolved: + t.Fatalf("resolved callback fired for retained provider incident %q", id) + case <-time.After(50 * time.Millisecond): + } +} + func TestReevaluateClearsDockerContainerAlertWhenIgnoredPrefixAdded(t *testing.T) { m := newTestManager(t) diff --git a/internal/alerts/config_runtime.go b/internal/alerts/config_runtime.go index c45f48752..ec2b06dad 100644 --- a/internal/alerts/config_runtime.go +++ b/internal/alerts/config_runtime.go @@ -150,10 +150,14 @@ func (m *Manager) applyGlobalOfflineSettingsLocked() { if m.config.DisableAllNodesOffline { var nodeAlerts []string for storageKey, alert := range m.activeAlerts { - if alert != nil && alert.CanonicalKind == string(alertspecs.AlertSpecKindConnectivity) { - if resourceType, _ := alert.Metadata["resourceType"].(string); resourceType == "node" { - nodeAlerts = append(nodeAlerts, storageKey) - } + if alert == nil { + continue + } + resourceType, _ := alert.Metadata["resourceType"].(string) + isCanonicalNodeConnectivity := alert.CanonicalKind == string(alertspecs.AlertSpecKindConnectivity) && resourceType == "node" + isLegacyNodeConnectivity := alert.Type == "connectivity" && (strings.HasPrefix(alert.ID, "node-offline-") || strings.HasPrefix(storageKey, "node-offline-")) + if isCanonicalNodeConnectivity || isLegacyNodeConnectivity { + nodeAlerts = append(nodeAlerts, storageKey) } } for _, alertID := range nodeAlerts { @@ -322,6 +326,15 @@ func alertResourceTypeKeysContain(keys []string, target string) bool { return false } +func isConfigReevaluatedMetricType(metricType string) bool { + switch metricType { + case "cpu", "memory", "disk", "diskRead", "diskWrite", "networkIn", "networkOut", "temperature", "usage": + return true + default: + return false + } +} + // reevaluateActiveAlertsLocked re-evaluates all active alerts against the current configuration. // This should only be called with m.mu already locked. func (m *Manager) reevaluateActiveAlertsLocked() { @@ -524,6 +537,14 @@ func (m *Manager) reevaluateActiveAlertsLocked() { threshold = getThresholdForMetric(guestThresholds, metricType) } + // Provider-owned incidents and other non-metric alerts are not threshold + // observations. An unrelated config save must leave their lifecycle and + // acknowledgement state intact; only their evaluator (or an explicit + // resource-disable policy handled above) can resolve them. + if !isConfigReevaluatedMetricType(metricType) { + continue + } + if threshold == nil || threshold.Trigger <= 0 { alertsToResolve = append(alertsToResolve, alertID) continue diff --git a/internal/api/alerts.go b/internal/api/alerts.go index 4dbef8bdc..281c58f31 100644 --- a/internal/api/alerts.go +++ b/internal/api/alerts.go @@ -211,8 +211,9 @@ func (h *AlertHandlers) UpdateAlertConfig(w http.ResponseWriter, r *http.Request notificationMgr := h.getMonitor(r.Context()).GetNotificationManager() notificationMgr.SetEnabled(updatedConfig.Enabled && updatedConfig.ActivationState == alerts.ActivationActive) notificationMgr.SetCooldown(updatedConfig.Schedule.Cooldown) - notificationMgr.SetGroupingWindow(updatedConfig.Schedule.Grouping.Window) - notificationMgr.SetGroupingOptions( + notificationMgr.SetGroupingConfig( + updatedConfig.Schedule.Grouping.Enabled, + updatedConfig.Schedule.Grouping.Window, updatedConfig.Schedule.Grouping.ByNode, updatedConfig.Schedule.Grouping.ByGuest, ) diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index b80f797c5..27b7f66cd 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1739,8 +1739,9 @@ func New(cfg *config.Config) (*Monitor, error) { // Apply schedule settings to notification manager m.notificationMgr.SetEnabled(alertConfig.Enabled && alertConfig.ActivationState == alerts.ActivationActive) m.notificationMgr.SetCooldown(alertConfig.Schedule.Cooldown) - m.notificationMgr.SetGroupingWindow(alertConfig.Schedule.Grouping.Window) - m.notificationMgr.SetGroupingOptions( + m.notificationMgr.SetGroupingConfig( + alertConfig.Schedule.Grouping.Enabled, + alertConfig.Schedule.Grouping.Window, alertConfig.Schedule.Grouping.ByNode, alertConfig.Schedule.Grouping.ByGuest, ) diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index bdd5124f0..d37b81f0f 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -241,6 +241,7 @@ type NotificationManager struct { lastNotified map[string]notificationRecord deliveryReceipts map[string]struct{} groupWindow time.Duration + groupingEnabled bool pendingAlerts []*alerts.Alert groupTimer *time.Timer groupByNode bool @@ -726,6 +727,7 @@ func NewNotificationManagerWithDataDir(publicURL string, dataDir string) *Notifi APIKeyHeader: "X-API-KEY", }, groupWindow: 30 * time.Second, + groupingEnabled: true, tenantID: strings.TrimSpace(os.Getenv("PULSE_TENANT_ID")), tenantName: strings.TrimSpace(os.Getenv("PULSE_TENANT_NAME")), pendingAlerts: make([]*alerts.Alert, 0), @@ -925,6 +927,54 @@ func (n *NotificationManager) SetGroupingWindow(seconds int) { log.Info().Int("seconds", seconds).Msg("updated notification grouping window") } +// SetGroupingConfig atomically applies the complete grouping policy. Disabling +// grouping (or selecting a zero-second window) flushes already-pending alerts +// as individual deliveries so a live config change cannot merge or lose them. +func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNode, byGuest bool) { + if seconds < 0 { + seconds = 0 + } + + n.mu.Lock() + n.groupingEnabled = enabled + n.groupWindow = time.Duration(seconds) * time.Second + n.groupByNode = byNode + n.groupByGuest = byGuest + + var pending []*alerts.Alert + var emailConfig EmailConfig + var webhooks []WebhookConfig + var appriseConfig AppriseConfig + var initialTarget notificationDeliveryTarget + var queue *NotificationQueue + if !enabled || seconds == 0 { + pending = append(pending, n.pendingAlerts...) + n.pendingAlerts = n.pendingAlerts[:0] + if n.groupTimer != nil { + n.groupTimer.Stop() + n.groupTimer = nil + } + emailConfig = copyEmailConfig(n.emailConfig) + webhooks = copyWebhookConfigs(n.webhooks) + appriseConfig = copyAppriseConfig(n.appriseConfig) + initialTarget = n.initialTarget + queue = n.queue + } + n.mu.Unlock() + + for _, alert := range pending { + n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, []*alerts.Alert{alert}, initialTarget, queue) + } + + log.Info(). + Bool("enabled", enabled). + Int("seconds", seconds). + Bool("byNode", byNode). + Bool("byGuest", byGuest). + Int("flushedIndividually", len(pending)). + Msg("updated notification grouping configuration") +} + // SetGroupingOptions updates grouping options func (n *NotificationManager) SetGroupingOptions(byNode, byGuest bool) { n.mu.Lock() @@ -1109,32 +1159,14 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp return } - if options.immediate { + if options.immediate || !n.groupingEnabled || n.groupWindow <= 0 { emailConfig := copyEmailConfig(n.emailConfig) webhooks := copyWebhookConfigs(n.webhooks) appriseConfig := copyAppriseConfig(n.appriseConfig) queue := n.queue n.mu.Unlock() - alertsToSend := []*alerts.Alert{alert} - jobs := buildNotificationDeliveryJobsForTarget( - emailConfig, - webhooks, - appriseConfig, - alertsToSend, - eventAlert, - time.Time{}, - options.target, - ) - if len(jobs) == 0 { - n.markAlertsNotified(alertsToSend, time.Now()) - return - } - if queue != nil { - n.enqueueNotificationJobs(queue, jobs) - } else { - n.dispatchNotificationJobsAsync(jobs) - } + n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, []*alerts.Alert{alert}, options.target, queue) return } @@ -1180,6 +1212,34 @@ func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, s n.mu.Unlock() } +func (n *NotificationManager) dispatchFiringAlerts( + emailConfig EmailConfig, + webhooks []WebhookConfig, + appriseConfig AppriseConfig, + alertsToSend []*alerts.Alert, + target notificationDeliveryTarget, + queue *NotificationQueue, +) { + jobs := buildNotificationDeliveryJobsForTarget( + emailConfig, + webhooks, + appriseConfig, + alertsToSend, + eventAlert, + time.Time{}, + target, + ) + if len(jobs) == 0 { + n.markAlertsNotified(alertsToSend, time.Now()) + return + } + if queue != nil { + n.enqueueNotificationJobs(queue, jobs) + return + } + n.dispatchNotificationJobsAsync(jobs) +} + func notificationDeliveryDestinationKey(job notificationDeliveryJob) string { var identity string switch job.Type { @@ -1467,30 +1527,7 @@ func (n *NotificationManager) sendGroupedAlerts() { queue := n.queue n.mu.Unlock() - jobs := buildNotificationDeliveryJobsForTarget( - emailConfig, - webhooks, - appriseConfig, - alertsToSend, - eventAlert, - time.Time{}, - initialTarget, - ) - if len(jobs) == 0 { - // Preserve cooldown semantics when notifications are globally enabled - // but no destination is configured. Delivery receipts remain empty, so - // a later recovery is still correctly suppressed. - n.markAlertsNotified(alertsToSend, time.Now()) - return - } - - // Use persistent queue if available, otherwise send directly - if queue != nil { - n.enqueueNotificationJobs(queue, jobs) - // Note: Cooldown will be marked after successful dequeue and send - } else { - n.dispatchNotificationJobsAsync(jobs) - } + n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, alertsToSend, initialTarget, queue) } func buildNotificationDeliveryJobs(emailConfig EmailConfig, webhooks []WebhookConfig, appriseConfig AppriseConfig, alertsToSend []*alerts.Alert, event notificationEvent, resolvedAt time.Time) []notificationDeliveryJob { @@ -2413,6 +2450,55 @@ func (n *NotificationManager) renderWebhookPayloadJSON(webhook WebhookConfig, da return fallback() } +func withNtfyAlertHeaders(webhook WebhookConfig, alertList []*alerts.Alert) WebhookConfig { + webhook = copyWebhookConfig(webhook) + if webhook.Headers == nil { + webhook.Headers = make(map[string]string) + } + + level := alerts.AlertLevelWarning + var primary *alerts.Alert + for _, alert := range alertList { + if alert == nil { + continue + } + if primary == nil { + primary = alert + } + if alert.Level == alerts.AlertLevelCritical { + level = alerts.AlertLevelCritical + break + } + } + + levelLabel := "WARNING" + priority := "high" + severityTag := "warning" + if level == alerts.AlertLevelCritical { + levelLabel = "CRITICAL" + priority = "urgent" + severityTag = "rotating_light" + } + + titleSubject := fmt.Sprintf("%d alerts", len(alertList)) + typeTag := "grouped" + if len(alertList) == 1 && primary != nil { + titleSubject = primary.ResourceName + if primary.Node != "" { + titleSubject += " on " + primary.Node + } + if primary.Type != "" { + typeTag = primary.Type + } + } + + webhook.Headers["Content-Type"] = "text/plain" + webhook.Headers["Title"] = fmt.Sprintf("%s: %s", levelLabel, titleSubject) + webhook.Headers["Priority"] = priority + webhook.Headers["Tags"] = fmt.Sprintf("%s,pulse,%s", severityTag, typeTag) + return webhook +} + // sendGroupedWebhook sends a grouped webhook notification func (n *NotificationManager) sendGroupedWebhook(webhook WebhookConfig, alertList []*alerts.Alert) error { if len(alertList) == 0 { @@ -2424,57 +2510,44 @@ func (n *NotificationManager) sendGroupedWebhook(webhook WebhookConfig, alertLis originalPrimary := alertList[0] alertCopy := *originalPrimary primaryAlert := &alertCopy + usesRenderedAlert := strings.TrimSpace(webhook.Template) != "" || (webhook.Service != "" && webhook.Service != "generic") + if usesRenderedAlert && len(alertList) > 1 { + otherAlerts := make([]string, 0, len(alertList)-1) + for i := 1; i < len(alertList); i++ { + if alertList[i] == nil { + continue + } + alertLabel := alertList[i].ResourceName + if alertList[i].Node != "" { + alertLabel += " on " + alertList[i].Node + } + if alertList[i].Message != "" { + otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %s", alertLabel, alertList[i].Message)) + } else { + otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %.1f%%", alertLabel, alertList[i].Value)) + } + } + if len(otherAlerts) > 0 { + if webhook.Service == "discord" { + primaryAlert.Message = fmt.Sprintf("%s | %d alerts: %s", primaryAlert.Message, len(alertList), strings.Join(otherAlerts, ", ")) + } else if strings.TrimSpace(webhook.Template) != "" { + primaryAlert.Message = fmt.Sprintf("%s\\n\\nAll %d alerts:\\n%s", primaryAlert.Message, len(alertList), strings.Join(otherAlerts, "\\n")) + } else { + primaryAlert.Message = fmt.Sprintf("%s\n\nAll %d alerts:\n%s", primaryAlert.Message, len(alertList), strings.Join(otherAlerts, "\n")) + } + } + } + + // Prepare template data only after enriching the primary alert. Preparing it + // first captured the original message and silently omitted grouped alerts + // from built-in service payloads. customFields := convertWebhookCustomFields(webhook.CustomFields) data := n.prepareWebhookData(primaryAlert, customFields) data.AlertCount = len(alertList) data.Alerts = alertList data.Mention = webhook.Mention - - // Check if webhook has a custom template first - // Only use custom template if it's not empty - if webhook.Template != "" && strings.TrimSpace(webhook.Template) != "" && len(alertList) > 0 { - // Use custom template with enhanced message for grouped alerts - alert := primaryAlert - if len(alertList) > 1 { - // Build a full list of all alerts - summary := alert.Message - otherAlerts := []string{} - for i := 1; i < len(alertList); i++ { // Show ALL alerts - otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %.1f%%", alertList[i].ResourceName, alertList[i].Value)) - } - if len(otherAlerts) > 0 { - // For custom templates, we need to escape newlines since they're likely - // used in shell commands or other contexts that need escaping - alert.Message = fmt.Sprintf("%s\\n\\nAll %d alerts:\\n%s", summary, len(alertList), strings.Join(otherAlerts, "\\n")) - } - } - } - - if webhook.Service != "" && webhook.Service != "generic" && len(alertList) > 0 { - // For service-specific webhooks, use the first alert with a note about others - // For simplicity, send the first alert with a note about others - // Most webhook services work better with single structured payloads - alert := primaryAlert - - // Modify message if multiple alerts - but format differently for Discord - if len(alertList) > 1 { - summary := alert.Message - otherAlerts := []string{} - for i := 1; i < len(alertList); i++ { - otherAlerts = append(otherAlerts, fmt.Sprintf("• %s: %.1f%%", alertList[i].ResourceName, alertList[i].Value)) - } - if len(otherAlerts) > 0 { - // For Discord, format as a single line list to avoid newline issues - // Discord embeds don't render \n in description anyway - if webhook.Service == "discord" { - // Use comma-separated list for Discord - alert.Message = fmt.Sprintf("%s | %d alerts: %s", summary, len(alertList), strings.Join(otherAlerts, ", ")) - } else { - // For other services, escape newlines properly - alert.Message = fmt.Sprintf("%s\\n\\nAll %d alerts:\\n%s", summary, len(alertList), strings.Join(otherAlerts, "\\n")) - } - } - } + if webhook.Service == "ntfy" { + webhook = withNtfyAlertHeaders(webhook, alertList) } var err error diff --git a/internal/notifications/notifications_additional_test.go b/internal/notifications/notifications_additional_test.go index f55815d14..4a2b08bf7 100644 --- a/internal/notifications/notifications_additional_test.go +++ b/internal/notifications/notifications_additional_test.go @@ -129,6 +129,92 @@ func TestSendGroupedWebhookGeneric(t *testing.T) { } } +func TestSendGroupedWebhookNtfyIncludesLiveMetadataAndEveryAlert(t *testing.T) { + var gotTitle string + var gotPriority string + var gotTags string + var gotAuthorization string + var gotBody string + + server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotTitle = r.Header.Get("Title") + gotPriority = r.Header.Get("Priority") + gotTags = r.Header.Get("Tags") + gotAuthorization = r.Header.Get("Authorization") + body, _ := io.ReadAll(r.Body) + gotBody = string(body) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + manager := NewNotificationManager("https://pulse.example") + defer manager.Stop() + manager.webhookClient = server.Client() + if err := manager.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatalf("allowlist: %v", err) + } + + alertList := []*alerts.Alert{ + { + ID: "alert-critical", + Type: "cpu", + Level: alerts.AlertLevelCritical, + ResourceID: "vm-101", + ResourceName: "database", + Node: "node-a", + Message: "CPU is critical", + Value: 99, + Threshold: 90, + StartTime: time.Now().Add(-2 * time.Minute), + }, + { + ID: "alert-warning", + Type: "memory", + Level: alerts.AlertLevelWarning, + ResourceID: "vm-102", + ResourceName: "cache", + Node: "node-b", + Message: "Memory is high", + Value: 88, + Threshold: 80, + StartTime: time.Now().Add(-time.Minute), + }, + } + + webhook := WebhookConfig{ + Name: "ntfy-live", + URL: server.URL + "/topic", + Enabled: true, + Service: "ntfy", + Headers: map[string]string{"Authorization": "Bearer secret"}, + } + + if err := manager.sendGroupedWebhook(webhook, alertList); err != nil { + t.Fatalf("sendGroupedWebhook error: %v", err) + } + + if gotTitle != "CRITICAL: 2 alerts" { + t.Fatalf("Title = %q, want %q", gotTitle, "CRITICAL: 2 alerts") + } + if gotPriority != "urgent" { + t.Fatalf("Priority = %q, want urgent", gotPriority) + } + if !strings.Contains(gotTags, "rotating_light") || !strings.Contains(gotTags, "pulse") { + t.Fatalf("Tags = %q, want critical Pulse tags", gotTags) + } + if gotAuthorization != "Bearer secret" { + t.Fatalf("Authorization = %q, want configured header preserved", gotAuthorization) + } + for _, resourceName := range []string{"database", "cache"} { + if !strings.Contains(gotBody, resourceName) { + t.Fatalf("ntfy body %q does not include alert resource %q", gotBody, resourceName) + } + } + if !strings.Contains(gotBody, "Memory is high") { + t.Fatalf("ntfy body %q does not include the second alert message", gotBody) + } +} + func TestSendGroupedWebhookDiscordEscapesSpecialCharacters(t *testing.T) { var gotBody []byte diff --git a/internal/notifications/notifications_test.go b/internal/notifications/notifications_test.go index fc078c5ef..83acc15c1 100644 --- a/internal/notifications/notifications_test.go +++ b/internal/notifications/notifications_test.go @@ -237,6 +237,84 @@ func TestSetGroupingWindowClampsNegativeValues(t *testing.T) { nm.mu.RUnlock() } +func TestGroupingDisabledFlushesPendingAndDeliversIndividually(t *testing.T) { + t.Setenv("PULSE_DATA_DIR", t.TempDir()) + + originalSpawn := spawnAsync + spawnAsync = func(f func()) { f() } + t.Cleanup(func() { spawnAsync = originalSpawn }) + + var requestMu sync.Mutex + requestCount := 0 + server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestMu.Lock() + requestCount++ + requestMu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + nm := NewNotificationManager("") + defer nm.Stop() + nm.webhookClient = server.Client() + if err := nm.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatalf("allowlist: %v", err) + } + nm.mu.Lock() + if nm.queue != nil { + _ = nm.queue.Stop() + nm.queue = nil + } + nm.mu.Unlock() + nm.AddWebhook(WebhookConfig{Name: "capture", URL: server.URL, Enabled: true}) + nm.SetGroupingConfig(true, 3600, true, false) + + newAlert := func(id string) *alerts.Alert { + return &alerts.Alert{ + ID: id, + Type: "cpu", + Level: alerts.AlertLevelWarning, + ResourceID: id, + ResourceName: id, + Message: "CPU is high", + Value: 90, + Threshold: 80, + StartTime: time.Now(), + } + } + + nm.SendAlert(newAlert("vm-1")) + nm.SendAlert(newAlert("vm-2")) + requestMu.Lock() + if requestCount != 0 { + requestMu.Unlock() + t.Fatalf("grouped alerts delivered before the grouping window elapsed: %d", requestCount) + } + requestMu.Unlock() + + nm.SetGroupingConfig(false, 30, true, false) + requestMu.Lock() + if requestCount != 2 { + requestMu.Unlock() + t.Fatalf("disabling grouping delivered %d requests, want 2 individual requests", requestCount) + } + requestMu.Unlock() + + nm.mu.RLock() + pendingCount := len(nm.pendingAlerts) + nm.mu.RUnlock() + if pendingCount != 0 { + t.Fatalf("pending alerts after disabling grouping = %d, want 0", pendingCount) + } + + nm.SendAlert(newAlert("vm-3")) + requestMu.Lock() + defer requestMu.Unlock() + if requestCount != 3 { + t.Fatalf("alert sent with grouping disabled produced %d total requests, want 3", requestCount) + } +} + func TestSendGroupedAppriseInvokesExecutor(t *testing.T) { t.Setenv("PULSE_DATA_DIR", t.TempDir()) nm := NewNotificationManager("") diff --git a/internal/notifications/webhook_enhanced.go b/internal/notifications/webhook_enhanced.go index 47285b5b7..a41e897d0 100644 --- a/internal/notifications/webhook_enhanced.go +++ b/internal/notifications/webhook_enhanced.go @@ -568,57 +568,8 @@ func (n *NotificationManager) TestEnhancedWebhook(webhook EnhancedWebhookConfig) } if webhook.Service == "ntfy" { - headers := make(map[string]string, len(webhook.Headers)+4) - for key, value := range webhook.Headers { - if !strings.Contains(value, "{{") { - headers[key] = value - } - } - webhook.Headers = headers - webhook.Headers["Content-Type"] = "text/plain" - - // Set dynamic headers based on alert level - title := fmt.Sprintf("%s: %s", - func() string { - switch testAlert.Level { - case alerts.AlertLevelCritical: - return "CRITICAL" - case alerts.AlertLevelWarning: - return "WARNING" - default: - return "INFO" - } - }(), - testAlert.ResourceName, - ) - webhook.Headers["Title"] = title - - priority := func() string { - switch testAlert.Level { - case alerts.AlertLevelCritical: - return "urgent" - case alerts.AlertLevelWarning: - return "high" - default: - return "default" - } - }() - webhook.Headers["Priority"] = priority - - tags := fmt.Sprintf("%s,pulse,%s", - func() string { - switch testAlert.Level { - case alerts.AlertLevelCritical: - return "rotating_light" - case alerts.AlertLevelWarning: - return "warning" - default: - return "white_check_mark" - } - }(), - testAlert.Type, - ) - webhook.Headers["Tags"] = tags + canonical := withNtfyAlertHeaders(canonicalWebhookConfigForEnhanced(webhook), []*alerts.Alert{testAlert}) + webhook.Headers = canonical.Headers } result, err := n.executeEnhancedWebhookRequest(webhook, payload, WebhookTestTimeout, "Pulse-Monitoring/2.0 (Test)", webhookEventID(testAlert.ID, "alert"))