mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 18:45:53 +00:00
525 lines
14 KiB
Go
525 lines
14 KiB
Go
package alerts
|
|
|
|
import (
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
// LifecycleEvent is an alert-state transition independent of notification
|
|
// delivery. Consumers use this stream for durable incident history; delivery
|
|
// callbacks remain governed by activation, quiet-hours, and suppression policy.
|
|
type LifecycleEvent struct {
|
|
Type string
|
|
OccurredAt time.Time
|
|
Alert *Alert
|
|
Details map[string]string
|
|
Persisted bool
|
|
}
|
|
|
|
type callbackBus struct {
|
|
mu sync.RWMutex
|
|
|
|
onAlert func(alert *Alert)
|
|
alertSubs map[int]func(alert *Alert)
|
|
onAlertForAI func(alert *Alert)
|
|
alertForAISubs map[int]func(alert *Alert)
|
|
onResolved func(alertID string)
|
|
resolvedSubs map[int]func(alertID string)
|
|
lifecycleSubs map[int]func(event LifecycleEvent)
|
|
|
|
onAcknowledged func(alert *Alert, user string)
|
|
onUnacknowledged func(alert *Alert, user string)
|
|
onEscalate func(alert *Alert, level int)
|
|
|
|
onFlappingDetected func(alert *Alert, trackingKey string)
|
|
|
|
nextCallbackID int
|
|
}
|
|
|
|
func newCallbackBus() callbackBus {
|
|
return callbackBus{
|
|
alertSubs: make(map[int]func(*Alert)),
|
|
alertForAISubs: make(map[int]func(*Alert)),
|
|
resolvedSubs: make(map[int]func(string)),
|
|
lifecycleSubs: make(map[int]func(LifecycleEvent)),
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) subscribeLifecycleCallback(cb func(event LifecycleEvent)) func() {
|
|
if cb == nil {
|
|
return func() {}
|
|
}
|
|
|
|
b.mu.Lock()
|
|
if b.lifecycleSubs == nil {
|
|
b.lifecycleSubs = make(map[int]func(LifecycleEvent))
|
|
}
|
|
id := b.nextIDLocked()
|
|
b.lifecycleSubs[id] = cb
|
|
b.mu.Unlock()
|
|
|
|
return func() {
|
|
b.mu.Lock()
|
|
delete(b.lifecycleSubs, id)
|
|
b.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) nextIDLocked() int {
|
|
b.nextCallbackID++
|
|
return b.nextCallbackID
|
|
}
|
|
|
|
func (b *callbackBus) ensureAlertSubsLocked() {
|
|
if b.alertSubs == nil {
|
|
b.alertSubs = make(map[int]func(*Alert))
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) ensureAlertForAISubsLocked() {
|
|
if b.alertForAISubs == nil {
|
|
b.alertForAISubs = make(map[int]func(*Alert))
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) ensureResolvedSubsLocked() {
|
|
if b.resolvedSubs == nil {
|
|
b.resolvedSubs = make(map[int]func(string))
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) setAlertCallback(cb func(alert *Alert)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.onAlert = cb
|
|
}
|
|
|
|
func (b *callbackBus) subscribeAlertCallback(cb func(alert *Alert)) func() {
|
|
if cb == nil {
|
|
return func() {}
|
|
}
|
|
|
|
b.mu.Lock()
|
|
b.ensureAlertSubsLocked()
|
|
id := b.nextIDLocked()
|
|
b.alertSubs[id] = cb
|
|
b.mu.Unlock()
|
|
|
|
return func() {
|
|
b.mu.Lock()
|
|
delete(b.alertSubs, id)
|
|
b.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) setAlertForAICallback(cb func(alert *Alert)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.onAlertForAI = cb
|
|
}
|
|
|
|
func (b *callbackBus) subscribeAlertForAICallback(cb func(alert *Alert)) func() {
|
|
if cb == nil {
|
|
return func() {}
|
|
}
|
|
|
|
b.mu.Lock()
|
|
b.ensureAlertForAISubsLocked()
|
|
id := b.nextIDLocked()
|
|
b.alertForAISubs[id] = cb
|
|
b.mu.Unlock()
|
|
|
|
return func() {
|
|
b.mu.Lock()
|
|
delete(b.alertForAISubs, id)
|
|
b.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) setResolvedCallback(cb func(alertID string)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.onResolved = cb
|
|
}
|
|
|
|
func (b *callbackBus) subscribeResolvedCallback(cb func(alertID string)) func() {
|
|
if cb == nil {
|
|
return func() {}
|
|
}
|
|
|
|
b.mu.Lock()
|
|
b.ensureResolvedSubsLocked()
|
|
id := b.nextIDLocked()
|
|
b.resolvedSubs[id] = cb
|
|
b.mu.Unlock()
|
|
|
|
return func() {
|
|
b.mu.Lock()
|
|
delete(b.resolvedSubs, id)
|
|
b.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (b *callbackBus) setAcknowledgedCallback(cb func(alert *Alert, user string)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.onAcknowledged = cb
|
|
}
|
|
|
|
func (b *callbackBus) setUnacknowledgedCallback(cb func(alert *Alert, user string)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.onUnacknowledged = cb
|
|
}
|
|
|
|
func (b *callbackBus) setEscalateCallback(cb func(alert *Alert, level int)) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
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
|
|
b.mu.RUnlock()
|
|
return cb
|
|
}
|
|
|
|
func (b *callbackBus) alertCallbacks() []func(alert *Alert) {
|
|
b.mu.RLock()
|
|
defer b.mu.RUnlock()
|
|
|
|
callbacks := make([]func(alert *Alert), 0, len(b.alertSubs)+1)
|
|
if b.onAlert != nil {
|
|
callbacks = append(callbacks, b.onAlert)
|
|
}
|
|
for _, cb := range b.alertSubs {
|
|
if cb != nil {
|
|
callbacks = append(callbacks, cb)
|
|
}
|
|
}
|
|
return callbacks
|
|
}
|
|
|
|
func (b *callbackBus) alertForAICallback() func(alert *Alert) {
|
|
b.mu.RLock()
|
|
cb := b.onAlertForAI
|
|
b.mu.RUnlock()
|
|
return cb
|
|
}
|
|
|
|
func (b *callbackBus) alertForAICallbacks() []func(alert *Alert) {
|
|
b.mu.RLock()
|
|
defer b.mu.RUnlock()
|
|
|
|
callbacks := make([]func(alert *Alert), 0, len(b.alertForAISubs)+1)
|
|
if b.onAlertForAI != nil {
|
|
callbacks = append(callbacks, b.onAlertForAI)
|
|
}
|
|
for _, cb := range b.alertForAISubs {
|
|
if cb != nil {
|
|
callbacks = append(callbacks, cb)
|
|
}
|
|
}
|
|
return callbacks
|
|
}
|
|
|
|
func (b *callbackBus) resolvedCallbacks() []func(alertID string) {
|
|
b.mu.RLock()
|
|
defer b.mu.RUnlock()
|
|
|
|
callbacks := make([]func(alertID string), 0, len(b.resolvedSubs)+1)
|
|
if b.onResolved != nil {
|
|
callbacks = append(callbacks, b.onResolved)
|
|
}
|
|
for _, cb := range b.resolvedSubs {
|
|
if cb != nil {
|
|
callbacks = append(callbacks, cb)
|
|
}
|
|
}
|
|
return callbacks
|
|
}
|
|
|
|
func (b *callbackBus) lifecycleCallbacks() []func(event LifecycleEvent) {
|
|
b.mu.RLock()
|
|
defer b.mu.RUnlock()
|
|
|
|
callbacks := make([]func(LifecycleEvent), 0, len(b.lifecycleSubs))
|
|
for _, cb := range b.lifecycleSubs {
|
|
if cb != nil {
|
|
callbacks = append(callbacks, cb)
|
|
}
|
|
}
|
|
return callbacks
|
|
}
|
|
|
|
func (b *callbackBus) acknowledgedCallback() func(alert *Alert, user string) {
|
|
b.mu.RLock()
|
|
cb := b.onAcknowledged
|
|
b.mu.RUnlock()
|
|
return cb
|
|
}
|
|
|
|
func (b *callbackBus) unacknowledgedCallback() func(alert *Alert, user string) {
|
|
b.mu.RLock()
|
|
cb := b.onUnacknowledged
|
|
b.mu.RUnlock()
|
|
return cb
|
|
}
|
|
|
|
func (b *callbackBus) escalateCallback() func(alert *Alert, level int) {
|
|
b.mu.RLock()
|
|
cb := b.onEscalate
|
|
b.mu.RUnlock()
|
|
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)
|
|
}
|
|
|
|
// SubscribeAlertCallback registers an additional alert callback without
|
|
// replacing the legacy single callback slot. The returned function removes the
|
|
// subscription when called.
|
|
func (m *Manager) SubscribeAlertCallback(cb func(alert *Alert)) func() {
|
|
return m.callbacks.subscribeAlertCallback(cb)
|
|
}
|
|
|
|
// SetAlertForAICallback sets a callback for AI analysis when alerts are created.
|
|
// Unlike SetAlertCallback, this callback is invoked unconditionally - it bypasses
|
|
// activation state, quiet hours, and other notification suppression checks.
|
|
// This allows AI to analyze alerts even when the user hasn't finished setup.
|
|
func (m *Manager) SetAlertForAICallback(cb func(alert *Alert)) {
|
|
m.callbacks.setAlertForAICallback(cb)
|
|
log.Info().Msg("alert-for-AI callback registered (bypasses notification suppression)")
|
|
}
|
|
|
|
// SubscribeAlertForAICallback registers an additional AI alert callback without
|
|
// replacing the legacy single callback slot. The returned function removes the
|
|
// subscription when called.
|
|
func (m *Manager) SubscribeAlertForAICallback(cb func(alert *Alert)) func() {
|
|
return m.callbacks.subscribeAlertForAICallback(cb)
|
|
}
|
|
|
|
// SetResolvedCallback sets the callback for resolved alerts.
|
|
func (m *Manager) SetResolvedCallback(cb func(alertID string)) {
|
|
m.callbacks.setResolvedCallback(cb)
|
|
}
|
|
|
|
// SubscribeResolvedCallback registers an additional resolved-alert callback
|
|
// without replacing the legacy single callback slot. The returned function
|
|
// removes the subscription when called.
|
|
func (m *Manager) SubscribeResolvedCallback(cb func(alertID string)) func() {
|
|
return m.callbacks.subscribeResolvedCallback(cb)
|
|
}
|
|
|
|
// SubscribeLifecycleCallback registers a delivery-independent lifecycle
|
|
// consumer. It is invoked synchronously at the canonical event-emission seam;
|
|
// callbacks must not re-enter the alert manager.
|
|
func (m *Manager) SubscribeLifecycleCallback(cb func(event LifecycleEvent)) func() {
|
|
return m.callbacks.subscribeLifecycleCallback(cb)
|
|
}
|
|
|
|
// SetAcknowledgedCallback sets the callback for acknowledged alerts.
|
|
func (m *Manager) SetAcknowledgedCallback(cb func(alert *Alert, user string)) {
|
|
m.callbacks.setAcknowledgedCallback(cb)
|
|
}
|
|
|
|
// SetUnacknowledgedCallback sets the callback for unacknowledged alerts.
|
|
func (m *Manager) SetUnacknowledgedCallback(cb func(alert *Alert, user string)) {
|
|
m.callbacks.setUnacknowledgedCallback(cb)
|
|
}
|
|
|
|
// SetEscalateCallback sets the callback for escalated alerts.
|
|
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()
|
|
}
|
|
|
|
func (m *Manager) getAlertCallbacks() []func(alert *Alert) {
|
|
return m.callbacks.alertCallbacks()
|
|
}
|
|
|
|
func (m *Manager) getAlertForAICallback() func(alert *Alert) {
|
|
return m.callbacks.alertForAICallback()
|
|
}
|
|
|
|
func (m *Manager) getAlertForAICallbacks() []func(alert *Alert) {
|
|
return m.callbacks.alertForAICallbacks()
|
|
}
|
|
|
|
func (m *Manager) getResolvedCallbacks() []func(alertID string) {
|
|
return m.callbacks.resolvedCallbacks()
|
|
}
|
|
|
|
func (m *Manager) getLifecycleCallbacks() []func(event LifecycleEvent) {
|
|
return m.callbacks.lifecycleCallbacks()
|
|
}
|
|
|
|
func (m *Manager) getAcknowledgedCallback() func(alert *Alert, user string) {
|
|
return m.callbacks.acknowledgedCallback()
|
|
}
|
|
|
|
func (m *Manager) getUnacknowledgedCallback() func(alert *Alert, user string) {
|
|
return m.callbacks.unacknowledgedCallback()
|
|
}
|
|
|
|
func (m *Manager) getEscalateCallback() func(alert *Alert, level int) {
|
|
return m.callbacks.escalateCallback()
|
|
}
|
|
|
|
// safeCallResolvedAlertCallback invokes onResolved with panic recovery while
|
|
// preserving canonical state as the internal identity and emitting the public
|
|
// alert ID to external callbacks for compatibility.
|
|
func (m *Manager) safeCallResolvedAlertCallback(alert *Alert, fallbackID string, async bool) {
|
|
// Record the resolution before the callback guard: this funnel is the
|
|
// resolve seam for every lifecycle path, with or without subscribers.
|
|
m.recordAlertEvent(eventlog.TypeResolved, alert, fallbackID, "", "Alert resolved.", nil)
|
|
|
|
callbacks := m.getResolvedCallbacks()
|
|
if len(callbacks) == 0 {
|
|
return
|
|
}
|
|
|
|
publicID := exportedAlertID(alert, fallbackID)
|
|
trackingKey := canonicalTrackingKeyForAlert(alert)
|
|
|
|
callbackFunc := func() {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Error().
|
|
Interface("panic", r).
|
|
Str("alertID", publicID).
|
|
Str("trackingKey", trackingKey).
|
|
Msg("Panic in onResolved callback")
|
|
}
|
|
}()
|
|
for _, callback := range callbacks {
|
|
callback(publicID)
|
|
}
|
|
}
|
|
|
|
if async {
|
|
go callbackFunc()
|
|
} else {
|
|
callbackFunc()
|
|
}
|
|
}
|
|
|
|
// safeCallAcknowledgedCallback invokes onAcknowledged with panic recovery and alert cloning.
|
|
func (m *Manager) safeCallAcknowledgedCallback(alert *Alert, user string) {
|
|
m.recordAlertEvent(eventlog.TypeAcknowledged, alert, "", "", "Alert acknowledged.",
|
|
map[string]string{"user": user})
|
|
|
|
callback := m.getAcknowledgedCallback()
|
|
if callback == nil || alert == nil {
|
|
return
|
|
}
|
|
|
|
alertCopy := cloneAlertForOutput(alert)
|
|
go func(a *Alert, u string) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Error().
|
|
Interface("panic", r).
|
|
Str("alertID", a.ID).
|
|
Msg("Panic in onAcknowledged callback")
|
|
}
|
|
}()
|
|
callback(a, u)
|
|
}(alertCopy, user)
|
|
}
|
|
|
|
// safeCallUnacknowledgedCallback invokes onUnacknowledged with panic recovery and alert cloning.
|
|
func (m *Manager) safeCallUnacknowledgedCallback(alert *Alert, user string) {
|
|
m.recordAlertEvent(eventlog.TypeUnacknowledged, alert, "", "", "Alert acknowledgement removed.",
|
|
map[string]string{"user": user})
|
|
|
|
callback := m.getUnacknowledgedCallback()
|
|
if callback == nil || alert == nil {
|
|
return
|
|
}
|
|
|
|
alertCopy := cloneAlertForOutput(alert)
|
|
go func(a *Alert, u string) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Error().
|
|
Interface("panic", r).
|
|
Str("alertID", a.ID).
|
|
Msg("Panic in onUnacknowledged callback")
|
|
}
|
|
}()
|
|
callback(a, u)
|
|
}(alertCopy, user)
|
|
}
|
|
|
|
// safeCallEscalateCallback invokes onEscalate with panic recovery and alert cloning.
|
|
func (m *Manager) safeCallEscalateCallback(alert *Alert, level int) {
|
|
m.safeCallEscalateCallbackWithKind(alert, level, false)
|
|
}
|
|
|
|
func (m *Manager) safeCallEscalationRepeatCallback(alert *Alert, level int) {
|
|
m.safeCallEscalateCallbackWithKind(alert, level, true)
|
|
}
|
|
|
|
func (m *Manager) safeCallEscalateCallbackWithKind(alert *Alert, level int, repeat bool) {
|
|
details := map[string]string{"level": strconv.Itoa(level)}
|
|
message := "Alert escalated."
|
|
if repeat {
|
|
details["repeat"] = "true"
|
|
message = "Critical alert escalation repeated."
|
|
}
|
|
m.recordAlertEvent(eventlog.TypeEscalated, alert, "", "", message, details)
|
|
|
|
callback := m.getEscalateCallback()
|
|
if callback == nil {
|
|
return
|
|
}
|
|
|
|
alertCopy := cloneAlertForOutput(alert)
|
|
go func(a *Alert, lvl int) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Error().
|
|
Interface("panic", r).
|
|
Str("alertID", a.ID).
|
|
Int("level", lvl).
|
|
Msg("Panic in onEscalate callback")
|
|
}
|
|
}()
|
|
callback(a, lvl)
|
|
}(alertCopy, level)
|
|
}
|