Keep guest-memory provenance bound to its original observation

Expose current, last-known and unavailable memory through raw guest and selected unified metrics without renewing QGA evidence during repeated backup deferrals. Preserve independent PVE/linked-agent source times, numeric selection, History and alert policy. Annotate selected metrics as well as raw facets so cross-source fallback cannot inherit another source freshness. Add connected observation, source ownership, clone and wire controls and update the shared subsystem contracts.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-04 17:00:24 +01:00
parent c50b8340fd
commit 43ccc6c39a
21 changed files with 434 additions and 31 deletions
@@ -15,6 +15,18 @@
## Purpose
### Memory annotation is not agent admission
The shared Memory model's optional guest `observation` records selected metric
state/source/original known time. Missing annotation is legacy/unknown. It
changes neither numeric validity nor enrollment, identity, removal, token scope
or guest-command authority. Canonical agent metric annotations use server-owned
report/source receipt time rather than a client-supplied annotation, and a
PVE-refreshed guest row cannot renew the correlated agent observation.
`TestGuestMemoryObservationWireContract` pins omission, round-trip preservation
and unchanged numeric validity; linked-agent/source-selection controls pin
receipt ownership and offline exclusion. No native liveness/thaw is asserted.
### TrueNAS probes are not fleet or inventory liveness
A separate settings connection probe does not establish runtime provider
@@ -17,6 +17,28 @@
## Purpose
### Guest memory observations are source-owned
QEMU and LXC memory carry an optional `observation` with `state` (`current`,
`last-known`, `unavailable`), canonical `source`, and original RFC3339
`observedAt` when known. Normal TTL cache hits keep the original read time.
Repeated QGA/backup deferrals are last-known, never renewed by a poll or
diagnostic RetrievedAt; expired/missing usage and powered-off guests expose
unavailability without inventing a read time. Legacy unannotated values make
no freshness assertion. Independently selected PVE status/listing or linked
Pulse-agent memory remains distinct from disk or QGA collection state. Agent
receipt time belongs to its source status, not a guest row refreshed by PVE.
Annotation preserves numeric selection, capacity, retention, guest identity,
agent admission, alert thresholds and existing History recording. It grants
no live-check authority or proof of thaw. Existing legacy state, raw Proxmox
facets and selected canonical memory metrics preserve the discriminator;
selected metric provenance must not be borrowed from another merged facet.
`TestGuestMemoryObservationContract` exercises poll/cache/deferral/expiry/
resumption through unified and served conversions; wire and source-selection
controls cover omission, cloning and independent live agent data. These are
synthetic source proofs, not native QGA/HAOS acceptance or delivered relief.
### Unavailable backup-age evaluation is not silent recovery
PVE, PBS and mock backup checks share one failure-preserving evaluator. The
@@ -21,6 +21,17 @@
## Purpose
### Memory provenance does not imply recoverability
Canonical metric values may carry optional selected guest memory observation
state/source/original known time. The facet follows that numeric source through
cloning and frontend conversion; last-known or unavailable memory is not fresh
backup evidence, a storage-health verdict or confirmation of thaw. This
annotation changes no backup identity, retention, rollup selection, restore
policy or snapshot data. The connected monitoring observation and metric
source-selection controls verify the shared type without weakening recovery
evidence requirements.
### Latest provider observations do not scan retained evidence payloads
The protection-posture reader enumerates provider/scope keys through the existing
@@ -23,6 +23,23 @@ and sort the complete canonical change table while startup and ingestion wait.
## Purpose
### Guest memory provenance follows metric selection
Raw Proxmox `memory.observation` preserves the guest builder's state, source
and original known observation time. Canonical memory metrics carry their own
optional `observation` through clones, merges and served frontend conversion.
A retained platform facet cannot supply provenance for a different selected
agent metric. Agent memory annotations are authored from server-owned source
receipt/freshness, never agent-supplied annotation or merged guest LastSeen.
Linked-agent views retain that source time after host/guest correlation.
Absent annotations are unknown/legacy; absent observation times stay absent.
Numeric preference, unavailable same-source withdrawal, cross-source fallback,
identity and freshness policy remain unchanged. Guest memory and disk-read
provenance are independent. Adapter/source-selection and linked-view controls
exercise these boundaries, with the connected monitoring contract pinning the
actual served projection; this does not prove native recovery or publication.
### PBS capacity consumers preserve explicit read failure
`ResourcePBSDatastore.error` describes existing provider read-failure evidence,
+10 -8
View File
@@ -1217,10 +1217,11 @@ type ResourceConvertInput struct {
// ResourceMetricInput represents a metric value for resource conversion.
type ResourceMetricInput struct {
Current float64
Total *int64
Used *int64
Free *int64
Current float64
Total *int64
Used *int64
Free *int64
Observation MemoryObservation
}
type ResourceAlertInput struct {
@@ -1321,10 +1322,11 @@ func ConvertResourceToFrontend(input ResourceConvertInput) ResourceFrontend {
if input.Memory != nil {
rf.Memory = &ResourceMetricFrontend{
Current: input.Memory.Current,
Total: input.Memory.Total,
Used: input.Memory.Used,
Free: input.Memory.Free,
Current: input.Memory.Current,
Total: input.Memory.Total,
Used: input.Memory.Used,
Free: input.Memory.Free,
Observation: input.Memory.Observation,
}
}
+37
View File
@@ -7,6 +7,43 @@ import (
"time"
)
func TestGuestMemoryObservationWireContract(t *testing.T) {
legacy := Memory{Total: 100, Used: 25, Free: 75, Usage: 25}
payload, err := json.Marshal(legacy)
if err != nil || strings.Contains(string(payload), "observation") {
t.Fatalf("legacy memory unexpectedly asserts freshness: %s / %v", payload, err)
}
at := time.Date(2026, time.October, 4, 12, 0, 0, 123, time.UTC)
for _, state := range []string{"current", "last-known", "unavailable"} {
t.Run(state, func(t *testing.T) {
memory := legacy
memory.Observation = MemoryObservation{State: state, Source: "guest-agent-meminfo", ObservedAt: at}
if state == "unavailable" {
memory.Observation.ObservedAt = time.Time{}
}
payload, err := json.Marshal(memory)
if err != nil {
t.Fatal(err)
}
var roundtrip Memory
if err := json.Unmarshal(payload, &roundtrip); err != nil || roundtrip != memory {
t.Fatalf("memory provenance changed in wire roundtrip: %s / %v", payload, err)
}
if strings.Contains(string(payload), "observedAt") != (state != "unavailable") {
t.Fatalf("unknown time was manufactured: %s", payload)
}
if memory.HasKnownUsage() != legacy.HasKnownUsage() {
t.Fatal("annotation changed numeric validity / threshold policy")
}
metric := ResourceMetricInput{Current: 25, Observation: memory.Observation}
front := ConvertResourceToFrontend(ResourceConvertInput{Memory: &metric, CPU: &metric, Disk: &metric})
if front.Memory == nil || front.Memory.Observation != memory.Observation || front.CPU.Observation.State != "" || front.Disk.Observation.State != "" {
t.Fatal("memory projection lost evidence or attributed it to another metric")
}
})
}
}
func TestMetricPoint_ZeroValue(t *testing.T) {
var point MetricPoint
+17 -5
View File
@@ -3085,6 +3085,17 @@ type PMGQueueStatus struct {
UpdatedAt time.Time `json:"updatedAt"` // When this queue data was collected
}
// MemoryObservation describes the provenance of a selected guest reading, not
// guest-agent liveness or permission to run a diagnostic. Current may include a
// normal TTL cache hit; ObservedAt remains the original source observation.
// Last-known values are retained evidence, not a new measurement. A missing
// observation is legacy/unknown, never an assertion of current freshness.
type MemoryObservation struct {
State string `json:"state"` // current, last-known, unavailable
Source string `json:"source"`
ObservedAt time.Time `json:"observedAt,omitzero"`
}
// Memory represents memory usage
type Memory struct {
Total int64 `json:"total"`
@@ -3095,11 +3106,12 @@ type Memory struct {
// UsageUnavailable is true when total capacity is known but no cache-aware
// source can establish current usage. Consumers must not interpret the
// zero-valued Used, Free, or Usage fields as a measurement in that state.
UsageUnavailable bool `json:"usageUnavailable,omitempty"`
Usage float64 `json:"usage"`
Balloon int64 `json:"balloon,omitempty"`
SwapUsed int64 `json:"swapUsed,omitempty"`
SwapTotal int64 `json:"swapTotal,omitempty"`
UsageUnavailable bool `json:"usageUnavailable,omitempty"`
Usage float64 `json:"usage"`
Balloon int64 `json:"balloon,omitempty"`
SwapUsed int64 `json:"swapUsed,omitempty"`
SwapTotal int64 `json:"swapTotal,omitempty"`
Observation MemoryObservation `json:"observation,omitzero"`
}
// HasKnownUsage reports whether this memory value is safe for live
+5 -4
View File
@@ -1140,10 +1140,11 @@ func (r ResourceFrontend) NormalizeCollections() ResourceFrontend {
// ResourceMetricFrontend represents a metric value for the frontend.
type ResourceMetricFrontend struct {
Current float64 `json:"current"`
Total *int64 `json:"total,omitempty"`
Used *int64 `json:"used,omitempty"`
Free *int64 `json:"free,omitempty"`
Current float64 `json:"current"`
Total *int64 `json:"total,omitempty"`
Used *int64 `json:"used,omitempty"`
Free *int64 `json:"free,omitempty"`
Observation MemoryObservation `json:"observation,omitzero"`
}
// ResourceNetworkFrontend represents network metrics for the frontend.
@@ -2981,7 +2981,9 @@ func TestGuestAgentBackupMonitoringContract(t *testing.T) {
}
func TestGuestMemoryObservationContract(t *testing.T) {
testGuestMemoryObservationLifecycle(t)
t.Run("poll-to-served-observation", testGuestMemoryObservationLifecycle)
t.Run("identity-and-origin-boundaries", testGuestMemoryObservationKeepsOriginsSeparate)
t.Run("independent-linked-agent", testGuestMemoryObservationPreservesIndependentLinkedAgent)
}
// Same-target, unambiguous lock evidence is required even with warm caches.
@@ -155,7 +155,7 @@ func testGuestAgentBackupMonitoringLifecycleWithUnverifiedConfig(t *testing.T, u
if locked.ID != initial.ID || locked.GuestAgentStatus != "deferred" || locked.DiskStatusReason != "prev-vm-locked" {
t.Fatalf("locked poll %d lost continuity/truth: %#v", poll, locked)
}
if locked.Memory != initial.Memory || source != "previous-snapshot" || locked.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(locked.NetworkInterfaces, initial.NetworkInterfaces) {
if guestMemoryValuesOnly(locked.Memory) != guestMemoryValuesOnly(initial.Memory) || source != "previous-snapshot" || locked.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(locked.NetworkInterfaces, initial.NetworkInterfaces) {
t.Fatalf("locked poll %d lost last-known evidence: source=%s memory=%#v want=%#v", poll, source, locked.Memory, initial.Memory)
}
}
@@ -176,7 +176,7 @@ func testGuestAgentBackupMonitoringLifecycleWithUnverifiedConfig(t *testing.T, u
t.Fatal(err)
}
locked, _ = build()
if calls.Load() != before || locked.GuestAgentStatus != "deferred" || locked.AgentVersion != initial.AgentVersion || locked.Memory != initial.Memory {
if calls.Load() != before || locked.GuestAgentStatus != "deferred" || locked.AgentVersion != initial.AgentVersion || guestMemoryValuesOnly(locked.Memory) != guestMemoryValuesOnly(initial.Memory) {
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.
@@ -240,3 +240,9 @@ func TestGuestAgentDeferralPreservesMetadataAndMemoryCache(t *testing.T) {
t.Fatal("deferral queued work, renewed cache or poisoned supported-OS evidence")
}
}
// Provenance is asserted separately; numeric continuity includes every old field.
func guestMemoryValuesOnly(memory models.Memory) models.Memory {
memory.Observation = models.MemoryObservation{}
return memory
}
@@ -0,0 +1,97 @@
package monitoring
import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
type guestMemoryObservationContext struct {
source string
providerObservedAt time.Time
deferred bool
previous *GuestMemorySnapshot
linkedAgent models.Host
}
// guestMemoryObservation annotates the value already selected by the builder.
// It does not select different numeric evidence, change alert/history policy,
// renew a cache, or run a guest command. In particular disk state is not memory
// provenance: PVE or a linked Pulse agent may still supply an independent read.
func (m *Monitor) guestMemoryObservation(instance, guestType, node string, vmid int, memory models.Memory, evidence guestMemoryObservationContext, now time.Time) models.MemoryObservation {
source := CanonicalMemorySource(evidence.source)
if !memory.HasKnownUsage() || source == "unavailable" || source == "powered-off" {
if source != "powered-off" {
source = "unavailable"
}
return models.MemoryObservation{State: "unavailable", Source: source}
}
observation := models.MemoryObservation{State: "current", Source: source}
switch source {
case "guest-agent-meminfo", "guest-agent-meminfo-derived", "previous-snapshot":
// The cache, rather than the repeatedly refreshed diagnostic/poll
// timestamp, owns the age of retained QGA evidence.
previous := evidence.previous
sameGuest := previous != nil && previous.Instance == instance && previous.GuestType == guestType && previous.Node == node && previous.VMID == vmid
qgaOrigin := source != "previous-snapshot" || (sameGuest &&
(previous.Memory.Observation.Source == "guest-agent-meminfo" || previous.Memory.Observation.Source == "guest-agent-meminfo-derived" ||
(previous.Memory.Observation.State == "" && (CanonicalMemorySource(previous.MemorySource) == "guest-agent-meminfo" || CanonicalMemorySource(previous.MemorySource) == "guest-agent-meminfo-derived"))))
if guestType == "qemu" && qgaOrigin {
if original, ok := m.cachedGuestMemoryObservation(instance, node, vmid, memory, now); ok {
observation = original
}
}
if source == "previous-snapshot" {
observation.State = "last-known"
if observation.ObservedAt.IsZero() && sameGuest && previous.Memory.Total == memory.Total && previous.Memory.Used == memory.Used {
// An annotated prior source carries its original observation.
// A legacy previous-snapshot's RetrievedAt cannot supply it.
if previous.Memory.Observation.State == "current" || previous.Memory.Observation.State == "last-known" {
observation = previous.Memory.Observation
observation.State = "last-known"
}
}
} else if evidence.deferred || (!observation.ObservedAt.IsZero() && now.Sub(observation.ObservedAt) >= vmAgentMemCacheTTL) {
observation.State = "last-known"
}
case "agent":
observation.ObservedAt = evidence.linkedAgent.LastSeen
if original := evidence.linkedAgent.Memory.Observation; original.State != "" {
observation = original
}
case "available-field", "derived-free-buffers-cached", "derived-total-minus-used", "status-mem", "status-freemem", "cluster-resources":
observation.ObservedAt = evidence.providerObservedAt
default:
// Unknown provenance cannot be upgraded to a current observation by
// a new poll. Keep the selected numbers and expose the uncertainty.
observation.State = "last-known"
}
if observation.ObservedAt.IsZero() || observation.ObservedAt.After(now) {
observation.ObservedAt = time.Time{}
observation.State = "last-known"
}
return observation
}
func (m *Monitor) cachedGuestMemoryObservation(instance, node string, vmid int, memory models.Memory, now time.Time) (models.MemoryObservation, bool) {
m.rrdCacheMu.RLock()
entry, ok := m.vmAgentMemCache[guestMemoryCacheKey(instance, node, vmid)]
m.rrdCacheMu.RUnlock()
if !ok || entry.negative || entry.fetchedAt.IsZero() || entry.fetchedAt.After(now) || now.Sub(entry.fetchedAt) > vmAgentMemCleanupMaxAge || memory.Total <= 0 || entry.info.EffectiveAvailable > uint64(memory.Total) || memory.Used != memory.Total-int64(entry.info.EffectiveAvailable) {
return models.MemoryObservation{}, false
}
expected := models.Memory{Free: int64(entry.info.EffectiveAvailable)}
splitReclaimableMemory(&expected, entry.info.Free)
if memory.Free != expected.Free || memory.Cache != expected.Cache {
return models.MemoryObservation{}, false
}
source := "guest-agent-meminfo"
switch entry.info.Source {
case "meminfo-available":
case "meminfo-derived":
source = "guest-agent-meminfo-derived"
default:
return models.MemoryObservation{}, false
}
return models.MemoryObservation{State: "current", Source: source, ObservedAt: entry.fetchedAt}, true
}
@@ -46,7 +46,8 @@ func testGuestMemoryObservationLifecycle(t *testing.T) {
const gib = uint64(1024 * 1024 * 1024)
identity := makeGuestID("provenance", "node", 105)
registry := unifiedresources.NewRegistry(nil)
m := &Monitor{guestMetadataLimiter: make(map[string]time.Time), rateTracker: NewRateTracker(), metricsHistory: NewMetricsHistory(32, time.Hour), resourceStore: unifiedresources.NewMonitorAdapter(registry)}
store := unifiedresources.NewMonitorAdapter(registry)
m := &Monitor{guestMetadataLimiter: make(map[string]time.Time), rateTracker: NewRateTracker(), metricsHistory: NewMetricsHistory(32, time.Hour), resourceStore: store}
client := &vmMemoryTrustStubClient{stubPVEClient: &stubPVEClient{}, vmAgentMemAvailable: 5 * gib,
vmStatus: &proxmox.VMStatus{Status: "running", MaxMem: 8 * gib, Mem: 8 * gib, Agent: proxmox.VMAgentField{Value: 1}}}
res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "guest", VMID: 105, Status: "running", MaxMem: 8 * gib, Mem: 8 * gib}
@@ -59,8 +60,8 @@ func testGuestMemoryObservationLifecycle(t *testing.T) {
}
m.recordGuestSnapshot("provenance", "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}})
views := registry.VMs()
store.PopulateFromSnapshot(models.StateSnapshot{VMs: []models.VM{vm}})
views := store.VMs()
if len(views) != 1 {
t.Fatal("guest did not reach unified view")
}
@@ -84,6 +85,7 @@ func testGuestMemoryObservationLifecycle(t *testing.T) {
t.Fatal(err)
}
var wire []struct {
Memory *models.ResourceMetricFrontend `json:"memory"`
Proxmox struct {
Memory models.Memory `json:"memory"`
} `json:"proxmox"`
@@ -92,6 +94,14 @@ func testGuestMemoryObservationLifecycle(t *testing.T) {
t.Fatalf("broadcast guest memory unavailable: %s / %v", data, err)
}
assertGuestMemoryObservation(t, wire[0].Proxmox.Memory, state, source, at)
if vm.Memory.HasKnownUsage() {
if wire[0].Memory == nil {
t.Fatal("selected memory metric disappeared")
}
assertGuestMemoryObservation(t, models.Memory{Observation: wire[0].Memory.Observation}, state, source, at)
} else if wire[0].Memory != nil {
t.Fatal("unavailable memory became a numeric metric")
}
if wire[0].Proxmox.Memory.Used != vm.Memory.Used || wire[0].Proxmox.Memory.UsageUnavailable != vm.Memory.UsageUnavailable {
t.Fatal("provenance changed numeric/unknown memory projection")
}
@@ -153,3 +163,75 @@ func testGuestMemoryObservationLifecycle(t *testing.T) {
t.Fatal("powered-off numeric compatibility changed")
}
}
func testGuestMemoryObservationKeepsOriginsSeparate(t *testing.T) {
now := time.Now()
at := now.Add(-2 * time.Minute)
observed := models.Memory{Total: 100, Used: 25, Free: 75, Usage: 25}
previous := GuestMemorySnapshot{Instance: "site", GuestType: "qemu", Node: "node", VMID: 105, Status: "running", RetrievedAt: now,
MemorySource: "previous-snapshot", Memory: observed}
previous.Memory.Observation = models.MemoryObservation{State: "last-known", Source: "available-field", ObservedAt: at}
m := &Monitor{vmAgentMemCache: map[string]agentMemCacheEntry{guestMemoryCacheKey("site", "node", 105): {
fetchedAt: now.Add(-time.Second), info: proxmox.LinuxMemoryAvailability{Source: "meminfo-available", EffectiveAvailable: 75}}}}
for _, tc := range []struct {
name string
change func(*GuestMemorySnapshot)
source string
at time.Time
}{
{name: "identical numeric QGA cache cannot replace PVE provenance", source: "available-field", at: at},
{name: "poll timestamp does not renew legacy retained evidence", source: "previous-snapshot", change: func(p *GuestMemorySnapshot) { p.Memory.Observation = models.MemoryObservation{} }},
{name: "foreign guest snapshot cannot lend evidence", source: "previous-snapshot", change: func(p *GuestMemorySnapshot) { p.VMID++ }},
{name: "foreign site snapshot cannot lend evidence", source: "previous-snapshot", change: func(p *GuestMemorySnapshot) { p.Instance = "other" }},
{name: "future original time remains unknown", source: "available-field", change: func(p *GuestMemorySnapshot) { p.Memory.Observation.ObservedAt = now.Add(time.Second) }},
} {
t.Run(tc.name, func(t *testing.T) {
prior := previous
if tc.change != nil {
tc.change(&prior)
}
memory := observed
memory.Observation = m.guestMemoryObservation("site", "qemu", "node", 105, memory, guestMemoryObservationContext{source: "previous-snapshot", previous: &prior}, now)
assertGuestMemoryObservation(t, memory, "last-known", tc.source, tc.at)
})
}
}
func testGuestMemoryObservationPreservesIndependentLinkedAgent(t *testing.T) {
at := time.Now().Add(-30 * time.Second)
resource := &unifiedresources.Resource{Type: unifiedresources.ResourceTypeVM, LastSeen: time.Now(),
Agent: &unifiedresources.AgentData{Memory: &unifiedresources.AgentMemoryMeta{Total: 100, Used: 40, Free: 60}},
SourceStatus: map[unifiedresources.DataSource]unifiedresources.SourceStatus{unifiedresources.SourceAgent: {Status: "online", LastSeen: at}}}
view := unifiedresources.NewVMView(resource)
agentMemory, ok := view.LinkedAgentMemory()
if !ok {
t.Fatal("source-owned linked memory missing")
}
for _, guestType := range []string{"qemu", "lxc"} {
t.Run(guestType, func(t *testing.T) {
m := &Monitor{rateTracker: NewRateTracker(), guestMetadataLimiter: make(map[string]time.Time)}
client := &vmMemoryTrustStubClient{stubPVEClient: &stubPVEClient{}, vmStatus: &proxmox.VMStatus{Status: "running", Lock: "backup", MaxMem: 100, Mem: 100}}
res := proxmox.ClusterResource{Type: guestType, Node: "node", VMID: 105, Status: "running", Lock: "backup", MaxMem: 100, Mem: 100, ObservedAt: time.Now()}
identity := makeGuestID("site", "node", 105)
linked := map[string]models.Host{identity: {Memory: agentMemory, Status: "online"}}
var memory models.Memory
if guestType == "qemu" {
vm, _, source, _, _, ok := m.buildVMFromClusterResource(context.Background(), "site", res, client, identity, linked, nil)
if !ok || source != "agent" {
t.Fatalf("linked VM agent was not selected: %s", source)
}
memory = vm.Memory
} else {
ct, _, source, _, ok := m.buildContainerFromClusterResource(context.Background(), "site", res, client, nil, linked)
if !ok || source != "agent" {
t.Fatalf("linked LXC agent was not selected: %s", source)
}
memory = ct.Memory
}
assertGuestMemoryObservation(t, memory, "current", "agent", at)
if memory.Used != 40 || client.vmAgentMemCalls != 0 {
t.Fatal("independent memory changed or caused a guest-agent read")
}
})
}
}
+1 -1
View File
@@ -6985,7 +6985,7 @@ func monitorMetricInput(metric *unifiedresources.MetricValue) *models.ResourceMe
current = math.Max(metric.Percent, metric.Value)
}
result := &models.ResourceMetricInput{Current: current}
result := &models.ResourceMetricInput{Current: current, Observation: metric.Observation}
if metric.Total != nil {
total := *metric.Total
result.Total = &total
@@ -468,6 +468,15 @@ func (m *Monitor) buildVMFromClusterResource(
snapshotNotes = append(snapshotNotes, "preserved-original-guest-memory-during-deferral")
}
}
providerObservedAt := observedAtOr(res.ObservedAt, sampleTime)
if state.detailedStatus != nil && CanonicalMemorySource(state.memorySource) != "cluster-resources" {
providerObservedAt = observedAtOr(state.detailedStatus.ObservedAt, sampleTime)
}
memory.Observation = m.guestMemoryObservation(instanceName, "qemu", res.Node, res.VMID, memory, guestMemoryObservationContext{
source: state.memorySource, providerObservedAt: providerObservedAt,
deferred: state.guestAgentStatus == "deferred", previous: prevSnapshot,
linkedAgent: vmIDToHostAgent[guestID],
}, sampleTime)
vm := models.VM{
ID: guestID,
+6 -1
View File
@@ -99,8 +99,9 @@ func (m *Monitor) buildContainerFromClusterResource(
agentHost,
hasAgent,
)
previousMemory := m.previousGuestSnapshot(instanceName, "lxc", res.Node, res.VMID)
memUsed, memorySource, _ = stabilizeGuestLowTrustMemory(
m.previousGuestSnapshot(instanceName, "lxc", res.Node, res.VMID),
previousMemory,
res.Status,
memorySource,
memTotal,
@@ -129,6 +130,10 @@ func (m *Monitor) buildContainerFromClusterResource(
Usage: safePercentage(float64(clampedMemUsed), float64(memTotal)),
}
}
memory.Observation = m.guestMemoryObservation(instanceName, "lxc", res.Node, res.VMID, memory, guestMemoryObservationContext{
source: memorySource, providerObservedAt: observedAtOr(res.ObservedAt, sampleTime),
previous: previousMemory, linkedAgent: agentHost,
}, time.Now())
diskUsed := res.Disk
if diskUsed > res.MaxDisk && res.MaxDisk > 0 {
diskUsed = res.MaxDisk
+8 -2
View File
@@ -512,13 +512,19 @@ func TestCloneProxmoxData_NetworkInterfaceIsolation(t *testing.T) {
}
func TestCloneProxmoxDataMemoryIsolation(t *testing.T) {
observation := models.MemoryObservation{State: "last-known", Source: "guest-agent-meminfo", ObservedAt: time.Now().Add(-time.Minute)}
original := &ProxmoxData{
Memory: &models.Memory{Total: 8 << 30, Used: 4 << 30, Usage: 50},
Memory: &models.Memory{Total: 8 << 30, Used: 4 << 30, Usage: 50, Observation: observation},
}
cloned := cloneProxmoxData(original)
if cloned.Memory.Observation != observation {
t.Fatal("clone lost original memory observation")
}
cloned.Memory.Used = 7 << 30
if original.Memory.Used != 4<<30 {
cloned.Memory.Observation.State = "current"
cloned.Memory.Observation.ObservedAt = time.Now()
if original.Memory.Used != 4<<30 || original.Memory.Observation != observation {
t.Error("mutating cloned memory should not affect original")
}
}
+14 -3
View File
@@ -5,6 +5,7 @@ import (
"math"
"math/rand"
"strings"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/mockmode"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
@@ -35,9 +36,19 @@ func hasObservedDiskUsage(disk models.Disk) bool {
}
func metricsFromHost(host models.Host) *ResourceMetrics {
// Author freshness from server-owned report receipt time, never from a
// guest row refreshed by another source or an agent-supplied annotation.
memory := host.Memory
memory.Observation = models.MemoryObservation{State: "current", Source: "agent", ObservedAt: host.LastSeen}
if host.Status != "online" || host.LastSeen.IsZero() || host.LastSeen.After(time.Now()) {
memory.Observation.State = "last-known"
if host.LastSeen.After(time.Now()) {
memory.Observation.ObservedAt = time.Time{}
}
}
return buildHostMetricPayload(
host.CPUUsage,
host.Memory,
memory,
host.Disks,
host.NetInRate,
host.NetOutRate,
@@ -99,7 +110,7 @@ func buildHostMetricPayload(
metrics.CPU = &MetricValue{Value: cpuPercent, Percent: cpuPercent, Unit: "percent", Source: source}
if memory.Total > 0 && memory.HasKnownUsage() {
percent := percentFromReportedPercent(memory.Usage)
metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source}
metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source, Observation: memory.Observation}
}
if len(disks) > 0 {
disk := disks[0]
@@ -199,7 +210,7 @@ func buildVMMetricPayload(
metrics.CPU = &MetricValue{Value: cpuPercent, Percent: cpuPercent, Unit: "percent", Source: source}
if memory.Total > 0 && memory.HasKnownUsage() {
percent := percentFromUsage(memory.Usage)
metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source}
metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source, Observation: memory.Observation}
}
if hasObservedDiskUsage(disk) {
percent := percentFromUsage(disk.Usage)
+35
View File
@@ -544,3 +544,38 @@ func TestCanonicalDiskUsagePreservesUnavailableAndObservedZero(t *testing.T) {
}
}
}
func TestGuestMemoryMetricObservationFollowsSelectedSource(t *testing.T) {
at := time.Now().Add(-time.Minute)
retained := models.Memory{Total: 100, Used: 25, Free: 75, Usage: 25,
Observation: models.MemoryObservation{State: "last-known", Source: "guest-agent-meminfo", ObservedAt: at}}
vm, _ := resourceFromVM(models.VM{VMID: 105, Memory: retained})
ct, _ := resourceFromContainer(models.Container{VMID: 106, Memory: retained})
for _, resource := range []Resource{vm, ct} {
if resource.Metrics.Memory.Observation != retained.Observation || resource.Proxmox.Memory.Observation != retained.Observation {
t.Fatal("guest adapter lost selected/raw observation")
}
copy := cloneResource(&resource)
copy.Metrics.Memory.Observation.State = "current"
copy.Proxmox.Memory.Observation.Source = "changed"
if resource.Metrics.Memory.Observation != retained.Observation || resource.Proxmox.Memory.Observation != retained.Observation {
t.Fatal("clone aliases memory provenance")
}
}
// Missing platform usage must not mark a surviving live agent metric
// unavailable just because the retained platform facet says so.
unavailable := models.UnavailableMemory(100)
unavailable.Observation = models.MemoryObservation{State: "unavailable", Source: "unavailable"}
platform, _ := resourceFromVM(models.VM{VMID: 105, Memory: unavailable})
agent := metricsFromHost(models.Host{Status: "online", LastSeen: at,
Memory: models.Memory{Total: 100, Used: 40, Free: 60, Usage: 40,
Observation: models.MemoryObservation{State: "current", Source: "untrusted", ObservedAt: time.Now()}}})
merged := mergeMetrics(&platform, agent, platform.Metrics, SourceProxmox, time.Now(), nil, nil)
if merged.Memory == nil || merged.Memory.Observation.State != "current" || merged.Memory.Observation.Source != "agent" || !merged.Memory.Observation.ObservedAt.Equal(at) || *merged.Memory.Used != 40 {
t.Fatalf("cross-source selection lost agent receipt or trusts supplied provenance: %+v", merged.Memory)
}
}
// Proxmox guest adapters must publish the node-independent identity key so
// canonical ID derivation survives live migration (#1669); guests without an
// instance+VMID stay keyless and fall back to source-specific derivation.
+3
View File
@@ -347,6 +347,9 @@ type MetricValue struct {
Percent float64 `json:"percent,omitempty"`
Unit string `json:"unit,omitempty"`
Source DataSource `json:"-"`
// Memory provenance travels with the selected numeric metric, not a
// different platform facet retained on a cross-source merged guest.
Observation models.MemoryObservation `json:"observation,omitzero"`
}
// ProxmoxData contains Proxmox-specific data for a resource.
+5 -1
View File
@@ -88,7 +88,8 @@ func linkedAgentMemoryFromResource(r *Resource) (models.Memory, bool) {
if r == nil || r.Agent == nil || r.Agent.Stale || r.Agent.Memory == nil {
return models.Memory{}, false
}
if status, ok := r.SourceStatus[SourceAgent]; !ok || status.Status != "online" {
status, ok := r.SourceStatus[SourceAgent]
if !ok || status.Status != "online" {
return models.Memory{}, false
}
m := r.Agent.Memory
@@ -97,6 +98,9 @@ func linkedAgentMemoryFromResource(r *Resource) (models.Memory, bool) {
return models.Memory{}, false
}
memory.Usage = float64(m.Used) / float64(m.Total) * 100
// Guest-row LastSeen can be renewed by PVE alone. The surviving agent
// sample keeps the agent source's timestamp after host/guest correlation.
memory.Observation = models.MemoryObservation{State: "current", Source: "agent", ObservedAt: status.LastSeen}
return memory, memory.HasKnownUsage()
}
+29
View File
@@ -12,6 +12,35 @@ import (
func ptrInt64(v int64) *int64 { return &v }
func ptrInt(v int) *int { return &v }
func TestLinkedAgentMemoryObservationUsesItsOwnSourceTime(t *testing.T) {
observedAt := time.Now().Add(-time.Minute)
resource := &Resource{
Type: ResourceTypeVM, LastSeen: time.Now(),
Agent: &AgentData{Memory: &AgentMemoryMeta{Total: 100, Used: 25, Free: 75}},
SourceStatus: map[DataSource]SourceStatus{SourceAgent: {Status: "online", LastSeen: observedAt}},
}
for _, guest := range []string{"qemu", "lxc"} {
t.Run(guest, func(t *testing.T) {
var memory models.Memory
var ok bool
if guest == "qemu" {
view := NewVMView(resource)
memory, ok = view.LinkedAgentMemory()
} else {
view := NewContainerView(resource)
memory, ok = view.LinkedAgentMemory()
}
if !ok || memory.Observation.State != "current" || memory.Observation.Source != "agent" || !memory.Observation.ObservedAt.Equal(observedAt) {
t.Fatalf("agent memory uses guest-row time or loses provenance: %+v", memory)
}
})
}
resource.SourceStatus[SourceAgent] = SourceStatus{Status: "offline", LastSeen: observedAt}
if _, ok := linkedAgentMemoryFromResource(resource); ok {
t.Fatal("provenance made an offline agent sample selectable")
}
}
func assertStringSlice(t *testing.T, got, want []string) {
t.Helper()
if len(got) != len(want) {