From 1183b87fa13b3c4de1fd939e234527bad1d76f2e Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 7 Nov 2025 09:12:28 +0000 Subject: [PATCH] Fix critical alert system concurrency and memory leak issues This commit addresses 7 critical issues identified during the alert system audit: **P0 Critical - Race Conditions Fixed:** 1. **dispatchAlert race in NotifyExistingAlert** (lines 5486-5497) - Changed from RLock to Lock to hold mutex during dispatchAlert call - dispatchAlert calls checkFlapping which writes to maps (flappingHistory, flappingActive, suppressedUntil) - Previous code: grabbed RLock, got alert pointer, released lock, then called dispatchAlert (RACE) - Fixed: hold Lock through dispatchAlert call 2. **dispatchAlert race in LoadActiveAlerts startup** (lines 8216-8235) - Startup goroutines called dispatchAlert without holding lock - Added m.mu.Lock/Unlock around dispatchAlert call in goroutine - Also added cancellation via escalationStop channel to prevent goroutine leaks on shutdown 3. **checkFlapping documentation** (line 738) - Added clear comment that checkFlapping requires caller to hold m.mu - Prevents future race conditions from improper usage **P1 Important - Data Loss Prevention:** 4. **History save race condition** (lines 177-180 in history.go) - Added saveMu mutex to serialize disk writes - Previous: concurrent saves could interleave, causing newer data to be overwritten by older snapshots - Fixed: saveMu.Lock at start of saveHistoryWithRetry ensures atomic disk writes - Newer snapshots now always win over older ones **P2 Memory Leak Prevention:** 5. **PMG anomaly tracker cleanup** (lines 7318-7331) - Added cleanup for pmgAnomalyTrackers map (24 hour TTL based on LastSampleTime) - Prevents unbounded growth from decommissioned/transient PMG instances - Each tracker: ~1-2KB (48 samples + baselines) 6. **PMG quarantine history cleanup** (lines 7333-7354) - Added cleanup for pmgQuarantineHistory map (7 day TTL based on last snapshot) - Prevents memory leak for deleted PMG instances - Removes both empty histories and very old histories **P2 Goroutine Leak Prevention:** 7. **Startup notification goroutine cancellation** (lines 8218-8234) - Added select with escalationStop channel to cancel startup notifications - Prevents goroutines from continuing after Stop() is called - Scales with number of restored critical alerts All fixes maintain proper lock ordering and prevent deadlocks by ensuring locks are held when accessing shared maps. --- internal/alerts/alerts.go | 73 ++++++++++++++++++++++++++++++++------ internal/alerts/history.go | 5 +++ 2 files changed, 68 insertions(+), 10 deletions(-) diff --git a/internal/alerts/alerts.go b/internal/alerts/alerts.go index d09894b..38961da 100644 --- a/internal/alerts/alerts.go +++ b/internal/alerts/alerts.go @@ -734,6 +734,8 @@ func (m *Manager) safeCallEscalateCallback(alert *Alert, level int) { // dispatchAlert delivers an alert to the configured callback, cloning it first to // prevent concurrent mutations from racing with consumers. // checkFlapping checks if an alert is flapping (changing state too rapidly) +// checkFlapping detects alert flapping and returns true if alert should be suppressed. +// IMPORTANT: Caller MUST hold m.mu (writes to flappingHistory, flappingActive, suppressedUntil maps) func (m *Manager) checkFlapping(alertID string) bool { if !m.config.FlappingEnabled { return false @@ -5483,15 +5485,16 @@ func (m *Manager) GetActiveAlerts() []Alert { // NotifyExistingAlert re-dispatches a notification for an existing active alert // Used when activation state changes from pending to active func (m *Manager) NotifyExistingAlert(alertID string) { - m.mu.RLock() - alert, exists := m.activeAlerts[alertID] - m.mu.RUnlock() + m.mu.Lock() + defer m.mu.Unlock() + alert, exists := m.activeAlerts[alertID] if !exists { return } - // Dispatch notification for existing alert + // Dispatch notification for existing alert while holding lock + // dispatchAlert expects caller to hold m.mu for checkFlapping safety m.dispatchAlert(alert, true) } @@ -7311,6 +7314,44 @@ func (m *Manager) Cleanup(maxAge time.Duration) { Msg("Cleaned up stale Docker restart tracking entry") } } + + // Clean up stale PMG anomaly trackers (no samples in 24h) + // Prevents memory leak from decommissioned or transient PMG instances + staleTrackerAge := 24 * time.Hour + for pmgID, tracker := range m.pmgAnomalyTrackers { + if tracker != nil && !tracker.LastSampleTime.IsZero() { + if now.Sub(tracker.LastSampleTime) > staleTrackerAge { + delete(m.pmgAnomalyTrackers, pmgID) + log.Debug(). + Str("pmgID", pmgID). + Time("lastSampleTime", tracker.LastSampleTime). + Msg("Cleaned up stale PMG anomaly tracker") + } + } + } + + // Clean up stale PMG quarantine history (no recent snapshots in 7 days) + // Prevents memory leak from deleted PMG instances + staleHistoryAge := 7 * 24 * time.Hour + for pmgID, snapshots := range m.pmgQuarantineHistory { + // If no snapshots remain or last snapshot is very old + if len(snapshots) == 0 { + delete(m.pmgQuarantineHistory, pmgID) + log.Debug(). + Str("pmgID", pmgID). + Msg("Cleaned up empty PMG quarantine history") + continue + } + + lastSnapshot := snapshots[len(snapshots)-1] + if now.Sub(lastSnapshot.Timestamp) > staleHistoryAge { + delete(m.pmgQuarantineHistory, pmgID) + log.Debug(). + Str("pmgID", pmgID). + Time("lastSnapshot", lastSnapshot.Timestamp). + Msg("Cleaned up stale PMG quarantine history") + } + } } // convertLegacyThreshold converts a legacy float64 threshold to HysteresisThreshold @@ -8211,12 +8252,24 @@ func (m *Manager) LoadActiveAlerts() error { // Use a goroutine and add a small delay to avoid notification spam on startup alertCopy := alert.Clone() go func(a *Alert) { - time.Sleep(10 * time.Second) // Wait for system to stabilize after restart - log.Info(). - Str("alertID", a.ID). - Str("resource", a.ResourceName). - Msg("Attempting to send notification for restored critical alert") - m.dispatchAlert(a, false) // Use dispatchAlert to respect activation state and quiet hours + // Wait for system to stabilize or cancellation + select { + case <-time.After(10 * time.Second): + log.Info(). + Str("alertID", a.ID). + Str("resource", a.ResourceName). + Msg("Attempting to send notification for restored critical alert") + + // Acquire lock before calling dispatchAlert (it accesses maps) + m.mu.Lock() + m.dispatchAlert(a, false) // Use dispatchAlert to respect activation state and quiet hours + m.mu.Unlock() + case <-m.escalationStop: + log.Debug(). + Str("alertID", a.ID). + Msg("Cancelled startup notification due to shutdown") + return + } }(alertCopy) } } diff --git a/internal/alerts/history.go b/internal/alerts/history.go index 0e56537..adc673a 100644 --- a/internal/alerts/history.go +++ b/internal/alerts/history.go @@ -30,6 +30,7 @@ type HistoryEntry struct { // HistoryManager manages persistent alert history type HistoryManager struct { mu sync.RWMutex + saveMu sync.Mutex // Serializes disk writes to prevent save race condition dataDir string historyFile string backupFile string @@ -174,6 +175,10 @@ func (hm *HistoryManager) saveHistory() error { // saveHistoryWithRetry saves history with exponential backoff retry func (hm *HistoryManager) saveHistoryWithRetry(maxRetries int) error { + // Serialize all disk writes to prevent concurrent saves from overwriting each other + hm.saveMu.Lock() + defer hm.saveMu.Unlock() + hm.mu.RLock() snapshot := make([]HistoryEntry, len(hm.history)) copy(snapshot, hm.history)