diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 94908c24f..5929522a6 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -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 diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 0c3cd3697..d5b9ade01 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -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; diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 2501500b8..7dd113875 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -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 diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 1571d5c00..e9514d955 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -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 diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index 9c5af5928..103b64d7e 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -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" ] }, diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index e69141847..50c033ef9 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -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. diff --git a/frontend-modern/src/components/Workloads/GuestRow.tsx b/frontend-modern/src/components/Workloads/GuestRow.tsx index 27c042ecc..d24092e95 100644 --- a/frontend-modern/src/components/Workloads/GuestRow.tsx +++ b/frontend-modern/src/components/Workloads/GuestRow.tsx @@ -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())); diff --git a/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx b/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx index a6aca115f..90f399670 100644 --- a/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx +++ b/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx @@ -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(); diff --git a/frontend-modern/src/components/Workloads/__tests__/guestDrawerModel.branchcov2.test.ts b/frontend-modern/src/components/Workloads/__tests__/guestDrawerModel.branchcov2.test.ts index ac2ded60e..4d981b204 100644 --- a/frontend-modern/src/components/Workloads/__tests__/guestDrawerModel.branchcov2.test.ts +++ b/frontend-modern/src/components/Workloads/__tests__/guestDrawerModel.branchcov2.test.ts @@ -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)', () => { diff --git a/frontend-modern/src/components/Workloads/guestDrawerModel.ts b/frontend-modern/src/components/Workloads/guestDrawerModel.ts index d32c68c76..0d6b626a8 100644 --- a/frontend-modern/src/components/Workloads/guestDrawerModel.ts +++ b/frontend-modern/src/components/Workloads/guestDrawerModel.ts @@ -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 => { - 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 => diff --git a/frontend-modern/src/components/Workloads/workloadSelectors.ts b/frontend-modern/src/components/Workloads/workloadSelectors.ts index 5149ed5d5..dc7527ebe 100644 --- a/frontend-modern/src/components/Workloads/workloadSelectors.ts +++ b/frontend-modern/src/components/Workloads/workloadSelectors.ts @@ -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; diff --git a/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts b/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts index 5d35654c0..b08cc4432 100644 --- a/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts +++ b/frontend-modern/src/hooks/__tests__/useWorkloads.test.ts @@ -115,6 +115,51 @@ describe('useWorkloads', () => { expect(useWorkloadsSource).toContain('createSignal'); }); + 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 | 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: [], diff --git a/frontend-modern/src/hooks/useWorkloads.ts b/frontend-modern/src/hooks/useWorkloads.ts index faddbfa7b..c6bbb76c7 100644 --- a/frontend-modern/src/hooks/useWorkloads.ts +++ b/frontend-modern/src/hooks/useWorkloads.ts @@ -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); diff --git a/frontend-modern/src/utils/__tests__/workloads.test.ts b/frontend-modern/src/utils/__tests__/workloads.test.ts index 04579cb32..c4b18ff56 100644 --- a/frontend-modern/src/utils/__tests__/workloads.test.ts +++ b/frontend-modern/src/utils/__tests__/workloads.test.ts @@ -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' }; diff --git a/frontend-modern/src/utils/searchQuery.ts b/frontend-modern/src/utils/searchQuery.ts index 1e3562772..06e65cf26 100644 --- a/frontend-modern/src/utils/searchQuery.ts +++ b/frontend-modern/src/utils/searchQuery.ts @@ -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; diff --git a/frontend-modern/src/utils/workloads.ts b/frontend-modern/src/utils/workloads.ts index 3b3aeb604..cad2e2e9e 100644 --- a/frontend-modern/src/utils/workloads.ts +++ b/frontend-modern/src/utils/workloads.ts @@ -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. diff --git a/internal/ai/issue1597_metric_authority_test.go b/internal/ai/issue1597_metric_authority_test.go new file mode 100644 index 000000000..e9b6c010a --- /dev/null +++ b/internal/ai/issue1597_metric_authority_test.go @@ -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) + } + } +} diff --git a/internal/ai/patrol_ai.go b/internal/ai/patrol_ai.go index f50efeb9e..26f8f2c60 100644 --- a/internal/ai/patrol_ai.go +++ b/internal/ai/patrol_ai.go @@ -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, diff --git a/internal/alerts/guest_snapshot.go b/internal/alerts/guest_snapshot.go index 5d124e62f..1d3ea8378 100644 --- a/internal/alerts/guest_snapshot.go +++ b/internal/alerts/guest_snapshot.go @@ -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, diff --git a/internal/alerts/guest_snapshot_test.go b/internal/alerts/guest_snapshot_test.go index 9ca7cb9a7..d9a9679af 100644 --- a/internal/alerts/guest_snapshot_test.go +++ b/internal/alerts/guest_snapshot_test.go @@ -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 ", diff --git a/internal/api/resources_test.go b/internal/api/resources_test.go index 081acf71c..51c0658f2 100644 --- a/internal/api/resources_test.go +++ b/internal/api/resources_test.go @@ -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{ diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 5ed57ab1c..e09f48c3c 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -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)`, }, }, { diff --git a/internal/monitoring/docker_detection.go b/internal/monitoring/docker_detection.go index 8caa316e9..54718f628 100644 --- a/internal/monitoring/docker_detection.go +++ b/internal/monitoring/docker_detection.go @@ -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 diff --git a/internal/monitoring/issue1597_metric_authority_test.go b/internal/monitoring/issue1597_metric_authority_test.go new file mode 100644 index 000000000..3b5e950e1 --- /dev/null +++ b/internal/monitoring/issue1597_metric_authority_test.go @@ -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 +} diff --git a/internal/monitoring/memory_source_catalog_test.go b/internal/monitoring/memory_source_catalog_test.go index d138d0db3..56d49985d 100644 --- a/internal/monitoring/memory_source_catalog_test.go +++ b/internal/monitoring/memory_source_catalog_test.go @@ -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) { diff --git a/internal/monitoring/monitor_polling_containers.go b/internal/monitoring/monitor_polling_containers.go index 76de799fc..d2b8bb58f 100644 --- a/internal/monitoring/monitor_polling_containers.go +++ b/internal/monitoring/monitor_polling_containers.go @@ -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) } } diff --git a/internal/monitoring/monitor_polling_vm.go b/internal/monitoring/monitor_polling_vm.go index 98af66aa9..fb8e77b3a 100644 --- a/internal/monitoring/monitor_polling_vm.go +++ b/internal/monitoring/monitor_polling_vm.go @@ -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) } } diff --git a/internal/monitoring/monitor_pve_guest_cpu_authority_test.go b/internal/monitoring/monitor_pve_guest_cpu_authority_test.go new file mode 100644 index 000000000..589f44bda --- /dev/null +++ b/internal/monitoring/monitor_pve_guest_cpu_authority_test.go @@ -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) + } + }) + } +} diff --git a/internal/monitoring/monitor_pve_guest_helpers.go b/internal/monitoring/monitor_pve_guest_helpers.go index e30e94ead..fbdbb6f4a 100644 --- a/internal/monitoring/monitor_pve_guest_helpers.go +++ b/internal/monitoring/monitor_pve_guest_helpers.go @@ -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) } } } diff --git a/internal/monitoring/monitor_unified_state_test.go b/internal/monitoring/monitor_unified_state_test.go index 5fc5c2558..67595876e 100644 --- a/internal/monitoring/monitor_unified_state_test.go +++ b/internal/monitoring/monitor_unified_state_test.go @@ -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 { diff --git a/internal/unifiedresources/metrics.go b/internal/unifiedresources/metrics.go index 3198b436c..c8ae33dc7 100644 --- a/internal/unifiedresources/metrics.go +++ b/internal/unifiedresources/metrics.go @@ -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) } diff --git a/internal/unifiedresources/metrics_test.go b/internal/unifiedresources/metrics_test.go index f893a1413..d50d33e14 100644 --- a/internal/unifiedresources/metrics_test.go +++ b/internal/unifiedresources/metrics_test.go @@ -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{ diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index ea10491bf..8cdb90588 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -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 diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index aac128b02..508317d33 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -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