From aa139b73fb2be0e56f552eaf7f87dfa928798145 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 8 Mar 2026 22:14:09 +0000 Subject: [PATCH] Fix intermittent VM disappearance from dashboard (#555) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two root causes: (1) When Proxmox cluster/resources returns a partial response (e.g. during migration or transient API issue), VMs missing from a responsive node were silently dropped because the node appeared in nodesWithResources, bypassing grace-period preservation. Now preserves recently-seen guests from online nodes for up to the grace window. (2) The task queue allowed overlapping polls for the same PVE instance — a slower stale poll could overwrite a newer complete VM list. Added per-instance execution lock to skip duplicate scheduled tasks. --- internal/monitoring/monitor.go | 128 ++++++++++++++++++ .../monitoring/monitor_extra_coverage_test.go | 122 +++++++++++++++++ 2 files changed, 250 insertions(+) diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index ee8d8e65d..b8df159fe 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -879,6 +879,7 @@ type Monitor struct { kubernetesTokenBindings map[string]string // Track token ID -> agent ID bindings to enforce uniqueness hostTokenBindings map[string]string // Track tokenID:hostname -> host identity bindings lastHostRuntimePersist map[string]time.Time // Track last persisted host runtime write per host ID + runningTasks map[string]struct{} // Prevent overlapping polls for the same instance dockerCommands map[string]*dockerHostCommand dockerCommandIndex map[string]string guestMetadataMu sync.RWMutex @@ -984,6 +985,103 @@ func makeGuestID(instanceName string, node string, vmid int) string { return fmt.Sprintf("%s:%s:%d", instanceName, node, vmid) } +func (m *Monitor) beginTaskExecution(task ScheduledTask) bool { + if m == nil { + return true + } + + key := schedulerKey(task.InstanceType, task.InstanceName) + + m.mu.Lock() + defer m.mu.Unlock() + + if m.runningTasks == nil { + m.runningTasks = make(map[string]struct{}) + } + + if _, exists := m.runningTasks[key]; exists { + return false + } + + m.runningTasks[key] = struct{}{} + return true +} + +func (m *Monitor) finishTaskExecution(task ScheduledTask) { + if m == nil { + return + } + + key := schedulerKey(task.InstanceType, task.InstanceName) + + m.mu.Lock() + defer m.mu.Unlock() + + if m.runningTasks != nil { + delete(m.runningTasks, key) + } +} + +func preserveRecentMissingGuestsFromResponsiveNodes( + instanceName string, + prevState models.StateSnapshot, + currentVMs []models.VM, + currentContainers []models.Container, + nodeEffectiveStatus map[string]string, + nodesWithResources map[string]bool, + now time.Time, +) ([]models.VM, []models.Container, int, int) { + currentGuestVMIDs := make(map[int]struct{}, len(currentVMs)+len(currentContainers)) + for _, vm := range currentVMs { + currentGuestVMIDs[vm.VMID] = struct{}{} + } + for _, ct := range currentContainers { + currentGuestVMIDs[ct.VMID] = struct{}{} + } + + preservedVMs := 0 + for _, vm := range prevState.VMs { + if vm.Instance != instanceName { + continue + } + if nodeEffectiveStatus[vm.Node] != "online" || !nodesWithResources[vm.Node] { + continue + } + if !vm.LastSeen.IsZero() && now.Sub(vm.LastSeen) >= nodeOfflineGracePeriod { + continue + } + if _, exists := currentGuestVMIDs[vm.VMID]; exists { + continue + } + + currentVMs = append(currentVMs, vm) + currentGuestVMIDs[vm.VMID] = struct{}{} + preservedVMs++ + } + + preservedContainers := 0 + for _, ct := range prevState.Containers { + if ct.Instance != instanceName { + continue + } + if nodeEffectiveStatus[ct.Node] != "online" || !nodesWithResources[ct.Node] { + continue + } + if !ct.LastSeen.IsZero() && now.Sub(ct.LastSeen) >= nodeOfflineGracePeriod { + continue + } + if _, exists := currentGuestVMIDs[ct.VMID]; exists { + continue + } + + currentContainers = append(currentContainers, ct) + currentGuestVMIDs[ct.VMID] = struct{}{} + preservedContainers++ + } + + return currentVMs, currentContainers, preservedVMs, preservedContainers +} + // parseDurationEnv parses a duration from an environment variable, returning defaultVal if not set or invalid func parseDurationEnv(key string, defaultVal time.Duration) time.Duration { val := os.Getenv(key) @@ -5315,6 +5413,17 @@ func (m *Monitor) executeScheduledTask(ctx context.Context, task ScheduledTask) return } + if !m.beginTaskExecution(task) { + if logging.IsLevelEnabled(zerolog.DebugLevel) { + log.Debug(). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Msg("Skipping overlapping scheduled task for instance already being polled") + } + return + } + defer m.finishTaskExecution(task) + if m.pollMetrics != nil { wait := time.Duration(0) if !task.NextRun.IsZero() { @@ -7983,6 +8092,25 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam Msg("Grace period preservation complete") } + partialPreservedVMs := 0 + partialPreservedContainers := 0 + allVMs, allContainers, partialPreservedVMs, partialPreservedContainers = preserveRecentMissingGuestsFromResponsiveNodes( + instanceName, + prevState, + allVMs, + allContainers, + nodeEffectiveStatus, + nodesWithResources, + time.Now(), + ) + if partialPreservedVMs > 0 || partialPreservedContainers > 0 { + log.Warn(). + Str("instance", instanceName). + Int("preservedMissingVMs", partialPreservedVMs). + Int("preservedMissingContainers", partialPreservedContainers). + Msg("Preserved recently seen guests missing from partial cluster/resources response") + } + m.logSuspiciousRepeatedVMMemoryUsage(instanceName, allVMs, prevInstanceVMs) // Always update state when using efficient polling path diff --git a/internal/monitoring/monitor_extra_coverage_test.go b/internal/monitoring/monitor_extra_coverage_test.go index 1cdf43d95..cccb43df0 100644 --- a/internal/monitoring/monitor_extra_coverage_test.go +++ b/internal/monitoring/monitor_extra_coverage_test.go @@ -5,6 +5,7 @@ import ( "fmt" "path/filepath" "strings" + "sync" "testing" "time" @@ -451,6 +452,70 @@ func TestMonitor_PollVMsAndContainersEfficient_UsesVMRRDMemUsedWhenStatusUnavail } } +func TestMonitor_PollVMsAndContainersEfficient_PreservesMissingGuestsFromPartialResources(t *testing.T) { + now := time.Now() + + m := &Monitor{ + state: models.NewState(), + guestAgentFSInfoTimeout: time.Second, + guestAgentRetries: 1, + guestAgentNetworkTimeout: time.Second, + guestAgentOSInfoTimeout: time.Second, + guestAgentVersionTimeout: time.Second, + guestMetadataCache: make(map[string]guestMetadataCacheEntry), + guestMetadataLimiter: make(map[string]time.Time), + rateTracker: NewRateTracker(), + metricsHistory: NewMetricsHistory(100, time.Hour), + alertManager: alerts.NewManager(), + stalenessTracker: NewStalenessTracker(nil), + nodeRRDMemCache: make(map[string]rrdMemCacheEntry), + vmRRDMemCache: make(map[string]rrdMemCacheEntry), + vmAgentMemCache: make(map[string]agentMemCacheEntry), + } + defer m.alertManager.Stop() + + m.state.UpdateVMsForInstance("pve1", []models.VM{ + {ID: "pve1:node1:100", Instance: "pve1", Node: "node1", VMID: 100, Name: "vm100", Type: "qemu", Status: "stopped", LastSeen: now}, + {ID: "pve1:node1:101", Instance: "pve1", Node: "node1", VMID: 101, Name: "vm101", Type: "qemu", Status: "stopped", LastSeen: now}, + }) + + client := &mockPVEClientExtra{ + resources: []proxmox.ClusterResource{ + {Type: "qemu", VMID: 100, Name: "vm100", Node: "node1", Status: "stopped", MaxMem: 2048, Mem: 1024}, + }, + } + + success := m.pollVMsAndContainersEfficient( + context.Background(), + "pve1", + "", + false, + client, + map[string]string{"node1": "online"}, + ) + if !success { + t.Fatal("pollVMsAndContainersEfficient failed") + } + + state := m.GetState() + if len(state.VMs) != 2 { + t.Fatalf("expected 2 VMs after partial preservation, got %d", len(state.VMs)) + } + + found := false + for _, vm := range state.VMs { + if vm.VMID == 101 { + found = true + if vm.LastSeen.IsZero() { + t.Fatal("preserved VM should keep last seen timestamp") + } + } + } + if !found { + t.Fatal("missing VM from partial cluster/resources response was not preserved") + } +} + func TestMonitor_PollVMsWithNodes_UsesVMRRDMemUsedWhenStatusUnavailable(t *testing.T) { const total = uint64(8 << 30) const inflatedUsed = uint64(6 << 30) @@ -990,11 +1055,29 @@ func (m *mockPVEClientFailNodes) GetVMMemAvailableFromAgent(ctx context.Context, } type mockExecutor struct { + mu sync.Mutex executed []PollTask + started chan struct{} + release chan struct{} } func (m *mockExecutor) Execute(ctx context.Context, task PollTask) { + m.mu.Lock() m.executed = append(m.executed, task) + started := m.started + release := m.release + m.mu.Unlock() + + if started != nil { + select { + case <-started: + default: + close(started) + } + } + if release != nil { + <-release + } } func TestMonitor_ExecuteScheduledTask_Extra(t *testing.T) { @@ -1022,6 +1105,45 @@ func TestMonitor_ExecuteScheduledTask_Extra(t *testing.T) { } } +func TestMonitor_ExecuteScheduledTask_SkipsOverlappingInstanceRuns(t *testing.T) { + m := &Monitor{ + pveClients: map[string]PVEClientInterface{"pve1": &mockPVEClientExtra{}}, + } + + exec := &mockExecutor{ + started: make(chan struct{}), + release: make(chan struct{}), + } + m.SetExecutor(exec) + + task := ScheduledTask{InstanceName: "pve1", InstanceType: InstanceTypePVE} + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + m.executeScheduledTask(context.Background(), task) + }() + + select { + case <-exec.started: + case <-time.After(2 * time.Second): + t.Fatal("first task did not start execution") + } + + m.executeScheduledTask(context.Background(), task) + + exec.mu.Lock() + executed := len(exec.executed) + exec.mu.Unlock() + if executed != 1 { + t.Fatalf("expected overlapping execution to be skipped, got %d executions", executed) + } + + close(exec.release) + wg.Wait() +} + func TestMonitor_Start_Extra(t *testing.T) { t.Setenv("PULSE_MOCK_TRENDS_SEED_DURATION", "5m") t.Setenv("PULSE_MOCK_TRENDS_SAMPLE_INTERVAL", "5m")