diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 9e88edc95..eef80263c 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -3142,7 +3142,7 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou existing.ParentID = rr.resolveCanonicalParentID(existing) existing.Status = chooseStatus(existing.Status, incoming.Status, source) - existing.Metrics = mergeMetrics(existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus) + existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus) // Prefer agent naming when available if incoming.Name != "" { @@ -3728,7 +3728,7 @@ func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource primary.Ceph = other.Ceph } - primary.Metrics = mergeMetrics(primary.Metrics, other.Metrics, SourceAgent, time.Now().UTC(), primary.SourceStatus) + primary.Metrics = mergeMetrics(primary, primary.Metrics, other.Metrics, SourceAgent, time.Now().UTC(), primary.SourceStatus) primary.Status = aggregateStatus(primary) } @@ -4682,7 +4682,7 @@ func addSources(sources []DataSource, more []DataSource) []DataSource { return out } -func mergeMetrics(existing *ResourceMetrics, incoming *ResourceMetrics, source DataSource, now time.Time, status map[DataSource]SourceStatus) *ResourceMetrics { +func mergeMetrics(target *Resource, existing *ResourceMetrics, incoming *ResourceMetrics, source DataSource, now time.Time, status map[DataSource]SourceStatus) *ResourceMetrics { if existing == nil { return incoming } @@ -4690,13 +4690,13 @@ func mergeMetrics(existing *ResourceMetrics, incoming *ResourceMetrics, source D return existing } merged := *existing - merged.CPU = mergeMetric(existing.CPU, incoming.CPU, source, now, status) - merged.Memory = mergeMetric(existing.Memory, incoming.Memory, source, now, status) - merged.Disk = mergeMetric(existing.Disk, incoming.Disk, source, now, status) - merged.NetIn = mergeMetric(existing.NetIn, incoming.NetIn, source, now, status) - merged.NetOut = mergeMetric(existing.NetOut, incoming.NetOut, source, now, status) - merged.DiskRead = mergeMetric(existing.DiskRead, incoming.DiskRead, source, now, status) - merged.DiskWrite = mergeMetric(existing.DiskWrite, incoming.DiskWrite, source, now, status) + merged.CPU = mergeMetric(target, existing.CPU, incoming.CPU, source, now, status) + merged.Memory = mergeMetric(target, existing.Memory, incoming.Memory, source, now, status) + merged.Disk = mergeMetric(target, existing.Disk, incoming.Disk, source, now, status) + merged.NetIn = mergeMetric(target, existing.NetIn, incoming.NetIn, source, now, status) + merged.NetOut = mergeMetric(target, existing.NetOut, incoming.NetOut, source, now, status) + merged.DiskRead = mergeMetric(target, existing.DiskRead, incoming.DiskRead, source, now, status) + merged.DiskWrite = mergeMetric(target, existing.DiskWrite, incoming.DiskWrite, source, now, status) return &merged } @@ -4718,7 +4718,7 @@ func metricSourceStale(now time.Time, status map[DataSource]SourceStatus, source return now.Sub(st.LastSeen) > threshold } -func mergeMetric(existing *MetricValue, incoming *MetricValue, source DataSource, now time.Time, status map[DataSource]SourceStatus) *MetricValue { +func mergeMetric(target *Resource, existing *MetricValue, incoming *MetricValue, source DataSource, now time.Time, status map[DataSource]SourceStatus) *MetricValue { if incoming == nil { return existing } @@ -4739,12 +4739,42 @@ func mergeMetric(existing *MetricValue, incoming *MetricValue, source DataSource } return existing } - if sourcePriority(source) >= sourcePriority(existing.Source) { + if metricMergePriority(target, source) >= metricMergePriority(target, existing.Source) { return &incomingCopy } return existing } +// hypervisorManagedGuest reports whether the resource is a guest workload whose +// utilisation is accounted by a hypervisor platform (Proxmox VM/LXC, VMware VM). +func hypervisorManagedGuest(r *Resource) bool { + if r == nil { + return false + } + switch r.Type { + case ResourceTypeVM, ResourceTypeSystemContainer: + return true + } + if r.Proxmox != nil && (r.Proxmox.VMID > 0 || strings.TrimSpace(r.Proxmox.ContainerType) != "") { + return true + } + return false +} + +// metricMergePriority resolves the priority a source's utilisation metrics get +// on a specific resource. On hypervisor-managed guests the platform's numbers +// are authoritative — an agent inside an LXC measures through the shared +// kernel and reports the node's utilisation, and history charts follow the +// platform series — so the agent must not outrank the platform there. The +// freshness gate in mergeMetric still lets a live agent cover for a stale +// platform source. +func metricMergePriority(target *Resource, source DataSource) int { + if source == SourceAgent && hypervisorManagedGuest(target) { + return 1 + } + return sourcePriority(source) +} + func sourcePriority(source DataSource) int { switch source { case SourceAgent: diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index db422e1c9..73b51bdb7 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -4390,13 +4390,14 @@ func TestMergeMetric_LiveSourceOverridesStaleHigherPriority(t *testing.T) { existing := &MetricValue{Value: 0, Percent: 0, Source: SourceAgent} incoming := &MetricValue{Value: 6.2, Percent: 6.2, Source: SourceProxmox} - got := mergeMetric(existing, incoming, SourceProxmox, now, status) + got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceProxmox, now, status) if got == nil || got.Percent != 6.2 || got.Source != SourceProxmox { t.Fatalf("expected live proxmox CPU to override stale agent CPU, got %+v", got) } } -// When both sources are fresh, static priority still decides: the agent wins. +// When both sources are fresh, static priority still decides: the agent wins +// on a host-shaped resource (its own kernel, its own numbers). func TestMergeMetric_FreshHigherPriorityStillWins(t *testing.T) { now := time.Now().UTC() status := map[DataSource]SourceStatus{ @@ -4406,12 +4407,78 @@ func TestMergeMetric_FreshHigherPriorityStillWins(t *testing.T) { existing := &MetricValue{Percent: 6.2, Source: SourceProxmox} incoming := &MetricValue{Percent: 12.5, Source: SourceAgent} - got := mergeMetric(existing, incoming, SourceAgent, now, status) + got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceAgent, now, status) if got == nil || got.Percent != 12.5 || got.Source != SourceAgent { t.Fatalf("expected fresh agent CPU to win by priority, got %+v", got) } } +// Issue #1597: an agent inside an LXC measures through the shared kernel and +// reports the node's CPU, not the container's. On a hypervisor-managed guest +// the platform metric must survive an agent merge even though the agent +// outranks Proxmox on host-shaped resources. +func TestMergeMetric_GuestKeepsPlatformCPUOverInGuestAgent(t *testing.T) { + now := time.Now().UTC() + status := map[DataSource]SourceStatus{ + SourceAgent: {Status: "online", LastSeen: now}, + SourceProxmox: {Status: "online", LastSeen: now}, + } + guest := &Resource{ + Type: ResourceTypeSystemContainer, + Proxmox: &ProxmoxData{VMID: 301, ContainerType: "lxc"}, + } + existing := &MetricValue{Percent: 0.58, Source: SourceProxmox} + incoming := &MetricValue{Percent: 58, Source: SourceAgent} + + got := mergeMetric(guest, existing, incoming, SourceAgent, now, status) + if got == nil || got.Percent != 0.58 || got.Source != SourceProxmox { + t.Fatalf("expected platform CPU to survive in-guest agent merge, got %+v", got) + } +} + +// Reverse ingest order for the same guest: the platform metric arrives while +// the agent's value is already in place. The platform must take over. +func TestMergeMetric_GuestPlatformOverridesEarlierAgentValue(t *testing.T) { + now := time.Now().UTC() + status := map[DataSource]SourceStatus{ + SourceAgent: {Status: "online", LastSeen: now}, + SourceProxmox: {Status: "online", LastSeen: now}, + } + guest := &Resource{ + Type: ResourceTypeSystemContainer, + Proxmox: &ProxmoxData{VMID: 301, ContainerType: "lxc"}, + } + existing := &MetricValue{Percent: 58, Source: SourceAgent} + incoming := &MetricValue{Percent: 0.58, Source: SourceProxmox} + + got := mergeMetric(guest, existing, incoming, SourceProxmox, now, status) + if got == nil || got.Percent != 0.58 || got.Source != SourceProxmox { + t.Fatalf("expected platform CPU to override earlier agent value on guest, got %+v", got) + } +} + +// The freshness gate still lets a live in-guest agent cover for a platform +// source that went stale (e.g. PVE API unreachable while the guest keeps +// reporting). +func TestMergeMetric_GuestStalePlatformFallsBackToAgent(t *testing.T) { + now := time.Now().UTC() + status := map[DataSource]SourceStatus{ + SourceAgent: {Status: "online", LastSeen: now}, + SourceProxmox: {Status: "stale", LastSeen: now.Add(-2 * time.Hour)}, + } + guest := &Resource{ + Type: ResourceTypeSystemContainer, + Proxmox: &ProxmoxData{VMID: 301, ContainerType: "lxc"}, + } + existing := &MetricValue{Percent: 0.58, Source: SourceProxmox} + incoming := &MetricValue{Percent: 12.5, Source: SourceAgent} + + got := mergeMetric(guest, existing, incoming, SourceAgent, now, status) + if got == nil || got.Percent != 12.5 || got.Source != SourceAgent { + t.Fatalf("expected live agent CPU to cover for stale platform source, got %+v", got) + } +} + // A stale source must not clobber a metric currently held by a live source. func TestMergeMetric_StaleSourceDoesNotClobberLive(t *testing.T) { now := time.Now().UTC() @@ -4422,7 +4489,7 @@ func TestMergeMetric_StaleSourceDoesNotClobberLive(t *testing.T) { existing := &MetricValue{Percent: 6.2, Source: SourceProxmox} incoming := &MetricValue{Percent: 0, Source: SourceAgent} - got := mergeMetric(existing, incoming, SourceAgent, now, status) + got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceAgent, now, status) if got == nil || got.Percent != 6.2 || got.Source != SourceProxmox { t.Fatalf("expected live proxmox CPU to survive stale agent merge, got %+v", got) }