From 134f2e5adaebb42f9a7bb53f3d61c2a5f4898bef Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 10:10:44 +0100 Subject: [PATCH] Refresh the resource store at most once per 2s of agent reports Every accepted agent report refreshed the canonical resource store, and every refresh covers the whole estate, so refresh work grew with agent count times estate size (#2199). The previous commit made each pass linear and cheaper; this bounds how many passes reports can trigger. Accepted host, Docker and Kubernetes reports, including rejected-report liveness, now refresh through a throttle. The first report after a quiet window is published before its handler returns. Reports within 2 seconds of the previous refresh fold into one trailing refresh at the window's end, which reads the latest state and broadcasts it. Two seconds matches readPathRegistryFreshness, the staleness read paths already accept. Agent removal and host-agent evaluation still refresh immediately. Monitor.Stop cancels a pending trailing refresh and waits for a running one before closing stores. Monitors built without New keep a zero window and refresh on every report. At 2,080 resources a refresh costs about 82 ms, so 200 agents reporting every 30 seconds went from roughly 55% of a core in report-driven refreshes to at most about 4%, however many agents report. This relaxes the agent-lifecycle guarantee that a report is visible when its handler returns: it is now visible within the window. The founder approved that trade on 24 September 2026 and the contract records it. The issue-1753 end-to-end test posts two reports back to back, so it now waits for the published rows instead of reading them at once. Refs #2199 --- .../v6/internal/subsystems/agent-lifecycle.md | 28 +++-- .../v6/internal/subsystems/monitoring.md | 3 +- ...issue1753_agent_ingest_integration_test.go | 44 +++++-- internal/monitoring/kubernetes_agents.go | 2 +- internal/monitoring/monitor.go | 113 +++++++++++++++++- internal/monitoring/monitor_agents.go | 10 +- .../monitoring/monitor_host_agents_test.go | 103 +++++++++++++++- 7 files changed, 266 insertions(+), 37 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 2dc2cee1e4..fcdd2d1559 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -53,17 +53,25 @@ is exposed as unavailable, rather than replaced with a synthetic healthy state. Delivery-log diagnostic text is a notification-attempt projection, not an agent admission or liveness verdict. Masking embedded destination credentials preserves failure context without changing agent identities, credentials, session replacement or removal policy. Consumers must use the retained failureClass and outcome as delivery evidence only; a transport error does not establish that the monitored agent is offline. -### Accepted reports refresh the canonical store synchronously — issue #2199 +### Accepted reports refresh the canonical store at most once per window — issue #2199 -An accepted host, Docker or Kubernetes report, and an agent removal, refresh the -canonical resource store before the handler returns, so ReadState consumers see -agent-backed truth without waiting for a broadcast. The #2199 reductions keep -that contract: they make each refresh pass cheaper (node-indexed guest parents -in the unified-resources contract, one registry snapshot per pass in the -monitoring contract) without deferring or coalescing reports. A pass is still -estate-wide, so its cost scales with agent report volume times estate size; -coalescing reports would trade this immediate visibility and needs its own -decision here first. +Every canonical store refresh is estate-wide, so refreshing on each accepted +report made total cost grow with agent count times estate size. Accepted host, +Docker and Kubernetes reports, including rejected-report liveness, therefore +refresh through `refreshUnifiedResourceStoreAfterAgentReport`, throttled to one +refresh per `agentReportRefreshInterval` (2 seconds, the same staleness the +read paths accept through `readPathRegistryFreshness`). The first report after +a quiet window is published before the handler returns. Reports inside the +window fold into one trailing refresh at the window's end, which reads the +latest state and broadcasts it, so a report reaches ReadState consumers within +the window rather than immediately. Agent removal and host-agent evaluation +still refresh immediately, so removed inventory retires at once. `Monitor.Stop` +cancels a pending trailing refresh and waits for a running one before closing +stores. Monitors built without `New` keep a zero window and refresh on every +report. `TestAgentReportRefreshFoldsReportsWithinWindow` and +`TestStopCancelsPendingAgentReportRefresh` pin the leading, trailing and +shutdown behaviour. The founder approved trading immediate visibility for this +bound on 24 September 2026. ### Canonical Patrol and Assistant continuation, 2026-09-07 diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 314c5786fe..9286d0f8e2 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -42,7 +42,8 @@ checks unchanged. `TestAgentReportRefreshClonesRegistryOnce` requires one clone per agent-report refresh and `TestResourceSnapshotStoreClonesOncePerPass` pins the wrapper. With node-indexed guest parents, a synthetic agent-report refresh at 2,080 resources moved from 188 to 82 ms; the pass still covers the whole -estate, and it stays synchronous (see the agent-lifecycle contract). +estate, and report-driven passes run at most once per 2-second window (see +the agent-lifecycle contract). ### Linked Pulse agent memory for Proxmox LXC — issue #2148 (22 September 2026) diff --git a/internal/api/issue1753_agent_ingest_integration_test.go b/internal/api/issue1753_agent_ingest_integration_test.go index 482d40ac89..5f0d3d9845 100644 --- a/internal/api/issue1753_agent_ingest_integration_test.go +++ b/internal/api/issue1753_agent_ingest_integration_test.go @@ -98,23 +98,43 @@ func TestIssue1753SameNameProxmoxAgentsAuthenticateAndStayDistinctEndToEnd(t *te t.Fatalf("provider links = %+v, want each site linked to its own authenticated agent", linkedByInstance) } - resources, _ := monitor.UnifiedResourceSnapshot() - byInstance := make(map[string]unifiedresources.Resource) - for _, resource := range resources { - if resource.Proxmox != nil && resource.Proxmox.NodeName != "" { - byInstance[resource.Proxmox.Instance] = resource - } - } - if len(byInstance) != 2 { - t.Fatalf("presentation provider rows = %d, want 2: %+v", len(byInstance), resources) - } - for instance, want := range map[string]struct { + wantRows := map[string]struct { agentID string name string }{ "staging": {agentID: "machine-staging", name: "Staging"}, "production": {agentID: "machine-production", name: "Production"}, - } { + } + presentationRows := func() ([]unifiedresources.Resource, map[string]unifiedresources.Resource) { + resources, _ := monitor.UnifiedResourceSnapshot() + byInstance := make(map[string]unifiedresources.Resource) + for _, resource := range resources { + if resource.Proxmox != nil && resource.Proxmox.NodeName != "" { + byInstance[resource.Proxmox.Instance] = resource + } + } + return resources, byInstance + } + rowsLinked := func(byInstance map[string]unifiedresources.Resource) bool { + for instance, want := range wantRows { + resource := byInstance[instance] + if resource.Agent == nil || resource.Agent.AgentID != want.agentID { + return false + } + } + return true + } + // Reports inside one refresh window are published together by a trailing + // refresh (agent-lifecycle contract, #2199), so wait for the canonical rows. + resources, byInstance := presentationRows() + for deadline := time.Now().Add(5 * time.Second); !rowsLinked(byInstance) && time.Now().Before(deadline); { + time.Sleep(20 * time.Millisecond) + resources, byInstance = presentationRows() + } + if len(byInstance) != 2 { + t.Fatalf("presentation provider rows = %d, want 2: %+v", len(byInstance), resources) + } + for instance, want := range wantRows { resource := byInstance[instance] if resource.Agent == nil || resource.Agent.AgentID != want.agentID || resource.Name != want.name { t.Fatalf("%s presentation row = %+v, want agent %q and name %q", instance, resource, want.agentID, want.name) diff --git a/internal/monitoring/kubernetes_agents.go b/internal/monitoring/kubernetes_agents.go index e0a75790ec..a98a1c6493 100644 --- a/internal/monitoring/kubernetes_agents.go +++ b/internal/monitoring/kubernetes_agents.go @@ -344,7 +344,7 @@ func (m *Monitor) ApplyKubernetesReport(report agentsk8s.Report, tokenRecord *co m.state.UpsertKubernetesCluster(cluster) m.state.SetConnectionHealth(kubernetesConnectionPrefix+identifier, true) m.recordKubernetesPodMetrics(cluster, receivedAt) - m.refreshUnifiedResourceStoreAfterAgentStateChange() + m.refreshUnifiedResourceStoreAfterAgentReport() return cluster, nil } diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 76c58be0f9..c71cd30883 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1301,6 +1301,11 @@ type Monitor struct { // data, which is the dominant write amplification in #1966. unifiedMetricSyncMu sync.Mutex unifiedMetricSyncLast map[unifiedMetricSampleKey]unifiedMetricSample + + // agentReportRefreshWindow folds accepted agent reports into at most one + // canonical store refresh per window; zero refreshes on every report. + agentReportRefreshWindow time.Duration + agentReportRefresh agentReportRefreshState } func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) { @@ -1738,6 +1743,7 @@ func New(cfg *config.Config) (*Monitor, error) { m := &Monitor{ config: cfg, state: models.NewState(), + agentReportRefreshWindow: agentReportRefreshInterval, pveClients: make(map[string]PVEClientInterface), pbsClients: make(map[string]*pbs.Client), pmgClients: make(map[string]*pmg.Client), @@ -5236,8 +5242,9 @@ func (s *resourceSnapshotStore) MetricsTargetForResource(resourceID string) *uni return s.targets.MetricsTargetForResource(resourceID) } -// refreshUnifiedResourceStoreAfterAgentStateChange makes accepted agent -// ingest and removal immediately visible to canonical ReadState consumers. +// refreshUnifiedResourceStoreAfterAgentStateChange makes agent removal and +// host-agent evaluation immediately visible to canonical ReadState consumers; +// accepted reports go through refreshUnifiedResourceStoreAfterAgentReport. // WebSocket broadcasts may also rebuild the store for their own hydrate path, // but client presence must never be the trigger that publishes agent-backed // runtime truth or retires removed inventory. @@ -5248,6 +5255,105 @@ func (m *Monitor) refreshUnifiedResourceStoreAfterAgentStateChange() { m.updateResourceStore(m.GetState()) } +// agentReportRefreshInterval bounds how often accepted agent reports refresh +// the canonical store. It matches readPathRegistryFreshness, the staleness the +// read paths already accept. +const agentReportRefreshInterval = readPathRegistryFreshness + +// agentReportRefreshState throttles report-driven store refreshes to a leading +// refresh plus at most one trailing refresh per window. +type agentReportRefreshState struct { + mu sync.Mutex + lastRun time.Time + running bool + pending *time.Timer + stopped bool + inFlight sync.WaitGroup +} + +// refreshUnifiedResourceStoreAfterAgentReport publishes an accepted agent +// report to the canonical store. Every refresh is estate-wide, so refreshing +// on each report made total cost grow with agents times resources (#2199). +// The first report after a quiet window still refreshes before the handler +// returns. Reports within agentReportRefreshInterval of the previous refresh +// fold into one trailing refresh at the window's end, which reads the latest +// state and broadcasts it. Agent removal and host-agent evaluation keep +// refreshing immediately through refreshUnifiedResourceStoreAfterAgentStateChange. +func (m *Monitor) refreshUnifiedResourceStoreAfterAgentReport() { + if m == nil || m.state == nil { + return + } + window := m.agentReportRefreshWindow + if window <= 0 { + m.updateResourceStore(m.GetState()) + return + } + + r := &m.agentReportRefresh + r.mu.Lock() + if r.stopped || r.pending != nil { + // A due trailing refresh reads state after this report was applied. + r.mu.Unlock() + return + } + now := time.Now() + if !r.running && now.Sub(r.lastRun) >= window { + r.running = true + r.lastRun = now + r.mu.Unlock() + m.updateResourceStore(m.GetState()) + r.mu.Lock() + r.running = false + r.mu.Unlock() + return + } + r.inFlight.Add(1) + r.pending = time.AfterFunc(max(window-now.Sub(r.lastRun), 0), m.runTrailingAgentReportRefresh) + r.mu.Unlock() +} + +func (m *Monitor) runTrailingAgentReportRefresh() { + r := &m.agentReportRefresh + defer r.inFlight.Done() + r.mu.Lock() + r.pending = nil + if r.stopped { + r.mu.Unlock() + return + } + r.running = true + r.lastRun = time.Now() + r.mu.Unlock() + + m.updateResourceStore(m.GetState()) + + r.mu.Lock() + r.running = false + r.mu.Unlock() + + // No handler follows a trailing refresh, so publish it to clients here. + m.mu.RLock() + hub := m.wsHub + m.mu.RUnlock() + if hub != nil { + m.broadcastCurrentState(hub) + } +} + +// stopAgentReportRefresh cancels a pending trailing refresh and waits for one +// already running, so none touches stores Stop is about to close. +func (m *Monitor) stopAgentReportRefresh() { + r := &m.agentReportRefresh + r.mu.Lock() + r.stopped = true + if r.pending != nil && r.pending.Stop() { + r.pending = nil + r.inFlight.Done() + } + r.mu.Unlock() + r.inFlight.Wait() +} + func recordSupplementalResourceChanges(store ResourceStoreInterface, changes []unifiedresources.ResourceChange) { if store == nil || len(changes) == 0 { return @@ -7445,6 +7551,9 @@ const guestMetadataDrainTimeout = 2 * time.Second func (m *Monitor) Stop() { log.Info().Msg("stopping monitor") + // A trailing agent-report refresh must not run against stores closed below. + m.stopAgentReportRefresh() + if m.deadMan != nil { m.deadMan.stop(time.Now().UTC(), m.alertManager) } diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 4bfe270a24..461f6b1d22 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -1743,7 +1743,7 @@ func (m *Monitor) applyRejectedHostReportLiveness( // clear an offline alert, or renew the accepted telemetry lease. m.state.SetConnectionHealth(hostConnectionPrefix+host.ID, true) m.persistHostContinuity(host, report, order) - m.refreshUnifiedResourceStoreAfterAgentStateChange() + m.refreshUnifiedResourceStoreAfterAgentReport() log.Debug(). Str("hostID", host.ID). @@ -2206,7 +2206,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con if existing.ID != "" { m.state.SetConnectionHealth(dockerConnectionPrefix+existing.ID, true) } - m.refreshUnifiedResourceStoreAfterAgentStateChange() + m.refreshUnifiedResourceStoreAfterAgentReport() log.Debug(). Str("dockerHostID", identifier). Str("sequenceId", report.SequenceID). @@ -2604,7 +2604,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con Str("dockerHost", host.Hostname). Str("sequenceId", report.SequenceID). Msg("Docker helper status processed without refreshing inventory-derived alerts or metrics") - m.refreshUnifiedResourceStoreAfterAgentStateChange() + m.refreshUnifiedResourceStoreAfterAgentReport() return host, nil } @@ -2700,7 +2700,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con Int("containers", len(containers)). Msg("Docker host report processed") - m.refreshUnifiedResourceStoreAfterAgentStateChange() + m.refreshUnifiedResourceStoreAfterAgentReport() return host, nil } @@ -3800,7 +3800,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. // committed, because ownership is checked against that host ID. m.ApplyProbeAvailabilityResults(host.ID, probeAvailabilityResultsFromReport(report.AvailabilityResults)) m.persistHostContinuity(host, report, reportOrder) - m.refreshUnifiedResourceStoreAfterAgentStateChange() + m.refreshUnifiedResourceStoreAfterAgentReport() return host, nil } diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index db06f95852..c13e1f9781 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -7,6 +7,7 @@ import ( "net" "os" "strings" + "sync/atomic" "testing" "time" @@ -6412,11 +6413,11 @@ func TestApplyHostReportHonoursRemovalBlockWhenIdentityWouldFork(t *testing.T) { // countingResourceStore counts registry clones handed out by the live store. type countingResourceStore struct { *unifiedresources.MonitorAdapter - getAll int + getAll atomic.Int32 } func (c *countingResourceStore) GetAll() []unifiedresources.Resource { - c.getAll++ + c.getAll.Add(1) return c.MonitorAdapter.GetAll() } @@ -6440,8 +6441,8 @@ func TestResourceSnapshotStoreClonesOncePerPass(t *testing.T) { store := newResourceSnapshotStore(counting) first := store.GetAll() second := store.GetAll() - if counting.getAll != 1 { - t.Fatalf("underlying GetAll calls = %d, want 1", counting.getAll) + if got := counting.getAll.Load(); got != 1 { + t.Fatalf("underlying GetAll calls = %d, want 1", got) } if len(first) != 1 || len(second) != 1 || &first[0] != &second[0] { t.Fatalf("pass consumers did not share one snapshot: %d and %d resources", len(first), len(second)) @@ -6480,7 +6481,97 @@ func TestAgentReportRefreshClonesRegistryOnce(t *testing.T) { } m.refreshUnifiedResourceStoreAfterAgentStateChange() - if counting.getAll != 1 { - t.Fatalf("registry clones for one agent report refresh = %d, want 1", counting.getAll) + if got := counting.getAll.Load(); got != 1 { + t.Fatalf("registry clones for one agent report refresh = %d, want 1", got) + } +} + +// newAgentReportThrottleMonitor returns a monitor whose report-driven store +// refreshes are throttled to window, with one online host agent. +func newAgentReportThrottleMonitor(t *testing.T, window time.Duration) (*Monitor, *countingResourceStore) { + t.Helper() + state := models.NewState() + state.UpsertHost(models.Host{ID: "agent-1", Hostname: "pve1.example", Status: "online", CPUUsage: 1, LastSeen: time.Now().UTC()}) + counting := &countingResourceStore{MonitorAdapter: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil))} + m := &Monitor{ + state: state, + resourceStore: counting, + metricsHistory: NewMetricsHistory(32, time.Hour), + agentReportRefreshWindow: window, + } + t.Cleanup(m.stopAgentReportRefresh) + return m, counting +} + +func reportAgentCPU(m *Monitor, cpu float64) { + m.state.UpsertHost(models.Host{ID: "agent-1", Hostname: "pve1.example", Status: "online", CPUUsage: cpu, LastSeen: time.Now().UTC()}) + m.refreshUnifiedResourceStoreAfterAgentReport() +} + +func publishedAgentCPU(t *testing.T, store *countingResourceStore) float64 { + t.Helper() + for _, resource := range store.MonitorAdapter.GetAll() { + if resource.Type == unifiedresources.ResourceTypeAgent && resource.Metrics != nil && resource.Metrics.CPU != nil { + return resource.Metrics.CPU.Value + } + } + t.Fatal("no agent CPU in the canonical store") + return 0 +} + +// Accepted agent reports refresh the estate-wide store at most once per +// window: the first report after a quiet window is published before the call +// returns, and reports inside the window fold into one trailing refresh that +// publishes the latest state (#2199). +func TestAgentReportRefreshFoldsReportsWithinWindow(t *testing.T) { + const window = 500 * time.Millisecond + m, store := newAgentReportThrottleMonitor(t, window) + + reportAgentCPU(m, 10) + if got := store.getAll.Load(); got != 1 { + t.Fatalf("refreshes after the leading report = %d, want 1", got) + } + if got := publishedAgentCPU(t, store); got != 10 { + t.Fatalf("leading report published CPU %v, want 10", got) + } + + for cpu := 20.0; cpu <= 60; cpu += 10 { + reportAgentCPU(m, cpu) + } + if got := store.getAll.Load(); got != 1 { + t.Fatalf("refreshes during the window = %d, want the burst deferred", got) + } + + deadline := time.Now().Add(5 * time.Second) + for store.getAll.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := store.getAll.Load(); got != 2 { + t.Fatalf("refreshes after the window = %d, want one trailing refresh", got) + } + if got := publishedAgentCPU(t, store); got != 60 { + t.Fatalf("trailing refresh published CPU %v, want the latest report's 60", got) + } + + time.Sleep(window + 100*time.Millisecond) + reportAgentCPU(m, 70) + if got := store.getAll.Load(); got != 3 { + t.Fatalf("refreshes after a quiet window = %d, want a synchronous leading refresh", got) + } + if got := publishedAgentCPU(t, store); got != 70 { + t.Fatalf("leading report after a quiet window published CPU %v, want 70", got) + } +} + +func TestStopCancelsPendingAgentReportRefresh(t *testing.T) { + const window = 100 * time.Millisecond + m, store := newAgentReportThrottleMonitor(t, window) + + reportAgentCPU(m, 10) + reportAgentCPU(m, 20) + m.stopAgentReportRefresh() + time.Sleep(3 * window) + if got := store.getAll.Load(); got != 1 { + t.Fatalf("refreshes after stop = %d, want the pending trailing refresh cancelled", got) } }