vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-09 16:00:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Track alert save workers during shutdown

Browse source
This commit is contained in:
rcourtman 2026-06-28 01:39:12 +01:00
parent 4fc56162f6
commit 85c46cfcd1
12 changed files with 66 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

22
internal/alerts/active_cleanup.go
View file

@ -310,16 +310,7 @@ func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) {
if removedCount > 0 {
log.Debug().Int("removed", removedCount).Int("remaining", len(m.activeAlerts)).Msg("cleaned up alerts for non-existent nodes")
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Msg("panic in SaveActiveAlerts goroutine (cleanup)")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("failed to save alerts after cleanup")
}
}()
m.saveActiveAlertsAsync("node cleanup")
} else {
log.Debug().Msg("no alerts needed cleanup")
}
@ -358,14 +349,5 @@ func (m *Manager) ClearActiveAlerts() {
log.Info().Msg("cleared all active and pending alerts")
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Msg("panic in SaveActiveAlerts goroutine (clear)")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("failed to persist cleared alerts")
}
}()
m.saveActiveAlertsAsync("clear active alerts")
}

42
internal/alerts/active_persistence.go
View file

@ -14,15 +14,30 @@ import (
"github.com/rs/zerolog/log"
)
func asyncSaveActiveAlerts(reason string, save func() error) {
func (m *Manager) saveActiveAlertsAsync(context string) {
m.stopMu.RLock()
if m.stopping {
m.stopMu.RUnlock()
return
}
m.workerWG.Add(1)
m.stopMu.RUnlock()
go func() {
defer m.workerWG.Done()
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Str("reason", reason).Msg("panic in SaveActiveAlerts goroutine")
log.Error().
Interface("panic", r).
Str("context", context).
Msg("Panic in SaveActiveAlerts goroutine")
}
}()
if err := save(); err != nil {
log.Error().Err(err).Str("reason", reason).Msg("failed to save active alerts")
if err := m.SaveActiveAlerts(); err != nil {
log.Error().
Err(err).
Str("context", context).
Msg("Failed to save active alerts")
}
}()
}
@ -107,25 +122,6 @@ func (m *Manager) SaveActiveAlerts() error {
return nil
}
func (m *Manager) saveActiveAlertsAsync(context string) {
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().
Interface("panic", r).
Str("context", context).
Msg("Panic in SaveActiveAlerts goroutine")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().
Err(err).
Str("context", context).
Msg("Failed to save active alerts")
}
}()
}
// LoadActiveAlerts restores active alerts from disk.
func (m *Manager) LoadActiveAlerts() error {
m.mu.Lock()

22
internal/alerts/alerts_test.go
View file

@ -101,28 +101,8 @@ func newTestManager(t *testing.T) *Manager {
m := NewManager()
// Restore env var and release mutex when test completes.
// We also stop the history manager's background goroutines (but not the
// full manager Stop which includes a 100ms sleep) to prevent writes to
// the temp directory after the test completes.
t.Cleanup(func() {
// Stop the history manager to halt background save routines
m.historyManager.Stop()
// Close escalation channel to stop that goroutine too
select {
case <-m.escalationStop:
// Already closed
default:
close(m.escalationStop)
}
// Close cleanup channel
select {
case <-m.cleanupStop:
// Already closed
default:
close(m.cleanupStop)
}
// Brief pause to let goroutines finish any in-flight operations
time.Sleep(10 * time.Millisecond)
m.Stop()
if hadOld {
os.Setenv("PULSE_DATA_DIR", oldVal)

4
internal/alerts/backup_snapshot.go
View file

@ -525,7 +525,7 @@ func (m *Manager) CheckSnapshotsForInstance(instanceName string, snapshots []mod
},
})
if result.Transition != nil && result.Transition.Kind == alertspecs.EvaluationTransitionActivated {
asyncSaveActiveAlerts("snapshot", m.SaveActiveAlerts)
m.saveActiveAlertsAsync("snapshot")
}
}
@ -913,7 +913,7 @@ func (m *Manager) CheckBackupsWithInventory(
},
})
if result.Transition != nil && result.Transition.Kind == alertspecs.EvaluationTransitionActivated {
asyncSaveActiveAlerts("backup", m.SaveActiveAlerts)
m.saveActiveAlertsAsync("backup")
}
}

6
internal/alerts/canonical_metric.go
View file

@ -113,7 +113,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
migratedAlertIdentity := false
defer func() {
if migratedAlertIdentity {
asyncSaveActiveAlerts("canonical guest metric node move", m.SaveActiveAlerts)
m.saveActiveAlertsAsync("canonical guest metric node move")
}
}()
defer m.mu.Unlock()
@ -237,7 +237,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
m.recentAlerts[trackingKey] = alert
m.historyManager.AddAlert(*alert)
asyncSaveActiveAlerts("canonical metric create", m.SaveActiveAlerts)
m.saveActiveAlertsAsync("canonical metric create")
if alertForAICallback := m.getAlertForAICallback(); alertForAICallback != nil {
alertCopy := cloneAlertForOutput(alert)
@ -315,7 +315,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
ResolvedTime: observedAt,
}
m.removeActiveAlertNoLock(storageKey)
asyncSaveActiveAlerts("canonical metric resolution", m.SaveActiveAlerts)
m.saveActiveAlertsAsync("canonical metric resolution")
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
m.safeCallResolvedAlertCallback(existingAlert, storageKey, true)
}

11
internal/alerts/config_runtime.go
View file

@ -538,16 +538,7 @@ func (m *Manager) reevaluateActiveAlertsLocked() {
}
if len(alertsToResolve) > 0 {
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Msg("panic in SaveActiveAlerts goroutine (config update)")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("failed to save active alerts after config update")
}
}()
m.saveActiveAlertsAsync("config update")
}
}

11
internal/alerts/history.go
View file

@ -54,6 +54,7 @@ type HistoryManager struct {
saveInterval time.Duration
stopChan chan struct{}
saveTicker *time.Ticker
workerWG sync.WaitGroup
callbacks []AlertCallback // Called when alerts are added
}
@ -118,7 +119,11 @@ func NewHistoryManager(dataDir string) *HistoryManager {
hm.startPeriodicSave()
// Start cleanup routine
go hm.cleanupRoutine()
hm.workerWG.Add(1)
go func() {
defer hm.workerWG.Done()
hm.cleanupRoutine()
}()
return hm
}
@ -462,7 +467,9 @@ func syncDir(dirPath string) error {
func (hm *HistoryManager) startPeriodicSave() {
hm.saveTicker = time.NewTicker(hm.saveInterval)
hm.workerWG.Add(1)
go func() {
defer hm.workerWG.Done()
for {
select {
case <-hm.saveTicker.C:
@ -643,6 +650,8 @@ func (hm *HistoryManager) Stop() {
hm.saveTicker.Stop()
}
hm.workerWG.Wait()
// Save one final time
if err := hm.saveHistory(); err != nil {
log.Error().Err(err).Msg("Failed to save alert history on shutdown")

11
internal/alerts/lifecycle.go
View file

@ -1,23 +1,24 @@
package alerts
import (
"time"
"github.com/rs/zerolog/log"
)
// Stop stops the alert manager and saves history
func (m *Manager) Stop() {
m.stopOnce.Do(func() {
m.stopMu.Lock()
m.stopping = true
closeSignalChannel(m.escalationStop)
closeSignalChannel(m.cleanupStop)
m.stopMu.Unlock()
m.workerWG.Wait()
if m.historyManager != nil {
m.historyManager.Stop()
}
// Give background goroutines time to exit cleanly
time.Sleep(100 * time.Millisecond)
// Save active alerts before stopping
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("Failed to save active alerts on stop")

25
internal/alerts/manager.go
View file

@ -86,6 +86,9 @@ type Manager struct {
quietHoursLoc *time.Location
now func() time.Time
stopOnce sync.Once
stopMu sync.RWMutex
stopping bool
workerWG sync.WaitGroup
}
type ackRecord struct {
@ -151,14 +154,20 @@ func NewManagerWithDataDir(dataDir string) *Manager {
log.Error().Err(err).Msg("failed to load active alerts")
}
// Start escalation checker
go m.escalationChecker()
// Start periodic save of active alerts
go m.periodicSaveAlerts()
// Start periodic cleanup of stale tracking map entries
go m.trackingMapCleanup()
// Start background workers.
m.workerWG.Add(3)
go func() {
defer m.workerWG.Done()
m.escalationChecker()
}()
go func() {
defer m.workerWG.Done()
m.periodicSaveAlerts()
}()
go func() {
defer m.workerWG.Done()
m.trackingMapCleanup()
}()
return m
}

26
internal/alerts/metric_runtime.go
View file

@ -175,7 +175,7 @@ func (m *Manager) checkMetric(resourceID, resourceName, node, instance, resource
migratedAlertIdentity := false
defer func() {
if migratedAlertIdentity {
asyncSaveActiveAlerts("guest metric node move", m.SaveActiveAlerts)
m.saveActiveAlertsAsync("guest metric node move")
}
}()
defer m.mu.Unlock()
@ -342,17 +342,7 @@ func (m *Manager) checkMetric(resourceID, resourceName, node, instance, resource
m.recentAlerts[trackingKey] = alert
m.historyManager.AddAlert(*alert)
// Save active alerts after adding new one
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Msg("panic in SaveActiveAlerts goroutine")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("failed to save active alerts after creation")
}
}()
m.saveActiveAlertsAsync("metric create")
log.Warn().
Str("alertID", alertID).
@ -505,17 +495,7 @@ func (m *Manager) checkMetric(resourceID, resourceName, node, instance, resource
// Remove from active alerts
m.removeActiveAlertNoLock(alertID)
// Save active alerts after resolution
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Msg("panic in SaveActiveAlerts goroutine (resolution)")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("failed to save active alerts after resolution")
}
}()
m.saveActiveAlertsAsync("metric resolution")
// Add to recently resolved while preventing lock-order inversions
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)

11
internal/alerts/tracking_cleanup.go
View file

@ -223,16 +223,7 @@ func (m *Manager) cleanupStaleMaps() {
}
if staleResolved > 0 {
go func() {
defer func() {
if r := recover(); r != nil {
log.Error().Interface("panic", r).Msg("panic in SaveActiveAlerts goroutine (stale cleanup)")
}
}()
if err := m.SaveActiveAlerts(); err != nil {
log.Error().Err(err).Msg("failed to save active alerts after stale cleanup")
}
}()
m.saveActiveAlertsAsync("stale cleanup")
log.Info().
Int("count", staleResolved).
Msg("Auto-resolved stale alerts")

9
internal/alerts/unified_eval_parity_test.go
View file

@ -2,7 +2,6 @@ package alerts
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
@ -11,14 +10,8 @@ func newUnifiedEvalParityManager(t *testing.T) *Manager {
t.Helper()
m := NewManagerWithDataDir(t.TempDir())
// Stop background workers to avoid post-test writes into temp dirs.
t.Cleanup(func() {
if m.historyManager != nil {
m.historyManager.Stop()
}
closeSignalChannel(m.escalationStop)
closeSignalChannel(m.cleanupStop)
time.Sleep(10 * time.Millisecond)
m.Stop()
})
return m