diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 4715dcd9c6..c258f4881e 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8746,6 +8746,17 @@ nodes/controller members, and confirmed inventory removal. `TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata. This is synthetic runtime evidence, not USB hardware or reporter acceptance. +### Coalesced host readings follow configured source freshness + +The websocket broadcast coalesces a Proxmox node with a Pulse agent that keeps +its own registry row, and now judges which source's readings the row shows by +the stale thresholds the view's registry used, the monitor's configured ones +on the resource store (`internal/monitoring/monitor.go`). +This changes which readings a coalesced row shows, not agent admission, +registration, linking, heartbeat or removal. Proof: +`TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds` in +`internal/monitoring/monitor_host_agents_test.go`. + ### Quick security setup preserves unrelated settings Authenticated force setup in `internal/api/security_setup_fix.go` retains the diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index ca46405a4e..39e2e31eac 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -9570,6 +9570,22 @@ That same resource-handler seed contract must also stay on canonical unified resource ownership for tenant-scoped requests: once a tenant state provider implements `UnifiedResourceSnapshotForTenant`, `/api/resources` may not fall back to raw tenant `StateSnapshot` seeding when that unified seed is empty. +The seed contract carries source freshness too. A provider that implements +`UnifiedResourceSnapshotWithStaleThresholds` (the default monitor) or +`UnifiedResourceSnapshotWithStaleThresholdsForTenant` (the tenant state +provider) returns the seed together with the stale thresholds its registry +generation judged it by, captured with the listing, and the registry +`/api/resources` rebuilds is built with them. Its stale pass and +presentation coalesce then judge as the websocket broadcast does: a Proxmox node +polled every two minutes stays online with a current sighting between polls +instead of reading stale with warning status. The registry cache is keyed by +its own copy of those thresholds as well as the freshness marker. A provider without those +methods leaves the registry defaults, and the resources API reads them +through optional interface checks, so `internal/api/router_helpers.go` +asserts the monitor and the tenant state provider implement them. Proof: +`TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast` (default +and tenant monitors on a resource store, and a monitor without one, each +against the same monitor's broadcast). That same mock/runtime contract now also governs chart payloads under `internal/api/chartapi/service.go`: when demo or mock presentation is enabled, `/api/charts`, `/api/charts/infrastructure`, and `/api/storage-charts` must diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 9f57b18290..d4eb93380d 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1193,6 +1193,18 @@ Monitoring owns source freshness cadence for Proxmox, PBS, and PMG resources: the stale threshold is derived from the configured polling interval with a minimum floor, so API-facing resource status must not degrade merely because a healthy source is between normal poll cycles. +Each unified view carries the thresholds its listed registry generation was +judged by, read with the listing from the same generation: the resource +store's configured ones, or none (the defaults) for a view built from mock +fixtures or a bare state snapshot. +`UnifiedResourceSnapshotWithStaleThresholds` hands them, with the seed, to +consumers that rebuild a registry from the monitor's resources (the resources +API reads each tenant monitor's), and the websocket broadcast coalesces split +host rows with them, so a coalesced row prefers readings by the same +freshness rule and thresholds as the view's own merges. Regression coverage: +`TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds` and +`TestUnifiedViewCarriesItsGenerationsStaleThresholds` in +`internal/monitoring/monitor_host_agents_test.go`. PBS and PMG configured instances also have one monitoring-owned runtime resource identity constructor. Poll publication, connection status, setup and auto-registration checks, canonical alias resolution, and alert-policy bridges diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index ccb8f2ea1d..1855dc3029 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -6506,6 +6506,15 @@ unchanged and older builds ignore the fields. `internal/api/ai_handlers.go` reads them for the usage export only. No new file, directory, retention window, backup, migration or recovery authority is introduced. +### Resource freshness thresholds open no storage or recovery path + +`internal/api/resourceapi/resources.go` builds the `/api/resources` registry +with the stale thresholds its seed was judged by, and +`internal/api/router_helpers.go` hands a tenant monitor's seed and thresholds +to it. +Both are read-path changes to source freshness and presentation; no storage, +retention, backup or recovery path is added or moved. + ### Webhook configuration never persists the API mask `internal/api/alerting/notifications.go` resolves each `***REDACTED***` diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index c284a54e4a..b59dacb186 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -1615,6 +1615,45 @@ 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. +The resources API's registry rebuild is resource seeding too, and the seed +carries the thresholds its own registry generation judged it by. The monitor +adapter records them per generation (`MonitorAdapter.StaleThresholds`): +thresholds set since a rebuild wait for the next one, and the read-state +overlays and live supplemental refreshes of a generation are judged by that +generation's thresholds, since a listing already marked stale cannot be +cleared by re-judging it. A monitor view lists the store's current +generation together with those thresholds +(`GetAllWithMetricsTargetsAndStaleThresholds` reads both from one registry +pointer); a view the monitor builds from mock fixtures or a bare state +snapshot was judged by the defaults and carries none. `UnifiedResourceSnapshotWithStaleThresholds` +(per tenant, `UnifiedResourceSnapshotWithStaleThresholdsForTenant`) hands the +view's listing, freshness marker and thresholds to the resources API, which +builds its registry with `NewRegistryWithStaleThresholds`: every ingest or +stale pass on that registry that passes no thresholds of its own uses them, +and a cached registry built with other thresholds is rebuilt even at the same +freshness marker. Presentation coalescing picks the source of a merged host +row's readings with the metric merge's own rule (`metricSourceStale`): +`ListForPresentation` passes the thresholds its registry was constructed +with, and the websocket broadcast passes its view's, so on mock and snapshot +views both follow the defaults and still agree. Before this the resources API judged by +the defaults everywhere, so a Proxmox node polled every two minutes read +stale, with warning status, in `/api/resources` ninety seconds after its poll +while the broadcast called it online, and both views showed a silent agent's +readings on a node whose agent keeps its own registry row. Patrol's scoped +runtime state still rebuilds registries with the defaults. Regression +coverage: `TestRegistrySeededFromMonitorJudgesFreshnessByItsThresholds` and +`TestRegistryIngestThresholdsOverrideConstructionThresholds` in +`registry_merge_policy_test.go`, +`TestMonitorAdapterReportsTheThresholdsItsGenerationWasJudgedBy` in +`monitor_adapter_read_state_test.go`, +`TestUnifiedViewCarriesItsGenerationsStaleThresholds` in +`internal/monitoring/monitor_host_agents_test.go`, +`TestResourceAPIRebuildsWithTheSeedingMonitorStaleThresholds` in +`code_standards_test.go`, +`TestResourceListRebuildsWhenSeedStaleThresholdsChange` in +`internal/api/resourceapi/resources_test.go`, and +`TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast` in +`internal/api/contract_test.go`. Service-discovery readiness is a unified-resource payload contract, not a drawer-local decoration. Resource list/detail payloads that expose a diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 34c3b3bfba..625e525639 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -14983,6 +14983,182 @@ func TestContract_ResourceListUsesTenantStateProviderAtStartup(t *testing.T) { } } +// The resources API rebuilds its registry from a monitor's unified seed, so it +// judges source freshness by the thresholds that seed's registry judged it by, +// for the default monitor and a tenant monitor alike. A Proxmox estate polled +// every two minutes, read ninety seconds after its last poll, lists as the +// websocket broadcast shows it: every node online on a current poll, with the +// node's own readings over those of an agent silent for seventy-five seconds. +// With the registry defaults the API called each poll stale (warning), and +// both views showed the silent agent's readings on a node whose agent keeps +// its own row. A monitor without a resource store builds its view with the +// defaults, which call the ninety-second polls stale, and both views then +// follow the defaults: no row takes a reading from a source its own sighting +// calls stale over a live one. +func TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast(t *testing.T) { + now := time.Now().UTC() + polledMonitor := func(t *testing.T, withStore bool) *monitoring.Monitor { + t.Helper() + monitor, state, _ := newTestMonitor(t) + setUnexportedField(t, monitor, "config", &config.Config{PVEPollingInterval: 2 * time.Minute}) + polled := now.Add(-90 * time.Second) + agentSilent := now.Add(-75 * time.Second) + state.UpdateNodesForInstance("homelab", []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: polled, + }, + { + ID: "homelab-pve2", Name: "pve2", Instance: "homelab", Status: "online", CPU: 0.30, + Memory: models.Memory{Total: 64 << 30, Used: 32 << 30, Free: 32 << 30, Usage: 50}, + LastSeen: polled, + }, + { + ID: "homelab-pve3", Name: "pve3", Instance: "homelab", Status: "online", CPU: 0.20, + Memory: models.Memory{Total: 64 << 30, Used: 8 << 30, Free: 56 << 30, Usage: 12.5}, + LastSeen: polled, + }, + }) + state.UpsertHost(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: agentSilent, IntervalSeconds: 30, + }) + state.UpsertHost(models.Host{ + ID: "host-pve3", MachineID: "machine-pve3", Hostname: "pve3", + Status: "online", CPUUsage: 70, + Memory: models.Memory{Total: 64 << 30, Used: 40 << 30, Free: 24 << 30, Usage: 62.5}, + LastSeen: agentSilent, IntervalSeconds: 30, + }) + if withStore { + adapter := unifiedresources.NewMonitorAdapter(nil) + monitor.SetResourceStore(adapter) + adapter.PopulateFromSnapshot(state.GetSnapshot()) + } + return monitor + } + + type nodeRow struct { + status string + cpu, memory float64 + } + nodeNames := map[string]bool{"pve1": true, "pve2": true, "pve3": true} + broadcastRows := func(t *testing.T, monitor *monitoring.Monitor) map[string]nodeRow { + t.Helper() + rows := make(map[string]nodeRow) + for _, resource := range monitor.BuildBroadcastFrontendState().Resources { + if !nodeNames[resource.Name] { + continue + } + if _, dup := rows[resource.Name]; dup { + t.Fatalf("broadcast shows node %s twice", resource.Name) + } + row := nodeRow{status: resource.Status} + if row.status == "degraded" { + // The broadcast spells a host's warning status as degraded. + row.status = string(unifiedresources.StatusWarning) + } + if resource.CPU != nil { + row.cpu = resource.CPU.Current + } + if resource.Memory != nil { + row.memory = resource.Memory.Current + } + rows[resource.Name] = row + } + return rows + } + // listedRows also returns each row's Proxmox sighting, which the + // broadcast payload does not carry. + listedRows := func(t *testing.T, h *ResourceHandlers, orgID string) (map[string]nodeRow, map[string]string) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/resources?type=agent", nil) + if orgID != "" { + req = req.WithContext(context.WithValue(req.Context(), OrgIDContextKey, orgID)) + } + rec := httptest.NewRecorder() + h.HandleListResources(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + var response ResourcesResponse + if err := json.NewDecoder(rec.Body).Decode(&response); err != nil { + t.Fatalf("decode response: %v", err) + } + rows := make(map[string]nodeRow) + sightings := make(map[string]string) + for _, resource := range response.Data { + if !nodeNames[resource.Name] { + continue + } + if _, dup := rows[resource.Name]; dup { + t.Fatalf("resources API lists node %s twice", resource.Name) + } + row := nodeRow{status: string(resource.Status)} + if resource.Metrics != nil && resource.Metrics.CPU != nil { + row.cpu = resource.Metrics.CPU.Percent + } + if resource.Metrics != nil && resource.Metrics.Memory != nil { + row.memory = resource.Metrics.Memory.Percent + } + rows[resource.Name] = row + sightings[resource.Name] = resource.SourceStatus[unifiedresources.SourceProxmox].Status + } + return rows, sightings + } + current := map[string]nodeRow{ + "pve1": {status: "online", cpu: 12, memory: 25}, + "pve2": {status: "online", cpu: 30, memory: 50}, + "pve3": {status: "online", cpu: 20, memory: 12.5}, + } + check := func(t *testing.T, monitor *monitoring.Monitor, h *ResourceHandlers, orgID string, want map[string]nodeRow, wantSighting string) { + t.Helper() + broadcast := broadcastRows(t, monitor) + if !reflect.DeepEqual(broadcast, want) { + t.Fatalf("broadcast node rows = %+v, want %+v", broadcast, want) + } + listed, sightings := listedRows(t, h, orgID) + for name, sighting := range sightings { + if sighting != wantSighting { + t.Errorf("resources API %s Proxmox sighting = %q, want %q", name, sighting, wantSighting) + } + } + if !reflect.DeepEqual(listed, broadcast) { + t.Fatalf("resources API node rows = %+v, want the broadcast's %+v", listed, broadcast) + } + } + + t.Run("default monitor", func(t *testing.T) { + monitor := polledMonitor(t, true) + h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) + h.SetStateProvider(monitor) + check(t, monitor, h, "", current, "online") + }) + t.Run("tenant monitor", func(t *testing.T) { + tenant := polledMonitor(t, true) + defaultMonitor, _, _ := newTestMonitor(t) + mtm := &monitoring.MultiTenantMonitor{} + setUnexportedField(t, mtm, "monitors", map[string]*monitoring.Monitor{"acme": tenant}) + h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) + h.SetStateProvider(defaultMonitor) + h.SetTenantStateProvider(NewMultiTenantStateProvider(mtm, defaultMonitor)) + check(t, tenant, h, "acme", current, "online") + }) + t.Run("monitor without a resource store", func(t *testing.T) { + monitor := polledMonitor(t, false) + h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) + h.SetStateProvider(monitor) + check(t, monitor, h, "", map[string]nodeRow{ + "pve1": {status: "warning", cpu: 80, memory: 75}, + "pve2": {status: "warning", cpu: 30, memory: 50}, + "pve3": {status: "warning", cpu: 70, memory: 62.5}, + }, "stale") + }) +} + func TestContract_ResourceCapabilitiesJSONSnapshot(t *testing.T) { payload := struct { ResourceID string `json:"resourceId"` diff --git a/internal/api/resourceapi/resources.go b/internal/api/resourceapi/resources.go index 8a0a600690..b6c2bcb214 100644 --- a/internal/api/resourceapi/resources.go +++ b/internal/api/resourceapi/resources.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "maps" "net/http" "sort" "strconv" @@ -66,6 +67,22 @@ type UnifiedResourceSnapshotProvider interface { UnifiedResourceSnapshot() ([]unified.Resource, time.Time) } +// UnifiedResourceSnapshotWithStaleThresholdsProvider supplies the default +// tenant's unified-resource seed with the stale thresholds the seed's registry +// judged its sightings by (nil: the registry defaults). The registry rebuilt +// from the seed runs its own stale pass, merges and presentation coalesce, +// which must judge by the same thresholds or a source the monitor calls +// current between configured polls reads stale here. +type UnifiedResourceSnapshotWithStaleThresholdsProvider interface { + UnifiedResourceSnapshotWithStaleThresholds() ([]unified.Resource, time.Time, map[unified.DataSource]time.Duration) +} + +// TenantUnifiedResourceSnapshotWithStaleThresholdsProvider is the +// tenant-scoped form of UnifiedResourceSnapshotWithStaleThresholdsProvider. +type TenantUnifiedResourceSnapshotWithStaleThresholdsProvider interface { + UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID string) ([]unified.Resource, time.Time, map[unified.DataSource]time.Duration) +} + // ResourceDiscoveryReadinessProvider projects service-discovery state onto a // unified resource without coupling the resource API to discovery storage. type ResourceDiscoveryReadinessProvider interface { @@ -77,6 +94,9 @@ type registrySeed struct { resources []unified.Resource lastUpdate time.Time unifiedSource bool + // staleThresholds are those the seed's registry judged it by; nil + // means the registry defaults. + staleThresholds map[unified.DataSource]time.Duration } // SupplementalRecordsProvider provides out-of-band ingest records for a specific source. @@ -1087,7 +1107,8 @@ func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, e h.cacheMu.Lock() entry, ok := h.registryCache[key] - if ok && entry.registry != nil && !seed.lastUpdate.IsZero() && entry.lastUpdate.Equal(seed.lastUpdate) { + if ok && entry.registry != nil && !seed.lastUpdate.IsZero() && entry.lastUpdate.Equal(seed.lastUpdate) && + maps.Equal(entry.staleThresholds, seed.staleThresholds) { h.cacheMu.Unlock() return entry.registry, nil } @@ -1100,7 +1121,11 @@ func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, e } h.supplementalMu.RUnlock() - registry := unified.NewRegistry(store) + // The rebuild runs its own stale pass, merges and presentation coalesce, + // so it judges freshness by the thresholds the seed was judged by; with + // the defaults a Proxmox node polled every two minutes read stale here + // between polls while the monitor called it current. + registry := unified.NewRegistryWithStaleThresholds(store, seed.staleThresholds) ownedSources := supplementalSnapshotOwnedSources(supplementalProviders, orgID) supplementalSources := sortedSupplementalSources(supplementalProviders) if seed.unifiedSource { @@ -1143,7 +1168,7 @@ func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, e if h.registryCache == nil { h.registryCache = make(map[string]registryCacheEntry) } - h.registryCache[key] = registryCacheEntry{registry: registry, lastUpdate: seed.lastUpdate} + h.registryCache[key] = registryCacheEntry{registry: registry, lastUpdate: seed.lastUpdate, staleThresholds: seed.staleThresholds} h.cacheMu.Unlock() return registry, nil @@ -1182,14 +1207,28 @@ func (h *QueryService) registrySeed(orgID string) (registrySeed, error) { if h.tenantStateProvider == nil { return seed, errors.New("tenant state provider unavailable") } - resources, lastUpdate := h.tenantStateProvider.UnifiedResourceSnapshotForTenant(orgID) - seed.resources = resources - seed.lastUpdate = lastUpdate + if provider, ok := any(h.tenantStateProvider).(TenantUnifiedResourceSnapshotWithStaleThresholdsProvider); ok { + var thresholds map[unified.DataSource]time.Duration + seed.resources, seed.lastUpdate, thresholds = provider.UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID) + seed.staleThresholds = maps.Clone(thresholds) + } else { + seed.resources, seed.lastUpdate = h.tenantStateProvider.UnifiedResourceSnapshotForTenant(orgID) + } seed.unifiedSource = true return seed, nil } - if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotProvider); ok { + if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotWithStaleThresholdsProvider); ok { + resources, lastUpdate, thresholds := provider.UnifiedResourceSnapshotWithStaleThresholds() + if len(resources) > 0 || !lastUpdate.IsZero() { + seed.resources = resources + seed.lastUpdate = lastUpdate + seed.unifiedSource = true + // Owned copy: the cache compares it with the next seed's. + seed.staleThresholds = maps.Clone(thresholds) + return seed, nil + } + } else if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotProvider); ok { resources, lastUpdate := provider.UnifiedResourceSnapshot() if len(resources) > 0 || !lastUpdate.IsZero() { seed.resources = resources @@ -1655,6 +1694,9 @@ type StorageIncidentSection struct { type registryCacheEntry struct { registry *unified.ResourceRegistry lastUpdate time.Time + // staleThresholds are those the registry was built with; a seed judged + // by others needs a rebuild even at the same lastUpdate. + staleThresholds map[unified.DataSource]time.Duration // rawList and presentation are lazily built once per registry generation // and SHARED between requests. Both the slices and the nested data of // their elements are read-only: take a flat copy (flatCopyResources) diff --git a/internal/api/resourceapi/resources_test.go b/internal/api/resourceapi/resources_test.go index 50a59ddba2..cd5009d2f8 100644 --- a/internal/api/resourceapi/resources_test.go +++ b/internal/api/resourceapi/resources_test.go @@ -694,6 +694,72 @@ func TestResourceListInvalidatesUnifiedSeedCacheOnFreshnessChange(t *testing.T) } } +type staleThresholdSeedProvider struct { + resources []unified.Resource + freshness time.Time + thresholds map[unified.DataSource]time.Duration +} + +func (p *staleThresholdSeedProvider) ReadSnapshot() models.StateSnapshot { + return models.StateSnapshot{} +} + +func (p *staleThresholdSeedProvider) UnifiedResourceSnapshotWithStaleThresholds() ([]unified.Resource, time.Time, map[unified.DataSource]time.Duration) { + out := make([]unified.Resource, len(p.resources)) + copy(out, p.resources) + return out, p.freshness, p.thresholds +} + +// A cached registry was built with one seed's stale thresholds; a seed judged +// by others needs a rebuild even when its freshness marker is unchanged. The +// provider changes its threshold map in place, so the cache must keep its own +// copy to notice. +func TestResourceListRebuildsWhenSeedStaleThresholdsChange(t *testing.T) { + now := time.Now().UTC() + polled := now.Add(-90 * time.Second) + provider := &staleThresholdSeedProvider{ + resources: []unified.Resource{{ + ID: "agent-pve2", Type: unified.ResourceTypeAgent, Name: "pve2", Status: unified.StatusOnline, + LastSeen: polled, UpdatedAt: polled, + Sources: []unified.DataSource{unified.SourceProxmox}, + SourceStatus: map[unified.DataSource]unified.SourceStatus{unified.SourceProxmox: {Status: "online", LastSeen: polled}}, + Proxmox: &unified.ProxmoxData{NodeName: "pve2", Instance: "homelab"}, + Identity: unified.ResourceIdentity{Hostnames: []string{"pve2"}}, + }}, + freshness: now, + thresholds: map[unified.DataSource]time.Duration{unified.SourceProxmox: 4 * time.Minute}, + } + h := NewQueryService(&config.Config{DataPath: t.TempDir()}) + h.SetStateProvider(provider) + + list := func(t *testing.T) unified.Resource { + t.Helper() + rec := httptest.NewRecorder() + h.HandleListResources(rec, httptest.NewRequest(http.MethodGet, "/api/resources?type=agent", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + var response ResourcesResponse + if err := json.NewDecoder(rec.Body).Decode(&response); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(response.Data) != 1 { + t.Fatalf("resources = %#v, want the one node", response.Data) + } + return response.Data[0] + } + + if got := list(t); got.Status != unified.StatusOnline || got.SourceStatus[unified.SourceProxmox].Status != "online" { + t.Fatalf("under a four-minute threshold: status %q, Proxmox sighting %q; want online for both", + got.Status, got.SourceStatus[unified.SourceProxmox].Status) + } + provider.thresholds[unified.SourceProxmox] = 60 * time.Second + if got := list(t); got.SourceStatus[unified.SourceProxmox].Status != "stale" { + t.Fatalf("under sixty seconds at the same freshness: Proxmox sighting %q, want stale; the cached registry was reused", + got.SourceStatus[unified.SourceProxmox].Status) + } +} + func TestResourceListMergesOneSidedLinkedHostWhenHostnameCorroborates(t *testing.T) { now := time.Now().UTC() node := models.Node{ diff --git a/internal/api/router_helpers.go b/internal/api/router_helpers.go index 024110d0db..63eb2887b3 100644 --- a/internal/api/router_helpers.go +++ b/internal/api/router_helpers.go @@ -5,6 +5,7 @@ import ( "net/http" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/api/resourceapi" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rs/zerolog/log" @@ -80,6 +81,25 @@ func (p *MultiTenantStateProvider) UnifiedResourceSnapshotForTenant(orgID string return monitor.UnifiedResourceSnapshot() } +// The resources API reads both seeds-with-thresholds through optional +// interface checks, so a signature drift would silently restore the registry +// defaults. +var ( + _ resourceapi.UnifiedResourceSnapshotWithStaleThresholdsProvider = (*monitoring.Monitor)(nil) + _ resourceapi.TenantUnifiedResourceSnapshotWithStaleThresholdsProvider = (*MultiTenantStateProvider)(nil) +) + +// UnifiedResourceSnapshotWithStaleThresholdsForTenant returns a tenant's +// unified-resource seed with the stale thresholds its registry judged it by, +// read from one monitor in one call. +func (p *MultiTenantStateProvider) UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID string) ([]unifiedresources.Resource, time.Time, map[unifiedresources.DataSource]time.Duration) { + monitor := p.monitorForTenant(orgID) + if monitor == nil { + return nil, time.Time{}, nil + } + return monitor.UnifiedResourceSnapshotWithStaleThresholds() +} + // ResourceHealthAlertsForTenant returns only the minimal alert facts needed by // the resource health projection, resolved from the same tenant monitor as the // resource seed. diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index e9646c53c8..f723e78dc7 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -650,8 +650,9 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) { for _, snippet := range []string{ "metricsTargetResolver := unifiedView.metricsTargets", "metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)", - "broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources)", - "if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {", + "broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources, unifiedView.staleThresholds)", + "if coalesced, ok := coalescer.CoalesceForPresentation(resources, staleThresholds); ok {", + "return unifiedresources.CoalescePresentationHostResourcesWithStaleThresholds(resources, staleThresholds)", "broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)", "attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)", "frontendState.CapabilityCatalog = broadcastCatalogs.capabilities", diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 550f922c64..be4573eb4b 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -4479,7 +4479,9 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot mode if metricsTargetResolver == nil { metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState) } - broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources) + // Coalescing picks each merged metric's source by freshness, judged by + // the thresholds the view's own registry judged its sightings by. + broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources, unifiedView.staleThresholds) // Coalescing owns the outer slice. Decorate that one projection in place, // not three full-resource copies; nested store data is still read-only. healthAlerts := resourceHealthAlerts(frontendState.ActiveAlerts) @@ -4505,7 +4507,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot mode // presentationCoalescingResourceStore is implemented by resource stores whose // registry can hold the operator's merge exclusions (MonitorAdapter). type presentationCoalescingResourceStore interface { - CoalesceForPresentation(resources []unifiedresources.Resource) ([]unifiedresources.Resource, bool) + CoalesceForPresentation(resources []unifiedresources.Resource, thresholds map[unifiedresources.DataSource]time.Duration) ([]unifiedresources.Resource, bool) } // coalesceResourcesForPresentation applies the presentation host coalesce with @@ -4515,10 +4517,16 @@ type presentationCoalescingResourceStore interface { // the rows supplies them when its registry is store-backed, as the resource // store's adapter and a host-continuity overlay are; the mock view and other // views built from an unified list carry none, so the resource store's -// adapter supplies them instead. -func (m *Monitor) coalesceResourcesForPresentation(readState unifiedresources.ReadState, resources []unifiedresources.Resource) []unifiedresources.Resource { +// adapter supplies them instead. staleThresholds are those the listing's +// registry generation judged its sightings by; the coalesce judges each +// merged metric's freshness by them. +func (m *Monitor) coalesceResourcesForPresentation( + readState unifiedresources.ReadState, + resources []unifiedresources.Resource, + staleThresholds map[unifiedresources.DataSource]time.Duration, +) []unifiedresources.Resource { if coalescer, ok := readState.(presentationCoalescingResourceStore); ok { - if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok { + if coalesced, ok := coalescer.CoalesceForPresentation(resources, staleThresholds); ok { return coalesced } } @@ -4527,12 +4535,12 @@ func (m *Monitor) coalesceResourcesForPresentation(readState unifiedresources.Re store := m.resourceStore m.mu.RUnlock() if coalescer, ok := store.(presentationCoalescingResourceStore); ok { - if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok { + if coalesced, ok := coalescer.CoalesceForPresentation(resources, staleThresholds); ok { return coalesced } } } - return unifiedresources.CoalescePresentationHostResources(resources) + return unifiedresources.CoalescePresentationHostResourcesWithStaleThresholds(resources, staleThresholds) } // GetLiveStateSnapshot returns the underlying monitor state snapshot without @@ -5086,6 +5094,9 @@ type monitorUnifiedStateView struct { readState unifiedresources.ReadState freshness time.Time metricsTargets MetricsTargetResourceStore + // staleThresholds are those the listed resources' registry generation + // was judged by; nil means the registry defaults. + staleThresholds map[unifiedresources.DataSource]time.Duration } type unifiedResourceReadStateLister interface { @@ -5155,9 +5166,10 @@ func (m *Monitor) unifiedStateViewWithStandaloneHostContinuity(view monitorUnifi return view } - resources, targets := unifiedProjectionResources(lister) + resources, targets, thresholds := unifiedProjectionResourcesWithStaleThresholds(lister) view.resources = resources view.metricsTargets = targets + view.staleThresholds = thresholds if view.freshness.IsZero() { view.freshness = latestUnifiedResourceLastSeen(resources) } @@ -5263,11 +5275,11 @@ func unifiedResourceFreshness(store ResourceStoreInterface, state *models.State) // associated freshness marker. In mock mode it returns the shared mock // unified-resource fixture graph rather than the live resource store. func (m *Monitor) UnifiedResourceSnapshot() ([]unifiedresources.Resource, time.Time) { - view := m.currentUnifiedStateView() // REST /api/resources seeds its registry from this snapshot. Apply the // same user-metadata hydration (container customUrl) as the websocket // broadcast path, so the two payload shapes cannot drift. - return m.applyPersistedMetadataToUnifiedResources(view.resources), view.freshness + resources, freshness, _ := m.UnifiedResourceSnapshotWithStaleThresholds() + return resources, freshness } // GetUnifiedReadState returns a typed unified read-state provider when the diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 268d542f9d..e439c9fe06 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -279,6 +279,83 @@ func TestBroadcastSlimmingKeepsHostAgentSupersededIdentityResolvable(t *testing. } } +// The broadcast coalesces an agent that keeps its own registry row into its +// Proxmox node by hostname, and picks each reading's source by freshness +// judged by the configured thresholds, as the registry's own merges are. A +// node polled every two minutes, read ninety seconds after its poll, shows its +// own readings rather than those of its agent, silent for seventy-five +// seconds. Polled at the default cadence, the node is stale at ninety seconds +// too, so the agent's readings stand. +func TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds(t *testing.T) { + now := time.Now().UTC() + for _, tc := range []struct { + name string + interval time.Duration + cpu, memory float64 + }{ + {"two-minute polls", 2 * time.Minute, 20, 12.5}, + {"default polls", 10 * time.Second, 70, 62.5}, + } { + t.Run(tc.name, func(t *testing.T) { + m := &Monitor{state: models.NewState(), config: &config.Config{PVEPollingInterval: tc.interval}} + m.state.UpdateNodesForInstance("homelab", []models.Node{{ + ID: "homelab-pve3", Name: "pve3", Instance: "homelab", Status: "online", CPU: 0.20, + Memory: models.Memory{Total: 64 << 30, Used: 8 << 30, Free: 56 << 30, Usage: 12.5}, + LastSeen: now.Add(-90 * time.Second), + }}) + m.state.UpsertHost(models.Host{ + ID: "host-pve3", MachineID: "machine-pve3", Hostname: "pve3", Status: "online", CPUUsage: 70, + Memory: models.Memory{Total: 64 << 30, Used: 40 << 30, Free: 24 << 30, Usage: 62.5}, + LastSeen: now.Add(-75 * time.Second), IntervalSeconds: 30, + }) + m.SetResourceStore(unifiedresources.NewMonitorAdapter(nil)) + + var rows []models.ResourceFrontend + for _, resource := range m.BuildBroadcastFrontendState().Resources { + if resource.Name == "pve3" { + rows = append(rows, resource) + } + } + if len(rows) != 1 { + t.Fatalf("broadcast shows %d pve3 rows, want the node and its agent coalesced into one", len(rows)) + } + if rows[0].CPU == nil || rows[0].Memory == nil { + t.Fatalf("pve3 lost its readings: cpu %+v memory %+v", rows[0].CPU, rows[0].Memory) + } + if rows[0].CPU.Current != tc.cpu || rows[0].Memory.Current != tc.memory { + t.Fatalf("pve3 cpu %.1f%%, memory %.1f%%; want %.1f%% and %.1f%%", + rows[0].CPU.Current, rows[0].Memory.Current, tc.cpu, tc.memory) + } + }) + } +} + +// A monitor view pairs its listing with the thresholds that registry +// generation was judged by. Thresholds set on the store since its last +// rebuild belong to the next generation, so the view and the seed it hands +// the resources API keep the ones the listing was judged by. +func TestUnifiedViewCarriesItsGenerationsStaleThresholds(t *testing.T) { + now := time.Now().UTC() + m := &Monitor{state: models.NewState(), config: &config.Config{PVEPollingInterval: 2 * time.Minute}} + m.state.UpdateNodesForInstance("homelab", []models.Node{{ + ID: "homelab-pve2", Name: "pve2", Instance: "homelab", Status: "online", CPU: 0.30, + Memory: models.Memory{Total: 64 << 30, Used: 32 << 30, Free: 32 << 30, Usage: 50}, + LastSeen: now.Add(-90 * time.Second), + }}) + adapter := unifiedresources.NewMonitorAdapter(nil) + m.SetResourceStore(adapter) + adapter.PopulateFromSnapshot(m.state.GetSnapshot()) + adapter.SetStaleThresholds(map[unifiedresources.DataSource]time.Duration{unifiedresources.SourceProxmox: time.Minute}) + + resources, _, thresholds := m.UnifiedResourceSnapshotWithStaleThresholds() + if got := thresholds[unifiedresources.SourceProxmox]; got != 4*time.Minute { + t.Fatalf("seed thresholds = %v, want the four minutes its generation was judged by", thresholds) + } + if len(resources) != 1 || resources[0].SourceStatus[unifiedresources.SourceProxmox].Status != "online" { + t.Fatalf("seed = %+v, want pve2 with the Proxmox sighting its generation judged online", resources) + } +} + func TestHostAgentCorrelationKeepsSameNamedProxmoxProvidersDistinct(t *testing.T) { now := time.Now() state := &models.State{ @@ -6926,6 +7003,10 @@ func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargets() ([]unified s.reads++ return s.MonitorAdapter.GetAllWithMetricsTargets() } +func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargetsAndStaleThresholds() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget, map[unifiedresources.DataSource]time.Duration) { + s.reads++ + return s.MonitorAdapter.GetAllWithMetricsTargetsAndStaleThresholds() +} func (*broadcastProjectionCountingStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool { return false } @@ -7079,10 +7160,10 @@ func TestBroadcastPresentationCoalescePrefersTheListingReadState(t *testing.T) { overlay := unifiedresources.NewRegistry(store) overlay.IngestResources(resources) listing := unifiedresources.NewMonitorAdapter(overlay) - if got := len(m.coalesceResourcesForPresentation(listing, listing.GetAll())); got != 2 { + if got := len(m.coalesceResourcesForPresentation(listing, listing.GetAll(), nil)); got != 2 { t.Fatalf("coalesced %d rows with the listing's exclusion, want 2", got) } - if got := len(m.coalesceResourcesForPresentation(nil, listing.GetAll())); got != 1 { + if got := len(m.coalesceResourcesForPresentation(nil, listing.GetAll(), nil)); got != 1 { t.Fatalf("without a listing read state the store generation decides: %d rows, want 1", got) } @@ -7090,7 +7171,7 @@ func TestBroadcastPresentationCoalescePrefersTheListingReadState(t *testing.T) { // operator decisions, so the resource store's adapter supplies them. mockView := monitorUnifiedStateViewFromResources(resources, now) withSplit := &Monitor{resourceStore: listing} - if got := len(withSplit.coalesceResourcesForPresentation(mockView.readState, mockView.resources)); got != 2 { + if got := len(withSplit.coalesceResourcesForPresentation(mockView.readState, mockView.resources, mockView.staleThresholds)); got != 2 { t.Fatalf("mock-style view coalesced %d rows, want the store's split to keep 2", got) } } diff --git a/internal/monitoring/resource_stale_thresholds.go b/internal/monitoring/resource_stale_thresholds.go index 9e5d639806..be89dcea08 100644 --- a/internal/monitoring/resource_stale_thresholds.go +++ b/internal/monitoring/resource_stale_thresholds.go @@ -57,6 +57,38 @@ func resourceStaleThresholdsForConfig( return thresholds } +// UnifiedResourceSnapshotWithStaleThresholds is UnifiedResourceSnapshot +// plus the stale thresholds the snapshot's registry judged its sightings by. +// A consumer that rebuilds a registry from the snapshot, as the resources API +// does, judges by the same thresholds, so its stale pass and presentation +// agree with the monitor's views. +func (m *Monitor) UnifiedResourceSnapshotWithStaleThresholds() ([]unifiedresources.Resource, time.Time, map[unifiedresources.DataSource]time.Duration) { + view := m.currentUnifiedStateView() + return m.applyPersistedMetadataToUnifiedResources(view.resources), view.freshness, view.staleThresholds +} + +// unifiedResourceThresholdProjectionLister lists a read state with the stale +// thresholds the listed registry generation was judged by (MonitorAdapter, +// and any wrapper embedding one). +type unifiedResourceThresholdProjectionLister interface { + GetAllWithMetricsTargetsAndStaleThresholds() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget, map[unifiedresources.DataSource]time.Duration) +} + +// unifiedProjectionResourcesWithStaleThresholds lists a read state with the +// thresholds the listed generation was judged by: the resource store's +// configured ones, or nil (the registry defaults) for a read state that +// reports none, such as a view built from mock fixtures or a bare snapshot. +func unifiedProjectionResourcesWithStaleThresholds( + lister unifiedResourceReadStateLister, +) ([]unifiedresources.Resource, MetricsTargetResourceStore, map[unifiedresources.DataSource]time.Duration) { + if projection, ok := lister.(unifiedResourceThresholdProjectionLister); ok { + resources, targets, thresholds := projection.GetAllWithMetricsTargetsAndStaleThresholds() + return resources, projectionMetricsTargets(targets), thresholds + } + resources, targets := unifiedProjectionResources(lister) + return resources, targets, nil +} + func (m *Monitor) resourceStaleThresholds() map[unifiedresources.DataSource]time.Duration { if m == nil { return ResourceStaleThresholdsForConfig(nil) diff --git a/internal/unifiedresources/code_standards_test.go b/internal/unifiedresources/code_standards_test.go index 696a6ce507..8cc0b23cf7 100644 --- a/internal/unifiedresources/code_standards_test.go +++ b/internal/unifiedresources/code_standards_test.go @@ -729,6 +729,34 @@ func TestResourceAPIUsesCanonicalTenantUnifiedSeed(t *testing.T) { } } +// The resources API rebuilds its registry from a monitor's seed, so the +// rebuild must judge source freshness by the thresholds that seed was judged +// by, delivered with the seed in one call, and must not reuse a registry built +// with other thresholds. A bare NewRegistry falls back to the defaults and +// marks a slowly polled source stale between its configured polls. +func TestResourceAPIRebuildsWithTheSeedingMonitorStaleThresholds(t *testing.T) { + data, err := os.ReadFile(filepath.Join("..", "api", "resourceapi", "resources.go")) + if err != nil { + t.Fatalf("failed to read resources.go: %v", err) + } + source := string(data) + + for _, snippet := range []string{ + "registry := unified.NewRegistryWithStaleThresholds(store, seed.staleThresholds)", + "seed.resources, seed.lastUpdate, thresholds = provider.UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID)", + "resources, lastUpdate, thresholds := provider.UnifiedResourceSnapshotWithStaleThresholds()", + "seed.staleThresholds = maps.Clone(thresholds)", + "maps.Equal(entry.staleThresholds, seed.staleThresholds)", + } { + if !strings.Contains(source, snippet) { + t.Fatalf("internal/api/resourceapi/resources.go must contain %q", snippet) + } + } + if strings.Contains(source, "unified.NewRegistry(store)") { + t.Fatal("internal/api/resourceapi/resources.go must not rebuild a tenant registry with the default stale thresholds") + } +} + // Report-merge undoes an operator link through the registry's fold record. A // candidate ID derived from the merged resource's type names neither side of // a link, and the resources API's registry, seeded from the monitor's diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index cced6ca6e2..fa996645e2 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -26,7 +26,11 @@ type MonitorAdapter struct { activeAlerts []models.Alert lastRebuiltAt time.Time staleThresholds map[DataSource]time.Duration - legacyHistory legacyHistoryBackfill + // registryStaleThresholds are the thresholds the current registry + // generation was built and judged with. staleThresholds may already hold + // newer ones, which reach the registry at the next rebuild. + registryStaleThresholds map[DataSource]time.Duration + legacyHistory legacyHistoryBackfill overlays overlayReadStateCache } @@ -64,8 +68,9 @@ func NewMonitorAdapterWithStaleThresholds(registry *ResourceRegistry, staleThres } return &MonitorAdapter{ - registry: registry, - staleThresholds: cloneStaleThresholds(staleThresholds), + registry: registry, + staleThresholds: cloneStaleThresholds(staleThresholds), + registryStaleThresholds: cloneStaleThresholds(staleThresholds), } } @@ -80,6 +85,20 @@ func (a *MonitorAdapter) SetStaleThresholds(staleThresholds map[DataSource]time. a.mu.Unlock() } +// StaleThresholds returns the source freshness thresholds the current +// registry generation was judged by; nil means the registry defaults. +// Thresholds set since that rebuild are not reported until the next one +// judges a generation by them. +func (a *MonitorAdapter) StaleThresholds() map[DataSource]time.Duration { + if a == nil { + return nil + } + a.mu.RLock() + thresholds := cloneStaleThresholds(a.registryStaleThresholds) + a.mu.RUnlock() + return thresholds +} + func (a *MonitorAdapter) currentStaleThresholds() map[DataSource]time.Duration { if a == nil { return nil @@ -117,7 +136,10 @@ func readStateWithRecords(readState ReadState, source DataSource, records []Inge return readState } - registry := adapter.currentRegistry() + // The overlay re-judges the generation's listing, so it takes the + // thresholds that generation was judged by, not ones set since: those + // cannot clear a stale verdict the listing already carries. + registry, thresholds := adapter.currentRegistryAndStaleThresholds() if registry == nil { return readState } @@ -133,7 +155,6 @@ 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, thresholds) overlay := NewMonitorAdapterWithStaleThresholds(cloned, thresholds) @@ -184,6 +205,19 @@ func (c *overlayReadStateCache) store(registry *ResourceRegistry, rebuiltAt time c.overlay = overlay } +// currentRegistryAndStaleThresholds reads the current registry generation +// together with the thresholds it was judged by. +func (a *MonitorAdapter) currentRegistryAndStaleThresholds() (*ResourceRegistry, map[DataSource]time.Duration) { + if a == nil { + return nil, nil + } + a.mu.RLock() + registry := a.registry + thresholds := cloneStaleThresholds(a.registryStaleThresholds) + a.mu.RUnlock() + return registry, thresholds +} + func (a *MonitorAdapter) currentRegistry() *ResourceRegistry { if a == nil { return nil @@ -496,6 +530,7 @@ func (a *MonitorAdapter) replaceRegistryLocked(snapshot models.StateSnapshot, re a.mu.Lock() a.registry = rebuilt + a.registryStaleThresholds = staleThresholds a.activeAlerts = append([]models.Alert(nil), snapshot.ActiveAlerts...) a.lastRebuiltAt = rebuiltAt a.mu.Unlock() @@ -571,6 +606,19 @@ func (a *MonitorAdapter) GetAllWithMetricsTargets() ([]Resource, map[string]Metr return registry.ListWithMetricsTargets() } +// GetAllWithMetricsTargetsAndStaleThresholds is GetAllWithMetricsTargets +// plus the stale thresholds the listed registry generation was judged by, +// read with the registry pointer, so a rebuild publishing meanwhile cannot +// pair one generation's listing with another's thresholds. +func (a *MonitorAdapter) GetAllWithMetricsTargetsAndStaleThresholds() ([]Resource, map[string]MetricsTarget, map[DataSource]time.Duration) { + registry, thresholds := a.currentRegistryAndStaleThresholds() + if registry == nil { + return nil, nil, nil + } + resources, targets := registry.ListWithMetricsTargets() + return resources, targets, thresholds +} + func (a *MonitorAdapter) unifiedAIAdapter() *UnifiedAIAdapter { registry := a.currentRegistry() if registry == nil { @@ -684,12 +732,14 @@ func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records a.mutationMu.Lock() defer a.mutationMu.Unlock() - registry := a.currentRegistry() + // Records merge into the current generation, so they are judged by the + // thresholds it was; thresholds set since wait for the next rebuild. + registry, thresholds := a.currentRegistryAndStaleThresholds() if registry == nil || len(records) == 0 || strings.TrimSpace(string(source)) == "" { return } before := registry.List() - registry.IngestRecordsWithStaleThresholds(source, records, a.currentStaleThresholds()) + registry.IngestRecordsWithStaleThresholds(source, records, thresholds) 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 55a8298d70..0ab82b5b43 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -628,6 +628,61 @@ func TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds(t *testing.T) { }) } +// The adapter reports the thresholds its current registry generation was +// judged by. Thresholds set since then wait for the rebuild that judges a +// generation by them, so a consumer rebuilding from the current listing never +// re-judges it by thresholds it was not judged by. A read-state overlay of +// that generation is judged by the same thresholds: re-judging its listing by +// newer, shorter ones would mark polls stale that the generation calls +// current. +func TestMonitorAdapterReportsTheThresholdsItsGenerationWasJudgedBy(t *testing.T) { + snapshot := staleThresholdEstateSnapshot(time.Now().UTC()) + configured := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute} + adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), configured) + adapter.PopulateFromSnapshot(snapshot) + if got := adapter.StaleThresholds(); got[SourceProxmox] != 4*time.Minute { + t.Fatalf("after the rebuild StaleThresholds = %v, want the four minutes it was judged by", got) + } + + adapter.SetStaleThresholds(map[DataSource]time.Duration{SourceProxmox: 60 * time.Second}) + if got := adapter.StaleThresholds(); got[SourceProxmox] != 4*time.Minute { + t.Fatalf("before the next rebuild StaleThresholds = %v, want the current generation's four minutes", got) + } + for name, row := range presentedNodeRows(t, adapter.GetAll()) { + if poll := row.SourceStatus[SourceProxmox]; poll.Status != "online" { + t.Fatalf("current generation %s Proxmox sighting = %q, want online as judged", name, poll.Status) + } + } + overlay, ok := ReadStateWithRecords(adapter, SourceAgent, []IngestRecord{{ + SourceID: "host-elsewhere", + Resource: Resource{ + Type: ResourceTypeAgent, Name: "elsewhere", Status: StatusOnline, LastSeen: time.Now().UTC(), + Agent: &AgentData{AgentID: "host-elsewhere", Hostname: "elsewhere"}, + }, + Identity: ResourceIdentity{MachineID: "machine-elsewhere", Hostnames: []string{"elsewhere"}}, + }}).(*MonitorAdapter) + if !ok || overlay == adapter { + t.Fatal("ReadStateWithRecords did not build an overlay") + } + if got := overlay.StaleThresholds(); got[SourceProxmox] != 4*time.Minute { + t.Fatalf("overlay StaleThresholds = %v, want the generation's four minutes", got) + } + overlayRows := presentedNodeRows(t, overlay.GetAll()) + if len(overlayRows) != 3 { + t.Fatalf("overlay lists %d Proxmox node rows, want 3", len(overlayRows)) + } + for name, row := range overlayRows { + if poll := row.SourceStatus[SourceProxmox]; poll.Status != "online" { + t.Fatalf("overlay %s Proxmox sighting = %q, want online as the generation judged it", name, poll.Status) + } + } + + adapter.PopulateFromSnapshot(snapshot) + if got := adapter.StaleThresholds(); got[SourceProxmox] != 60*time.Second { + t.Fatalf("after the next rebuild StaleThresholds = %v, want the sixty seconds it was judged by", got) + } +} + 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/presentation_coalesce.go b/internal/unifiedresources/presentation_coalesce.go index fd76f1abbe..995f2dd51c 100644 --- a/internal/unifiedresources/presentation_coalesce.go +++ b/internal/unifiedresources/presentation_coalesce.go @@ -11,7 +11,18 @@ import ( // monitored host when a runtime/platform view and the Pulse agent view share a // canonical hostname. func CoalescePresentationHostResources(resources []Resource) []Resource { - return CoalescePresentationHostResourcesWithExclusions(resources, nil) + return coalescePresentationHostResources(resources, nil, nil) +} + +// CoalescePresentationHostResourcesWithStaleThresholds applies the +// presentation host coalesce, judging which source's metrics are current by +// the caller's stale thresholds: those the registry that produced the +// resources judged its sightings by. +func CoalescePresentationHostResourcesWithStaleThresholds( + resources []Resource, + thresholds map[DataSource]time.Duration, +) []Resource { + return coalescePresentationHostResources(resources, nil, thresholds) } // CoalescePresentationHostResourcesWithExclusions applies the presentation @@ -20,9 +31,17 @@ func CoalescePresentationHostResourcesWithExclusions( resources []Resource, excluded func(left, right Resource) bool, ) []Resource { - coalesced := coalescePresentationHostResourcesOnce(resources, excluded) + return coalescePresentationHostResources(resources, excluded, nil) +} + +func coalescePresentationHostResources( + resources []Resource, + excluded func(left, right Resource) bool, + thresholds map[DataSource]time.Duration, +) []Resource { + coalesced := coalescePresentationHostResourcesOnce(resources, excluded, thresholds) for len(coalesced) < len(resources) { - next := coalescePresentationHostResourcesOnce(coalesced, excluded) + next := coalescePresentationHostResourcesOnce(coalesced, excluded, thresholds) if len(next) == len(coalesced) { return refreshPresentationProxmoxChildActionAgents(next) } @@ -35,6 +54,7 @@ func CoalescePresentationHostResourcesWithExclusions( func coalescePresentationHostResourcesOnce( resources []Resource, excluded func(left, right Resource) bool, + thresholds map[DataSource]time.Duration, ) []Resource { if len(resources) == 0 { return resources @@ -74,7 +94,7 @@ func coalescePresentationHostResourcesOnce( if !shouldMergePresentationHostResources(existing, resource) { continue } - mergedResource := mergePresentationHostResources(existing, resource) + mergedResource := mergePresentationHostResources(existing, resource, thresholds) coalesced[existingIndex] = mergedResource addPresentationParentRedirect(parentRedirects, existing.ID, mergedResource.ID) addPresentationParentRedirect(parentRedirects, resource.ID, mergedResource.ID) @@ -414,7 +434,7 @@ func mergePresentationSources(left, right []DataSource) []DataSource { return merged } -func mergePresentationHostResources(left, right Resource) Resource { +func mergePresentationHostResources(left, right Resource, thresholds map[DataSource]time.Duration) Resource { primary, secondary := left, right if preferPresentationHostPrimary(right, left) { primary, secondary = right, left @@ -467,8 +487,12 @@ func mergePresentationHostResources(left, right Resource) Resource { for source, status := range primary.SourceStatus { combinedStatus[source] = status } + // Judged by the producing registry's thresholds and the rule its + // stale pass and metric merge read, so where both rows carry a + // reading this merge prefers a current source over a stale one as + // the registry's own merges do. stale := func(source DataSource) bool { - return presentationSourceStale(now, combinedStatus, source) + return metricSourceStale(now, combinedStatus, source, thresholds) } merged.Metrics = mergePresentationMetrics(merged.Metrics, secondary.Metrics, stale) } @@ -568,24 +592,6 @@ func uniquePresentationStrings(values []string) []string { return unique } -// presentationSourceStale 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 coalescing never demotes a source on missing information. -func presentationSourceStale(now time.Time, status map[DataSource]SourceStatus, source DataSource) bool { - if status == nil { - return false - } - st, ok := status[source] - if !ok || st.LastSeen.IsZero() { - return false - } - threshold, ok := defaultStaleThresholds[source] - if !ok { - threshold = 60 * time.Second - } - return now.Sub(st.LastSeen) > threshold -} - // mergePresentationMetric keeps the primary (left) value, except that a metric // from a stale source must never win over a metric from a live one. Without // this, a Proxmox node whose Pulse Agent has gone offline shows the agent's diff --git a/internal/unifiedresources/presentation_coalesce_test.go b/internal/unifiedresources/presentation_coalesce_test.go index a35335840a..fffa671d8b 100644 --- a/internal/unifiedresources/presentation_coalesce_test.go +++ b/internal/unifiedresources/presentation_coalesce_test.go @@ -390,7 +390,7 @@ func TestMonitorAdapterCoalesceForPresentationHonoursExclusions(t *testing.T) { if split { want = 2 } - rows, ok := adapter.CoalesceForPresentation(adapter.GetAll()) + rows, ok := adapter.CoalesceForPresentation(adapter.GetAll(), nil) if !ok || len(rows) != want { t.Fatalf("split=%v: adapter coalesced %d rows (ok=%v), want %d", split, len(rows), ok, want) } @@ -403,7 +403,7 @@ func TestMonitorAdapterCoalesceForPresentationHonoursExclusions(t *testing.T) { // rather than coalescing as if none existed. storeless := NewRegistry(nil) storeless.IngestResources(resources) - if _, ok := NewMonitorAdapter(storeless).CoalesceForPresentation(resources); ok { + if _, ok := NewMonitorAdapter(storeless).CoalesceForPresentation(resources, nil); ok { t.Fatal("a store-less adapter claimed to carry operator decisions") } } diff --git a/internal/unifiedresources/presentation_exclusions.go b/internal/unifiedresources/presentation_exclusions.go index 9ceb598389..b00f0dd4c2 100644 --- a/internal/unifiedresources/presentation_exclusions.go +++ b/internal/unifiedresources/presentation_exclusions.go @@ -1,5 +1,7 @@ package unifiedresources +import "time" + // presentationExclusionFilter reports whether the presentation host coalesce // must keep two resources apart because the operator split the pair (unlink // or report-merge). It reads a copy of the registry's exclusions, so callers @@ -31,14 +33,15 @@ func (rr *ResourceRegistry) presentationExclusionFilter() func(left, right Resou // listed from this adapter's current registry generation, honouring the // operator's merge exclusions through the same filter that // ResourceRegistry.ListForPresentation uses, so the websocket broadcast keeps -// a pair the operator split apart as the resources API does. It reports -// false, leaving resources untouched, when the registry has no store and so -// carries no operator decisions: the mock view and other read states built -// from an already-unified list. -func (a *MonitorAdapter) CoalesceForPresentation(resources []Resource) ([]Resource, bool) { +// a pair the operator split apart as the resources API does. Metric +// freshness is judged by thresholds: those the listing's registry generation +// judged its sightings by. It reports false, leaving resources untouched, +// when the registry has no store and so carries no operator decisions: the +// mock view and other read states built from an already-unified list. +func (a *MonitorAdapter) CoalesceForPresentation(resources []Resource, thresholds map[DataSource]time.Duration) ([]Resource, bool) { registry := a.currentRegistry() if registry == nil || registry.store == nil { return resources, false } - return CoalescePresentationHostResourcesWithExclusions(resources, registry.presentationExclusionFilter()), true + return coalescePresentationHostResources(resources, registry.presentationExclusionFilter(), thresholds), true } diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 02928ca3a0..5ced144b64 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -177,6 +177,12 @@ type ResourceRegistry struct { // stale pass do. nil outside an ingest and for callers without // thresholds, where all three use the defaults. ingestStaleThresholds map[DataSource]time.Duration + + // staleThresholds are the freshness thresholds of the monitor whose + // resources this registry holds, fixed at construction. An ingest or + // stale pass without thresholds of its own and the presentation listing + // judge sightings by them; nil means the defaults. + staleThresholds map[DataSource]time.Duration } // agentNodeCandidate pairs a resources-map key with its entry so the indexed @@ -215,6 +221,13 @@ func (rr *ResourceRegistry) ensureCanonicalMetadataLocked() { // NewRegistry creates a new registry using the provided store for overrides. func NewRegistry(store ResourceStore) *ResourceRegistry { + return NewRegistryWithStaleThresholds(store, nil) +} + +// NewRegistryWithStaleThresholds creates a registry that judges source +// freshness by the given thresholds wherever a caller passes none, for a +// registry rebuilt from a monitor's resources, such as the resources API's. +func NewRegistryWithStaleThresholds(store ResourceStore, thresholds map[DataSource]time.Duration) *ResourceRegistry { rr := &ResourceRegistry{ resources: make(map[string]*Resource), bySource: make(map[DataSource]map[string]string), @@ -222,6 +235,7 @@ func NewRegistry(store ResourceStore) *ResourceRegistry { store: store, exclusions: make(map[string]struct{}), canonicalMetadataDirty: true, + staleThresholds: cloneStaleThresholds(thresholds), } rr.bySource[SourceProxmox] = make(map[string]string) @@ -275,6 +289,7 @@ func (rr *ResourceRegistry) IngestSnapshotWithStaleThresholds(snapshot models.St } func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresholds map[DataSource]time.Duration) { + thresholds = rr.thresholdsOrOwn(thresholds) rr.mu.Lock() rr.ingestStaleThresholds = thresholds rr.mu.Unlock() @@ -658,6 +673,7 @@ func (rr *ResourceRegistry) IngestRecordsWithStaleThresholds(source DataSource, } func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool, thresholds map[DataSource]time.Duration) { + thresholds = rr.thresholdsOrOwn(thresholds) rr.mu.Lock() rr.ingestStaleThresholds = thresholds rr.mu.Unlock() @@ -916,6 +932,7 @@ func (rr *ResourceRegistry) IngestResourcesWithStaleThresholds(resources []Resou } func (rr *ResourceRegistry) ingestResources(resources []Resource, thresholds map[DataSource]time.Duration) { + thresholds = rr.thresholdsOrOwn(thresholds) seededIDs := make([]string, 0, len(resources)) for _, incoming := range resources { resource := cloneResourcePtr(&incoming) @@ -1503,10 +1520,11 @@ func (rr *ResourceRegistry) listMaterialized(withTargets bool) ([]Resource, map[ // ListForPresentation returns resources in the canonical API/broadcast // presentation shape, including top-level host coalescing that respects manual -// merge exclusions. +// merge exclusions and judges source freshness by the thresholds the registry +// was constructed with (not those an individual ingest passed). func (rr *ResourceRegistry) ListForPresentation() []Resource { resources := rr.List() - return CoalescePresentationHostResourcesWithExclusions(resources, rr.presentationExclusionFilter()) + return coalescePresentationHostResources(resources, rr.presentationExclusionFilter(), rr.staleThresholds) } // ListByType returns all resources of the provided type. @@ -1865,13 +1883,22 @@ func (rr *ResourceRegistry) Stats() ResourceStats { } // MarkStale marks sources as stale based on last seen timestamps. -// If thresholds is nil, default thresholds are used. +// If thresholds is nil, the registry's own thresholds are used. func (rr *ResourceRegistry) MarkStale(now time.Time, thresholds map[DataSource]time.Duration) { + thresholds = rr.thresholdsOrOwn(thresholds) rr.mu.Lock() defer rr.mu.Unlock() rr.markStaleLocked(now, thresholds) } +// thresholdsOrOwn returns a caller's thresholds, else the registry's own. +func (rr *ResourceRegistry) thresholdsOrOwn(thresholds map[DataSource]time.Duration) map[DataSource]time.Duration { + if thresholds != nil { + return thresholds + } + return rr.staleThresholds +} + // sourceSightingStatus derives the per-source delivery status from the last // time that source actually delivered the resource. A zero sighting means the // source has never delivered it (synthesized offline placeholders, instances diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index b48ad5d8cc..8d2cba07cc 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -356,6 +356,158 @@ func TestMetricMergeFreshnessMatchesTheStalePass(t *testing.T) { } } +// staleThresholdEstateSnapshot is a Proxmox estate polled every two minutes, +// read ninety seconds after its last poll: a node merged with its auto-linked +// agent, silent for seventy-five seconds; a node with no agent; and a node +// whose agent the registry keeps as its own row, which presentation coalesces +// into the node by hostname. +func staleThresholdEstateSnapshot(now time.Time) models.StateSnapshot { + polled := now.Add(-90 * time.Second) + agentSilent := now.Add(-75 * time.Second) + return 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: polled, + }, + { + ID: "homelab-pve2", Name: "pve2", Instance: "homelab", Status: "online", CPU: 0.30, + Memory: models.Memory{Total: 64 << 30, Used: 32 << 30, Free: 32 << 30, Usage: 50}, + LastSeen: polled, + }, + { + ID: "homelab-pve3", Name: "pve3", Instance: "homelab", Status: "online", CPU: 0.20, + Memory: models.Memory{Total: 64 << 30, Used: 8 << 30, Free: 56 << 30, Usage: 12.5}, + LastSeen: polled, + }, + }, + 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: agentSilent, IntervalSeconds: 30, + }, + { + ID: "host-pve3", MachineID: "machine-pve3", Hostname: "pve3", + Status: "online", CPUUsage: 70, + Memory: models.Memory{Total: 64 << 30, Used: 40 << 30, Free: 24 << 30, Usage: 62.5}, + LastSeen: agentSilent, IntervalSeconds: 30, + }, + }, + LastUpdate: now, + } +} + +// presentedNodeRows indexes the coalesced rows that carry a Proxmox sighting +// by name. +func presentedNodeRows(t *testing.T, resources []Resource) map[string]Resource { + t.Helper() + rows := make(map[string]Resource) + for _, resource := range resources { + if _, ok := resource.SourceStatus[SourceProxmox]; !ok { + continue + } + if _, dup := rows[resource.Name]; dup { + t.Fatalf("two presented rows carry Proxmox node %q", resource.Name) + } + rows[resource.Name] = resource + } + return rows +} + +// A registry rebuilt from the monitor's resources, as the resources API +// rebuilds one, judges source freshness by the monitor's configured +// thresholds: its stale pass, and the presentation coalesce of both views, +// agree with the monitor's own registry. With the defaults the rebuild called +// every node's poll stale at ninety seconds (warning status), and both views +// showed the silent agent's readings on the node whose agent stays its own +// row. +func TestRegistrySeededFromMonitorJudgesFreshnessByItsThresholds(t *testing.T) { + now := time.Now().UTC() + thresholds := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute} + snapshot := staleThresholdEstateSnapshot(now) + + adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), thresholds) + adapter.PopulateFromSnapshot(snapshot) + adapter.PopulateFromSnapshot(snapshot) + + broadcast := presentedNodeRows(t, CoalescePresentationHostResourcesWithStaleThresholds(adapter.GetAll(), thresholds)) + seeded := NewRegistryWithStaleThresholds(nil, thresholds) + seeded.IngestResources(adapter.GetAll()) + api := presentedNodeRows(t, seeded.ListForPresentation()) + + want := map[string]struct { + cpu, memory float64 + agent bool + }{ + "pve1": {cpu: 12, memory: 25, agent: true}, + "pve2": {cpu: 30, memory: 50}, + "pve3": {cpu: 20, memory: 12.5, agent: true}, + } + for view, rows := range map[string]map[string]Resource{"broadcast": broadcast, "resources API": api} { + if len(rows) != len(want) { + t.Fatalf("%s presents %d node rows, want %d", view, len(rows), len(want)) + } + for name, expected := range want { + row, ok := rows[name] + if !ok { + t.Fatalf("%s lost node %s", view, name) + } + if poll := row.SourceStatus[SourceProxmox]; poll.Status != "online" { + t.Errorf("%s %s Proxmox sighting = %q, want online: ninety seconds is within the configured four minutes", view, name, poll.Status) + } + if row.Status != StatusOnline { + t.Errorf("%s %s status = %q, want online", view, name, row.Status) + } + if agent, ok := row.SourceStatus[SourceAgent]; ok != expected.agent || (ok && agent.Status != "stale") { + t.Errorf("%s %s agent sighting = %+v (present %v), want a stale one present %v", view, name, agent, ok, expected.agent) + } + if row.Metrics == nil || row.Metrics.CPU == nil || row.Metrics.Memory == nil { + t.Fatalf("%s %s lost its readings: %+v", view, name, row.Metrics) + } + if cpu, memory := row.Metrics.CPU, row.Metrics.Memory; cpu.Source != SourceProxmox || cpu.Percent != expected.cpu || + memory.Source != SourceProxmox || memory.Percent != expected.memory { + t.Errorf("%s %s cpu %.1f%% from %s, memory %.1f%% from %s; want the current poll's %.1f%% and %.1f%%", + view, name, cpu.Percent, cpu.Source, memory.Percent, memory.Source, expected.cpu, expected.memory) + } + } + } + + // The fixture sits where the thresholds decide: a rebuild judged by the + // defaults calls the same polls stale. + defaults := NewRegistry(nil) + defaults.IngestResources(adapter.GetAll()) + defaultRows := presentedNodeRows(t, defaults.ListForPresentation()) + if len(defaultRows) != len(want) { + t.Fatalf("default-threshold rebuild presents %d node rows, want %d", len(defaultRows), len(want)) + } + for name, row := range defaultRows { + if poll := row.SourceStatus[SourceProxmox]; poll.Status != "stale" { + t.Fatalf("default-threshold rebuild %s Proxmox sighting = %q, want stale; the fixture no longer exercises the thresholds", name, poll.Status) + } + } +} + +// An ingest that passes thresholds of its own keeps them over the registry's. +func TestRegistryIngestThresholdsOverrideConstructionThresholds(t *testing.T) { + now := time.Now().UTC() + snapshot := staleThresholdEstateSnapshot(now) + registry := NewRegistryWithStaleThresholds(nil, map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute}) + registry.IngestSnapshotWithStaleThresholds(snapshot, map[DataSource]time.Duration{SourceProxmox: 60 * time.Second}) + rows := presentedNodeRows(t, registry.List()) + if len(rows) != 3 { + t.Fatalf("registry lists %d Proxmox node rows, want 3", len(rows)) + } + for name, row := range rows { + if poll := row.SourceStatus[SourceProxmox]; poll.Status != "stale" { + t.Fatalf("%s Proxmox sighting = %q, want stale under the ingest's own sixty seconds", name, poll.Status) + } + } +} + // 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.