From b0590c0ccf6701adefc16fdf1c826aec474eb6e7 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 7 Oct 2026 20:54:29 +0100 Subject: [PATCH 1/2] Join operator links on resources that arrive as records The monitor rebuilds its registry from the state snapshot and then ingests supplemental records (vSphere, TrueNAS). Snapshot and resource ingest applied the operator's manual links, but record ingest did not, so a link whose side exists only as a record (an agent inside a vSphere or TrueNAS VM, an agent on a TrueNAS host) never joined in the registry behind the websocket broadcast and /api/state. The resources API seeds through resource ingest and showed the same pair merged, and the published inventory also flipped to merged whenever continuity hydration cloned the registry for an unrelated absent machine. Record ingest now applies the links, with the adapter's stale thresholds on the rebuild, live supplemental and overlay paths. Continuity records stay out of it: a saved enrollment only fills an absent machine and must not lend its linked guest an offline verdict. A link merge keeps the TrueNAS and vSphere payloads the primary lacks. Joining links in the rebuild exposed how identity treats a folded resource. Its ID was no longer live, so once an agent-type primary was pinned with the folded agent's machine key, pin succession re-keyed the link onto the primary itself and the next rebuild split the pair. That already broke snapshot links with a Proxmox node as primary. The registry now remembers each folded resource, and pin-driven and record-declared successions treat it as observed. Mock mode still builds its view from fixtures without the link store, so the mock broadcast is unchanged here. --- .../v6/internal/subsystems/monitoring.md | 11 +- .../internal/subsystems/unified-resources.md | 65 +++- .../issue1913_host_continuity_test.go | 117 +++++++ .../unifiedresources/canonical_id_pins.go | 11 +- internal/unifiedresources/monitor_adapter.go | 6 +- .../monitor_adapter_read_state_test.go | 81 +++++ internal/unifiedresources/registry.go | 55 ++- .../registry_merge_policy_test.go | 313 ++++++++++++++++++ 8 files changed, 643 insertions(+), 16 deletions(-) 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{ From 3a482c610c2212966290f8e2dea9fd015a9b1f27 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 7 Oct 2026 21:28:01 +0100 Subject: [PATCH 2/2] Judge source merges by the ingest's stale thresholds mergeInto always passed nil thresholds to mergeMetrics, so every source merge into an existing row judged metric freshness by the registry defaults while the stale pass, the manual links and the monitor's configuration used the caller's thresholds. A Proxmox node polled every two minutes (four-minute threshold) merged with its auto-linked agent showed the silent agent's readings while the stale pass called the node's poll current, and a supplemental refresh of a linked vSphere VM kept the in-guest agent's memory over vSphere's current reading. Snapshot and record ingest now hold the caller's thresholds on the registry while they run, the way the guest loop holds its agent index, and mergeInto reads them. The ~50 snapshot helpers that reach mergeInto through ingest() keep their signatures. The stale pass and the metric merge now share one threshold rule, sourceStaleThreshold, so the merge also honours a sighting's expected update interval (slow Proxmox disk inventory) as the stale pass already did, and a source without a default falls back to 120s in both (the merge used 60s). Every defined source has a default, so that fallback reaches none. Outside mock mode the monitor configures only Proxmox, PBS and PMG, so those are the production sources this changes; mock mode also configures TrueNAS, vSphere and availability. --- .../v6/internal/subsystems/monitoring.md | 10 +- .../internal/subsystems/unified-resources.md | 24 ++- .../monitor_adapter_read_state_test.go | 147 ++++++++++++++++++ internal/unifiedresources/registry.go | 65 +++++--- .../registry_merge_policy_test.go | 108 +++++++++++++ 5 files changed, 323 insertions(+), 31 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index d7b496972e..4871c3d8c3 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2679,10 +2679,12 @@ truthfulness, not native thaw, containing-release or workload acceptance. 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 + operator links and the freshness gate of every metric merge, a link's or + a source's into an existing row, reads them (unified-resources contract, + "Operator links reach record-ingested resources"). Regression + coverage: + `TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds` and + `TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds` in `internal/unifiedresources/monitor_adapter_read_state_test.go` and `TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity` in `internal/monitoring/issue1913_host_continuity_test.go`. diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 95805c44a6..189b3ddc69 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -1609,7 +1609,12 @@ Source freshness is a separate unified-resource status input, not a fixed global health timer. Snapshot rebuilds, resource seeding, supplemental record overlays, and cloned read-state overlays must preserve the monitoring-provided stale thresholds for Proxmox, PBS, and PMG sources so resources do not flap to -warning/degraded between successful configured poll cycles. +warning/degraded between successful configured poll cycles. A source merge +into an existing row judges metric freshness by the ingest's thresholds and +the stale pass's own threshold rule, so the merge's freshness gate never +calls a source current that the stale pass marks stale, or the reverse. The +gate decides only where no stronger rule does: a hypervisor-managed guest +keeps its platform CPU even while that source is stale. Service-discovery readiness is a unified-resource payload contract, not a drawer-local decoration. Resource list/detail payloads that expose a @@ -6556,8 +6561,21 @@ 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. + snapshot-time link does. Ordinary source merges into an existing row use + the same thresholds on every ingest path: snapshot and record ingest hold + the caller's thresholds while they run, and `mergeInto` judges each + metric's freshness with them through `sourceStaleThreshold`, the rule the + stale pass reads. Before this, those merges used the defaults, so a + Proxmox node polled every two minutes showed a silent auto-linked agent's + readings while the stale pass called the node's poll current, and a + supplemental refresh of a linked vSphere VM kept the agent's memory over + vSphere's current reading. Outside mock mode the monitor configures only + Proxmox, PBS and PMG; mock mode also configures TrueNAS, vSphere and + availability. Regression coverage: + `TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds` (rebuild, + supplemental refresh, overlay, node with auto-linked agent), + `TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds` and + `TestMetricMergeFreshnessMatchesTheStalePass`. - 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 diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go index dd7bd9da7b..55a8298d70 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -481,6 +481,153 @@ func TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds(t *testin } } +// Every source merge into an existing row judges metric freshness by the +// adapter's configured thresholds, as the rebuild's links and stale pass do. +// A linked vSphere VM first reports no memory, so its in-guest agent supplies +// it; a refresh then brings vSphere memory observed three minutes ago, current +// under a five-minute threshold. The live supplemental refresh and the +// read-state overlay merge that record into the existing row, and must show +// vSphere's memory as a rebuild from the same records does. A Proxmox node +// polled every two minutes is likewise current at ninety seconds, so the +// rebuild keeps its readings over those of its auto-linked agent, silent for +// seventy-five seconds. +func TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds(t *testing.T) { + now := time.Now().UTC() + thresholds := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute, SourceVMware: 5 * time.Minute} + + t.Run("linked vSphere VM refresh", func(t *testing.T) { + 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, + } + vmRecord := func(lastSeen time.Time, metrics *ResourceMetrics) []IngestRecord { + return []IngestRecord{{ + SourceID: "vc-1:vm:vm-42", + Resource: Resource{ + Type: ResourceTypeVM, + Technology: "vmware", + Name: "app-guest", + Status: StatusOnline, + LastSeen: lastSeen, + Metrics: metrics, + VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"}, + }, + }} + } + initial := vmRecord(now.Add(-4*time.Minute), &ResourceMetrics{CPU: &MetricValue{Percent: 10, Source: SourceVMware}}) + refreshedAt := now.Add(-3 * time.Minute) + refreshed := vmRecord(refreshedAt, &ResourceMetrics{ + CPU: &MetricValue{Percent: 10, Source: SourceVMware}, + Memory: &MetricValue{Percent: 40, Source: SourceVMware}, + }) + + unlinked := NewMonitorAdapter(NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial}) + 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: refreshed}) + return adapter + }}, + {"supplemental", func(adapter *MonitorAdapter) ReadState { + adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial}) + adapter.PopulateSupplementalRecords(SourceVMware, refreshed) + return adapter + }}, + {"overlay", func(adapter *MonitorAdapter) ReadState { + adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial}) + return ReadStateWithRecords(adapter, SourceVMware, refreshed) + }}, + } { + 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)) + + vms := readState.VMs() + if len(vms) != 1 || vms[0].ID() != link.ResourceA || len(readState.Hosts()) != 0 { + t.Fatalf("estate = %d VMs, %d standalone agents, want only the linked VM %s", len(vms), len(readState.Hosts()), link.ResourceA) + } + if got := vms[0].r.SourceStatus[SourceVMware]; got.Status != "online" || !got.LastSeen.Equal(refreshedAt) { + t.Fatalf("vSphere sighting = %+v, want the refresh, current under the configured threshold", got) + } + 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) + } + }) + } + }) + + t.Run("node with auto-linked agent", func(t *testing.T) { + adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), thresholds) + adapter.PopulateFromSnapshot(models.StateSnapshot{ + Nodes: []models.Node{{ + ID: "homelab-pve1", + Name: "pve1", + Instance: "homelab", + Status: "online", + LinkedAgentID: "host-pve1", + CPU: 0.12, + Memory: models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25}, + LastSeen: now.Add(-90 * time.Second), + }}, + Hosts: []models.Host{{ + ID: "host-pve1", + MachineID: "machine-pve1", + Hostname: "pve1", + LinkedNodeID: "homelab-pve1", + Status: "online", + CPUUsage: 80, + Memory: models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75}, + LastSeen: now.Add(-75 * time.Second), + IntervalSeconds: 30, + }}, + LastUpdate: now, + }) + + nodes := adapter.Nodes() + if len(nodes) != 1 { + t.Fatalf("nodes = %d, want the node merged with its agent", len(nodes)) + } + node := nodes[0].r + if poll := node.SourceStatus[SourceProxmox]; poll.Status != "online" || !poll.LastSeen.Equal(now.Add(-90*time.Second)) { + t.Fatalf("Proxmox sighting = %+v, want the current poll from ninety seconds ago", poll) + } + if got := node.SourceStatus[SourceAgent].Status; got != "stale" { + t.Fatalf("agent sighting = %q, want stale", got) + } + if node.Metrics == nil || node.Metrics.CPU == nil || node.Metrics.Memory == nil { + t.Fatalf("merged node lost its readings: %+v", node.Metrics) + } + if cpu := node.Metrics.CPU; cpu.Source != SourceProxmox || cpu.Percent != 12 { + t.Fatalf("node CPU = %.0f%% from %s, want the current poll's 12%%", cpu.Percent, cpu.Source) + } + if memory := node.Metrics.Memory; memory.Source != SourceProxmox || memory.Percent != 25 { + t.Fatalf("node memory = %.0f%% from %s, want the current poll's 25%%", 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 c1b036344f..ac5891cacd 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -152,6 +152,13 @@ type ResourceRegistry struct { // what the full scan would see; scores stay live through the shared // pointers. agentNodeScanIndex map[string][]agentNodeCandidate + + // ingestStaleThresholds holds the caller-owned freshness thresholds of + // the snapshot or record ingest in progress, so each source merge into an + // existing row judges metric freshness as that ingest's manual links and + // stale pass do. nil outside an ingest and for callers without + // thresholds, where all three use the defaults. + ingestStaleThresholds map[DataSource]time.Duration } // agentNodeCandidate pairs a resources-map key with its entry so the indexed @@ -247,6 +254,10 @@ func (rr *ResourceRegistry) IngestSnapshotWithStaleThresholds(snapshot models.St } func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresholds map[DataSource]time.Duration) { + rr.mu.Lock() + rr.ingestStaleThresholds = thresholds + rr.mu.Unlock() + hostByID := make(map[string]*models.Host, len(snapshot.Hosts)) for i := range snapshot.Hosts { host := snapshot.Hosts[i] @@ -610,6 +621,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh rr.refreshCanonicalIdentitiesLocked() rr.markStaleLocked(time.Now().UTC(), thresholds) rr.invalidateViewsLocked() + rr.ingestStaleThresholds = nil rr.mu.Unlock() } @@ -625,6 +637,10 @@ func (rr *ResourceRegistry) IngestRecordsWithStaleThresholds(source DataSource, } func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool, thresholds map[DataSource]time.Duration) { + rr.mu.Lock() + rr.ingestStaleThresholds = thresholds + rr.mu.Unlock() + var successions []CanonicalIDSuccession supersededSeen := make(map[string]struct{}) for _, record := range records { @@ -677,6 +693,7 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec rr.refreshProxmoxSensorSetupLocked() rr.refreshCanonicalIdentitiesLocked() rr.invalidateViewsLocked() + rr.ingestStaleThresholds = nil rr.mu.Unlock() } @@ -1805,26 +1822,15 @@ func sourceVerdictsRecorded(sightings map[DataSource]SourceStatus) bool { } func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSource]time.Duration) { - thresholds = effectiveStaleThresholds(thresholds) - changed := false for _, resource := range rr.resources { previousStatus := resource.Status staleFound := false for source, status := range resource.SourceStatus { - threshold, ok := thresholds[source] - if !ok { - threshold = 120 * time.Second - } - if status.ExpectedUpdateIntervalSeconds > 0 { - // Slow inventory polls have their own cadence. A source must miss - // two expected intervals before its retained observation is stale. - threshold = max(threshold, 2*time.Duration(status.ExpectedUpdateIntervalSeconds)*time.Second) - } if status.LastSeen.IsZero() { continue } - if now.Sub(status.LastSeen) > threshold { + if now.Sub(status.LastSeen) > sourceStaleThreshold(source, status, thresholds) { changed = changed || status.Status != "stale" status.Status = "stale" resource.SourceStatus[source] = status @@ -3973,7 +3979,7 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou existing.ParentID = rr.resolveCanonicalParentID(existing) existing.Status = chooseStatus(existing.Status, incoming.Status, source, existing.Sources) - existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, nil) + existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, rr.ingestStaleThresholds) existing.Metrics = clearUnavailableSourceMemoryMetric(existing.Metrics, &incoming, source) // Prefer agent naming when available @@ -6327,6 +6333,27 @@ func mergeMetrics( return &merged } +// sourceStaleThreshold is how long a source may go without delivering before +// its sighting is stale: the caller's threshold for the source, else the +// default. The stale pass and the metric merge's freshness gate both read it, +// so the gate never calls a source current that the pass marks stale, or the +// reverse. +func sourceStaleThreshold(source DataSource, sighting SourceStatus, thresholds map[DataSource]time.Duration) time.Duration { + threshold := thresholds[source] + if threshold <= 0 { + threshold = defaultStaleThresholds[source] + } + if threshold <= 0 { + threshold = 120 * time.Second + } + if sighting.ExpectedUpdateIntervalSeconds > 0 { + // Slow inventory polls have their own cadence. A source must miss + // two expected intervals before its retained observation is stale. + threshold = max(threshold, 2*time.Duration(sighting.ExpectedUpdateIntervalSeconds)*time.Second) + } + return threshold +} + // 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. @@ -6343,17 +6370,7 @@ func metricSourceStale( if !ok || st.LastSeen.IsZero() { return false } - 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 + return now.Sub(st.LastSeen) > sourceStaleThreshold(source, st, thresholds) } func mergeMetric( diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index c133a0e533..6f04cb36b7 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -247,6 +247,114 @@ func TestLeaseExpiredAgentOnLiveProxmoxNodeStaysOnline(t *testing.T) { } } +// A source merge into an existing row judges metric freshness by the ingest's +// stale thresholds, as the stale pass that ends the ingest does. With PVE +// polled every two minutes the monitor allows Proxmox four minutes, so a node +// polled ninety seconds ago is current while its agent, silent for seventy-five +// seconds, is stale, and the row shows the node's readings. Without thresholds +// both sources are stale by the defaults and the agent keeps its precedence. +func TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + Nodes: []models.Node{{ + ID: "homelab-pve1", + Name: "pve1", + Instance: "homelab", + Status: "online", + LinkedAgentID: "host-pve1", + CPU: 0.12, + Memory: models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25}, + LastSeen: now.Add(-90 * time.Second), + }}, + Hosts: []models.Host{{ + ID: "host-pve1", + MachineID: "machine-pve1", + Hostname: "pve1", + LinkedNodeID: "homelab-pve1", + Status: "online", + CPUUsage: 80, + Memory: models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75}, + LastSeen: now.Add(-75 * time.Second), + IntervalSeconds: 30, + }}, + } + + for _, tc := range []struct { + name string + thresholds map[DataSource]time.Duration + proxmoxPoll string + want DataSource + wantCPU float64 + wantMemory float64 + }{ + {"configured", map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute}, "online", SourceProxmox, 12, 25}, + {"defaults", nil, "stale", SourceAgent, 80, 75}, + } { + t.Run(tc.name, func(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshotWithStaleThresholds(snapshot, tc.thresholds) + + resource := onlyResourceOfType(t, rr, ResourceTypeAgent) + if got := resource.SourceStatus[SourceProxmox].Status; got != tc.proxmoxPoll { + t.Fatalf("Proxmox sighting = %q, want %q: %+v", got, tc.proxmoxPoll, resource.SourceStatus) + } + if got := resource.SourceStatus[SourceAgent].Status; got != "stale" { + t.Fatalf("agent sighting = %q, want stale", got) + } + if resource.Metrics == nil || resource.Metrics.CPU == nil || resource.Metrics.Memory == nil { + t.Fatalf("merged row lost its readings: %+v", resource.Metrics) + } + if cpu := resource.Metrics.CPU; cpu.Source != tc.want || cpu.Percent != tc.wantCPU { + t.Fatalf("CPU = %.0f%% from %s, want %.0f%% from %s", cpu.Percent, cpu.Source, tc.wantCPU, tc.want) + } + if memory := resource.Metrics.Memory; memory.Source != tc.want || memory.Percent != tc.wantMemory { + t.Fatalf("memory = %.0f%% from %s, want %.0f%% from %s", memory.Percent, memory.Source, tc.wantMemory, tc.want) + } + }) + } +} + +// The metric merge and the stale pass read one threshold rule, so a source +// sighting the merge treats as current is never one the pass marks stale. +func TestMetricMergeFreshnessMatchesTheStalePass(t *testing.T) { + now := time.Now().UTC() + configured := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute, SourceVMware: 5 * time.Minute} + for _, tc := range []struct { + name string + source DataSource + sighting SourceStatus + thresholds map[DataSource]time.Duration + wantStale bool + }{ + {"proxmox by default", SourceProxmox, SourceStatus{LastSeen: now.Add(-90 * time.Second)}, nil, true}, + {"proxmox by configuration", SourceProxmox, SourceStatus{LastSeen: now.Add(-90 * time.Second)}, configured, false}, + {"vsphere by default", SourceVMware, SourceStatus{LastSeen: now.Add(-3 * time.Minute)}, nil, true}, + {"vsphere by configuration", SourceVMware, SourceStatus{LastSeen: now.Add(-3 * time.Minute)}, configured, false}, + {"agent without its own threshold", SourceAgent, SourceStatus{LastSeen: now.Add(-75 * time.Second)}, configured, true}, + {"slow disk inventory", SourceProxmox, SourceStatus{LastSeen: now.Add(-5 * time.Minute), ExpectedUpdateIntervalSeconds: 600}, nil, false}, + {"missed two disk inventory polls", SourceProxmox, SourceStatus{LastSeen: now.Add(-21 * time.Minute), ExpectedUpdateIntervalSeconds: 600}, nil, true}, + } { + t.Run(tc.name, func(t *testing.T) { + status := map[DataSource]SourceStatus{tc.source: tc.sighting} + if got := metricSourceStale(now, status, tc.source, tc.thresholds); got != tc.wantStale { + t.Fatalf("metric merge stale = %v, want %v", got, tc.wantStale) + } + + rr := NewRegistry(nil) + rr.resources["resource"] = &Resource{ + ID: "resource", + Status: StatusOnline, + Sources: []DataSource{tc.source}, + SourceStatus: map[DataSource]SourceStatus{tc.source: tc.sighting}, + } + rr.MarkStale(now, tc.thresholds) + if got := rr.resources["resource"].SourceStatus[tc.source].Status == "stale"; got != tc.wantStale { + t.Fatalf("stale pass stale = %v, want %v", got, tc.wantStale) + } + }) + } +} + // Docker hosts hold a shorter lease than the registry's Docker stale // threshold, so a silent Docker host was offline for a while and then // flipped to warning once its sighting went stale.