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.