diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 22156b30f7..e4f103a1c8 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -433,10 +433,18 @@ pending backlog through the same canonical batch path instead of relying on a separate direct-send shortcut or waiting for an unrelated timer tick. Alert delivery cooldown is also owned at this boundary. Normal alert delivery must suppress duplicate sends for the same active alert occurrence when -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. +cooldown is disabled or still active. A manager-admitted increase above the +highest delivered severity for that occurrence is new information and must pass +this gate, while retaining normal destination selection and grouping. Late +lower-severity delivery receipts must not lower that high-water mark and permit +repeat critical sends. A new occurrence has its own severity history. This +exception does not bypass manager acknowledgement, snooze, flapping or hourly +budget admission, nor destination routing or rate limits. Scheduled escalation +retains its dedicated send path so the alert schedule, not transport cooldown, +controls escalation cadence and channel targeting. +`TestMonitorSeverityDelivery` covers observation-to-HTTP delivery and the +persisted delivery log for custom sensors, CPU metrics and ZFS health; callback +success alone is not delivery evidence. 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 diff --git a/internal/monitoring/monitor_severity_delivery_test.go b/internal/monitoring/monitor_severity_delivery_test.go new file mode 100644 index 0000000000..0b1650b33f --- /dev/null +++ b/internal/monitoring/monitor_severity_delivery_test.go @@ -0,0 +1,195 @@ +package monitoring + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/notifications" + "github.com/stretchr/testify/require" +) + +// Issue 1801: exercise actual observation, manager admission, monitor callback, +// persistent queue, HTTP destination and delivery history, not just callbacks. +func TestMonitorSeverityDelivery(t *testing.T) { + for _, tc := range []struct { + cooldown, group int + policy string + }{ + {0, 0, "ready"}, {30, 0, "ready"}, {30, 1, "ready"}, + {30, 0, "acknowledged"}, {30, 0, "snoozed"}, {30, 0, "rate-limited"}, + {30, 0, "routing"}, {30, 0, "metric"}, {30, 0, "health"}, + } { + t.Run(fmt.Sprintf("%s/cooldown%d/group%d", tc.policy, tc.cooldown, tc.group), func(t *testing.T) { + requests := make(chan string, 20) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var payload struct { + Level string `json:"level"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Error(err) + } + requests <- payload.Level + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + nm := notifications.NewNotificationManagerWithDataDir("", t.TempDir()) + defer nm.Stop() + require.NotNil(t, nm.GetQueue()) + nm.SetCooldown(tc.cooldown) + nm.SetGroupingWindow(tc.group) + require.NoError(t, nm.UpdateAllowedPrivateCIDRs("127.0.0.1/32")) + nm.AddWebhook(notifications.WebhookConfig{ID: "hook", Name: "hook", URL: server.URL, Enabled: true, Template: `{"level":"{{.Level}}"}`}) + mgr := alerts.NewManagerWithDataDir(t.TempDir()) + defer mgr.Stop() + cfg := mgr.GetConfig() + cfg.Enabled = true + cfg.ActivationState = alerts.ActivationActive + cfg.FlappingEnabled = false + cfg.Schedule.MaxAlertsHour = 100 + // Disable fixture activation delay explicitly; absent entries default to five seconds. + cfg.TimeThresholds = map[string]int{"agent": 0, "storage": 0} + cfg.SuppressionWindow = 0 + cfg.MetricEvaluationWindows = map[string]map[string]int{"agent": {"cpu": 0}} + if tc.policy == "metric" { + cfg.AgentDefaults.CPU = &alerts.HysteresisThreshold{Trigger: 80, Clear: 75} + } + if tc.policy == "rate-limited" { + cfg.Schedule.MaxAlertsHour = 1 + } + mgr.UpdateConfig(cfg) + monitor := &Monitor{alertManager: mgr, notificationMgr: nm} + mgr.SetAlertCallback(monitor.handleAlertFired) + host := models.Host{ID: "sensor-host", Hostname: "sensor-host", Sensors: models.HostSensorSummary{ + Custom: []models.HostCustomSensorMetric{{ID: "probe", Name: "Probe", Status: "warning", ObservedAt: time.Now()}}, + }} + alertType := "custom-sensor" + storage := models.Storage{ID: "pool-store", Name: "tank", Status: "online", ZFSPool: &models.ZFSPool{Name: "tank", State: "DEGRADED"}} + if tc.policy == "metric" { + host.Sensors.Custom = nil + host.CPUUsage = 85 + alertType = "cpu" + } + if tc.policy == "health" { + alertType = "zfs-pool-state" + } + observe := func(critical bool) { + if tc.policy == "health" { + if critical { + storage.ZFSPool.State = "FAULTED" + } else { + storage.ZFSPool.State = "DEGRADED" + } + mgr.CheckStorage(storage) + } else { + if tc.policy == "metric" { + if critical { + host.CPUUsage = 95 + } else { + host.CPUUsage = 85 + } + } else { + if critical { + host.Sensors.Custom[0].Status = "critical" + } else { + host.Sensors.Custom[0].Status = "warning" + } + } + mgr.CheckHost(host) + } + } + ready := tc.policy == "ready" || tc.policy == "metric" || tc.policy == "health" + await := func(level string, sent int) { + t.Helper() + select { + case got := <-requests: + require.Equal(t, level, got) + case <-time.After(4 * time.Second): + t.Fatalf("missing %s destination delivery; history=%+v", level, deliveryHistory(t, nm)) + } + // Wait for successful queue completion (which follows durable receipts and + // the cooldown marker), not merely arrival at the HTTP handler. + require.Eventually(t, func() bool { + stats, err := nm.GetQueueStats() + return err == nil && stats["sent"] == sent && len(deliveryHistory(t, nm)) == sent + }, 3*time.Second, 10*time.Millisecond) + history := deliveryHistory(t, nm) + require.Len(t, history, sent) + for _, row := range history { + require.True(t, row.Success) + require.Equal(t, notifications.DeliveryOutcomeSent, row.Outcome) + } + } + observe(false) + await("warning", 1) + var original alerts.Alert + for _, a := range mgr.GetActiveAlerts() { + if a.Type == alertType { + original = a + } + } + require.NotEmpty(t, original.ID) + switch tc.policy { + case "acknowledged": + require.NoError(t, mgr.AcknowledgeAlert(original.ID, "test")) + case "snoozed": + require.NoError(t, mgr.SnoozeAlert(original.ID, "test", time.Now().Add(time.Hour))) + case "routing": + nm.SetInitialNotifyTarget("email") // no email configured; must not widen to webhook + } + observe(false) + observe(true) + want := 1 + if ready { + await("critical", 2) + want = 2 + } + found := false + for _, a := range mgr.GetActiveAlerts() { + if a.ID == original.ID { + found = true + require.Equal(t, alerts.AlertLevelCritical, a.Level) + require.Equal(t, original.StartTime, a.StartTime) + } + } + require.True(t, found, "same occurrence must remain active") + observe(true) + observe(false) + // Also probe notification-layer repeat/downgrade suppression directly once + // the ready path has completed both sends. + if ready { + nm.SendAlert(&original) + critical := original.Clone() + critical.Level = alerts.AlertLevelCritical + nm.SendAlert(critical) + } + select { + case extra := <-requests: + t.Fatalf("unexpected repeat/suppressed delivery: %s", extra) + case <-time.After(1200 * time.Millisecond): + } + history := deliveryHistory(t, nm) + require.Len(t, history, want) + for _, row := range history { + require.Equal(t, []string{original.ID}, row.AlertIDs) + require.Equal(t, "webhook:hook", row.DestinationID) + require.Equal(t, 1, row.Attempts) + } + dlq, err := nm.GetQueue().GetDLQ(100) + require.NoError(t, err) + require.Empty(t, dlq) + }) + } +} + +func deliveryHistory(t *testing.T, nm *notifications.NotificationManager) []notifications.DeliveryLogEntry { + t.Helper() + rows, err := nm.GetDeliveryLog(time.Time{}, 100) + require.NoError(t, err) + return rows +} diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 8bfb445a17..032a3e6d22 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -550,6 +550,7 @@ func defaultAppriseExec(ctx context.Context, args []string) ([]byte, error) { type notificationRecord struct { lastSent time.Time alertStart time.Time + level alerts.AlertLevel // Highest delivered severity for this occurrence. } // Alert represents an alert (interface to avoid circular dependency) @@ -1164,10 +1165,12 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp } // Check cooldown. A disabled cooldown means "do not send repeat - // notifications for the same active alert occurrence"; escalation uses an - // explicit bypass because its cadence is owned by the alert schedule. + // notifications for the same active alert occurrence". A higher severity is + // new information, not a repeat. Scheduled escalation retains its explicit + // bypass because its cadence is owned by the alert schedule. record, exists := n.lastNotified[alert.ID] - if !options.bypassCooldown && exists && record.alertStart.Equal(alert.StartTime) && (n.cooldown <= 0 || time.Since(record.lastSent) < n.cooldown) { + severityIncreased := notificationSeverityRank(record.level) > 0 && notificationSeverityRank(alert.Level) > notificationSeverityRank(record.level) + if !options.bypassCooldown && !severityIncreased && exists && record.alertStart.Equal(alert.StartTime) && (n.cooldown <= 0 || time.Since(record.lastSent) < n.cooldown) { elapsed := time.Since(record.lastSent) remainingCooldown := time.Duration(0) if n.cooldown > elapsed { @@ -1221,6 +1224,20 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp n.mu.Unlock() } +// Unknown levels cannot opt into the severity cooldown exception. +func notificationSeverityRank(level alerts.AlertLevel) int { + switch level { + case alerts.AlertLevelInfo: + return 1 + case alerts.AlertLevelWarning: + return 2 + case alerts.AlertLevelCritical: + return 3 + default: + return 0 + } +} + func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, sentAt time.Time) { n.mu.Lock() if n.lastNotified == nil { @@ -1230,7 +1247,14 @@ func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, s if alert == nil { continue } + level := alert.Level + // Destinations and retries can complete out of order. A late warning + // receipt must not re-enable critical delivery for this occurrence. + if previous, ok := n.lastNotified[alert.ID]; ok && previous.alertStart.Equal(alert.StartTime) && notificationSeverityRank(previous.level) > notificationSeverityRank(level) { + level = previous.level + } n.lastNotified[alert.ID] = notificationRecord{ + level: level, lastSent: sentAt, alertStart: alert.StartTime, } diff --git a/internal/notifications/severity_cooldown_test.go b/internal/notifications/severity_cooldown_test.go new file mode 100644 index 0000000000..bb33b00510 --- /dev/null +++ b/internal/notifications/severity_cooldown_test.go @@ -0,0 +1,59 @@ +package notifications + +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/stretchr/testify/require" +) + +func TestNotificationSeverityCooldown(t *testing.T) { + for _, cooldown := range []time.Duration{0, time.Hour} { + for _, tc := range []struct { + from, to alerts.AlertLevel + send bool + }{ + {alerts.AlertLevelInfo, alerts.AlertLevelWarning, true}, + {alerts.AlertLevelWarning, alerts.AlertLevelCritical, true}, + {alerts.AlertLevelCritical, alerts.AlertLevelCritical, false}, + {alerts.AlertLevelCritical, alerts.AlertLevelWarning, false}, + {alerts.AlertLevelWarning, alerts.AlertLevelWarning, false}, + {"", alerts.AlertLevelCritical, false}, + {alerts.AlertLevelWarning, "unknown", false}, + } { + t.Run(cooldown.String()+"/"+string(tc.from)+"-"+string(tc.to), func(t *testing.T) { + nm := NewNotificationManagerWithDataDir("", t.TempDir()) + defer nm.Stop() + nm.cooldown = cooldown + nm.SetGroupingWindow(3600) + a := &alerts.Alert{ID: "severity", StartTime: time.Now(), Level: tc.from} + nm.markAlertsNotified([]*alerts.Alert{a}, time.Now()) + a.Level = tc.to + nm.SendAlert(a) + nm.mu.RLock() + count := len(nm.pendingAlerts) + nm.mu.RUnlock() + if tc.send { + require.Equal(t, 1, count) + } else { + require.Zero(t, count) + } + }) + } + } +} + +func TestNotificationSeverityReceiptOrder(t *testing.T) { + nm := NewNotificationManagerWithDataDir("", t.TempDir()) + defer nm.Stop() + a := &alerts.Alert{ID: "severity", StartTime: time.Now(), Level: alerts.AlertLevelCritical} + nm.markAlertsNotified([]*alerts.Alert{a}, time.Now()) + a.Level = alerts.AlertLevelWarning + nm.markAlertsNotified([]*alerts.Alert{a}, time.Now()) + require.Equal(t, alerts.AlertLevelCritical, nm.lastNotified[a.ID].level) + // A new occurrence starts its own severity history. + a.StartTime = a.StartTime.Add(time.Minute) + nm.markAlertsNotified([]*alerts.Alert{a}, time.Now()) + require.Equal(t, alerts.AlertLevelWarning, nm.lastNotified[a.ID].level) +}