mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-26 21:47:20 +00:00
Integrate reviewed retained-history replay CPU correction
Accept exact Core commits after reviewing the write-only materialization path and retained normal/race, build, and ordinary-child evidence. Preserve canonical user reads and checkpoint/retention behavior. This reduces controlled startup CPU, not RSS or the reported memory baseline. Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
@@ -534,9 +534,26 @@ func (s *IncidentStore) RecordAlertResolved(alert *alerts.Alert, resolvedAt time
|
||||
// canonical events exist, QueryIncidents replaces these snapshot-derived
|
||||
// lifecycle breadcrumbs with the durable projection.
|
||||
func (s *IncidentStore) EnsureAlertOccurrence(alert *alerts.Alert, resolvedAt *time.Time) *Incident {
|
||||
if s == nil || alert == nil || strings.TrimSpace(alert.ID) == "" {
|
||||
if !s.materializeAlertOccurrence(alert, resolvedAt) {
|
||||
return nil
|
||||
}
|
||||
return s.GetTimelineByAlertAt(alert.ID, alert.StartTime)
|
||||
}
|
||||
|
||||
// MaterializeAlertOccurrence repairs the same snapshot-derived occurrence as
|
||||
// EnsureAlertOccurrence without reading its projected timeline. Lifecycle
|
||||
// replay writes canonical changes separately and discards the returned
|
||||
// timeline, so hydrating it for every imported snapshot only adds history
|
||||
// reads and copies. Callers serving a timeline must use EnsureAlertOccurrence
|
||||
// or QueryIncidents, retaining canonical evidence and read-error semantics.
|
||||
func (s *IncidentStore) MaterializeAlertOccurrence(alert *alerts.Alert, resolvedAt *time.Time) {
|
||||
s.materializeAlertOccurrence(alert, resolvedAt)
|
||||
}
|
||||
|
||||
func (s *IncidentStore) materializeAlertOccurrence(alert *alerts.Alert, resolvedAt *time.Time) bool {
|
||||
if s == nil || alert == nil || strings.TrimSpace(alert.ID) == "" {
|
||||
return false
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
shell := s.findIncidentByAlertAtLocked(alert.ID, alert.StartTime)
|
||||
@@ -589,7 +606,7 @@ func (s *IncidentStore) EnsureAlertOccurrence(alert *alerts.Alert, resolvedAt *t
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
return s.GetTimelineByAlertAt(alert.ID, alert.StartTime)
|
||||
return true
|
||||
}
|
||||
|
||||
// RecordAnalysis adds an AI analysis event to the incident for an alert.
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestIncidentMaterializationPreservesSnapshotCheckpointAndReadRepair(t *testing.T) {
|
||||
for _, canonical := range []bool{false, true} {
|
||||
name := "local"
|
||||
if canonical {
|
||||
name = "canonical"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
cfg := IncidentStoreConfig{DataDir: t.TempDir(), MaxIncidents: 1, MaxAgeDays: 30}
|
||||
store := NewIncidentStore(cfg)
|
||||
t.Cleanup(store.flush)
|
||||
var history *unifiedresources.MemoryStore
|
||||
if canonical {
|
||||
history = unifiedresources.NewMemoryStore()
|
||||
store.SetResourceTimelineStore(history)
|
||||
}
|
||||
start := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
|
||||
ack, resolved := start.Add(time.Minute), start.Add(2*time.Minute)
|
||||
alert := &alerts.Alert{ID: "snapshot", ResourceID: "resource", Type: "cpu", StartTime: start,
|
||||
LastSeen: resolved, Acknowledged: true, AckTime: &ack, AckUser: "operator"}
|
||||
store.MaterializeAlertOccurrence(alert, &resolved)
|
||||
store.flush()
|
||||
first := store.GetTimelineByAlertAt(alert.ID, start)
|
||||
require.NotNil(t, first)
|
||||
require.Len(t, first.Events, 3)
|
||||
require.Equal(t, IncidentStatusResolved, first.Status)
|
||||
require.True(t, first.Acknowledged)
|
||||
require.Equal(t, alert.AckUser, first.AckUser)
|
||||
require.Equal(t, &resolved, first.ClosedAt)
|
||||
before, err := os.Stat(store.filePath)
|
||||
require.NoError(t, err)
|
||||
for i := 0; i < 8; i++ {
|
||||
store.MaterializeAlertOccurrence(alert, &resolved)
|
||||
}
|
||||
store.flush()
|
||||
after, err := os.Stat(store.filePath)
|
||||
require.NoError(t, err)
|
||||
require.True(t, os.SameFile(before, after), "unchanged materialization must not rewrite its checkpoint")
|
||||
reloaded := NewIncidentStore(cfg)
|
||||
t.Cleanup(reloaded.flush)
|
||||
if canonical {
|
||||
reloaded.SetResourceTimelineStore(history)
|
||||
}
|
||||
restored := reloaded.GetTimelineByAlertAt(alert.ID, start)
|
||||
require.NotNil(t, restored)
|
||||
require.Equal(t, first.ID, restored.ID)
|
||||
require.Equal(t, first.Events, restored.Events)
|
||||
require.Equal(t, first.OpenedAt, restored.OpenedAt)
|
||||
require.Equal(t, first.ClosedAt, restored.ClosedAt)
|
||||
require.Equal(t, first.Status, restored.Status)
|
||||
require.Equal(t, first.Acknowledged, restored.Acknowledged)
|
||||
require.Equal(t, first.AckUser, restored.AckUser)
|
||||
require.Equal(t, first.AckTime, restored.AckTime)
|
||||
// Query coverage windows belong to each read, not the checkpoint.
|
||||
require.Equal(t, first.History.Source, restored.History.Source)
|
||||
|
||||
// Eviction is still bounded and the existing read-facing method can
|
||||
// repair a requested retained occurrence beyond that cache.
|
||||
newer := &alerts.Alert{ID: "newer", ResourceID: "other-resource", StartTime: start.Add(10 * time.Minute)}
|
||||
reloaded.MaterializeAlertOccurrence(newer, nil)
|
||||
require.Len(t, reloaded.incidents, 1)
|
||||
require.Equal(t, newer.ID, reloaded.incidents[0].AlertIdentifier)
|
||||
repaired := reloaded.EnsureAlertOccurrence(alert, &resolved)
|
||||
require.NotNil(t, repaired)
|
||||
require.Len(t, repaired.Events, 3)
|
||||
require.Len(t, reloaded.incidents, 1)
|
||||
require.Equal(t, alert.ID, reloaded.incidents[0].AlertIdentifier)
|
||||
|
||||
// Age retention is not replaced by count-only retention.
|
||||
expired := &alerts.Alert{ID: "expired", StartTime: start.Add(-45 * 24 * time.Hour)}
|
||||
expiredEnd := expired.StartTime.Add(time.Minute)
|
||||
reloaded.MaterializeAlertOccurrence(expired, &expiredEnd)
|
||||
require.Len(t, reloaded.incidents, 1)
|
||||
require.Equal(t, alert.ID, reloaded.incidents[0].AlertIdentifier)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestIncidentMaterializationDoesNotMaskReadFailure(t *testing.T) {
|
||||
store := NewIncidentStore(IncidentStoreConfig{DataDir: t.TempDir()})
|
||||
t.Cleanup(store.flush)
|
||||
store.SetResourceTimelineStore(failingIncidentHistory{unifiedresources.NewMemoryStore()})
|
||||
alert := &alerts.Alert{ID: "read-failure", ResourceID: "resource", StartTime: time.Now().UTC().Add(-time.Minute)}
|
||||
store.MaterializeAlertOccurrence(alert, nil)
|
||||
store.flush()
|
||||
require.Len(t, store.incidents, 1, "the writer still creates the recovery shell")
|
||||
_, err := store.QueryIncidents(IncidentQuery{AlertIdentifier: alert.ID})
|
||||
require.ErrorContains(t, err, "history read denied", "an actual read must still expose unavailable canonical history")
|
||||
require.Nil(t, store.EnsureAlertOccurrence(alert, nil))
|
||||
store.SetResourceTimelineStore(unifiedresources.NewMemoryStore())
|
||||
repaired := store.EnsureAlertOccurrence(alert, nil)
|
||||
require.NotNil(t, repaired)
|
||||
require.Len(t, repaired.Events, 1)
|
||||
}
|
||||
|
||||
func TestIncidentMaterializationRejectsInvalidSnapshots(t *testing.T) {
|
||||
var missing *IncidentStore
|
||||
missing.MaterializeAlertOccurrence(&alerts.Alert{ID: "valid"}, nil)
|
||||
require.Nil(t, missing.EnsureAlertOccurrence(&alerts.Alert{ID: "valid"}, nil))
|
||||
store := NewIncidentStore(IncidentStoreConfig{})
|
||||
for _, alert := range []*alerts.Alert{nil, {}, {ID: " \t"}} {
|
||||
store.MaterializeAlertOccurrence(alert, nil)
|
||||
require.Nil(t, store.EnsureAlertOccurrence(alert, nil))
|
||||
}
|
||||
require.Empty(t, store.incidents)
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
package monitoring
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/ai/memory"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// Count only the public history reads, not store implementation details. Replay
|
||||
// writes the occurrence shell and canonical changes; it has no returned
|
||||
// timeline to serve and must not hydrate one for each imported snapshot.
|
||||
type replayHistoryReadCounter struct {
|
||||
memory.IncidentTimelineStore
|
||||
filteredReads atomic.Int64
|
||||
identityReads atomic.Int64
|
||||
recentReads atomic.Int64
|
||||
}
|
||||
|
||||
func (s *replayHistoryReadCounter) GetRecentChangesFiltered(id string, since time.Time, limit int, filters unifiedresources.ResourceChangeFilters) ([]unifiedresources.ResourceChange, error) {
|
||||
s.filteredReads.Add(1)
|
||||
return s.IncidentTimelineStore.GetRecentChangesFiltered(id, since, limit, filters)
|
||||
}
|
||||
|
||||
func (s *replayHistoryReadCounter) ResourceHistoryIDs(id string) ([]string, error) {
|
||||
s.identityReads.Add(1)
|
||||
return s.IncidentTimelineStore.ResourceHistoryIDs(id)
|
||||
}
|
||||
|
||||
func (s *replayHistoryReadCounter) GetRecentChanges(id string, since time.Time, limit int) ([]unifiedresources.ResourceChange, error) {
|
||||
s.recentReads.Add(1)
|
||||
return s.IncidentTimelineStore.GetRecentChanges(id, since, limit)
|
||||
}
|
||||
|
||||
func importReplaySnapshots(t *testing.T, manager *alerts.Manager, snapshots []alerts.Alert) {
|
||||
t.Helper()
|
||||
store, err := eventlog.Open(t.TempDir())
|
||||
require.NoError(t, err)
|
||||
manager.SetEventLog(store) // the manager owns Close
|
||||
events := make([]eventlog.Event, len(snapshots))
|
||||
for i, alert := range snapshots {
|
||||
payload, err := json.Marshal(alert)
|
||||
require.NoError(t, err)
|
||||
events[i] = eventlog.Event{Type: eventlog.TypeHistoryImported, AlertID: alert.ID,
|
||||
ResourceID: alert.ResourceID, OccurredAt: alert.LastSeen, Snapshot: payload}
|
||||
}
|
||||
require.NoError(t, store.ImportEvents(events))
|
||||
}
|
||||
|
||||
func TestImportedTimelineReplayDoesNotReadDiscardedProjections(t *testing.T) {
|
||||
manager := alerts.NewManagerWithDataDir(t.TempDir(), alerts.WithoutPersistedAlertRestore())
|
||||
t.Cleanup(manager.Stop)
|
||||
start := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
|
||||
snapshots := make([]alerts.Alert, 8)
|
||||
for i := range snapshots {
|
||||
opened := start.Add(time.Duration(i) * time.Minute)
|
||||
ack := opened.Add(time.Second)
|
||||
snapshots[i] = alerts.Alert{ID: fmt.Sprintf("import-%d", i), ResourceID: fmt.Sprintf("resource-%d", i),
|
||||
StartTime: opened, LastSeen: opened.Add(2 * time.Second), Type: "cpu", Level: alerts.AlertLevelWarning,
|
||||
Acknowledged: true, AckTime: &ack, AckUser: "synthetic-operator"}
|
||||
}
|
||||
importReplaySnapshots(t, manager, snapshots)
|
||||
store := unifiedresources.NewMemoryStore()
|
||||
adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store))
|
||||
history := &replayHistoryReadCounter{IncidentTimelineStore: adapter}
|
||||
incidents := memory.NewIncidentStore(memory.IncidentStoreConfig{})
|
||||
incidents.SetResourceTimelineStore(history)
|
||||
m := &Monitor{alertManager: manager, resourceStore: adapter, incidentStore: incidents}
|
||||
m.replayAlertLifecycleProjections()
|
||||
m.replayAlertLifecycleProjections() // the watermark must make this a no-op
|
||||
reads := history.filteredReads.Load() + history.identityReads.Load() + history.recentReads.Load()
|
||||
t.Logf("imported=%d discarded-projection reads: filtered=%d identities=%d recent=%d", len(snapshots),
|
||||
history.filteredReads.Load(), history.identityReads.Load(), history.recentReads.Load())
|
||||
|
||||
// Explicit user reads still hydrate canonical timelines, including with a
|
||||
// completely missing incident checkpoint. A writer optimisation must not
|
||||
// remove these reads or turn recovered evidence into empty history.
|
||||
recovered := memory.NewIncidentStore(memory.IncidentStoreConfig{})
|
||||
recovered.SetResourceTimelineStore(adapter)
|
||||
for _, alert := range snapshots {
|
||||
page, err := recovered.QueryIncidents(memory.IncidentQuery{AlertIdentifier: alert.ID, StartedAt: alert.StartTime})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, page.Incidents, 1)
|
||||
incident := page.Incidents[0]
|
||||
require.Equal(t, memory.IncidentStatusResolved, incident.Status)
|
||||
require.Equal(t, &alert.LastSeen, incident.ClosedAt)
|
||||
require.True(t, incident.Acknowledged)
|
||||
require.Equal(t, alert.AckUser, incident.AckUser)
|
||||
require.Len(t, incident.Events, 3)
|
||||
require.Equal(t, "canonical_resource_history", page.History.Source)
|
||||
changes, err := store.GetRecentChanges(alert.ResourceID, time.Time{}, 10)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, changes, 3, "replay must remain idempotent")
|
||||
}
|
||||
require.Zero(t, reads, "replay must not query a projection it discards")
|
||||
}
|
||||
|
||||
type replayCheckpointRecorder struct {
|
||||
*unifiedresources.MonitorAdapter
|
||||
manager *alerts.Manager
|
||||
imports int
|
||||
checkpoint int64
|
||||
}
|
||||
|
||||
func (s *replayCheckpointRecorder) RecordChange(change unifiedresources.ResourceChange) error {
|
||||
if change.Kind == unifiedresources.ChangeAlertFired {
|
||||
if s.imports == alertProjectionCheckpointEvery {
|
||||
s.checkpoint = s.manager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer)
|
||||
}
|
||||
s.imports++
|
||||
}
|
||||
return s.MonitorAdapter.RecordChange(change)
|
||||
}
|
||||
|
||||
func TestImportedTimelineReplayCheckpointAndCacheRecovery(t *testing.T) {
|
||||
manager := alerts.NewManagerWithDataDir(t.TempDir(), alerts.WithoutPersistedAlertRestore())
|
||||
t.Cleanup(manager.Stop)
|
||||
start := time.Now().UTC().Add(-2 * time.Hour).Truncate(time.Second)
|
||||
snapshots := make([]alerts.Alert, alertProjectionCheckpointEvery+1)
|
||||
for i := range snapshots {
|
||||
opened := start.Add(time.Duration(i) * time.Second)
|
||||
snapshots[i] = alerts.Alert{ID: fmt.Sprintf("checkpoint-%d", i), ResourceID: "shared-resource",
|
||||
StartTime: opened, LastSeen: opened.Add(time.Second), Type: "cpu", Level: alerts.AlertLevelWarning}
|
||||
}
|
||||
importReplaySnapshots(t, manager, snapshots)
|
||||
store := unifiedresources.NewMemoryStore()
|
||||
adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store))
|
||||
recorder := &replayCheckpointRecorder{MonitorAdapter: adapter, manager: manager}
|
||||
incidents := memory.NewIncidentStore(memory.IncidentStoreConfig{MaxIncidents: 2})
|
||||
incidents.SetResourceTimelineStore(adapter)
|
||||
m := &Monitor{alertManager: manager, resourceStore: recorder, incidentStore: incidents}
|
||||
m.replayAlertLifecycleProjections()
|
||||
require.Equal(t, len(snapshots), recorder.imports)
|
||||
require.Positive(t, recorder.checkpoint, "checkpoint must persist before the next import")
|
||||
complete := manager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer)
|
||||
require.Greater(t, complete, recorder.checkpoint)
|
||||
|
||||
// Resume from the actually captured mid-pass checkpoint with a new bounded
|
||||
// cache. Only the remaining event is visited and its canonical rows dedupe.
|
||||
manager.StoreLifecycleProjectionWatermark(alertLifecycleProjectionConsumer, recorder.checkpoint)
|
||||
repaired := memory.NewIncidentStore(memory.IncidentStoreConfig{MaxIncidents: 2})
|
||||
repaired.SetResourceTimelineStore(adapter)
|
||||
resume := &Monitor{alertManager: manager, resourceStore: recorder, incidentStore: repaired}
|
||||
before := recorder.imports
|
||||
resume.replayAlertLifecycleProjections()
|
||||
require.Equal(t, before+1, recorder.imports)
|
||||
require.Equal(t, complete, manager.LifecycleProjectionWatermark(alertLifecycleProjectionConsumer))
|
||||
changes, err := store.GetRecentChanges("shared-resource", time.Time{}, 2*len(snapshots)+1)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, changes, 2*len(snapshots))
|
||||
|
||||
// Both the evicted earliest occurrence and the resumed tail remain readable
|
||||
// from canonical evidence, not just from the two-entry snapshot cache.
|
||||
for _, i := range []int{0, len(snapshots) - 1} {
|
||||
alert := snapshots[i]
|
||||
page, err := repaired.QueryIncidents(memory.IncidentQuery{AlertIdentifier: alert.ID, StartedAt: alert.StartTime})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, page.Incidents, 1)
|
||||
require.Equal(t, memory.IncidentStatusResolved, page.Incidents[0].Status)
|
||||
require.Len(t, page.Incidents[0].Events, 2)
|
||||
}
|
||||
}
|
||||
@@ -415,7 +415,7 @@ func (m *Monitor) materializeImportedAlertTimeline(alert *alerts.Alert, imported
|
||||
}
|
||||
}
|
||||
|
||||
m.incidentStore.EnsureAlertOccurrence(alert, resolvedAt)
|
||||
m.incidentStore.MaterializeAlertOccurrence(alert, resolvedAt)
|
||||
firedAt := alert.StartTime
|
||||
if firedAt.IsZero() {
|
||||
firedAt = importedAt
|
||||
|
||||
Reference in New Issue
Block a user