diff --git a/docs/UNIFIED_AGENT.md b/docs/UNIFIED_AGENT.md index f3889468a5..f4110e4457 100644 --- a/docs/UNIFIED_AGENT.md +++ b/docs/UNIFIED_AGENT.md @@ -85,6 +85,7 @@ curl -fsSL http://:7655/install.sh | \ - **Docker Monitoring**: Container metrics, health checks, Swarm support (when enabled) - **Kubernetes Monitoring**: Cluster, node, pod, and deployment health (when enabled) - **Libvirt/KVM Monitoring**: Read-only VM inventory, state, vCPU, memory, and disk/network rates when the Linux host exposes `virsh` +- **XCP-ng Monitoring**: Read-only pool and VM inventory, power state, vCPU, and memory when an XCP-ng control domain exposes `xe` - **External Probes** (Pro): runs availability checks assigned to this agent from the Pulse server and reports the results back — see below - **Auto-Update**: Automatically updates when a new version is released - **Multi-Platform**: Linux, macOS, Windows support @@ -102,6 +103,18 @@ make its Container Station libvirt client/socket available to the agent service; the presence of KVM processes alone is not enough to establish a readable libvirt connection. +On XCP-ng, the host module automatically checks for the local `xe` CLI. It +uses only bounded `pool-list`, `host-list`, and `vm-list` queries: no XAPI +credentials or VM lifecycle authority are added. The XCP-ng pool becomes the +host's cluster grouping, and pool-wide VMs are de-duplicated by UUID and +parented to the resident Pulse host when that node also reports. If several +pool nodes run the Unified Agent, their identical pool views coalesce rather +than creating duplicate workloads. A failed `xe` query preserves the last +successful inventory while normal host metrics continue to report. + +This local integration covers one XCP-ng pool. Multi-pool deployments that +need a central Xen Orchestra connection remain a separate integration surface. + ## External Probes (Pro) With the Pro `external_probe` entitlement, availability checks configured in diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index e278e89c34..98b150a2e4 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -97,6 +97,7 @@ observation. 5d. `internal/kubernetesagent/agent.go` 5e. `internal/agentexec/verifier_postconditions.go` 5f. `pkg/agents/docker/report_limits.go` + 5g. `internal/hostagent/xcpng.go` 6. `cmd/pulse-agent/main.go` 7. `scripts/install.sh` 8. `scripts/install.ps1` @@ -1528,6 +1529,15 @@ the intentionally sparse public response. every guest disappear, while a successful empty inventory removes them. This path does not grant lifecycle, console, XML, snapshot, migration, or storage mutation authority. + XCP-ng inventory follows the same observation-only boundary. On Linux, + `internal/hostagent/xcpng.go` may auto-detect only the local `xe` client and + issue fixed, bounded `pool-list`, `host-list`, and `vm-list` requests under + one eight-second deadline. The report may carry one pool identity, the + reporting host's XCP UUID, and at most 1024 non-template, non-control-domain + VMs with UUID, power state, vCPU, configured/actual memory, and resident + host UUID. No server-authored xe argument, credential, lifecycle call, or + Xen Orchestra authority enters this path. Missing or invalid output omits + the new sample without failing normal host telemetry. Authenticated server ingest may reduce those typed samples into bounded host-level GPU utilization, VRAM pressure, and GPU temperature history on the existing agent identity, but that monitoring projection does not change diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 8d071ef45e..469e6ab1d4 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -428,6 +428,7 @@ changes. 60. `internal/config/host_continuity.go` 61. `internal/monitoring/docker_metadata_migration.go` 62. `internal/monitoring/kubernetes_metadata_migration.go` +62a. `internal/monitoring/monitor_xcpng.go` 63. `internal/monitoring/metadata_stores.go` 63a. `internal/config/docker_metadata.go` 63b. `internal/config/guest_metadata.go` @@ -915,6 +916,15 @@ authoritative removal. Domain metrics use unified-resource lookup. No libvirt lifecycle or configuration operation is part of the monitoring contract. +### XCP-ng xe inventory is a bounded pool observation + +Authenticated host reports may carry one normalized XCP-ng pool inventory +with at most 1024 VMs. Server ingest validates UUIDs and names, clamps vCPU and +memory values, de-duplicates VMs, and preserves the last successful inventory +when a later local `xe` query fails. A present empty VM list is authoritative. +No XAPI action, console, migration, snapshot, or configuration operation is +part of this monitoring contract. + ### Custom sensor evidence stays typed through host monitoring Authenticated host reports now copy bounded `sensors.custom` entries into the diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index da2eb72c39..bf7c8441de 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -126,6 +126,7 @@ about the same disk cannot diverge. 8. `internal/unifiedresources/metrics.go` 9. `internal/unifiedresources/metrics_targets.go` 10. `internal/unifiedresources/registry.go` +10a. `internal/unifiedresources/xcpng.go` 11. `internal/unifiedresources/resolve.go` 12. `internal/unifiedresources/resolve_context.go` 13. `internal/unifiedresources/resolved_host_set.go` @@ -1692,6 +1693,16 @@ positive maximum. The last successful libvirt collection time, rather than the host heartbeat, controls guest source freshness. This facet must not borrow Proxmox or VMware identifiers and carries no mutation capability. +### Agent XCP-ng pool inventories coalesce by platform UUID + +The registry projects XCP-ng guests through the same provider-neutral +`virtualMachine` facet with `technology=xcp-ng`. Pool-wide copies reported by +several agents use one `xcpng::vm:` source identity, so a +VM appears once. The resident XCP host UUID selects a real agent resource as +parent when that node reports, and the pool name supplies host cluster +grouping. Power state, vCPU, and bounded memory facts are observational only; +the adapter advertises no lifecycle capability or nonexistent history target. + ### Agent custom sensors remain typed, cloned host metadata `resourceFromHost` projects `models.HostSensorSummary.Custom` into diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index 1a8f082019..ced33f634c 100644 --- a/internal/hostagent/agent.go +++ b/internal/hostagent/agent.go @@ -1115,6 +1115,11 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { // an unavailable hypervisor cannot delay the rest of the host report. libvirtData := a.collectLibvirtInventory(ctx) + // XCP-ng exposes pool inventory through the local, read-only xe CLI. It has + // an independent bounded deadline so an unavailable XAPI cannot delay the + // normal host report. + xcpngData := a.collectXCPNGInventory(ctx) + // Collect S.M.A.R.T. disk data after topology owners. Enumeration and each // device probe have their own deadlines; a single shared 10-second budget // caused later disks on larger arrays to be cancelled before their probe. @@ -1151,6 +1156,14 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { UpdatedAt: libvirtData.CollectedAt, }) } + if xcpngData != nil { + moduleStatus = append(moduleStatus, agentshost.ModuleStatus{ + Name: "xcp-ng", + Enabled: true, + State: "running", + UpdatedAt: xcpngData.CollectedAt, + }) + } report := agentshost.Report{ Agent: agentshost.AgentInfo{ @@ -1197,6 +1210,7 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { Unraid: unraidData, Ceph: cephData, Libvirt: libvirtData, + XCPNG: xcpngData, ClusterSensors: clusterSensors, // Results stay queued until the primary destination accepts this // report, so a delivery failure retries them instead of losing them. diff --git a/internal/hostagent/xcpng.go b/internal/hostagent/xcpng.go new file mode 100644 index 0000000000..9374040443 --- /dev/null +++ b/internal/hostagent/xcpng.go @@ -0,0 +1,232 @@ +package hostagent + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "sort" + "strconv" + "strings" + "time" + "unicode/utf8" + + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" +) + +const ( + xcpngQueryTimeout = 8 * time.Second + xcpngMaxPoolOutputBytes = 64 * 1024 + xcpngMaxHostOutputBytes = 256 * 1024 + xcpngMaxVMOutputBytes = 1024 * 1024 + xcpngMaxVMs = 1024 + xcpngMaxNameBytes = 256 +) + +var ( + xcpngPoolListArgs = []string{"pool-list", "params=uuid,name-label,master"} + xcpngHostListArgs = []string{"host-list", "params=uuid,name-label,hostname"} + xcpngVMListArgs = []string{ + "vm-list", + "is-control-domain=false", + "is-a-template=false", + "is-a-snapshot=false", + "params=uuid,name-label,power-state,VCPUs-number,memory-actual,memory-static-max,resident-on", + } +) + +// collectXCPNGInventory auto-detects the pool-wide, read-only xe CLI exposed +// by an XCP-ng control domain. Hosts without xe continue reporting normally. +func (a *Agent) collectXCPNGInventory(ctx context.Context) *agentshost.XCPNGInventory { + if a.collector.GOOS() != "linux" { + return nil + } + path, err := a.collector.LookPath("xe") + if err != nil { + if !errors.Is(err, exec.ErrNotFound) && !os.IsNotExist(err) { + a.logger.Debug().Err(err).Msg("Failed to locate xe for XCP-ng inventory") + } + return nil + } + + queryCtx, cancel := context.WithTimeout(ctx, xcpngQueryTimeout) + defer cancel() + + poolOutput, err := collectCommandOutputLimited( + queryCtx, a.collector, xcpngMaxPoolOutputBytes, path, xcpngPoolListArgs..., + ) + if err != nil { + a.logger.Debug().Err(err).Msg("Failed to list XCP-ng pools") + return nil + } + pools, err := parseXERecords(poolOutput, xcpngMaxPoolOutputBytes, 2) + if err != nil || len(pools) != 1 { + a.logger.Debug().Err(err).Int("pools", len(pools)).Msg("Failed to parse one XCP-ng pool") + return nil + } + poolUUID := canonicalXCPNGUUID(pools[0]["uuid"]) + if poolUUID == "" { + return nil + } + + hostOutput, err := collectCommandOutputLimited( + queryCtx, a.collector, xcpngMaxHostOutputBytes, path, xcpngHostListArgs..., + ) + if err != nil { + a.logger.Debug().Err(err).Msg("Failed to list XCP-ng hosts") + return nil + } + hosts, err := parseXERecords(hostOutput, xcpngMaxHostOutputBytes, 64) + if err != nil { + a.logger.Debug().Err(err).Msg("Failed to parse XCP-ng hosts") + return nil + } + + vmOutput, err := collectCommandOutputLimited( + queryCtx, a.collector, xcpngMaxVMOutputBytes, path, xcpngVMListArgs..., + ) + if err != nil { + a.logger.Debug().Err(err).Msg("Failed to list XCP-ng virtual machines") + return nil + } + records, err := parseXERecords(vmOutput, xcpngMaxVMOutputBytes, xcpngMaxVMs) + if err != nil { + a.logger.Debug().Err(err).Msg("Failed to parse XCP-ng virtual machines") + return nil + } + + inventory := &agentshost.XCPNGInventory{ + PoolUUID: poolUUID, + PoolName: safeXCPNGName(pools[0]["name-label"]), + MasterUUID: canonicalXCPNGUUID(pools[0]["master"]), + LocalHostUUID: localXCPNGHostUUID(hosts, a.hostname), + VMs: make([]agentshost.XCPNGVM, 0, len(records)), + CollectedAt: a.collector.Now().UTC(), + } + seen := make(map[string]struct{}, len(records)) + for _, record := range records { + uuid := canonicalXCPNGUUID(record["uuid"]) + name := safeXCPNGName(record["name-label"]) + if uuid == "" || name == "" { + continue + } + if _, exists := seen[uuid]; exists { + continue + } + seen[uuid] = struct{}{} + vcpus, _ := strconv.Atoi(strings.TrimSpace(record["vcpus-number"])) + memory, _ := strconv.ParseInt(strings.TrimSpace(record["memory-actual"]), 10, 64) + memoryMax, _ := strconv.ParseInt(strings.TrimSpace(record["memory-static-max"]), 10, 64) + inventory.VMs = append(inventory.VMs, agentshost.XCPNGVM{ + UUID: uuid, + Name: name, + PowerState: normalizeXCPNGPowerState(record["power-state"]), + VCPUs: max(0, min(vcpus, 4096)), + MemoryActual: max(int64(0), memory), + MemoryStaticMax: max(int64(0), memoryMax), + ResidentHostUUID: canonicalXCPNGUUID(record["resident-on"]), + }) + } + sort.Slice(inventory.VMs, func(i, j int) bool { + return strings.ToLower(inventory.VMs[i].Name) < strings.ToLower(inventory.VMs[j].Name) + }) + return inventory +} + +func parseXERecords(output string, maxBytes, maxRecords int) ([]map[string]string, error) { + if len(output) > maxBytes { + return nil, fmt.Errorf("xe output exceeds %d bytes", maxBytes) + } + records := make([]map[string]string, 0) + current := make(map[string]string) + flush := func() { + if len(current) == 0 || len(records) == maxRecords { + current = make(map[string]string) + return + } + records = append(records, current) + current = make(map[string]string) + } + for _, raw := range strings.Split(output, "\n") { + line := strings.TrimSpace(strings.TrimSuffix(raw, "\r")) + if line == "" { + flush() + continue + } + closeParen := strings.IndexByte(line, ')') + if closeParen < 0 { + continue + } + colon := strings.IndexByte(line[closeParen+1:], ':') + if colon < 0 { + continue + } + colon += closeParen + 1 + key := strings.ToLower(strings.TrimSpace(strings.SplitN(line[:closeParen], "(", 2)[0])) + value := strings.TrimSpace(line[colon+1:]) + if key == "" || len(key) > 64 || len(value) > xcpngMaxNameBytes*4 || !utf8.ValidString(value) { + continue + } + current[key] = value + } + flush() + return records, nil +} + +func localXCPNGHostUUID(hosts []map[string]string, hostname string) string { + hostname = strings.TrimSpace(hostname) + short := strings.SplitN(hostname, ".", 2)[0] + for _, host := range hosts { + for _, candidate := range []string{host["hostname"], host["name-label"]} { + candidate = strings.TrimSpace(candidate) + candidateShort := strings.SplitN(candidate, ".", 2)[0] + if candidate != "" && + (strings.EqualFold(candidate, hostname) || strings.EqualFold(candidateShort, short)) { + return canonicalXCPNGUUID(host["uuid"]) + } + } + } + if len(hosts) == 1 { + return canonicalXCPNGUUID(hosts[0]["uuid"]) + } + return "" +} + +func canonicalXCPNGUUID(value string) string { + value = strings.ToLower(strings.TrimSpace(value)) + if len(value) != 36 || value[8] != '-' || value[13] != '-' || value[18] != '-' || value[23] != '-' { + return "" + } + for i, r := range value { + if i == 8 || i == 13 || i == 18 || i == 23 { + continue + } + if !((r >= '0' && r <= '9') || (r >= 'a' && r <= 'f')) { + return "" + } + } + return value +} + +func safeXCPNGName(value string) string { + value = strings.TrimSpace(value) + if value == "" || len(value) > xcpngMaxNameBytes || !utf8.ValidString(value) { + return "" + } + for _, r := range value { + if r < 0x20 || r == 0x7f { + return "" + } + } + return value +} + +func normalizeXCPNGPowerState(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "running", "halted", "paused", "suspended": + return strings.ToLower(strings.TrimSpace(value)) + default: + return "unknown" + } +} diff --git a/internal/hostagent/xcpng_test.go b/internal/hostagent/xcpng_test.go new file mode 100644 index 0000000000..0e478be8f9 --- /dev/null +++ b/internal/hostagent/xcpng_test.go @@ -0,0 +1,111 @@ +package hostagent + +import ( + "context" + "os" + "slices" + "strings" + "testing" + "time" + + "github.com/rs/zerolog" +) + +func TestCollectXCPNGInventoryUsesBoundedReadOnlyLists(t *testing.T) { + const xePath = "/opt/xensource/bin/xe" + now := time.Date(2026, 7, 30, 15, 0, 0, 0, time.UTC) + var calls [][]string + collector := &mockCollector{ + goos: "linux", + nowFn: func() time.Time { return now }, + lookPathFn: func(file string) (string, error) { + if file == "xe" { + return xePath, nil + } + return "", os.ErrNotExist + }, + commandCombinedOutputFn: func(_ context.Context, name string, args ...string) (string, error) { + if name != xePath { + t.Fatalf("command = %q, want %q", name, xePath) + } + calls = append(calls, append([]string(nil), args...)) + switch { + case slices.Equal(args, xcpngPoolListArgs): + return `uuid ( RO) : 11111111-1111-1111-1111-111111111111 + name-label ( RW): Lab Pool + master ( RO): aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa +`, nil + case slices.Equal(args, xcpngHostListArgs): + return `uuid ( RO) : aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa + name-label ( RW): xcp-one + hostname ( RO): xcp-one.local + +uuid ( RO) : bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb + name-label ( RW): xcp-two + hostname ( RO): xcp-two.local +`, nil + case slices.Equal(args, xcpngVMListArgs): + return `uuid ( RO) : cccccccc-cccc-cccc-cccc-cccccccccccc + name-label ( RW): Database + power-state ( RO): running + VCPUs-number ( RO): 4 + memory-actual ( RO): 4294967296 + memory-static-max ( RW): 8589934592 + resident-on ( RO): aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa + +uuid ( RO) : dddddddd-dddd-dddd-dddd-dddddddddddd + name-label ( RW): Worker + power-state ( RO): halted + VCPUs-number ( RO): 2 + memory-actual ( RO): 2147483648 + memory-static-max ( RW): 4294967296 + resident-on ( RO): bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb +`, nil + default: + t.Fatalf("unexpected xe args: %#v", args) + return "", nil + } + }, + } + agent := &Agent{ + logger: zerolog.Nop(), + collector: collector, + hostname: "xcp-one.local", + } + + got := agent.collectXCPNGInventory(context.Background()) + if got == nil || len(got.VMs) != 2 || len(calls) != 3 { + t.Fatalf("inventory = %+v, calls = %#v", got, calls) + } + if got.PoolName != "Lab Pool" || + got.LocalHostUUID != "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" || + !got.CollectedAt.Equal(now) { + t.Fatalf("inventory metadata = %+v", got) + } + if got.VMs[0].Name != "Database" || + got.VMs[0].PowerState != "running" || + got.VMs[0].VCPUs != 4 || + got.VMs[0].MemoryActual != 4<<30 || + got.VMs[0].MemoryStaticMax != 8<<30 { + t.Fatalf("database VM = %+v", got.VMs[0]) + } +} + +func TestXCPNGParsingRejectsUnsafeAndOversizedInput(t *testing.T) { + records, err := parseXERecords( + "uuid ( RO): ok\nignored line\nname-label ( RW): safe\n\n", + 1024, + 4, + ) + if err != nil || len(records) != 1 || records[0]["name-label"] != "safe" { + t.Fatalf("records = %#v, err = %v", records, err) + } + if _, err := parseXERecords(strings.Repeat("x", 1025), 1024, 1); err == nil { + t.Fatal("expected oversized xe output to fail") + } + if canonicalXCPNGUUID("not-a-uuid") != "" || + safeXCPNGName("bad\x00name") != "" || + normalizeXCPNGPowerState("migrating") != "unknown" { + t.Fatal("unsafe XCP-ng values were accepted") + } +} diff --git a/internal/models/deepcopy.go b/internal/models/deepcopy.go index fd7949c7f4..55a4a595ea 100644 --- a/internal/models/deepcopy.go +++ b/internal/models/deepcopy.go @@ -403,6 +403,7 @@ func cloneHost(src Host) Host { dest.Unraid = cloneHostUnraidStorage(src.Unraid) dest.Ceph = cloneHostCephCluster(src.Ceph) dest.Libvirt = cloneHostLibvirtInventory(src.Libvirt) + dest.XCPNG = cloneHostXCPNGInventory(src.XCPNG) dest.PackageUpdates = cloneHostPackageUpdateStatus(src.PackageUpdates) dest.StorageCleanup = cloneHostStorageCleanupStatus(src.StorageCleanup) dest.TokenLastUsedAt = cloneTimePtr(src.TokenLastUsedAt) @@ -421,6 +422,15 @@ func cloneHostLibvirtInventory(src *HostLibvirtInventory) *HostLibvirtInventory return &dest } +func cloneHostXCPNGInventory(src *HostXCPNGInventory) *HostXCPNGInventory { + if src == nil { + return nil + } + dest := *src + dest.VMs = append([]HostXCPNGVM(nil), src.VMs...) + return &dest +} + func cloneHostIdentityConflict(src *HostIdentityConflict) *HostIdentityConflict { if src == nil { return nil diff --git a/internal/models/deepcopy_test.go b/internal/models/deepcopy_test.go index 486c608778..896c420b43 100644 --- a/internal/models/deepcopy_test.go +++ b/internal/models/deepcopy_test.go @@ -437,3 +437,18 @@ func TestCloneHostLibvirtInventoryIsolation(t *testing.T) { t.Fatal("clone libvirt inventory aliases source domains") } } + +func TestCloneHostXCPNGInventoryIsolation(t *testing.T) { + src := Host{XCPNG: &HostXCPNGInventory{ + PoolUUID: "11111111-1111-1111-1111-111111111111", + VMs: []HostXCPNGVM{{UUID: "vm-a", Name: "app"}}, + }} + dst := cloneHost(src) + if dst.XCPNG == nil || len(dst.XCPNG.VMs) != 1 { + t.Fatalf("cloned XCP-ng inventory = %+v", dst.XCPNG) + } + dst.XCPNG.VMs[0].Name = "mutated" + if src.XCPNG.VMs[0].Name != "app" { + t.Fatal("cloned XCP-ng inventory aliases source VMs") + } +} diff --git a/internal/models/models.go b/internal/models/models.go index ce05a9460e..47d09d2b21 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -305,6 +305,7 @@ type Host struct { Unraid *HostUnraidStorage `json:"unraid,omitempty"` Ceph *HostCephCluster `json:"ceph,omitempty"` Libvirt *HostLibvirtInventory `json:"libvirt,omitempty"` + XCPNG *HostXCPNGInventory `json:"xcpng,omitempty"` Status string `json:"status"` UptimeSeconds int64 `json:"uptimeSeconds,omitempty"` IntervalSeconds int `json:"intervalSeconds,omitempty"` @@ -443,6 +444,10 @@ func (h Host) NormalizeCollections() Host { libvirt := h.Libvirt.NormalizeCollections() h.Libvirt = &libvirt } + if h.XCPNG != nil { + xcpng := h.XCPNG.NormalizeCollections() + h.XCPNG = &xcpng + } if h.Tags == nil { h.Tags = []string{} } @@ -493,6 +498,34 @@ type HostLibvirtDomain struct { IORatesValid bool `json:"-"` } +// HostXCPNGInventory is the normalized server-side view of the XCP-ng pool +// visible to a Unified Agent running in an XCP-ng control domain. +type HostXCPNGInventory struct { + PoolUUID string `json:"poolUuid"` + PoolName string `json:"poolName,omitempty"` + MasterUUID string `json:"masterUuid,omitempty"` + LocalHostUUID string `json:"localHostUuid,omitempty"` + VMs []HostXCPNGVM `json:"vms"` + CollectedAt time.Time `json:"collectedAt"` +} + +func (i HostXCPNGInventory) NormalizeCollections() HostXCPNGInventory { + if i.VMs == nil { + i.VMs = []HostXCPNGVM{} + } + return i +} + +type HostXCPNGVM struct { + UUID string `json:"uuid"` + Name string `json:"name"` + PowerState string `json:"powerState"` + VCPUs int `json:"vcpus,omitempty"` + MemoryActual int64 `json:"memoryActualBytes,omitempty"` + MemoryStaticMax int64 `json:"memoryStaticMaxBytes,omitempty"` + ResidentHostUUID string `json:"residentHostUuid,omitempty"` +} + // HostIdentityConflict records evidence that reports from more than one // physical machine are being folded into a single host agent record. The // usual cause is template-deployed machines that still share the same diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 6657f25eeb..429c69d906 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -2734,6 +2734,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. previousHostModel, observedAt, ) + xcpngData := normalizeAgentXCPNGInventory(report.XCPNG, previousHostModel, observedAt) agentUpdate := mergeAgentUpdateStatus( previousHostAgentUpdate(m.state.GetHosts(), identifier), @@ -2786,6 +2787,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. Unraid: unraidData, Ceph: cephData, Libvirt: libvirtData, + XCPNG: xcpngData, Status: "online", UptimeSeconds: report.Host.UptimeSeconds, IntervalSeconds: report.Agent.IntervalSeconds, diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 1b82667f31..5308d6a380 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -4678,3 +4678,46 @@ func TestIssue1654LivePreexistingAgentIsNotSuperseded(t *testing.T) { t.Fatalf("host count = %d, want both live identities preserved", got) } } + +func TestNormalizeAgentXCPNGInventoryBoundsAndPreservesFailures(t *testing.T) { + now := time.Date(2026, 7, 30, 15, 0, 0, 0, time.UTC) + report := &agentshost.XCPNGInventory{ + PoolUUID: "11111111-1111-1111-1111-111111111111", + PoolName: " Lab Pool ", + LocalHostUUID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + VMs: []agentshost.XCPNGVM{ + { + UUID: "cccccccc-cccc-cccc-cccc-cccccccccccc", + Name: " Database ", + PowerState: "RUNNING", + VCPUs: 5000, + MemoryActual: 10 << 30, + MemoryStaticMax: 8 << 30, + ResidentHostUUID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + }, + {UUID: "cccccccc-cccc-cccc-cccc-cccccccccccc", Name: "duplicate"}, + {UUID: "bad", Name: "invalid"}, + }, + } + got := normalizeAgentXCPNGInventory(report, nil, now) + if got == nil || len(got.VMs) != 1 { + t.Fatalf("inventory = %+v", got) + } + if got.PoolName != "Lab Pool" || + got.VMs[0].Name != "Database" || + got.VMs[0].PowerState != "running" || + got.VMs[0].VCPUs != 4096 || + got.VMs[0].MemoryActual != 8<<30 { + t.Fatalf("normalized inventory = %+v", got) + } + + previous := &models.Host{XCPNG: got} + preserved := normalizeAgentXCPNGInventory(nil, previous, now.Add(time.Minute)) + if preserved == nil || len(preserved.VMs) != 1 || !preserved.CollectedAt.Equal(now) { + t.Fatalf("preserved inventory = %+v", preserved) + } + preserved.VMs[0].Name = "mutated" + if previous.XCPNG.VMs[0].Name != "Database" { + t.Fatal("preserved inventory aliases previous host state") + } +} diff --git a/internal/monitoring/monitor_xcpng.go b/internal/monitoring/monitor_xcpng.go new file mode 100644 index 0000000000..bc3d5c73d8 --- /dev/null +++ b/internal/monitoring/monitor_xcpng.go @@ -0,0 +1,110 @@ +package monitoring + +import ( + "sort" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" +) + +const maxAgentXCPNGVMs = 1024 + +func normalizeAgentXCPNGInventory( + report *agentshost.XCPNGInventory, + previous *models.Host, + observedAt time.Time, +) *models.HostXCPNGInventory { + if report == nil { + if previous != nil && previous.XCPNG != nil { + preserved := *previous.XCPNG + preserved.VMs = append([]models.HostXCPNGVM(nil), previous.XCPNG.VMs...) + return &preserved + } + return nil + } + + poolUUID := canonicalAgentXCPNGUUID(report.PoolUUID) + if poolUUID == "" { + return nil + } + result := &models.HostXCPNGInventory{ + PoolUUID: poolUUID, + PoolName: safeAgentXCPNGName(report.PoolName), + MasterUUID: canonicalAgentXCPNGUUID(report.MasterUUID), + LocalHostUUID: canonicalAgentXCPNGUUID(report.LocalHostUUID), + VMs: make([]models.HostXCPNGVM, 0, min(len(report.VMs), maxAgentXCPNGVMs)), + CollectedAt: observedAt.UTC(), + } + seen := make(map[string]struct{}, len(report.VMs)) + for _, incoming := range report.VMs { + if len(result.VMs) == maxAgentXCPNGVMs { + break + } + uuid := canonicalAgentXCPNGUUID(incoming.UUID) + name := safeAgentXCPNGName(incoming.Name) + if uuid == "" || name == "" { + continue + } + if _, exists := seen[uuid]; exists { + continue + } + seen[uuid] = struct{}{} + result.VMs = append(result.VMs, models.HostXCPNGVM{ + UUID: uuid, + Name: name, + PowerState: normalizeAgentXCPNGPowerState(incoming.PowerState), + VCPUs: max(0, min(incoming.VCPUs, 4096)), + MemoryActual: max(int64(0), incoming.MemoryActual), + MemoryStaticMax: max(int64(0), incoming.MemoryStaticMax), + ResidentHostUUID: canonicalAgentXCPNGUUID(incoming.ResidentHostUUID), + }) + last := &result.VMs[len(result.VMs)-1] + if last.MemoryStaticMax > 0 && last.MemoryActual > last.MemoryStaticMax { + last.MemoryActual = last.MemoryStaticMax + } + } + sort.Slice(result.VMs, func(i, j int) bool { + return strings.ToLower(result.VMs[i].Name) < strings.ToLower(result.VMs[j].Name) + }) + return result +} + +func canonicalAgentXCPNGUUID(value string) string { + value = strings.ToLower(strings.TrimSpace(value)) + if len(value) != 36 || value[8] != '-' || value[13] != '-' || value[18] != '-' || value[23] != '-' { + return "" + } + for i, r := range value { + if i == 8 || i == 13 || i == 18 || i == 23 { + continue + } + if !((r >= '0' && r <= '9') || (r >= 'a' && r <= 'f')) { + return "" + } + } + return value +} + +func safeAgentXCPNGName(value string) string { + value = strings.TrimSpace(value) + if value == "" || len(value) > 256 { + return "" + } + for _, r := range value { + if r < 0x20 || r == 0x7f { + return "" + } + } + return value +} + +func normalizeAgentXCPNGPowerState(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "running", "halted", "paused", "suspended": + return strings.ToLower(strings.TrimSpace(value)) + default: + return "unknown" + } +} diff --git a/internal/unifiedresources/adapters.go b/internal/unifiedresources/adapters.go index 572e0588ee..1f5391bb75 100644 --- a/internal/unifiedresources/adapters.go +++ b/internal/unifiedresources/adapters.go @@ -161,6 +161,13 @@ func resourceFromHost(host models.Host) (Resource, ResourceIdentity) { identity := identityFromHost(host) hostProfile := agentHostProfileForHost(host) platform := agentRuntimePlatformForHost(host, hostProfile) + if host.XCPNG != nil { + platform = "xcp-ng" + identity.ClusterName = firstNonEmpty( + strings.TrimSpace(host.XCPNG.PoolName), + strings.TrimSpace(host.XCPNG.PoolUUID), + ) + } agent := &AgentData{ AgentID: host.ID, @@ -556,13 +563,26 @@ func resourceFromHost(host models.Host) (Resource, ResourceIdentity) { Uptime: host.UptimeSeconds, Temperature: agent.Temperature, Agent: agent, - Tags: host.Tags, + Tags: appendXCPNGTag(host.Tags, host.XCPNG != nil), Capabilities: hostActionCapabilities(host), } return resource, identity } +func appendXCPNGTag(tags []string, enabled bool) []string { + result := append([]string(nil), tags...) + if !enabled { + return result + } + for _, tag := range result { + if strings.EqualFold(strings.TrimSpace(tag), "xcp-ng") { + return result + } + } + return append(result, "xcp-ng") +} + const hostPackageUpdateHandler = "host.package_updates" const hostStorageCleanupHandler = "host.storage_cleanup" diff --git a/internal/unifiedresources/adapters_test.go b/internal/unifiedresources/adapters_test.go index 563863ca6b..1fc6e28da0 100644 --- a/internal/unifiedresources/adapters_test.go +++ b/internal/unifiedresources/adapters_test.go @@ -41,6 +41,29 @@ func TestResourceFromProxmoxNodeIncludesTemperature(t *testing.T) { } } +func TestResourceFromHostProjectsXCPNGPoolIdentity(t *testing.T) { + host := models.Host{ + ID: "agent-one", + Hostname: "xcp-one", + Platform: "linux", + Status: "online", + LastSeen: time.Now().UTC(), + Tags: []string{"lab"}, + XCPNG: &models.HostXCPNGInventory{ + PoolUUID: "11111111-1111-1111-1111-111111111111", + PoolName: "Lab Pool", + }, + } + resource, identity := resourceFromHost(host) + if resource.Technology != "xcp-ng" || + resource.Agent == nil || + resource.Agent.Platform != "xcp-ng" || + identity.ClusterName != "Lab Pool" || + !reflect.DeepEqual(resource.Tags, []string{"lab", "xcp-ng"}) { + t.Fatalf("XCP-ng host projection = resource %+v identity %+v", resource, identity) + } +} + func TestResourceFromHostPreservesCustomSensorMeta(t *testing.T) { value := 12.5 observedAt := time.Date(2026, 7, 30, 19, 0, 0, 0, time.UTC) diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index ed945d3db6..94acab9239 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -198,6 +198,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh rr.ingestHost(host) rr.ingestHostLibvirtDomains(host) } + rr.ingestHostXCPNGVMs(snapshot.Hosts) for _, host := range snapshot.Hosts { rr.ingestHostUnraidStorage(host) } diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index 757993b6fe..6fc26cf50a 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -66,6 +66,99 @@ func TestMemoryStore_RecordActionAuditAppliesRedaction(t *testing.T) { } } +func TestRegistryProjectsAndDeduplicatesXCPNGPoolVMs(t *testing.T) { + now := time.Now().UTC() + poolUUID := "11111111-1111-1111-1111-111111111111" + hostOneUUID := "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" + hostTwoUUID := "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" + vm := models.HostXCPNGVM{ + UUID: "cccccccc-cccc-cccc-cccc-cccccccccccc", + Name: "database", + PowerState: "running", + VCPUs: 4, + MemoryActual: 4 << 30, + MemoryStaticMax: 8 << 30, + ResidentHostUUID: hostTwoUUID, + } + hosts := []models.Host{ + { + ID: "agent-one", + Hostname: "xcp-one", + Platform: "linux", + Status: "online", + LastSeen: now, + XCPNG: &models.HostXCPNGInventory{ + PoolUUID: poolUUID, + PoolName: "Lab Pool", + LocalHostUUID: hostOneUUID, + VMs: []models.HostXCPNGVM{vm}, + CollectedAt: now, + }, + }, + { + ID: "agent-two", + Hostname: "xcp-two", + Platform: "linux", + Status: "online", + LastSeen: now.Add(time.Second), + XCPNG: &models.HostXCPNGInventory{ + PoolUUID: poolUUID, + PoolName: "Lab Pool", + LocalHostUUID: hostTwoUUID, + VMs: []models.HostXCPNGVM{vm}, + CollectedAt: now.Add(time.Second), + }, + }, + } + + registry := NewRegistry(nil) + registry.IngestSnapshot(models.StateSnapshot{Hosts: hosts}) + vms := registry.ListByType(ResourceTypeVM) + if len(vms) != 1 { + t.Fatalf("XCP-ng VMs = %d, want one de-duplicated VM: %+v", len(vms), vms) + } + projected := vms[0] + if projected.Technology != "xcp-ng" || + projected.Status != StatusOnline || + projected.VirtualMachine == nil || + projected.VirtualMachine.Hypervisor != "xcp-ng" || + projected.VirtualMachine.VCPUs != 4 { + t.Fatalf("XCP-ng VM = %+v", projected) + } + if projected.Metrics == nil || + projected.Metrics.Memory == nil || + projected.Metrics.Memory.Percent != 50 { + t.Fatalf("XCP-ng VM memory = %+v", projected.Metrics) + } + if projected.ParentID == nil { + t.Fatal("XCP-ng VM has no resident parent") + } + parent, ok := registry.Get(*projected.ParentID) + if !ok || parent.Name != "xcp-two" || parent.Technology != "xcp-ng" { + t.Fatalf("resident parent = %+v, found=%v", parent, ok) + } + if parent.Identity.ClusterName != "Lab Pool" { + t.Fatalf("parent cluster identity = %+v", parent.Identity) + } + if BuildMetricsTargetForRegistry(registry, projected.ID) != nil { + t.Fatal("inventory-only XCP-ng VM must not advertise nonexistent metrics history") + } +} + +func TestXCPNGResourceStatusMapping(t *testing.T) { + for state, want := range map[string]ResourceStatus{ + "running": StatusOnline, + "halted": StatusOffline, + "paused": StatusWarning, + "suspended": StatusWarning, + "unknown": StatusUnknown, + } { + if got := xcpngResourceStatus(state); got != want { + t.Errorf("xcpngResourceStatus(%q) = %q, want %q", state, got, want) + } + } +} + func TestResourceRegistry_GetByReferenceResolvesSourceIDAndCanonicalAlias(t *testing.T) { rr := NewRegistry(nil) rr.IngestRecords(SourceProxmox, []IngestRecord{{ diff --git a/internal/unifiedresources/xcpng.go b/internal/unifiedresources/xcpng.go new file mode 100644 index 0000000000..8848232e57 --- /dev/null +++ b/internal/unifiedresources/xcpng.go @@ -0,0 +1,110 @@ +package unifiedresources + +import ( + "math" + "sort" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +func xcpngVMSourceID(poolUUID, vmUUID string) string { + return "xcpng:" + strings.TrimSpace(poolUUID) + ":vm:" + strings.TrimSpace(vmUUID) +} + +// ingestHostXCPNGVMs coalesces identical pool-wide xe inventories reported by +// several real host agents. The newest sighting wins and each VM is parented +// to the agent whose local XCP-ng host UUID matches resident-on. +func (rr *ResourceRegistry) ingestHostXCPNGVMs(hosts []models.Host) { + parentByHostUUID := make(map[string]string) + for _, host := range hosts { + if host.XCPNG == nil || strings.TrimSpace(host.XCPNG.LocalHostUUID) == "" { + continue + } + if parentID := rr.sourceResourceID(SourceAgent, host.ID); parentID != "" { + parentByHostUUID[strings.TrimSpace(host.XCPNG.LocalHostUUID)] = parentID + } + } + + type candidate struct { + host models.Host + vm models.HostXCPNGVM + } + bySourceID := make(map[string]candidate) + for _, host := range hosts { + if host.XCPNG == nil { + continue + } + for _, vm := range host.XCPNG.VMs { + sourceID := xcpngVMSourceID(host.XCPNG.PoolUUID, vm.UUID) + if strings.TrimSpace(host.XCPNG.PoolUUID) == "" || strings.TrimSpace(vm.UUID) == "" { + continue + } + existing, ok := bySourceID[sourceID] + if !ok || host.XCPNG.CollectedAt.After(existing.host.XCPNG.CollectedAt) { + bySourceID[sourceID] = candidate{host: host, vm: vm} + } + } + } + sourceIDs := make([]string, 0, len(bySourceID)) + for sourceID := range bySourceID { + sourceIDs = append(sourceIDs, sourceID) + } + sort.Strings(sourceIDs) + for _, sourceID := range sourceIDs { + item := bySourceID[sourceID] + resource, identity := resourceFromHostXCPNGVM(item.host, item.vm) + if parentID := parentByHostUUID[strings.TrimSpace(item.vm.ResidentHostUUID)]; parentID != "" { + resource.ParentID = &parentID + } + rr.ingest(SourceAgent, sourceID, resource, identity) + } +} + +func resourceFromHostXCPNGVM(host models.Host, vm models.HostXCPNGVM) (Resource, ResourceIdentity) { + lastSeen := host.LastSeen + if host.XCPNG != nil && !host.XCPNG.CollectedAt.IsZero() { + lastSeen = host.XCPNG.CollectedAt + } + var metrics *ResourceMetrics + if vm.MemoryStaticMax > 0 { + used := max(int64(0), min(vm.MemoryActual, vm.MemoryStaticMax)) + total := vm.MemoryStaticMax + metrics = &ResourceMetrics{Memory: &MetricValue{ + Used: &used, + Total: &total, + Percent: math.Max(0, math.Min(100, float64(used)/float64(total)*100)), + Unit: "bytes", + Source: SourceAgent, + }} + } + return Resource{ + Type: ResourceTypeVM, + Technology: "xcp-ng", + Name: strings.TrimSpace(vm.Name), + Status: xcpngResourceStatus(vm.PowerState), + LastSeen: lastSeen, + UpdatedAt: time.Now().UTC(), + Metrics: metrics, + Tags: []string{"xcp-ng", "xen"}, + VirtualMachine: &VirtualMachineData{ + RuntimeState: strings.TrimSpace(vm.PowerState), + Hypervisor: "xcp-ng", + VCPUs: vm.VCPUs, + }, + }, ResourceIdentity{Hostnames: uniqueStrings([]string{vm.Name})} +} + +func xcpngResourceStatus(state string) ResourceStatus { + switch strings.ToLower(strings.TrimSpace(state)) { + case "running": + return StatusOnline + case "halted": + return StatusOffline + case "paused", "suspended": + return StatusWarning + default: + return StatusUnknown + } +} diff --git a/pkg/agents/host/report.go b/pkg/agents/host/report.go index a29e7eaa2f..107f05d937 100644 --- a/pkg/agents/host/report.go +++ b/pkg/agents/host/report.go @@ -54,6 +54,7 @@ type Report struct { Unraid *UnraidStorage `json:"unraid,omitempty"` Ceph *CephCluster `json:"ceph,omitempty"` Libvirt *LibvirtInventory `json:"libvirt,omitempty"` + XCPNG *XCPNGInventory `json:"xcpng,omitempty"` ClusterSensors []ClusterNodeSensors `json:"clusterSensors,omitempty"` // AvailabilityResults carries availability checks the agent executed on // behalf of the server for targets assigned to it. The server owns failure @@ -88,6 +89,29 @@ type LibvirtDomain struct { DiskWriteBytes uint64 `json:"diskWriteBytes,omitempty"` } +// XCPNGInventory is the bounded, read-only pool inventory collected through +// the xe CLI available in an XCP-ng control domain. +type XCPNGInventory struct { + PoolUUID string `json:"poolUuid"` + PoolName string `json:"poolName,omitempty"` + MasterUUID string `json:"masterUuid,omitempty"` + LocalHostUUID string `json:"localHostUuid,omitempty"` + VMs []XCPNGVM `json:"vms"` + CollectedAt time.Time `json:"collectedAt"` +} + +// XCPNGVM contains stable inventory fields returned by xe vm-list. Collection +// is intentionally observational; no XAPI lifecycle operation is exposed. +type XCPNGVM struct { + UUID string `json:"uuid"` + Name string `json:"name"` + PowerState string `json:"powerState"` + VCPUs int `json:"vcpus,omitempty"` + MemoryActual int64 `json:"memoryActualBytes,omitempty"` + MemoryStaticMax int64 `json:"memoryStaticMaxBytes,omitempty"` + ResidentHostUUID string `json:"residentHostUuid,omitempty"` +} + // AvailabilityProbeResult is one completed availability check reported by a // remote probe agent. Outcome mirrors the shared probe vocabulary // ("reachable", "unreachable", "indeterminate"); anything else is treated as diff --git a/pkg/agents/host/report_test.go b/pkg/agents/host/report_test.go index 0ba245267d..57b61b1130 100644 --- a/pkg/agents/host/report_test.go +++ b/pkg/agents/host/report_test.go @@ -685,3 +685,38 @@ func TestReportLibvirtInventoryJSONRoundTrip(t *testing.T) { t.Fatalf("nil libvirt inventory should be omitted: %s", bare) } } + +func TestReportXCPNGInventoryJSONRoundTrip(t *testing.T) { + collectedAt := time.Date(2026, 7, 30, 15, 0, 0, 0, time.UTC) + report := Report{XCPNG: &XCPNGInventory{ + PoolUUID: "11111111-1111-1111-1111-111111111111", + PoolName: "Lab Pool", + LocalHostUUID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + CollectedAt: collectedAt, + VMs: []XCPNGVM{{ + UUID: "cccccccc-cccc-cccc-cccc-cccccccccccc", + Name: "database", + PowerState: "running", + VCPUs: 4, + MemoryActual: 4 << 30, + MemoryStaticMax: 8 << 30, + ResidentHostUUID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + }}, + }} + data, err := json.Marshal(report) + if err != nil { + t.Fatalf("marshal report: %v", err) + } + var decoded Report + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("unmarshal report: %v", err) + } + if decoded.XCPNG == nil || + decoded.XCPNG.PoolName != "Lab Pool" || + !decoded.XCPNG.CollectedAt.Equal(collectedAt) || + len(decoded.XCPNG.VMs) != 1 || + decoded.XCPNG.VMs[0].MemoryActual != 4<<30 || + decoded.XCPNG.VMs[0].MemoryStaticMax != 8<<30 { + t.Fatalf("XCP-ng inventory = %+v", decoded.XCPNG) + } +}