gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-22 03:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep platform CPU authoritative on hypervisor-managed guests

Browse Source 
An agent running inside an LXC measures /proc through the shared kernel
and reports the node's utilisation, not the container's. Because
SourceAgent outranked SourceProxmox in the metric merge, that value
overrode the hypervisor's cgroup-accounted CPU on the guest's canonical
resource, so the workloads row disagreed with the (Proxmox-sourced)
history chart by orders of magnitude on idle guests.

Demote agent-sourced utilisation metrics below the platform source when
the target resource is a hypervisor-managed guest (vm /
system-container). The freshness gate still lets a live agent cover for
a stale platform source.

Fixes #1597

Contract-Neutral: behavioral fix: demote in-guest agent utilisation below platform source on hypervisor-managed guests (issue #1597); no public contract delta
This commit is contained in:
rcourtman
2026-07-20 16:34:58 +01:00
parent b6a74576bc
commit a4d800fe3e
2 changed files with 113 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/unifiedresources/registry.go
+42 -12
View File
@@ -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:
internal/unifiedresources/registry_test.go
+71 -4
View File
@@ -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)
}