mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-19 10:33:29 +00:00
feat(alerts): re-fire start restoration in the reducer with parity harness
Phase 1 slice 4 of docs/ALERT_ENGINE_EVOLUTION.md. The reducer now keeps a resolved-occurrence ledger: an activation inside RefireRetention (5 minutes, on the signal clock) consumes the record, restores the original occurrence's StartedAt, and emits EventRefired — mirroring consumeRecentlyResolvedForRefireWithPrimaryLock, where a re-fire within the recently-resolved window reactivates the same occurrence without a new history entry. Outside the window the re-fire is a fresh occurrence with a fresh start. reducer_parity_refire_test.go anchors the simulated epoch at wall time and backdates recentlyResolved timestamps per step so the manager's wall-clock retention check follows simulated time; StartTime is asserted exactly on both engines. First parity slice with no manager defect found. The wall-vs-evidence clock mix in the manager's retention check is recorded as a deliberately deferred nuance.
This commit is contained in:
parent
07aa2116d1
commit
3d48649867
5 changed files with 316 additions and 10 deletions
|
|
@ -58,6 +58,16 @@ confirmation counts directly (bypassing the evaluator path) left a stale
|
|||
previous run's first observation. Fixed by re-stamping whenever an
|
||||
observation starts a new run.
|
||||
|
||||
Remaining for later slices: re-fire-within-retention start restoration,
|
||||
ack/snooze lifecycle, intent-policy interaction, and a shadow-mode
|
||||
runtime feed.
|
||||
Fourth slice (2026-08-26): re-fire-within-retention start restoration —
|
||||
a resolve followed by a re-activation inside the recently-resolved
|
||||
retention (5 minutes) restores the original occurrence's StartTime and is
|
||||
treated as the same occurrence (EventRefired in the reducer; the manager
|
||||
reactivates without a new history entry); outside the window the re-fire
|
||||
is a fresh occurrence. First slice where parity passed with no manager
|
||||
defect found. Known un-characterized nuance, deliberately deferred: the
|
||||
manager's retention check mixes wall-clock now against evidence-stamped
|
||||
resolve times, which the reducer normalizes to ObservedAt throughout —
|
||||
that difference disappears at cutover.
|
||||
|
||||
Remaining for later slices: ack/snooze lifecycle, intent-policy
|
||||
interaction, and a shadow-mode runtime feed.
|
||||
|
|
|
|||
|
|
@ -45,6 +45,27 @@ type DiscreteRule struct {
|
|||
Disabled bool
|
||||
}
|
||||
|
||||
// recordResolved remembers a resolved occurrence for re-fire restoration.
|
||||
func (s *State) recordResolved(key string, incident *Incident, resolvedAt time.Time) {
|
||||
s.resolved[key] = resolvedRecord{StartedAt: incident.StartedAt, ResolvedAt: resolvedAt}
|
||||
}
|
||||
|
||||
// consumeRefire consumes a resolved occurrence still inside RefireRetention
|
||||
// and returns its start time for restoration. A record too old to
|
||||
// reactivate is dropped (the manager keeps it only for its history
|
||||
// bookkeeping, then prunes it on the same retention).
|
||||
func (s *State) consumeRefire(key string, observedAt time.Time) (time.Time, bool) {
|
||||
record, ok := s.resolved[key]
|
||||
if !ok {
|
||||
return time.Time{}, false
|
||||
}
|
||||
delete(s.resolved, key)
|
||||
if !record.ResolvedAt.After(observedAt.Add(-RefireRetention)) {
|
||||
return time.Time{}, false
|
||||
}
|
||||
return record.StartedAt, true
|
||||
}
|
||||
|
||||
// ApplyDiscrete advances the state for one discrete observation under one
|
||||
// rule and returns the transition events, in order. Deterministic: time
|
||||
// enters only through the signal's ObservedAt.
|
||||
|
|
@ -70,6 +91,7 @@ func (s *State) ApplyDiscrete(signal DiscreteSignal, rule DiscreteRule) []Event
|
|||
}
|
||||
delete(s.incidents, key)
|
||||
if incident.State == StateFiring {
|
||||
s.recordResolved(key, incident, signal.ObservedAt)
|
||||
return []Event{event(EventResolved, incident.Severity)}
|
||||
}
|
||||
return []Event{event(EventPendingCleared, "")}
|
||||
|
|
@ -90,6 +112,7 @@ func (s *State) ApplyDiscrete(signal DiscreteSignal, rule DiscreteRule) []Event
|
|||
requiredRecovery := rule.RecoveryConfirmations
|
||||
if requiredRecovery <= 1 {
|
||||
delete(s.incidents, key)
|
||||
s.recordResolved(key, incident, signal.ObservedAt)
|
||||
return []Event{event(EventResolved, incident.Severity)}
|
||||
}
|
||||
incident.RecoveryCount++
|
||||
|
|
@ -98,6 +121,7 @@ func (s *State) ApplyDiscrete(signal DiscreteSignal, rule DiscreteRule) []Event
|
|||
return nil
|
||||
}
|
||||
delete(s.incidents, key)
|
||||
s.recordResolved(key, incident, signal.ObservedAt)
|
||||
return []Event{event(EventResolved, incident.Severity)}
|
||||
}
|
||||
|
||||
|
|
@ -136,12 +160,16 @@ func (s *State) ApplyDiscrete(signal DiscreteSignal, rule DiscreteRule) []Event
|
|||
incident.State = StateFiring
|
||||
incident.Confirmations = required
|
||||
incident.StartedAt = incident.PendingSince
|
||||
if restored, ok := s.consumeRefire(key, signal.ObservedAt); ok {
|
||||
incident.StartedAt = restored
|
||||
return []Event{event(EventRefired, incident.Severity)}
|
||||
}
|
||||
return []Event{event(EventFired, incident.Severity)}
|
||||
}
|
||||
|
||||
// First matching observation.
|
||||
if required <= 1 {
|
||||
s.incidents[key] = &Incident{
|
||||
created := &Incident{
|
||||
ResourceID: signal.ResourceID,
|
||||
Key: signal.Key,
|
||||
State: StateFiring,
|
||||
|
|
@ -150,6 +178,11 @@ func (s *State) ApplyDiscrete(signal DiscreteSignal, rule DiscreteRule) []Event
|
|||
Confirmations: 1,
|
||||
LastObservedAt: signal.ObservedAt,
|
||||
}
|
||||
s.incidents[key] = created
|
||||
if restored, ok := s.consumeRefire(key, signal.ObservedAt); ok {
|
||||
created.StartedAt = restored
|
||||
return []Event{event(EventRefired, signal.Severity)}
|
||||
}
|
||||
return []Event{event(EventFired, signal.Severity)}
|
||||
}
|
||||
s.incidents[key] = &Incident{
|
||||
|
|
|
|||
|
|
@ -130,8 +130,13 @@ func TestDiscreteRefireRequiresFullConfirmations(t *testing.T) {
|
|||
t.Fatalf("events = %+v, want pending after recovery", events)
|
||||
}
|
||||
events = state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 4*time.Minute), rule)
|
||||
if len(events) != 1 || events[0].Type != EventFired {
|
||||
t.Fatalf("events = %+v, want re-fire after full confirmations", events)
|
||||
if len(events) != 1 || events[0].Type != EventRefired {
|
||||
t.Fatalf("events = %+v, want refired after full confirmations within retention", events)
|
||||
}
|
||||
// Within RefireRetention the original occurrence's start is restored.
|
||||
incident, _ := state.Incident("node-1", "connectivity")
|
||||
if !incident.StartedAt.Equal(t0) {
|
||||
t.Fatalf("StartedAt = %v, want restored original start %v", incident.StartedAt, t0)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -209,3 +214,50 @@ func TestDiscreteDisableBypassesRecoveryGate(t *testing.T) {
|
|||
t.Fatalf("events = %+v, want immediate resolve on disable", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDiscreteRefireOutsideRetentionStartsFresh(t *testing.T) {
|
||||
state := NewState()
|
||||
rule := DiscreteRule{Confirmations: 1}
|
||||
|
||||
state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule)
|
||||
state.ApplyDiscrete(discreteSignalAt(false, "", time.Minute), rule)
|
||||
|
||||
// Beyond RefireRetention the resolved occurrence is no longer
|
||||
// consumable: a re-fire is a new occurrence with a fresh start.
|
||||
refireAt := time.Minute + RefireRetention + time.Second
|
||||
events := state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, refireAt), rule)
|
||||
if len(events) != 1 || events[0].Type != EventFired {
|
||||
t.Fatalf("events = %+v, want plain fired outside retention", events)
|
||||
}
|
||||
incident, _ := state.Incident("node-1", "connectivity")
|
||||
if !incident.StartedAt.Equal(t0.Add(refireAt)) {
|
||||
t.Fatalf("StartedAt = %v, want fresh start %v", incident.StartedAt, t0.Add(refireAt))
|
||||
}
|
||||
}
|
||||
|
||||
func TestDiscreteRefireRecordConsumedOnce(t *testing.T) {
|
||||
state := NewState()
|
||||
rule := DiscreteRule{Confirmations: 1}
|
||||
|
||||
state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule)
|
||||
state.ApplyDiscrete(discreteSignalAt(false, "", time.Minute), rule)
|
||||
|
||||
// First re-fire restores; its own later resolve records a NEW
|
||||
// occurrence whose start is the restored (original) start.
|
||||
events := state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 2*time.Minute), rule)
|
||||
if len(events) != 1 || events[0].Type != EventRefired {
|
||||
t.Fatalf("events = %+v, want refired", events)
|
||||
}
|
||||
state.ApplyDiscrete(discreteSignalAt(false, "", 3*time.Minute), rule)
|
||||
|
||||
// Second re-fire within retention restores again — from the second
|
||||
// occurrence's record, which carries the same original start.
|
||||
events = state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 4*time.Minute), rule)
|
||||
if len(events) != 1 || events[0].Type != EventRefired {
|
||||
t.Fatalf("events = %+v, want refired from the newer record", events)
|
||||
}
|
||||
incident, _ := state.Incident("node-1", "connectivity")
|
||||
if !incident.StartedAt.Equal(t0) {
|
||||
t.Fatalf("StartedAt = %v, want original start carried through %v", incident.StartedAt, t0)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -87,13 +87,24 @@ type Incident struct {
|
|||
type EventType string
|
||||
|
||||
const (
|
||||
EventPending EventType = "pending"
|
||||
EventPendingCleared EventType = "pending_cleared"
|
||||
EventFired EventType = "fired"
|
||||
EventPending EventType = "pending"
|
||||
EventPendingCleared EventType = "pending_cleared"
|
||||
EventFired EventType = "fired"
|
||||
// EventRefired marks an activation that consumed a recently resolved
|
||||
// occurrence within RefireRetention: the incident keeps the original
|
||||
// occurrence's StartedAt, and downstream consumers treat it as the
|
||||
// same occurrence (the manager reactivates without a new history
|
||||
// entry, while still notifying).
|
||||
EventRefired EventType = "refired"
|
||||
EventSeverityChanged EventType = "severity_changed"
|
||||
EventResolved EventType = "resolved"
|
||||
)
|
||||
|
||||
// RefireRetention is how long a resolved occurrence remains consumable by a
|
||||
// re-fire, mirroring the manager's recentlyResolvedRetention. Time is
|
||||
// measured on the signal's ObservedAt.
|
||||
const RefireRetention = 5 * time.Minute
|
||||
|
||||
// Event is one transition emitted by Apply.
|
||||
type Event struct {
|
||||
Type EventType
|
||||
|
|
@ -104,15 +115,27 @@ type Event struct {
|
|||
At time.Time
|
||||
}
|
||||
|
||||
// resolvedRecord remembers one resolved occurrence so a re-fire inside
|
||||
// RefireRetention can restore its start time (the confirmation family's
|
||||
// canonical lifecycle behavior).
|
||||
type resolvedRecord struct {
|
||||
StartedAt time.Time
|
||||
ResolvedAt time.Time
|
||||
}
|
||||
|
||||
// State holds every incident the reducer tracks. It is not safe for
|
||||
// concurrent use; the owner serializes Apply calls.
|
||||
type State struct {
|
||||
incidents map[string]*Incident
|
||||
resolved map[string]resolvedRecord
|
||||
}
|
||||
|
||||
// NewState returns an empty reducer state.
|
||||
func NewState() *State {
|
||||
return &State{incidents: make(map[string]*Incident)}
|
||||
return &State{
|
||||
incidents: make(map[string]*Incident),
|
||||
resolved: make(map[string]resolvedRecord),
|
||||
}
|
||||
}
|
||||
|
||||
func incidentKey(resourceID, subKey string) string {
|
||||
|
|
|
|||
188
internal/alerts/reducer_parity_refire_test.go
Normal file
188
internal/alerts/reducer_parity_refire_test.go
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
package alerts
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"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/unifiedresources"
|
||||
)
|
||||
|
||||
// Re-fire parity: a resolve followed by a re-fire inside the
|
||||
// recently-resolved retention window restores the ORIGINAL occurrence's
|
||||
// StartTime (consumeRecentlyResolvedForRefireWithPrimaryLock); outside the
|
||||
// window the re-fire is a fresh occurrence. The manager's retention check
|
||||
// compares wall-clock now against ResolvedTime, so the harness anchors the
|
||||
// simulated epoch at wall time and backdates recentlyResolved entries by
|
||||
// each step's advance — the same trick the metric harness uses for
|
||||
// pendingAlerts — making retention behave per simulated time while
|
||||
// StartTime assertions stay exact.
|
||||
|
||||
type refireParityStep struct {
|
||||
matched bool
|
||||
advance time.Duration
|
||||
|
||||
wantFiring bool
|
||||
wantStartOffset time.Duration // -1 skips
|
||||
}
|
||||
|
||||
type refireParityScenario struct {
|
||||
name string
|
||||
confirmations int
|
||||
steps []refireParityStep
|
||||
}
|
||||
|
||||
const refireParityResourceID = "parity-refire-1"
|
||||
|
||||
type managerRefireParityEngine struct {
|
||||
manager *Manager
|
||||
clock time.Time
|
||||
conf int
|
||||
}
|
||||
|
||||
func (e *managerRefireParityEngine) observe(t *testing.T, step refireParityStep) {
|
||||
t.Helper()
|
||||
e.clock = e.clock.Add(step.advance)
|
||||
|
||||
if step.advance > 0 {
|
||||
// Shift resolved timestamps back so the wall-clock retention check
|
||||
// sees the simulated gap.
|
||||
e.manager.resolvedMutex.Lock()
|
||||
for _, resolved := range e.manager.recentlyResolved {
|
||||
if resolved != nil {
|
||||
resolved.ResolvedTime = resolved.ResolvedTime.Add(-step.advance)
|
||||
}
|
||||
}
|
||||
e.manager.resolvedMutex.Unlock()
|
||||
}
|
||||
|
||||
spec, err := buildCanonicalDiscreteStateSpec(
|
||||
refireParityResourceID, "parity-refire", unifiedresources.ResourceTypeAgent,
|
||||
AlertLevelWarning, e.conf, false, "parity-state", []string{"bad"},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("build spec: %v", err)
|
||||
}
|
||||
observed := "ok"
|
||||
if step.matched {
|
||||
observed = "bad"
|
||||
}
|
||||
if _, ok := e.manager.evaluateCanonicalLifecycleAlert(canonicalLifecycleAlertParams{
|
||||
Spec: spec,
|
||||
Evidence: alertspecs.AlertEvidence{
|
||||
ObservedAt: e.clock,
|
||||
DiscreteState: &alertspecs.DiscreteStateEvidence{StateKey: "parity-state", Observed: observed},
|
||||
},
|
||||
Tracking: e.manager.offlineConfirmations,
|
||||
TrackingKey: "parity:" + refireParityResourceID,
|
||||
AlertID: canonicalDiscreteStateStateID(refireParityResourceID, "parity-state"),
|
||||
AlertType: "parity-state",
|
||||
ResourceID: refireParityResourceID,
|
||||
ResourceName: "parity-refire",
|
||||
Message: "parity refire condition",
|
||||
}); !ok {
|
||||
t.Fatal("evaluateCanonicalLifecycleAlert rejected the parity spec")
|
||||
}
|
||||
}
|
||||
|
||||
func refireParityScenarios() []refireParityScenario {
|
||||
skip := time.Duration(-1)
|
||||
return []refireParityScenario{
|
||||
{
|
||||
name: "re-fire within retention restores the original start",
|
||||
confirmations: 1,
|
||||
steps: []refireParityStep{
|
||||
{matched: true, wantFiring: true, wantStartOffset: 0},
|
||||
{matched: false, advance: time.Minute, wantFiring: false, wantStartOffset: skip},
|
||||
{matched: true, advance: time.Minute, wantFiring: true, wantStartOffset: 0},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "confirmation runs also restore across a short resolve gap",
|
||||
confirmations: 2,
|
||||
steps: []refireParityStep{
|
||||
{matched: true, wantFiring: false, wantStartOffset: skip},
|
||||
{matched: true, advance: 30 * time.Second, wantFiring: true, wantStartOffset: 0},
|
||||
{matched: false, advance: 30 * time.Second, wantFiring: false, wantStartOffset: skip},
|
||||
{matched: true, advance: 30 * time.Second, wantFiring: false, wantStartOffset: skip},
|
||||
{matched: true, advance: 30 * time.Second, wantFiring: true, wantStartOffset: 0},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "re-fire outside retention is a fresh occurrence",
|
||||
confirmations: 1,
|
||||
steps: []refireParityStep{
|
||||
{matched: true, wantFiring: true, wantStartOffset: 0},
|
||||
{matched: false, advance: time.Minute, wantFiring: false, wantStartOffset: skip},
|
||||
{matched: true, advance: 6 * time.Minute, wantFiring: true, wantStartOffset: 7 * time.Minute},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestReducerParityWithManagerRefire(t *testing.T) {
|
||||
for _, scenario := range refireParityScenarios() {
|
||||
scenario := scenario
|
||||
t.Run(scenario.name, func(t *testing.T) {
|
||||
// Anchor at wall time so the manager's wall-clock retention
|
||||
// check and the simulated evidence clock agree.
|
||||
epoch := time.Now().UTC().Truncate(time.Second)
|
||||
|
||||
manager := NewManagerWithDataDir(t.TempDir(), WithoutPersistedAlertRestore())
|
||||
t.Cleanup(manager.Stop)
|
||||
manager.mu.Lock()
|
||||
manager.config.Enabled = true
|
||||
manager.config.FlappingEnabled = false
|
||||
manager.mu.Unlock()
|
||||
|
||||
managerEngine := &managerRefireParityEngine{manager: manager, clock: epoch, conf: scenario.confirmations}
|
||||
reducerState := reducer.NewState()
|
||||
reducerClock := epoch
|
||||
rule := reducer.DiscreteRule{Confirmations: scenario.confirmations}
|
||||
|
||||
for i, step := range scenario.steps {
|
||||
managerEngine.observe(t, step)
|
||||
reducerClock = reducerClock.Add(step.advance)
|
||||
reducerState.ApplyDiscrete(reducer.DiscreteSignal{
|
||||
ResourceID: refireParityResourceID,
|
||||
Key: "parity-state",
|
||||
Matched: step.matched,
|
||||
Severity: reducer.SeverityWarning,
|
||||
ObservedAt: reducerClock,
|
||||
}, rule)
|
||||
|
||||
manager.mu.Lock()
|
||||
alert, exists := manager.getActiveAlertNoLock(
|
||||
canonicalDiscreteStateStateID(refireParityResourceID, "parity-state"))
|
||||
manager.mu.Unlock()
|
||||
managerFiring := exists && alert != nil
|
||||
incident, ok := reducerState.Incident(refireParityResourceID, "parity-state")
|
||||
reducerFiring := ok && incident.State == reducer.StateFiring
|
||||
|
||||
label := fmt.Sprintf("step %d (matched=%v advance=%s)", i, step.matched, step.advance)
|
||||
|
||||
if managerFiring != step.wantFiring {
|
||||
t.Fatalf("%s: manager firing = %v, scenario expects %v (characterization drift)",
|
||||
label, managerFiring, step.wantFiring)
|
||||
}
|
||||
if managerFiring && step.wantStartOffset >= 0 {
|
||||
wantStart := epoch.Add(step.wantStartOffset)
|
||||
if !alert.StartTime.Equal(wantStart) {
|
||||
t.Fatalf("%s: manager start = %v, scenario expects %v", label, alert.StartTime, wantStart)
|
||||
}
|
||||
}
|
||||
|
||||
if reducerFiring != managerFiring {
|
||||
t.Fatalf("%s: PARITY DIVERGENCE — reducer firing = %v, manager = %v",
|
||||
label, reducerFiring, managerFiring)
|
||||
}
|
||||
if managerFiring && step.wantStartOffset >= 0 && !incident.StartedAt.Equal(alert.StartTime) {
|
||||
t.Fatalf("%s: PARITY DIVERGENCE — reducer start = %v, manager = %v",
|
||||
label, incident.StartedAt, alert.StartTime)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue