mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
Keep alert acknowledgements through long-running incidents (#2358)
## What changed and why Keep manual and automatic acknowledgements on incidents that are still active during hourly tracking cleanup. Previously the canonical acknowledgement could expire after 24 hours even while the incident continued, causing a short recovery and recurrence to notify again. Active canonical identities now receive the same protection as legacy identities and ordinary cleanup. Inactive expiry and explicit unacknowledgement are unchanged. This repairs a reproduced defect in existing alert behavior. It is not a diagnosis of the whole daily alert cycle reported in #2353 or a claim of installed notification recovery. The periodic-history persistence test also now uses virtual time with the real ticker, verifies two exact persisted snapshots before shutdown, and drains its worker. This replaces a scheduling-sensitive 750 ms poll without changing production intervals or persistence behavior. ## Included source Complete range `7667114fe6dc45e9542d1c7531511964ca6f35ed..1c8f51bf1ff12d559cfbd0a47cf25b879c710fdd`: - Runtime and regression candidate `e256e9fc509cf35143b9bcde638d24baa562f1ae`. - Test-only correction `da685d6d12147d945692df81be13392124bd36cd`. - Reviewed additive integration `1c8f51bf1ff12d559cfbd0a47cf25b879c710fdd`. All reviewed identities and ancestry are preserved. Current upstream remains the range base. ## Validation and limits Recorded baseline `source-zk09jitb` reproduces acknowledgement loss and duplicate dispatch. Final offline proof `source-406x6wbt` passes ten focused race repetitions, complete alerts/notifications/monitoring race suites, vet and affected-library builds. Eighty modeled recurrence cycles cover manual/automatic acknowledgement, JSON/durable checkpoint restart, zero duplicate dispatches, explicit unacknowledgement and seven retention controls. The earlier full proof `source-74ad7ohd` failed the periodic-history polling deadline; the additive test correction is included, not an unchanged favorable rerun. The final integrated alert, notification and monitoring source matches the proved candidate; integration adds only already-published frontend work outside these paths. Complete-range provenance/whitespace, per-commit canonical contract guards and browser applicability checks pass (no frontend delta). Sufficient source evidence was reused rather than repeated solely for publication. Proof Go was 1.26.7 rather than declared 1.26.8; required exact-head CI still must validate the declared toolchain and source before landing. Independent review and exact-head checks govern merge. Source checks do not establish installed recovery, complete reporter acceptance or release qualification. No release, tag or deployment is changed. [How Pulse handles triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)
This commit is contained in:
@@ -43,6 +43,21 @@ callbacks remain outside the manager lock. No retention duration is extended.
|
||||
sample and a short recovery/refire. This is synthetic lifecycle proof, not
|
||||
installed notification-destination acceptance.
|
||||
|
||||
Hourly tracking-map cleanup retains canonical acknowledgement records while
|
||||
their incident is active, including legacy-keyed active snapshots and active
|
||||
recurrences with an older inactive timestamp. A decision does not expire from
|
||||
its acknowledgement age while the condition continues. Recovery starts the
|
||||
existing inactive-retention window; the ordinary cleanup's one-hour expiry,
|
||||
hourly stale-record fallback and explicit unacknowledgement remain unchanged.
|
||||
`TestTrackingCleanupPreservesAcknowledgedProviderRecurrence` in
|
||||
`internal/alerts/reducer_parity_ack_test.go` verifies manual and automatic
|
||||
acknowledgement through provider-native pool incident reconciliation, affirmative
|
||||
recovery, short recurrence, dispatch callbacks and JSON/durable checkpoint restart.
|
||||
`TestTrackingCleanupCanonicalAckRetentionBounds` verifies active/legacy identity,
|
||||
recent inactivity, both existing expiry paths and missing-timestamp fallback.
|
||||
Modeled elapsed time and local callbacks do not prove installed recovery or
|
||||
delivery to an external notification destination.
|
||||
|
||||
Mobile incident drawers transfer their exact context to Assistant and close
|
||||
through the shared explicit handoff callback. Keeping the source drawer above
|
||||
Assistant, or dropping its occurrence identity to make navigation work, fails
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
package alerts
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -97,25 +99,54 @@ func TestCleanupRoutine_ReturnsImmediatelyWhenStopped(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestStartPeriodicSave_PersistsHistoryOnTicker(t *testing.T) {
|
||||
hm := newTestHistoryManager(t)
|
||||
hm.saveInterval = 10 * time.Millisecond
|
||||
hm.history = []HistoryEntry{
|
||||
{Alert: Alert{ID: "periodic-save-alert"}, Timestamp: time.Now()},
|
||||
}
|
||||
|
||||
hm.startPeriodicSave()
|
||||
// Stop waits for the save worker, so an in-flight save cannot race the
|
||||
// TempDir cleanup.
|
||||
defer hm.Stop()
|
||||
|
||||
deadline := time.Now().Add(750 * time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
data, err := os.ReadFile(hm.historyFile)
|
||||
if err == nil && strings.Contains(string(data), "periodic-save-alert") {
|
||||
return
|
||||
// Exercise the real ticker and persistence worker using virtual time.
|
||||
// The old 750 ms filesystem-poll deadline measured wall-clock scheduling
|
||||
// and I/O as well as periodic-save behaviour. Wait drains each tick's real
|
||||
// file I/O before assertions, without adding a production callback or
|
||||
// letting Stop's final save satisfy the periodic-save assertion.
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
hm := newTestHistoryManager(t)
|
||||
hm.history = []HistoryEntry{
|
||||
{Alert: Alert{ID: "periodic-save-alert"}, Timestamp: time.Now()},
|
||||
}
|
||||
time.Sleep(15 * time.Millisecond)
|
||||
}
|
||||
hm.startPeriodicSave()
|
||||
defer hm.Stop()
|
||||
|
||||
t.Fatal("periodic save did not persist history data before deadline")
|
||||
synctest.Wait()
|
||||
if _, err := os.Stat(hm.historyFile); !os.IsNotExist(err) {
|
||||
t.Fatalf("history written before first tick: %v", err)
|
||||
}
|
||||
assertPersisted := func(wantIDs ...string) {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(hm.historyFile)
|
||||
if err != nil {
|
||||
t.Fatalf("read periodic history: %v", err)
|
||||
}
|
||||
var entries []HistoryEntry
|
||||
if err := json.Unmarshal(data, &entries); err != nil {
|
||||
t.Fatalf("decode periodic history: %v", err)
|
||||
}
|
||||
if len(entries) != len(wantIDs) {
|
||||
t.Fatalf("periodic history has %d entries, want %d", len(entries), len(wantIDs))
|
||||
}
|
||||
for i, id := range wantIDs {
|
||||
if entries[i].Alert.ID != id {
|
||||
t.Errorf("entry %d ID = %q, want %q", i, entries[i].Alert.ID, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
time.Sleep(hm.saveInterval)
|
||||
synctest.Wait()
|
||||
assertPersisted("periodic-save-alert")
|
||||
|
||||
hm.mu.Lock()
|
||||
hm.history = append(hm.history, HistoryEntry{
|
||||
Alert: Alert{ID: "second-tick-alert"}, Timestamp: time.Now(),
|
||||
})
|
||||
hm.mu.Unlock()
|
||||
time.Sleep(hm.saveInterval)
|
||||
synctest.Wait()
|
||||
assertPersisted("periodic-save-alert", "second-tick-alert")
|
||||
})
|
||||
}
|
||||
|
||||
@@ -2,11 +2,13 @@ package alerts
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts/reducer"
|
||||
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
||||
)
|
||||
|
||||
@@ -262,3 +264,173 @@ func TestAutoAcknowledgementSurvivesMetricEvaluation(t *testing.T) {
|
||||
evaluate(90)
|
||||
assertAcknowledged("short recovery and refire")
|
||||
}
|
||||
|
||||
// Age only the acknowledgement timestamps, keeping detector observations fresh.
|
||||
// The hourly sweep must not discard a long-running operator decision before a
|
||||
// real recovery can start the existing inactive-retention window.
|
||||
func TestTrackingCleanupPreservesAcknowledgedProviderRecurrence(t *testing.T) {
|
||||
for _, durable := range []bool{false, true} {
|
||||
for _, automatic := range []bool{false, true} {
|
||||
name := "JSON"
|
||||
if durable {
|
||||
name = "durable"
|
||||
}
|
||||
if automatic {
|
||||
name += "/automatic"
|
||||
} else {
|
||||
name += "/manual"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
newManager := func() *Manager {
|
||||
return newCleanupRetentionManager(t, dir, durable)
|
||||
}
|
||||
m := newManager()
|
||||
resource := unifiedresources.Resource{
|
||||
ID: "storage:ack-tank", Type: unifiedresources.ResourceTypeStorage,
|
||||
Name: "ack-tank", ParentName: "truenas-main",
|
||||
Sources: []unifiedresources.DataSource{unifiedresources.SourceTrueNAS},
|
||||
Storage: &unifiedresources.StorageMeta{Platform: "truenas", Topology: "pool", Protection: "zfs", IsZFS: true},
|
||||
Incidents: []unifiedresources.ResourceIncident{{
|
||||
Provider: "truenas", NativeID: "ack-pool-alert", Code: "truenas_volume_status",
|
||||
Severity: storagehealth.RiskWarning, Summary: "Pool ack-tank is DEGRADED",
|
||||
StartedAt: time.Now().Add(-26 * time.Hour),
|
||||
}},
|
||||
}
|
||||
healthy := resource
|
||||
healthy.Incidents = nil
|
||||
evaluate := func(m *Manager, observed unifiedresources.Resource) {
|
||||
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{observed})
|
||||
}
|
||||
var deliveries atomic.Int32
|
||||
m.SetAlertCallback(func(*Alert) { deliveries.Add(1) })
|
||||
evaluate(m, resource)
|
||||
initial := m.GetActiveAlerts()
|
||||
if len(initial) != 1 || deliveries.Load() != 1 {
|
||||
t.Fatalf("initial provider condition: %d alerts / %d dispatches, want 1 / 1", len(initial), deliveries.Load())
|
||||
}
|
||||
id := initial[0].ID
|
||||
user := "operator"
|
||||
if automatic {
|
||||
user = "system-auto"
|
||||
m.mu.Lock()
|
||||
m.config.AutoAcknowledgeAfterHours = 1
|
||||
m.mu.Unlock()
|
||||
m.Cleanup(time.Hour)
|
||||
} else if err := m.AcknowledgeAlert(id, user); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ackAt := time.Now().Add(-25 * time.Hour)
|
||||
m.mu.Lock()
|
||||
active, ok := m.getActiveAlertNoLock(id)
|
||||
if !ok || !active.Acknowledged {
|
||||
m.mu.Unlock()
|
||||
t.Fatal("detector acknowledgement was not established")
|
||||
}
|
||||
active.AckTime = &ackAt
|
||||
record := m.ackStateByCanonical[id]
|
||||
record.time = ackAt
|
||||
m.ackStateByCanonical[id] = record
|
||||
m.mirrorAcknowledgeNoLock(active, user, ackAt)
|
||||
m.setActiveAlertNoLock(id, active)
|
||||
m.mu.Unlock()
|
||||
|
||||
checkCycle := func(m *Manager, count *atomic.Int32, stage string) {
|
||||
t.Helper()
|
||||
before := count.Load()
|
||||
m.cleanupStaleMaps()
|
||||
m.mu.RLock()
|
||||
record, retained := m.ackStateByCanonical[id]
|
||||
m.mu.RUnlock()
|
||||
if !retained || record.user != user || !record.time.Equal(ackAt) {
|
||||
t.Errorf("%s: hourly sweep lost the active acknowledgement: %+v, retained=%v", stage, record, retained)
|
||||
}
|
||||
evaluate(m, healthy) // Observed pool with no incident: affirmative recovery.
|
||||
if got := len(m.GetActiveAlerts()); got != 0 {
|
||||
t.Fatalf("%s: real recovery left %d active alerts", stage, got)
|
||||
}
|
||||
m.Cleanup(time.Hour) // Old acknowledgement, newly inactive: retain it.
|
||||
evaluate(m, resource)
|
||||
alerts := m.GetActiveAlerts()
|
||||
if len(alerts) != 1 || alerts[0].ID != id || !alerts[0].Acknowledged || alerts[0].AckUser != user || alerts[0].AckTime == nil || !alerts[0].AckTime.Equal(ackAt) {
|
||||
t.Errorf("%s: short recurrence lost acknowledgement identity: %+v", stage, alerts)
|
||||
}
|
||||
if got := count.Load() - before; got != 0 {
|
||||
t.Errorf("%s: acknowledged recovery/recurrence dispatched %d duplicate notifications", stage, got)
|
||||
}
|
||||
t.Logf("%s: recovery/recurrence dispatch delta %d", stage, count.Load()-before)
|
||||
}
|
||||
checkCycle(m, &deliveries, "before restart")
|
||||
m.Stop() // The actual final checkpoint, not a hand-built JSON fixture.
|
||||
restarted := newManager()
|
||||
var afterRestart atomic.Int32
|
||||
restarted.SetAlertCallback(func(*Alert) { afterRestart.Add(1) })
|
||||
checkCycle(restarted, &afterRestart, "after restart")
|
||||
if err := restarted.UnacknowledgeAlert(id); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := afterRestart.Load()
|
||||
evaluate(restarted, healthy)
|
||||
evaluate(restarted, resource)
|
||||
if got := afterRestart.Load() - before; got != 1 {
|
||||
t.Errorf("explicit unacknowledge/recovery/recurrence dispatched %d notifications, want 1", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrackingCleanupCanonicalAckRetentionBounds(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
active bool
|
||||
legacy bool
|
||||
inactive time.Duration
|
||||
wantHourly bool
|
||||
wantNormal bool
|
||||
}{
|
||||
{name: "active canonical", active: true, wantHourly: true, wantNormal: true},
|
||||
{name: "active legacy identity", active: true, legacy: true, wantHourly: true, wantNormal: true},
|
||||
{name: "active after earlier recurrence", active: true, inactive: 25 * time.Hour, wantHourly: true, wantNormal: true},
|
||||
{name: "recently inactive old acknowledgement", inactive: 30 * time.Minute, wantHourly: true, wantNormal: true},
|
||||
{name: "existing one-hour expiry", inactive: 2 * time.Hour, wantHourly: true},
|
||||
{name: "stale inactive", inactive: 25 * time.Hour},
|
||||
{name: "legacy missing inactive timestamp"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
m := newTestManager(t)
|
||||
now := time.Now()
|
||||
old := now.Add(-25 * time.Hour)
|
||||
id, alert := testNewCanonicalAlert("ack-bounds", "metric-threshold:cpu", "metric-threshold", "cpu")
|
||||
alert.StartTime, alert.LastSeen = old, now
|
||||
alert.Acknowledged, alert.AckUser, alert.AckTime = true, "operator", &old
|
||||
record := ackRecord{acknowledged: true, user: "operator", time: old}
|
||||
if tc.inactive > 0 {
|
||||
record.inactiveAt = now.Add(-tc.inactive)
|
||||
}
|
||||
m.mu.Lock()
|
||||
m.ackStateByCanonical[id] = record
|
||||
if tc.active {
|
||||
storageKey := id
|
||||
if tc.legacy {
|
||||
storageKey = "legacy-cpu-ack-bounds"
|
||||
}
|
||||
m.activeAlerts[storageKey] = alert
|
||||
}
|
||||
m.mu.Unlock()
|
||||
assertRetained := func(stage string, want bool) {
|
||||
t.Helper()
|
||||
m.mu.RLock()
|
||||
got, retained := m.ackStateByCanonical[id]
|
||||
m.mu.RUnlock()
|
||||
if retained != want || (retained && got != record) {
|
||||
t.Errorf("%s: acknowledgement retained=%v (%+v), want %v with unchanged record", stage, retained, got, want)
|
||||
}
|
||||
}
|
||||
m.cleanupStaleMaps()
|
||||
assertRetained("hourly sweep", tc.wantHourly)
|
||||
m.Cleanup(time.Hour)
|
||||
assertRetained("normal one-hour inactive expiry", tc.wantNormal)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -139,6 +139,11 @@ func (m *Manager) cleanupStaleMaps() {
|
||||
}
|
||||
}
|
||||
for canonicalID, record := range m.ackStateByCanonical {
|
||||
// As with legacy records and Cleanup, an active incident owns its
|
||||
// acknowledgement until recovery starts the inactive-retention window.
|
||||
if m.hasActiveAlertTrackingKeyNoLock(canonicalID) {
|
||||
continue
|
||||
}
|
||||
checkTime := record.inactiveAt
|
||||
if checkTime.IsZero() {
|
||||
checkTime = record.time
|
||||
|
||||
Reference in New Issue
Block a user