diff --git a/internal/monitoring/monitor_extra_coverage_test.go b/internal/monitoring/monitor_extra_coverage_test.go index 7bf3bd0995..52cf0cd0bb 100644 --- a/internal/monitoring/monitor_extra_coverage_test.go +++ b/internal/monitoring/monitor_extra_coverage_test.go @@ -336,6 +336,13 @@ type mockPVEClientExtra struct { vmRRDPoints []proxmox.GuestRRDPoint } +type rotatingLegacyGuestAgentClient struct { + mockPVEClientExtra + fsDelay time.Duration + fsInfoCalls []int + fsInfoCallsMu sync.Mutex +} + func (m *mockPVEClientExtra) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) { return m.resources, nil } @@ -351,6 +358,35 @@ func (m *mockPVEClientExtra) GetVMFSInfo(ctx context.Context, node string, vmid return m.fsInfo, nil } +func (m *rotatingLegacyGuestAgentClient) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]proxmox.VMFileSystem, error) { + m.fsInfoCallsMu.Lock() + m.fsInfoCalls = append(m.fsInfoCalls, vmid) + m.fsInfoCallsMu.Unlock() + + select { + case <-time.After(m.fsDelay): + case <-ctx.Done(): + return nil, ctx.Err() + } + + return []proxmox.VMFileSystem{{ + Mountpoint: "/", + Type: "ext4", + TotalBytes: 100 * 1024 * 1024 * 1024, + UsedBytes: 40 * 1024 * 1024 * 1024, + Disk: "/dev/vda", + }}, nil +} + +func (m *rotatingLegacyGuestAgentClient) takeFSInfoCalls() []int { + m.fsInfoCallsMu.Lock() + defer m.fsInfoCallsMu.Unlock() + + calls := append([]int(nil), m.fsInfoCalls...) + m.fsInfoCalls = nil + return calls +} + func (m *mockPVEClientExtra) GetVMNetworkInterfaces(ctx context.Context, node string, vmid int) ([]proxmox.VMNetworkInterface, error) { return m.netIfaces, nil } @@ -896,6 +932,84 @@ func TestPollVMsWithNodes_UsesLinkedHostAgentDiskFallback(t *testing.T) { } } +func TestPollVMsWithNodes_RotatesGuestAgentPriorityAcrossPolls(t *testing.T) { + t.Setenv("PULSE_DATA_DIR", t.TempDir()) + + m := &Monitor{ + state: models.NewState(), + guestAgentFSInfoTimeout: 250 * time.Millisecond, + guestAgentRetries: 0, + guestAgentNetworkTimeout: 250 * time.Millisecond, + guestAgentOSInfoTimeout: 250 * time.Millisecond, + guestAgentVersionTimeout: 250 * time.Millisecond, + 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), + guestAgentWorkSlots: make(chan struct{}, 1), + guestAgentPollCursor: make(map[string]int), + } + defer m.alertManager.Stop() + + client := &rotatingLegacyGuestAgentClient{ + mockPVEClientExtra: mockPVEClientExtra{ + vms: []proxmox.VM{ + {VMID: 100, Name: "vm100", Node: "node1", Status: "running", MaxMem: 8 * 1024, Mem: 4 * 1024, MaxDisk: 100 * 1024 * 1024 * 1024}, + {VMID: 101, Name: "vm101", Node: "node1", Status: "running", MaxMem: 8 * 1024, Mem: 4 * 1024, MaxDisk: 100 * 1024 * 1024 * 1024}, + {VMID: 102, Name: "vm102", Node: "node1", Status: "running", MaxMem: 8 * 1024, Mem: 4 * 1024, MaxDisk: 100 * 1024 * 1024 * 1024}, + }, + vmStatus: &proxmox.VMStatus{ + Status: "running", + Agent: proxmox.VMAgentField{Value: 1}, + MaxMem: 8 * 1024, + Mem: 4 * 1024, + }, + }, + fsDelay: 60 * time.Millisecond, + } + + checkResolved := func(expectedVMID int) { + state := m.GetState() + if len(state.VMs) != 3 { + t.Fatalf("expected 3 VMs, got %d", len(state.VMs)) + } + + vmByID := make(map[int]models.VM, len(state.VMs)) + for _, vm := range state.VMs { + vmByID[vm.VMID] = vm + } + + if vmByID[expectedVMID].Disk.Usage <= 0 { + t.Fatalf("expected VM %d to get a real disk reading, got usage=%.2f reason=%q", expectedVMID, vmByID[expectedVMID].Disk.Usage, vmByID[expectedVMID].DiskStatusReason) + } + } + + for _, expectedVMID := range []int{100, 101, 102} { + ctx, cancel := context.WithTimeout(context.Background(), 75*time.Millisecond) + m.pollVMsWithNodes( + ctx, + "pve1", + "", + false, + client, + []proxmox.Node{{Node: "node1", Status: "online"}}, + map[string]string{"node1": "online"}, + ) + cancel() + + calls := client.takeFSInfoCalls() + if len(calls) == 0 || calls[0] != expectedVMID { + t.Fatalf("expected VM %d to be first guest-agent disk query, got calls %v", expectedVMID, calls) + } + checkResolved(expectedVMID) + } +} + func TestMonitor_PollGuestSnapshots_Extra(t *testing.T) { m := &Monitor{ state: models.NewState(), diff --git a/internal/monitoring/monitor_polling.go b/internal/monitoring/monitor_polling.go index db811a0480..44ec150acf 100644 --- a/internal/monitoring/monitor_polling.go +++ b/internal/monitoring/monitor_polling.go @@ -249,6 +249,38 @@ func matchZFSPoolForStorage(storage models.Storage, zfsPoolMap map[string]*model return nil } +type indexedLegacyVM struct { + order int + vm proxmox.VM +} + +type orderedNodeVMResult struct { + order int + vm models.VM + alertVM models.VM + snap GuestMemorySnapshot +} + +func rotateIndexedLegacyVMs(vms []indexedLegacyVM, offset int) []indexedLegacyVM { + count := len(vms) + if count <= 1 { + return append([]indexedLegacyVM(nil), vms...) + } + + offset %= count + if offset < 0 { + offset += count + } + if offset == 0 { + return append([]indexedLegacyVM(nil), vms...) + } + + rotated := make([]indexedLegacyVM, 0, count) + rotated = append(rotated, vms[offset:]...) + rotated = append(rotated, vms[:offset]...) + return rotated +} + // pollVMsWithNodes polls VMs from all nodes in parallel using goroutines // When the instance is part of a cluster, the cluster name is used for guest IDs to prevent duplicates // when multiple cluster nodes are configured as separate PVE instances. @@ -325,16 +357,27 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu return } - var nodeVMs []models.VM - var nodeAlertVMs []models.VM - var nodeSnapshots []GuestMemorySnapshot - - // Process each VM - for _, vm := range vms { - // Skip templates + indexedVMs := make([]indexedLegacyVM, 0, len(vms)) + for idx, vm := range vms { if vm.Template == 1 { continue } + indexedVMs = append(indexedVMs, indexedLegacyVM{ + order: idx, + vm: vm, + }) + } + scheduledVMs := rotateIndexedLegacyVMs( + indexedVMs, + m.nextGuestAgentPollOffset(instanceName+":"+n.Node, len(indexedVMs)), + ) + nodeResults := make([]orderedNodeVMResult, 0, len(scheduledVMs)) + + // Process each VM while rotating guest-agent priority across polls. + // This keeps the same higher-VMID guests from always falling to the + // back of the queue under timeouts or bounded guest-agent work slots. + for _, indexedVM := range scheduledVMs { + vm := indexedVM.vm // Parse tags var tags []string @@ -980,18 +1023,33 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, vm.VMID) } - nodeVMs = append(nodeVMs, modelVM) - nodeAlertVMs = append(nodeAlertVMs, modelVM) - nodeSnapshots = append(nodeSnapshots, GuestMemorySnapshot{ - Name: vm.Name, - Status: vm.Status, - RetrievedAt: sampleTime, - MemorySource: memorySource, - Memory: modelVM.Memory, - Raw: guestRaw, + nodeResults = append(nodeResults, orderedNodeVMResult{ + order: indexedVM.order, + vm: modelVM, + alertVM: modelVM, + snap: GuestMemorySnapshot{ + Name: vm.Name, + Status: vm.Status, + RetrievedAt: sampleTime, + MemorySource: memorySource, + Memory: modelVM.Memory, + Raw: guestRaw, + }, }) } + sort.Slice(nodeResults, func(i, j int) bool { + return nodeResults[i].order < nodeResults[j].order + }) + nodeVMs := make([]models.VM, 0, len(nodeResults)) + nodeAlertVMs := make([]models.VM, 0, len(nodeResults)) + nodeSnapshots := make([]GuestMemorySnapshot, 0, len(nodeResults)) + for _, result := range nodeResults { + nodeVMs = append(nodeVMs, result.vm) + nodeAlertVMs = append(nodeAlertVMs, result.alertVM) + nodeSnapshots = append(nodeSnapshots, result.snap) + } + nodeDuration := time.Since(nodeStart) log.Debug(). Str("node", n.Node).