gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-12 15:28:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rotate legacy guest-agent VM priority across polls (#1319)

Browse Source
This commit is contained in:
rcourtman
2026-03-27 16:17:48 +00:00
parent 3d214c80a7
commit 0abbb8ba92
2 changed files with 188 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/monitoring/monitor_extra_coverage_test.go
+114
View File
@@ -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(),
internal/monitoring/monitor_polling.go
+74 -16
View File
@@ -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).