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.
This commit is contained in:
parent
99e5a38534
commit
1183b87fa1
2 changed files with 68 additions and 10 deletions
|
|
@ -734,6 +734,8 @@ func (m *Manager) safeCallEscalateCallback(alert *Alert, level int) {
|
||||||
// dispatchAlert delivers an alert to the configured callback, cloning it first to
|
// dispatchAlert delivers an alert to the configured callback, cloning it first to
|
||||||
// prevent concurrent mutations from racing with consumers.
|
// prevent concurrent mutations from racing with consumers.
|
||||||
// checkFlapping checks if an alert is flapping (changing state too rapidly)
|
// 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 {
|
func (m *Manager) checkFlapping(alertID string) bool {
|
||||||
if !m.config.FlappingEnabled {
|
if !m.config.FlappingEnabled {
|
||||||
return false
|
return false
|
||||||
|
|
@ -5483,15 +5485,16 @@ func (m *Manager) GetActiveAlerts() []Alert {
|
||||||
// NotifyExistingAlert re-dispatches a notification for an existing active alert
|
// NotifyExistingAlert re-dispatches a notification for an existing active alert
|
||||||
// Used when activation state changes from pending to active
|
// Used when activation state changes from pending to active
|
||||||
func (m *Manager) NotifyExistingAlert(alertID string) {
|
func (m *Manager) NotifyExistingAlert(alertID string) {
|
||||||
m.mu.RLock()
|
m.mu.Lock()
|
||||||
alert, exists := m.activeAlerts[alertID]
|
defer m.mu.Unlock()
|
||||||
m.mu.RUnlock()
|
|
||||||
|
|
||||||
|
alert, exists := m.activeAlerts[alertID]
|
||||||
if !exists {
|
if !exists {
|
||||||
return
|
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)
|
m.dispatchAlert(alert, true)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -7311,6 +7314,44 @@ func (m *Manager) Cleanup(maxAge time.Duration) {
|
||||||
Msg("Cleaned up stale Docker restart tracking entry")
|
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
|
// 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
|
// Use a goroutine and add a small delay to avoid notification spam on startup
|
||||||
alertCopy := alert.Clone()
|
alertCopy := alert.Clone()
|
||||||
go func(a *Alert) {
|
go func(a *Alert) {
|
||||||
time.Sleep(10 * time.Second) // Wait for system to stabilize after restart
|
// Wait for system to stabilize or cancellation
|
||||||
log.Info().
|
select {
|
||||||
Str("alertID", a.ID).
|
case <-time.After(10 * time.Second):
|
||||||
Str("resource", a.ResourceName).
|
log.Info().
|
||||||
Msg("Attempting to send notification for restored critical alert")
|
Str("alertID", a.ID).
|
||||||
m.dispatchAlert(a, false) // Use dispatchAlert to respect activation state and quiet hours
|
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)
|
}(alertCopy)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,7 @@ type HistoryEntry struct {
|
||||||
// HistoryManager manages persistent alert history
|
// HistoryManager manages persistent alert history
|
||||||
type HistoryManager struct {
|
type HistoryManager struct {
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
|
saveMu sync.Mutex // Serializes disk writes to prevent save race condition
|
||||||
dataDir string
|
dataDir string
|
||||||
historyFile string
|
historyFile string
|
||||||
backupFile string
|
backupFile string
|
||||||
|
|
@ -174,6 +175,10 @@ func (hm *HistoryManager) saveHistory() error {
|
||||||
|
|
||||||
// saveHistoryWithRetry saves history with exponential backoff retry
|
// saveHistoryWithRetry saves history with exponential backoff retry
|
||||||
func (hm *HistoryManager) saveHistoryWithRetry(maxRetries int) error {
|
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()
|
hm.mu.RLock()
|
||||||
snapshot := make([]HistoryEntry, len(hm.history))
|
snapshot := make([]HistoryEntry, len(hm.history))
|
||||||
copy(snapshot, hm.history)
|
copy(snapshot, hm.history)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue