diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 796fc46ca0..d7b496972e 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2676,7 +2676,16 @@ truthfulness, not native thaw, containing-release or workload acceptance. ingestion so canonical host IDs stay stable across restarts (see the unified-resources contract's durable identity-pin obligation). Rebuild paths added to the adapter must keep that persistence step; ephemeral - snapshot-bridge adapters stay read-only. + snapshot-bridge adapters stay read-only. The rebuild, the live + supplemental refresh and the read-state overlay ingest records with the + adapter's configured stale thresholds, because record ingest joins + operator links and a link's metric merge judges each side's freshness + by them (unified-resources contract, "Operator links reach + record-ingested resources"). Regression coverage: + `TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds` in + `internal/unifiedresources/monitor_adapter_read_state_test.go` and + `TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity` in + `internal/monitoring/issue1913_host_continuity_test.go`. 11. The TrueNAS provider projects pools with `Storage.Topology` fixed to `pool` and the ZFS data vdev layout in `Storage.VDevLayout`. The diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 69b3cf77e3..95805c44a6 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -2803,7 +2803,9 @@ application resource-provider or WebSocket lifecycle. ID and drops the superseded pin row (`ApplyCanonicalIDSuccessions` in `internal/unifiedresources/canonical_id_succession.go`). Successions never fire while the old ID still belongs to a live - resource, never follow a contradicting machine key (a reinstalled + resource or to one a manual link folded into its primary (see + "Operator links reach record-ingested resources"), never follow a + contradicting machine key (a reinstalled machine mints fresh, it does not absorb the old host's operator state), and never rewrite change-journal rows. Known limitation: machine-keyless hosts whose history already merged under a collapsed @@ -2884,8 +2886,9 @@ application resource-provider or WebSocket lifecycle. `ResourceRegistry.IngestRecords` applies the same `ApplyCanonicalIDSuccessions` semantics as pin-driven successions (operator state and action audits re-key, the superseded pin row - drops, never while the old ID belongs to a live resource, journal - rows never rewritten). Regression coverage: + drops, never while the old ID belongs to a live resource or one a + manual link folded into its primary, journal rows never rewritten). + Regression coverage: `TestRegistryIngestRecordsKeepsSameHostnameSystemsDistinct` / `TestIngestRecordsSucceedLegacyHostnameScopedCanonicalIDs` / `TestIngestRecordsDoNotCompleteTrueNASIdentityFromPins` in @@ -6535,6 +6538,62 @@ full heading. This shared drawer rule applies across resource types. The 390px browser qualification checks heading fit and keyboard expansion; no data model, authorisation, action dispatch or resource admission contract changes. +### Operator links reach record-ingested resources + +Every registry ingest path applies the operator's manual links: snapshot +ingest, resource ingest and record ingest (`IngestRecords`, +`IngestRecordsWithStaleThresholds` and the read-state overlay's record pass). +Either side of a link can arrive only as a supplemental record, such as a +vSphere or TrueNAS VM with a Pulse agent inside, or an agent on a TrueNAS host, +and the monitor's rebuild ingests those records after the snapshot. While only +snapshot and resource ingest applied links, the rebuilt registry behind the +websocket broadcast and `/api/state` kept both rows. The resources API, which +seeds through resource ingest, showed them merged, and so did the published +inventory whenever an unrelated absent machine made continuity hydration +clone the registry. Mock mode builds its unified view from fixtures without +the link store, so links do not reach the mock broadcast; that boundary is +separate. + +- The monitor adapter passes its configured stale thresholds to record + ingest, so a link joined there judges each side's metrics the way a + snapshot-time link does. Ordinary source merges into an existing row still + use the default thresholds on every ingest path. +- Continuity records are the one record ingest that never joins links. A + saved enrollment only fills an absent machine, so it stays its own row + instead of lending its linked guest an offline verdict and an old agent + payload. +- A link merge keeps the TrueNAS and vSphere payloads the primary lacks, so + an agent chosen as primary over a TrueNAS system still carries the + system's facet, and the system's pools re-parent to the merged row. +- The registry remembers each resource a manual link folded into its + primary. A folded resource is still observed, so neither pin-driven nor + record-declared succession may succeed its ID. An agent-type primary + carries the folded resource's machine keys and is pinned with them, and + succeeding the folded ID re-keyed the link onto the primary itself, which + split the pair on the next rebuild. That already broke snapshot links whose + primary is a machine-keyless agent-type row, such as a Proxmox node. +- Fold history is registry-local: resource seeding and overlay clones do not + carry it. A reference to the folded resource's canonical ID, such as an + availability check configured against an agent later linked into its + guest, does not resolve to the primary, so the check loses its target in + the broadcast as it already did in the resources API. + +`registry_merge_policy_test.go` pins the record links in both directions +through two rebuilds and the live supplemental refresh, and checks that a +registry seeded from the monitor's listing through resource ingest, as the +resources API seeds, lists the same rows +(`TestManualLinksJoinRecordIngestedResourcesInMonitorRebuild`). The same file +pins the node case (`TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence`), +record-declared succession (`TestRecordDeclaredSuccessionSparesALinkFoldedResource`) +and both payloads (`TestManualLinkKeepsProviderPayloadsThePrimaryLacks`). +`monitor_adapter_read_state_test.go` +(`TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds`) pins the +thresholds on the rebuild, supplemental and overlay paths, and +`internal/monitoring/issue1913_host_continuity_test.go` +(`TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity`) pins the +published inventory with no continuity, an unrelated absent machine and the +linked agent's own saved enrollment. + ### Provider link network corroboration An existing reciprocal host/provider link remains authoritative, including a diff --git a/internal/monitoring/issue1913_host_continuity_test.go b/internal/monitoring/issue1913_host_continuity_test.go index 5134592e7b..192540bc5c 100644 --- a/internal/monitoring/issue1913_host_continuity_test.go +++ b/internal/monitoring/issue1913_host_continuity_test.go @@ -89,3 +89,120 @@ func TestHostContinuityPreservesLiveAgentAfterReenrollment(t *testing.T) { }) } } + +// Continuity hydration clones the registry through resource ingest, which +// applies operator links, while the rebuild ingests supplemental records after +// the snapshot. A link to a record-sourced guest (an agent inside a vSphere VM) +// must hold in the published inventory either way, or the state the websocket +// broadcasts and the resources API seeds from flips with an unrelated absent +// machine. A saved enrollment only fills an absent machine: when the linked +// agent itself is gone, its continuity row stays separate and the guest keeps +// what vSphere reports. +func TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity(t *testing.T) { + now := time.Now().UTC() + guestAgent := models.Host{ + ID: "host-app-guest", MachineID: "machine-app-guest", Hostname: "app-guest", + Platform: "linux", Status: "online", LastSeen: now, IntervalSeconds: 30, + } + vmRecords := []unifiedresources.IngestRecord{{ + SourceID: "vc-1:vm:vm-42", + Resource: unifiedresources.Resource{ + Type: unifiedresources.ResourceTypeVM, + Technology: "vmware", + Name: "app-guest", + Status: unifiedresources.StatusOnline, + LastSeen: now, + VMware: &unifiedresources.VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"}, + }, + }} + + unlinked := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental( + models.StateSnapshot{Hosts: []models.Host{guestAgent}, LastUpdate: now}, + map[unifiedresources.DataSource][]unifiedresources.IngestRecord{unifiedresources.SourceVMware: vmRecords}, + ) + if len(unlinked.VMs()) != 1 || len(unlinked.Hosts()) != 1 { + t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(unlinked.VMs()), len(unlinked.Hosts())) + } + vmID, agentID := unlinked.VMs()[0].ID(), unlinked.Hosts()[0].ID() + + for _, tc := range []struct { + name string + reporting bool + continuity []config.HostContinuityEntry + }{ + {name: "no-continuity", reporting: true}, + { + name: "absent-machine-continuity", + reporting: true, + continuity: []config.HostContinuityEntry{{ + HostID: "absent-machine", MachineID: "other-machine", Hostname: "other-host", + Platform: "linux", LastSeen: now.Add(-time.Hour), + }}, + }, + { + name: "linked-agent-continuity-only", + continuity: []config.HostContinuityEntry{{ + HostID: guestAgent.ID, MachineID: guestAgent.MachineID, Hostname: guestAgent.Hostname, + Platform: "linux", LastSeen: now.Add(-10 * time.Minute), + }}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + links := unifiedresources.NewMemoryStore() + if err := links.AddLink(unifiedresources.ResourceLink{ResourceA: vmID, ResourceB: agentID, PrimaryID: vmID}); err != nil { + t.Fatalf("add link: %v", err) + } + state := models.NewState() + if tc.reporting { + state.Hosts = []models.Host{guestAgent} + } + m := &Monitor{state: state} + if len(tc.continuity) > 0 { + continuity := config.NewHostContinuityStore(t.TempDir(), nil) + for _, entry := range tc.continuity { + if err := continuity.Upsert(entry); err != nil { + t.Fatal(err) + } + } + m.hostContinuityStore = continuity + } + m.SetResourceStore(unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(links))) + m.SetSupplementalRecordsProvider(unifiedresources.SourceVMware, &testSupplementalInventoryReadinessProvider{ + source: unifiedresources.SourceVMware, + records: vmRecords, + settled: true, + readyAt: now, + }) + + resources, _ := m.UnifiedResourceSnapshot() + var guest, agent *unifiedresources.Resource + for i := range resources { + switch resources[i].ID { + case agentID: + agent = &resources[i] + case vmID: + guest = &resources[i] + } + } + if guest == nil { + t.Fatalf("linked VM %s missing from published inventory", vmID) + } + if tc.reporting { + if agent != nil { + t.Fatalf("linked agent %s still published standalone", agentID) + } + if guest.VMware == nil || guest.Agent == nil { + t.Fatalf("published VM vmware=%t agent=%t, want the VM carrying its agent", guest.VMware != nil, guest.Agent != nil) + } + return + } + if agent == nil || agent.Status != unifiedresources.StatusOffline { + t.Fatalf("saved enrollment row = %+v, want its own offline row", agent) + } + if guest.Agent != nil || guest.Status != unifiedresources.StatusOnline { + t.Fatalf("published VM agent=%t status=%s, want vSphere's online VM untouched by the saved enrollment", guest.Agent != nil, guest.Status) + } + }) + } +} diff --git a/internal/unifiedresources/canonical_id_pins.go b/internal/unifiedresources/canonical_id_pins.go index ece08ab9b7..c48aa0cee3 100644 --- a/internal/unifiedresources/canonical_id_pins.go +++ b/internal/unifiedresources/canonical_id_pins.go @@ -337,8 +337,9 @@ func (index *identityPinIndex) successionsFor(pin ResourceIdentityPin) []Canonic // the new canonical ID before the pin write, so operator intent // (never-auto-remediate, maintenance windows) and action-audit history // survive the era change. Successions are skipped while the old canonical ID -// still belongs to a live resource: a genuinely short-named host must not -// have its rows stolen by an FQDN sibling. Change-journal rows are never +// still belongs to a live resource, or to one a manual link folded into +// another: a genuinely short-named host must not have its rows stolen by an +// FQDN sibling. Change-journal rows are never // rewritten; EraIDs merges those at read time. func (rr *ResourceRegistry) PersistIdentityPins() { if rr.store == nil { @@ -358,7 +359,11 @@ func (rr *ResourceRegistry) PersistIdentityPins() { } pins = append(pins, pin) for _, succession := range rr.identityPins.successionsFor(pin) { - if _, live := rr.resources[succession.OldCanonicalID]; live { + // A primary that absorbed a linked resource also carries its + // machine keys. That resource is still observed, and succeeding + // it would re-key the link onto the primary itself, splitting + // the pair again on the next rebuild. + if rr.canonicalIDObservedLocked(succession.OldCanonicalID) { continue } successions = append(successions, succession) diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index 6abf5dfdbf..910a61f5a3 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -135,7 +135,7 @@ func readStateWithRecords(readState ReadState, source DataSource, records []Inge cloned := NewRegistry(registry.store) thresholds := adapter.currentStaleThresholds() cloned.IngestResourcesWithStaleThresholds(registry.List(), thresholds) - cloned.ingestRecords(source, records, onlyMissing) + cloned.ingestRecords(source, records, onlyMissing, thresholds) overlay := NewMonitorAdapterWithStaleThresholds(cloned, thresholds) if keyed { adapter.overlays.store(registry, rebuiltAt, key, now, overlay) @@ -450,7 +450,7 @@ func (a *MonitorAdapter) replaceRegistryLocked(snapshot models.StateSnapshot, re if len(records) == 0 || strings.TrimSpace(string(source)) == "" { continue } - rebuilt.IngestRecords(source, records) + rebuilt.IngestRecordsWithStaleThresholds(source, records, staleThresholds) } // IngestSnapshot runs its stale pass before the record sources above are // ingested, so record-sourced resources would otherwise keep their @@ -665,7 +665,7 @@ func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records return } before := registry.List() - registry.IngestRecords(source, records) + registry.IngestRecordsWithStaleThresholds(source, records, a.currentStaleThresholds()) rebuiltAt := time.Now().UTC() recordRegistryChanges(registry.store, before, registry.List(), rebuiltAt, nil, SourcePlatformEvent, changeSourceAdapterForDataSource(source)) registry.PersistIdentityPins() diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go index 7eefcb5b99..dd7bd9da7b 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -400,6 +400,87 @@ func TestReadStateWithRecordsPreservesConfiguredStaleThresholds(t *testing.T) { } } +// A link whose side arrives as a record joins during record ingest, so it +// must judge freshness by the adapter's thresholds as a snapshot-time link +// does. Under a five-minute vSphere threshold a VM polled three minutes ago +// is current, and its own memory reading outranks its in-guest agent's. +func TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + Hosts: []models.Host{{ + ID: "host-app-guest", + Hostname: "app-guest", + MachineID: "machine-app-guest", + Status: "online", + LastSeen: now, + Memory: models.Memory{Total: 8 << 30, Used: 6 << 30, Free: 2 << 30, Usage: 75}, + }}, + LastUpdate: now, + } + vmRecords := []IngestRecord{{ + SourceID: "vc-1:vm:vm-42", + Resource: Resource{ + Type: ResourceTypeVM, + Technology: "vmware", + Name: "app-guest", + Status: StatusOnline, + LastSeen: now.Add(-3 * time.Minute), + Metrics: &ResourceMetrics{Memory: &MetricValue{Percent: 40, Source: SourceVMware}}, + VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"}, + }, + }} + thresholds := map[DataSource]time.Duration{SourceVMware: 5 * time.Minute} + + unlinked := NewMonitorAdapter(NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: vmRecords}) + vms, hosts := unlinked.VMs(), unlinked.Hosts() + if len(vms) != 1 || len(hosts) != 1 { + t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(vms), len(hosts)) + } + link := ResourceLink{ResourceA: vms[0].ID(), ResourceB: hosts[0].ID(), PrimaryID: hosts[0].ID()} + + for _, path := range []struct { + name string + read func(adapter *MonitorAdapter) ReadState + }{ + {"rebuild", func(adapter *MonitorAdapter) ReadState { + adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: vmRecords}) + return adapter + }}, + {"supplemental", func(adapter *MonitorAdapter) ReadState { + adapter.PopulateFromSnapshot(snapshot) + adapter.PopulateSupplementalRecords(SourceVMware, vmRecords) + return adapter + }}, + {"overlay", func(adapter *MonitorAdapter) ReadState { + adapter.PopulateFromSnapshot(snapshot) + return ReadStateWithRecords(adapter, SourceVMware, vmRecords) + }}, + } { + t.Run(path.name, func(t *testing.T) { + store := NewMemoryStore() + if err := store.AddLink(link); err != nil { + t.Fatalf("add link: %v", err) + } + readState := path.read(NewMonitorAdapterWithStaleThresholds(NewRegistry(store), thresholds)) + + if hosts := readState.Hosts(); len(hosts) != 0 { + t.Fatalf("linked agent still listed standalone: %s", hosts[0].ID()) + } + vms := readState.VMs() + if len(vms) != 1 || vms[0].ID() != link.ResourceA { + t.Fatalf("VMs = %d, want the linked VM %s", len(vms), link.ResourceA) + } + if vms[0].r.Metrics == nil || vms[0].r.Metrics.Memory == nil { + t.Fatal("linked VM lost its memory reading") + } + if memory := vms[0].r.Metrics.Memory; memory.Source != SourceVMware || memory.Percent != 40 { + t.Fatalf("linked VM memory = %.0f%% from %s, want vSphere's current 40%%", memory.Percent, memory.Source) + } + }) + } +} + func TestHostContinuityCannotOverwriteCurrentResource(t *testing.T) { now := time.Now().UTC() for _, mode := range []string{"machine-identity", "source-identity", "future-timestamp", "provider-only"} { diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 48f2148fcd..c1b036344f 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -95,6 +95,10 @@ type ResourceRegistry struct { exclusions map[string]struct{} identityPins *identityPinIndex pbsBackups []models.PBSBackup + // linkMergedIDs holds the resources a manual link folded into their + // primary. They are still observed, so identity succession treats them + // as live. + linkMergedIDs map[string]struct{} // supersededIndex maps record-declared retired canonical IDs to the live // resource that superseded them, so references persisted under a retired // ID (availability links, API reads) keep resolving. An empty value marks @@ -611,10 +615,16 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh // IngestRecords ingests normalized records for a single source. func (rr *ResourceRegistry) IngestRecords(source DataSource, records []IngestRecord) { - rr.ingestRecords(source, records, false) + rr.ingestRecords(source, records, false, nil) } -func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool) { +// IngestRecordsWithStaleThresholds ingests normalized records for a single +// source and merges operator-linked resources with caller-owned thresholds. +func (rr *ResourceRegistry) IngestRecordsWithStaleThresholds(source DataSource, records []IngestRecord, thresholds map[DataSource]time.Duration) { + rr.ingestRecords(source, records, false, thresholds) +} + +func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool, thresholds map[DataSource]time.Duration) { var successions []CanonicalIDSuccession supersededSeen := make(map[string]struct{}) for _, record := range records { @@ -649,6 +659,16 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec rr.applyRecordSuccessions(successions) rr.mu.Lock() + // Either side of an operator link can arrive as a record: a vSphere or + // TrueNAS VM with a Pulse agent inside, or an agent on a TrueNAS host. + // Snapshot and resource ingest apply the links, so record ingest must too, + // or the monitor's rebuilt registry keeps both rows while REST, which + // seeds from that registry, shows them merged. Continuity records only + // fill absent machines: a saved enrollment joined to its linked guest + // would lend the guest its offline verdict and old agent payload. + if !onlyMissing { + rr.applyManualLinks(thresholds) + } rr.refreshStorageConsumersLocked() rr.refreshPBSRollupsLocked() rr.refreshStoragePostureLocked() @@ -746,12 +766,25 @@ func (rr *ResourceRegistry) proxmoxGuestResourceIDForSourceRefLocked(ref string) return uniqueID } +// canonicalIDObservedLocked reports whether a canonical ID still names an +// observed resource: a live row, or one a manual link folded into its primary. +// Succeeding an observed ID would hand its operator rows, and the link that +// folded it, to another resource. +func (rr *ResourceRegistry) canonicalIDObservedLocked(id string) bool { + if _, live := rr.resources[id]; live { + return true + } + _, folded := rr.linkMergedIDs[id] + return folded +} + // applyRecordSuccessions re-keys operator-owned store rows from canonical IDs // that ingested records declared superseded (IngestRecord.SupersededCanonicalIDs) // onto the records' current canonical IDs. Mirrors the guards of pin-driven -// successions in PersistIdentityPins: a superseded ID still held by a live -// resource is skipped so a genuinely distinct sibling never has its rows -// stolen. Re-runs are cheap no-ops once the old rows are gone. +// successions in PersistIdentityPins: a superseded ID still observed, live or +// folded into another resource by a manual link, is skipped so a genuinely +// distinct sibling never has its rows stolen. Re-runs are cheap no-ops once +// the old rows are gone. func (rr *ResourceRegistry) applyRecordSuccessions(successions []CanonicalIDSuccession) { if len(successions) == 0 || rr.store == nil { return @@ -764,7 +797,7 @@ func (rr *ResourceRegistry) applyRecordSuccessions(successions []CanonicalIDSucc rr.mu.RLock() kept := make([]CanonicalIDSuccession, 0, len(successions)) for _, succession := range successions { - if _, live := rr.resources[succession.OldCanonicalID]; live { + if rr.canonicalIDObservedLocked(succession.OldCanonicalID) { continue } kept = append(kept, succession) @@ -4811,6 +4844,10 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura rr.mergeResourceData(primary, other, thresholds) delete(rr.resources, otherID) + if rr.linkMergedIDs == nil { + rr.linkMergedIDs = make(map[string]struct{}) + } + rr.linkMergedIDs[otherID] = struct{}{} rr.updateSourceMappings(otherID, primaryID) } } @@ -4876,6 +4913,12 @@ func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource if primary.Ceph == nil { primary.Ceph = other.Ceph } + if primary.TrueNAS == nil { + primary.TrueNAS = other.TrueNAS + } + if primary.VMware == nil { + primary.VMware = other.VMware + } // Manual links combine already-normalized resources. Preserve each metric's // recorded source instead of flattening the linked resource to SourceAgent. diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index e4d7080211..c133a0e533 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -441,6 +441,319 @@ func TestManualLinkKeepsTheLiveSightingOfASharedSource(t *testing.T) { } } +// Either side of an operator link can arrive as a supplemental record: a +// vSphere VM with a Pulse agent inside, or an agent on a TrueNAS host. The +// monitor's rebuild ingests records after the snapshot, so it must join them +// there too, keep the record side's provider payload, and list what the +// resources API lists after seeding from it. +func TestManualLinksJoinRecordIngestedResourcesInMonitorRebuild(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + Hosts: []models.Host{ + {ID: "host-app-guest", Hostname: "app-guest", MachineID: "machine-app-guest", Status: "online", LastSeen: now}, + {ID: "host-nas", Hostname: "nas-a-mgmt", MachineID: "machine-nas-a", Status: "online", LastSeen: now}, + }, + LastUpdate: now, + } + vmRecord := IngestRecord{ + SourceID: "vc-1:vm:vm-42", + Resource: Resource{ + Type: ResourceTypeVM, + Technology: "vmware", + Name: "app-guest", + Status: StatusOnline, + LastSeen: now, + VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"}, + }, + Identity: ResourceIdentity{Hostnames: []string{"app-guest"}}, + } + records := map[DataSource][]IngestRecord{ + SourceVMware: {vmRecord}, + SourceTrueNAS: { + { + SourceID: "system:tn-1", + Resource: Resource{ + Type: ResourceTypeAgent, + Name: "nas-a", + Status: StatusOnline, + LastSeen: now, + TrueNAS: &TrueNASData{Hostname: "nas-a"}, + }, + Identity: ResourceIdentity{Hostnames: []string{"nas-a"}}, + }, + { + SourceID: "system:tn-1:pool:tank", + ParentSourceID: "system:tn-1", + Resource: Resource{ + Type: ResourceTypeStorage, + Name: "tank", + Status: StatusOnline, + LastSeen: now, + Storage: &StorageMeta{Type: "zfs-pool", Platform: "truenas", Topology: "pool"}, + }, + }, + }, + } + + for _, agentPrimary := range []bool{false, true} { + name := "record-primary" + if agentPrimary { + name = "agent-primary" + } + t.Run(name, func(t *testing.T) { + store := NewMemoryStore() + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateSnapshotAndSupplemental(snapshot, records) + + var vmID, guestAgentID, systemID, nasAgentID string + for _, resource := range adapter.GetAll() { + switch { + case resource.VMware != nil: + vmID = resource.ID + case resource.TrueNAS != nil: + systemID = resource.ID + case resource.Name == "app-guest": + guestAgentID = resource.ID + case resource.Name == "nas-a-mgmt": + nasAgentID = resource.ID + } + } + if vmID == "" || guestAgentID == "" || systemID == "" || nasAgentID == "" { + t.Fatalf("unlinked estate = %v, want the VM, both agents and the TrueNAS system", resourceIDs(adapter.GetAll())) + } + guestPrimary, nasPrimary := vmID, systemID + if agentPrimary { + guestPrimary, nasPrimary = guestAgentID, nasAgentID + } + for _, link := range []ResourceLink{ + {ResourceA: vmID, ResourceB: guestAgentID, PrimaryID: guestPrimary}, + {ResourceA: systemID, ResourceB: nasAgentID, PrimaryID: nasPrimary}, + } { + if err := store.AddLink(link); err != nil { + t.Fatalf("add link: %v", err) + } + } + + assertJoined := func(stage string) { + t.Helper() + resources := adapter.GetAll() + if len(resources) != 3 { + t.Fatalf("%s: resources = %v, want the linked VM, the linked TrueNAS system and its pool", stage, resourceIDs(resources)) + } + // An agent inside a guest supplements the guest, whichever + // side the operator chose as primary. + guest, ok := adapter.currentRegistry().Get(vmID) + if !ok { + t.Fatalf("%s: linked vSphere VM %s missing from %v", stage, vmID, resourceIDs(resources)) + } + if guest.Type != ResourceTypeVM || guest.VMware == nil || guest.Agent == nil { + t.Fatalf("%s: linked VM type=%s vmware=%t agent=%t, want the VM carrying its agent", stage, guest.Type, guest.VMware != nil, guest.Agent != nil) + } + if !slices.Contains(guest.Sources, SourceVMware) || !slices.Contains(guest.Sources, SourceAgent) { + t.Fatalf("%s: linked VM sources = %v, want vmware and agent", stage, guest.Sources) + } + system, ok := adapter.currentRegistry().Get(nasPrimary) + if !ok { + t.Fatalf("%s: linked TrueNAS system %s missing from %v", stage, nasPrimary, resourceIDs(resources)) + } + if system.TrueNAS == nil || system.Agent == nil { + t.Fatalf("%s: linked system truenas=%t agent=%t, want the primary carrying both payloads", stage, system.TrueNAS != nil, system.Agent != nil) + } + if !slices.Contains(system.Sources, SourceTrueNAS) || !slices.Contains(system.Sources, SourceAgent) { + t.Fatalf("%s: linked system sources = %v, want truenas and agent", stage, system.Sources) + } + pool := onlyResourceOfType(t, adapter.currentRegistry(), ResourceTypeStorage) + if pool.ParentID == nil || *pool.ParentID != nasPrimary { + t.Fatalf("%s: pool parent = %v, want the linked system %s", stage, pool.ParentID, nasPrimary) + } + + // Identity pins persisted by the rebuild must not re-key a link + // onto its own primary. + links, err := store.GetLinks() + if err != nil { + t.Fatalf("%s: get links: %v", stage, err) + } + for _, link := range links { + if link.ResourceA == link.ResourceB { + t.Fatalf("%s: link collapsed onto %s", stage, link.ResourceA) + } + } + + // The resources API seeds its registry from this listing + // through resource ingest and must agree with it. + rest := NewRegistry(store) + rest.IngestResources(resources) + if got, want := resourceIDs(rest.List()), resourceIDs(resources); !slices.Equal(got, want) { + t.Fatalf("%s: a registry seeded from the monitor lists %v, monitor lists %v", stage, got, want) + } + } + + next := snapshot + next.LastUpdate = now.Add(time.Second) + adapter.PopulateSnapshotAndSupplemental(next, records) + assertJoined("rebuild") + + adapter.PopulateSupplementalRecords(SourceVMware, records[SourceVMware]) + adapter.PopulateSupplementalRecords(SourceTrueNAS, records[SourceTrueNAS]) + assertJoined("supplemental refresh") + + next.LastUpdate = now.Add(2 * time.Second) + adapter.PopulateSnapshotAndSupplemental(next, records) + assertJoined("second rebuild") + }) + } +} + +// A Proxmox node that absorbs a linked agent carries the agent's machine key, +// so the rebuild pins the node with it. The agent is still observed: that pin +// must not succeed the agent's ID onto the node, which re-keyed the link onto +// the node itself and split the pair on the next rebuild. +func TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + Nodes: []models.Node{{ + ID: "lab-pve1", Name: "pve1", Instance: "lab", + Host: "https://192.0.2.10:8006", Status: "online", LastSeen: now, + }}, + Hosts: []models.Host{{ + ID: "host-box", Hostname: "box-x", MachineID: "machine-box", Status: "online", LastSeen: now, + }}, + LastUpdate: now, + } + store := NewMemoryStore() + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateSnapshotAndSupplemental(snapshot, nil) + var nodeID, agentID string + for _, resource := range adapter.GetAll() { + if resource.Proxmox != nil { + nodeID = resource.ID + } else if resource.Agent != nil { + agentID = resource.ID + } + } + if nodeID == "" || agentID == "" || nodeID == agentID { + t.Fatalf("unlinked estate = %v, want a node and a separate agent", resourceIDs(adapter.GetAll())) + } + if err := store.AddLink(ResourceLink{ResourceA: nodeID, ResourceB: agentID, PrimaryID: nodeID}); err != nil { + t.Fatalf("add link: %v", err) + } + + for rebuild := 1; rebuild <= 3; rebuild++ { + next := snapshot + next.LastUpdate = now.Add(time.Duration(rebuild) * time.Second) + adapter.PopulateSnapshotAndSupplemental(next, nil) + if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{nodeID}) { + t.Fatalf("rebuild %d: resources = %v, want only the linked node %s", rebuild, got, nodeID) + } + links, err := store.GetLinks() + if err != nil { + t.Fatalf("get links: %v", err) + } + if len(links) != 1 || links[0].ResourceA != nodeID || links[0].ResourceB != agentID { + t.Fatalf("rebuild %d: links = %+v, want %s -> %s kept", rebuild, links, nodeID, agentID) + } + } +} + +// A record may name an older canonical ID as superseded on every ingest. When +// that ID belongs to a resource a manual link folded into the record's own +// resource, it is still observed: succeeding it on a later refresh would +// re-key the link onto its primary. +func TestRecordDeclaredSuccessionSparesALinkFoldedResource(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + Hosts: []models.Host{{ID: "host-nas", Hostname: "nas-a-mgmt", MachineID: "machine-nas-a", Status: "online", LastSeen: now}}, + LastUpdate: now, + } + agentID := MachineIdentityCanonicalID(ResourceTypeAgent, "machine-nas-a") + records := []IngestRecord{{ + SourceID: "system:tn-1", + SupersededCanonicalIDs: []string{agentID}, + Resource: Resource{ + Type: ResourceTypeAgent, + Name: "nas-a", + Status: StatusOnline, + LastSeen: now, + TrueNAS: &TrueNASData{Hostname: "nas-a"}, + }, + Identity: ResourceIdentity{Hostnames: []string{"nas-a"}}, + }} + + unlinked := NewMonitorAdapter(NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceTrueNAS: records}) + var systemID string + for _, resource := range unlinked.GetAll() { + if resource.TrueNAS != nil { + systemID = resource.ID + } + } + if systemID == "" || !slices.Contains(resourceIDs(unlinked.GetAll()), agentID) { + t.Fatalf("unlinked estate = %v, want agent %s and a TrueNAS system", resourceIDs(unlinked.GetAll()), agentID) + } + + store := NewMemoryStore() + if err := store.AddLink(ResourceLink{ResourceA: systemID, ResourceB: agentID, PrimaryID: systemID}); err != nil { + t.Fatalf("add link: %v", err) + } + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceTrueNAS: records}) + adapter.PopulateSupplementalRecords(SourceTrueNAS, records) + + if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{systemID}) { + t.Fatalf("resources = %v, want only the linked system %s", got, systemID) + } + links, err := store.GetLinks() + if err != nil { + t.Fatalf("get links: %v", err) + } + if len(links) != 1 || links[0].ResourceA != systemID || links[0].ResourceB != agentID { + t.Fatalf("links = %+v, want %s -> %s kept", links, systemID, agentID) + } +} + +// A link merge folds the other resource's provider payloads into the primary +// where the primary has none, whichever side the operator chose. +func TestManualLinkKeepsProviderPayloadsThePrimaryLacks(t *testing.T) { + now := time.Now().UTC() + store := NewMemoryStore() + for _, link := range []ResourceLink{ + {ResourceA: "agent-nas", ResourceB: "truenas-system", PrimaryID: "agent-nas"}, + {ResourceA: "agent-esxi", ResourceB: "vmware-host", PrimaryID: "agent-esxi"}, + } { + if err := store.AddLink(link); err != nil { + t.Fatalf("add link: %v", err) + } + } + host := func(id, name string, sources ...DataSource) Resource { + status := make(map[DataSource]SourceStatus, len(sources)) + for _, source := range sources { + status[source] = SourceStatus{Status: "online", LastSeen: now} + } + return Resource{ID: id, Type: ResourceTypeAgent, Name: name, Status: StatusOnline, LastSeen: now, Sources: sources, SourceStatus: status} + } + agentNAS := host("agent-nas", "nas-a-mgmt", SourceAgent) + agentNAS.Agent = &AgentData{AgentID: "agent-nas", Hostname: "nas-a-mgmt"} + system := host("truenas-system", "nas-a", SourceTrueNAS) + system.TrueNAS = &TrueNASData{Hostname: "nas-a"} + agentESXi := host("agent-esxi", "esxi-01", SourceAgent) + agentESXi.Agent = &AgentData{AgentID: "agent-esxi", Hostname: "esxi-01"} + esxi := host("vmware-host", "esxi-01.lab", SourceVMware) + esxi.VMware = &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "host-101", EntityType: "host"} + + rr := NewRegistry(store) + rr.IngestResources([]Resource{agentNAS, system, agentESXi, esxi}) + + if got := resourceIDs(rr.List()); !slices.Equal(got, []string{"agent-esxi", "agent-nas"}) { + t.Fatalf("resources = %v, want the two linked primaries", got) + } + if nas, _ := rr.Get("agent-nas"); nas.TrueNAS == nil || nas.Agent == nil { + t.Fatalf("linked TrueNAS system truenas=%t agent=%t, want both payloads", nas.TrueNAS != nil, nas.Agent != nil) + } + if host, _ := rr.Get("agent-esxi"); host.VMware == nil || host.Agent == nil { + t.Fatalf("linked vSphere host vmware=%t agent=%t, want both payloads", host.VMware != nil, host.Agent != nil) + } +} + func TestLeaseExpiredKubernetesClusterStaysOffline(t *testing.T) { rr := NewRegistry(nil) rr.IngestSnapshot(models.StateSnapshot{