diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 199b9bdc55..6aa2d738a2 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -15,6 +15,31 @@ ## Purpose +### Connection confirmations cannot span unknown or disabled observations + +The existing PVE/PBS/PMG/vSphere/TrueNAS connection-degraded detector requires +three consecutive degraded observations to activate and three consecutive +healthy observations to recover. Pending or unknown connection state is neither +degradation nor recovery: it drops pending activation and intent grace, restarts +recovery confirmation, and retains an active occurrence's identity, severity, +acknowledgement, trusted metadata and observation time without fire/resolve +callbacks. Paused connections, disabled connections and owning-resource offline +policy disablement clear active alerts immediately and reset pending activation +even when no active alert exists. Healthy evidence also ends pending intent grace. + +Resets address exactly the connection's canonical discrete-state key, not another +connection, condition or metric. Core and optional shadow reducer observe the same +reset. Intent bookkeeping and its persisted checkpoint lose the interrupted run; +re-enabled detection must establish fresh confirmations and any explicit grace. +No new alert family, threshold, grace policy, route or external probe is added. + +`connection_confirmation_gap_test.go` covers all five platforms, six interruption +paths, exact-connection isolation, fresh activation/grace, acknowledged recovery +and exactly-once callbacks with shadow parity. `TestInterruptDiscreteRun` in +`reducer/discrete_test.go` pins exact-key isolation, retained firing state, +acknowledgement, no fabricated resolution and idempotent interruption. These +synthetic source checks do not establish native outage or destination delivery. + ### Retained guest observations are not alert evidence Guest memory marked `last-known`, `unavailable` or an unknown observation state diff --git a/internal/alerts/connection.go b/internal/alerts/connection.go index 9c0e7e855a..8772531d51 100644 --- a/internal/alerts/connection.go +++ b/internal/alerts/connection.go @@ -155,11 +155,31 @@ func (m *Manager) suppressConnectionDegradedAlert(snap ConnectionSnapshot) { alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey) m.mu.Lock() + m.interruptConnectionDegradedRunNoLock(snap.ID) m.mu.Unlock() m.clearAlert(alertID) } +// interruptConnectionDegradedRunNoLock ends confirmation and intent runs for +// this connection only. Unknown evidence is neither degradation nor recovery: +// keep any firing occurrence, but require fresh consecutive healthy polls. +// Callers hold m.mu. +func (m *Manager) interruptConnectionDegradedRunNoLock(resourceID string) { + specKey := canonicalDiscreteStateSpecID(resourceID, connectionDegradedStateKey) + for _, state := range m.mirrorStatesNoLock() { + state.InterruptDiscreteRun(resourceID, specKey) + } + m.clearConnectionDegradedIntentNoLock(resourceID) +} + +func (m *Manager) clearConnectionDegradedIntentNoLock(resourceID string) { + trackingKey := canonicalDiscreteStateStateID(resourceID, connectionDegradedStateKey) + if m.clearIntentPendingNoLock(trackingKey) { + m.saveActiveAlertsAsync("connection confirmation interrupted") + } +} + // CheckConnection raises or clears the connection-degraded alert for one // platform connection. Severity scales with observed state: stale → warning, // unreachable / unauthorized → critical. State=active runs through the @@ -187,10 +207,10 @@ func (m *Manager) CheckConnection(snap ConnectionSnapshot) { case ConnectionStateStale, ConnectionStateUnreachable, ConnectionStateUnauthorized: // fall through and fire default: - // pending/unknown — no alert, but reset the consecutive count so a - // later degraded run starts from zero instead of accumulating across - // a transient pending blip. + // Pending/unknown breaks both confirmation runs without resolving an + // existing outage or carrying activation grace across the gap. m.mu.Lock() + m.interruptConnectionDegradedRunNoLock(snap.ID) m.mu.Unlock() return } @@ -200,11 +220,8 @@ func (m *Manager) CheckConnection(snap ConnectionSnapshot) { severity = AlertLevelCritical } - // Another degraded observation invalidates any in-flight recovery - // confirmations — recovery must build back up from zero. + // The reducer resets recovery confirmations on degraded observations. alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey) - m.mu.Lock() - m.mu.Unlock() spec, err := buildCanonicalDiscreteStateSpec( snap.ID, @@ -310,8 +327,8 @@ func (m *Manager) clearConnectionDegradedAlert(snap ConnectionSnapshot) { defer m.mu.Unlock() defer m.shadowObserveRecoveryNoLock(snap.ID, specKey, alertID, offlineRecoveryConfirmationsDefault) - // Reset the legacy degraded counter; the reducer core owns the - // confirmation run itself. - + // Healthy evidence ends pending activation grace as well as the reducer's + // activation count. A firing incident still uses the recovery gate below. + m.clearConnectionDegradedIntentNoLock(snap.ID) m.resolveDiscreteRecoveryNoLock(snap.ID, specKey, alertID, offlineRecoveryConfirmationsDefault, "Connection", snap.Name, snap.ID) } diff --git a/internal/alerts/connection_confirmation_gap_test.go b/internal/alerts/connection_confirmation_gap_test.go new file mode 100644 index 0000000000..a5b43b396d --- /dev/null +++ b/internal/alerts/connection_confirmation_gap_test.go @@ -0,0 +1,201 @@ +package alerts + +import ( + "encoding/json" + "os" + "path/filepath" + "reflect" + "sync/atomic" + "testing" + "time" +) + +func interruptConnectionSnapshot(m *Manager, snap ConnectionSnapshot, gap string) { + switch gap { + case "pending": + snap.State = ConnectionStatePending + case "unknown": + snap.State = ConnectionState("unknown") + case "healthy": + snap.State = ConnectionStateActive + case "paused": + snap.State = ConnectionStatePaused + case "disabled": + snap.Enabled = false + case "offline-policy": + m.mu.Lock() + m.config.Overrides[snap.ID] = ThresholdConfig{DisableConnectivity: true} + m.mu.Unlock() + } + m.CheckConnection(snap) + if gap == "offline-policy" { + m.mu.Lock() + delete(m.config.Overrides, snap.ID) + m.mu.Unlock() + } +} + +func TestConnectionConfirmationInterrupted(t *testing.T) { + for _, platform := range []ConnectionType{ConnectionTypePVE, ConnectionTypePBS, ConnectionTypePMG, ConnectionTypeVMware, ConnectionTypeTrueNAS} { + for _, gap := range []string{"pending", "unknown", "healthy", "paused", "disabled", "offline-policy"} { + t.Run(string(platform)+"/"+gap, func(t *testing.T) { + m := newShadowFeedManager(t) + snap := platformConnectionSnapshot("connection-a", "Test connection", ConnectionStateStale) + snap.Type = platform + other := snap + other.ID = "connection-ab" + key := canonicalDiscreteStateSpecID(snap.ID, connectionDegradedStateKey) + alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey) + for range 2 { + m.CheckConnection(snap) + m.CheckConnection(other) + } + interruptConnectionSnapshot(m, snap, gap) + m.mu.RLock() + pending := testCoreHasIncident(m, snap.ID, key) + otherCount := testCoreConfirmations(m, other.ID, canonicalDiscreteStateSpecID(other.ID, connectionDegradedStateKey)) + m.mu.RUnlock() + if pending { + t.Error("interrupted observation retained activation confirmations") + } + if otherCount != 2 { + t.Errorf("interruption changed another connection's count to %d", otherCount) + } + for range 2 { + m.CheckConnection(snap) + if testHasActiveAlert(t, m, alertID) { + t.Fatal("non-consecutive degraded observations fired an alert") + } + } + m.CheckConnection(snap) + testRequireActiveAlert(t, m, alertID) + m.CheckConnection(other) + testRequireActiveAlert(t, m, canonicalDiscreteStateStateID(other.ID, connectionDegradedStateKey)) + if n := m.ShadowDivergences(); n != 0 { + t.Fatalf("shadow diverged %d times", n) + } + }) + } + } +} + +func TestConnectionUnknownInterruptsRecoveryWithoutResolving(t *testing.T) { + for _, gap := range []string{"pending", "unknown"} { + t.Run(gap, func(t *testing.T) { + m := newShadowFeedManager(t) + var fires, resolutions atomic.Int32 + resolved := make(chan struct{}, 2) + m.mu.Lock() + m.config.ActivationState = ActivationActive + m.mu.Unlock() + m.SetAlertCallback(func(*Alert) { fires.Add(1) }) + m.SetResolvedCallback(func(string) { + resolutions.Add(1) + resolved <- struct{}{} + }) + snap := platformConnectionSnapshot("connection-a", "Test connection", ConnectionStateStale) + alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey) + for range 3 { + m.CheckConnection(snap) + } + if err := m.AcknowledgeAlert(alertID, "test-operator"); err != nil { + t.Fatal(err) + } + snap.State = ConnectionStateActive + for range 2 { + m.CheckConnection(snap) + } + before := testRequireActiveAlert(t, m, alertID).Clone() + interruptConnectionSnapshot(m, snap, gap) + after := testRequireActiveAlert(t, m, alertID).Clone() + if !reflect.DeepEqual(before, after) { + t.Fatal("unknown observation changed the acknowledged active occurrence") + } + for range 2 { + m.CheckConnection(snap) + if !testHasActiveAlert(t, m, alertID) { + t.Fatal("non-consecutive healthy observations resolved the outage") + } + } + if fires.Load() != 1 || resolutions.Load() != 0 { + t.Fatalf("unexpected callbacks: fires=%d resolutions=%d", fires.Load(), resolutions.Load()) + } + m.CheckConnection(snap) + if testHasActiveAlert(t, m, alertID) { + t.Fatal("three fresh healthy observations did not resolve") + } + select { + case <-resolved: + case <-time.After(2 * time.Second): + t.Fatal("confirmed recovery did not dispatch its resolved callback") + } + if fires.Load() != 1 || resolutions.Load() != 1 { + t.Fatalf("unexpected final callbacks: fires=%d resolutions=%d", fires.Load(), resolutions.Load()) + } + if n := m.ShadowDivergences(); n != 0 { + t.Fatalf("shadow diverged %d times", n) + } + }) + } +} + +func TestConnectionInterruptionRestartsIntentGrace(t *testing.T) { + for _, gap := range []string{"pending", "unknown", "healthy", "paused", "disabled", "offline-policy"} { + t.Run(gap, func(t *testing.T) { + m := newShadowFeedManager(t) + var tick time.Duration + m.mu.Lock() + m.intentClock = func() time.Duration { return tick } + m.mu.Unlock() + grace := 60 + document := NewAlertIntentPolicyDocument() + document.Defaults[string(AlertIntentSignalOffline)] = AlertIntentRule{GraceSeconds: &grace} + if err := m.LoadIntentPolicies(document); err != nil { + t.Fatal(err) + } + snap := platformConnectionSnapshot("connection-a", "Test connection", ConnectionStateStale) + alertID := canonicalDiscreteStateStateID(snap.ID, connectionDegradedStateKey) + m.CheckConnection(snap) + tick = 40 * time.Second + m.CheckConnection(snap) + if err := m.SaveActiveAlerts(); err != nil { + t.Fatal(err) + } + interruptConnectionSnapshot(m, snap, gap) + m.mu.RLock() + _, pending := m.intentPending[alertID] + _, ticking := m.intentRuntimeTicks[alertID] + m.mu.RUnlock() + if pending || ticking { + t.Error("interrupted observation retained intent grace bookkeeping") + } + if err := m.SaveActiveAlerts(); err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(filepath.Join(m.getAlertsDir(), intentPendingFileName)) + if err != nil { + t.Fatal(err) + } + var saved []IntentPendingState + if err := json.Unmarshal(data, &saved); err != nil { + t.Fatal(err) + } + if len(saved) != 0 { + t.Fatal("checkpoint retained interrupted intent grace for restart") + } + for _, next := range []time.Duration{70, 90, 129} { + tick = next * time.Second + m.CheckConnection(snap) + if testHasActiveAlert(t, m, alertID) { + t.Fatal("intent grace accrued across an observation gap") + } + } + tick = 130 * time.Second + m.CheckConnection(snap) + testRequireActiveAlert(t, m, alertID) + if n := m.ShadowDivergences(); n != 0 { + t.Fatalf("shadow diverged %d times", n) + } + }) + } +} diff --git a/internal/alerts/reducer/discrete.go b/internal/alerts/reducer/discrete.go index 28e64029a7..cf7d77daa8 100644 --- a/internal/alerts/reducer/discrete.go +++ b/internal/alerts/reducer/discrete.go @@ -12,6 +12,26 @@ package reducer import "time" +// InterruptDiscreteRun ends the consecutive-observation evidence for one key +// without treating missing evidence as healthy. Pending activation (including +// intent grace) is dropped. Firing incidents keep their occurrence and +// acknowledgement, but recovery must restart from zero. No event is emitted. +// It reports whether a confirmation run was interrupted. +func (s *State) InterruptDiscreteRun(resourceID, subKey string) bool { + key := incidentKey(resourceID, subKey) + incident := s.incidents[key] + if incident == nil { + return false + } + if incident.State == StatePending { + delete(s.incidents, key) + return true + } + interrupted := incident.RecoveryCount != 0 + incident.RecoveryCount = 0 + return interrupted +} + // DiscreteSignal is one observation of a discrete condition on a resource: // Matched reports whether the trigger condition was observed (offline, // powered-state mismatch, state in the trigger set). diff --git a/internal/alerts/reducer/discrete_test.go b/internal/alerts/reducer/discrete_test.go index abf2538c6e..0ae975595e 100644 --- a/internal/alerts/reducer/discrete_test.go +++ b/internal/alerts/reducer/discrete_test.go @@ -5,6 +5,76 @@ import ( "time" ) +func TestInterruptDiscreteRun(t *testing.T) { + t.Run("pending confirmation and grace restart for the exact key", func(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 3, Intent: &DiscreteIntent{Explicit: true, GraceSeconds: 120}} + signal := discreteSignalAt(true, SeverityWarning, 0) + state.ApplyDiscrete(signal, rule) + signal.ObservedAt = signal.ObservedAt.Add(time.Minute) + state.ApplyDiscrete(signal, rule) + other := signal + other.Key = "other-condition" + state.ApplyDiscrete(other, rule) + other.ResourceID = "node-10" + state.ApplyDiscrete(other, rule) + if !state.InterruptDiscreteRun("node-1", "connectivity") { + t.Fatal("pending run was not interrupted") + } + if _, ok := state.Incident("node-1", "connectivity"); ok { + t.Fatal("pending run survived interruption") + } + for _, id := range []string{"node-1", "node-10"} { + if _, ok := state.Incident(id, "other-condition"); !ok { + t.Fatalf("interruption removed unrelated key for %s", id) + } + } + if state.InterruptDiscreteRun("node-1", "connectivity") { + t.Fatal("absent run reported an interruption") + } + signal.ObservedAt = signal.ObservedAt.Add(time.Hour) + events := state.ApplyDiscrete(signal, rule) + incident, ok := state.Incident(signal.ResourceID, signal.Key) + if len(events) != 1 || events[0].Type != EventPending || !ok || incident.Confirmations != 1 || !incident.PendingSince.Equal(signal.ObservedAt) { + t.Fatalf("activation did not restart: events=%+v incident=%+v", events, incident) + } + }) + + t.Run("firing occurrence and acknowledgement survive a recovery gap", func(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 1, RecoveryConfirmations: 3} + signal := discreteSignalAt(true, SeverityCritical, 0) + state.ApplyDiscrete(signal, rule) + if !state.Acknowledge(signal.ResourceID, signal.Key, "test-operator", signal.ObservedAt) { + t.Fatal("acknowledgement failed") + } + signal.Matched = false + state.ApplyDiscrete(signal, rule) + state.ApplyDiscrete(signal, rule) + before, _ := state.Incident(signal.ResourceID, signal.Key) + if before.RecoveryCount != 2 || !state.InterruptDiscreteRun(signal.ResourceID, signal.Key) { + t.Fatal("recovery run was not interrupted") + } + want := before + want.RecoveryCount = 0 + got, ok := state.Incident(signal.ResourceID, signal.Key) + if !ok || got != want { + t.Fatalf("interruption changed the firing occurrence: got=%+v want=%+v", got, want) + } + if len(state.resolved) != 0 || len(state.acks) != 1 || state.InterruptDiscreteRun(signal.ResourceID, signal.Key) { + t.Fatal("interruption manufactured resolution, lost acknowledgement or was not idempotent") + } + for range 2 { + if events := state.ApplyDiscrete(signal, rule); len(events) != 0 { + t.Fatalf("non-consecutive recovery emitted %+v", events) + } + } + if events := state.ApplyDiscrete(signal, rule); len(events) != 1 || events[0].Type != EventResolved { + t.Fatalf("fresh recovery did not resolve: %+v", events) + } + }) +} + func discreteSignalAt(matched bool, severity Severity, offset time.Duration) DiscreteSignal { return DiscreteSignal{ ResourceID: "node-1",