diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index dd1d022b02..37ccb65f0e 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -744,6 +744,16 @@ through `GetLiveHostsSnapshot`, which copies only hosts (see the monitoring contract); it must not route through a full state snapshot, which copied every guest on each agent report. +The mock toggle keeps the same boundary for evaluations already in flight. +`SetMockMode` ends the mock-mode epoch before it clears alerts and forgets the +fixture agents (see the monitoring contract), so a fixture agent pass that read +the estate before the toggle cannot reopen the agent's alerts or re-register +its hostname deduplication afterwards. The toggle waits for the +alert-manager calls already running, not for whole passes. Report admission is +not fenced: a live report already past its mock-mode check when mock mode is +switched on can still be evaluated after the clear, and its alerts can persist +until mock mode is left. + Physical-disk evidence collected by a host agent must survive projection back into monitoring's models. Absent evidence has to carry its declared sentinel rather than a zero value that reads as a real measurement: an absent diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 1019d8b72f..10fb0c6582 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1663,10 +1663,12 @@ boundaries: - Leaving mock mode routes the fixture agents through `HandleHostRemoved`. `ClearActiveAlerts` drops their alerts but not their hostname registrations, and a real node named like a fixture agent (`pve1`) would otherwise keep its - metric alerts suppressed with no agent to own them. A pass whose snapshot - predates the switch finds mock mode off under the same lock and evaluates - nothing. Two overlapping passes can still evaluate an older snapshot after - a newer one; the next pass reconciles the set again. + metric alerts suppressed with no agent to own them. `SetMockMode` ends the + mock-mode epoch before it forgets them, so a pass whose snapshot predates the + switch finds its epoch ended under the same lock and evaluates nothing, and + an agent whose departure the switch refused stays in the forgotten set. Two + overlapping passes in one epoch can still evaluate an older snapshot after a + newer one; the next pass reconciles the set again. Fixture agents must carry readings a real agent could report. Mock Kubernetes pods share 0.7 single-pod memory footprints per node, which keeps a node's pods @@ -1721,14 +1723,115 @@ boundary around it: matches node names only, so it relies on fixture node names being unique across instances, which they are. -The disk loop relies on the pass's single up-front mode check, like every -loop in `checkMockAlerts` except the host-agent step, which re-checks under its -own lock. A pass already past that check when mock mode is switched off can -reopen its fixture alerts after `SetMockMode(false)` clears them, and -other paths that evaluate fixture data (frontend refresh into unified alert -evaluation, backup rollups) have the same window. That mode-exit race predates -the disk loop and is not closed here; a lock around `checkMockAlerts` alone -would leave the other paths open. +Switching mock mode fences the alert evaluations that read mode-dependent data +(`mockModeFence`, `internal/monitoring/mock_mode_fence.go`). `GetState`, the +fixture graph, the unified read view, the recovery rollups and the connection +ledger all change source when the mode flips. A pass already holding one side's +data used to reach the alert manager after `SetMockMode` cleared it, and its +fixture alerts then stayed in live mode, with nothing left to evaluate or +resolve them, until the 24-hour stale sweep. Each of these paths takes a scope +before it reads anything mode-dependent and sends its alert-manager calls +through it: + +- `checkMockAlerts`, for every loop including the host-agent and disk steps + and the mock alert snapshot cache; +- `checkBackupAlerts`, shared by the PVE, PBS and mock polls, and + `pollGuestSnapshots`, whose guest inventory is the fixture read view in mock + mode; +- `checkConnectionAlerts`, whose ledger lists fixture connections in mock mode; +- `pruneStaleDockerAlerts`, which removes Docker alerts for hosts missing from + the read view; +- the resource-store refresh. `updateResourceStore` and + `updateResourceStoreForRead` take the scope of the state they rebuild from + (`currentStateWithScope` for `GetState`), and the unified alert sync + (`syncUnifiedResourceAlertsToState`) runs its policy reconcile, identity and + availability-link migrations, per-resource metrics and incident sync under + it. + +`SetMockMode` flips the mode, ends the epoch, waits for the alert-manager calls +already admitted under it and only then clears, so nothing these paths read +before the flip lands after the clear, in either direction. Switches are +serialized. The wait covers the calls already running, not whole passes, but it +has no fixed wall-clock bound: backup evaluation and unified incident sync are +estate-wide calls, and an admitted call can include metric-window reads and +migration persistence. Unified metrics are evaluated one resource per call, and +reads, capacity-trend queries and registry rebuilds stay outside the wait. +Admission never blocks, so a call made inside an admitted one cannot deadlock a +waiting switch; `SetMockMode` itself must not be called from inside an +evaluation. + +A registry rebuild publishes shared state, so a rebuild holding the mode the +monitor left can finish after a newer refresh published and replace the +registry that refresh reads back. A rebuild whose epoch has ended does not +start, and one already running is tracked: a refresh evaluates the registry it +read back only if no ended-epoch rebuild overlapped it, and otherwise leaves the +alerts to the next refresh. Even without a late rebuild, the registry still +holds the fixture estate right after leaving mock mode, and may hold it until an +undisturbed rebuild in the new epoch has replaced it +(`mockModeFence.registryCurrent`); in mock mode the read paths use the fixture +view instead. The live readers that act on that inventory (Docker pruning, the +backup-age guest lookup, the guest inventory of the backup and snapshot +pollers, and the pollers' previous-node and previous-guest carry, which writes +back into live state) read it through `currentModeReadState`, which serves a +view of current monitor state meanwhile. That view omits provider-owned +supplemental resources and persisted manual agent links, so a carry in the +window can fall back to platform guest memory for one poll, and it copies the +estate, so only callers outside per-resource loops use it. Every other live +registry reader (`GetUnifiedReadStateOrSnapshot` callers such as metrics +targets and system usage, and the broadcast view) may still see the fixture +estate until the next rebuild, as before, and metric samples synced from a late +rebuild stay in metric history. + +The fixture sources behind the router's mock supplemental adapters +(`mock.SupplementalRecords`, `mock.SupplementalChanges`) serve nothing while +mock mode is off. The router swaps those adapters for the real TrueNAS and +vSphere pollers only after `SetMockMode` returns, one source at a time with a +refresh after each, and the adapters used to fall back to default fixtures, so +a single switch to live mode raised fixture vSphere incidents. Two concurrent +mock-mode requests can still leave the router's provider set disagreeing with +the mode (fixture adapters serving nothing in live mode, or real pollers +feeding mock mode); neither raises fixture alerts in live mode. + +Report admission and the live pollers outside the paths above are not fenced: +a live report or poll already past its mock-mode check when mock mode is +switched on can still evaluate after that clear, and its alerts can persist +until mock mode is left. + +The fence and the clear are per monitor while the mode is process-wide. A +tenant monitor other than the one `SetMockMode` was called on neither drains +nor clears, so fixture alerts it raised outlive the switch as they did before +the fence; switching mock mode with several tenant monitors running is not +covered. + +The proofs park a real evaluation at a fixed point, switch modes and release +it. `TestLeavingMockModeRefusesTheRestOfAnInFlightPass` holds a mock pass at its +first storage capacity read, `TestLeavingMockModeDiscardsAnInFlightFrontendRefresh` +holds a frontend read in its supplemental-records supply, +`TestLeavingMockModeDiscardsAnInFlightBackupEvaluation` holds a backup +evaluation at its guest lookup, and +`TestLeavingMockModeDiscardsAnInFlightConnectionCheck` holds the ledger read. +Against the unfenced code they left about 40 fixture storage, disk and Docker +alerts, 5 vSphere incidents and 137 backup-age alerts in live mode, and counted +a mock-mode observation toward a live connection's three-observation +confirmation. `TestLeavingMockModeWaitsForAnInFlightEvaluation` holds an +evaluation in its CPU window read, outside the alert manager's lock, and fails +if the switch returns first; `TestLiveRefreshIgnoresARegistryRepublishedFromMockMode` +lets a mock-mode rebuild republish while a live refresh is held before reading +the registry back, then checks the next live refresh evaluates live records. +Each of those two fails against a fence without its wait or its overlap check. +`TestDockerPruneIgnoresTheRegistryMockModeLeft` raises a live Docker alert +before any live rebuild and fails if pruning reads the fixture registry, and +`TestPreviousStateCarryIgnoresTheRegistryMockModeLeft` fails if the +previous-state carry returns fixture nodes or guests. All eight are in +`internal/monitoring/monitor_mock_alerts_test.go`; +`TestSupplementalFixturesServeNothingOutsideMockMode` in +`internal/mock/platform_fixtures_test.go` pins the fixture sources; +`TestMockModeFence*` in +`internal/monitoring/mock_mode_fence_test.go` pin the fence itself, and +`TestStoreRefreshTakesItsMockModeScopeBeforeReadingState` in +`internal/monitoring/canonical_guardrails_test.go` rejects a call that reads +state in an argument ahead of `m.mockModeFence.begin()` and pins the +end-epoch-then-clear order in `SetMockMode`. The disk evaluation runs on the mock alert tick, not the poller's disk interval (five minutes by default, `PhysicalDiskPollingMinutes` per instance). diff --git a/internal/mock/platform_fixtures.go b/internal/mock/platform_fixtures.go index 96023a8d5b..19bb49cf2b 100644 --- a/internal/mock/platform_fixtures.go +++ b/internal/mock/platform_fixtures.go @@ -119,37 +119,24 @@ func defaultTrueNASConnectionFixture(fixtures PlatformFixtures) TrueNASConnectio } } +// SupplementalRecords serves the fixture graph's provider-backed records while +// mock mode is on and nothing once it is off. The router's mock adapters stay +// registered for a moment after leaving mock mode, and a live refresh that +// still reached them must not ingest and alert on fixture resources. func SupplementalRecords(source unifiedresources.DataSource) []unifiedresources.IngestRecord { - if IsMockEnabled() { - return CurrentFixtureGraph().SupplementalRecords(source) - } - - platformFixtures := defaultPlatformFixtures() - now := time.Now().UTC() - switch normalizeSupplementalSource(source) { - case unifiedresources.SourceTrueNAS: - return truenas.FixtureRecords(platformFixtures.TrueNAS) - case unifiedresources.SourceVMware: - return vmware.FixtureRecords(platformFixtures.VMware) - case unifiedresources.SourceAvailability: - return availabilityFixtureRecords(defaultAvailabilityFixtures(now), now) - default: + if !IsMockEnabled() { return nil } + return CurrentFixtureGraph().SupplementalRecords(source) } +// SupplementalChanges serves the fixture graph's provider activity while mock +// mode is on and nothing once it is off, for the same reason. func SupplementalChanges(source unifiedresources.DataSource) []unifiedresources.ResourceChange { - if IsMockEnabled() { - return CurrentFixtureGraph().SupplementalChanges(source) - } - - platformFixtures := defaultPlatformFixtures() - switch normalizeSupplementalSource(source) { - case unifiedresources.SourceVMware: - return vmware.FixtureActivityChanges(platformFixtures.VMware) - default: + if !IsMockEnabled() { return nil } + return CurrentFixtureGraph().SupplementalChanges(source) } func SupplementalOwnedSources() []unifiedresources.DataSource { diff --git a/internal/mock/platform_fixtures_test.go b/internal/mock/platform_fixtures_test.go index 947388c7a1..d7243b8704 100644 --- a/internal/mock/platform_fixtures_test.go +++ b/internal/mock/platform_fixtures_test.go @@ -232,13 +232,42 @@ func TestUnifiedResourceSnapshotParentsDemoProxmoxWorkloads(t *testing.T) { } func TestSupplementalRecordsNormalizesVMwareAlias(t *testing.T) { + previous := IsMockEnabled() + mustSetEnabled(t, true) + t.Cleanup(func() { mustSetEnabled(t, previous) }) + records := SupplementalRecords(unifiedresources.DataSource("vmware-vsphere")) if len(records) == 0 { t.Fatal("expected records for vmware-vsphere alias") } } +// The router's mock adapters outlive mock mode briefly while it swaps the real +// pollers back in; a live refresh that reaches them must find no fixtures. +func TestSupplementalFixturesServeNothingOutsideMockMode(t *testing.T) { + previous := IsMockEnabled() + mustSetEnabled(t, false) + t.Cleanup(func() { mustSetEnabled(t, previous) }) + + for _, source := range []unifiedresources.DataSource{ + unifiedresources.SourceTrueNAS, + unifiedresources.SourceVMware, + unifiedresources.SourceAvailability, + } { + if records := SupplementalRecords(source); len(records) != 0 { + t.Errorf("SupplementalRecords(%s) served %d fixture records outside mock mode", source, len(records)) + } + if changes := SupplementalChanges(source); len(changes) != 0 { + t.Errorf("SupplementalChanges(%s) served %d fixture changes outside mock mode", source, len(changes)) + } + } +} + func TestSupplementalChangesNormalizesVMwareAlias(t *testing.T) { + previous := IsMockEnabled() + mustSetEnabled(t, true) + t.Cleanup(func() { mustSetEnabled(t, previous) }) + changes := SupplementalChanges(unifiedresources.DataSource("vmware-vsphere")) if len(changes) == 0 { t.Fatal("expected activity changes for vmware-vsphere alias") diff --git a/internal/monitoring/availability_poller.go b/internal/monitoring/availability_poller.go index 9f13bf861a..1684535a7a 100644 --- a/internal/monitoring/availability_poller.go +++ b/internal/monitoring/availability_poller.go @@ -415,7 +415,7 @@ func (m *Monitor) RefreshAvailabilityTargets() { } m.refreshInstanceInfoCacheFromProviders() - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) } func (m *Monitor) pollAvailabilityTarget(ctx context.Context, target config.AvailabilityTarget) { @@ -425,7 +425,7 @@ func (m *Monitor) pollAvailabilityTarget(ctx context.Context, target config.Avai latency := time.Since(start) checkedAt := time.Now().UTC() m.applyAvailabilityObservationDetailed(target, uuid.NewString(), checkedAt, latency, result.Outcome, result.TransportOutcome, result.Application, err, result.Certificate, "", time.Time{}) - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) } // applyAvailabilityObservation records one availability observation regardless of diff --git a/internal/monitoring/availability_probe_agent.go b/internal/monitoring/availability_probe_agent.go index b5c0adbb9f..a0175d6107 100644 --- a/internal/monitoring/availability_probe_agent.go +++ b/internal/monitoring/availability_probe_agent.go @@ -205,7 +205,7 @@ func (m *Monitor) applyProbeAvailabilityResultsAt(hostID string, results []Probe if applied == 0 { return } - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) } func applicationResultFromReport(entry agentshost.AvailabilityProbeResult) *availabilityprobe.ApplicationResult { diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go index 02e18e99cc..eddb0df1c4 100644 --- a/internal/monitoring/broadcast_metrics_snapshot_test.go +++ b/internal/monitoring/broadcast_metrics_snapshot_test.go @@ -90,7 +90,7 @@ func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) { if freshness := latestUnifiedResourceLastSeen(adapter.GetAll()); !freshness.IsZero() { want.LastUpdate = freshness.UnixMilli() } - got := m.buildBroadcastFrontendStateFromSnapshotWithClock(graph.State, func() time.Time { return evaluationTime }) + got := m.buildBroadcastFrontendStateFromSnapshotWithClock(graph.State, m.mockModeFence.begin(), func() time.Time { return evaluationTime }) wantJSON, err := json.Marshal(want) if err != nil { t.Fatal(err) @@ -142,7 +142,7 @@ func TestBroadcastMetricsSnapshotHealthClockBoundary(t *testing.T) { detail string }{{59 * time.Second, "0m"}, {61 * time.Second, "1m"}} { calls := 0 - got := m.buildBroadcastFrontendStateFromSnapshotWithClock(models.EmptyStateSnapshot(), func() time.Time { + got := m.buildBroadcastFrontendStateFromSnapshotWithClock(models.EmptyStateSnapshot(), m.mockModeFence.begin(), func() time.Time { calls++ return original.Add(tc.age) }) @@ -301,7 +301,7 @@ func BenchmarkBroadcastMetricsSnapshot(b *testing.B) { b.StartTimer() } if bulk { - completeProjectionSink = m.buildBroadcastFrontendStateFromSnapshot(graph.State) + completeProjectionSink = m.buildBroadcastFrontendStateFromSnapshot(graph.State, m.mockModeFence.begin()) } else { completeProjectionSink = parentFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC()) } diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 6234aff22a..fd27c2215b 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -498,7 +498,7 @@ func TestUnifiedResourceAlertSyncEvaluatesMetricsBeforeIncidents(t *testing.T) { } source := string(data) - metricIndex := strings.Index(source, "CheckUnifiedResourceMetricsWithCapacityTrends(resources") + metricIndex := strings.Index(source, "CheckUnifiedResourceWithCapacityTrend(input, capacityTrends[input.ID])") incidentIndex := strings.Index(source, "SyncUnifiedResourceIncidents(resources)") if metricIndex < 0 { t.Fatalf("monitor alert sync must run unified resource metric evaluation") @@ -511,6 +511,43 @@ func TestUnifiedResourceAlertSyncEvaluatesMetricsBeforeIncidents(t *testing.T) { } } +// Monitor state and live snapshots switch to the fixture graph in mock mode, +// so a store refresh must take its mock-mode scope before it reads them. +// Go evaluates call arguments left to right: a scope begun in the argument +// after the read belongs to the epoch the snapshot may already have left. +func TestStoreRefreshTakesItsMockModeScopeBeforeReadingState(t *testing.T) { + files, err := filepath.Glob("*.go") + if err != nil { + t.Fatalf("glob monitoring sources: %v", err) + } + for _, file := range files { + if strings.HasSuffix(file, "_test.go") { + continue + } + data, err := os.ReadFile(file) + if err != nil { + t.Fatalf("read %s: %v", file, err) + } + source := string(data) + for _, forbidden := range []string{ + "GetState(), m.mockModeFence.begin()", + "GetSnapshot(), m.mockModeFence.begin()", + } { + if strings.Contains(source, forbidden) { + t.Errorf("%s reads state before taking its mock-mode scope (%q); use currentStateWithScope or begin the scope first", file, forbidden) + } + } + } + + data, err := os.ReadFile("monitor.go") + if err != nil { + t.Fatalf("failed to read monitor.go: %v", err) + } + if got := strings.Count(string(data), "m.mockModeFence.advance()\n\t\tm.alertManager.ClearActiveAlerts()"); got != 2 { + t.Fatalf("SetMockMode must end the mock-mode epoch immediately before clearing alerts in both directions, found %d", got) + } +} + func TestStateBroadcastTreatsTypedNilHubAsAbsent(t *testing.T) { data, err := os.ReadFile("monitor.go") if err != nil { @@ -2959,7 +2996,7 @@ func TestBroadcastProjectionMatchesPreviousPipeline(t *testing.T) { if !view.freshness.IsZero() { want.LastUpdate = view.freshness.UnixMilli() } - got := m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, func() time.Time { return now }) + got := m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, m.mockModeFence.begin(), func() time.Time { return now }) wantJSON, err := json.Marshal(want) if err != nil { t.Fatal(err) diff --git a/internal/monitoring/guest_memory_observation_test.go b/internal/monitoring/guest_memory_observation_test.go index 41fb665b42..f1b09836ce 100644 --- a/internal/monitoring/guest_memory_observation_test.go +++ b/internal/monitoring/guest_memory_observation_test.go @@ -79,7 +79,7 @@ func testGuestMemoryObservationLifecycle(t *testing.T) { assertGuestMemoryObservation(t, *legacy.Memory, state, source, at) // REST and WebSocket state use this production conversion, including // registry cloning and the nested Proxmox payload the drawer consumes. - front := m.buildBroadcastFrontendStateFromSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + front := m.buildBroadcastFrontendStateFromSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}, m.mockModeFence.begin()) data, err := json.Marshal(front.Resources) if err != nil { t.Fatal(err) diff --git a/internal/monitoring/mock_mode_fence.go b/internal/monitoring/mock_mode_fence.go new file mode 100644 index 0000000000..c552480bbe --- /dev/null +++ b/internal/monitoring/mock_mode_fence.go @@ -0,0 +1,196 @@ +package monitoring + +import ( + "sync" + "sync/atomic" +) + +// mockModeFence keeps alert evaluations of mode-dependent data inside the +// mock-mode epoch they read it in. GetState, the fixture graph, recovery +// rollups and the connection ledger all switch source when mock mode flips, +// and a pass already holding one side's data can reach the alert manager +// after SetMockMode cleared it. Its fixture alerts then outlive mock mode, +// with nothing left to evaluate or resolve them. +// +// A pass takes a scope before it reads anything mode-dependent and sends each +// alert-manager call through scope.run. SetMockMode flips the mode, advances +// the fence and only then clears: advance refuses every later call made under +// an older epoch and returns once the calls already admitted have finished, +// so nothing a pass read before the flip lands after the clear. advance waits +// for the alert-manager calls already admitted, not for whole passes; reads, +// metric queries and registry rebuilds stay outside it. +// +// A registry rebuild publishes shared state that a later refresh reads back +// and evaluates, so a rebuild still running when the epoch ends can replace a +// newer refresh's registry with the mode the monitor left. Rebuilds go +// through scope.publish, which advance does not wait for (a rebuild can spend +// seconds in SQLite), and a refresh evaluates what it read back only while +// undisturbedSince confirms no ended-epoch publish overlapped it. Readers +// that use the registry outside a refresh ask registryCurrent: after a switch +// the registry still holds the estate of the mode the monitor left, until an +// undisturbed publish in the new epoch replaces it. +// +// Admission never blocks, so a call that re-enters the fence (an alert +// callback that refreshes the resource store) cannot deadlock against a +// waiting advance; an inner call made under an ended epoch is refused like +// any other. advance must not be reached from inside an admitted call. The +// zero value is ready to use. +type mockModeFence struct { + mu sync.Mutex + drained *sync.Cond + epoch uint64 + inFlight map[uint64]int + // publishing counts registry publishes by epoch; lateFinishes counts + // publishes that finished after their epoch ended. + publishing map[uint64]int + lateFinishes uint64 + // cleanEpoch and cleanLate record the last undisturbed replacement; + // registryStale caches whether the registry is current for lock-free + // readers. Their zero values make a never-switched monitor's registry + // current. + cleanEpoch uint64 + cleanLate uint64 + registryStale atomic.Bool +} + +// mockModeScope is the epoch one pass read its data in. +type mockModeScope struct { + fence *mockModeFence + epoch uint64 +} + +// begin opens a scope in the current epoch. Callers must take it before they +// read the mock mode or any data whose source depends on it. +func (f *mockModeFence) begin() mockModeScope { + f.mu.Lock() + defer f.mu.Unlock() + return mockModeScope{fence: f, epoch: f.epoch} +} + +// advance ends the current epoch and waits until no call admitted under an +// ended epoch is still running. +func (f *mockModeFence) advance() { + f.mu.Lock() + defer f.mu.Unlock() + f.epoch++ + f.registryStale.Store(true) + for f.staleInFlightLocked() { + if f.drained == nil { + f.drained = sync.NewCond(&f.mu) + } + f.drained.Wait() + } +} + +func (f *mockModeFence) staleInFlightLocked() bool { + for epoch := range f.inFlight { + if epoch < f.epoch { + return true + } + } + return false +} + +// current reports whether the scope's epoch is still the fence's epoch. +func (s mockModeScope) current() bool { + s.fence.mu.Lock() + defer s.fence.mu.Unlock() + return s.fence.epoch == s.epoch +} + +// run calls fn if the scope's epoch is still current and reports whether it +// did. A mode change that starts while fn runs waits for it to return. +func (s mockModeScope) run(fn func()) bool { + f := s.fence + f.mu.Lock() + if f.epoch != s.epoch { + f.mu.Unlock() + return false + } + if f.inFlight == nil { + f.inFlight = make(map[uint64]int) + } + f.inFlight[s.epoch]++ + f.mu.Unlock() + + defer func() { + f.mu.Lock() + defer f.mu.Unlock() + if f.inFlight[s.epoch]--; f.inFlight[s.epoch] == 0 { + delete(f.inFlight, s.epoch) + } + if f.drained != nil { + f.drained.Broadcast() + } + }() + fn() + return true +} + +// publish runs fn, which replaces state that later refreshes read back (the +// unified resource registry) and reports whether it did, unless the scope's +// epoch has ended. It returns the mark undisturbedSince compares against. +func (s mockModeScope) publish(fn func() bool) (mark uint64, ok bool) { + f := s.fence + f.mu.Lock() + if f.epoch != s.epoch { + f.mu.Unlock() + return 0, false + } + if f.publishing == nil { + f.publishing = make(map[uint64]int) + } + f.publishing[s.epoch]++ + mark = f.lateFinishes + f.mu.Unlock() + + replaced := false + defer func() { + f.mu.Lock() + defer f.mu.Unlock() + if f.publishing[s.epoch]--; f.publishing[s.epoch] == 0 { + delete(f.publishing, s.epoch) + } + switch { + case f.epoch != s.epoch: + f.lateFinishes++ + case replaced && f.lateFinishes == mark && !f.olderPublishingLocked(): + f.cleanEpoch = s.epoch + f.cleanLate = f.lateFinishes + } + f.registryStale.Store(!f.registryCurrentLocked()) + }() + replaced = fn() + return mark, true +} + +// undisturbedSince reports whether no publish from an ended epoch has finished +// since mark or is still running. A refresh that read the registry back after +// its own publish evaluates it only then; otherwise the registry may hold the +// data of the mode the monitor left. +func (s mockModeScope) undisturbedSince(mark uint64) bool { + f := s.fence + f.mu.Lock() + defer f.mu.Unlock() + return f.lateFinishes == mark && !f.olderPublishingLocked() +} + +// registryCurrent reports whether the registry's last replacement was an +// undisturbed publish in the current epoch and no ended-epoch publish is still +// running. Until then it may hold the estate of the mode the monitor left. +func (f *mockModeFence) registryCurrent() bool { + return !f.registryStale.Load() +} + +func (f *mockModeFence) registryCurrentLocked() bool { + return f.cleanEpoch == f.epoch && f.cleanLate == f.lateFinishes && !f.olderPublishingLocked() +} + +func (f *mockModeFence) olderPublishingLocked() bool { + for epoch := range f.publishing { + if epoch < f.epoch { + return true + } + } + return false +} diff --git a/internal/monitoring/mock_mode_fence_test.go b/internal/monitoring/mock_mode_fence_test.go new file mode 100644 index 0000000000..02fb975f06 --- /dev/null +++ b/internal/monitoring/mock_mode_fence_test.go @@ -0,0 +1,120 @@ +package monitoring + +import ( + "testing" + "time" +) + +func TestMockModeFenceRefusesCallsFromAnEndedEpoch(t *testing.T) { + var fence mockModeFence + stale := fence.begin() + fence.advance() + + if stale.current() { + t.Fatal("a scope from an ended epoch still reads as current") + } + if stale.run(func() { t.Fatal("a call from an ended epoch ran") }) { + t.Fatal("run reported a refused call as admitted") + } + ran := false + if !fence.begin().run(func() { ran = true }) || !ran { + t.Fatal("a scope taken after the epoch ended was refused") + } +} + +func TestMockModeFenceAdvanceWaitsForAdmittedCalls(t *testing.T) { + var fence mockModeFence + scope := fence.begin() + entered, release := make(chan struct{}), make(chan struct{}) + nestedAdmitted := make(chan bool, 1) + callDone := make(chan struct{}) + go func() { + defer close(callDone) + scope.run(func() { + close(entered) + <-release + // A call made from inside an admitted one, after the epoch + // ended, is refused rather than deadlocking the waiting advance. + nestedAdmitted <- scope.run(func() {}) + }) + }() + waitForSignal(t, entered, 10*time.Second, "admitted call never started") + + advanced := make(chan struct{}) + go func() { + defer close(advanced) + fence.advance() + }() + waitForCondition(t, 10*time.Second, func() bool { return !scope.current() }, "advance never ended the epoch") + select { + case <-advanced: + t.Fatal("advance returned while a call admitted under the ended epoch was still running") + default: + } + if scope.run(func() { t.Error("a call started after the epoch ended ran") }) { + t.Fatal("run admitted a call after the epoch ended") + } + + close(release) + waitForSignal(t, callDone, 10*time.Second, "admitted call did not finish") + waitForSignal(t, advanced, 10*time.Second, "advance did not return once the admitted call finished") + if <-nestedAdmitted { + t.Fatal("a nested call made after the epoch ended was admitted") + } +} + +func TestMockModeFenceFlagsReadsOverlappedByAnEndedEpochPublish(t *testing.T) { + var fence mockModeFence + if !fence.registryCurrent() { + t.Fatal("a monitor that never switched modes distrusts its registry") + } + stale := fence.begin() + entered, release := make(chan struct{}), make(chan struct{}) + staleDone := make(chan struct{}) + go func() { + defer close(staleDone) + stale.publish(func() bool { + close(entered) + <-release + return true + }) + }() + waitForSignal(t, entered, 10*time.Second, "publish never started") + + // advance does not wait for publishes. + advanced := make(chan struct{}) + go func() { + defer close(advanced) + fence.advance() + }() + waitForSignal(t, advanced, 10*time.Second, "advance waited for a publish") + if fence.registryCurrent() { + t.Fatal("the registry counted as current right after a switch") + } + if _, ok := stale.publish(func() bool { t.Error("a publish from an ended epoch ran"); return true }); ok { + t.Fatal("publish admitted an ended epoch") + } + + live := fence.begin() + mark, ok := live.publish(func() bool { return true }) + if !ok { + t.Fatal("publish refused the current epoch") + } + if live.undisturbedSince(mark) || fence.registryCurrent() { + t.Fatal("a publish overlapped by a running ended-epoch publish counted as undisturbed") + } + close(release) + waitForSignal(t, staleDone, 10*time.Second, "ended-epoch publish did not finish") + if live.undisturbedSince(mark) || fence.registryCurrent() { + t.Fatal("a publish overlapped by a finished ended-epoch publish counted as undisturbed") + } + + // A refresh that skipped its rebuild replaced nothing. + if _, ok := live.publish(func() bool { return false }); !ok || fence.registryCurrent() { + t.Fatal("a publish that replaced nothing made the registry current") + } + next, ok := live.publish(func() bool { return true }) + if !ok || !live.undisturbedSince(next) || !fence.registryCurrent() { + t.Fatal("a rebuild after the ended-epoch publish finished did not make the registry current") + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index f91fb9fc3f..c81e4d50fa 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1282,6 +1282,7 @@ type Monitor struct { mockMetricsWg sync.WaitGroup mockHostAgentsMu sync.Mutex mockHostAgents map[string]models.Host // Fixture agents evaluated by the last mock alert pass + mockModeFence mockModeFence // Keeps mode-dependent alert evaluations inside the epoch they read in dockerChecker DockerChecker // Optional Docker checker for LXC containers dockerCheckerConfiguredAt time.Time // Last time the Docker checker was configured dockerCheckAllowedVMIDs map[int]struct{} // Optional VMID allowlist gating the LXC Docker socket probe; empty means all guests @@ -4360,30 +4361,30 @@ func (m *Monitor) PBSBackupsSnapshot() []models.PBSBackup { // BuildFrontendState returns the current state converted to frontend format. // This replaces the GetState().ToFrontend() pattern in consumer code. func (m *Monitor) BuildFrontendState() models.StateFrontend { - return m.buildBroadcastFrontendStateFromSnapshot(m.GetState()) + return m.buildBroadcastFrontendStateFromSnapshot(m.currentStateWithScope()) } // BuildBroadcastFrontendState returns frontend state ready for websocket // broadcasts, including the unified resource payload when a resource store is // configured. func (m *Monitor) BuildBroadcastFrontendState() models.StateFrontend { - return m.buildBroadcastFrontendStateFromSnapshot(m.GetState()) + return m.buildBroadcastFrontendStateFromSnapshot(m.currentStateWithScope()) } func buildFrontendStateFromSnapshot(snapshot models.StateSnapshot) models.StateFrontend { return snapshot.ToFrontend() } -func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateSnapshot) models.StateFrontend { - return m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, time.Now) +func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateSnapshot, scope mockModeScope) models.StateFrontend { + return m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, scope, time.Now) } // The clock is sampled at the same post-read health-evaluation boundary as // ordinary broadcasts. Tests can compare complete projections at one instant // without mistaking a naturally advancing health-age label for lost content. -func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot models.StateSnapshot, healthClock func() time.Time) models.StateFrontend { +func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot models.StateSnapshot, scope mockModeScope, healthClock func() time.Time) models.StateFrontend { frontendState := buildFrontendStateFromSnapshot(snapshot) - m.updateResourceStoreForRead(snapshot) + m.updateResourceStoreForRead(snapshot, scope) if m != nil && m.alertManager != nil { if liveAlerts := m.activeAlertsSnapshot(); len(liveAlerts) > 0 || len(frontendState.ActiveAlerts) > 0 { frontendState.ActiveAlerts = liveAlerts @@ -4576,19 +4577,33 @@ func (m *Monitor) broadcastEscalatedAlert(hub *websocket.Hub, alert *alerts.Aler hub.BroadcastAlertToTenant(m.GetOrgID(), alert) } +// mockModeSwitchMu serializes mock-mode switches. The mode is process-wide, +// and a switch is several steps (flip, end the epoch, clear, reset state), so +// two interleaved switches could clear what the later one just admitted. +var mockModeSwitchMu sync.Mutex + // SetMockMode switches between mock data and real infrastructure data at runtime. +// It must not be called from inside an alert evaluation, whose completion it +// may wait for. func (m *Monitor) SetMockMode(enable bool) error { + mockModeSwitchMu.Lock() + defer mockModeSwitchMu.Unlock() + current := mock.IsMockEnabled() if current == enable { log.Info().Bool("mockMode", enable).Msg("mock mode already in desired state") return nil } + // Every evaluation of mode-dependent data that started before the flip + // must finish or be refused before the clear, or it reopens alerts for + // the side the monitor just left (see mockModeFence). if enable { m.stopMockMetricsSampler() if err := mock.SetEnabled(true); err != nil { return err } + m.mockModeFence.advance() m.alertManager.ClearActiveAlerts() m.mu.Lock() m.resetStateLocked() @@ -4607,6 +4622,7 @@ func (m *Monitor) SetMockMode(enable bool) error { if err := mock.SetEnabled(false); err != nil { return err } + m.mockModeFence.advance() m.alertManager.ClearActiveAlerts() m.forgetMockHostAgents() m.mu.Lock() @@ -4750,7 +4766,7 @@ func (m *Monitor) SetResourceStore(store ResourceStoreInterface) { // Guard against minimally initialized monitors (e.g., test fixtures // with bare &Monitor{}) where m.state may be nil. if store != nil && m.state != nil { - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) } } @@ -4777,7 +4793,7 @@ func (m *Monitor) SetSupplementalRecordsProvider(source unifiedresources.DataSou } m.mu.Unlock() - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) } // SetLicenseChecker wires the commercial feature gate used by monitoring-owned @@ -5153,6 +5169,25 @@ func (m *Monitor) GetUnifiedReadStateOrSnapshot() unifiedresources.ReadState { return m.currentUnifiedStateView().readState } +// currentModeReadState is GetUnifiedReadStateOrSnapshot for callers that act +// on the inventory it lists: those that open or remove alerts from it (Docker +// pruning, backup and snapshot guest lookups) and the pollers' previous-state +// carry, which writes it back into live state. After leaving mock mode the +// registry may still hold the fixture estate until an undisturbed rebuild in +// the new epoch replaces it, so they get a view of current state meanwhile. +// The view lists the state snapshot without provider-owned supplemental +// resources or persisted manual links and copies the whole estate, so it is +// only for callers outside per-resource loops. +func (m *Monitor) currentModeReadState() unifiedresources.ReadState { + if m == nil { + return nil + } + if !mock.IsMockEnabled() && m.GetUnifiedReadState() != nil && !m.mockModeFence.registryCurrent() { + return m.unifiedStateViewWithStandaloneHostContinuity(monitorUnifiedStateViewFromSnapshot(m.GetState())).readState + } + return m.GetUnifiedReadStateOrSnapshot() +} + // shouldSkipNodeMetrics returns true if we should skip detailed metric polling // for the given node because a host agent is providing richer data. // This helps reduce API load when agents are active. @@ -5194,7 +5229,7 @@ type readRefreshResourceStore interface { // above all must not queue behind an in-flight rebuild whose change-record // and identity-pin persistence can take seconds per transaction on slow // volumes (#1665: /api/state stuck for minutes with SQLite on NFS). -func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot) { +func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot, scope mockModeScope) { m.mu.RLock() store := m.resourceStore m.mu.RUnlock() @@ -5204,7 +5239,7 @@ func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot) { } readStore, ok := store.(readRefreshResourceStore) if !ok { - m.updateResourceStore(state) + m.updateResourceStore(state, scope) return } @@ -5215,20 +5250,46 @@ func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot) { if ownedSources := m.providerOwnedSnapshotSources(); len(ownedSources) > 0 { snapshotForStore = unifiedresources.SnapshotWithoutSources(state, ownedSources) } - rebuilt := readStore.TryReplaceRegistryForRead(snapshotForStore, readPathRegistryFreshness, m.collectSupplementalRecordsBySource) - if !rebuilt { + rebuilt := false + mark, published := scope.publish(func() bool { + rebuilt = readStore.TryReplaceRegistryForRead(snapshotForStore, readPathRegistryFreshness, m.collectSupplementalRecordsBySource) + if rebuilt { + recordSupplementalResourceChanges(store, m.collectSupplementalChanges()) + } + return rebuilt + }) + if !published || !rebuilt { return } - recordSupplementalResourceChanges(store, m.collectSupplementalChanges()) + m.syncPublishedResourceStore(store, scope, mark) +} + +// syncPublishedResourceStore runs the metric and alert syncs over the registry +// a refresh just published under scope. A rebuild from an ended mock-mode +// epoch that overlapped the refresh may have replaced that registry with the +// mode the monitor left, so the alert sync runs only if none did. +func (m *Monitor) syncPublishedResourceStore(store ResourceStoreInterface, scope mockModeScope, mark uint64) { store = newResourceSnapshotStore(store) + resources := store.GetAll() m.syncAllUnifiedMetrics(store) - m.syncUnifiedResourceAlertsToState(store.GetAll()) + if scope.undisturbedSince(mark) { + m.syncUnifiedResourceAlertsToState(resources, scope) + } +} + +// currentStateWithScope reads the monitor state together with the mock-mode +// scope it was read in. GetState serves the fixture graph in mock mode, so a +// refresh that evaluates alerts from it must take the scope first. +func (m *Monitor) currentStateWithScope() (models.StateSnapshot, mockModeScope) { + scope := m.mockModeFence.begin() + return m.GetState(), scope } // updateResourceStore populates the canonical resource store from current // monitoring state. Callers use it at accepted-ingest boundaries and before // broadcast hydration so every ReadState consumer observes the same snapshot. -func (m *Monitor) updateResourceStore(state models.StateSnapshot) { +// The scope must predate the read of state (see currentStateWithScope). +func (m *Monitor) updateResourceStore(state models.StateSnapshot, scope mockModeScope) { m.mu.RLock() store := m.resourceStore m.mu.RUnlock() @@ -5263,46 +5324,48 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) { Msg("[Resources] Suppressing legacy snapshot slices for provider-owned sources") } - recordsBySource := m.collectSupplementalRecordsBySource() - supplementalChanges := m.collectSupplementalChanges() - if atomicStore, ok := store.(AtomicSnapshotResourceStore); ok { - atomicStore.PopulateSnapshotAndSupplemental(snapshotForStore, recordsBySource) - recordSupplementalResourceChanges(store, supplementalChanges) - store = newResourceSnapshotStore(store) - m.syncAllUnifiedMetrics(store) - for source, records := range recordsBySource { - if len(records) == 0 { - continue + mark, published := scope.publish(func() bool { + recordsBySource := m.collectSupplementalRecordsBySource() + supplementalChanges := m.collectSupplementalChanges() + if atomicStore, ok := store.(AtomicSnapshotResourceStore); ok { + atomicStore.PopulateSnapshotAndSupplemental(snapshotForStore, recordsBySource) + recordSupplementalResourceChanges(store, supplementalChanges) + for source, records := range recordsBySource { + if len(records) == 0 { + continue + } + log.Debug(). + Str("source", string(source)). + Int("records", len(records)). + Msg("[Resources] Atomically ingested supplemental records") } - log.Debug(). - Str("source", string(source)). - Int("records", len(records)). - Msg("[Resources] Atomically ingested supplemental records") + return true } - m.syncUnifiedResourceAlertsToState(store.GetAll()) + + store.PopulateFromSnapshot(snapshotForStore) + + supplementalStore, ok := store.(SupplementalRecordStore) + if ok { + for source, records := range recordsBySource { + if len(records) == 0 { + continue + } + supplementalStore.PopulateSupplementalRecords(source, records) + log.Debug(). + Str("source", string(source)). + Int("records", len(records)). + Msg("[Resources] Ingested supplemental records") + } + } + + recordSupplementalResourceChanges(store, supplementalChanges) + return true + }) + if !published { + // The snapshot belongs to the mode the monitor just left. return } - - store.PopulateFromSnapshot(snapshotForStore) - - supplementalStore, ok := store.(SupplementalRecordStore) - if ok { - for source, records := range recordsBySource { - if len(records) == 0 { - continue - } - supplementalStore.PopulateSupplementalRecords(source, records) - log.Debug(). - Str("source", string(source)). - Int("records", len(records)). - Msg("[Resources] Ingested supplemental records") - } - } - - recordSupplementalResourceChanges(store, supplementalChanges) - store = newResourceSnapshotStore(store) - m.syncAllUnifiedMetrics(store) - m.syncUnifiedResourceAlertsToState(store.GetAll()) + m.syncPublishedResourceStore(store, scope, mark) } // resourceSnapshotStore serves one GetAll clone to every consumer of a single @@ -5356,7 +5419,7 @@ func (m *Monitor) refreshUnifiedResourceStoreAfterAgentStateChange() { if m == nil || m.state == nil { return } - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) } // agentReportRefreshInterval bounds how often accepted agent reports refresh @@ -5389,7 +5452,7 @@ func (m *Monitor) refreshUnifiedResourceStoreAfterAgentReport() { } window := m.agentReportRefreshWindow if window <= 0 { - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) return } @@ -5405,7 +5468,7 @@ func (m *Monitor) refreshUnifiedResourceStoreAfterAgentReport() { r.running = true r.lastRun = now r.mu.Unlock() - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) r.mu.Lock() r.running = false r.mu.Unlock() @@ -5429,7 +5492,7 @@ func (m *Monitor) runTrailingAgentReportRefresh() { r.lastRun = time.Now() r.mu.Unlock() - m.updateResourceStore(m.GetState()) + m.updateResourceStore(m.currentStateWithScope()) r.mu.Lock() r.running = false diff --git a/internal/monitoring/monitor_alert_override_migration_test.go b/internal/monitoring/monitor_alert_override_migration_test.go index 87aca1c906..e1e4b49b03 100644 --- a/internal/monitoring/monitor_alert_override_migration_test.go +++ b/internal/monitoring/monitor_alert_override_migration_test.go @@ -210,7 +210,7 @@ func TestSyncUnifiedResourceAlertsPersistsAndEvaluatesTrueNASOverrideSuccession( trueNASMemoryAlertResource(secondID, "", sharedName, 90), trueNASMemoryAlertResource(newID, oldID, sharedName, 90), } - monitor.syncUnifiedResourceAlertsToState(resources) + monitor.syncUnifiedResourceAlertsToState(resources, monitor.mockModeFence.begin()) inMemory := manager.GetConfig() if _, exists := inMemory.Overrides[oldID]; exists { @@ -246,7 +246,7 @@ func TestSyncUnifiedResourceAlertsPersistsAndEvaluatesTrueNASOverrideSuccession( trueNASMemoryAlertResource(newID, oldID, sharedName, 96), trueNASMemoryAlertResource(secondID, "", sharedName, 70), } - monitor.syncUnifiedResourceAlertsToState(resources) + monitor.syncUnifiedResourceAlertsToState(resources, monitor.mockModeFence.begin()) warning := waitForAlertNotification(t, notifications, newID, alerts.AlertLevelWarning) waitForAlertResolution(t, resolutions, secondID) if warning.Threshold != overridePct { @@ -254,7 +254,7 @@ func TestSyncUnifiedResourceAlertsPersistsAndEvaluatesTrueNASOverrideSuccession( } resources[0] = trueNASMemoryAlertResource(newID, oldID, sharedName, 99) - monitor.syncUnifiedResourceAlertsToState(resources) + monitor.syncUnifiedResourceAlertsToState(resources, monitor.mockModeFence.begin()) critical := waitForAlertNotification(t, notifications, newID, alerts.AlertLevelCritical) if critical.Threshold != overridePct { t.Fatalf("critical notification threshold = %.1f, want %.1f", critical.Threshold, overridePct) @@ -264,7 +264,7 @@ func TestSyncUnifiedResourceAlertsPersistsAndEvaluatesTrueNASOverrideSuccession( } resources[0] = trueNASMemoryAlertResource(newID, oldID, sharedName, 89) - monitor.syncUnifiedResourceAlertsToState(resources) + monitor.syncUnifiedResourceAlertsToState(resources, monitor.mockModeFence.begin()) if _, exists := activeMemoryAlertForResource(manager, newID); exists { t.Fatalf("TrueNAS alert did not clear below the persisted 90%% recovery threshold") } @@ -284,7 +284,7 @@ func TestSyncUnifiedResourceAlertsPersistsAndEvaluatesTrueNASOverrideSuccession( restartedMonitor.syncUnifiedResourceAlertsToState([]unifiedresources.Resource{ trueNASMemoryAlertResource(newID, "", sharedName, 90), trueNASMemoryAlertResource(secondID, "", sharedName, 70), - }) + }, restartedMonitor.mockModeFence.begin()) if _, exists := activeMemoryAlertForResource(restartedManager, newID); exists { t.Fatalf("restarted manager evaluated %s at the default threshold", newID) } @@ -434,7 +434,7 @@ func TestSyncUnifiedResourceAlertsMigratesDockerContainerOverrideKeys(t *testing monitor.syncUnifiedResourceAlertsToState([]unifiedresources.Resource{ dockerApp("app-container-1111111111111111", "host-1", "aaaaaaaaaaaa", "media-server"), dockerApp("app-container-0011223344556677", "host-1", "cccccccccccc", "proxy"), - }) + }, monitor.mockModeFence.begin()) inMemory := manager.GetConfig() for _, gone := range []string{ diff --git a/internal/monitoring/monitor_alert_restore_test.go b/internal/monitoring/monitor_alert_restore_test.go index aa18515c16..619fc2d011 100644 --- a/internal/monitoring/monitor_alert_restore_test.go +++ b/internal/monitoring/monitor_alert_restore_test.go @@ -134,7 +134,7 @@ func TestResourcePublicationReconcilesRestoredNativeAlertAliases(t *testing.T) { if got := len(manager.GetActiveAlerts()); got != 3 { t.Fatalf("alerts before resource publication = %d, want 3", got) } - monitor.updateResourceStore(snapshot) + monitor.updateResourceStore(snapshot, monitor.mockModeFence.begin()) active := manager.GetActiveAlerts() if len(active) != 1 || active[0].ResourceID != "unaffected" { t.Fatalf("resource publication did not reconcile restored aliases: %+v", active) diff --git a/internal/monitoring/monitor_alert_sync.go b/internal/monitoring/monitor_alert_sync.go index 9fc1154af9..b85b085e15 100644 --- a/internal/monitoring/monitor_alert_sync.go +++ b/internal/monitoring/monitor_alert_sync.go @@ -94,15 +94,39 @@ func (m *Monitor) recentlyResolvedAlertsSnapshot() []models.ResolvedAlert { return m.alertManager.GetRecentlyResolved() } -func (m *Monitor) syncUnifiedResourceAlertsToState(resources []unifiedresources.Resource) { +// syncUnifiedResourceAlertsToState evaluates resources read under scope. A +// mock-mode flip since that read refuses every remaining step, so a snapshot +// of the mode the monitor left can neither open or resolve alerts nor rewrite +// persisted identities and links in the mode it entered. +func (m *Monitor) syncUnifiedResourceAlertsToState(resources []unifiedresources.Resource, scope mockModeScope) { if m == nil || m.alertManager == nil { return } // Resource publication can make persisted canonical policy resolvable for // restored native-ID alerts that preceded the first observation at startup. - m.alertManager.ReconcileOperatorIntentState() + scope.run(func() { m.alertManager.ReconcileOperatorIntentState() }) + scope.run(func() { m.migrateAlertIdentities(resources) }) + scope.run(func() { m.migrateAvailabilityLinksToCanonicalIDs(resources) }) + // Metrics go one resource per fenced call, as + // CheckUnifiedResourceMetricsWithCapacityTrends would iterate them, so a + // mode change waits for one resource's evaluation rather than the estate's. + capacityTrends := m.unifiedStorageCapacityTrends(resources, time.Now()) + for _, resource := range resources { + input, ok := alerts.UnifiedResourceInputFromResource(resource) + if !ok { + continue + } + scope.run(func() { m.alertManager.CheckUnifiedResourceWithCapacityTrend(input, capacityTrends[input.ID]) }) + } + scope.run(func() { m.alertManager.SyncUnifiedResourceIncidents(resources) }) + m.syncAlertsToState() +} + +// migrateAlertIdentities persists the alert identity migration that resources +// make resolvable. +func (m *Monitor) migrateAlertIdentities(resources []unifiedresources.Resource) { config := m.alertManager.GetConfig() migrationPlan := alerts.PlanAlertIdentityMigration(config, resources) if migrationPlan.UnsupportedVersion { @@ -136,12 +160,6 @@ func (m *Monitor) syncUnifiedResourceAlertsToState(resources []unifiedresources. Msg("persisted alert identity migration plan") } } - - m.migrateAvailabilityLinksToCanonicalIDs(resources) - - m.alertManager.CheckUnifiedResourceMetricsWithCapacityTrends(resources, m.unifiedStorageCapacityTrends(resources, time.Now())) - m.alertManager.SyncUnifiedResourceIncidents(resources) - m.syncAlertsToState() } // pruneStaleDockerAlerts removes docker alerts that reference hosts no longer present in state. @@ -150,7 +168,10 @@ func (m *Monitor) pruneStaleDockerAlerts() bool { return false } - readState := m.GetUnifiedReadStateOrSnapshot() + // The inventory is the fixture read view in mock mode. A prune that read + // it must not remove Docker alerts the other mode raised after a switch. + scope := m.mockModeFence.begin() + readState := m.currentModeReadState() if readState == nil { return false } @@ -216,7 +237,9 @@ func (m *Monitor) pruneStaleDockerAlerts() bool { host.Hostname = hostID } - m.alertManager.HandleDockerHostRemoved(host) + if !scope.run(func() { m.alertManager.HandleDockerHostRemoved(host) }) { + return cleared + } processed[hostID] = struct{}{} cleared = true } diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index a75e7dd557..afb150ad62 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -181,8 +181,10 @@ func (m *Monitor) checkConnectionAlerts() { if lister == nil || m.alertManager == nil { return } + // The lister serves the mock connection ledger while mock mode is on. + scope := m.mockModeFence.begin() for _, snap := range lister() { - m.alertManager.CheckConnection(snap) + scope.run(func() { m.alertManager.CheckConnection(snap) }) } } @@ -628,6 +630,10 @@ func (m *Monitor) broadcastStateUpdate() { func (m *Monitor) checkMockAlerts() { defer recoverFromPanic("checkMockAlerts") + // Passes overlap and can still be reading the fixture when the monitor + // leaves mock mode, so every evaluation below runs under the scope taken + // before the mode check and stops once SetMockMode ends its epoch. + scope := m.mockModeFence.begin() log.Debug().Bool("mockEnabled", mock.IsMockEnabled()).Msg("checkMockAlerts called") if !mock.IsMockEnabled() { log.Debug().Msg("mock mode not enabled, skipping mock alert check") @@ -661,7 +667,7 @@ func (m *Monitor) checkMockAlerts() { log.Debug(). Int("trackedNodes", len(existingNodes)). Msg("Collecting resources for alert cleanup in mock mode") - m.alertManager.CleanupAlertsForNodes(existingNodes) + scope.run(func() { m.alertManager.CleanupAlertsForNodes(existingNodes) }) m.checkBackupAlerts(context.Background()) @@ -678,7 +684,7 @@ func (m *Monitor) checkMockAlerts() { Msg("Reached guest check limit for this cycle") break } - m.alertManager.CheckGuest(vm, "mock") + scope.run(func() { m.alertManager.CheckGuest(vm, "mock") }) guestsChecked++ } @@ -687,7 +693,7 @@ func (m *Monitor) checkMockAlerts() { if guestsChecked >= maxGuestsPerCycle { break } - m.alertManager.CheckGuest(container, "mock") + scope.run(func() { m.alertManager.CheckGuest(container, "mock") }) guestsChecked++ } @@ -699,11 +705,11 @@ func (m *Monitor) checkMockAlerts() { // and disk alerts to its agent on the first tick, as the host-agent // hostname deduplication does in production. log.Debug().Int("hostCount", len(state.Hosts)).Msg("checking host agent alerts") - m.evaluateMockHostAgents(state.Hosts, state.Nodes) + m.evaluateMockHostAgents(scope, state.Hosts, state.Nodes) // Check alerts for each node for _, node := range state.Nodes { - m.alertManager.CheckNode(node) + scope.run(func() { m.alertManager.CheckNode(node) }) } // Check alerts for storage @@ -713,7 +719,8 @@ func (m *Monitor) checkMockAlerts() { Str("name", storage.Name). Float64("usage", storage.Usage). Msg("Checking storage for alerts") - m.alertManager.CheckStorageWithCapacityTrend(storage, m.storageCapacityTrend(storage, time.Now())) + trend := m.storageCapacityTrend(storage, time.Now()) + scope.run(func() { m.alertManager.CheckStorageWithCapacityTrend(storage, trend) }) } // Check alerts for physical disks. Live disks are evaluated by the @@ -722,18 +729,18 @@ func (m *Monitor) checkMockAlerts() { // this pass the estate shows failing disks with no disk-health or // disk-wearout alert. log.Debug().Int("diskCount", len(state.PhysicalDisks)).Msg("checking physical disk alerts") - m.checkMockPhysicalDiskAlerts(state.PhysicalDisks, state.Nodes, state.Hosts) + m.checkMockPhysicalDiskAlerts(scope, state.PhysicalDisks, state.Nodes, state.Hosts) // Check alerts for PBS instances log.Debug().Int("pbsCount", len(state.PBSInstances)).Msg("checking PBS alerts") for _, pbsInst := range state.PBSInstances { - m.alertManager.CheckPBS(pbsInst) + scope.run(func() { m.alertManager.CheckPBS(pbsInst) }) } // Check alerts for PMG instances log.Debug().Int("pmgCount", len(state.PMGInstances)).Msg("checking PMG alerts") for _, pmgInst := range state.PMGInstances { - m.alertManager.CheckPMG(pmgInst) + scope.run(func() { m.alertManager.CheckPMG(pmgInst) }) } // Check alerts for Docker hosts (container state/health/metrics/updates and @@ -742,25 +749,29 @@ func (m *Monitor) checkMockAlerts() { // against mock data. log.Debug().Int("dockerHostCount", len(state.DockerHosts)).Msg("checking docker alerts") for _, dockerHost := range state.DockerHosts { - m.checkMockDockerHostAlerts(dockerHost) + scope.run(func() { m.checkMockDockerHostAlerts(dockerHost) }) } // Cache the latest alert snapshots directly in the mock data so the API can serve // mock state without needing to grab the alert manager lock again. - mock.UpdateAlertSnapshots(m.alertManager.GetActiveAlerts(), m.alertManager.GetRecentlyResolved()) + scope.run(func() { + mock.UpdateAlertSnapshots(m.alertManager.GetActiveAlerts(), m.alertManager.GetRecentlyResolved()) + }) } // evaluateMockHostAgents evaluates every fixture agent and remembers the set. // A runtime mock config change rebuilds the estate, so an agent can leave it // between passes; it then goes through HandleHostRemoved, as a deleted live // agent does, or its alerts and hostname deduplication would outlive it. -func (m *Monitor) evaluateMockHostAgents(hosts []models.Host, nodes []models.Node) { +func (m *Monitor) evaluateMockHostAgents(scope mockModeScope, hosts []models.Host, nodes []models.Node) { m.mockHostAgentsMu.Lock() defer m.mockHostAgentsMu.Unlock() // A pass that took its snapshot before the monitor left mock mode must // not re-register fixture agents after forgetMockHostAgents ran. - if !mock.IsMockEnabled() { + // SetMockMode ends the epoch before it forgets them, so either this check + // fails or forgetMockHostAgents waits for this pass to record its set. + if !scope.current() { return } @@ -773,12 +784,17 @@ func (m *Monitor) evaluateMockHostAgents(hosts []models.Host, nodes []models.Nod // Remove first so a departed agent cannot unregister a hostname that a // remaining agent registers on this same pass. for id, host := range m.mockHostAgents { - if _, ok := current[id]; !ok { - m.alertManager.HandleHostRemoved(host) + if _, ok := current[id]; ok { + continue + } + if !scope.run(func() { m.alertManager.HandleHostRemoved(host) }) { + // Leaving mock mode refused the removal; keep the agent so + // forgetMockHostAgents still removes it. + current[id] = host } } for _, host := range hosts { - m.checkMockHostAlerts(host, nodes) + scope.run(func() { m.checkMockHostAlerts(host, nodes) }) } m.mockHostAgents = current } @@ -786,9 +802,9 @@ func (m *Monitor) evaluateMockHostAgents(hosts []models.Host, nodes []models.Nod // forgetMockHostAgents removes the fixture agents when the monitor leaves mock // mode. ClearActiveAlerts drops their alerts but not their hostname // deduplication, so a real node named like a fixture agent (pve1) would keep -// its CPU, memory and disk alerts suppressed with no agent to own them. The -// removal also clears anything a pass already in flight recreated; mock mode -// is off before this runs, so no later pass can register them again. +// its CPU, memory and disk alerts suppressed with no agent to own them. +// SetMockMode ends the mock epoch before this runs, so the set it removes is +// final: no pass of that epoch can register an agent or evaluate one again. func (m *Monitor) forgetMockHostAgents() { m.mockHostAgentsMu.Lock() defer m.mockHostAgentsMu.Unlock() @@ -831,7 +847,7 @@ func (m *Monitor) checkMockDockerHostAlerts(host models.DockerHost) { // evaluated as healthy. The poller's wait for host-agent links to settle after // a restart does not apply: fixture links and exclusions are complete from the // first pass. -func (m *Monitor) checkMockPhysicalDiskAlerts(disks []models.PhysicalDisk, nodes []models.Node, hosts []models.Host) { +func (m *Monitor) checkMockPhysicalDiskAlerts(scope mockModeScope, disks []models.PhysicalDisk, nodes []models.Node, hosts []models.Host) { type nodeKey struct{ instance, name string } excludeByHost := make(map[string][]string, len(hosts)) @@ -855,6 +871,6 @@ func (m *Monitor) checkMockPhysicalDiskAlerts(disks []models.PhysicalDisk, nodes if !onlineNodes[key] { continue } - m.checkPhysicalDiskAlerts(disk.Instance, disk, excludeByNode[key]) + scope.run(func() { m.checkPhysicalDiskAlerts(disk.Instance, disk, excludeByNode[key]) }) } } diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index d5475857eb..fdbcbeac11 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -568,25 +568,27 @@ func (m *Monitor) syncGuestBackupTimesAndResourceStore() { } m.state.SyncGuestBackupTimes() - m.updateResourceStore(m.state.GetSnapshot()) + scope := m.mockModeFence.begin() + m.updateResourceStore(m.state.GetSnapshot(), scope) } func (m *Monitor) backupReadStateForInstance(instanceName string) unifiedresources.ReadState { if m == nil { return nil } - readState := m.GetUnifiedReadStateOrSnapshot() + readState := m.currentModeReadState() if backupReadStateHasGuestForInstance(readState, instanceName) || m.state == nil { return readState } + scope := m.mockModeFence.begin() snapshot := m.state.GetSnapshot() if !backupSnapshotHasGuestForInstance(snapshot, instanceName) { return readState } - m.updateResourceStore(snapshot) - readState = m.GetUnifiedReadStateOrSnapshot() + m.updateResourceStore(snapshot, scope) + readState = m.currentModeReadState() if backupReadStateHasGuestForInstance(readState, instanceName) { return readState } @@ -950,6 +952,8 @@ func (m *Monitor) pollPVEBackupsAndSnapshots(parentCtx context.Context, instance func (m *Monitor) pollGuestSnapshots(ctx context.Context, instanceName string, client PVEClientInterface) { log.Debug().Str("instance", instanceName).Msg("polling guest snapshots") + // The guest inventory comes from the fixture read view in mock mode. + scope := m.mockModeFence.begin() readState := m.backupReadStateForInstance(instanceName) var vms []models.VM for _, vm := range readState.VMs() { @@ -1270,7 +1274,7 @@ func (m *Monitor) pollGuestSnapshots(ctx context.Context, instanceName string, c } if m.alertManager != nil { - m.alertManager.CheckSnapshotsForInstance(instanceName, allSnapshots, guestLookups) + scope.run(func() { m.alertManager.CheckSnapshotsForInstance(instanceName, allSnapshots, guestLookups) }) } log.Debug(). diff --git a/internal/monitoring/monitor_full_coverage_test.go b/internal/monitoring/monitor_full_coverage_test.go index bce1b5d0b8..a7a563e632 100644 --- a/internal/monitoring/monitor_full_coverage_test.go +++ b/internal/monitoring/monitor_full_coverage_test.go @@ -434,7 +434,7 @@ func TestMonitor_ResourceUpdate(t *testing.T) { } // updateResourceStore - m.updateResourceStore(models.StateSnapshot{}) + m.updateResourceStore(models.StateSnapshot{}, m.mockModeFence.begin()) // PopulateFromSnapshot called (no-op in mock, but covered) // getResourcesForBroadcast diff --git a/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go b/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go index 7459a3c610..dca2a68b1c 100644 --- a/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go +++ b/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go @@ -823,9 +823,9 @@ func TestMockHostAgentLeavingFixtureUsesRemovalLifecycle(t *testing.T) { kept := models.Host{ID: "host-linux-1", Hostname: "apollo-114", Status: "online", Memory: memory} departed := models.Host{ID: "host-node-pve9", Hostname: "pve9", Status: "online", Memory: memory} - monitor.evaluateMockHostAgents([]models.Host{kept, departed}, nil) + monitor.evaluateMockHostAgents(monitor.mockModeFence.begin(), []models.Host{kept, departed}, nil) // A runtime mock config change rebuilt the estate without pve9. - monitor.evaluateMockHostAgents([]models.Host{kept}, nil) + monitor.evaluateMockHostAgents(monitor.mockModeFence.begin(), []models.Host{kept}, nil) keptAlert := false for _, alert := range manager.GetActiveAlerts() { diff --git a/internal/monitoring/monitor_mock_alerts_test.go b/internal/monitoring/monitor_mock_alerts_test.go index 3c5c933339..3c02565ece 100644 --- a/internal/monitoring/monitor_mock_alerts_test.go +++ b/internal/monitoring/monitor_mock_alerts_test.go @@ -1,13 +1,23 @@ package monitoring import ( + "bytes" + "context" + "fmt" + "runtime" + "slices" + "sort" "strings" + "sync" + "sync/atomic" "testing" "time" "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog" "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) func TestCheckMockDockerHostAlertsUsesHostLifecycleForOfflineFixtures(t *testing.T) { @@ -209,21 +219,25 @@ func TestLeavingMockModeReleasesFixtureAgentHostnames(t *testing.T) { func TestMockHostAgentPassAfterLeavingMockModeRegistersNothing(t *testing.T) { manager := newMockHostAlertTestManager(t) monitor := &Monitor{alertManager: manager} - mustSetMockEnabled(t, false) // A pass whose snapshot predates leaving mock mode reaches the agent step - // after forgetMockHostAgents ran. + // after SetMockMode ended its epoch and forgetMockHostAgents ran. + scope := monitor.mockModeFence.begin() + monitor.mockModeFence.advance() stale := models.Host{ ID: "host-node-mock-cluster-1-pve1", Hostname: "pve1", Status: "online", Memory: models.Memory{Total: 32 << 30, Used: 30 << 30, Free: 2 << 30, Usage: 93.75}, } - monitor.evaluateMockHostAgents([]models.Host{stale}, nil) + monitor.evaluateMockHostAgents(scope, []models.Host{stale}, nil) if active := manager.GetActiveAlerts(); len(active) != 0 { t.Fatalf("stale mock pass opened %d alerts in live mode", len(active)) } + if len(monitor.mockHostAgents) != 0 { + t.Fatalf("stale mock pass registered fixture agents %v after they were forgotten", monitor.mockHostAgents) + } node := models.Node{ ID: "real-pve1", Name: "pve1", @@ -240,6 +254,561 @@ func TestMockHostAgentPassAfterLeavingMockModeRegistersNothing(t *testing.T) { t.Fatal("stale mock pass registered fixture hostname pve1 and suppressed the real node's memory alert") } +// pinDefaultMockEstate enables mock mode on the default estate, whatever +// PULSE_MOCK_* the shell exports, and restores the previous mode and config. +func pinDefaultMockEstate(t *testing.T) { + t.Helper() + previousEnabled, previousConfig := mock.IsMockEnabled(), mock.GetConfig() + t.Cleanup(func() { + mustSetMockEnabled(t, false) + mock.SetMockConfig(previousConfig) + mustSetMockEnabled(t, previousEnabled) + }) + mustSetMockEnabled(t, false) + cfg := mock.DefaultConfig + cfg.UpdateInterval = 5 * time.Minute + mock.SetMockConfig(cfg) +} + +// signalFirstFiring closes the returned channel when manager fires its first +// alert. Lifecycle callbacks run inside the evaluation that fired, so the +// pass that fired is past its fixture read when the channel closes. +func signalFirstFiring(t *testing.T, manager *alerts.Manager) <-chan struct{} { + t.Helper() + fired := make(chan struct{}) + var once sync.Once + unsubscribe := manager.SubscribeLifecycleCallback(func(event alerts.LifecycleEvent) { + if event.Type == eventlog.TypeFired { + once.Do(func() { close(fired) }) + } + }) + t.Cleanup(unsubscribe) + return fired +} + +func describeAlerts(active []alerts.Alert) string { + keys := make([]string, 0, len(active)) + for _, alert := range active { + keys = append(keys, alert.Type+" "+alert.ResourceID) + } + sort.Strings(keys) + return strings.Join(keys, ", ") +} + +// switchMockModeAsync runs SetMockMode on its own goroutine, so a test that is +// holding an evaluation parked can still bound how long the switch takes. +func switchMockModeAsync(t *testing.T, monitor *Monitor, enable bool) <-chan struct{} { + t.Helper() + done := make(chan struct{}) + go func() { + defer close(done) + if err := monitor.SetMockMode(enable); err != nil { + t.Errorf("SetMockMode(%v): %v", enable, err) + } + }() + return done +} + +// goroutineIn reports whether one goroutine's stack holds every frame. +func goroutineIn(frames ...string) bool { + buf := make([]byte, 4<<20) + n := runtime.Stack(buf, true) + for _, goroutine := range bytes.Split(buf[:n], []byte("\n\n")) { + all := true + for _, frame := range frames { + all = all && bytes.Contains(goroutine, []byte(frame)) + } + if all { + return true + } + } + return false +} + +// waitForGoroutineIn waits until one goroutine's stack holds every frame. A +// test holding a lock uses it to see the goroutine it started parked on that +// lock at a known point. +func waitForGoroutineIn(t *testing.T, frames ...string) { + t.Helper() + waitForCondition(t, 30*time.Second, func() bool { return goroutineIn(frames...) }, + fmt.Sprintf("no goroutine reached %s", strings.Join(frames, " via "))) +} + +func TestLeavingMockModeRefusesTheRestOfAnInFlightPass(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + } + mustSetMonitorMockMode(t, monitor, true) + + // Park a real pass at its first storage capacity read: it has read the + // fixture and evaluated guests, agents and nodes; storage, disks, PBS, + // PMG and Docker are still ahead of it. + monitor.metricsHistory.capacityForecastMu.Lock() + parked := true + defer func() { + if parked { + monitor.metricsHistory.capacityForecastMu.Unlock() + } + }() + passDone := make(chan struct{}) + go func() { + defer close(passDone) + monitor.checkMockAlerts() + }() + waitForGoroutineIn(t, "monitoring.(*Monitor).storageCapacityTrendFor", "monitoring.(*Monitor).checkMockAlerts") + + waitForSignal(t, switchMockModeAsync(t, monitor, false), 30*time.Second, "leaving mock mode waited on a pass parked outside any evaluation") + monitor.metricsHistory.capacityForecastMu.Unlock() + parked = false + waitForSignal(t, passDone, 30*time.Second, "mock pass did not finish after it was released") + + if active := manager.GetActiveAlerts(); len(active) != 0 { + t.Fatalf("a mock pass in flight when mock mode ended left %d fixture alerts in live mode: %s", + len(active), describeAlerts(active)) + } +} + +func TestLeavingMockModeWaitsForAnInFlightEvaluation(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + } + mustSetMonitorMockMode(t, monitor, true) + + // Park a real pass inside its first guest evaluation, in the CPU window + // read the alert manager makes outside its own lock, so nothing but the + // fence holds the switch back. + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + manager.SetMetricWindowProvider(func(alerts.MetricWindowRequest) ([]alerts.MetricWindowPoint, error) { + once.Do(func() { + close(entered) + <-release + }) + return nil, nil + }) + released := false + defer func() { + if !released { + close(release) + } + }() + passDone := make(chan struct{}) + go func() { + defer close(passDone) + monitor.checkMockAlerts() + }() + waitForSignal(t, entered, 30*time.Second, "mock pass never evaluated a windowed metric") + + // The switch must stop in advance until the evaluation returns. One + // that finishes first cleared alerts the evaluation can still raise. + leftMockMode := switchMockModeAsync(t, monitor, false) + waitForCondition(t, 30*time.Second, func() bool { + select { + case <-leftMockMode: + return true + default: + } + return goroutineIn("monitoring.(*mockModeFence).advance", "monitoring.(*Monitor).SetMockMode") + }, "leaving mock mode neither finished nor waited for the running evaluation") + select { + case <-leftMockMode: + t.Fatal("leaving mock mode cleared alerts while a fixture evaluation was still running") + default: + } + + close(release) + released = true + waitForSignal(t, leftMockMode, 30*time.Second, "leaving mock mode did not finish once the evaluation returned") + waitForSignal(t, passDone, 30*time.Second, "mock pass did not finish after it was released") + if active := manager.GetActiveAlerts(); len(active) != 0 { + t.Fatalf("the evaluation running when mock mode ended left %d fixture alerts in live mode: %s", + len(active), describeAlerts(active)) + } +} + +// blockingFixtureRecords serves one source's fixture records the way the +// router's mock adapter does, reading them on each call, and holds its first +// call until released. Later calls are not held, not even while the first is. +type blockingFixtureRecords struct { + source unifiedresources.DataSource + entered chan struct{} + release chan struct{} + held atomic.Bool +} + +func newBlockingFixtureRecords(source unifiedresources.DataSource) *blockingFixtureRecords { + return &blockingFixtureRecords{source: source, entered: make(chan struct{}), release: make(chan struct{})} +} + +func (p *blockingFixtureRecords) SupplementalRecords(_ *Monitor, _ string) []unifiedresources.IngestRecord { + records := mock.SupplementalRecords(p.source) + if p.held.CompareAndSwap(false, true) { + close(p.entered) + <-p.release + } + return records +} + +func TestLeavingMockModeDiscardsAnInFlightFrontendRefresh(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + provider := newBlockingFixtureRecords(unifiedresources.SourceVMware) + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + resourceStore: unifiedresources.NewMonitorAdapter(nil), + supplementalProviders: map[unifiedresources.DataSource]MonitorSupplementalRecordsProvider{ + unifiedresources.SourceVMware: provider, + }, + } + mustSetMonitorMockMode(t, monitor, true) + + // Hold a frontend read in its registry rebuild: it has the fixture + // snapshot and the vSphere fixture records, whose datastore, host and + // network incidents its unified alert sync would raise. + released := false + defer func() { + if !released { + close(provider.release) + } + }() + readDone := make(chan struct{}) + go func() { + defer close(readDone) + monitor.BuildFrontendState() + }() + waitForSignal(t, provider.entered, 30*time.Second, "frontend read never reached the supplemental records") + + waitForSignal(t, switchMockModeAsync(t, monitor, false), 30*time.Second, "leaving mock mode waited on a registry rebuild") + close(provider.release) + released = true + waitForSignal(t, readDone, 30*time.Second, "frontend read did not finish after it was released") + + if active := manager.GetActiveAlerts(); len(active) != 0 { + t.Fatalf("a frontend read in flight when mock mode ended left %d fixture alerts in live mode: %s", + len(active), describeAlerts(active)) + } +} + +// gatedListStore holds the first registry listing made after open is called, +// so a test can publish another generation between a refresh's own publish +// and its read back. +type gatedListStore struct { + *unifiedresources.MonitorAdapter + mu sync.Mutex + armed bool + entered chan struct{} + release chan struct{} +} + +func (s *gatedListStore) arm() { + s.mu.Lock() + defer s.mu.Unlock() + s.armed = true +} + +func (s *gatedListStore) GetAll() []unifiedresources.Resource { + s.mu.Lock() + hold := s.armed + s.armed = false + s.mu.Unlock() + if hold { + close(s.entered) + <-s.release + } + return s.MonitorAdapter.GetAll() +} + +// fixedRecords stands in for a live provider: it serves the same records in +// either mode. +type fixedRecords []unifiedresources.IngestRecord + +func (r fixedRecords) SupplementalRecords(_ *Monitor, _ string) []unifiedresources.IngestRecord { + return r +} + +func alertTypes(active []alerts.Alert) map[string]int { + types := make(map[string]int) + for _, alert := range active { + types[alert.Type]++ + } + return types +} + +func TestLiveRefreshIgnoresARegistryRepublishedFromMockMode(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + mustSetMockEnabled(t, true) + // TrueNAS records stand in for a live provider's data; the vSphere + // fixture source serves nothing once mock mode is off. + liveTrueNAS := fixedRecords(mock.SupplementalRecords(unifiedresources.SourceTrueNAS)) + mustSetMockEnabled(t, false) + if len(liveTrueNAS) == 0 { + t.Fatal("mock fixture has no TrueNAS records") + } + provider := newBlockingFixtureRecords(unifiedresources.SourceVMware) + store := &gatedListStore{ + MonitorAdapter: unifiedresources.NewMonitorAdapter(nil), + entered: make(chan struct{}), + release: make(chan struct{}), + } + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + resourceStore: store, + supplementalProviders: map[unifiedresources.DataSource]MonitorSupplementalRecordsProvider{ + unifiedresources.SourceVMware: provider, + unifiedresources.SourceTrueNAS: liveTrueNAS, + }, + } + mustSetMonitorMockMode(t, monitor, true) + releasedProvider, releasedList := false, false + defer func() { + if !releasedProvider { + close(provider.release) + } + if !releasedList { + close(store.release) + } + }() + + // A mock-mode refresh holds the fixture snapshot and vSphere records and + // has not published them yet. + staleDone := make(chan struct{}) + go func() { + defer close(staleDone) + monitor.updateResourceStore(monitor.currentStateWithScope()) + }() + waitForSignal(t, provider.entered, 30*time.Second, "mock-mode refresh never reached the supplemental records") + waitForSignal(t, switchMockModeAsync(t, monitor, false), 30*time.Second, "leaving mock mode waited on a registry rebuild") + + // A live refresh publishes the live TrueNAS records alone and is held + // before reading the registry back. + store.arm() + liveDone := make(chan struct{}) + go func() { + defer close(liveDone) + monitor.updateResourceStore(monitor.currentStateWithScope()) + }() + waitForSignal(t, store.entered, 30*time.Second, "live refresh never read the registry back") + for _, resource := range store.MonitorAdapter.GetAll() { + if slices.Contains(resource.Sources, unifiedresources.SourceVMware) { + t.Fatalf("the live refresh published vSphere fixture resource %s", resource.ID) + } + } + + // The mock-mode refresh now publishes the fixture estate over it. + close(provider.release) + releasedProvider = true + waitForSignal(t, staleDone, 30*time.Second, "mock-mode refresh did not finish after it was released") + close(store.release) + releasedList = true + waitForSignal(t, liveDone, 30*time.Second, "live refresh did not finish after it was released") + + if active := manager.GetActiveAlerts(); len(active) != 0 { + t.Fatalf("a live refresh evaluated a registry the mock-mode refresh republished, leaving %d fixture alerts: %s", + len(active), describeAlerts(active)) + } + if monitor.mockModeFence.registryCurrent() { + t.Fatal("the registry the mock-mode refresh republished counted as current") + } + + // The next live refresh is undisturbed: it replaces the registry and + // evaluates the live records. + monitor.updateResourceStore(monitor.currentStateWithScope()) + if !monitor.mockModeFence.registryCurrent() { + t.Fatal("an undisturbed live rebuild did not make the registry current") + } + types := alertTypes(manager.GetActiveAlerts()) + if types["zfs-pool-state"] == 0 { + t.Fatalf("the undisturbed live refresh raised no alert from the live TrueNAS records: %v", types) + } + for _, fixtureType := range []string{"resource-incident", "storage-incident"} { + if types[fixtureType] != 0 { + t.Fatalf("vSphere fixture incidents reached live mode: %v", types) + } + } +} + +func TestDockerPruneIgnoresTheRegistryMockModeLeft(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + resourceStore: unifiedresources.NewMonitorAdapter(nil), + } + mustSetMonitorMockMode(t, monitor, true) + monitor.updateResourceStore(monitor.currentStateWithScope()) + mustSetMonitorMockMode(t, monitor, false) + + // Before any live rebuild replaces the fixture registry, a live Docker + // host reports and then goes offline. + host := models.DockerHost{ID: "live-docker-1", Hostname: "live-docker-1", DisplayName: "live-docker-1", Status: "offline"} + monitor.state.UpsertDockerHost(host) + for range 3 { + manager.HandleDockerHostOffline(host) + } + if types := alertTypes(manager.GetActiveAlerts()); types["docker-host-offline"] != 1 { + t.Fatalf("live Docker host raised no offline alert: %v", types) + } + + monitor.SyncAlertState() + if types := alertTypes(manager.GetActiveAlerts()); types["docker-host-offline"] != 1 { + t.Fatalf("Docker pruning removed a live host's alert against the fixture registry: %v", types) + } +} + +func TestPreviousStateCarryIgnoresTheRegistryMockModeLeft(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + monitor := &Monitor{ + state: models.NewState(), + alertManager: newMockHostAlertTestManager(t), + metricsHistory: NewMetricsHistory(10, time.Hour), + resourceStore: unifiedresources.NewMonitorAdapter(nil), + } + mustSetMonitorMockMode(t, monitor, true) + monitor.updateResourceStore(monitor.currentStateWithScope()) + fixture := mock.CurrentFixtureGraph().State + if len(fixture.Nodes) == 0 || len(fixture.VMs) == 0 { + t.Fatal("mock fixture has no nodes or VMs") + } + instance := fixture.VMs[0].Instance + mustSetMonitorMockMode(t, monitor, false) + + // A live instance named like a fixture one polls before any live rebuild + // replaces the fixture registry. Its previous-state carry, which writes + // back into live state, must not see the fixture nodes and guests. + if nodes := monitor.previousNodesForInstance(instance); len(nodes) != 0 { + t.Fatalf("previous-node carry for %q returned %d fixture nodes after leaving mock mode", instance, len(nodes)) + } + if guests := monitor.previousGuestContextForInstance(instance); len(guests.vms) != 0 || len(guests.containers) != 0 { + t.Fatalf("previous-guest carry for %q returned %d fixture VMs and %d containers after leaving mock mode", + instance, len(guests.vms), len(guests.containers)) + } +} + +func TestLeavingMockModeDiscardsAnInFlightBackupEvaluation(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + config := manager.GetConfig() + config.BackupDefaults.Enabled = true + config.BackupDefaults.WarningDays = 1 + config.BackupDefaults.CriticalDays = 2 + manager.UpdateConfig(config) + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + } + mustSetMonitorMockMode(t, monitor, true) + // Cache the fixture read view the guest lookup will be served from. + monitor.GetUnifiedReadStateOrSnapshot() + + // Park a real evaluation at its guest lookup, after it read the fixture + // recovery rollups whose stale backups raise backup-age alerts. + monitor.mockUnifiedViewMu.Lock() + parked := true + defer func() { + if parked { + monitor.mockUnifiedViewMu.Unlock() + } + }() + evalDone := make(chan struct{}) + go func() { + defer close(evalDone) + monitor.checkBackupAlerts(context.Background()) + }() + waitForGoroutineIn(t, "monitoring.(*Monitor).currentUnifiedStateView", "monitoring.(*Monitor).checkBackupAlerts") + + waitForSignal(t, switchMockModeAsync(t, monitor, false), 30*time.Second, "leaving mock mode waited on a backup evaluation parked outside the alert manager") + monitor.mockUnifiedViewMu.Unlock() + parked = false + waitForSignal(t, evalDone, 30*time.Second, "backup evaluation did not finish after it was released") + + if active := manager.GetActiveAlerts(); len(active) != 0 { + t.Fatalf("a backup evaluation in flight when mock mode ended left %d fixture alerts in live mode: %.500s", + len(active), describeAlerts(active)) + } +} + +func TestLeavingMockModeDiscardsAnInFlightConnectionCheck(t *testing.T) { + pinDefaultMockEstate(t) + manager := newMockHostAlertTestManager(t) + monitor := &Monitor{ + state: models.NewState(), + alertManager: manager, + metricsHistory: NewMetricsHistory(10, time.Hour), + } + mustSetMonitorMockMode(t, monitor, true) + + // A hybrid mock ledger lists real connections beside fixture ones, so the + // same connection can be read degraded in mock mode and again live. + degraded := alerts.ConnectionSnapshot{ + ID: "pve-cluster-a", + Name: "cluster-a", + Type: alerts.ConnectionTypePVE, + State: alerts.ConnectionStateUnreachable, + Enabled: true, + } + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + monitor.SetConnectionsSnapshotLister(func() []alerts.ConnectionSnapshot { + once.Do(func() { + close(entered) + <-release + }) + return []alerts.ConnectionSnapshot{degraded} + }) + released := false + defer func() { + if !released { + close(release) + } + }() + checkDone := make(chan struct{}) + go func() { + defer close(checkDone) + monitor.checkConnectionAlerts() + }() + waitForSignal(t, entered, 30*time.Second, "connection check never read the ledger") + + waitForSignal(t, switchMockModeAsync(t, monitor, false), 30*time.Second, "leaving mock mode waited on a connection check parked in its ledger read") + close(release) + released = true + waitForSignal(t, checkDone, 30*time.Second, "connection check did not finish after it was released") + + // The connection-degraded alert confirms on its third consecutive + // observation; the one read in mock mode must not be one of them. + monitor.checkConnectionAlerts() + monitor.checkConnectionAlerts() + if active := manager.GetActiveAlerts(); len(active) != 0 { + t.Fatalf("a connection observation read in mock mode counted toward live confirmation: %s", describeAlerts(active)) + } + monitor.checkConnectionAlerts() + if active := manager.GetActiveAlerts(); len(active) != 1 { + t.Fatalf("three live degraded observations raised %d alerts, want 1: %s", len(active), describeAlerts(active)) + } +} + // diskAlertKeys lists the active disk-health and disk-wearout alerts as // "type instance/node device" so assertions name the disk, not the alert ID. func diskAlertKeys(manager *alerts.Manager) map[string]bool { @@ -331,7 +900,7 @@ func TestCheckMockPhysicalDiskAlertsFollowsPollerBoundary(t *testing.T) { {Instance: "east", Node: "pve1", DevPath: "/dev/sdb", Model: "WD Red Pro 8TB", Type: "sata", Health: "FAILED", Wearout: -1}, } - monitor.checkMockPhysicalDiskAlerts(disks, nodes, hosts) + monitor.checkMockPhysicalDiskAlerts(monitor.mockModeFence.begin(), disks, nodes, hosts) got := diskAlertKeys(manager) if len(got) != 2 || !got["disk-health west/pve1 /dev/sda"] || !got["disk-wearout west/pve1 /dev/nvme0n1"] { t.Fatalf("disk alerts = %v, want only FAILED health on west/pve1 /dev/sda and wearout on west/pve1 /dev/nvme0n1", got) @@ -342,7 +911,7 @@ func TestCheckMockPhysicalDiskAlertsFollowsPollerBoundary(t *testing.T) { // recovery needs three healthy-looking passes. hosts[0].DiskExclude = []string{"/dev/sda", "nvme0n1"} for i := 0; i < 3; i++ { - monitor.checkMockPhysicalDiskAlerts(disks, nodes, hosts) + monitor.checkMockPhysicalDiskAlerts(monitor.mockModeFence.begin(), disks, nodes, hosts) } if got := diskAlertKeys(manager); len(got) != 0 { t.Fatalf("excluded devices kept disk alerts %v", got) @@ -351,7 +920,7 @@ func TestCheckMockPhysicalDiskAlertsFollowsPollerBoundary(t *testing.T) { // A node that leaves the estate takes its disk alerts with it through the // node cleanup the mock pass runs first. hosts[0].DiskExclude = nil - monitor.checkMockPhysicalDiskAlerts(disks, nodes, hosts) + monitor.checkMockPhysicalDiskAlerts(monitor.mockModeFence.begin(), disks, nodes, hosts) if got := diskAlertKeys(manager); len(got) != 2 { t.Fatalf("disk alerts after removing the exclusion = %v, want both west/pve1 alerts back", got) } diff --git a/internal/monitoring/monitor_pbs_coverage_test.go b/internal/monitoring/monitor_pbs_coverage_test.go index fa77a4eb2c..20f1674150 100644 --- a/internal/monitoring/monitor_pbs_coverage_test.go +++ b/internal/monitoring/monitor_pbs_coverage_test.go @@ -694,7 +694,7 @@ func TestPBSPolledCapacityRequiresObservedRecovery(t *testing.T) { for range 5 { monitor.pollPBSInstance(context.Background(), instance.Name, client) adapter.PopulateFromSnapshot(monitor.state.GetSnapshot()) - monitor.syncUnifiedResourceAlertsToState(adapter.GetAll()) + monitor.syncUnifiedResourceAlertsToState(adapter.GetAll(), monitor.mockModeFence.begin()) } } // Exercise the production poll-to-storage conversion, not just the diff --git a/internal/monitoring/monitor_polling_test.go b/internal/monitoring/monitor_polling_test.go index bde1dfa401..ab82cc4010 100644 --- a/internal/monitoring/monitor_polling_test.go +++ b/internal/monitoring/monitor_polling_test.go @@ -351,7 +351,7 @@ func TestUpdateResourceStoreSyncsUnifiedIncidentAlerts(t *testing.T) { }, } - monitor.updateResourceStore(models.StateSnapshot{}) + monitor.updateResourceStore(models.StateSnapshot{}, monitor.mockModeFence.begin()) active := alertManager.GetActiveAlerts() if len(active) != 1 { @@ -412,7 +412,7 @@ func TestUpdateResourceStoreSyncsCanonicalStorageMetrics(t *testing.T) { }, } - monitor.updateResourceStore(models.StateSnapshot{}) + monitor.updateResourceStore(models.StateSnapshot{}, monitor.mockModeFence.begin()) memory := monitor.GetStorageMetrics("pool:tank", time.Hour) if got := len(memory["usage"]); got == 0 { @@ -459,7 +459,7 @@ func TestUpdateResourceStoreSyncsCanonicalStorageMetrics(t *testing.T) { metricsStore: store, } - monitor.updateResourceStore(snapshot) + monitor.updateResourceStore(snapshot, monitor.mockModeFence.begin()) var storageResourceID string for _, resource := range resourceStore.GetAll() { @@ -524,7 +524,7 @@ func TestBuildBroadcastFrontendStateIncludesUnifiedIncidentAlerts(t *testing.T) }, } - frontend := monitor.buildBroadcastFrontendStateFromSnapshot(models.StateSnapshot{}) + frontend := monitor.buildBroadcastFrontendStateFromSnapshot(models.StateSnapshot{}, monitor.mockModeFence.begin()) if len(frontend.ActiveAlerts) != 1 { t.Fatalf("expected 1 active alert in frontend state, got %d", len(frontend.ActiveAlerts)) @@ -1270,7 +1270,7 @@ func TestBuildBroadcastFrontendStatePrefersLiveAlertManagerOverSnapshotAlerts(t }}, } - frontend := monitor.buildBroadcastFrontendStateFromSnapshot(staleSnapshot) + frontend := monitor.buildBroadcastFrontendStateFromSnapshot(staleSnapshot, monitor.mockModeFence.begin()) if len(frontend.ActiveAlerts) != 1 { t.Fatalf("expected 1 live active alert in frontend state, got %d", len(frontend.ActiveAlerts)) @@ -1396,7 +1396,7 @@ func TestBuildBroadcastFrontendStatePreservesSnapshotAlertsWithoutAlertManager(t }}, } - frontend := monitor.buildBroadcastFrontendStateFromSnapshot(snapshot) + frontend := monitor.buildBroadcastFrontendStateFromSnapshot(snapshot, monitor.mockModeFence.begin()) if len(frontend.ActiveAlerts) != 1 { t.Fatalf("expected snapshot alert to be preserved, got %#v", frontend.ActiveAlerts) } @@ -2274,7 +2274,7 @@ func TestUpdateResourceStore_IngestsSupplementalRecords(t *testing.T) { t.Fatalf("RegisterPollProvider failed: %v", err) } - monitor.updateResourceStore(models.StateSnapshot{}) + monitor.updateResourceStore(models.StateSnapshot{}, monitor.mockModeFence.begin()) if store.snapshotCalls != 1 { t.Fatalf("expected PopulateFromSnapshot to be called once, got %d", store.snapshotCalls) @@ -2328,7 +2328,7 @@ func TestUpdateResourceStore_RecordsSupplementalChanges(t *testing.T) { t.Fatalf("RegisterPollProvider failed: %v", err) } - monitor.updateResourceStore(models.StateSnapshot{}) + monitor.updateResourceStore(models.StateSnapshot{}, monitor.mockModeFence.begin()) if len(store.recordedChanges) != 1 { t.Fatalf("expected 1 supplemental change recorded, got %d", len(store.recordedChanges)) @@ -2389,7 +2389,7 @@ func TestUpdateResourceStore_SuppressesProviderOwnedSnapshotSources(t *testing.T PhysicalDisks: []models.PhysicalDisk{{}}, CephClusters: []models.CephCluster{{}}, Hosts: []models.Host{{}}, - }) + }, monitor.mockModeFence.begin()) if store.snapshotCalls != 1 { t.Fatalf("expected PopulateFromSnapshot to be called once, got %d", store.snapshotCalls) @@ -2435,7 +2435,7 @@ func TestUpdateResourceStore_IngestsRegisteredSupplementalProvider(t *testing.T) monitor.SetSupplementalRecordsProvider(unifiedresources.SourceTrueNAS, provider) store.snapshotCalls = 0 store.recordsBySource = nil - monitor.updateResourceStore(models.StateSnapshot{}) + monitor.updateResourceStore(models.StateSnapshot{}, monitor.mockModeFence.begin()) if store.snapshotCalls != 1 { t.Fatalf("expected PopulateFromSnapshot to be called once, got %d", store.snapshotCalls) @@ -2483,7 +2483,7 @@ func TestUpdateResourceStore_UsesAtomicStoreReplacementWhenAvailable(t *testing. Hosts: []models.Host{{ID: "host-1", Hostname: "minipc", Status: "online"}}, } - monitor.updateResourceStore(snapshot) + monitor.updateResourceStore(snapshot, monitor.mockModeFence.begin()) if store.atomicCalls != 1 { t.Fatalf("expected atomic populate to be called once, got %d", store.atomicCalls) @@ -2535,7 +2535,7 @@ func TestUpdateResourceStore_SuppressesSnapshotForRegisteredSupplementalOwnershi VMs: []models.VM{{}}, Containers: []models.Container{{}}, Hosts: []models.Host{{}}, - }) + }, monitor.mockModeFence.begin()) if len(store.lastSnapshot.Nodes) != 0 || len(store.lastSnapshot.VMs) != 0 || len(store.lastSnapshot.Containers) != 0 { t.Fatalf("expected proxmox slices to be suppressed for direct provider ownership") diff --git a/internal/monitoring/monitor_previous_state.go b/internal/monitoring/monitor_previous_state.go index 79568436ef..a186647fab 100644 --- a/internal/monitoring/monitor_previous_state.go +++ b/internal/monitoring/monitor_previous_state.go @@ -27,7 +27,7 @@ func (m *Monitor) previousGuestContextForInstance(instanceName string) previousG hostAgentsByVMID: make(map[string]models.Host), } - readState := m.GetUnifiedReadStateOrSnapshot() + readState := m.currentModeReadState() if readState == nil { return ctx } @@ -93,7 +93,7 @@ func (m *Monitor) previousGuestContextForInstance(instanceName string) previousG func (m *Monitor) previousNodesForInstance(instanceName string) []models.Node { prevInstanceNodes := make([]models.Node, 0) - readState := m.GetUnifiedReadStateOrSnapshot() + readState := m.currentModeReadState() if readState == nil { return prevInstanceNodes } diff --git a/internal/monitoring/recovery_rollups.go b/internal/monitoring/recovery_rollups.go index 34120189d6..2905df5f3f 100644 --- a/internal/monitoring/recovery_rollups.go +++ b/internal/monitoring/recovery_rollups.go @@ -101,6 +101,9 @@ func (m *Monitor) checkBackupAlerts(ctx context.Context) { } defer m.backupAlertEvalMu.Unlock() + // Rollups and the guest lookup come from the fixture graph in mock mode; + // an evaluation that read them must not land after a mode change. + scope := m.mockModeFence.begin() cfg := m.alertManager.GetConfig() if !cfg.Enabled || !cfg.BackupDefaults.Enabled || (cfg.BackupDefaults.WarningDays <= 0 && cfg.BackupDefaults.CriticalDays <= 0) { @@ -119,19 +122,23 @@ func (m *Monitor) checkBackupAlerts(ctx context.Context) { // deadline, including connection-pool waiting, still needs visibility. if !errors.Is(err, context.Canceled) { log.Warn().Err(err).Msg("Failed to list recovery rollups for backup alerts") - m.alertManager.RaiseSystemAlert(alerts.SystemAlertInput{ - Type: alerts.BackupEvaluationAlertType, - Level: alerts.AlertLevelWarning, - Fingerprint: "recovery-rollups-unavailable", - Message: "Backup-age alerts were not evaluated because recovery data could not be read. Existing backup alerts have been kept; check Pulse's logs.", + scope.run(func() { + m.alertManager.RaiseSystemAlert(alerts.SystemAlertInput{ + Type: alerts.BackupEvaluationAlertType, + Level: alerts.AlertLevelWarning, + Fingerprint: "recovery-rollups-unavailable", + Message: "Backup-age alerts were not evaluated because recovery data could not be read. Existing backup alerts have been kept; check Pulse's logs.", + }) }) } return } - guestsByKey, guestsByVMID := buildGuestLookupsFromReadState(m.GetUnifiedReadStateOrSnapshot(), m.guestMetadataStore) - m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts()) - // Only a complete read/evaluation (including a genuinely empty result) - // resolves the outage. In-flight checks are coalesced, not queued, so an - // older successful read cannot later erase a newer failure. - m.alertManager.ClearSystemAlert(alerts.BackupEvaluationAlertType) + guestsByKey, guestsByVMID := buildGuestLookupsFromReadState(m.currentModeReadState(), m.guestMetadataStore) + scope.run(func() { + m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts()) + // Only a complete read/evaluation (including a genuinely empty result) + // resolves the outage. In-flight checks are coalesced, not queued, so an + // older successful read cannot later erase a newer failure. + m.alertManager.ClearSystemAlert(alerts.BackupEvaluationAlertType) + }) }