diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 9660e69b59..24cf2bb889 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,47 @@ ## Purpose +### Guest-agent coordination around backups — issue #2439 + +All PVE QEMU guest-agent reads, including disk diagnostics and legacy meminfo +callers, share fail-fast per-guest admission within this Pulse process. Clients +for the same endpoint and configured cluster aliases coordinate across methods; +VMIDs remain serial through node migration. Other VMs and independent endpoints +remain independent. No guest payload or credentials enter coordination state. +Before each command and before returning its fully received bounded payload, +Pulse reads the operation lock from VM config. Any nonempty lock (not just +backup), missing/malformed config or an unverifiable lock defers the command. +Known locks in node inventory, cluster resources or detailed status suppress +polling enrichment even when status fails and recent agent evidence exists. + +Admission is held through the complete response, not just headers. A transport +failure, timeout, wrong-command-ID response or incomplete body defers every +command for that guest for a minute; callers neither immediately retry nor +replay through another cluster endpoint. Coordination retains at most 4,096 +active/cooling endpoint-guest entries and expires cooldowns. Explicit API +permission refusals and unsupported commands remain errors, not successes. + +Deferrals retain the original metadata/memory cache timestamps and supported-OS +failure evidence. Locked guests retain labelled last-known disks/identity and +previous-snapshot memory, or unavailable memory without trustworthy evidence. +Current non-QGA memory, CPU and I/O may still be observed. History never writes +retained disk values or deferred cached memory as newly observed telemetry; +normal unlocked polling resumes and replaces them with fresh data. + +`TestGuestAgentBackupMonitoringContract` exercises healthy → status lock → +resource lock with failed status → node fallback → unlocked resumption through +the real client/builders and History. The client controls cover every read, +unknown lock state, diagnostic overlap, aliases/source/VM isolation, in-flight +lock changes, incomplete bodies, cooldown and absence of failover replay. +These reproduce missing Pulse safety controls, **not** the reported native +PVE/QGA command-ID collision or successful backup thaw. HTTP lock checks cannot +atomically reserve PVE's serial channel against a backup beginning afterwards, +cancel work already running in pvedaemon, detect a guest left frozen after its +lock clears, or coordinate separate Pulse processes. Native freeze/thaw acceptance +on a containing build remains required. Until then, stopping Pulse before a +freeze-enabled backup and restarting only after verified thaw remains the safe +workaround, with monitoring and alerts unavailable during that window. + ### Exact TrueNAS subscription termination — issue #2396 JSON-RPC stream readers and the subscription-acknowledgement wait recognise diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index abb02e0309..3a37762ae5 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2969,3 +2969,8 @@ func TestBroadcastProjectionMatchesPreviousPipeline(t *testing.T) { t.Fatal("broadcast decoration mutated registry") } } + +// The runtime contract includes both poll builders, protocol admission and truthful History. +func TestGuestAgentBackupMonitoringContract(t *testing.T) { + testGuestAgentBackupMonitoringLifecycle(t) +} diff --git a/internal/monitoring/guest_agent_backup_integration_test.go b/internal/monitoring/guest_agent_backup_integration_test.go new file mode 100644 index 0000000000..3afe1c8a27 --- /dev/null +++ b/internal/monitoring/guest_agent_backup_integration_test.go @@ -0,0 +1,190 @@ +package monitoring + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +func testGuestAgentBackupMonitoringLifecycle(t *testing.T) { + const mib = uint64(1024 * 1024) + var phase, calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + p := phase.Load() + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + if p == 1 || p == 2 { + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + } else { + fmt.Fprint(w, `{"data":{}}`) + } + case strings.HasSuffix(r.URL.Path, "/status/current"): + if p == 2 { + http.Error(w, "unavailable", 503) + return + } + lock := "" + if p == 1 { + lock = `,"lock":"backup"` + } + fmt.Fprintf(w, `{"data":{"status":"running","agent":1,"maxmem":%d,"mem":%d%s}}`, 8*mib, 8*mib, lock) + case strings.Contains(r.URL.Path, "/agent/"): + calls.Add(1) + switch { + case strings.HasSuffix(r.URL.Path, "file-read"): + available := 5 * 1024 + if p == 3 { + available = 4 * 1024 + } + json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"content": fmt.Sprintf("MemTotal: 8192 kB\nMemAvailable: %d kB\n", available)}}) + case strings.HasSuffix(r.URL.Path, "get-fsinfo"): + used := 300 * mib + if p == 3 { + used = 400 * mib + } + fmt.Fprintf(w, `{"data":{"result":[{"mountpoint":"/","type":"ext4","total-bytes":%d,"used-bytes":%d}]}}`, 1000*mib, used) + case strings.HasSuffix(r.URL.Path, "network-get-interfaces"): + fmt.Fprint(w, `{"data":{"result":[{"name":"eth0","hardware-address":"02:00:00:00:00:01","ip-addresses":[{"ip-address":"192.0.2.10","ip-address-type":"ipv4","prefix":24}]}]}}`) + case strings.HasSuffix(r.URL.Path, "get-osinfo"): + fmt.Fprint(w, `{"data":{"name":"Linux","version":"fixture"}}`) + default: + fmt.Fprint(w, `{"data":{"result":{"version":"1.0"}}}`) + } + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + m := &Monitor{config: &config.Config{}, rateTracker: NewRateTracker(), metricsHistory: NewMetricsHistory(32, time.Hour), guestMetadataLimiter: make(map[string]time.Time)} + res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "guest", VMID: 105, Status: "running", MaxMem: 8 * mib, Mem: 8 * mib, MaxDisk: 1000 * mib, CPU: 0.1} + var previous *models.VM + identity := "fixture:node:105" + build := func() (models.VM, string) { + vm, raw, source, notes, at, ok := m.buildVMFromClusterResource(context.Background(), "fixture", res, client, identity, nil, previous) + if !ok { + t.Fatal("VM disappeared") + } + m.recordGuestSnapshot("fixture", "qemu", "node", 105, GuestMemorySnapshot{Name: vm.Name, Status: vm.Status, RetrievedAt: at, MemorySource: source, Memory: vm.Memory, Raw: raw, Notes: notes}) + m.recordGuestMetrics([]models.VM{vm}, nil, time.Now().Add(-time.Second)) + previous = &vm + return vm, source + } + initial, source := build() + if source != "guest-agent-meminfo" || initial.Memory.Used != int64(3*mib) || initial.Disk.Used != int64(300*mib) { + t.Fatalf("healthy readings absent: %s %#v %#v", source, initial.Memory, initial.Disk) + } + before := calls.Load() + if before != 5 { + t.Fatalf("healthy agent command count = %d, want five", before) + } + metadataKey := guestMetadataCacheKey("fixture", "node", 105) + memKey := guestMemoryCacheKey("fixture", "node", 105) + originalMetadata := m.guestMetadataCache[metadataKey] + originalMemory := m.vmAgentMemCache[memKey] + phase.Store(1) + res.CPU = 0.2 + locked, source := build() + if locked.ID != initial.ID || locked.GuestAgentStatus != "deferred" || locked.DiskStatusReason != "prev-vm-locked" { + t.Fatalf("lock lost continuity/truth: %#v", locked) + } + if locked.Memory.Used != initial.Memory.Used || source != "previous-snapshot" || locked.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(locked.NetworkInterfaces, initial.NetworkInterfaces) { + t.Fatalf("last known evidence lost: %s %#v", source, locked) + } + if calls.Load() != before { + t.Fatalf("status-lock sent agent commands: %d -> %d", before, calls.Load()) + } + if !reflect.DeepEqual(m.guestMetadataCache[metadataKey], originalMetadata) || !reflect.DeepEqual(m.vmAgentMemCache[memKey], originalMemory) { + t.Fatal("lock renewed/replaced cached observations") + } + for metric, want := range map[string]int{"cpu": 2, "memory": 1, "memoryused": 1, "disk": 1} { + if got := len(m.metricsHistory.GetGuestMetrics(identity, metric, time.Hour)); got != want { + t.Errorf("%s history points = %d, want %d", metric, got, want) + } + } + // Cluster lock must survive a failed status call and the recent-evidence fallback. + phase.Store(2) + if err := json.Unmarshal([]byte(`{"type":"qemu","node":"node","name":"guest","vmid":105,"status":"running","lock":"backup","maxmem":8388608,"mem":8388608,"maxdisk":1048576000}`), &res); err != nil { + t.Fatal(err) + } + locked, _ = build() + if calls.Load() != before || locked.GuestAgentStatus != "deferred" || locked.AgentVersion != initial.AgentVersion { + t.Fatalf("resource-lock fallback queried/lost identity: calls=%d %#v", calls.Load(), locked) + } + // The per-node inventory fallback must propagate its lock to the same builder. + manager := alerts.NewManagerWithDataDir(t.TempDir()) + defer manager.Stop() + m.alertManager = manager + nodeVMs, _ := m.pollNodeVMsWithClusterResourceBuilder(context.Background(), "fixture", "node", []proxmox.VM{{VMID: 105, Name: "guest", Status: "running", Lock: "backup", MaxMem: 8 * mib, Mem: 8 * mib, MaxDisk: 1000 * mib}}, client, map[string]models.VM{makeGuestID("fixture", "node", 105): locked}, nil) + if len(nodeVMs) != 1 || nodeVMs[0].GuestAgentStatus != "deferred" || calls.Load() != before { + t.Fatalf("node fallback lost lock or sent guest commands: %#v calls=%d", nodeVMs, calls.Load()) + } + // Resume from expired, unrenewed caches; new data must replace the retained view. + phase.Store(3) + res.Lock = "" + cached := m.vmAgentMemCache[memKey] + cached.fetchedAt = time.Now().Add(-2 * vmAgentMemCacheTTL) + m.vmAgentMemCache[memKey] = cached + metadata := m.guestMetadataCache[metadataKey] + metadata.fetchedAt = time.Now().Add(-2 * guestMetadataCacheTTL) + m.guestMetadataCache[metadataKey] = metadata + m.guestMetadataLimiter = make(map[string]time.Time) + resumed, source := build() + if resumed.ID != initial.ID || resumed.GuestAgentStatus != "available" || resumed.DiskStatusReason != "" || resumed.Disk.Used != int64(400*mib) || resumed.Memory.Used != int64(4*mib) || source != "guest-agent-meminfo" { + t.Fatalf("healthy resumption failed: %s %#v", source, resumed) + } + if calls.Load() != before+5 { + t.Errorf("fresh resume commands = %d, want %d", calls.Load(), before+5) + } + if got := len(m.metricsHistory.GetGuestMetrics(identity, "disk", time.Hour)); got != 2 { + t.Errorf("cached disk was written as new telemetry: %d points", got) + } +} + +func TestGuestAgentDeferralPreservesMetadataAndMemoryCache(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + calls.Add(1) + http.Error(w, "QEMU command timed out", 500) + })) + defer server.Close() + client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + now := time.Now().Add(-10 * time.Minute) + metadata := guestMetadataCacheEntry{ipAddresses: []string{"192.0.2.10"}, osName: "Linux", agentVersion: "1.0", fetchedAt: now, osInfoFailureCount: 2} + memory := agentMemCacheEntry{available: 1024, fetchedAt: now, info: proxmox.LinuxMemoryAvailability{Source: "meminfo-available", EffectiveAvailable: 1024}} + mk, gk := guestMemoryCacheKey("fixture", "node", 105), guestMetadataCacheKey("fixture", "node", 105) + m := &Monitor{guestMetadataCache: map[string]guestMetadataCacheEntry{gk: metadata}, vmAgentMemCache: map[string]agentMemCacheEntry{mk: memory}, guestMetadataLimiter: make(map[string]time.Time)} + ips, _, os, _, version := m.fetchGuestAgentMetadata(context.Background(), client, "fixture", "node", "guest", 105, &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}}, false) + if len(ips) != 1 || os != "Linux" || version != "1.0" { + t.Fatal("deferral lost last known metadata") + } + if _, err := m.getVMAgentMemoryAvailability(context.Background(), client, "fixture", "node", 105); proxmox.GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Fatalf("memory retried deferred VM: %v", err) + } + if calls.Load() != 1 || !reflect.DeepEqual(m.guestMetadataCache[gk], metadata) || !reflect.DeepEqual(m.vmAgentMemCache[mk], memory) { + t.Fatal("deferral queued work, renewed cache or poisoned supported-OS evidence") + } +} diff --git a/internal/monitoring/guest_agent_evidence.go b/internal/monitoring/guest_agent_evidence.go index caf19fd1ce..001c33d176 100644 --- a/internal/monitoring/guest_agent_evidence.go +++ b/internal/monitoring/guest_agent_evidence.go @@ -39,7 +39,7 @@ func hasRecentGuestAgentEvidence(prev *models.VM, now time.Time) bool { func shouldQueryGuestAgent(vmStatus *proxmox.VMStatus, prev *models.VM, now time.Time) bool { if vmStatus != nil { - return vmStatus.Agent.IsAvailable() + return vmStatus.Lock == "" && vmStatus.Agent.IsAvailable() } return hasRecentGuestAgentEvidence(prev, now) } diff --git a/internal/monitoring/guest_disk_stability.go b/internal/monitoring/guest_disk_stability.go index 87cb04f5b0..53a99390d5 100644 --- a/internal/monitoring/guest_disk_stability.go +++ b/internal/monitoring/guest_disk_stability.go @@ -4,6 +4,8 @@ import ( "strings" "time" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" + "github.com/rcourtman/pulse-go-rewrite/internal/models" ) @@ -19,6 +21,10 @@ func classifyGuestAgentDiskStatusError(err error) string { return "" } + if reason := proxmox.GuestAgentDeferredReason(err); reason != "" { + return reason + } + errStr := err.Error() errStrLower := strings.ToLower(errStr) diff --git a/internal/monitoring/guest_memory_agent.go b/internal/monitoring/guest_memory_agent.go index 8b29b23279..3439899636 100644 --- a/internal/monitoring/guest_memory_agent.go +++ b/internal/monitoring/guest_memory_agent.go @@ -2,6 +2,7 @@ package monitoring import ( "context" + "errors" "fmt" "strings" "time" @@ -93,6 +94,12 @@ func (m *Monitor) getVMAgentMemoryAvailability(ctx context.Context, client PVECl return proxmox.LinuxMemoryAvailability{}, fmt.Errorf("guest agent meminfo fallback unsupported") } + if errors.Is(err, proxmox.ErrGuestAgentDeferred) { + // A coordination pause is not a failed guest observation. Preserve the + // last successful cache and its original timestamp for normal resumption. + return proxmox.LinuxMemoryAvailability{}, err + } + m.rrdCacheMu.Lock() defer m.rrdCacheMu.Unlock() if m.vmAgentMemCache == nil { diff --git a/internal/monitoring/guest_memory_sources.go b/internal/monitoring/guest_memory_sources.go index d26d7c48f0..709038169c 100644 --- a/internal/monitoring/guest_memory_sources.go +++ b/internal/monitoring/guest_memory_sources.go @@ -173,7 +173,7 @@ func selectGuestLowTrustUsedMemory(memTotal uint64, status *proxmox.VMStatus) (u } func shouldPreferGuestAgentMemAvailable(status *proxmox.VMStatus, memTotal uint64) bool { - if status == nil || !status.Agent.IsAvailable() || status.Mem == 0 { + if status == nil || status.Lock != "" || !status.Agent.IsAvailable() || status.Mem == 0 { return false } if guestStatusFreeMem(status) > 0 { @@ -311,7 +311,7 @@ func (m *Monitor) resolveGuestStatusMemory( // only in node RRD, so a guest RRD lookup can never yield memory evidence // (#1634). - if !hasMemAvailable && !hasSelectedUsed && status.Agent.IsAvailable() && !triedGuestAgentMemAvailable { + if !hasMemAvailable && !hasSelectedUsed && status.Lock == "" && status.Agent.IsAvailable() && !triedGuestAgentMemAvailable { if agentAvailable, agentSource, ok := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw); ok { memAvailable = agentAvailable hasMemAvailable = true diff --git a/internal/monitoring/guest_metadata.go b/internal/monitoring/guest_metadata.go index ec955ed3cc..ce22b3636c 100644 --- a/internal/monitoring/guest_metadata.go +++ b/internal/monitoring/guest_metadata.go @@ -3,6 +3,7 @@ package monitoring import ( "context" "encoding/json" + "errors" "fmt" "math" "sort" @@ -206,7 +207,7 @@ func (m *Monitor) retryGuestAgentCall(ctx context.Context, timeout time.Duration lastErr = err // Don't retry non-timeout errors or if this was the last attempt - if attempt >= maxRetries || !strings.Contains(err.Error(), "timeout") { + if errors.Is(err, proxmox.ErrGuestAgentDeferred) || attempt >= maxRetries || !strings.Contains(err.Error(), "timeout") { break } @@ -229,6 +230,11 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI cached, ok := m.guestMetadataCache[key] m.guestMetadataMu.RUnlock() + if vmStatus != nil && vmStatus.Lock != "" { + // Deliberately retained identity, not a newly fetched observation. + return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion + } + agentAvailable := client != nil && ((vmStatus != nil && vmStatus.Agent.IsAvailable()) || allowWithoutStatus) if !agentAvailable { if ok && now.Sub(cached.fetchedAt) < guestMetadataCacheEntryTTL(cached) { @@ -274,6 +280,10 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI interfaces, err := m.retryGuestAgentCall(ctx, m.guestAgentNetworkTimeout, m.guestAgentRetries, func(ctx context.Context) (interface{}, error) { return client.GetVMNetworkInterfaces(ctx, nodeName, vmid) }) + if errors.Is(err, proxmox.ErrGuestAgentDeferred) { + m.deferGuestMetadataRetry(key, time.Now()) + return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion + } if err != nil { log.Debug(). Str("instance", instanceName). @@ -311,6 +321,10 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI agentInfoRaw, err := m.retryGuestAgentCall(ctx, m.guestAgentOSInfoTimeout, m.guestAgentRetries, func(ctx context.Context) (interface{}, error) { return client.GetVMAgentInfo(ctx, nodeName, vmid) }) + if errors.Is(err, proxmox.ErrGuestAgentDeferred) { + m.deferGuestMetadataRetry(key, time.Now()) + return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion + } if err != nil { if isGuestAgentOSInfoUnsupportedError(err) { osInfoSkip = true @@ -365,6 +379,10 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI versionRaw, err := m.retryGuestAgentCall(ctx, m.guestAgentVersionTimeout, m.guestAgentRetries, func(ctx context.Context) (interface{}, error) { return client.GetVMAgentVersion(ctx, nodeName, vmid) }) + if errors.Is(err, proxmox.ErrGuestAgentDeferred) { + m.deferGuestMetadataRetry(key, time.Now()) + return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion + } if err != nil { log.Debug(). Str("instance", instanceName). diff --git a/internal/monitoring/memory_trust_characterization_test.go b/internal/monitoring/memory_trust_characterization_test.go index 2d12f5091a..60c583c3a1 100644 --- a/internal/monitoring/memory_trust_characterization_test.go +++ b/internal/monitoring/memory_trust_characterization_test.go @@ -2,6 +2,8 @@ package monitoring import ( "context" + "fmt" + "strings" "testing" "time" @@ -1171,3 +1173,30 @@ func TestGracePeriodGuestContributesNoMemorySample(t *testing.T) { t.Fatalf("grace-period guest recorded %d memory sample(s): %+v", len(got), got) } } + +func TestBackupLockMemoryTrustKeepsNativeEvidenceWithoutAgentReads(t *testing.T) { + const gib = uint64(1024 * 1024 * 1024) + for _, native := range []bool{false, true} { + t.Run(fmt.Sprintf("native-%t", native), func(t *testing.T) { + m := &Monitor{} + client := &vmMemoryTrustStubClient{stubPVEClient: &stubPVEClient{}, vmAgentMemAvailable: 5 * gib} + status := &proxmox.VMStatus{Status: "running", Agent: proxmox.VMAgentField{Value: 1}, Lock: "backup", MaxMem: 8 * gib, Mem: 8 * gib} + if native { + status.MemInfo = &proxmox.VMMemInfo{Total: 8 * gib, Available: 5 * gib} + } + _, used, source := m.resolveGuestStatusMemory(context.Background(), client, "fixture", "guest", "node", 105, "fixture:node:105", status, nil, 8*gib, "", &VMMemoryRaw{}) + if client.vmAgentMemCalls != 0 || strings.HasPrefix(source, "guest-agent-meminfo") { + t.Fatalf("backup used fresh/cached guest-agent evidence: calls=%d source=%s", client.vmAgentMemCalls, source) + } + if native && (used != 3*gib || source != "available-field") { + t.Fatalf("current native memory lost: %d %s", used, source) + } + status.Lock = "" + status.MemInfo = nil + _, _, source = m.resolveGuestStatusMemory(context.Background(), client, "fixture", "guest", "node", 105, "fixture:node:105", status, nil, 8*gib, "", &VMMemoryRaw{}) + if client.vmAgentMemCalls != 1 || source != "guest-agent-meminfo" { + t.Fatalf("unlocked memory did not resume: %d %s", client.vmAgentMemCalls, source) + } + }) + } +} diff --git a/internal/monitoring/monitor_pve_guest_builders.go b/internal/monitoring/monitor_pve_guest_builders.go index af16aa7dbc..3f1423fa2c 100644 --- a/internal/monitoring/monitor_pve_guest_builders.go +++ b/internal/monitoring/monitor_pve_guest_builders.go @@ -58,6 +58,12 @@ func (m *Monitor) applyVMStatusDetails( return } + if res.Lock != "" { + copy := *status + copy.Lock = res.Lock + status = © + } + res.Lock = status.Lock state.detailedStatus = status state.guestAgentStatus, state.guestAgentExpected = vmGuestAgentRuntimeState(status, recentGuestAgentEvidence) state.memTotal, state.memUsed, state.memorySource = m.resolveGuestStatusMemory( @@ -111,6 +117,9 @@ func (m *Monitor) applyVMStatusDetails( if len(fsDisks) > 0 { state.individualDisks = fsDisks } + if guestAgentDiskDeferred(state.diskStatusReason) { + state.guestAgentStatus = "deferred" + } } else { // Agent disabled - show allocated disk size if state.diskTotal > 0 { @@ -172,6 +181,9 @@ func vmGuestAgentRuntimeState(status *proxmox.VMStatus, recentGuestAgentEvidence if status == nil { return "", false } + if status.Lock != "" { + return "deferred", status.Agent.IsEnabled() + } if status.Agent.IsAvailable() { return "available", true } @@ -265,10 +277,13 @@ func (m *Monitor) buildVMFromClusterResource( Msg("Could not get VM status, using cluster/resources disk data") } + if state.detailedStatus != nil && state.detailedStatus.Lock != "" { + res.Lock = state.detailedStatus.Lock + } now := time.Now() guestAgentAvailable := shouldQueryGuestAgent(state.detailedStatus, prevVM, now) || m.hasRecentGuestMetadataEvidence(instanceName, res.Node, res.VMID, now) - if guestAgentAvailable && state.detailedStatus == nil { + if guestAgentAvailable && res.Lock == "" && state.detailedStatus == nil { guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion := m.fetchGuestAgentMetadata( ctx, client, @@ -308,12 +323,31 @@ func (m *Monitor) buildVMFromClusterResource( if len(fsDisks) > 0 { state.individualDisks = fsDisks } + if guestAgentDiskDeferred(state.diskStatusReason) { + state.guestAgentStatus = "deferred" + } state.guestAgentExpected = true - if len(guestIPs) > 0 || len(guestIfaces) > 0 || guestOSName != "" || guestOSVersion != "" || guestAgentVersion != "" || state.diskFromAgent { + if state.guestAgentStatus != "deferred" && (len(guestIPs) > 0 || len(guestIfaces) > 0 || guestOSName != "" || guestOSVersion != "" || guestAgentVersion != "" || state.diskFromAgent) { state.guestAgentStatus = "available" } } + if res.Lock != "" { + state.guestAgentStatus = "deferred" + state.diskStatusReason = "vm-locked" + state.diskFromAgent = false + state.diskUsage = -1 + if guestMemorySourceReliability(state.memorySource) == guestMemoryReliabilityLow { + state.memorySource = "unavailable" + state.memUsed = 0 + } + if prevVM != nil { + state.ipAddresses = cloneStringSlice(prevVM.IPAddresses) + state.networkInterfaces = cloneGuestNetworkInterfaces(prevVM.NetworkInterfaces) + state.osName, state.osVersion, state.agentVersion = prevVM.OSName, prevVM.OSVersion, prevVM.AgentVersion + } + } + var preferred bool state.diskTotal, state.diskUsed, state.diskFree, state.diskUsage, state.individualDisks, state.diskStatusReason, preferred = preferLinkedHostAgentDiskInventory( guestID, @@ -449,6 +483,7 @@ func (m *Monitor) buildVMFromClusterResource( Disks: state.individualDisks, DiskStatusReason: state.diskStatusReason, GuestAgentStatus: state.guestAgentStatus, + Lock: res.Lock, GuestAgentExpected: state.guestAgentExpected, IPAddresses: state.ipAddresses, OSName: state.osName, @@ -531,6 +566,9 @@ type vmFSInfoSummary struct { } func (m *Monitor) fetchVMFSInfo(ctx context.Context, instanceName string, res proxmox.ClusterResource, client PVEClientInterface) ([]proxmox.VMFileSystem, string, bool) { + if res.Lock != "" { + return nil, "vm-locked", false + } // Use retry logic for guest agent calls to handle transient timeouts (refs #630) fsInfoRaw, err := m.retryGuestAgentCall(ctx, m.guestAgentFSInfoTimeout, m.guestAgentRetries, func(ctx context.Context) (interface{}, error) { return client.GetVMFSInfo(ctx, res.Node, res.VMID) @@ -542,6 +580,9 @@ func (m *Monitor) fetchVMFSInfo(ctx context.Context, instanceName string, res pr } } if err != nil { + if reason := proxmox.GuestAgentDeferredReason(err); reason != "" { + return nil, reason, false + } // Log more helpful error messages based on the error type errMsg := err.Error() if strings.Contains(errMsg, "500") || strings.Contains(errMsg, "QEMU guest agent is not running") { @@ -847,3 +888,11 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo( return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, nil, false, "special-filesystems-only" } + +func guestAgentDiskDeferred(reason string) bool { + switch strings.TrimPrefix(reason, "prev-") { + case "vm-locked", "lock-unverified", "agent-busy", "agent-cooldown", "agent-capacity", "agent-timeout", "agent-response-incomplete": + return true + } + return false +} diff --git a/internal/monitoring/monitor_pve_guest_helpers.go b/internal/monitoring/monitor_pve_guest_helpers.go index 65885359ff..86fe424d2f 100644 --- a/internal/monitoring/monitor_pve_guest_helpers.go +++ b/internal/monitoring/monitor_pve_guest_helpers.go @@ -1,6 +1,7 @@ package monitoring import ( + "strings" "time" "github.com/rcourtman/pulse-go-rewrite/internal/models" @@ -134,7 +135,17 @@ func (m *Monitor) recordGuestMetrics(allVMs []models.VM, allContainers []models. } if vm.Status == "running" { 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) + memoryUsage, memoryUsed, diskUsage := historyMemoryUsage(vm.Memory), historyMemoryUsed(vm.Memory), vm.Disk.Usage + if strings.HasPrefix(vm.DiskStatusReason, "prev-") { + diskUsage = -1 + } + if vm.GuestAgentStatus == "deferred" { + snapshot := m.previousGuestSnapshot(vm.Instance, "qemu", vm.Node, vm.VMID) + if snapshot == nil || snapshot.MemorySource == "previous-snapshot" || strings.HasPrefix(CanonicalMemorySource(snapshot.MemorySource), "guest-agent-meminfo") { + memoryUsage, memoryUsed = -1, -1 + } + } + m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), memoryUsage, memoryUsed, diskUsage, diskRead, diskWrite, networkIn, networkOut, now) } } for _, ct := range allContainers { diff --git a/internal/monitoring/monitor_pve_node_vm_builder.go b/internal/monitoring/monitor_pve_node_vm_builder.go index 755c249e48..c5182072be 100644 --- a/internal/monitoring/monitor_pve_node_vm_builder.go +++ b/internal/monitoring/monitor_pve_node_vm_builder.go @@ -51,6 +51,7 @@ func (m *Monitor) pollNodeVMsWithClusterResourceBuilder( Uptime: vm.Uptime, Template: vm.Template, Tags: vm.Tags, + Lock: vm.Lock, IOCounters: vm.IOCounters, ObservedAt: vm.ObservedAt, }, diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 5dc74c8d6f..1b3dcee276 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -640,6 +640,12 @@ func (c *Client) requestWithRetry(ctx context.Context, method, path string, data // get performs a GET request func (c *Client) get(ctx context.Context, path string) (*http.Response, error) { + if node, vmid, ok := guestAgentPath(path); ok { + return c.getGuestAgent(ctx, path, node, vmid) + } + if strings.Contains(path, "/agent/") { + return nil, &guestAgentDeferredError{reason: "invalid-guest-key"} + } return c.request(ctx, "GET", path, nil) } @@ -2432,6 +2438,7 @@ type ClusterResource struct { Tags string `json:"tags,omitempty"` IOCounters IOCounterPresence `json:"-"` ObservedAt time.Time `json:"-"` + Lock string `json:"lock,omitempty"` } // GetClusterResources returns all resources (VMs, containers) across the cluster @@ -2715,6 +2722,7 @@ type VMStatus struct { Agent VMAgentField `json:"agent"` IOCounters IOCounterPresence `json:"-"` ObservedAt time.Time `json:"-"` + Lock string `json:"lock,omitempty"` } // GetZFSPoolStatus gets the status of ZFS pools on a node diff --git a/pkg/proxmox/client_api_more3_test.go b/pkg/proxmox/client_api_more3_test.go index 12a4aafc58..240dce4cef 100644 --- a/pkg/proxmox/client_api_more3_test.go +++ b/pkg/proxmox/client_api_more3_test.go @@ -2,12 +2,19 @@ package proxmox import ( "context" + "fmt" "net/http" + "strings" "testing" ) func TestClientVMFSInfoParsing(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo": writeJSON(t, w, map[string]interface{}{ @@ -56,6 +63,11 @@ func TestClientVMFSInfoParsing(t *testing.T) { func TestClientVMFSInfoParsingCentOSXFSResult(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/pve1/qemu/1376/agent/get-fsinfo": w.Header().Set("Content-Type", "application/json") @@ -87,6 +99,11 @@ func TestClientVMFSInfoParsingCentOSXFSResult(t *testing.T) { func TestClientVMFSInfoParsingPrivilegedCapacityFallback(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo": writeJSON(t, w, map[string]interface{}{ @@ -126,6 +143,11 @@ func TestClientVMFSInfoParsingPrivilegedCapacityFallback(t *testing.T) { func TestClientVMFSInfoParsingWindowsVolumeGUIDMountpointFallback(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo": writeJSON(t, w, map[string]interface{}{ @@ -164,6 +186,11 @@ func TestClientVMFSInfoParsingWindowsVolumeGUIDMountpointFallback(t *testing.T) func TestClientVMFSInfoParsingIssue1319WindowsVolumePayload(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/pve7/qemu/116/agent/get-fsinfo": w.Header().Set("Content-Type", "application/json") @@ -200,6 +227,11 @@ func TestClientVMFSInfoParsingIssue1319WindowsVolumePayload(t *testing.T) { func TestClientVMFSInfoSkipsMalformedEntries(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo": writeJSON(t, w, map[string]interface{}{ @@ -242,6 +274,11 @@ func TestClientVMFSInfoSkipsMalformedEntries(t *testing.T) { func TestClientVMFSInfoObjectResult(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo": writeJSON(t, w, map[string]interface{}{ @@ -266,6 +303,11 @@ func TestClientVMFSInfoObjectResult(t *testing.T) { func TestClientVMFSInfoObjectFilesystemResult(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo": writeJSON(t, w, map[string]interface{}{ diff --git a/pkg/proxmox/client_api_more_test.go b/pkg/proxmox/client_api_more_test.go index 095d6cc9a3..bf66d6c96f 100644 --- a/pkg/proxmox/client_api_more_test.go +++ b/pkg/proxmox/client_api_more_test.go @@ -3,8 +3,10 @@ package proxmox import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" + "strings" "testing" "time" ) @@ -217,6 +219,11 @@ func TestClientClusterAndAgentInfo(t *testing.T) { func TestClientGetVMNetworkInterfaces_ObjectAndPartialPayloads(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/network-get-interfaces": writeJSON(t, w, map[string]interface{}{ @@ -340,6 +347,11 @@ func TestClientStatusAndResources(t *testing.T) { func TestClientGetVMMemAvailableFromAgent(t *testing.T) { client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes/node1/qemu/100/agent/file-read": if got := r.URL.Query().Get("file"); got != "/proc/meminfo" { diff --git a/pkg/proxmox/client_vma_version_test.go b/pkg/proxmox/client_vma_version_test.go index cc8a57c18f..4d957a5f03 100644 --- a/pkg/proxmox/client_vma_version_test.go +++ b/pkg/proxmox/client_vma_version_test.go @@ -5,11 +5,17 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "testing" ) func TestClient_GetVMAgentVersion_MapResult(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") if r.URL.Path == "/api2/json/nodes/node1/qemu/100/agent/info" { fmt.Fprint(w, `{"data":{"result":{"version":{"version":"3.1"}}}}`) @@ -40,6 +46,11 @@ func TestClient_GetVMAgentVersion_MapResult(t *testing.T) { func TestClient_GetVMAgentVersion_NoVersion(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") if r.URL.Path == "/api2/json/nodes/node1/qemu/100/agent/info" { fmt.Fprint(w, `{"data":{"result":{"foo":"bar"}}}`) diff --git a/pkg/proxmox/cluster_client.go b/pkg/proxmox/cluster_client.go index be6d44c381..64338a616d 100644 --- a/pkg/proxmox/cluster_client.go +++ b/pkg/proxmox/cluster_client.go @@ -172,6 +172,7 @@ func sanitizeEndpointError(errMsg string) string { // endpointFingerprints is an optional map of endpoint URL -> TLS fingerprint for per-node certificate verification. // This enables TOFU (Trust On First Use) for clusters with unique self-signed certs per node. func NewClusterClient(name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string) *ClusterClient { + registerGuestAgentEndpoints(config.Host, endpoints) if endpointFingerprints == nil { endpointFingerprints = make(map[string]string) } @@ -1373,7 +1374,7 @@ func (cc *ClusterClient) GetVMConfig(ctx context.Context, node string, vmid int) func (cc *ClusterClient) GetVMAgentInfo(ctx context.Context, node string, vmid int) (map[string]interface{}, error) { var result map[string]interface{} - err := cc.executeWithFailover(ctx, func(client *Client) error { + err := cc.executeGuestAgent(ctx, func(client *Client) error { info, err := client.GetVMAgentInfo(ctx, node, vmid) if err != nil { return err @@ -1387,7 +1388,7 @@ func (cc *ClusterClient) GetVMAgentInfo(ctx context.Context, node string, vmid i // GetVMAgentVersion returns the guest agent version for the VM. func (cc *ClusterClient) GetVMAgentVersion(ctx context.Context, node string, vmid int) (string, error) { var version string - err := cc.executeWithFailover(ctx, func(client *Client) error { + err := cc.executeGuestAgent(ctx, func(client *Client) error { v, err := client.GetVMAgentVersion(ctx, node, vmid) if err != nil { return err @@ -1401,7 +1402,7 @@ func (cc *ClusterClient) GetVMAgentVersion(ctx context.Context, node string, vmi // GetVMFSInfo returns filesystem information from QEMU guest agent func (cc *ClusterClient) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]VMFileSystem, error) { var result []VMFileSystem - err := cc.executeWithFailover(ctx, func(client *Client) error { + err := cc.executeGuestAgent(ctx, func(client *Client) error { info, err := client.GetVMFSInfo(ctx, node, vmid) if err != nil { return err @@ -1415,7 +1416,7 @@ func (cc *ClusterClient) GetVMFSInfo(ctx context.Context, node string, vmid int) // GetVMNetworkInterfaces returns guest network interfaces from the QEMU agent func (cc *ClusterClient) GetVMNetworkInterfaces(ctx context.Context, node string, vmid int) ([]VMNetworkInterface, error) { var result []VMNetworkInterface - err := cc.executeWithFailover(ctx, func(client *Client) error { + err := cc.executeGuestAgent(ctx, func(client *Client) error { interfaces, err := client.GetVMNetworkInterfaces(ctx, node, vmid) if err != nil { return err @@ -1429,7 +1430,7 @@ func (cc *ClusterClient) GetVMNetworkInterfaces(ctx context.Context, node string // GetVMMemAvailableFromAgent reads /proc/meminfo via the QEMU guest agent to get MemAvailable. func (cc *ClusterClient) GetVMMemAvailableFromAgent(ctx context.Context, node string, vmid int) (uint64, error) { var result uint64 - err := cc.executeWithFailover(ctx, func(client *Client) error { + err := cc.executeGuestAgent(ctx, func(client *Client) error { available, err := client.GetVMMemAvailableFromAgent(ctx, node, vmid) if err != nil { return err @@ -1444,7 +1445,7 @@ func (cc *ClusterClient) GetVMMemAvailableFromAgent(ctx context.Context, node st // through the first healthy cluster client. func (cc *ClusterClient) GetVMMemoryAvailabilityFromAgent(ctx context.Context, node string, vmid int) (LinuxMemoryAvailability, error) { var result LinuxMemoryAvailability - err := cc.executeWithFailover(ctx, func(client *Client) error { + err := cc.executeGuestAgent(ctx, func(client *Client) error { availability, err := client.GetVMMemoryAvailabilityFromAgent(ctx, node, vmid) if err != nil { return err diff --git a/pkg/proxmox/cluster_client_additional_api6_test.go b/pkg/proxmox/cluster_client_additional_api6_test.go index d969ed0461..ca085c0376 100644 --- a/pkg/proxmox/cluster_client_additional_api6_test.go +++ b/pkg/proxmox/cluster_client_additional_api6_test.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "testing" "time" ) @@ -33,6 +34,11 @@ func TestClusterClient_GetNodePendingUpdates_NoHealthyNodes(t *testing.T) { func TestClusterClient_GetVMAgentVersion_QemuGA(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") if r.URL.Path == "/api2/json/nodes" { fmt.Fprint(w, `{"data":[{"node":"node1","status":"online"}]}`) diff --git a/pkg/proxmox/cluster_client_additional_api_test.go b/pkg/proxmox/cluster_client_additional_api_test.go index 17d825eeb3..0c6d65ccff 100644 --- a/pkg/proxmox/cluster_client_additional_api_test.go +++ b/pkg/proxmox/cluster_client_additional_api_test.go @@ -146,6 +146,11 @@ func TestClusterClient_GetVMConfig(t *testing.T) { func TestClusterClient_GetVMAgentInfo(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") if r.URL.Path == "/api2/json/nodes" { fmt.Fprint(w, `{"data":[{"node":"node1","status":"online"}]}`) @@ -173,6 +178,11 @@ func TestClusterClient_GetVMAgentInfo(t *testing.T) { func TestClusterClient_GetVMAgentVersion(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") if r.URL.Path == "/api2/json/nodes" { fmt.Fprint(w, `{"data":[{"node":"node1","status":"online"}]}`) @@ -200,6 +210,11 @@ func TestClusterClient_GetVMAgentVersion(t *testing.T) { func TestClusterClient_GetVMNetworkInterfaces(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") if r.URL.Path == "/api2/json/nodes" { fmt.Fprint(w, `{"data":[{"node":"node1","status":"online"}]}`) diff --git a/pkg/proxmox/cluster_client_api_test.go b/pkg/proxmox/cluster_client_api_test.go index 516f373a48..bbf27e5310 100644 --- a/pkg/proxmox/cluster_client_api_test.go +++ b/pkg/proxmox/cluster_client_api_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "sync/atomic" "testing" "time" @@ -37,6 +38,11 @@ func TestClusterClient_GetCephStatus(t *testing.T) { func TestClusterClient_GetVMMemoryAvailabilityFromAgent(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } w.Header().Set("Content-Type", "application/json") switch r.URL.Path { case "/api2/json/nodes": diff --git a/pkg/proxmox/cluster_client_test.go b/pkg/proxmox/cluster_client_test.go index e5eaf068fe..19ebf4f59c 100644 --- a/pkg/proxmox/cluster_client_test.go +++ b/pkg/proxmox/cluster_client_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "sync/atomic" "testing" "time" @@ -67,6 +68,11 @@ func TestClusterClientHandlesRateLimitWithoutMarkingUnhealthy(t *testing.T) { func TestClusterClientIgnoresGuestAgentTimeoutForHealth(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes": w.Header().Set("Content-Type", "application/json") @@ -295,6 +301,11 @@ func TestIsEndpointConnectivityError(t *testing.T) { func TestClusterClientIgnoresPlainGuestAgent500ForHealth(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guest reads require fresh operation-lock evidence, independently of their payload fixture. + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } switch r.URL.Path { case "/api2/json/nodes": w.Header().Set("Content-Type", "application/json") diff --git a/pkg/proxmox/guest_agent_backup_test.go b/pkg/proxmox/guest_agent_backup_test.go new file mode 100644 index 0000000000..f4b1d3ff1a --- /dev/null +++ b/pkg/proxmox/guest_agent_backup_test.go @@ -0,0 +1,206 @@ +package proxmox + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func backupTestClient(t *testing.T, host string) *Client { + t.Helper() + c, err := NewClient(ClientConfig{Host: host, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + return c +} + +func backupAgentReads() map[string]func(context.Context, *Client, int) error { + return map[string]func(context.Context, *Client, int) error{ + "os": func(ctx context.Context, c *Client, id int) error { + _, err := c.GetVMAgentInfo(ctx, "node", id) + return err + }, + "version": func(ctx context.Context, c *Client, id int) error { + _, err := c.GetVMAgentVersion(ctx, "node", id) + return err + }, + "filesystem": func(ctx context.Context, c *Client, id int) error { + _, err := c.GetVMFSInfo(ctx, "node", id) + return err + }, + "network": func(ctx context.Context, c *Client, id int) error { + _, err := c.GetVMNetworkInterfaces(ctx, "node", id) + return err + }, + "meminfo": func(ctx context.Context, c *Client, id int) error { + _, err := c.GetVMMemoryAvailabilityFromAgent(ctx, "node", id) + return err + }, + "legacy-memory": func(ctx context.Context, c *Client, id int) error { + _, err := c.GetVMMemAvailableFromAgent(ctx, "node", id) + return err + }, + } +} + +func backupAgentPayload(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "file-read"): + fmt.Fprint(w, `{"data":{"content":"MemTotal: 1024 kB\nMemAvailable: 512 kB\n"}}`) + case strings.HasSuffix(r.URL.Path, "info") && !strings.HasSuffix(r.URL.Path, "fsinfo") && !strings.HasSuffix(r.URL.Path, "osinfo"): + fmt.Fprint(w, `{"data":{"result":{"version":"1.0"}}}`) + case strings.HasSuffix(r.URL.Path, "get-osinfo"): + fmt.Fprint(w, `{"data":{"id":"linux"}}`) + default: + fmt.Fprint(w, `{"data":{"result":[]}}`) + } +} + +func TestGuestAgentBackupLockBlocksEveryReadAndResumes(t *testing.T) { + var locked atomic.Bool + locked.Store(true) + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if strings.HasSuffix(r.URL.Path, "/config") { + if locked.Load() { + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + } else { + fmt.Fprint(w, `{"data":{}}`) + } + return + } + if strings.Contains(r.URL.Path, "/agent/") { + calls.Add(1) + backupAgentPayload(w, r) + return + } + http.NotFound(w, r) + })) + defer server.Close() + c := backupTestClient(t, server.URL) + for name, read := range backupAgentReads() { + t.Run(name, func(t *testing.T) { + if err := read(context.Background(), c, 105); err == nil { + t.Fatal("backup-locked guest was queried") + } + }) + } + if got := calls.Load(); got != 0 { + t.Errorf("agent requests while locked = %d, want zero", got) + } + locked.Store(false) + for name, read := range backupAgentReads() { + t.Run("resumed-"+name, func(t *testing.T) { + if err := read(context.Background(), c, 105); err != nil { + t.Fatal(err) + } + }) + } + if got := calls.Load(); got != 6 { + t.Errorf("agent requests after unlock = %d, want six", got) + } +} + +func TestGuestAgentUnknownLockStateFailsClosed(t *testing.T) { + for _, body := range []string{`{"data":null}`, `{}`, `{"data":{"lock":true}}`, `{"data":{"lock":"snapshot"}}`} { + t.Run(body, func(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, body) + return + } + calls.Add(1) + backupAgentPayload(w, r) + })) + defer server.Close() + _, err := backupTestClient(t, server.URL).GetVMFSInfo(context.Background(), "node", 105) + if err == nil || calls.Load() != 0 { + t.Fatalf("unverified lock state queried agent: err=%v calls=%d", err, calls.Load()) + } + }) + } +} + +func TestGuestAgentConcurrentDiagnosticsDoNotOverlap(t *testing.T) { + var active, peak atomic.Int32 + entered := make(chan struct{}, 10) + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + n := active.Add(1) + for old := peak.Load(); n > old; old = peak.Load() { + if peak.CompareAndSwap(old, n) { + break + } + } + entered <- struct{}{} + <-release + active.Add(-1) + backupAgentPayload(w, r) + })) + defer server.Close() + // Separate Client instances model a diagnostic alongside ordinary polling. + c1, c2 := backupTestClient(t, server.URL), backupTestClient(t, server.URL) + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); _, _ = c1.GetVMFSInfo(context.Background(), "node", 105) }() + <-entered + wg.Add(1) + done := make(chan struct{}) + go func() { + defer wg.Done() + _, _ = c2.GetVMNetworkInterfaces(context.Background(), "node", 105) + close(done) + }() + select { + case <-done: + case <-entered: + case <-time.After(100 * time.Millisecond): + } + close(release) + wg.Wait() + if got := peak.Load(); got != 1 { + t.Fatalf("simultaneous guest-agent requests = %d, want one", got) + } +} + +func TestGuestAgentTimeoutDefersOtherCommands(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + calls.Add(1) + if strings.HasSuffix(r.URL.Path, "get-fsinfo") { + <-r.Context().Done() + return + } + backupAgentPayload(w, r) + })) + defer server.Close() + c := backupTestClient(t, server.URL) + ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond) + defer cancel() + if _, err := c.GetVMFSInfo(ctx, "node", 105); err == nil { + t.Fatal("missing timeout") + } + if _, err := c.GetVMAgentInfo(context.Background(), "node", 105); err == nil { + t.Error("another command was sent immediately after timeout") + } + if got := calls.Load(); got != 1 { + t.Fatalf("guest-agent requests = %d, want one", got) + } +} diff --git a/pkg/proxmox/guest_agent_guard.go b/pkg/proxmox/guest_agent_guard.go new file mode 100644 index 0000000000..d273d7ae02 --- /dev/null +++ b/pkg/proxmox/guest_agent_guard.go @@ -0,0 +1,210 @@ +package proxmox + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/securityutil" +) + +// ErrGuestAgentDeferred means no further guest-agent commands should be queued +// by the caller. In particular, a timed-out HTTP request may still be running +// in pvedaemon; another command is not a safe way to recover it. +var ErrGuestAgentDeferred = errors.New("guest agent deferred") + +type guestAgentDeferredError struct { + reason string + cause error +} + +func (e *guestAgentDeferredError) Error() string { return "guest agent deferred: " + e.reason } +func (e *guestAgentDeferredError) Is(target error) bool { return target == ErrGuestAgentDeferred } +func (e *guestAgentDeferredError) Unwrap() error { return e.cause } + +// GuestAgentDeferredReason returns a fixed reason, never provider response text. +func GuestAgentDeferredReason(err error) string { + var deferred *guestAgentDeferredError + if errors.As(err, &deferred) { + return deferred.reason + } + return "" +} + +const guestAgentUncertainCooldown = time.Minute +const maxGuestAgentGuardEntries = 4096 + +type guestAgentGuardKey struct { + endpoint string + vmid int // VMIDs are cluster-wide, including while a VM migrates nodes. +} +type guestAgentGuardEntry struct { + busy bool + until time.Time +} + +// Share admission with independently constructed diagnostic clients, not just +// the polling client. Only coordination is shared: no credentials or guest data. +var guestAgentGuards = struct { + sync.Mutex + aliases map[string][]string + entries map[guestAgentGuardKey]guestAgentGuardEntry + nextCleanup time.Time +}{aliases: make(map[string][]string), entries: make(map[guestAgentGuardKey]guestAgentGuardEntry)} + +func registerGuestAgentEndpoints(host string, endpoints []string) { + guestAgentGuards.Lock() + defer guestAgentGuards.Unlock() + group := make(map[string]bool) + for _, endpoint := range append([]string{host}, endpoints...) { + u, err := securityutil.NormalizeHTTPBaseURL(endpoint, "https") + if err != nil { + continue + } + key := securityutil.AppendURLPath(u, "api2", "json").String() + group[key] = true + for _, alias := range guestAgentGuards.aliases[key] { + group[alias] = true + } + } + keys := make([]string, 0, len(group)) + for key := range group { + keys = append(keys, key) + } + for key := range group { + guestAgentGuards.aliases[key] = keys + } +} + +func acquireGuestAgent(endpoint string, vmid int) (func(bool), error) { + guestAgentGuards.Lock() + defer guestAgentGuards.Unlock() + now := time.Now() + if !now.Before(guestAgentGuards.nextCleanup) || len(guestAgentGuards.entries) >= maxGuestAgentGuardEntries { + for key, entry := range guestAgentGuards.entries { + if !entry.busy && !now.Before(entry.until) { + delete(guestAgentGuards.entries, key) + } + } + guestAgentGuards.nextCleanup = now.Add(time.Minute) + } + endpoints := guestAgentGuards.aliases[endpoint] + if len(endpoints) == 0 { + endpoints = []string{endpoint} + } + keys := make([]guestAgentGuardKey, 0, len(endpoints)) + for _, endpoint := range endpoints { + key := guestAgentGuardKey{endpoint: endpoint, vmid: vmid} + entry := guestAgentGuards.entries[key] + if entry.busy { + return nil, &guestAgentDeferredError{reason: "agent-busy"} + } + if now.Before(entry.until) { + return nil, &guestAgentDeferredError{reason: "agent-cooldown"} + } + keys = append(keys, key) + } + if len(guestAgentGuards.entries)+len(keys) > maxGuestAgentGuardEntries { + return nil, &guestAgentDeferredError{reason: "agent-capacity"} + } + for _, key := range keys { + guestAgentGuards.entries[key] = guestAgentGuardEntry{busy: true} + } + return func(uncertain bool) { + guestAgentGuards.Lock() + defer guestAgentGuards.Unlock() + for _, key := range keys { + if uncertain { + guestAgentGuards.entries[key] = guestAgentGuardEntry{until: time.Now().Add(guestAgentUncertainCooldown)} + } else { + delete(guestAgentGuards.entries, key) + } + } + }, nil +} + +func guestAgentPath(path string) (string, int, bool) { + parts := strings.Split(strings.SplitN(path, "?", 2)[0], "/") + if len(parts) != 7 || parts[1] != "nodes" || parts[3] != "qemu" || parts[5] != "agent" { + return "", 0, false + } + vmid, err := strconv.Atoi(parts[4]) + return parts[2], vmid, err == nil && vmid > 0 +} + +func (c *Client) verifyGuestAgentUnlocked(ctx context.Context, node string, vmid int) error { + // The config endpoint contains the authoritative PVE operation lock even + // when a PVE version omits it from status/current or cluster/resources. + config, err := c.GetVMConfig(ctx, node, vmid) + if err != nil || config == nil { + return &guestAgentDeferredError{reason: "lock-unverified", cause: err} + } + if value, exists := config["lock"]; exists { + lock, ok := value.(string) + if !ok { + return &guestAgentDeferredError{reason: "lock-unverified"} + } + if strings.TrimSpace(lock) != "" { + return &guestAgentDeferredError{reason: "vm-locked"} + } + } + return nil +} + +func (c *Client) getGuestAgent(ctx context.Context, path, node string, vmid int) (*http.Response, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + release, err := acquireGuestAgent(c.baseURL, vmid) + if err != nil { + return nil, err + } + uncertain := false + defer func() { release(uncertain) }() + if err := c.verifyGuestAgentUnlocked(ctx, node, vmid); err != nil { + return nil, err + } + resp, err := c.request(ctx, http.MethodGet, path, nil) + if err != nil { + // Explicit refusals/unsupported commands remain ordinary errors. An + // uncertain completion blocks every command, not just this method. + lower := strings.ToLower(err.Error()) + uncertain = !strings.Contains(lower, "api error") || ctx.Err() != nil || strings.Contains(lower, "timeout") || strings.Contains(lower, "timed out") || strings.Contains(lower, "wrong command id") + if uncertain { + return nil, &guestAgentDeferredError{reason: "agent-timeout", cause: err} + } + return nil, err + } + // Hold admission until the entire bounded response has arrived. Returning + // at response headers would allow another command while the first stalls. + body, readErr := readResponseBodyLimited(resp.Body) + closeErr := resp.Body.Close() + if readErr != nil || closeErr != nil { + uncertain = true + return nil, &guestAgentDeferredError{reason: "agent-response-incomplete", cause: errors.Join(readErr, closeErr)} + } + // A backup may have started while the command was in flight. Do not publish + // its payload as fresh telemetry if lock clearance cannot still be verified. + if err := c.verifyGuestAgentUnlocked(ctx, node, vmid); err != nil { + return nil, err + } + resp.Body = io.NopCloser(bytes.NewReader(body)) + return resp, nil +} + +// Guest calls do not fail over or replay: another HTTP worker still addresses +// the same serial QGA channel. Transport recovery belongs to ordinary API reads. +func (cc *ClusterClient) executeGuestAgent(ctx context.Context, fn func(*Client) error) error { + client, err := cc.getHealthyClient(ctx) + if err != nil { + return fmt.Errorf("select guest agent endpoint: %w", err) + } + return fn(client) +} diff --git a/pkg/proxmox/guest_agent_guard_test.go b/pkg/proxmox/guest_agent_guard_test.go new file mode 100644 index 0000000000..bf27f30353 --- /dev/null +++ b/pkg/proxmox/guest_agent_guard_test.go @@ -0,0 +1,191 @@ +package proxmox + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestGuestAgentClusterAliasesAndVMIsolation(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + first := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/nodes"): + fmt.Fprint(w, `{"data":[]}`) + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{}}`) + case strings.Contains(r.URL.Path, "/agent/"): + close(entered) + <-release + backupAgentPayload(w, r) + default: + http.NotFound(w, r) + } + })) + defer first.Close() + var calls atomic.Int32 + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/nodes") { + fmt.Fprint(w, `{"data":[]}`) + return + } + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + calls.Add(1) + backupAgentPayload(w, r) + }) + second := httptest.NewServer(handler) + defer second.Close() + independent := httptest.NewServer(handler) + defer independent.Close() + NewClusterClient("aliases", ClientConfig{Host: first.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}, []string{first.URL, second.URL}, nil) + c1, c2, c3 := backupTestClient(t, first.URL), backupTestClient(t, second.URL), backupTestClient(t, independent.URL) + done := make(chan error, 1) + go func() { _, err := c1.GetVMFSInfo(context.Background(), "node", 105); done <- err }() + <-entered + defer func() { + close(release) + if err := <-done; err != nil { + t.Error(err) + } + }() + if _, err := c2.GetVMAgentInfo(context.Background(), "migrated-node", 105); !errors.Is(err, ErrGuestAgentDeferred) { + t.Errorf("alias bypassed busy VM: %v", err) + } + if _, err := c2.GetVMAgentInfo(context.Background(), "node", 106); err != nil { + t.Errorf("different VM blocked: %v", err) + } + if _, err := c3.GetVMAgentInfo(context.Background(), "node", 105); err != nil { + t.Errorf("independent source blocked: %v", err) + } + if calls.Load() != 2 { + t.Errorf("healthy isolated calls = %d", calls.Load()) + } +} + +func TestGuestAgentLockAppearingInFlightDiscardsPayload(t *testing.T) { + var locked atomic.Bool + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + if locked.Load() { + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + } else { + fmt.Fprint(w, `{"data":{}}`) + } + return + } + calls.Add(1) + locked.Store(true) + backupAgentPayload(w, r) + })) + defer server.Close() + c := backupTestClient(t, server.URL) + if value, err := c.GetVMFSInfo(context.Background(), "node", 105); value != nil || GuestAgentDeferredReason(err) != "vm-locked" { + t.Fatalf("in-flight backup payload published: %#v %v", value, err) + } + if _, err := c.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "vm-locked" { + t.Fatal(err) + } + if calls.Load() != 1 { + t.Fatalf("another command queued: %d", calls.Load()) + } +} + +func TestGuestAgentIncompleteBodyBlocksWholeVM(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + calls.Add(1) + w.Header().Set("Content-Length", "1000") + fmt.Fprint(w, `{"data":{"result":[]}}`) + })) + defer server.Close() + c := backupTestClient(t, server.URL) + if _, err := c.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-response-incomplete" { + t.Fatalf("incomplete body not deferred: %v", err) + } + for name, read := range backupAgentReads() { + t.Run(name, func(t *testing.T) { + if err := read(context.Background(), c, 105); GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Fatal(err) + } + }) + } + if calls.Load() != 1 { + t.Fatalf("commands after incomplete body = %d", calls.Load()) + } +} + +func TestGuestAgentClusterDoesNotReplayUncertainCall(t *testing.T) { + var calls atomic.Int32 + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/nodes"): + fmt.Fprint(w, `{"data":[]}`) + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{}}`) + default: + calls.Add(1) + http.Error(w, "QEMU guest command timed out", 500) + } + }) + a, b := httptest.NewServer(handler), httptest.NewServer(handler) + defer a.Close() + defer b.Close() + cc := NewClusterClient("timeouts", ClientConfig{Host: a.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}, []string{a.URL, b.URL}, nil) + if _, err := cc.GetVMFSInfo(context.Background(), "node", 105); !errors.Is(err, ErrGuestAgentDeferred) { + t.Fatalf("uncertain completion lost: %v", err) + } + if _, err := cc.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Fatalf("another cluster command sent: %v", err) + } + if calls.Load() != 1 { + t.Fatalf("uncertain call replayed: %d", calls.Load()) + } + for endpoint, healthy := range cc.GetHealthStatus() { + if !healthy { + t.Errorf("VM timeout poisoned endpoint %s", endpoint) + } + } +} + +func TestGuestAgentCooldownExpiresWithoutSleepOrReplay(t *testing.T) { + endpoint := "http://fixture.invalid/api2/json" + release, err := acquireGuestAgent(endpoint, 105) + if err != nil { + t.Fatal(err) + } + release(true) + if _, err := acquireGuestAgent(endpoint, 105); GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Fatal(err) + } + guestAgentGuards.Lock() + key := guestAgentGuardKey{endpoint: endpoint, vmid: 105} + entry := guestAgentGuards.entries[key] + entry.until = time.Now().Add(-time.Second) + guestAgentGuards.entries[key] = entry + guestAgentGuards.Unlock() + release, err = acquireGuestAgent(endpoint, 105) + if err != nil { + t.Fatal(err) + } + release(false) + guestAgentGuards.Lock() + _, retained := guestAgentGuards.entries[key] + guestAgentGuards.Unlock() + if retained { + t.Fatal("healthy completed operation retained estate state") + } +} diff --git a/pkg/proxmox/io_counters_test.go b/pkg/proxmox/io_counters_test.go index 8259242b34..fc210c76ab 100644 --- a/pkg/proxmox/io_counters_test.go +++ b/pkg/proxmox/io_counters_test.go @@ -115,3 +115,26 @@ func TestInternalCounterMetadataNeverChangesProxmoxWireShape(t *testing.T) { } } } + +func TestBackupLocksRetainCumulativeCounterPresence(t *testing.T) { + payload := []byte(`{"lock":"backup","diskread":0,"diskwrite":null,"netin":42}`) + var listing VM + var resource ClusterResource + var status VMStatus + for _, dst := range []any{&listing, &resource, &status} { + if err := json.Unmarshal(payload, dst); err != nil { + t.Fatal(err) + } + } + for name, value := range map[string]struct { + lock string + presence IOCounterPresence + }{"listing": {listing.Lock, listing.IOCounters}, "resource": {resource.Lock, resource.IOCounters}, "status": {status.Lock, status.IOCounters}} { + t.Run(name, func(t *testing.T) { + p := value.presence.Effective() + if value.lock != "backup" || !p.DiskRead || !p.NetworkIn || p.DiskWrite || p.NetworkOut { + t.Fatalf("lock decoding corrupted presence: %#v", value) + } + }) + } +}