diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 131f24bb44..5f8cdcdef0 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -9735,3 +9735,20 @@ The accepted upstream `diagnostics_token_visibility_test.go` controls and API reference route checks cover caller filtering and the documented authority. Final-source checks must retain this upstream behavior beside the reviewed reporting/help repairs; source proof is not installed credential recovery. + +### PVE guest-cycle publication and independent agents (12 October 2026) + +Ending a PVE connection/runtime cancels its in-flight guest collection and rejects +late guest inventory, diagnostics, alerts, backup-template eligibility and +History. Collection ownership is separate from consumed guest-command ownership: +cancellation never proves a guest command stopped or clears a shared uncertainty +fence. Transport replacement alone does not invalidate accepted OS-command +safety or renew an existing cache's age. Independent Pulse-agent reports and +stored metric observations remain intact when a guest cycle is withdrawn. + +`TestGuestPollOwnershipRuntimeDoesNotWithdrawIndependentAgentEvidence` in +`internal/monitoring/monitor_host_agents_test.go` checks those independent +sources through a runtime transition. Production collection/late-reply and +publication-order controls are in `guest_poll_ownership_test.go`; diagnostics +and both History stores are covered in `diagnostic_snapshots_test.go`. These +controls establish no native guest restart, thaw or prior-workload restoration. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index ce4760aa7f..68618092f9 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -7283,3 +7283,34 @@ and never queue behind a warm ingest; a cold read still builds its first registr consume-once replacement-store controls, refused and skipped drains, read-provider re-entry, and duplicate/loss assertions, alongside the original initialization, tenant withdrawal, saved first-alert policy and joined-shutdown controls. + +### Guest-cycle publication ownership (12 October 2026) + +The production guest cycle stages diagnostic snapshots, alert evaluations, +backup-template eligibility and History until both VM/container collections are +ready. It admits publication only while its registered PVE client and cycle +remain current. Retirement, transport replacement and a new runtime cancel +outstanding guest cycles and order their state changes against any already +admitted publication; late inventories/details cannot recreate or erase the +replacement's guest readings. An interrupted inventory does not grant fallback. +One cycle per live binding owns collection until it completes, including when +its caller is cancelled. Another connection can still advance independently. + +Deep-enrichment expiry is distinct from parent-cycle cancellation: a complete +authoritative inventory still publishes within the reserved parent tail. An +accepted collection retains existing failed-node/grace rules, empty-inventory +handling and metric/source/age policy. Focused builder and collector entry points +retain their existing immediate effects. No guest operation lock, consumed-read +uncertainty, timeout, permission, TLS policy or automatic retry is relaxed. +Client replacement is not a guest reboot and does not erase accepted cache age +or OS-command safety state. Remote collection holds no publication lock; effects +run without the monitor state mutex held. The guest publication gate does not +claim ownership of node, storage, backup-job or native Docker-command collection. + +`guest_poll_ownership_test.go`, `diagnostic_snapshots_test.go`, +`monitor_host_agents_test.go`, `proxmox_guest_snapshot_coherence_test.go` and +`proxmox_large_cluster_poll_budget_test.go` cover late replies on both enumeration +routes, cancellation/retirement/runtime/client changes, complete staged readings, +History and breach continuity, independent agents, single-flight/recovery, +retirement ordering and bounded-detail inventory. These are controlled source +proofs, not native restart, independent thaw, installed or release acceptance. diff --git a/internal/monitoring/diagnostic_snapshots_test.go b/internal/monitoring/diagnostic_snapshots_test.go index d03f92e431..b6ec68cfdb 100644 --- a/internal/monitoring/diagnostic_snapshots_test.go +++ b/internal/monitoring/diagnostic_snapshots_test.go @@ -1,6 +1,7 @@ package monitoring import ( + "context" "encoding/json" "strings" "testing" @@ -1395,3 +1396,30 @@ func TestGuestMemorySampleCapacityContract(t *testing.T) { t.Run("selection", testGuestMemorySampleCapacitySelection) t.Run("cache-binding", testGuestMemorySampleCapacityCacheBinding) } + +func TestGuestPollOwnershipStagesAllReadingEffects(t *testing.T) { + c := &guestPollOwnershipClient{cpu: .95} + m := guestPollOwnershipMonitor(t, c) + ctx, owner := m.beginPVEGuestPoll(context.Background(), "site-a", c) + if owner == nil { + t.Fatal("registered cycle not admitted") + } + defer m.finishPVEGuestPoll("site-a", owner) + resources := c.resources() + vm, ok := m.handleClusterVMResource(ctx, "site-a", resources[0], "site-a:node:105", c, nil, nil) + if !ok { + t.Fatal("VM builder lost inventory") + } + ct, ok := m.handleClusterContainerResource(ctx, "site-a", resources[1], "site-a:node:106", c, nil, nil) + if !ok { + t.Fatal("container builder lost inventory") + } + stagePVEGuestEffect(ctx, func() { m.recordGuestMetrics([]models.VM{vm}, []models.Container{ct}, time.Time{}) }) + if len(m.GetDiagnosticSnapshots().Guests) != 0 || len(m.metricsHistory.GetGuestMetrics(vm.ID, "cpu", time.Hour)) != 0 || len(m.state.GetSnapshot().VMs) != 0 { + t.Fatal("a partial builder published before the complete guest generation") + } + m.publishPVEGuests(ctx, "site-a", []models.VM{vm}, []models.Container{ct}) + if len(m.GetDiagnosticSnapshots().Guests) != 2 || len(m.state.GetSnapshot().VMs) != 1 || len(m.metricsHistory.GetGuestMetrics(vm.ID, "cpu", time.Hour)) != 1 || len(guestHistoryStoredPoints(t, m, "vm", vm.ID, "cpu")) != 1 { + t.Fatal("accepted generation lost diagnostics or in-memory/persistent History") + } +} diff --git a/internal/monitoring/guest_poll_ownership_test.go b/internal/monitoring/guest_poll_ownership_test.go index e1537047d4..4136f96130 100644 --- a/internal/monitoring/guest_poll_ownership_test.go +++ b/internal/monitoring/guest_poll_ownership_test.go @@ -5,6 +5,7 @@ import ( "fmt" "reflect" "sync" + "sync/atomic" "testing" "time" @@ -25,6 +26,7 @@ type guestPollOwnershipClient struct { release chan struct{} block string once sync.Once + calls atomic.Int32 } func (c *guestPollOwnershipClient) wait(ctx context.Context, phase string) { @@ -46,6 +48,7 @@ func (c *guestPollOwnershipClient) resources() []proxmox.ClusterResource { } func (c *guestPollOwnershipClient) GetClusterResources(ctx context.Context, _ string) ([]proxmox.ClusterResource, error) { + c.calls.Add(1) if c.route == "node" { return nil, fmt.Errorf("cluster/resources unavailable") } @@ -53,6 +56,93 @@ func (c *guestPollOwnershipClient) GetClusterResources(ctx context.Context, _ st return c.resources(), nil } +func TestGuestPollOwnershipSingleFlightAndOrdinaryRecovery(t *testing.T) { + for _, canceled := range []bool{false, true} { + t.Run(fmt.Sprint(canceled), func(t *testing.T) { + c := &guestPollOwnershipClient{cpu: .1, block: "inventory", entered: make(chan context.Context, 1), release: make(chan struct{})} + m := guestPollOwnershipMonitor(t, c) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var once sync.Once + unblock := func() { once.Do(func() { close(c.release) }) } + defer unblock() + done := make(chan error, 1) + go func() { done <- runGuestOwnershipPoll(m, ctx, c) }() + select { + case <-c.entered: + case <-time.After(5 * time.Second): + t.Fatal("first inventory did not start") + } + if canceled { + cancel() + } + // A repeat cannot issue even an inventory GET while this binding's + // first cycle still owns the outstanding read. + if err := runGuestOwnershipPoll(m, context.Background(), c); err != nil { + t.Fatal(err) + } + if c.calls.Load() != 1 { + t.Error("overlapping/canceled cycles stacked inventory reads") + } + unblock() + waitGuestOwnership(t, done) + if err := runGuestOwnershipPoll(m, context.Background(), c); err != nil { + t.Fatal(err) + } + got := m.state.GetSnapshot() + if c.calls.Load() != 2 || len(got.VMs) != 1 || len(got.Containers) != 1 || len(m.pveGuestPolls) != 0 { + t.Fatal("ordinary completed-cycle recovery lost inventory or leaked ownership") + } + }) + } +} + +func TestGuestPollOwnershipRetirementWaitsForAdmittedPublication(t *testing.T) { + c := &guestPollOwnershipClient{} + m := guestPollOwnershipMonitor(t, c) + ctx, owner := m.beginPVEGuestPoll(context.Background(), "site-a", c) + if owner == nil { + t.Fatal("registered cycle not admitted") + } + defer m.finishPVEGuestPoll("site-a", owner) + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + stagePVEGuestEffect(ctx, func() { + // Effects may read the monitor; publication must not hold m.mu. + if _, ok := m.getPVEClient("site-a"); !ok { + t.Error("publication lost its connection before its admitted effects") + } + close(entered) + <-release + m.recordGuestSnapshot("site-a", "qemu", "node", 105, GuestMemorySnapshot{Name: "accepted"}) + }) + done := make(chan error, 1) + go func() { + m.publishPVEGuests(ctx, "site-a", []models.VM{{ID: "site-a:node:105", Instance: "site-a", Node: "node", VMID: 105}}, nil) + done <- nil + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("publication effect did not start or could not re-enter monitor") + } + retired := make(chan error, 1) + go func() { m.retirePVEInstanceRuntime("site-a"); retired <- nil }() + select { + case <-retired: + t.Error("retirement completed midway through an admitted guest batch") + case <-time.After(30 * time.Millisecond): + } + unblock() + waitGuestOwnership(t, done) + waitGuestOwnership(t, retired) + if len(m.state.GetSnapshot().VMs) != 0 || len(m.GetDiagnosticSnapshots().Guests) != 0 { + t.Error("publication resurrected retired guest state/diagnostics") + } +} + func (c *guestPollOwnershipClient) GetVMs(ctx context.Context, _ string) ([]proxmox.VM, error) { c.wait(ctx, "inventory") var rows []proxmox.VM diff --git a/internal/monitoring/guest_poll_publication.go b/internal/monitoring/guest_poll_publication.go new file mode 100644 index 0000000000..2d152453e4 --- /dev/null +++ b/internal/monitoring/guest_poll_publication.go @@ -0,0 +1,124 @@ +package monitoring + +import ( + "context" + "sync" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +// A guest cycle collects outside publication locks. Its effects are staged +// until both guest kinds are ready and the registered connection still owns +// this cycle. Child enrichment deadlines do not invalidate an authoritative +// inventory; cancellation of the parent cycle does. +type pveGuestPoll struct { + ctx context.Context + cancel context.CancelFunc + stopRuntime func() bool + client PVEClientInterface + mu sync.Mutex + effects []func() +} + +type pveGuestPollContextKey struct{} + +func (m *Monitor) beginPVEGuestPoll(ctx context.Context, instance string, client PVEClientInterface) (context.Context, *pveGuestPoll) { + m.pveGuestPublicationMu.RLock() + defer m.pveGuestPublicationMu.RUnlock() + m.mu.Lock() + defer m.mu.Unlock() + if ctx.Err() != nil || client == nil || m.pveClients[instance] != client || (m.runtimeCtx != nil && m.runtimeCtx.Err() != nil) { + return ctx, nil + } + if previous := m.pveGuestPolls[instance]; previous != nil { + if previous.client == client { + // Cancellation is not completion. Do not stack another inventory or + // release consumed-command fences while its owner is still running. + return ctx, nil + } + previous.cancel() + } + ownedCtx, cancel := context.WithCancel(ctx) + owner := &pveGuestPoll{ctx: ownedCtx, cancel: cancel, client: client} + if m.runtimeCtx != nil { + owner.stopRuntime = context.AfterFunc(m.runtimeCtx, cancel) + } + if m.pveGuestPolls == nil { + m.pveGuestPolls = make(map[string]*pveGuestPoll) + } + m.pveGuestPolls[instance] = owner + return context.WithValue(ownedCtx, pveGuestPollContextKey{}, owner), owner +} + +func (m *Monitor) finishPVEGuestPoll(instance string, owner *pveGuestPoll) { + owner.cancel() + if owner.stopRuntime != nil { + owner.stopRuntime() + } + m.mu.Lock() + if m.pveGuestPolls[instance] == owner { + delete(m.pveGuestPolls, instance) + } + m.mu.Unlock() +} + +// Focused builder/collector callers retain their existing immediate effects. +// Only the production connection cycle installs a batch, inherited by all its +// concurrent builders. A completed partial builder is not a published cycle. +func stagePVEGuestEffect(ctx context.Context, effect func()) { + if owner, _ := ctx.Value(pveGuestPollContextKey{}).(*pveGuestPoll); owner != nil { + owner.mu.Lock() + owner.effects = append(owner.effects, effect) + owner.mu.Unlock() + return + } + effect() +} + +func (m *Monitor) publishPVEGuests(ctx context.Context, instance string, vms []models.VM, containers []models.Container) { + owner, _ := ctx.Value(pveGuestPollContextKey{}).(*pveGuestPoll) + if owner == nil { + m.state.UpdateGuestsForInstance(instance, vms, containers) + return + } + // Retirement/client/runtime changes take the writer side and therefore + // cannot clear or replace the connection midway through an admitted batch. + // m.mu is released before effects (including alert callbacks) run. + m.pveGuestPublicationMu.RLock() + defer m.pveGuestPublicationMu.RUnlock() + m.mu.RLock() + current := owner.ctx.Err() == nil && m.pveGuestPolls[instance] == owner && m.pveClients[instance] == owner.client && (m.runtimeCtx == nil || m.runtimeCtx.Err() == nil) + m.mu.RUnlock() + if !current { + return + } + m.state.UpdateGuestsForInstance(instance, vms, containers) + owner.mu.Lock() + effects := owner.effects + owner.effects = nil + owner.mu.Unlock() + for _, effect := range effects { + effect() + } +} + +// Called with m.mu and the publication writer held. Replacing a transport is +// not a guest reboot: accepted cache ages and OS-command safety stay intact. +func (m *Monitor) cancelPVEGuestPollLocked(instance string) { + if owner := m.pveGuestPolls[instance]; owner != nil { + owner.cancel() + delete(m.pveGuestPolls, instance) + } +} + +func (m *Monitor) replacePVEClient(instance string, client PVEClientInterface) { + m.pveGuestPublicationMu.Lock() + defer m.pveGuestPublicationMu.Unlock() + m.mu.Lock() + defer m.mu.Unlock() + m.cancelPVEGuestPollLocked(instance) + if m.pveClients == nil { + m.pveClients = make(map[string]PVEClientInterface) + } + m.pveClients[instance] = client +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 4c98450f7e..625f7cd32a 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1306,6 +1306,8 @@ type Monitor struct { mockUnifiedViewFixtureAt time.Time // fixture freshness the published view was built from mockUnifiedViewValid bool pveClients map[string]PVEClientInterface + pveGuestPublicationMu sync.RWMutex // Orders guest publication with connection/runtime replacement and retirement. + pveGuestPolls map[string]*pveGuestPoll // Guarded by m.mu; canceled cycles retain ownership until completion unless the binding ends. pbsClients map[string]*pbs.Client pmgClients map[string]*pmg.Client availabilityStatuses map[string]AvailabilityProbeStatus @@ -1504,8 +1506,13 @@ type Monitor struct { } func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) { + m.pveGuestPublicationMu.Lock() + defer m.pveGuestPublicationMu.Unlock() m.mu.Lock() defer m.mu.Unlock() + for name := range m.pveGuestPolls { + m.cancelPVEGuestPollLocked(name) + } // A new runtime invalidates detached replication reads even when the old // caller has not yet canceled its parent context. for name, poll := range m.pveReplicationPolls { diff --git a/internal/monitoring/monitor_client_init.go b/internal/monitoring/monitor_client_init.go index 6e35ffb899..7a0b318dd0 100644 --- a/internal/monitoring/monitor_client_init.go +++ b/internal/monitoring/monitor_client_init.go @@ -44,7 +44,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) { endpoints, endpointFingerprints, ) - m.pveClients[pve.Name] = clusterClient + m.replacePVEClient(pve.Name, clusterClient) log.Info(). Str("instance", pve.Name). Str("cluster", pve.ClusterName). @@ -73,7 +73,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) { m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, false) continue } - m.pveClients[pve.Name] = client + m.replacePVEClient(pve.Name, client) log.Info().Str("instance", pve.Name).Msg("PVE client created successfully") // Set initial connection health to true m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, true) diff --git a/internal/monitoring/monitor_client_reconnect.go b/internal/monitoring/monitor_client_reconnect.go index e948f06713..f261258f0b 100644 --- a/internal/monitoring/monitor_client_reconnect.go +++ b/internal/monitoring/monitor_client_reconnect.go @@ -116,9 +116,7 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) { clientConfig.Timeout = connectionTimeout clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints) - m.mu.Lock() - m.pveClients[pve.Name] = clusterClient - m.mu.Unlock() + m.replacePVEClient(pve.Name, clusterClient) m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, true) log.Info(). @@ -140,9 +138,7 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) { continue } - m.mu.Lock() - m.pveClients[pve.Name] = client - m.mu.Unlock() + m.replacePVEClient(pve.Name, client) m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, true) log.Info(). diff --git a/internal/monitoring/monitor_cluster_dedupe.go b/internal/monitoring/monitor_cluster_dedupe.go index cc98b554d0..9068ba9ba6 100644 --- a/internal/monitoring/monitor_cluster_dedupe.go +++ b/internal/monitoring/monitor_cluster_dedupe.go @@ -76,6 +76,8 @@ func (m *Monitor) retirePVEInstanceRuntime(instanceName string) { if m == nil || instanceName == "" { return } + m.pveGuestPublicationMu.Lock() + defer m.pveGuestPublicationMu.Unlock() m.retireGuestMemoryCache(instanceName) if m.taskQueue != nil { @@ -88,6 +90,7 @@ func (m *Monitor) retirePVEInstanceRuntime(instanceName string) { key := schedulerKey(InstanceTypePVE, instanceName) m.mu.Lock() + m.cancelPVEGuestPollLocked(instanceName) if poll := m.pveReplicationPolls[instanceName]; poll != nil { poll.cancel() delete(m.pveReplicationPolls, instanceName) diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 8e1fa106e5..f78c83e85a 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -31,6 +31,33 @@ import ( "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" ) +func TestGuestPollOwnershipRuntimeDoesNotWithdrawIndependentAgentEvidence(t *testing.T) { + c := &guestPollOwnershipClient{cpu: .1} + m := guestPollOwnershipMonitor(t, c) + m.state.UpsertHost(models.Host{ID: "independent", Hostname: "agent", Status: "online", LastSeen: time.Now(), Memory: models.Memory{Total: 1000, Used: 300, Free: 700, Usage: 30}}) + hosts := m.state.GetSnapshot().Hosts + key := guestMetadataCacheKey("site-a", "node", 105) + metadata := guestMetadataCacheEntry{osName: "OpenBSD", osInfoFailureCount: guestAgentOSInfoFailureThreshold, osInfoSkip: true, fetchedAt: time.Now().Add(-time.Minute)} + m.guestMetadataCache[key] = metadata + memKey := guestMemoryCacheKey("site-a", "node", 105) + memory := agentMemCacheEntry{info: memoryLifecycleSample(4096), fetchedAt: time.Now().Add(-time.Minute)} + m.vmAgentMemCache = map[string]agentMemCacheEntry{memKey: memory} + ctx, owner := m.beginPVEGuestPoll(context.Background(), "site-a", c) + if owner == nil { + t.Fatal("registered cycle not admitted") + } + defer m.finishPVEGuestPoll("site-a", owner) + stagePVEGuestEffect(ctx, func() { m.recordGuestSnapshot("site-a", "qemu", "node", 105, GuestMemorySnapshot{Name: "obsolete"}) }) + m.setRuntimeContext(context.Background(), nil) + m.publishPVEGuests(ctx, "site-a", nil, nil) + if owner.ctx.Err() == nil || len(m.GetDiagnosticSnapshots().Guests) != 0 || !reflect.DeepEqual(m.state.GetSnapshot().Hosts, hosts) { + t.Fatal("runtime withdrawal admitted obsolete guest evidence or withdrew independent agent memory") + } + if !reflect.DeepEqual(m.guestMetadataCache[key], metadata) || m.vmAgentMemCache[memKey] != memory { + t.Fatal("runtime withdrawal erased OS-command safety or renewed accepted cache age") + } +} + func TestNewMonitorRoutesStartupCustomSensorWarningBeforeStart(t *testing.T) { monitor, err := New(&config.Config{DataPath: t.TempDir()}) if err != nil { diff --git a/internal/monitoring/monitor_polling_containers.go b/internal/monitoring/monitor_polling_containers.go index f9313d81a0..7fc4bff26b 100644 --- a/internal/monitoring/monitor_polling_containers.go +++ b/internal/monitoring/monitor_polling_containers.go @@ -298,24 +298,25 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s } // Trigger guest metadata migration if old format exists - if m.guestMetadataStore != nil { - m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, int(container.VMID)) - } + stagePVEGuestEffect(ctx, func() { + if m.guestMetadataStore != nil { + m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, int(container.VMID)) + } - m.recordGuestSnapshot(instanceName, modelContainer.Type, n.Node, int(container.VMID), GuestMemorySnapshot{ - Name: modelContainer.Name, - Status: modelContainer.Status, - RetrievedAt: modelContainer.LastSeen, - MemorySource: memorySource, - FallbackReason: guestMemoryFallbackReason(memorySource), - Memory: modelContainer.Memory, - Raw: guestRaw, + m.recordGuestSnapshot(instanceName, modelContainer.Type, n.Node, int(container.VMID), GuestMemorySnapshot{ + Name: modelContainer.Name, + Status: modelContainer.Status, + RetrievedAt: modelContainer.LastSeen, + MemorySource: memorySource, + FallbackReason: guestMemoryFallbackReason(memorySource), + Memory: modelContainer.Memory, + Raw: guestRaw, + }) + + // Check alerts + m.alertManager.CheckGuest(modelContainer, instanceName) }) - nodeContainers = append(nodeContainers, modelContainer) - - // Check alerts - m.alertManager.CheckGuest(modelContainer, instanceName) } nodeDuration := time.Since(nodeStart) @@ -355,7 +356,7 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s } } if failedNodes == 0 && successfulNodes > 0 { - m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects) + stagePVEGuestEffect(ctx, func() { m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects) }) } preservedContainers := 0 @@ -382,34 +383,36 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s // Record guest metrics history for running containers (enables sparkline/trends view) if !shouldSkipNativeMockStateMetricWrites() { now := time.Now() - for _, ct := range allContainers { - if !guestObservedInCycle(ct.LastSeen, startTime) { - continue + stagePVEGuestEffect(ctx, func() { + for _, ct := range allContainers { + if !guestObservedInCycle(ct.LastSeen, startTime) { + continue + } + if ct.Status != "running" { + continue + } + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates( + ct.DiskRead, + ct.DiskWrite, + ct.NetworkIn, + ct.NetworkOut, + ct.IORateValidity, + ) + m.recordGuestMetric( + "container", + ct.ID, + unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), + historyMemoryUsage(ct.Memory), + historyMemoryUsed(ct.Memory), + ct.Disk.Usage, + diskRead, + diskWrite, + networkIn, + networkOut, + now, + ) } - if ct.Status != "running" { - continue - } - diskRead, diskWrite, networkIn, networkOut := guestHistoryRates( - ct.DiskRead, - ct.DiskWrite, - ct.NetworkIn, - ct.NetworkOut, - ct.IORateValidity, - ) - m.recordGuestMetric( - "container", - ct.ID, - unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), - historyMemoryUsage(ct.Memory), - historyMemoryUsed(ct.Memory), - ct.Disk.Usage, - diskRead, - diskWrite, - networkIn, - networkOut, - now, - ) - } + }) } duration := time.Since(startTime) diff --git a/internal/monitoring/monitor_polling_vm.go b/internal/monitoring/monitor_polling_vm.go index 5c62784ad2..5ed8d32db2 100644 --- a/internal/monitoring/monitor_polling_vm.go +++ b/internal/monitoring/monitor_polling_vm.go @@ -100,7 +100,7 @@ func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string, } } if failedNodes == 0 && successfulNodes > 0 { - m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects) + stagePVEGuestEffect(ctx, func() { m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects) }) } preservedVMs := 0 @@ -122,34 +122,36 @@ func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string, if !shouldSkipNativeMockStateMetricWrites() { now := time.Now() - for _, vm := range allVMs { - if !guestObservedInCycle(vm.LastSeen, startTime) { - continue + stagePVEGuestEffect(ctx, func() { + for _, vm := range allVMs { + if !guestObservedInCycle(vm.LastSeen, startTime) { + continue + } + if vm.Status != "running" { + continue + } + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates( + vm.DiskRead, + vm.DiskWrite, + vm.NetworkIn, + vm.NetworkOut, + vm.IORateValidity, + ) + m.recordGuestMetric( + "vm", + vm.ID, + unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), + historyMemoryUsage(vm.Memory), + historyMemoryUsed(vm.Memory), + vm.Disk.Usage, + diskRead, + diskWrite, + networkIn, + networkOut, + now, + ) } - if vm.Status != "running" { - continue - } - diskRead, diskWrite, networkIn, networkOut := guestHistoryRates( - vm.DiskRead, - vm.DiskWrite, - vm.NetworkIn, - vm.NetworkOut, - vm.IORateValidity, - ) - m.recordGuestMetric( - "vm", - vm.ID, - unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), - historyMemoryUsage(vm.Memory), - historyMemoryUsed(vm.Memory), - vm.Disk.Usage, - diskRead, - diskWrite, - networkIn, - networkOut, - now, - ) - } + }) } duration := time.Since(startTime) diff --git a/internal/monitoring/monitor_pve.go b/internal/monitoring/monitor_pve.go index 25a6d5e1b0..3cac9a7b7c 100644 --- a/internal/monitoring/monitor_pve.go +++ b/internal/monitoring/monitor_pve.go @@ -441,8 +441,10 @@ func (m *Monitor) retryPVEPortFallback(ctx context.Context, instanceName string, var pbsInstances []config.PBSInstance var pmgInstances []config.PMGInstance + m.pveGuestPublicationMu.Lock() m.mu.Lock() instanceCfg.Host = persistHost + m.cancelPVEGuestPollLocked(instanceName) m.pveClients[instanceName] = fallbackClient // Update in-memory config so subsequent polls build clients against the working port. @@ -460,6 +462,7 @@ func (m *Monitor) retryPVEPortFallback(ctx context.Context, instanceName string, } persistence = m.persistence m.mu.Unlock() + m.pveGuestPublicationMu.Unlock() // Persist to disk so restarts keep the working endpoint. if persistence != nil { @@ -817,6 +820,11 @@ func (m *Monitor) pollGuestsWithFallback( return ctx.Err() default: } + ctx, owner := m.beginPVEGuestPoll(ctx, instanceName, client) + if owner == nil { + return nil // Another cycle or a replaced/ended binding owns guest work. + } + defer m.finishPVEGuestPoll(instanceName, owner) // Always try the efficient cluster/resources endpoint first // This endpoint works on both clustered and standalone nodes @@ -824,6 +832,9 @@ func (m *Monitor) pollGuestsWithFallback( useClusterEndpoint := m.pollVMsAndContainersEfficient(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodeEffectiveStatus) if !useClusterEndpoint { + if ctx.Err() != nil { + return ctx.Err() // An interrupted inventory is not a fallback grant. + } // Fall back to traditional polling only if cluster/resources not available // This should be rare - only for very old Proxmox versions log.Debug(). @@ -851,7 +862,7 @@ func (m *Monitor) pollGuestsWithFallback( if instanceCfg.MonitorContainers { containers = m.collectContainersWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus) } - m.state.UpdateGuestsForInstance(instanceName, vms, containers) + m.publishPVEGuests(ctx, instanceName, vms, containers) } return nil diff --git a/internal/monitoring/monitor_pve_cluster.go b/internal/monitoring/monitor_pve_cluster.go index d03383f925..965cf94905 100644 --- a/internal/monitoring/monitor_pve_cluster.go +++ b/internal/monitoring/monitor_pve_cluster.go @@ -682,7 +682,8 @@ func clusterEndpointIdentityChanged(existing, refreshed []config.ClusterEndpoint // rebuildPVEClusterClient swaps the failover client for a cluster instance so // it picks up the refreshed endpoint list. The in-flight poll keeps using the -// old client; the next poll cycle gets the new one. +// old client for node work; obsolete guest collections cannot publish. The +// next poll cycle uses the refreshed transport for guest collection too. func (m *Monitor) rebuildPVEClusterClient(instanceName string) { if m.config == nil || m.pveClients == nil { return @@ -704,9 +705,7 @@ func (m *Monitor) rebuildPVEClusterClient(instanceName string) { clientConfig.Timeout = m.config.ConnectionTimeout clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints) - m.mu.Lock() - m.pveClients[instanceName] = clusterClient - m.mu.Unlock() + m.replacePVEClient(instanceName, clusterClient) log.Info(). Str("instance", instanceName). diff --git a/internal/monitoring/monitor_pve_guest_poll.go b/internal/monitoring/monitor_pve_guest_poll.go index 5d3fd159c6..5263d5b121 100644 --- a/internal/monitoring/monitor_pve_guest_poll.go +++ b/internal/monitoring/monitor_pve_guest_poll.go @@ -53,7 +53,7 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam log.Debug().Err(err).Str("instance", instanceName).Msg("cluster/resources not available, falling back to traditional polling") return false } - m.updatePVEBackupTemplateSubjectsFromClusterResources(instanceName, resources) + stagePVEGuestEffect(ctx, func() { m.updatePVEBackupTemplateSubjectsFromClusterResources(instanceName, resources) }) enrichmentCtx, cancelEnrichment := pveGuestEnrichmentContext(ctx) defer cancelEnrichment() @@ -81,9 +81,8 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam // Publish the complete guest generation only after both VM and container // collection/enrichment has finished. Empty authoritative results still // remove genuinely deleted guests. - m.state.UpdateGuestsForInstance(instanceName, allVMs, allContainers) - - m.recordGuestMetrics(allVMs, allContainers, cycleStart) + stagePVEGuestEffect(ctx, func() { m.recordGuestMetrics(allVMs, allContainers, cycleStart) }) + m.publishPVEGuests(ctx, instanceName, allVMs, allContainers) log.Debug(). Str("instance", instanceName). @@ -387,23 +386,25 @@ func (m *Monitor) handleClusterVMResource( } // Trigger guest metadata migration if old format exists - if m.guestMetadataStore != nil { - m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID) - } + stagePVEGuestEffect(ctx, func() { + if m.guestMetadataStore != nil { + m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID) + } - m.recordGuestSnapshot(instanceName, vm.Type, res.Node, res.VMID, GuestMemorySnapshot{ - Name: vm.Name, - Status: vm.Status, - RetrievedAt: sampleTime, - MemorySource: memorySource, - FallbackReason: guestMemoryFallbackReason(memorySource), - Memory: vm.Memory, - Raw: guestRaw, - Notes: snapshotNotes, + m.recordGuestSnapshot(instanceName, vm.Type, res.Node, res.VMID, GuestMemorySnapshot{ + Name: vm.Name, + Status: vm.Status, + RetrievedAt: sampleTime, + MemorySource: memorySource, + FallbackReason: guestMemoryFallbackReason(memorySource), + Memory: vm.Memory, + Raw: guestRaw, + Notes: snapshotNotes, + }) + + m.checkGuestAlertsForVM(instanceName, vm) }) - m.checkGuestAlertsForVM(instanceName, vm) - return vm, true } @@ -422,22 +423,24 @@ func (m *Monitor) handleClusterContainerResource( } // Trigger guest metadata migration if old format exists - if m.guestMetadataStore != nil { - m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID) - } + stagePVEGuestEffect(ctx, func() { + if m.guestMetadataStore != nil { + m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID) + } - m.recordGuestSnapshot(instanceName, container.Type, res.Node, res.VMID, GuestMemorySnapshot{ - Name: container.Name, - Status: container.Status, - RetrievedAt: sampleTime, - MemorySource: memorySource, - FallbackReason: guestMemoryFallbackReason(memorySource), - Memory: container.Memory, - Raw: guestRaw, + m.recordGuestSnapshot(instanceName, container.Type, res.Node, res.VMID, GuestMemorySnapshot{ + Name: container.Name, + Status: container.Status, + RetrievedAt: sampleTime, + MemorySource: memorySource, + FallbackReason: guestMemoryFallbackReason(memorySource), + Memory: container.Memory, + Raw: guestRaw, + }) + + m.alertManager.CheckGuest(container, instanceName) }) - m.alertManager.CheckGuest(container, instanceName) - return container, true } diff --git a/internal/monitoring/proxmox_guest_snapshot_coherence_test.go b/internal/monitoring/proxmox_guest_snapshot_coherence_test.go index 3734866da9..4e57a3a34e 100644 --- a/internal/monitoring/proxmox_guest_snapshot_coherence_test.go +++ b/internal/monitoring/proxmox_guest_snapshot_coherence_test.go @@ -69,6 +69,7 @@ func TestPollGuestsWithFallbackRetainsOnlyFailedNodeGeneration(t *testing.T) { } nodeStatus := map[string]string{"node-a": "online", "node-b": "online"} cfg := &config.PVEInstance{MonitorVMs: true, MonitorContainers: true} + monitor.replacePVEClient("lab", client) if err := monitor.pollGuestsWithFallback(context.Background(), "lab", cfg, client, nodes, nodeStatus); err != nil { t.Fatalf("partial poll failed: %v", err) diff --git a/internal/monitoring/proxmox_large_cluster_poll_budget_test.go b/internal/monitoring/proxmox_large_cluster_poll_budget_test.go index 246030f7eb..b2f240b8ca 100644 --- a/internal/monitoring/proxmox_large_cluster_poll_budget_test.go +++ b/internal/monitoring/proxmox_large_cluster_poll_budget_test.go @@ -104,7 +104,10 @@ func TestLargePVEClusterPublishesCompleteGenerationWhenEnrichmentBudgetExpires(t stubPVEClient: &stubPVEClient{}, resources: largeClusterResources(), } - ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond) + monitor.replacePVEClient("large-cluster", client) + // Expire only deep enrichment. The parent still owns the authoritative + // inventory and its reserved publication tail, unlike explicit cancellation. + ctx, cancel := context.WithTimeout(context.Background(), pvePollTailReserve+40*time.Millisecond) defer cancel() started := time.Now()