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

Harden hypervisor guest CPU authority

Browse Source
This commit is contained in:
rcourtman
2026-07-23 22:19:28 +01:00
parent b8ea840f11
commit d618fb8b74
34 changed files with 1288 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/release-control/v6/internal/subsystems/ai-runtime.md
+5
View File
@@ -30,6 +30,11 @@ contracts. The subsystem also owns AI orchestration, runtime cost control,
shared AI transport surfaces, and browser-visible Assistant transcript actions
that define what visible operator/model text can leave the transcript without
exposing hidden provider/tool metadata.
Patrol guest inventory consumes canonical unified-resource CPU percent when
read-state is available and the shared Proxmox guest CPU-percent normalizer for
legacy snapshot fallback. It must not independently rescale by configured cores
or substitute a linked in-guest host-agent CPU value, so Patrol evaluates the
same guest observation as dashboard, alerts, and history.
Interactive Assistant invocation authority is fail-closed and request-local.
The `read_only` presentation means no model-invokable durable Pulse-state or
docs/release-control/v6/internal/subsystems/alerts.md
+5
View File
@@ -21,6 +21,11 @@ Docker and Podman container CPU thresholds evaluate host-capacity-normalized
CPU percent, not Docker's runtime-native per-core percent. Alert metadata may
carry the raw per-core value and reporting host CPU count for evidence, but the
threshold value and canonical `cpuPercent` metadata remain normalized.
Proxmox VM and LXC CPU thresholds consume the same canonical guest CPU-percent
normalizer as unified live state and guest history. Proxmox guest CPU is the
authoritative observation for that guest; a Pulse host agent running inside the
guest may retain its own agent alert identity, but its host CPU observation
must not replace or renormalize the Proxmox guest value.
Docker and Podman OOM alerts require authoritative runtime evidence: the
container must be stopped (`exited` or `dead`) and its reported `OOMKilled`
state must be explicitly true. Exit code 137 alone is only SIGKILL evidence;
docs/release-control/v6/internal/subsystems/monitoring.md
+7
View File
@@ -126,6 +126,13 @@ unified-resource CPU metrics are 0..100 percentages, while legacy
0..1 ratio fields. Monitoring-owned read-state conversion must divide canonical
Proxmox node, VM, and LXC CPU percentages before handing them back to legacy
snapshot/current-row paths.
Proxmox guest live state, alerts, and history share one guest CPU-percent
normalizer. Efficient cluster polling and traditional per-node polling must
write VM/LXC history under the Proxmox guest ID in that same 0..100 unit, with
no core-count division and no in-guest host-agent substitution. A linked host
agent may supplement metrics the platform does not provide, but guest CPU and
its `vm` / `system-container` history target remain platform-owned so dashboard,
details, API state, alerts, and history cannot select different authorities.
Tenant monitor enumeration is monitoring-owned runtime topology, not a
reporting source of truth. `MultiTenantMonitor.ListOrganizationIDs` may expose
persisted organization IDs to API-owned background workers, but it must not
docs/release-control/v6/internal/subsystems/performance-and-scalability.md
+16
View File
@@ -132,6 +132,10 @@ admission records a stable refusal without invoking executor or network code.
104. `frontend-modern/src/components/Workloads/WorkloadsSurface.tsx`
105. `frontend-modern/src/components/Workloads/nestedWorkloadContext.ts`
106. `frontend-modern/src/components/Workloads/__tests__/nestedWorkloadContext.test.ts`
107. `frontend-modern/src/utils/workloads.ts`
108. `frontend-modern/src/utils/__tests__/workloads.test.ts`
109. `frontend-modern/src/utils/searchQuery.ts`
110. `frontend-modern/src/utils/__tests__/searchQuery.test.ts`
## Shared Boundaries
@@ -1582,6 +1586,18 @@ tooltip lifecycle lives in
`frontend-modern/src/components/Workloads/useEnhancedCPUBarState.ts`. Future
CPU-bar runtime changes must extend through those owners instead of
reintroducing mixed tooltip state and formatting logic into the shell.
The Workloads CPU unit boundary is shared across rows, sorting, filtering, and
drawer history fallback. Canonical unified-resource APIs supply percent values;
`frontend-modern/src/hooks/useWorkloads.ts` converts that value exactly once
into the `WorkloadGuest.cpu` ratio, and
`frontend-modern/src/utils/workloads.ts` owns the inverse ratio-to-percent
projection used by `GuestRow.tsx`, `workloadSelectors.ts`,
`frontend-modern/src/utils/searchQuery.ts`, and `guestDrawerModel.ts`.
`WorkloadGuest.cpus` is allocated-core metadata only and must never divide or
multiply the CPU observation. Workloads consumers must not infer source
precedence, reinterpret ratios above an arbitrary threshold, or replace an
established provider history series; source authority and canonical identity
convergence belong upstream in the unified-resource and monitoring contracts.
The unified resource table hot path is now also governed as explicit
performance-owned runtime, with shared ownership against the unified-resource
docs/release-control/v6/internal/subsystems/registry.json
+6
View File
@@ -5991,7 +5991,9 @@
"frontend-modern/src/routing/routePreload.ts",
"frontend-modern/src/types/workloads.ts",
"frontend-modern/src/useAppRuntimeState.ts",
"frontend-modern/src/utils/searchQuery.ts",
"frontend-modern/src/utils/thresholdSliderPresentation.ts",
"frontend-modern/src/utils/workloads.ts",
"frontend-modern/src/utils/workloadsSummaryCache.ts",
"internal/api/router_bench_test.go",
"internal/api/slo.go",
@@ -6132,7 +6134,9 @@
"frontend-modern/src/components/Workloads/workloadUrlSyncModel.ts",
"frontend-modern/src/hooks/useWorkloads.ts",
"frontend-modern/src/types/workloads.ts",
"frontend-modern/src/utils/searchQuery.ts",
"frontend-modern/src/utils/thresholdSliderPresentation.ts",
"frontend-modern/src/utils/workloads.ts",
"frontend-modern/src/utils/workloadsSummaryCache.ts"
],
"allow_same_subsystem_tests": false,
@@ -6169,7 +6173,9 @@
"frontend-modern/src/components/Workloads/StackedMemoryBar.test.tsx",
"frontend-modern/src/components/Workloads/ThresholdSlider.test.tsx",
"frontend-modern/src/hooks/__tests__/useWorkloads.test.ts",
"frontend-modern/src/utils/__tests__/searchQuery.test.ts",
"frontend-modern/src/utils/__tests__/thresholdSliderPresentation.test.ts",
"frontend-modern/src/utils/__tests__/workloads.test.ts",
"frontend-modern/src/utils/__tests__/workloadsSummaryCache.test.ts"
]
},
docs/release-control/v6/internal/subsystems/unified-resources.md
+16 -1
View File
@@ -33,6 +33,16 @@ reported as 0..100 percentages. Unified-resource host metric payloads and
host-derived storage adapters must clamp those reported percentages directly
rather than passing them through ratio-to-percent normalization, which is
reserved for providers that report 0..1 usage ratios.
Proxmox VM and LXC CPU is a semantic-authority exception to ordinary
freshness-based metric selection. The Proxmox guest observation is normalized
from its capacity ratio to canonical 0..100 percent exactly once, without
dividing by configured cores, and remains the guest CPU value whenever that
observation exists—even if the Proxmox source is stale. Source freshness stays
visible through `SourceStatus`; an in-guest host agent cannot supply a
comparable fallback because LXC observes the shared kernel and QEMU accounting
is independently scoped. Agent CPU may fill the field only when the platform
has no CPU observation. Agent-only fields that the platform does not provide
remain eligible per metric.
Physical-disk resources own cross-source disk identity. When Proxmox inventory
and host-agent SMART telemetry describe the same device, the merged resource
must retain Proxmox node/instance source payloads while carrying SMART
@@ -1657,7 +1667,12 @@ two source facets contribute the same metric, source priority decides only if
both sources have equivalent freshness. A stale source must not hold CPU,
memory, disk, network, or disk I/O metrics against a live source, and a stale
incoming source must not clobber a live metric already attached to the
canonical resource. This freshness gate belongs in
canonical resource. The explicit exception is hypervisor-managed guest CPU:
when a Proxmox or VMware platform CPU observation exists, semantic authority
precedes freshness so an in-guest agent never replaces it. Manual guest-agent
links are direction-independent: the hypervisor guest remains the canonical
resource and metrics target, each selected metric retains its real source, and
agent-only fields are preserved. This selection policy belongs in
`internal/unifiedresources/registry.go` and
`internal/unifiedresources/presentation_coalesce.go`, not in platform-page
rendering or frontend fallback code.
frontend-modern/src/components/Workloads/GuestRow.tsx
+2 -2
View File
@@ -17,7 +17,7 @@ import {
import { StatusDot } from '@/components/shared/StatusDot';
import { TagBadges } from '@/components/shared/TagBadges';
import { WorkloadTypeBadge } from '@/components/shared/WorkloadTypeBadge';
import { resolveWorkloadType } from '@/utils/workloads';
import { getWorkloadCPUPercent, resolveWorkloadType } from '@/utils/workloads';
import { EnhancedCPUBar } from '@/components/Workloads/EnhancedCPUBar';
import { MetricMiniSparkline } from '@/components/Workloads/MetricMiniSparkline';
import { UpdateButton } from '@/components/shared/ContainerUpdateBadge';
@@ -100,7 +100,7 @@ export function GuestRow(props: GuestRowProps) {
availabilityPresentation,
} = useGuestRowState(props);
const cpuPercent = createMemo(() => (props.guest.cpu || 0) * 100);
const cpuPercent = createMemo(() => getWorkloadCPUPercent(props.guest.cpu) ?? 0);
const metricDisplayMode = createMemo(() => props.metricDisplayMode ?? 'bars');
const isSparklineMode = createMemo(() => metricDisplayMode() === 'sparklines');
const detailControlsId = createMemo(() => buildSummaryDisclosureControlsId(guestId()));
frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx
+10
View File
@@ -244,6 +244,16 @@ describe('GuestRow', () => {
expect(cpuBar.dataset.usage).toBe('75');
});
it.each([1, 4, 8])(
'does not renormalize authoritative guest CPU by %i allocated cores',
(cpus) => {
renderGuestRow({ guest: makeGuest({ cpu: 0.0058, cpus }) });
const cpuBar = screen.getByTestId('cpu-bar');
expect(Number(cpuBar.dataset.usage)).toBeCloseTo(0.58);
expect(cpuBar.dataset.cores).toBe(String(cpus));
},
);
it('renders memory bar', () => {
renderGuestRow({ guest: makeGuest() });
expect(screen.getByTestId('memory-bar')).toBeTruthy();
frontend-modern/src/components/Workloads/__tests__/guestDrawerModel.branchcov2.test.ts
+3 -3
View File
@@ -77,7 +77,7 @@ describe('guestDrawerModel (branch coverage)', () => {
});
describe('getGuestDrawerHistoryFallbackMetrics', () => {
it('scales a fractional cpu (<=1.5) by 100 and returns finite metric values', () => {
it('scales the canonical workload cpu ratio by 100 and returns finite metric values', () => {
const guest = makeGuest({
cpu: 1.0,
memory: { total: 100, used: 40, free: 60, usage: 0.4 },
@@ -98,8 +98,8 @@ describe('guestDrawerModel (branch coverage)', () => {
});
});
it('passes a cpu value greater than 1.5 through unchanged', () => {
expect(getGuestDrawerHistoryFallbackMetrics(makeGuest({ cpu: 2.0 })).cpu).toBe(2);
it('uses the same ratio contract above 100 percent', () => {
expect(getGuestDrawerHistoryFallbackMetrics(makeGuest({ cpu: 2.0 })).cpu).toBe(200);
});
it('treats the cpu boundary of exactly 1.5 as a ratio (150)', () => {
frontend-modern/src/components/Workloads/guestDrawerModel.ts
+6 -8
View File
@@ -8,7 +8,11 @@ import type {
import { formatHistoryChartTooltipValue } from '@/components/shared/historyChartModel';
import { formatBytes, formatPercent } from '@/utils/format';
import { getCanonicalWorkloadId, resolveWorkloadType } from '@/utils/workloads';
import {
getCanonicalWorkloadId,
getWorkloadCPUPercent,
resolveWorkloadType,
} from '@/utils/workloads';
import type { NestedWorkloadContext } from './nestedWorkloadContext';
type Guest = WorkloadGuest;
@@ -67,13 +71,7 @@ export const isGuestDrawerVM = (guest: Guest): guest is VM => resolveWorkloadTyp
export const getGuestDrawerHistoryFallbackMetrics = (
guest: Guest,
): Record<string, number | undefined> => {
const cpuRaw = typeof guest.cpu === 'number' ? guest.cpu : undefined;
const cpuPercent =
cpuRaw === undefined || !Number.isFinite(cpuRaw)
? undefined
: cpuRaw <= 1.5
? cpuRaw * 100
: cpuRaw;
const cpuPercent = getWorkloadCPUPercent(guest.cpu);
const memUsage = guest.memory?.usage;
const diskUsage = guest.disk?.usage;
const finite = (value: number | undefined): number | undefined =>
frontend-modern/src/components/Workloads/workloadSelectors.ts
+3 -2
View File
@@ -10,6 +10,7 @@ import { getNodeDisplayName } from '@/utils/nodes';
import {
isContainerWorkloadViewMode,
getCanonicalWorkloadId,
getWorkloadCPUPercent,
resolveWorkloadType,
workloadMatchesPlatformScope,
workloadMatchesViewMode,
@@ -253,8 +254,8 @@ export const createWorkloadSortComparator = (
let bVal: SortValue = null;
if (sortKey === 'cpu') {
aVal = a.cpu * 100;
bVal = b.cpu * 100;
aVal = getWorkloadCPUPercent(a.cpu) ?? 0;
bVal = getWorkloadCPUPercent(b.cpu) ?? 0;
} else if (sortKey === 'memory') {
aVal = a.memory ? a.memory.usage || 0 : 0;
bVal = b.memory ? b.memory.usage || 0 : 0;
frontend-modern/src/hooks/__tests__/useWorkloads.test.ts
+45
View File
@@ -115,6 +115,51 @@ describe('useWorkloads', () => {
expect(useWorkloadsSource).toContain('createSignal<WorkloadGuest[]>');
});
it('projects authoritative LXC and QEMU CPU once regardless of agent sources or cores', async () => {
apiFetchJSONMock.mockResolvedValueOnce({
data: [
{
...sampleResource,
id: 'cluster-a-pve1-101',
type: 'vm',
name: 'qemu-101',
sources: ['proxmox'],
metrics: { ...sampleResource.metrics, cpu: { percent: 0.58 } },
proxmox: { vmid: 101, nodeName: 'pve1', instance: 'cluster-a', cpus: 8 },
},
{
...sampleResource,
id: 'cluster-a-pve1-102',
type: 'system-container',
name: 'lxc-102',
vmid: 102,
sources: ['proxmox', 'agent'],
metrics: { ...sampleResource.metrics, cpu: { percent: 0.58 } },
proxmox: { vmid: 102, nodeName: 'pve1', instance: 'cluster-a', cpus: 1 },
},
],
meta: { totalPages: 1 },
});
let dispose = () => {};
let result: ReturnType<UseWorkloadsModule['useWorkloads']> | undefined;
createRoot((d) => {
dispose = d;
const [enabled] = createSignal(true);
result = useWorkloads(enabled);
});
await waitForWorkloadCount(() => result!.workloads().length, 2);
const byName = new Map(result!.workloads().map((workload) => [workload.name, workload]));
expect(byName.get('qemu-101')?.cpu).toBe(0.0058);
expect(byName.get('qemu-101')?.cpus).toBe(8);
expect(byName.get('lxc-102')?.cpu).toBe(0.0058);
expect(byName.get('lxc-102')?.cpus).toBe(1);
dispose();
});
it('handles empty responses without mutating into undefined state', async () => {
apiFetchJSONMock.mockResolvedValueOnce({
data: [],
frontend-modern/src/hooks/useWorkloads.ts
+2 -1
View File
@@ -10,6 +10,7 @@ import { canonicalDiscoveryResourceType } from '@/utils/discoveryTarget';
import { normalizeDiskArray } from '@/utils/format';
import {
buildCanonicalNodeScopedWorkloadId,
getWorkloadCPUFraction,
isDockerManagedAppContainer,
resolveWorkloadTypeFromString,
} from '@/utils/workloads';
@@ -460,7 +461,7 @@ const mapResourceToWorkload = (resource: APIResource): WorkloadGuest | null => {
: workloadType === 'pod'
? 'pod'
: 'app-container',
cpu: cpuPercent / 100,
cpu: getWorkloadCPUFraction(cpuPercent),
cpus: resource.proxmox?.cpus ?? 1,
memory: (() => {
const base = buildMetric(resource.metrics?.memory);
frontend-modern/src/utils/__tests__/workloads.test.ts
+23
View File
@@ -3,6 +3,8 @@ import {
buildCanonicalNodeScopedWorkloadId,
canonicalizeWorkloadFilterType,
getDiscoveryResourceTypeForWorkload,
getWorkloadCPUFraction,
getWorkloadCPUPercent,
normalizeWorkloadViewModeParam,
resolveWorkloadType,
resolveWorkloadTypeFromString,
@@ -16,6 +18,27 @@ import {
} from '@/utils/workloads';
import type { WorkloadGuest } from '@/types/workloads';
describe('workload CPU normalization', () => {
it('round-trips canonical percent without applying allocated core count', () => {
const canonicalPercent = 0.58;
const ratio = getWorkloadCPUFraction(canonicalPercent);
expect(ratio).toBe(0.0058);
expect(getWorkloadCPUPercent(ratio)).toBeCloseTo(canonicalPercent);
});
it('keeps ratios above one as percentages above 100 instead of reinterpreting units', () => {
expect(getWorkloadCPUPercent(2)).toBe(200);
});
it('rejects non-finite values and clamps negative observations to zero', () => {
expect(getWorkloadCPUPercent(Number.NaN)).toBeUndefined();
expect(getWorkloadCPUFraction(Number.POSITIVE_INFINITY)).toBe(0);
expect(getWorkloadCPUPercent(-0.5)).toBe(0);
expect(getWorkloadCPUFraction(-50)).toBe(0);
});
});
describe('resolveWorkloadType', () => {
it('returns workloadType when present', () => {
const guest = { workloadType: 'vm' as const, type: 'system-container' };
frontend-modern/src/utils/searchQuery.ts
+2 -2
View File
@@ -1,4 +1,5 @@
import type { VM, Container } from '@/types/api';
import { getWorkloadCPUPercent } from '@/utils/workloads';
// Exclusion-aware split of a free-text search. Terms prefixed with `-` hide
// matching rows ("-watchtower" hides anything whose haystack contains
@@ -143,8 +144,7 @@ function evaluateMetricCondition(guest: FilterableItem, condition: MetricConditi
switch (condition.field) {
case 'cpu':
// CPU is stored as decimal (0-1), convert to percentage
value = ('cpu' in guest ? guest.cpu || 0 : 0) * 100;
value = getWorkloadCPUPercent('cpu' in guest ? guest.cpu : undefined) ?? 0;
break;
case 'memory':
value = 'memory' in guest && guest.memory ? guest.memory.usage : 0;
frontend-modern/src/utils/workloads.ts
+15
View File
@@ -14,6 +14,21 @@ import {
normalizeSourcePlatformScopes,
} from '@/utils/sourcePlatforms';
// WorkloadGuest.cpu is a normalized utilization ratio. It is independent of
// the guest's allocated core count; convert it exactly once at presentation
// boundaries so rows, filters, drawers, and history fallbacks cannot drift.
export const getWorkloadCPUPercent = (cpu: unknown): number | undefined => {
if (typeof cpu !== 'number' || !Number.isFinite(cpu)) return undefined;
return Math.max(0, cpu * 100);
};
// Unified-resource APIs expose canonical CPU as a percentage. WorkloadGuest is
// the legacy presentation boundary that stores the same value as a ratio.
export const getWorkloadCPUFraction = (percent: unknown): number => {
if (typeof percent !== 'number' || !Number.isFinite(percent)) return 0;
return Math.max(0, percent) / 100;
};
/**
* Resolve a raw type string (from API or backend) to a semantic WorkloadType.
* Returns null when the value cannot be mapped to any known workload type.
internal/ai/issue1597_metric_authority_test.go
+34
View File
@@ -0,0 +1,34 @@
package ai
import (
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
func TestIssue1597PatrolGuestFallbackUsesCanonicalProxmoxCPUPercent(t *testing.T) {
rows := patrolGuestInventoryRows(patrolRuntimeState{
VMs: []models.VM{{
ID: "vm-301",
Name: "vm-301",
CPU: 0.0058,
CPUs: 8,
}},
Containers: []models.Container{{
ID: "ct-302",
Name: "ct-302",
Type: "lxc",
CPU: 0.0058,
CPUs: 1,
}},
}, nil, nil)
if len(rows) != 2 {
t.Fatalf("guest rows = %d, want VM and LXC: %+v", len(rows), rows)
}
for _, row := range rows {
if row.cpu != 0.58 {
t.Fatalf("%s CPU = %v, want canonical proxmox 0.58", row.id, row.cpu)
}
}
}
internal/ai/patrol_ai.go
+2 -2
View File
@@ -2314,7 +2314,7 @@ func patrolGuestInventoryRows(snap patrolRuntimeState, scopedSet map[string]bool
gType: "VM",
node: vm.Node,
status: vm.Status,
cpu: vm.CPU * 100,
cpu: unifiedresources.ProxmoxGuestCPUPercent(vm.CPU),
mem: vm.Memory.Usage,
disk: vm.Disk.Usage,
vmid: vm.VMID,
@@ -2335,7 +2335,7 @@ func patrolGuestInventoryRows(snap patrolRuntimeState, scopedSet map[string]bool
gType: "Container",
node: ct.Node,
status: ct.Status,
cpu: ct.CPU * 100,
cpu: unifiedresources.ProxmoxGuestCPUPercent(ct.CPU),
mem: ct.Memory.Usage,
disk: ct.Disk.Usage,
vmid: ct.VMID,
internal/alerts/guest_snapshot.go
+3 -2
View File
@@ -4,6 +4,7 @@ import (
"strings"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
type guestKind uint8
@@ -100,7 +101,7 @@ func guestSnapshotFromVM(vm models.VM) guestSnapshot {
Instance: vm.Instance,
Status: vm.Status,
Lock: vm.Lock,
CPUPercent: vm.CPU * 100,
CPUPercent: unifiedresources.ProxmoxGuestCPUPercent(vm.CPU),
MemUsage: vm.Memory.Usage,
DiskUsage: vm.Disk.Usage,
DiskRead: vm.DiskRead,
@@ -123,7 +124,7 @@ func guestSnapshotFromContainer(container models.Container) guestSnapshot {
Instance: container.Instance,
Status: container.Status,
Lock: container.Lock,
CPUPercent: container.CPU * 100,
CPUPercent: unifiedresources.ProxmoxGuestCPUPercent(container.CPU),
MemUsage: container.Memory.Usage,
DiskUsage: container.Disk.Usage,
DiskRead: container.DiskRead,
internal/alerts/guest_snapshot_test.go
+26
View File
@@ -19,6 +19,32 @@ func TestGuestSnapshotResourceTypeUsesCanonicalSystemContainer(t *testing.T) {
}
}
func TestGuestSnapshotUsesCanonicalProxmoxCPUPercent(t *testing.T) {
for _, tc := range []struct {
name string
cpu float64
want float64
}{
{name: "zero", cpu: 0, want: 0},
{name: "fractional-capacity", cpu: 0.0058, want: 0.58},
{name: "full-capacity", cpu: 1, want: 100},
{name: "already-percent", cpu: 12.5, want: 12.5},
} {
t.Run("vm/"+tc.name, func(t *testing.T) {
snapshot := guestSnapshotFromVM(models.VM{CPU: tc.cpu})
if snapshot.CPUPercent != tc.want {
t.Fatalf("VM CPU percent = %v, want %v", snapshot.CPUPercent, tc.want)
}
})
t.Run("lxc/"+tc.name, func(t *testing.T) {
snapshot := guestSnapshotFromContainer(models.Container{CPU: tc.cpu})
if snapshot.CPUPercent != tc.want {
t.Fatalf("LXC CPU percent = %v, want %v", snapshot.CPUPercent, tc.want)
}
})
}
}
func TestParsePulseTagsRecognizesGuestRuntimeControls(t *testing.T) {
settings := parsePulseTags([]string{
" PULSE-NO-ALERTS ",
internal/api/resources_test.go
+153
View File
@@ -1659,6 +1659,159 @@ func TestResourceLinkMergesResources(t *testing.T) {
}
}
func TestResourceLinkFromAgentKeepsHypervisorGuestAuthorityInAPIState(t *testing.T) {
now := time.Now().UTC()
tests := []struct {
name string
guestType unified.ResourceType
targetType string
snapshot models.StateSnapshot
guestSourceID string
wantTechnology string
}{
{
name: "lxc",
guestType: unified.ResourceTypeSystemContainer,
targetType: "system-container",
guestSourceID: "pve-a-node-1-301",
snapshot: models.StateSnapshot{
Containers: []models.Container{{
ID: "pve-a-node-1-301",
VMID: 301,
Name: "lxc-301",
Node: "node-1",
Instance: "pve-a",
Type: "lxc",
Status: "running",
CPU: 0.0058,
CPUs: 4,
LastSeen: now,
}},
Hosts: []models.Host{{
ID: "agent-lxc-301",
Hostname: "agent-for-lxc-301",
Status: "online",
CPUUsage: 94,
NetOutRate: 4096,
CPUCount: 4,
LastSeen: now,
Memory: models.Memory{Total: 100, Used: 70, Usage: 70},
ReportIP: "10.0.30.1",
AgentVersion: "6.1.1",
}},
},
wantTechnology: "lxc",
},
{
name: "qemu",
guestType: unified.ResourceTypeVM,
targetType: "vm",
guestSourceID: "pve-a-node-1-302",
snapshot: models.StateSnapshot{
VMs: []models.VM{{
ID: "pve-a-node-1-302",
VMID: 302,
Name: "vm-302",
Node: "node-1",
Instance: "pve-a",
Type: "qemu",
Status: "running",
CPU: 0.0058,
CPUs: 8,
LastSeen: now,
}},
Hosts: []models.Host{{
ID: "agent-vm-302",
Hostname: "agent-for-vm-302",
Status: "online",
CPUUsage: 94,
NetOutRate: 4096,
CPUCount: 8,
LastSeen: now,
Memory: models.Memory{Total: 100, Used: 70, Usage: 70},
ReportIP: "10.0.30.2",
AgentVersion: "6.1.1",
}},
},
wantTechnology: "qemu",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
cfg := &config.Config{DataPath: t.TempDir()}
h := NewResourceHandlers(cfg)
h.SetStateProvider(resourceStateProvider{snapshot: tc.snapshot})
list := func() ResourcesResponse {
t.Helper()
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/resources", nil)
h.HandleListResources(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("list status = %d, body=%s", rec.Code, rec.Body.String())
}
var response ResourcesResponse
if err := json.NewDecoder(rec.Body).Decode(&response); err != nil {
t.Fatalf("decode list response: %v", err)
}
return response
}
before := list()
var agentID, guestID string
for _, resource := range before.Data {
switch resource.Type {
case unified.ResourceTypeAgent:
agentID = resource.ID
case tc.guestType:
guestID = resource.ID
}
}
if agentID == "" || guestID == "" {
t.Fatalf("missing pre-link resources: agent=%q guest=%q data=%+v", agentID, guestID, before.Data)
}
payload, err := json.Marshal(map[string]string{
"targetId": guestID,
"reason": "agent runs inside guest",
})
if err != nil {
t.Fatalf("marshal link payload: %v", err)
}
linkRec := httptest.NewRecorder()
linkReq := httptest.NewRequest(http.MethodPost, "/api/resources/"+agentID+"/link", bytes.NewReader(payload))
h.HandleLink(linkRec, linkReq)
if linkRec.Code != http.StatusOK {
t.Fatalf("link status = %d, body=%s", linkRec.Code, linkRec.Body.String())
}
after := list()
if len(after.Data) != 1 {
t.Fatalf("resources after link = %d, want one guest: %+v", len(after.Data), after.Data)
}
resource := after.Data[0]
if resource.ID != guestID || resource.Type != tc.guestType || resource.Technology != tc.wantTechnology {
t.Fatalf("linked API resource = %s/%s/%s, want guest %s/%s/%s", resource.ID, resource.Type, resource.Technology, guestID, tc.guestType, tc.wantTechnology)
}
if resource.Proxmox == nil || resource.Agent == nil {
t.Fatalf("linked API resource lost source payloads: proxmox=%+v agent=%+v", resource.Proxmox, resource.Agent)
}
if resource.Metrics == nil || resource.Metrics.CPU == nil || resource.Metrics.CPU.Percent != 0.58 {
t.Fatalf("API CPU = %+v, want canonical proxmox 0.58", resource.Metrics)
}
if resource.Metrics.NetOut == nil || resource.Metrics.NetOut.Value != 4096 {
t.Fatalf("API state lost agent-only network metric: %+v", resource.Metrics.NetOut)
}
if resource.MetricsTarget == nil ||
resource.MetricsTarget.ResourceType != tc.targetType ||
resource.MetricsTarget.ResourceID != tc.guestSourceID {
t.Fatalf("API metrics target = %+v, want %s/%s", resource.MetricsTarget, tc.targetType, tc.guestSourceID)
}
})
}
}
func TestResourceReportMergeCreatesExclusions(t *testing.T) {
now := time.Now().UTC()
sharedInterfaces := []models.HostNetworkInterface{
internal/monitoring/canonical_guardrails_test.go
+2 -2
View File
@@ -1998,14 +1998,14 @@ func TestMockNativePollersDeferToCanonicalMockSampler(t *testing.T) {
file: "monitor_polling_vm.go",
snippets: []string{
"if !shouldSkipNativeMockStateMetricWrites() {",
`m.recordGuestMetric("vm", vm.ID, vm.CPU*100, vm.Memory.Usage, vm.Disk.Usage, -1, -1, -1, -1, now)`,
`m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), vm.Memory.Usage, vm.Disk.Usage, -1, -1, -1, -1, now)`,
},
},
{
file: "monitor_polling_containers.go",
snippets: []string{
"if !shouldSkipNativeMockStateMetricWrites() {",
`m.recordGuestMetric("container", ct.ID, ct.CPU*100, ct.Memory.Usage, ct.Disk.Usage, -1, -1, -1, -1, now)`,
`m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), ct.Memory.Usage, ct.Disk.Usage, -1, -1, -1, -1, now)`,
},
},
{
internal/monitoring/docker_detection.go
+2 -1
View File
@@ -11,6 +11,7 @@ import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
"github.com/rs/zerolog/log"
)
@@ -693,7 +694,7 @@ func enrichGuestDockerReportFromContainer(report *agentsdocker.Report, container
return
}
if report.Host.CPUUsagePercent == 0 && container.CPU > 0 {
report.Host.CPUUsagePercent = container.CPU * 100
report.Host.CPUUsagePercent = unifiedresources.ProxmoxGuestCPUPercent(container.CPU)
}
if report.Host.Memory.TotalBytes == 0 && container.Memory.Total > 0 {
report.Host.Memory.TotalBytes = container.Memory.Total
internal/monitoring/issue1597_metric_authority_test.go
+154
View File
@@ -0,0 +1,154 @@
package monitoring
import (
"encoding/json"
"math"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func TestIssue1597UnifiedGuestCPUFeedsDashboardDetailsAndAlerts(t *testing.T) {
now := time.Now().UTC()
store := unifiedresources.NewMemoryStore()
if err := store.AddLink(unifiedresources.ResourceLink{
ResourceA: "guest",
ResourceB: "agent",
PrimaryID: "agent",
}); err != nil {
t.Fatalf("AddLink() error = %v", err)
}
registry := unifiedresources.NewRegistry(store)
registry.IngestResources([]unifiedresources.Resource{
{
ID: "guest",
Type: unifiedresources.ResourceTypeSystemContainer,
Name: "database",
Status: unifiedresources.StatusOnline,
LastSeen: now.Add(-2 * time.Hour),
Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
SourceStatus: map[unifiedresources.DataSource]unifiedresources.SourceStatus{
unifiedresources.SourceProxmox: {
Status: "stale",
LastSeen: now.Add(-2 * time.Hour),
},
},
Metrics: &unifiedresources.ResourceMetrics{
CPU: &unifiedresources.MetricValue{
Value: 0.58,
Percent: 0.58,
Unit: "percent",
Source: unifiedresources.SourceProxmox,
},
},
Proxmox: &unifiedresources.ProxmoxData{
SourceID: "cluster-a:pve-1:301",
NodeName: "pve-1",
ClusterName: "cluster-a",
Instance: "connection-a",
VMID: 301,
ContainerType: "lxc",
CPUs: 4,
},
},
{
ID: "agent",
Type: unifiedresources.ResourceTypeAgent,
Name: "database-agent",
Status: unifiedresources.StatusOnline,
LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent},
SourceStatus: map[unifiedresources.DataSource]unifiedresources.SourceStatus{
unifiedresources.SourceAgent: {Status: "online", LastSeen: now},
},
Metrics: &unifiedresources.ResourceMetrics{
CPU: &unifiedresources.MetricValue{
Value: 58,
Percent: 58,
Unit: "percent",
Source: unifiedresources.SourceAgent,
},
NetIn: &unifiedresources.MetricValue{
Value: 4096,
Unit: "bytes_per_second",
Source: unifiedresources.SourceAgent,
},
},
Agent: &unifiedresources.AgentData{
AgentID: "agent",
Hostname: "database",
OSName: "Linux",
OSVersion: "6.8",
},
},
})
adapter := unifiedresources.NewMonitorAdapter(registry)
frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter)
if len(frontend) != 1 {
t.Fatalf("dashboard resource count = %d, want one unified guest", len(frontend))
}
if frontend[0].Type != "system-container" || frontend[0].CPU == nil {
t.Fatalf("dashboard guest shape = %+v, want system-container with CPU", frontend[0])
}
if math.Abs(frontend[0].CPU.Current-0.58) > 0.000001 {
t.Fatalf("dashboard CPU = %v, want authoritative Proxmox 0.58", frontend[0].CPU.Current)
}
if len(frontend[0].Proxmox) == 0 || len(frontend[0].Agent) == 0 {
t.Fatalf("details payload lost source facets: proxmox=%s agent=%s", frontend[0].Proxmox, frontend[0].Agent)
}
var metricsTarget unifiedresources.MetricsTarget
if err := json.Unmarshal(frontend[0].MetricsTarget, &metricsTarget); err != nil {
t.Fatalf("decode metrics target: %v", err)
}
if metricsTarget.ResourceType != "system-container" || metricsTarget.ResourceID != "cluster-a:pve-1:301" {
t.Fatalf("dashboard metrics target = %+v, want platform guest history", metricsTarget)
}
containers := adapter.Containers()
if len(containers) != 1 {
t.Fatalf("canonical container view count = %d, want one", len(containers))
}
alertGuest := containerFromReadStateView(containers[0])
alertGuest.Status = "running"
if math.Abs(alertGuest.CPU-0.0058) > 0.000001 || alertGuest.CPUs != 4 {
t.Fatalf("alert guest CPU/core shape = cpu:%v cpus:%d, want ratio 0.0058 and metadata-only four cores", alertGuest.CPU, alertGuest.CPUs)
}
alertManager := alerts.NewManagerWithDataDir(t.TempDir())
t.Cleanup(alertManager.Stop)
alertManager.UpdateConfig(alerts.AlertConfig{
Enabled: true,
ActivationState: alerts.ActivationActive,
GuestDefaults: alerts.ThresholdConfig{
CPU: &alerts.HysteresisThreshold{Trigger: 10, Clear: 5},
},
TimeThresholds: map[string]int{},
MetricTimeThresholds: map[string]map[string]int{},
})
alertManager.CheckGuest(alertGuest, "connection-a")
if cpuAlert := issue1597CPUAlert(alertManager.GetActiveAlerts()); cpuAlert != nil {
t.Fatalf("authoritative 0.58%% platform CPU incorrectly fired alert: %+v", cpuAlert)
}
// Prove the threshold would have fired if the 58% in-guest agent value had
// leaked into the canonical guest alert path.
alertGuest.CPU = 0.58
alertManager.CheckGuest(alertGuest, "connection-a")
if cpuAlert := issue1597CPUAlert(alertManager.GetActiveAlerts()); cpuAlert == nil {
t.Fatal("alert control did not fire for the rejected 58% agent CPU value")
}
}
func issue1597CPUAlert(active []alerts.Alert) *alerts.Alert {
for index := range active {
if active[index].Type == "cpu" {
return &active[index]
}
}
return nil
}
internal/monitoring/memory_source_catalog_test.go
+1 -1
View File
@@ -138,7 +138,7 @@ func TestMockVMPollingDefersMemoryHistoryToCanonicalSampler(t *testing.T) {
requiredSnippets := []string{
"if !shouldSkipNativeMockStateMetricWrites() {",
`m.recordGuestMetric("vm", vm.ID, vm.CPU*100, vm.Memory.Usage, vm.Disk.Usage, -1, -1, -1, -1, now)`,
`m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), vm.Memory.Usage, vm.Disk.Usage, -1, -1, -1, -1, now)`,
}
for _, snippet := range requiredSnippets {
if !strings.Contains(source, snippet) {
internal/monitoring/monitor_polling_containers.go
+2 -1
View File
@@ -8,6 +8,7 @@ import (
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
"github.com/rs/zerolog/log"
)
@@ -336,7 +337,7 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri
}
// IO/network series are not recorded on the traditional polling
// path (parity with the historical inline writes).
m.recordGuestMetric("container", ct.ID, ct.CPU*100, ct.Memory.Usage, ct.Disk.Usage, -1, -1, -1, -1, now)
m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), ct.Memory.Usage, ct.Disk.Usage, -1, -1, -1, -1, now)
}
}
internal/monitoring/monitor_polling_vm.go
+2 -1
View File
@@ -7,6 +7,7 @@ import (
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
"github.com/rs/zerolog/log"
)
@@ -123,7 +124,7 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
}
// IO/network series are not recorded on the traditional polling
// path (parity with the historical inline writes).
m.recordGuestMetric("vm", vm.ID, vm.CPU*100, vm.Memory.Usage, vm.Disk.Usage, -1, -1, -1, -1, now)
m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), vm.Memory.Usage, vm.Disk.Usage, -1, -1, -1, -1, now)
}
}
internal/monitoring/monitor_pve_guest_cpu_authority_test.go
+81
View File
@@ -0,0 +1,81 @@
package monitoring
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/mock"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
)
func TestRecordGuestMetricsKeepsLiveAndHistoricalProxmoxCPUUnitsAligned(t *testing.T) {
previousMockMode := mock.IsMockEnabled()
mustSetMockEnabled(t, false)
t.Cleanup(func() {
mustSetMockEnabled(t, previousMockMode)
})
cfg := metrics.DefaultConfig(t.TempDir())
cfg.WriteBufferSize = 32
cfg.FlushInterval = time.Hour
store, err := metrics.NewStore(cfg)
if err != nil {
t.Fatalf("new metrics store: %v", err)
}
t.Cleanup(func() {
_ = store.Close()
})
if err := store.WaitForMaintenance(5 * time.Second); err != nil {
t.Fatalf("wait for metrics maintenance: %v", err)
}
history := NewMetricsHistory(32, time.Hour)
monitor := &Monitor{
metricsHistory: history,
metricsStore: store,
}
monitor.recordGuestMetrics(
[]models.VM{{
ID: "cluster-a-node-1-301",
Status: "running",
CPU: 0.0058,
CPUs: 8,
LastSeen: time.Now().UTC(),
}},
[]models.Container{{
ID: "cluster-a-node-1-302",
Status: "running",
Type: "lxc",
CPU: 0.0058,
CPUs: 1,
LastSeen: time.Now().UTC(),
}},
)
store.Flush()
for _, tc := range []struct {
name string
resourceType string
resourceID string
}{
{name: "qemu", resourceType: "vm", resourceID: "cluster-a-node-1-301"},
{name: "lxc", resourceType: "container", resourceID: "cluster-a-node-1-302"},
} {
t.Run(tc.name, func(t *testing.T) {
historyPoints := history.GetGuestMetrics(tc.resourceID, "cpu", time.Hour)
if len(historyPoints) != 1 || historyPoints[0].Value != 0.58 {
t.Fatalf("in-memory CPU history = %+v, want one 0.58 point", historyPoints)
}
now := time.Now().UTC()
storePoints, err := store.Query(tc.resourceType, tc.resourceID, "cpu", now.Add(-time.Minute), now.Add(time.Minute), 0)
if err != nil {
t.Fatalf("query persistent CPU history: %v", err)
}
if len(storePoints) != 1 || storePoints[0].Value != 0.58 {
t.Fatalf("persistent CPU history = %+v, want one 0.58 point", storePoints)
}
})
}
}
internal/monitoring/monitor_pve_guest_helpers.go
+3 -2
View File
@@ -4,6 +4,7 @@ import (
"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"
"github.com/rs/zerolog/log"
)
@@ -109,12 +110,12 @@ func (m *Monitor) recordGuestMetrics(allVMs []models.VM, allContainers []models.
now := time.Now()
for _, vm := range allVMs {
if vm.Status == "running" {
m.recordGuestMetric("vm", vm.ID, vm.CPU*100, vm.Memory.Usage, vm.Disk.Usage, vm.DiskRead, vm.DiskWrite, vm.NetworkIn, vm.NetworkOut, now)
m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), vm.Memory.Usage, vm.Disk.Usage, vm.DiskRead, vm.DiskWrite, vm.NetworkIn, vm.NetworkOut, now)
}
}
for _, ct := range allContainers {
if ct.Status == "running" {
m.recordGuestMetric("container", ct.ID, ct.CPU*100, ct.Memory.Usage, ct.Disk.Usage, ct.DiskRead, ct.DiskWrite, ct.NetworkIn, ct.NetworkOut, now)
m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), ct.Memory.Usage, ct.Disk.Usage, ct.DiskRead, ct.DiskWrite, ct.NetworkIn, ct.NetworkOut, now)
}
}
}
internal/monitoring/monitor_unified_state_test.go
+111
View File
@@ -250,6 +250,117 @@ func TestConvertResourcesForBroadcastAttachesResolvedStorageMetricsTarget(t *tes
}
}
func TestConvertResourcesForBroadcastKeepsLinkedGuestCPUAndHistoryTargetAligned(t *testing.T) {
now := time.Now().UTC()
tests := []struct {
name string
resourceType unifiedresources.ResourceType
technology string
targetType string
containerType string
}{
{
name: "lxc",
resourceType: unifiedresources.ResourceTypeSystemContainer,
technology: "lxc",
targetType: "system-container",
containerType: "lxc",
},
{
name: "qemu",
resourceType: unifiedresources.ResourceTypeVM,
technology: "qemu",
targetType: "vm",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
guestID := string(tc.resourceType) + "-guest-301"
guestSourceID := "cluster-a-node-1-" + tc.name + "-301"
const agentID = "agent-inside-guest-301"
store := unifiedresources.NewMemoryStore()
if err := store.AddLink(unifiedresources.ResourceLink{
ResourceA: guestID,
ResourceB: agentID,
PrimaryID: agentID,
}); err != nil {
t.Fatalf("add link: %v", err)
}
registry := unifiedresources.NewRegistry(store)
registry.IngestResources([]unifiedresources.Resource{
{
ID: guestID,
Type: tc.resourceType,
Technology: tc.technology,
Name: tc.name + "-301",
Status: unifiedresources.StatusOnline,
LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
SourceStatus: map[unifiedresources.DataSource]unifiedresources.SourceStatus{
unifiedresources.SourceProxmox: {Status: "online", LastSeen: now},
},
Proxmox: &unifiedresources.ProxmoxData{
SourceID: guestSourceID,
VMID: 301,
ContainerType: tc.containerType,
CPUs: 8,
},
Metrics: &unifiedresources.ResourceMetrics{
CPU: &unifiedresources.MetricValue{
Value: 0.58,
Percent: 0.58,
Unit: "percent",
Source: unifiedresources.SourceProxmox,
},
},
},
{
ID: agentID,
Type: unifiedresources.ResourceTypeAgent,
Name: tc.name + "-301",
Status: unifiedresources.StatusOnline,
LastSeen: now,
Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent},
SourceStatus: map[unifiedresources.DataSource]unifiedresources.SourceStatus{
unifiedresources.SourceAgent: {Status: "online", LastSeen: now},
},
Agent: &unifiedresources.AgentData{AgentID: agentID, CPUCount: 8},
Metrics: &unifiedresources.ResourceMetrics{
CPU: &unifiedresources.MetricValue{
Value: 94,
Percent: 94,
Unit: "percent",
Source: unifiedresources.SourceAgent,
},
},
},
})
adapter := unifiedresources.NewMonitorAdapter(registry)
frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter)
if len(frontend) != 1 {
t.Fatalf("broadcast resources = %d, want one merged guest: %+v", len(frontend), frontend)
}
got := frontend[0]
if got.ID != guestID || got.Type != string(tc.resourceType) || got.Technology != tc.technology {
t.Fatalf("broadcast resource = %s/%s/%s, want %s/%s/%s", got.ID, got.Type, got.Technology, guestID, tc.resourceType, tc.technology)
}
if got.CPU == nil || got.CPU.Current != 0.58 {
t.Fatalf("broadcast CPU = %+v, want proxmox 0.58", got.CPU)
}
var target unifiedresources.MetricsTarget
if err := json.Unmarshal(got.MetricsTarget, &target); err != nil {
t.Fatalf("decode metrics target: %v", err)
}
if target.ResourceType != tc.targetType || target.ResourceID != guestSourceID {
t.Fatalf("broadcast metrics target = %+v, want %s/%s", target, tc.targetType, guestSourceID)
}
})
}
}
func hasFrontendResourceName(resources []models.ResourceFrontend, name string) bool {
for _, resource := range resources {
if resource.Name == name {
internal/unifiedresources/metrics.go
+10 -1
View File
@@ -184,7 +184,7 @@ func buildVMMetricPayload(
source DataSource,
) *ResourceMetrics {
metrics := &ResourceMetrics{}
cpuPercent := percentFromUsage(cpu)
cpuPercent := ProxmoxGuestCPUPercent(cpu)
metrics.CPU = &MetricValue{Value: cpuPercent, Percent: cpuPercent, Unit: "percent", Source: source}
if memory.Total > 0 {
percent := percentFromUsage(memory.Usage)
@@ -794,6 +794,15 @@ func percentFromUsage(value float64) float64 {
return value
}
// ProxmoxGuestCPUPercent converts the Proxmox guest capacity ratio to the
// canonical 0..100 percent unit shared by live state, alerts, and history.
// Values already expressed as percent are retained for compatibility with
// normalized fixtures and read-state adapters. Guest CPU is not divided by
// the configured core count: Proxmox already reports capacity utilization.
func ProxmoxGuestCPUPercent(value float64) float64 {
return percentFromUsage(value)
}
func percentFromReportedPercent(value float64) float64 {
return clampMetricValue(value, 0, 100)
}
internal/unifiedresources/metrics_test.go
+31
View File
@@ -87,6 +87,37 @@ func TestMetricsFromDockerContainerKeepsCapacityPercentValues(t *testing.T) {
}
}
func TestMetricsFromProxmoxGuestsNormalizesCPUOnceIndependentOfCoreCount(t *testing.T) {
for _, tc := range []struct {
name string
cores int
}{
{name: "one-core", cores: 1},
{name: "four-cores", cores: 4},
{name: "eight-cores", cores: 8},
} {
t.Run("vm-"+tc.name, func(t *testing.T) {
metrics := metricsFromVM(models.VM{CPU: 0.0058, CPUs: tc.cores})
if metrics.CPU == nil {
t.Fatal("VM CPU metric is nil")
}
if metrics.CPU.Percent != 0.58 || metrics.CPU.Value != 0.58 || metrics.CPU.Source != SourceProxmox {
t.Fatalf("VM CPU = %+v, want canonical proxmox percent 0.58", metrics.CPU)
}
})
t.Run("lxc-"+tc.name, func(t *testing.T) {
metrics := metricsFromContainer(models.Container{CPU: 0.0058, CPUs: tc.cores})
if metrics.CPU == nil {
t.Fatal("LXC CPU metric is nil")
}
if metrics.CPU.Percent != 0.58 || metrics.CPU.Value != 0.58 || metrics.CPU.Source != SourceProxmox {
t.Fatalf("LXC CPU = %+v, want canonical proxmox percent 0.58", metrics.CPU)
}
})
}
}
func TestMetricsFromKubernetesClusterAggregatesLinkedHostReportedPercents(t *testing.T) {
cluster := models.KubernetesCluster{ID: "cluster-1", Name: "cluster-1"}
hosts := []*models.Host{
internal/unifiedresources/registry.go
+122 -30
View File
@@ -498,7 +498,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
rr.mu.Lock()
rr.pbsBackups = clonePBSBackups(snapshot.PBSBackups)
rr.applyManualLinks()
rr.applyManualLinks(thresholds)
rr.refreshStorageConsumersLocked()
rr.refreshPBSRollupsLocked()
rr.refreshStoragePostureLocked()
@@ -667,7 +667,7 @@ func (rr *ResourceRegistry) ingestResources(resources []Resource, thresholds map
for _, resourceID := range seededIDs {
rr.seedSourceMappingsFromResourceLocked(rr.resources[resourceID])
}
rr.applyManualLinks()
rr.applyManualLinks(thresholds)
rr.refreshStorageConsumersLocked()
rr.refreshPBSRollupsLocked()
rr.refreshStoragePostureLocked()
@@ -3180,7 +3180,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, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus)
existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, nil)
// Prefer agent naming when available
if incoming.Name != "" {
@@ -3675,7 +3675,7 @@ func mergeVMwareData(existing *VMwareData, incoming *VMwareData) *VMwareData {
return &merged
}
func (rr *ResourceRegistry) applyManualLinks() {
func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Duration) {
if len(rr.links) == 0 {
return
}
@@ -3706,13 +3706,22 @@ func (rr *ResourceRegistry) applyManualLinks() {
continue
}
rr.mergeResourceData(primary, other)
// A manual link records operator intent to unify identities, but its
// direction must not change the semantic resource shape. In particular,
// an agent running inside a VM/LXC supplements that guest; it does not
// turn the guest into an agent resource with an agent metrics target.
if primary.Type == ResourceTypeAgent && hypervisorManagedGuest(other) {
primary, other = other, primary
primaryID, otherID = otherID, primaryID
}
rr.mergeResourceData(primary, other, thresholds)
delete(rr.resources, otherID)
rr.updateSourceMappings(otherID, primaryID)
}
}
func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource) {
func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource, thresholds map[DataSource]time.Duration) {
if other == nil || primary == nil {
return
}
@@ -3768,7 +3777,9 @@ func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource
primary.Ceph = other.Ceph
}
primary.Metrics = mergeMetrics(primary, primary.Metrics, other.Metrics, SourceAgent, time.Now().UTC(), primary.SourceStatus)
// Manual links combine already-normalized resources. Preserve each metric's
// recorded source instead of flattening the linked resource to SourceAgent.
primary.Metrics = mergeMetrics(primary, primary.Metrics, other.Metrics, "", time.Now().UTC(), primary.SourceStatus, thresholds)
primary.Status = aggregateStatus(primary)
}
@@ -4722,28 +4733,41 @@ func addSources(sources []DataSource, more []DataSource) []DataSource {
return out
}
func mergeMetrics(target *Resource, existing *ResourceMetrics, incoming *ResourceMetrics, source DataSource, now time.Time, status map[DataSource]SourceStatus) *ResourceMetrics {
if existing == nil {
return incoming
}
func mergeMetrics(
target *Resource,
existing *ResourceMetrics,
incoming *ResourceMetrics,
source DataSource,
now time.Time,
status map[DataSource]SourceStatus,
thresholds map[DataSource]time.Duration,
) *ResourceMetrics {
if incoming == nil {
return existing
}
merged := *existing
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)
var merged ResourceMetrics
if existing != nil {
merged = *existing
}
merged.CPU = mergeCPUMetric(target, merged.CPU, incoming.CPU, source, now, status, thresholds)
merged.Memory = mergeMetric(target, merged.Memory, incoming.Memory, source, now, status, thresholds)
merged.Disk = mergeMetric(target, merged.Disk, incoming.Disk, source, now, status, thresholds)
merged.NetIn = mergeMetric(target, merged.NetIn, incoming.NetIn, source, now, status, thresholds)
merged.NetOut = mergeMetric(target, merged.NetOut, incoming.NetOut, source, now, status, thresholds)
merged.DiskRead = mergeMetric(target, merged.DiskRead, incoming.DiskRead, source, now, status, thresholds)
merged.DiskWrite = mergeMetric(target, merged.DiskWrite, incoming.DiskWrite, source, now, status, thresholds)
return &merged
}
// metricSourceStale reports whether a source's most recent report is older than
// its stale threshold. A zero/unknown last-seen is treated as NOT stale so the
// merge never demotes a source on missing information.
func metricSourceStale(now time.Time, status map[DataSource]SourceStatus, source DataSource) bool {
func metricSourceStale(
now time.Time,
status map[DataSource]SourceStatus,
source DataSource,
thresholds map[DataSource]time.Duration,
) bool {
if status == nil {
return false
}
@@ -4751,19 +4775,37 @@ func metricSourceStale(now time.Time, status map[DataSource]SourceStatus, source
if !ok || st.LastSeen.IsZero() {
return false
}
threshold, ok := defaultStaleThresholds[source]
if !ok {
threshold := time.Duration(0)
if configured := thresholds[source]; configured > 0 {
threshold = configured
}
if threshold <= 0 {
threshold = defaultStaleThresholds[source]
}
if threshold <= 0 {
threshold = 60 * time.Second
}
return now.Sub(st.LastSeen) > threshold
}
func mergeMetric(target *Resource, 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,
thresholds map[DataSource]time.Duration,
) *MetricValue {
if incoming == nil {
return existing
}
incomingCopy := *incoming
incomingCopy.Source = source
incomingSource := source
if incomingSource == "" {
incomingSource = incoming.Source
}
incomingCopy.Source = incomingSource
if existing == nil {
return &incomingCopy
}
@@ -4771,20 +4813,70 @@ func mergeMetric(target *Resource, existing *MetricValue, incoming *MetricValue,
// metric against a live one, and a live source overrides a stale one
// regardless of priority. Mirrors the presentation-coalesce rule so the
// "live source wins a metric" invariant holds in both metric-merge paths.
existingStale := metricSourceStale(now, status, existing.Source)
incomingStale := metricSourceStale(now, status, source)
existingStale := metricSourceStale(now, status, existing.Source, thresholds)
incomingStale := metricSourceStale(now, status, incomingSource, thresholds)
if existingStale != incomingStale {
if existingStale {
return &incomingCopy
}
return existing
}
if metricMergePriority(target, source) >= metricMergePriority(target, existing.Source) {
if metricMergePriority(target, incomingSource) >= metricMergePriority(target, existing.Source) {
return &incomingCopy
}
return existing
}
// mergeCPUMetric enforces semantic authority before freshness. An in-guest
// agent cannot provide a comparable CPU observation for a hypervisor-managed
// guest: LXC sees the shared host kernel and QEMU uses different accounting.
// Once a platform CPU observation exists, retain it even while that source is
// stale so live state, alerts, and platform-keyed history cannot disagree.
// SourceStatus remains the separate signal that the observation is stale.
func mergeCPUMetric(
target *Resource,
existing *MetricValue,
incoming *MetricValue,
source DataSource,
now time.Time,
status map[DataSource]SourceStatus,
thresholds map[DataSource]time.Duration,
) *MetricValue {
if incoming == nil {
return existing
}
incomingSource := source
if incomingSource == "" {
incomingSource = incoming.Source
}
if existing != nil && hypervisorManagedGuest(target) {
existingAuthoritative := authoritativeGuestCPUSource(target, existing.Source)
incomingAuthoritative := authoritativeGuestCPUSource(target, incomingSource)
if existingAuthoritative != incomingAuthoritative {
if existingAuthoritative {
return existing
}
incomingCopy := *incoming
incomingCopy.Source = incomingSource
return &incomingCopy
}
}
return mergeMetric(target, existing, incoming, source, now, status, thresholds)
}
func authoritativeGuestCPUSource(target *Resource, source DataSource) bool {
if !hypervisorManagedGuest(target) {
return false
}
if target.Proxmox != nil {
return source == SourceProxmox
}
if target.VMware != nil {
return source == SourceVMware
}
return source == SourceProxmox || source == SourceVMware
}
// 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 {
@@ -4805,9 +4897,9 @@ func hypervisorManagedGuest(r *Resource) bool {
// 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.
// platform series — so the agent must not outrank the platform there. CPU has
// the stronger semantic rule in mergeCPUMetric; other overlapping metrics keep
// the ordinary freshness gate and this priority policy.
func metricMergePriority(target *Resource, source DataSource) int {
if source == SourceAgent && hypervisorManagedGuest(target) {
return 1
internal/unifiedresources/registry_test.go
+383 -14
View File
@@ -4390,7 +4390,7 @@ 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(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceProxmox, now, status)
got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceProxmox, now, status, nil)
if got == nil || got.Percent != 6.2 || got.Source != SourceProxmox {
t.Fatalf("expected live proxmox CPU to override stale agent CPU, got %+v", got)
}
@@ -4407,7 +4407,7 @@ func TestMergeMetric_FreshHigherPriorityStillWins(t *testing.T) {
existing := &MetricValue{Percent: 6.2, Source: SourceProxmox}
incoming := &MetricValue{Percent: 12.5, Source: SourceAgent}
got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceAgent, now, status)
got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceAgent, now, status, nil)
if got == nil || got.Percent != 12.5 || got.Source != SourceAgent {
t.Fatalf("expected fresh agent CPU to win by priority, got %+v", got)
}
@@ -4417,7 +4417,7 @@ func TestMergeMetric_FreshHigherPriorityStillWins(t *testing.T) {
// 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) {
func TestMergeCPUMetric_GuestKeepsPlatformCPUOverInGuestAgent(t *testing.T) {
now := time.Now().UTC()
status := map[DataSource]SourceStatus{
SourceAgent: {Status: "online", LastSeen: now},
@@ -4430,7 +4430,7 @@ func TestMergeMetric_GuestKeepsPlatformCPUOverInGuestAgent(t *testing.T) {
existing := &MetricValue{Percent: 0.58, Source: SourceProxmox}
incoming := &MetricValue{Percent: 58, Source: SourceAgent}
got := mergeMetric(guest, existing, incoming, SourceAgent, now, status)
got := mergeCPUMetric(guest, existing, incoming, SourceAgent, now, status, nil)
if got == nil || got.Percent != 0.58 || got.Source != SourceProxmox {
t.Fatalf("expected platform CPU to survive in-guest agent merge, got %+v", got)
}
@@ -4438,7 +4438,7 @@ func TestMergeMetric_GuestKeepsPlatformCPUOverInGuestAgent(t *testing.T) {
// 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) {
func TestMergeCPUMetric_GuestPlatformOverridesEarlierAgentValue(t *testing.T) {
now := time.Now().UTC()
status := map[DataSource]SourceStatus{
SourceAgent: {Status: "online", LastSeen: now},
@@ -4451,16 +4451,16 @@ func TestMergeMetric_GuestPlatformOverridesEarlierAgentValue(t *testing.T) {
existing := &MetricValue{Percent: 58, Source: SourceAgent}
incoming := &MetricValue{Percent: 0.58, Source: SourceProxmox}
got := mergeMetric(guest, existing, incoming, SourceProxmox, now, status)
got := mergeCPUMetric(guest, existing, incoming, SourceProxmox, now, status, nil)
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) {
// Source freshness is reported independently from semantic authority. A live
// in-guest agent must not replace a stale platform CPU observation because the
// two values use different accounting and history remains platform-keyed.
func TestMergeCPUMetric_GuestRetainsStalePlatformObservation(t *testing.T) {
now := time.Now().UTC()
status := map[DataSource]SourceStatus{
SourceAgent: {Status: "online", LastSeen: now},
@@ -4473,9 +4473,9 @@ func TestMergeMetric_GuestStalePlatformFallsBackToAgent(t *testing.T) {
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)
got := mergeCPUMetric(guest, existing, incoming, SourceAgent, now, status, nil)
if got == nil || got.Percent != 0.58 || got.Source != SourceProxmox {
t.Fatalf("expected stale platform CPU to remain authoritative, got %+v", got)
}
}
@@ -4489,12 +4489,381 @@ func TestMergeMetric_StaleSourceDoesNotClobberLive(t *testing.T) {
existing := &MetricValue{Percent: 6.2, Source: SourceProxmox}
incoming := &MetricValue{Percent: 0, Source: SourceAgent}
got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceAgent, now, status)
got := mergeMetric(&Resource{Type: ResourceTypeAgent}, existing, incoming, SourceAgent, now, status, nil)
if got == nil || got.Percent != 6.2 || got.Source != SourceProxmox {
t.Fatalf("expected live proxmox CPU to survive stale agent merge, got %+v", got)
}
}
func TestResourceRegistry_ManualGuestAgentLinksPreserveMetricAuthority(t *testing.T) {
now := time.Now().UTC()
type guestShape struct {
name string
resourceType ResourceType
technology string
targetType string
containerType string
guestID string
guestSourceID string
guestSourceCPU float64
}
guestShapes := []guestShape{
{
name: "lxc",
resourceType: ResourceTypeSystemContainer,
technology: "lxc",
targetType: "system-container",
containerType: "lxc",
guestID: "system-container-cluster-a-node-1-301",
guestSourceID: "cluster-a-node-1-301",
guestSourceCPU: 0.58,
},
{
name: "qemu",
resourceType: ResourceTypeVM,
technology: "qemu",
targetType: "vm",
guestID: "vm-cluster-a-node-1-302",
guestSourceID: "cluster-a-node-1-302",
guestSourceCPU: 37.5,
},
}
freshnessCases := []struct {
name string
platformLastSeen time.Time
agentLastSeen time.Time
thresholds map[DataSource]time.Duration
wantMemorySource DataSource
}{
{
name: "both-fresh",
platformLastSeen: now,
agentLastSeen: now,
wantMemorySource: SourceProxmox,
},
{
name: "platform-stale-agent-fresh",
platformLastSeen: now.Add(-2 * time.Hour),
agentLastSeen: now,
wantMemorySource: SourceAgent,
},
{
name: "platform-fresh-agent-stale",
platformLastSeen: now,
agentLastSeen: now.Add(-2 * time.Hour),
wantMemorySource: SourceProxmox,
},
{
name: "both-stale",
platformLastSeen: now.Add(-2 * time.Hour),
agentLastSeen: now.Add(-2 * time.Hour),
wantMemorySource: SourceProxmox,
},
{
name: "caller-threshold-keeps-platform-fresh",
platformLastSeen: now.Add(-2 * time.Minute),
agentLastSeen: now,
thresholds: map[DataSource]time.Duration{SourceProxmox: 5 * time.Minute},
wantMemorySource: SourceProxmox,
},
}
for _, shape := range guestShapes {
for _, freshness := range freshnessCases {
for _, primaryIsAgent := range []bool{false, true} {
direction := "guest-primary"
if primaryIsAgent {
direction = "agent-primary"
}
t.Run(shape.name+"/"+freshness.name+"/"+direction, func(t *testing.T) {
const agentID = "agent-inside-guest"
primaryID := shape.guestID
if primaryIsAgent {
primaryID = agentID
}
store := NewMemoryStore()
if err := store.AddLink(ResourceLink{
ResourceA: shape.guestID,
ResourceB: agentID,
PrimaryID: primaryID,
Reason: "same guest",
}); err != nil {
t.Fatalf("add link: %v", err)
}
rr := NewRegistry(store)
rr.IngestResourcesWithStaleThresholds([]Resource{
{
ID: shape.guestID,
Type: shape.resourceType,
Technology: shape.technology,
Name: shape.name + "-guest",
Status: StatusOnline,
LastSeen: freshness.platformLastSeen,
Sources: []DataSource{SourceProxmox},
SourceStatus: map[DataSource]SourceStatus{
SourceProxmox: {Status: "online", LastSeen: freshness.platformLastSeen},
},
Proxmox: &ProxmoxData{
SourceID: shape.guestSourceID,
Instance: "pve-a",
ClusterName: "cluster-a",
NodeName: "node-1",
VMID: 301,
ContainerType: shape.containerType,
CPUs: 8,
},
Metrics: &ResourceMetrics{
CPU: &MetricValue{Percent: shape.guestSourceCPU, Source: SourceProxmox},
Memory: &MetricValue{Percent: 40, Source: SourceProxmox},
},
},
{
ID: agentID,
Type: ResourceTypeAgent,
Name: shape.name + "-guest",
Status: StatusOnline,
LastSeen: freshness.agentLastSeen,
Sources: []DataSource{SourceAgent},
SourceStatus: map[DataSource]SourceStatus{
SourceAgent: {Status: "online", LastSeen: freshness.agentLastSeen},
},
Agent: &AgentData{AgentID: agentID, Hostname: shape.name + "-guest", CPUCount: 8},
Metrics: &ResourceMetrics{
CPU: &MetricValue{Percent: 94, Source: SourceAgent},
Memory: &MetricValue{Percent: 70, Source: SourceAgent},
NetOut: &MetricValue{Value: 4096, Unit: "bytes/s", Source: SourceAgent},
},
},
}, freshness.thresholds)
resources := rr.List()
if len(resources) != 1 {
t.Fatalf("resources = %d, want one merged guest: %+v", len(resources), resources)
}
got := resources[0]
if got.ID != shape.guestID || got.Type != shape.resourceType {
t.Fatalf("merged resource = %s/%s, want guest %s/%s", got.ID, got.Type, shape.guestID, shape.resourceType)
}
if got.Proxmox == nil || got.Agent == nil {
t.Fatalf("merged guest lost provider payloads: proxmox=%+v agent=%+v", got.Proxmox, got.Agent)
}
if got.Metrics == nil || got.Metrics.CPU == nil {
t.Fatalf("merged guest lost CPU metrics: %+v", got.Metrics)
}
if got.Metrics.CPU.Percent != shape.guestSourceCPU || got.Metrics.CPU.Source != SourceProxmox {
t.Fatalf("CPU = %+v, want authoritative proxmox %.2f", got.Metrics.CPU, shape.guestSourceCPU)
}
if got.Metrics.Memory == nil || got.Metrics.Memory.Source != freshness.wantMemorySource {
t.Fatalf("memory = %+v, want source %s", got.Metrics.Memory, freshness.wantMemorySource)
}
if got.Metrics.NetOut == nil || got.Metrics.NetOut.Value != 4096 || got.Metrics.NetOut.Source != SourceAgent {
t.Fatalf("agent-only network metric was not preserved: %+v", got.Metrics.NetOut)
}
target := BuildMetricsTargetForRegistry(rr, got.ID)
if target == nil || target.ResourceType != shape.targetType || target.ResourceID != shape.guestSourceID {
t.Fatalf("metrics target = %+v, want %s/%s", target, shape.targetType, shape.guestSourceID)
}
})
}
}
}
}
func TestResourceRegistry_ManualGuestAgentLinkUsesAgentCPUOnlyWhenPlatformHasNoCPU(t *testing.T) {
now := time.Now().UTC()
const (
guestID = "vm-cluster-a-node-1-401"
agentID = "agent-inside-vm-401"
)
store := NewMemoryStore()
if err := store.AddLink(ResourceLink{
ResourceA: guestID,
ResourceB: agentID,
PrimaryID: agentID,
}); err != nil {
t.Fatalf("add link: %v", err)
}
rr := NewRegistry(store)
rr.IngestResources([]Resource{
{
ID: guestID,
Type: ResourceTypeVM,
Name: "vm-401",
Status: StatusOnline,
LastSeen: now,
Sources: []DataSource{SourceProxmox},
SourceStatus: map[DataSource]SourceStatus{
SourceProxmox: {Status: "online", LastSeen: now},
},
Proxmox: &ProxmoxData{SourceID: "cluster-a-node-1-401", VMID: 401},
Metrics: &ResourceMetrics{
Memory: &MetricValue{Percent: 50, Source: SourceProxmox},
},
},
{
ID: agentID,
Type: ResourceTypeAgent,
Name: "vm-401",
Status: StatusOnline,
LastSeen: now,
Sources: []DataSource{SourceAgent},
SourceStatus: map[DataSource]SourceStatus{
SourceAgent: {Status: "online", LastSeen: now},
},
Agent: &AgentData{AgentID: agentID},
Metrics: &ResourceMetrics{
CPU: &MetricValue{Percent: 33, Source: SourceAgent},
},
},
})
got, ok := rr.Get(guestID)
if !ok || got.Metrics == nil || got.Metrics.CPU == nil {
t.Fatalf("merged guest missing fallback CPU: %+v", got)
}
if got.Metrics.CPU.Percent != 33 || got.Metrics.CPU.Source != SourceAgent {
t.Fatalf("CPU = %+v, want agent fallback when platform observation is absent", got.Metrics.CPU)
}
}
func TestResourceRegistry_ManualGuestAgentLinkReconnectRestoresFreshPlatformCPU(t *testing.T) {
now := time.Now().UTC()
const (
guestID = "system-container-cluster-a-node-1-451"
guestSourceID = "cluster-a-node-1-451"
agentID = "agent-inside-lxc-451"
)
store := NewMemoryStore()
if err := store.AddLink(ResourceLink{
ResourceA: guestID,
ResourceB: agentID,
PrimaryID: agentID,
}); err != nil {
t.Fatalf("add link: %v", err)
}
rr := NewRegistry(store)
ingest := func(platformCPU float64, platformLastSeen time.Time) {
t.Helper()
rr.IngestResources([]Resource{
{
ID: guestID,
Type: ResourceTypeSystemContainer,
Name: "lxc-451",
Status: StatusOnline,
LastSeen: platformLastSeen,
Sources: []DataSource{SourceProxmox},
SourceStatus: map[DataSource]SourceStatus{
SourceProxmox: {Status: "online", LastSeen: platformLastSeen},
},
Proxmox: &ProxmoxData{SourceID: guestSourceID, VMID: 451, ContainerType: "lxc"},
Metrics: &ResourceMetrics{
CPU: &MetricValue{Percent: platformCPU, Source: SourceProxmox},
},
},
{
ID: agentID,
Type: ResourceTypeAgent,
Name: "lxc-451",
Status: StatusOnline,
LastSeen: now,
Sources: []DataSource{SourceAgent},
SourceStatus: map[DataSource]SourceStatus{
SourceAgent: {Status: "online", LastSeen: now},
},
Agent: &AgentData{AgentID: agentID},
Metrics: &ResourceMetrics{
CPU: &MetricValue{Percent: 94, Source: SourceAgent},
},
},
})
}
ingest(0.58, now.Add(-2*time.Hour))
first, ok := rr.Get(guestID)
if !ok || first.Metrics == nil || first.Metrics.CPU == nil {
t.Fatalf("first merged guest missing CPU: %+v", first)
}
if first.Metrics.CPU.Percent != 0.58 || first.Metrics.CPU.Source != SourceProxmox {
t.Fatalf("stale platform CPU = %+v, want proxmox 0.58", first.Metrics.CPU)
}
ingest(12.5, now.Add(time.Second))
reconnected, ok := rr.Get(guestID)
if !ok || reconnected.Metrics == nil || reconnected.Metrics.CPU == nil {
t.Fatalf("reconnected guest missing CPU: %+v", reconnected)
}
if reconnected.Metrics.CPU.Percent != 12.5 || reconnected.Metrics.CPU.Source != SourceProxmox {
t.Fatalf("reconnected CPU = %+v, want fresh proxmox 12.5", reconnected.Metrics.CPU)
}
target := BuildMetricsTargetForRegistry(rr, guestID)
if target == nil || target.ResourceType != "system-container" || target.ResourceID != guestSourceID {
t.Fatalf("reconnected metrics target = %+v", target)
}
}
func TestResourceRegistry_LinkHintsKeepClusterGuestsDistinctAndPlatformAuthoritative(t *testing.T) {
now := time.Now().UTC()
rr := NewRegistry(nil)
rr.IngestSnapshot(models.StateSnapshot{
Nodes: []models.Node{
{ID: "cluster-a-node-1", Name: "node-1", Instance: "pve-a", ClusterName: "cluster-a", IsClusterMember: true, Status: "online", LastSeen: now},
{ID: "cluster-b-node-1", Name: "node-1", Instance: "pve-b", ClusterName: "cluster-b", IsClusterMember: true, Status: "online", LastSeen: now},
},
VMs: []models.VM{
{ID: "cluster-a-node-1-501", VMID: 501, Name: "duplicate-vmid", Node: "node-1", Instance: "pve-a", Status: "running", CPU: 0.0058, CPUs: 1, LastSeen: now},
{ID: "cluster-b-node-1-501", VMID: 501, Name: "duplicate-vmid", Node: "node-1", Instance: "pve-b", Status: "running", CPU: 0.375, CPUs: 8, LastSeen: now},
},
Containers: []models.Container{
{ID: "cluster-a-node-1-601", VMID: 601, Name: "lxc-with-agent", Node: "node-1", Instance: "pve-a", Type: "lxc", Status: "running", CPU: 0.0058, CPUs: 4, LastSeen: now},
},
Hosts: []models.Host{
{ID: "agent-vm-501", Hostname: "duplicate-vmid", LinkedVMID: "cluster-a-node-1-501", CPUUsage: 94, Status: "online", LastSeen: now},
{ID: "agent-lxc-601", Hostname: "lxc-with-agent", LinkedContainerID: "cluster-a-node-1-601", CPUUsage: 88, Status: "online", LastSeen: now},
},
})
vms := rr.ListByType(ResourceTypeVM)
if len(vms) != 2 {
t.Fatalf("VMs = %d, want two distinct cluster-scoped identities: %+v", len(vms), vms)
}
wantVMCPU := map[string]float64{
"cluster-a-node-1-501": 0.58,
"cluster-b-node-1-501": 37.5,
}
for _, vm := range vms {
if vm.Proxmox == nil || vm.Metrics == nil || vm.Metrics.CPU == nil {
t.Fatalf("VM missing proxmox CPU: %+v", vm)
}
want, ok := wantVMCPU[vm.Proxmox.SourceID]
if !ok {
t.Fatalf("unexpected VM source identity: %+v", vm.Proxmox)
}
if vm.Metrics.CPU.Percent != want || vm.Metrics.CPU.Source != SourceProxmox {
t.Fatalf("VM %s CPU = %+v, want proxmox %.2f", vm.Proxmox.SourceID, vm.Metrics.CPU, want)
}
}
containers := rr.ListByType(ResourceTypeSystemContainer)
if len(containers) != 1 || containers[0].Metrics == nil || containers[0].Metrics.CPU == nil {
t.Fatalf("LXC resources = %+v, want one with CPU", containers)
}
if containers[0].Metrics.CPU.Percent != 0.58 || containers[0].Metrics.CPU.Source != SourceProxmox {
t.Fatalf("LXC CPU = %+v, want proxmox 0.58", containers[0].Metrics.CPU)
}
hostAgentCount := 0
for _, resource := range rr.ListByType(ResourceTypeAgent) {
if resource.Agent != nil {
hostAgentCount++
}
}
if hostAgentCount != 2 {
t.Fatalf("link hints must not implicitly merge guest identities; host agents = %d", hostAgentCount)
}
}
// The merged-source host shape from the homelab canonical-ID bug: a PVE node
// record that only knows cluster+hostname, and a pulse-agent record that
// knows the machine ID. Whichever record mints the canonical resource decides