Read guest filesystems before optional QGA enrichment

Prevent optional Linux memory and metadata failures from starving supported filesystem usage. Carry shared admission deferral independently of a completed disk sample. Preserve backup locks, uncertain completion, no replay and original observation ages; bind both poll paths and truthful History to connected controls. This is source repair, not a native discussion #2538 cause or installed recovery.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-06 00:30:02 +01:00
parent 28c64e3808
commit 7b5cdfc5c3
13 changed files with 386 additions and 114 deletions
@@ -202,6 +202,33 @@ and units on 8- and 4-CPU hosts. Genuine high usage still breaches its configure
threshold; no clipping or smoothing masks spikes. These source controls do not
establish the reporter's native payload, installed cause or containing release.
### Filesystem collection precedes optional guest reads — discussion #2538
Both the detailed-status and recent-evidence/no-status QEMU builders collect
filesystem usage before optional Linux `/proc/meminfo` and guest metadata.
Windows/Android guests may answer filesystem queries without supporting every
optional command. A completed, verified filesystem response can therefore reach
the read projection and History even if a subsequent optional command starts a
shared pause. Memory and metadata helpers return admission deferral separately
from observations: a successful disk must not hide later uncertainty.
No completion-error exemption, operation-lock requirement, timeout, retry,
failover or process-local coordination rule changes. An unknown optional reply
still blocks every later command, including newly constructed clients and cold
collectors in the same process. Subsequent paused polls retain old values without
adding disk/memory History or renewing original memory evidence. Explicit
unsupported replies retain their existing terminal/error and negative-cache
semantics. Independent current provider counters remain usable. This does not
establish the reporter's original failed command, supported native filesystem
response, process-restart safety, thaw or installed/released repair.
`TestGuestAgentOptionalReadOrderingContract` drives actual single/cluster clients,
normal and recent-evidence/no-status builders, unified JSON readback and memory,
persistent and chart History for NTFS/ext4 success, terminal unsupported memory/OS
and uncertain memory/network/OS/version replies across repeated polls and cold
collectors. Existing backup, lock, transport and observation contracts remain
required; all responses in this fixture are synthetic.
### Guest-agent coordination around backups — issue #2439
All PVE QEMU guest-agent reads, including disk diagnostics and legacy meminfo
@@ -913,7 +913,7 @@ func TestProxmoxGuestMemoryFallbackUsesInstanceScopedCachesAndAgentMeminfo(t *te
// the resolver must not consult guest RRD between the preferred
// guest-agent pass and the retry pass.
"// No RRD fallback here: PVE guest rrddata carries only cache-inclusive",
"if agentAvailable, agentSource, ok := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw); ok {",
"agentAvailable, agentSource, ok, deferred := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw)",
"memorySource = agentSource",
"guestRaw.GuestAgentMemAvailable = agentAvailable",
`memorySource = "unavailable"`,
@@ -2981,6 +2981,11 @@ func TestGuestAgentBackupMonitoringContract(t *testing.T) {
testGuestAgentBackupMonitoringLifecycle(t)
}
func TestGuestAgentOptionalReadOrderingContract(t *testing.T) {
t.Run("current-status", func(t *testing.T) { testGuestAgentOptionalReadOrdering(t, false) })
t.Run("recent-evidence-without-status", func(t *testing.T) { testGuestAgentOptionalReadOrdering(t, true) })
}
func TestGuestMemoryObservationContract(t *testing.T) {
t.Run("poll-to-served-observation", testGuestMemoryObservationLifecycle)
t.Run("identity-and-origin-boundaries", testGuestMemoryObservationKeepsOriginsSeparate)
@@ -229,7 +229,7 @@ func TestGuestAgentDeferralPreservesMetadataAndMemoryCache(t *testing.T) {
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)
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")
}
@@ -0,0 +1,225 @@
package monitoring
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
)
// These are deliberately synthetic protocol outcomes, not captured responses
// from discussion #2538. Filesystem success is independent of optional Linux
// memory/metadata support, but an uncertain command still stops every later read.
func testGuestAgentOptionalReadOrdering(t *testing.T, withoutStatus bool) {
for _, clientKind := range []string{"client", "cluster"} {
for _, tc := range []struct {
name, failure, os, fs string
terminal bool
}{
{"windows-unsupported-memory", "file-read", "Microsoft Windows 11", "ntfs", true},
{"android-unsupported-os", "get-osinfo", "Android", "ext4", true},
{"uncertain-memory", "file-read", "Microsoft Windows 11", "ntfs", false},
{"uncertain-network", "network-get-interfaces", "Android", "ext4", false},
{"uncertain-os", "get-osinfo", "Android", "ext4", false},
{"uncertain-version", "info", "Microsoft Windows 11", "ntfs", false},
} {
if withoutStatus && tc.failure != "get-osinfo" {
continue // No-status enrichment does not issue a memory fallback.
}
t.Run(clientKind+"/"+tc.name, func(t *testing.T) {
const mib = uint64(1024 * 1024)
var poll atomic.Int32
var mu sync.Mutex
var commands []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" {
t.Error("lost configured authentication")
w.WriteHeader(http.StatusForbidden)
return
}
parts := strings.Split(r.URL.Path, "/")
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.HasSuffix(r.URL.Path, "/status/current"):
if withoutStatus {
w.WriteHeader(http.StatusForbidden)
return
}
fmt.Fprintf(w, `{"data":{"status":"running","agent":1,"maxmem":%d,"mem":%d}}`, 8*mib, 8*mib)
case strings.Contains(r.URL.Path, "/agent/"):
command := parts[len(parts)-1]
id, _ := strconv.Atoi(parts[len(parts)-3])
mu.Lock()
commands = append(commands, fmt.Sprintf("%d/%s", id, command))
mu.Unlock()
if id == 105 && command == tc.failure {
w.WriteHeader(http.StatusInternalServerError)
if tc.terminal {
fmt.Fprint(w, "unsupported command: guest-"+command)
} else {
// A complete but unexplained 500, even mentioning a
// missing Linux file, is NOT proof of completion.
fmt.Fprint(w, "optional guest read failed: /proc/meminfo unavailable")
}
return
}
switch command {
case "get-fsinfo":
json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"result": []any{map[string]any{"mountpoint": map[bool]string{true: `C:\`, false: "/"}[tc.fs == "ntfs"], "type": tc.fs, "total-bytes": 1000 * mib, "used-bytes": uint64(200+poll.Load()*100) * mib}}}})
case "file-read":
if r.URL.Query().Get("file") != "/proc/meminfo" {
t.Error("unexpected guest file")
}
json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"content": "MemTotal: 8192 kB\nMemAvailable: 5120 kB\n"}})
case "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 "get-osinfo":
json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"name": tc.os, "version": "fixture"}})
case "info":
fmt.Fprint(w, `{"data":{"result":{"version":"fixture"}}}`)
default:
t.Error("unexpected guest command", command)
http.NotFound(w, r)
}
default:
http.NotFound(w, r)
}
}))
defer server.Close()
newClient := func() PVEClientInterface {
cfg := proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}
if clientKind == "cluster" {
return proxmox.NewClusterClient("optional", cfg, []string{server.URL}, nil)
}
client, err := proxmox.NewClient(cfg)
if err != nil {
t.Fatal(err)
}
return client
}
client := newClient()
m := guestHistoryObservationMonitor(t)
m.rateTracker = NewRateTracker()
m.guestMetadataLimiter = make(map[string]time.Time)
registry := unifiedresources.NewRegistry(nil)
res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "guest", VMID: 105, Status: "running", MaxMem: 8 * mib, Mem: 8 * mib, MaxDisk: 1000 * mib, CPU: .1}
id := makeGuestID("optional", "node", 105)
var previous *models.VM
if withoutStatus {
previous = &models.VM{ID: id, GuestAgentStatus: "available", Status: "running", LastSeen: time.Now()}
}
build := func() models.VM {
vm, raw, source, notes, at, ok := m.buildVMFromClusterResource(context.Background(), "optional", res, client, id, nil, previous)
if !ok {
t.Fatal("guest disappeared")
}
m.recordGuestSnapshot("optional", "qemu", "node", 105, GuestMemorySnapshot{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))
registry.IngestSnapshot(models.StateSnapshot{VMs: []models.VM{vm}})
view := registry.VMs()[0]
if view.ID() != id || view.DiskUsed() != vm.Disk.Used || view.DiskStatusReason() != vm.DiskStatusReason {
t.Fatal("filesystem observation changed at the read boundary")
}
encoded, err := json.Marshal(registry.GetAll())
if err != nil || !strings.Contains(string(encoded), id) {
t.Fatal("guest identity missing from JSON read projection")
}
next := previousVMFromView(view)
previous = &next
return vm
}
poll.Store(1)
initial := build()
if initial.Disk.Used != int64(300*mib) || initial.Disk.Usage != 30 || initial.DiskStatusReason != "" {
t.Fatalf("optional %s starved independent filesystem success: used=%d usage=%v reason=%q", tc.failure, initial.Disk.Used, initial.Disk.Usage, initial.DiskStatusReason)
}
if !tc.terminal && initial.GuestAgentStatus != "deferred" {
t.Fatal("successful filesystem concealed later command uncertainty")
}
if tc.terminal && initial.GuestAgentStatus == "deferred" {
t.Fatal("completed unsupported command became a shared pause")
}
mu.Lock()
firstCommands := append([]string(nil), commands...)
mu.Unlock()
if firstCommands[0] != "105/get-fsinfo" {
t.Fatalf("first command = %q, want filesystem before optional reads", firstCommands[0])
}
memoryKey := guestMemoryCacheKey("optional", "node", 105)
originalMemory := m.vmAgentMemCache[memoryKey]
originalDisk := guestHistoryStoredPoints(t, m, "vm", id, "disk")
if len(originalDisk) != 1 || originalDisk[0].Value != 30 {
t.Fatal("initial filesystem did not reach persistent History")
}
for i := int32(2); i <= 3; i++ {
// Distinct persisted timestamps ensure a falsely renewed value
// cannot hide behind a same-second storage key.
time.Sleep(1100 * time.Millisecond)
poll.Store(i)
client = newClient() // A new diagnostic/cluster client cannot escape admission.
vm := build()
if tc.terminal {
if vm.Disk.Usage != float64(20+i*10) || vm.DiskStatusReason != "" || vm.GuestAgentStatus == "deferred" {
t.Fatal("completed unsupported reply prevented repeated fresh filesystem polls")
}
} else {
if vm.Disk.Used != initial.Disk.Used || vm.DiskStatusReason != "prev-agent-cooldown" || vm.GuestAgentStatus != "deferred" {
t.Fatalf("uncertainty lost truthful retained disk: %+v", vm.Disk)
}
if !reflect.DeepEqual(m.vmAgentMemCache[memoryKey], originalMemory) || !vm.Memory.Observation.ObservedAt.Equal(initial.Memory.Observation.ObservedAt) {
t.Fatal("cooldown renewed original memory evidence")
}
}
}
if !tc.terminal {
mu.Lock()
unchanged := reflect.DeepEqual(commands, firstCommands)
mu.Unlock()
if !unchanged || !reflect.DeepEqual(guestHistoryStoredPoints(t, m, "vm", id, "disk"), originalDisk) {
t.Fatal("cooldown dispatched another command or renewed persistent disk History")
}
// A cold collector on the same endpoint shares the live process's
// guard. This is NOT a new Pulse process or native restart proof.
cold := &Monitor{rateTracker: NewRateTracker(), guestMetadataLimiter: make(map[string]time.Time)}
vm, _, _, _, _, _ := cold.buildVMFromClusterResource(context.Background(), "optional", res, newClient(), id, nil, nil)
if !withoutStatus && (vm.Disk.Usage != -1 || vm.DiskStatusReason != "agent-cooldown" || vm.GuestAgentStatus != "deferred") {
t.Fatal("cold collector bypassed active cooldown or fabricated a disk")
}
mu.Lock()
unchanged = reflect.DeepEqual(commands, firstCommands)
mu.Unlock()
if !unchanged {
t.Fatal("cold collector sent a command during active cooldown")
}
}
wantDiskPoints := 1
if tc.terminal {
wantDiskPoints = 3
}
for _, points := range [][]MetricPoint{m.metricsHistory.GetGuestMetrics(id, "disk", time.Hour), m.GetGuestMetricsForChart(id, "vm", id, time.Hour)["disk"]} {
if len(points) != wantDiskPoints {
t.Fatalf("disk History/chart has %d points, want %d", len(points), wantDiskPoints)
}
}
if len(guestHistoryStoredPoints(t, m, "vm", id, "disk")) != wantDiskPoints || len(m.metricsHistory.GetGuestMetrics(id, "cpu", time.Hour)) != 3 {
t.Fatal("persistent disk or independent CPU history disagreed with polling")
}
})
}
}
}
@@ -111,7 +111,18 @@ func testGuestAgentTransportDeferralKeepsLastKnownHistory(t *testing.T) {
res.CPU = 0.2
for i := 0; i < 2; i++ {
deferred, source := build()
if deferred.ID != initial.ID || deferred.GuestAgentStatus != "deferred" || deferred.DiskStatusReason != "prev-agent-cooldown" || deferred.Memory.Used != initial.Memory.Used || source != "previous-snapshot" || deferred.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(deferred.NetworkInterfaces, initial.NetworkInterfaces) {
wantReason := "prev-agent-cooldown"
if i == 0 {
// Filesystem now receives the original error, before optional
// reads observe the resulting shared cooldown.
wantReason = "prev-agent-completion-unverified"
if kind == "lost reply" {
wantReason = "prev-agent-timeout"
} else if kind == "redirect" {
wantReason = "prev-agent-redirect"
}
}
if deferred.ID != initial.ID || deferred.GuestAgentStatus != "deferred" || deferred.DiskStatusReason != wantReason || deferred.Memory.Used != initial.Memory.Used || source != "previous-snapshot" || deferred.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(deferred.NetworkInterfaces, initial.NetworkInterfaces) {
t.Fatalf("uncertain command lost truthful continuity: source=%s vm=%+v", source, deferred)
}
if !reflect.DeepEqual(m.vmAgentMemCache[memKey], memory) || !reflect.DeepEqual(m.guestMetadataCache[metadataKey], metadata) {
@@ -66,7 +66,7 @@ func TestResolveGuestStatusMemoryNeverConsultsGuestRRD(t *testing.T) {
stubPVEClient: &stubPVEClient{},
}
total, used, source := mon.resolveGuestStatusMemory(
total, used, source, _ := mon.resolveGuestStatusMemory(
context.Background(),
client,
"pve-a",
@@ -105,7 +105,7 @@ func TestResolveGuestStatusMemoryAcceptsExplicitZeroGuestAgentAvailable(t *testi
},
}
total, used, source := mon.resolveGuestStatusMemory(
total, used, source, _ := mon.resolveGuestStatusMemory(
context.Background(),
client,
"pve-a",
@@ -318,7 +318,7 @@ func TestResolveGuestStatusMemoryUsesGuestAgentMeminfoFallback(t *testing.T) {
}
raw := &VMMemoryRaw{}
memTotal, memUsed, source := mon.resolveGuestStatusMemory(
memTotal, memUsed, source, _ := mon.resolveGuestStatusMemory(
context.Background(),
client,
"pve-a",
@@ -367,7 +367,7 @@ func TestResolveGuestStatusMemoryPrefersGuestAgentMeminfoForSaturatedStatus(t *t
}
raw := &VMMemoryRaw{}
memTotal, memUsed, source := mon.resolveGuestStatusMemory(
memTotal, memUsed, source, _ := mon.resolveGuestStatusMemory(
context.Background(),
client,
"cluster-a",
+17 -9
View File
@@ -2,6 +2,7 @@ package monitoring
import (
"context"
"errors"
"math"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
@@ -198,6 +199,7 @@ func guestMemoryFallbackReason(source string) string {
return MemorySourceFallbackReason(source)
}
// The final result is admission deferral, not a failed or successful sample.
func (m *Monitor) tryGuestAgentMemAvailable(
ctx context.Context,
client PVEClientInterface,
@@ -207,10 +209,10 @@ func (m *Monitor) tryGuestAgentMemAvailable(
vmid int,
memTotal uint64,
guestRaw *VMMemoryRaw,
) (uint64, string, bool) {
) (uint64, string, bool, bool) {
availability, agentErr := m.getVMAgentMemoryAvailability(ctx, client, instanceName, node, vmid)
if agentErr != nil || availability.Source == "" {
return 0, "", false
return 0, "", false, errors.Is(agentErr, proxmox.ErrGuestAgentDeferred)
}
agentAvailable := availability.EffectiveAvailable
if guestRaw != nil {
@@ -223,7 +225,7 @@ func (m *Monitor) tryGuestAgentMemAvailable(
guestRaw.GuestAgentDerived = availability.Source == "meminfo-derived"
}
if memTotal == 0 || agentAvailable > memTotal {
return 0, "", false
return 0, "", false, false
}
source := "guest-agent-meminfo"
if availability.Source == "meminfo-derived" {
@@ -237,9 +239,10 @@ func (m *Monitor) tryGuestAgentMemAvailable(
Uint64("available", agentAvailable).
Str("source", source).
Msg("QEMU memory: using guest agent /proc/meminfo fallback (excludes reclaimable cache)")
return agentAvailable, source, true
return agentAvailable, source, true, false
}
// Return shared guest-command deferral even when independent memory is usable.
func (m *Monitor) resolveGuestStatusMemory(
ctx context.Context,
client PVEClientInterface,
@@ -253,9 +256,9 @@ func (m *Monitor) resolveGuestStatusMemory(
memTotal uint64,
memorySource string,
guestRaw *VMMemoryRaw,
) (uint64, uint64, string) {
) (uint64, uint64, string, bool) {
if status == nil {
return memTotal, 0, memorySource
return memTotal, 0, memorySource, false
}
if guestRaw != nil {
@@ -297,9 +300,12 @@ func (m *Monitor) resolveGuestStatusMemory(
}
triedGuestAgentMemAvailable := false
guestReadDeferred := false
if !hasMemAvailable && shouldPreferGuestAgentMemAvailable(status, memTotal) {
triedGuestAgentMemAvailable = true
if agentAvailable, agentSource, ok := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw); ok {
agentAvailable, agentSource, ok, deferred := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw)
guestReadDeferred = guestReadDeferred || deferred
if ok {
memAvailable = agentAvailable
hasMemAvailable = true
memorySource = agentSource
@@ -312,7 +318,9 @@ func (m *Monitor) resolveGuestStatusMemory(
// (#1634).
if !hasMemAvailable && !hasSelectedUsed && status.Lock == "" && status.Agent.IsAvailable() && !triedGuestAgentMemAvailable {
if agentAvailable, agentSource, ok := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw); ok {
agentAvailable, agentSource, ok, deferred := m.tryGuestAgentMemAvailable(ctx, client, instanceName, guestName, node, vmid, memTotal, guestRaw)
guestReadDeferred = guestReadDeferred || deferred
if ok {
memAvailable = agentAvailable
hasMemAvailable = true
memorySource = agentSource
@@ -380,7 +388,7 @@ func (m *Monitor) resolveGuestStatusMemory(
memUsed = memTotal
}
return memTotal, memUsed, memorySource
return memTotal, memUsed, memorySource, guestReadDeferred
}
// preferLinkedAgentLXCMemory replaces a low-trust Proxmox LXC memory reading
+14 -12
View File
@@ -222,7 +222,9 @@ func (m *Monitor) retryGuestAgentCall(ctx context.Context, timeout time.Duration
return nil, lastErr
}
func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientInterface, instanceName, nodeName, vmName string, vmid int, vmStatus *proxmox.VMStatus, allowWithoutStatus bool) ([]string, []models.GuestNetworkInterface, string, string, string) {
// The final result reports command deferral separately from retained metadata.
// A previously successful filesystem must not conceal a later shared pause.
func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientInterface, instanceName, nodeName, vmName string, vmid int, vmStatus *proxmox.VMStatus, allowWithoutStatus bool) ([]string, []models.GuestNetworkInterface, string, string, string, bool) {
key := guestMetadataCacheKey(instanceName, nodeName, vmid)
now := time.Now()
@@ -232,30 +234,30 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI
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
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, false
}
agentAvailable := client != nil && ((vmStatus != nil && vmStatus.Agent.IsAvailable()) || allowWithoutStatus)
if !agentAvailable {
if ok && now.Sub(cached.fetchedAt) < guestMetadataCacheEntryTTL(cached) {
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, false
}
m.clearGuestMetadataCache(instanceName, nodeName, vmid)
return nil, nil, "", "", ""
return nil, nil, "", "", "", false
}
if ok && now.Sub(cached.fetchedAt) < guestMetadataCacheEntryTTL(cached) {
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, false
}
needsFetch := !ok || now.Sub(cached.fetchedAt) >= guestMetadataCacheEntryTTL(cached)
if !needsFetch {
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, false
}
reserved := m.tryReserveGuestMetadataFetch(key, now)
if !reserved && ok {
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, false
}
if !reserved && !ok {
reserved = true
@@ -272,7 +274,7 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI
if reserved {
if !m.acquireGuestMetadataSlot(ctx) {
m.deferGuestMetadataRetry(key, time.Now())
return ipAddresses, networkIfaces, osName, osVersion, agentVersion
return ipAddresses, networkIfaces, osName, osVersion, agentVersion, false
}
defer m.releaseGuestMetadataSlot()
}
@@ -283,7 +285,7 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI
})
if errors.Is(err, proxmox.ErrGuestAgentDeferred) {
m.deferGuestMetadataRetry(key, time.Now())
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, true
}
if err != nil {
log.Debug().
@@ -325,7 +327,7 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI
})
if errors.Is(err, proxmox.ErrGuestAgentDeferred) {
m.deferGuestMetadataRetry(key, time.Now())
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, true
}
if err != nil {
if isGuestAgentOSInfoUnsupportedError(err) {
@@ -384,7 +386,7 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI
})
if errors.Is(err, proxmox.ErrGuestAgentDeferred) {
m.deferGuestMetadataRetry(key, time.Now())
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion
return cloneStringSlice(cached.ipAddresses), cloneGuestNetworkInterfaces(cached.networkInterfaces), cached.osName, cached.osVersion, cached.agentVersion, true
}
if err != nil {
log.Debug().
@@ -432,7 +434,7 @@ func (m *Monitor) fetchGuestAgentMetadata(ctx context.Context, client PVEClientI
}
}
return ipAddresses, networkIfaces, osName, osVersion, agentVersion
return ipAddresses, networkIfaces, osName, osVersion, agentVersion, false
}
func guestMetadataCacheKey(instanceName, nodeName string, vmid int) string {
@@ -71,7 +71,7 @@ func TestGuestMetadataFailedRefreshPreservesObservation(t *testing.T) {
for attempt := 0; attempt < 2; attempt++ {
m.guestMetadataLimiter[key] = time.Now().Add(-time.Second)
start := time.Now()
ips, ifaces, name, version, agent := m.fetchGuestAgentMetadata(context.Background(), &client, "metadata", "node", "guest", 105, status, false)
ips, ifaces, name, version, agent, _ := m.fetchGuestAgentMetadata(context.Background(), &client, "metadata", "node", "guest", 105, status, false)
end := time.Now()
if !reflect.DeepEqual(ips, original.ipAddresses) || !reflect.DeepEqual(ifaces, original.networkInterfaces) || name != original.osName || version != original.osVersion || agent != original.agentVersion {
t.Error("unsuccessful refresh lost last-known identity")
@@ -223,7 +223,7 @@ func TestGuestMetadataObservationHTTPPollLifecycle(t *testing.T) {
m.guestMetadataCache[key] = original
m.guestMetadataLimiter[key] = time.Now().Add(-time.Second)
phase.Store(1)
ips, ifaces, name, version, agent := m.fetchGuestAgentMetadata(context.Background(), client, "metadata", "node", "guest", 105, &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}}, false)
ips, ifaces, name, version, agent, _ := m.fetchGuestAgentMetadata(context.Background(), client, "metadata", "node", "guest", 105, &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}}, false)
if !reflect.DeepEqual(ips, first.IPAddresses) || !reflect.DeepEqual(ifaces, first.NetworkInterfaces) || name != first.OSName || version != first.OSVersion || agent != first.AgentVersion {
t.Error("failed HTTP refresh lost last-known metadata")
}
+9 -9
View File
@@ -1004,7 +1004,7 @@ func TestFetchGuestAgentMetadataPreservesCachedValuesOnEmptyResponses(t *testing
guestMetadataLimiter: make(map[string]time.Time),
}
gotIPs, gotIfaces, gotOSName, gotOSVersion, gotAgentVersion := monitor.fetchGuestAgentMetadata(
gotIPs, gotIfaces, gotOSName, gotOSVersion, gotAgentVersion, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
&emptyGuestMetadataClient{},
"pve",
@@ -1093,7 +1093,7 @@ func TestFetchGuestAgentMetadataPreservesFreshCacheWhenAgentTemporarilyUnavailab
t.Parallel()
monitor := newMonitor()
gotIPs, gotIfaces, gotOSName, gotOSVersion, gotAgentVersion := monitor.fetchGuestAgentMetadata(
gotIPs, gotIfaces, gotOSName, gotOSVersion, gotAgentVersion, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
&emptyGuestMetadataClient{},
"pve",
@@ -1111,7 +1111,7 @@ func TestFetchGuestAgentMetadataPreservesFreshCacheWhenAgentTemporarilyUnavailab
t.Parallel()
monitor := newMonitor()
gotIPs, gotIfaces, gotOSName, gotOSVersion, gotAgentVersion := monitor.fetchGuestAgentMetadata(
gotIPs, gotIfaces, gotOSName, gotOSVersion, gotAgentVersion, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
&emptyGuestMetadataClient{},
"pve",
@@ -1137,7 +1137,7 @@ func TestFetchGuestAgentMetadataRetriesIdentityOnlyCacheSooner(t *testing.T) {
status := &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}}
firstIPs, firstIfaces, firstOSName, firstOSVersion, firstAgentVersion := monitor.fetchGuestAgentMetadata(
firstIPs, firstIfaces, firstOSName, firstOSVersion, firstAgentVersion, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
client,
"pve",
@@ -1163,7 +1163,7 @@ func TestFetchGuestAgentMetadataRetriesIdentityOnlyCacheSooner(t *testing.T) {
monitor.guestMetadataCache[key] = entry
monitor.guestMetadataLimiter[key] = time.Now().Add(-time.Second)
secondIPs, secondIfaces, secondOSName, secondOSVersion, secondAgentVersion := monitor.fetchGuestAgentMetadata(
secondIPs, secondIfaces, secondOSName, secondOSVersion, secondAgentVersion, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
client,
"pve",
@@ -1194,7 +1194,7 @@ func TestFetchGuestAgentMetadataRetriesIPOnlyCacheSooner(t *testing.T) {
status := &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}}
firstIPs, firstIfaces, _, _, _ := monitor.fetchGuestAgentMetadata(
firstIPs, firstIfaces, _, _, _, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
client,
"pve",
@@ -1220,7 +1220,7 @@ func TestFetchGuestAgentMetadataRetriesIPOnlyCacheSooner(t *testing.T) {
monitor.guestMetadataCache[key] = entry
monitor.guestMetadataLimiter[key] = time.Now().Add(-time.Second)
secondIPs, secondIfaces, _, _, _ := monitor.fetchGuestAgentMetadata(
secondIPs, secondIfaces, _, _, _, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
client,
"pve",
@@ -1252,7 +1252,7 @@ func TestFetchGuestAgentMetadataRetriesEmptyCacheSooner(t *testing.T) {
status := &proxmox.VMStatus{Agent: proxmox.VMAgentField{Value: 1}}
firstIPs, firstIfaces, _, _, _ := monitor.fetchGuestAgentMetadata(
firstIPs, firstIfaces, _, _, _, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
client,
"pve",
@@ -1272,7 +1272,7 @@ func TestFetchGuestAgentMetadataRetriesEmptyCacheSooner(t *testing.T) {
monitor.guestMetadataCache[key] = entry
monitor.guestMetadataLimiter[key] = time.Now().Add(-time.Second)
secondIPs, secondIfaces, _, _, _ := monitor.fetchGuestAgentMetadata(
secondIPs, secondIfaces, _, _, _, _ := monitor.fetchGuestAgentMetadata(
context.Background(),
client,
"pve",
@@ -1184,7 +1184,7 @@ func TestBackupLockMemoryTrustKeepsNativeEvidenceWithoutAgentReads(t *testing.T)
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{})
_, 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)
}
@@ -1193,7 +1193,7 @@ func TestBackupLockMemoryTrustKeepsNativeEvidenceWithoutAgentReads(t *testing.T)
}
status.Lock = ""
status.MemInfo = nil
_, _, source = m.resolveGuestStatusMemory(context.Background(), client, "fixture", "guest", "node", 105, "fixture:node:105", status, nil, 8*gib, "", &VMMemoryRaw{})
_, _, 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)
}
@@ -949,7 +949,7 @@ func TestCorrelatedGuestMemoryNextPoll(t *testing.T) {
mon := &Monitor{state: models.NewState(), rateTracker: NewRateTracker(), config: &config.Config{}, resourceStore: unifiedresources.NewMonitorAdapter(registry)}
prev := mon.previousGuestContextForInstance("pve-a")
raw := VMMemoryRaw{}
_, got, source := mon.resolveGuestStatusMemory(context.Background(), &stubPVEClient{}, "pve-a", "firewall", "node1", 111, guestID, &proxmox.VMStatus{MaxMem: 8000, Mem: 8100}, prev.hostAgentsByVMID, 8000, "", &raw)
_, got, source, _ := mon.resolveGuestStatusMemory(context.Background(), &stubPVEClient{}, "pve-a", "firewall", "node1", 111, guestID, &proxmox.VMStatus{MaxMem: 8000, Mem: 8100}, prev.hostAgentsByVMID, 8000, "", &raw)
if got != tc.want {
t.Fatalf("used=%d source=%s want=%d", got, source, tc.want)
}
@@ -1001,7 +1001,7 @@ func TestCorrelatedGuestMemoryNextPoll(t *testing.T) {
t.Fatal("correlated agent crossed instance boundary")
}
otherID := makeGuestID("pve-b", "node1", 111)
_, otherUsed, otherSource := mon.resolveGuestStatusMemory(context.Background(), &stubPVEClient{}, "pve-b", "other-firewall", "node1", 111, otherID, &proxmox.VMStatus{MaxMem: 8000, Mem: 8100}, other.hostAgentsByVMID, 8000, "", &VMMemoryRaw{})
_, otherUsed, otherSource, _ := mon.resolveGuestStatusMemory(context.Background(), &stubPVEClient{}, "pve-b", "other-firewall", "node1", 111, otherID, &proxmox.VMStatus{MaxMem: 8000, Mem: 8100}, other.hostAgentsByVMID, 8000, "", &VMMemoryRaw{})
if otherUsed != 8000 || otherSource != "status-mem" {
t.Fatalf("other instance inherited agent memory: used=%d source=%s", otherUsed, otherSource)
}
@@ -1087,7 +1087,7 @@ func TestAutomaticGuestMemoryLinkNextPoll(t *testing.T) {
{"pve-a", guestID, 2800, "agent"}, {"pve-b", otherID, 8000, "status-mem"},
} {
prev := mon.previousGuestContextForInstance(tc.instance)
_, used, source := mon.resolveGuestStatusMemory(context.Background(), &stubPVEClient{}, tc.instance, "firewall", "node1", 111, tc.id, &proxmox.VMStatus{MaxMem: 8000, Mem: 8100}, prev.hostAgentsByVMID, 8000, "", &VMMemoryRaw{})
_, used, source, _ := mon.resolveGuestStatusMemory(context.Background(), &stubPVEClient{}, tc.instance, "firewall", "node1", 111, tc.id, &proxmox.VMStatus{MaxMem: 8000, Mem: 8100}, prev.hostAgentsByVMID, 8000, "", &VMMemoryRaw{})
if used != tc.used || source != tc.source {
t.Fatalf("%s used=%d source=%s", tc.instance, used, source)
}
@@ -66,60 +66,11 @@ func (m *Monitor) applyVMStatusDetails(
res.Lock = status.Lock
state.detailedStatus = status
state.guestAgentStatus, state.guestAgentExpected = vmGuestAgentRuntimeState(status, recentGuestAgentEvidence)
state.memTotal, state.memUsed, state.memorySource = m.resolveGuestStatusMemory(
ctx,
client,
instanceName,
res.Name,
res.Node,
res.VMID,
guestID,
status,
vmIDToHostAgent,
state.memTotal,
state.memorySource,
&state.guestRaw,
)
mergeVMRuntimeCounters(state, status)
// Gather guest metadata from the agent when available
guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion := m.fetchGuestAgentMetadata(ctx, client, instanceName, res.Node, res.Name, res.VMID, status, false)
if len(guestIPs) > 0 {
state.ipAddresses = guestIPs
}
if len(guestIfaces) > 0 {
state.networkInterfaces = guestIfaces
}
if guestOSName != "" {
state.osName = guestOSName
}
if guestOSVersion != "" {
state.osVersion = guestOSVersion
}
if guestAgentVersion != "" {
state.agentVersion = guestAgentVersion
}
// Always try to get filesystem info if agent is enabled
// Prefer guest agent data over cluster/resources data for accuracy
// Filesystem usage is cross-platform. Read it before optional Linux memory
// and metadata commands can start a shared uncertainty pause. Every command
// still verifies the same operation lock and is admitted only once.
if status.Agent.IsAvailable() {
var fsDisks []models.Disk
state.diskTotal, state.diskUsed, state.diskFree, state.diskUsage, fsDisks, state.diskFromAgent, state.diskStatusReason = m.updateVMDisksFromGuestAgentFSInfo(
ctx,
instanceName,
res,
client,
state.diskTotal,
state.diskUsed,
state.diskUsage,
)
if len(fsDisks) > 0 {
state.individualDisks = fsDisks
}
if guestAgentDiskDeferred(state.diskStatusReason) {
state.guestAgentStatus = "deferred"
}
m.applyVMGuestAgentFSInfo(ctx, instanceName, res, client, state)
} else {
// Agent disabled - show allocated disk size
if state.diskTotal > 0 {
@@ -139,6 +90,61 @@ func (m *Monitor) applyVMStatusDetails(
Msg("VM guest agent is not currently queryable")
}
var memoryDeferred bool
state.memTotal, state.memUsed, state.memorySource, memoryDeferred = m.resolveGuestStatusMemory(
ctx,
client,
instanceName,
res.Name,
res.Node,
res.VMID,
guestID,
status,
vmIDToHostAgent,
state.memTotal,
state.memorySource,
&state.guestRaw,
)
if memoryDeferred {
state.guestAgentStatus = "deferred"
}
mergeVMRuntimeCounters(state, status)
// Gather guest metadata from the agent when available
guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion, metadataDeferred := m.fetchGuestAgentMetadata(ctx, client, instanceName, res.Node, res.Name, res.VMID, status, false)
if metadataDeferred {
state.guestAgentStatus = "deferred"
}
if len(guestIPs) > 0 {
state.ipAddresses = guestIPs
}
if len(guestIfaces) > 0 {
state.networkInterfaces = guestIfaces
}
if guestOSName != "" {
state.osName = guestOSName
}
if guestOSVersion != "" {
state.osVersion = guestOSVersion
}
if guestAgentVersion != "" {
state.agentVersion = guestAgentVersion
}
}
func (m *Monitor) applyVMGuestAgentFSInfo(ctx context.Context, instanceName string, res proxmox.ClusterResource, client PVEClientInterface, state *vmBuildState) {
var fsDisks []models.Disk
state.diskTotal, state.diskUsed, state.diskFree, state.diskUsage, fsDisks, state.diskFromAgent, state.diskStatusReason = m.updateVMDisksFromGuestAgentFSInfo(
ctx, instanceName, res, client, state.diskTotal, state.diskUsed, state.diskUsage,
)
if len(fsDisks) > 0 {
state.individualDisks = fsDisks
}
if guestAgentDiskDeferred(state.diskStatusReason) {
state.guestAgentStatus = "deferred"
}
}
func mergeVMRuntimeCounters(state *vmBuildState, status *proxmox.VMStatus) {
@@ -284,7 +290,8 @@ func (m *Monitor) buildVMFromClusterResource(
guestAgentAvailable := shouldQueryGuestAgent(state.detailedStatus, prevVM, now) ||
m.hasRecentGuestMetadataEvidence(instanceName, res.Node, res.VMID, now)
if guestAgentAvailable && res.Lock == "" && state.detailedStatus == nil {
guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion := m.fetchGuestAgentMetadata(
m.applyVMGuestAgentFSInfo(ctx, instanceName, res, client, &state)
guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion, metadataDeferred := m.fetchGuestAgentMetadata(
ctx,
client,
instanceName,
@@ -294,6 +301,9 @@ func (m *Monitor) buildVMFromClusterResource(
nil,
true,
)
if metadataDeferred {
state.guestAgentStatus = "deferred"
}
if len(guestIPs) > 0 {
state.ipAddresses = guestIPs
}
@@ -310,22 +320,6 @@ func (m *Monitor) buildVMFromClusterResource(
state.agentVersion = guestAgentVersion
}
var fsDisks []models.Disk
state.diskTotal, state.diskUsed, state.diskFree, state.diskUsage, fsDisks, state.diskFromAgent, state.diskStatusReason = m.updateVMDisksFromGuestAgentFSInfo(
ctx,
instanceName,
res,
client,
state.diskTotal,
state.diskUsed,
state.diskUsage,
)
if len(fsDisks) > 0 {
state.individualDisks = fsDisks
}
if guestAgentDiskDeferred(state.diskStatusReason) {
state.guestAgentStatus = "deferred"
}
state.guestAgentExpected = true
if state.guestAgentStatus != "deferred" && (len(guestIPs) > 0 || len(guestIfaces) > 0 || guestOSName != "" || guestOSVersion != "" || guestAgentVersion != "" || state.diskFromAgent) {
state.guestAgentStatus = "available"