diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index 0d266efcb..9559acf6e 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -4301,7 +4301,7 @@ func TestCheckFlapping(t *testing.T) { // Call checkFlappingLocked m.mu.Lock() - result := m.checkFlappingLocked(alertID) + result, _ := m.checkFlappingLocked(alertID) m.mu.Unlock() if result != tt.expectFlapping { @@ -4354,7 +4354,7 @@ func TestCheckFlappingAlreadyFlapping(t *testing.T) { // Call checkFlappingLocked - should return true but NOT update suppression m.mu.Lock() - result := m.checkFlappingLocked(alertID) + result, _ := m.checkFlappingLocked(alertID) m.mu.Unlock() if !result { @@ -4396,7 +4396,7 @@ func TestCheckFlappingWindowExpiry(t *testing.T) { // Call checkFlappingLocked - old entries should be pruned m.mu.Lock() - result := m.checkFlappingLocked(alertID) + result, _ := m.checkFlappingLocked(alertID) historyLen := len(m.flappingHistory[alertID]) m.mu.Unlock() diff --git a/internal/alerts/callbacks.go b/internal/alerts/callbacks.go index 66b4a8c67..1161a7f5a 100644 --- a/internal/alerts/callbacks.go +++ b/internal/alerts/callbacks.go @@ -20,6 +20,8 @@ type callbackBus struct { onUnacknowledged func(alert *Alert, user string) onEscalate func(alert *Alert, level int) + onFlappingDetected func(alert *Alert, trackingKey string) + nextCallbackID int } @@ -144,6 +146,12 @@ func (b *callbackBus) setEscalateCallback(cb func(alert *Alert, level int)) { b.onEscalate = cb } +func (b *callbackBus) setFlappingDetectedCallback(cb func(alert *Alert, trackingKey string)) { + b.mu.Lock() + defer b.mu.Unlock() + b.onFlappingDetected = cb +} + func (b *callbackBus) alertCallback() func(alert *Alert) { b.mu.RLock() cb := b.onAlert @@ -234,6 +242,13 @@ func (b *callbackBus) escalateCallback() func(alert *Alert, level int) { return cb } +func (b *callbackBus) flappingDetectedCallback() func(alert *Alert, trackingKey string) { + b.mu.RLock() + cb := b.onFlappingDetected + b.mu.RUnlock() + return cb +} + // SetAlertCallback sets the callback for new alerts. func (m *Manager) SetAlertCallback(cb func(alert *Alert)) { m.callbacks.setAlertCallback(cb) @@ -289,6 +304,16 @@ func (m *Manager) SetEscalateCallback(cb func(alert *Alert, level int)) { m.callbacks.setEscalateCallback(cb) } +// SetFlappingDetectedCallback registers a callback fired exactly once on the +// transition into flapping suppression for a given trackingKey. The callback +// is invoked from a goroutine -- the alerts manager lock is NOT held when it +// runs -- so the callback is free to take its own locks or schedule a patrol. +// It will NOT fire again for the same trackingKey while the flapping cooldown +// window is active; subsequent suppressed dispatches are silent. +func (m *Manager) SetFlappingDetectedCallback(cb func(alert *Alert, trackingKey string)) { + m.callbacks.setFlappingDetectedCallback(cb) +} + func (m *Manager) getAlertCallback() func(alert *Alert) { return m.callbacks.alertCallback() } diff --git a/internal/alerts/flapping_callback_test.go b/internal/alerts/flapping_callback_test.go new file mode 100644 index 000000000..e3a8458e4 --- /dev/null +++ b/internal/alerts/flapping_callback_test.go @@ -0,0 +1,165 @@ +package alerts + +import ( + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestFlappingDetectedCallback verifies that: +// 1. The callback fires exactly once on the transition into the flapping +// state for a given trackingKey. +// 2. Subsequent calls inside the cooldown window (alert still flapping) +// are silent -- the one-shot semantics hold. +// 3. The callback receives the canonical tracking key plus the cloned +// alert so the consumer can read identity / context fields directly. +// 4. The callback runs without the alerts manager lock held: it is safe +// to take the manager's own mutex from inside the callback. +func TestFlappingDetectedCallback(t *testing.T) { + m := newTestManager(t) + + var ( + mu sync.Mutex + calls int32 + recvKey string + recvAlrt *Alert + ) + done := make(chan struct{}, 8) + + m.SetFlappingDetectedCallback(func(a *Alert, trackingKey string) { + // Re-entering the manager from inside the callback must not deadlock: + // if the manager lock were still held when this fires, RLock() here + // would block forever and the test would time out. + m.mu.RLock() + _ = m.flappingActive[trackingKey] + m.mu.RUnlock() + + mu.Lock() + recvKey = trackingKey + recvAlrt = a + mu.Unlock() + atomic.AddInt32(&calls, 1) + done <- struct{}{} + }) + + m.mu.Lock() + m.config.FlappingEnabled = true + m.config.FlappingThreshold = 3 + m.config.FlappingWindowSeconds = 300 + m.config.FlappingCooldownMinutes = 15 + m.config.ActivationState = ActivationActive + m.mu.Unlock() + + // Subscribe an inert alert callback so dispatchAlert reaches the flapping + // check (it short-circuits when no alert callbacks are registered). + m.SetAlertCallback(func(*Alert) {}) + + alert := &Alert{ + ID: "vm-100-cpu", + Type: "cpu", + ResourceID: "vm-100", + ResourceName: "testvm", + CanonicalKind: "vm", + Level: AlertLevelWarning, + } + + // First two dispatches: below threshold, no transition, no callback. + for i := 0; i < 2; i++ { + m.mu.Lock() + m.dispatchAlert(alert, false) + m.mu.Unlock() + } + if got := atomic.LoadInt32(&calls); got != 0 { + t.Fatalf("expected 0 callback invocations before threshold, got %d", got) + } + + // Third dispatch crosses the threshold -- this is the transition. + m.mu.Lock() + m.dispatchAlert(alert, false) + m.mu.Unlock() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("flapping-detected callback did not fire after threshold crossing") + } + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected 1 callback invocation on transition, got %d", got) + } + + mu.Lock() + if recvKey == "" { + t.Errorf("expected non-empty trackingKey in callback") + } + if recvAlrt == nil { + t.Fatal("expected non-nil alert in callback") + } + if recvAlrt.ResourceID != "vm-100" { + t.Errorf("expected ResourceID=vm-100, got %q", recvAlrt.ResourceID) + } + if recvAlrt.CanonicalKind != "vm" { + t.Errorf("expected CanonicalKind=vm, got %q", recvAlrt.CanonicalKind) + } + if recvAlrt.Type != "cpu" { + t.Errorf("expected Type=cpu, got %q", recvAlrt.Type) + } + if recvAlrt.ResourceName != "testvm" { + t.Errorf("expected ResourceName=testvm, got %q", recvAlrt.ResourceName) + } + mu.Unlock() + + // Subsequent dispatches inside the cooldown window: alert remains + // flapping, callback must not refire. + for i := 0; i < 4; i++ { + m.mu.Lock() + m.dispatchAlert(alert, false) + m.mu.Unlock() + } + + // Give any stray goroutines a chance to run. + time.Sleep(50 * time.Millisecond) + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected callback to fire EXACTLY once on transition; got %d invocations", got) + } +} + +// TestFlappingDetectedCallbackNotInvokedWhenDisabled confirms the callback is +// not fired when flapping detection is disabled in config -- the early return +// in checkFlappingLocked must not surface a phantom transition. +func TestFlappingDetectedCallbackNotInvokedWhenDisabled(t *testing.T) { + m := newTestManager(t) + + var calls int32 + m.SetFlappingDetectedCallback(func(*Alert, string) { + atomic.AddInt32(&calls, 1) + }) + + m.mu.Lock() + m.config.FlappingEnabled = false + m.config.FlappingThreshold = 1 + m.config.ActivationState = ActivationActive + m.mu.Unlock() + m.SetAlertCallback(func(*Alert) {}) + + alert := &Alert{ + ID: "vm-101-mem", + Type: "memory", + ResourceID: "vm-101", + ResourceName: "testvm", + CanonicalKind: "vm", + Level: AlertLevelWarning, + } + for i := 0; i < 10; i++ { + m.mu.Lock() + m.dispatchAlert(alert, false) + m.mu.Unlock() + } + + time.Sleep(50 * time.Millisecond) + if got := atomic.LoadInt32(&calls); got != 0 { + t.Fatalf("expected 0 callback invocations when flapping detection disabled, got %d", got) + } +} diff --git a/internal/alerts/notification_policy.go b/internal/alerts/notification_policy.go index 3a0d770d3..3f4ad3fb2 100644 --- a/internal/alerts/notification_policy.go +++ b/internal/alerts/notification_policy.go @@ -8,12 +8,20 @@ import ( "github.com/rs/zerolog/log" ) -// checkFlappingLocked detects alert flapping and returns true if alert should be suppressed. +// checkFlappingLocked detects alert flapping and returns whether the alert +// should be suppressed and whether this call is the first transition into +// the flapping state for the current cooldown window. justTransitioned is +// the signal callers use to fire a one-shot postmortem hook on first +// detection -- it is true only on the call that flips +// flappingActive[trackingKey] from false to true. +// // It modifies flappingHistory, flappingActive, and suppressedUntil maps. -// IMPORTANT: Caller MUST hold m.mu before calling this function. -func (m *Manager) checkFlappingLocked(trackingKey string) bool { +// IMPORTANT: Caller MUST hold m.mu before calling this function. The +// transition signal is returned so the caller can dispatch the +// flapping-detected callback OUTSIDE the lock (typically via a goroutine). +func (m *Manager) checkFlappingLocked(trackingKey string) (suppress bool, justTransitioned bool) { if !m.config.FlappingEnabled { - return false + return false, false } now := time.Now() @@ -58,11 +66,12 @@ func (m *Manager) checkFlappingLocked(trackingKey string) bool { if recordAlertSuppressed != nil { recordAlertSuppressed("flapping") } + return true, true } - return true + return true, false } - return false + return false, false } func (m *Manager) dispatchAlert(alert *Alert, async bool) bool { @@ -82,8 +91,32 @@ func (m *Manager) dispatchAlert(alert *Alert, async bool) bool { trackingKey := canonicalTrackingKeyForAlert(alert) - // Check for flapping (caller must hold m.mu) - if m.checkFlappingLocked(trackingKey) { + // Check for flapping (caller must hold m.mu). When this call is the + // first transition into the flapping state for the current cooldown + // window, fire the flapping-detected callback so a postmortem patrol + // can be triggered. The callback runs in its own goroutine because + // the caller of dispatchAlert holds m.mu and the callback is allowed + // to take its own locks or re-enter the alerts package. + suppress, justTransitioned := m.checkFlappingLocked(trackingKey) + if justTransitioned { + cb := m.callbacks.flappingDetectedCallback() + if cb != nil { + alertCopy := cloneAlertForOutput(alert) + go func(a *Alert, key string, fn func(*Alert, string)) { + defer func() { + if r := recover(); r != nil { + log.Error(). + Interface("panic", r). + Str("alertID", a.ID). + Str("trackingKey", key). + Msg("Panic in onFlappingDetected callback") + } + }() + fn(a, key) + }(alertCopy, trackingKey, cb) + } + } + if suppress { log.Debug(). Str("alertID", alert.ID). Str("trackingKey", trackingKey).