From b815ea141e09df44c7d05ed821988bb12f6c48ca Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:22:17 +0100 Subject: [PATCH 01/14] Compare the renewed licence expiry in UTC in its test TestProviderMSPLicenseRefreshAppliesShorterPaidPeriod compared the refresher's UTC expiry ("...Z") with the expected time formatted in the machine's local zone ("...+01:00"), so it failed on any machine not set to UTC, including a maintainer Mac on British Summer Time, while CI in UTC passed. Format the expectation in UTC. It now passes under Europe/London, America/Los_Angeles and UTC. --- internal/cloudcp/provider_msp_license_refresh_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/cloudcp/provider_msp_license_refresh_test.go b/internal/cloudcp/provider_msp_license_refresh_test.go index 7106a46a92..41c3ee7db5 100644 --- a/internal/cloudcp/provider_msp_license_refresh_test.go +++ b/internal/cloudcp/provider_msp_license_refresh_test.go @@ -230,7 +230,7 @@ func TestProviderMSPLicenseRefreshAppliesShorterPaidPeriod(t *testing.T) { refresher.SetRestart(func() { restarted <- struct{}{} }) result, err := refresher.Refresh(context.Background()) - if err != nil || !result.Changed || !result.RestartScheduled || result.ExpiresAt != shorterExpiry.Format(time.RFC3339) { + if err != nil || !result.Changed || !result.RestartScheduled || result.ExpiresAt != shorterExpiry.UTC().Format(time.RFC3339) { t.Fatalf("shorter paid-period refresh = %+v, %v", result, err) } installed, err := os.ReadFile(ProviderMSPRenewedLicensePath(cfg.DataDir)) From 7f6ac2f27e9a6d53ed1d261408b70b412ba9ed7d Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:23:55 +0100 Subject: [PATCH 02/14] Remove ChangeDetector's unjoined legacy history writer DetectChanges launched an untracked goroutine to rewrite ai_changes.json. Tests backed by a temp directory slept 20ms or 100ms hoping it had finished; when the goroutine started late it wrote into a directory that cleanup was removing, which is the latent TempDir race already fixed for IncidentStore and KnowledgeStore. Nothing in the runtime has called DetectChanges since patrol moved to agentic execution (1b1c9bb2a3). Recent-change context comes from the unified-resource timeline, and the detector is only its read fallback, so the save goroutine was reachable only from tests. Joining it would harden a dead legacy writer, so remove it along with saveToDisk and the unused save-scheduling fields. The detector still loads legacy history at construction; detected changes stay in memory. Replace the sleeping tests with TestChangeDetector_LegacyHistoryIsReadOnly, which fails 210/300 runs against the old async writer. With the save delayed 0-150ms, 349 of 600 old-code saves ran after their test's directory was removed; none remain. Also drop two sleeps after RemediationLog.Log, which already persists before returning. --- .../v6/internal/subsystems/ai-runtime.md | 9 +++ internal/ai/memory/changes.go | 55 +++---------- internal/ai/memory/memory_coverage_test.go | 77 +------------------ internal/ai/memory/memory_extended_test.go | 61 ++++++++------- 4 files changed, 53 insertions(+), 149 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 005efb5512..31a32302e4 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -6633,6 +6633,15 @@ The patrol-local `memory.ChangeDetector.GetChangesSummary` path now also delegates to the shared memory recent-change presentation helper, so any future fallback summary entry point inherits the same heading, resource prefixing, and change-type labels without re-implementing the markdown shape. +That detector's on-disk history (`ai_changes.json`) is legacy and read-only. +Nothing in the runtime has called `DetectChanges` since patrol moved to +agentic execution, so the detector only loads that file at construction and +never writes it. Detected changes stay in memory, and there is no background +save to join before a data directory is removed. The unjoined save goroutine +that `DetectChanges` used to launch was reachable only from tests, and could +outlive a `t.TempDir` cleanup; it was removed rather than hardened. +`TestChangeDetector_LegacyHistoryIsReadOnly` proves detection leaves the file +and directory untouched. Those unified-resource action and export audit records are now also exposed through the enterprise audit read surface so operators can inspect the execution trail without reaching into storage internals. diff --git a/internal/ai/memory/changes.go b/internal/ai/memory/changes.go index 4896eaae79..17627d3d22 100644 --- a/internal/ai/memory/changes.go +++ b/internal/ai/memory/changes.go @@ -4,8 +4,6 @@ package memory import ( - "encoding/json" - "os" "sync" "sync/atomic" "time" @@ -53,20 +51,20 @@ type ResourceSnapshot struct { SnapshotTime time.Time `json:"snapshot_time"` } -// ChangeDetector tracks infrastructure changes over time +// ChangeDetector tracks infrastructure changes over time. +// +// Recent-change context comes from the unified-resource timeline; the detector +// is only its fallback. Its on-disk history (ai_changes.json) is legacy data +// from before patrol stopped calling DetectChanges: it is loaded once at +// construction and never written, so changes detected now stay in memory. type ChangeDetector struct { mu sync.RWMutex previousState map[string]ResourceSnapshot // resourceID -> snapshot changes []Change maxChanges int - // Persistence + // dataDir holds the legacy history read by loadFromDisk. dataDir string - - // saveStateMu guards asynchronous save scheduling state. - saveStateMu sync.Mutex - saveRunning bool - saveRequested bool } // ChangeDetectorConfig configures the change detector @@ -156,17 +154,10 @@ func (d *ChangeDetector) DetectChanges(currentSnapshots []ResourceSnapshot) []Ch } } - // Store new changes + // Store new changes in memory only; the legacy history is read-only. if len(newChanges) > 0 { d.changes = append(d.changes, newChanges...) d.trimChanges() - - // Persist asynchronously - go func() { - if err := d.saveToDisk(); err != nil { - log.Warn().Err(err).Msg("failed to save change history") - } - }() } return newChanges @@ -306,35 +297,7 @@ func (d *ChangeDetector) trimChanges() { } } -// saveToDisk persists changes to JSON file -func (d *ChangeDetector) saveToDisk() error { - if d.dataDir == "" { - return nil - } - - d.mu.RLock() - changes := make([]Change, len(d.changes)) - copy(changes, d.changes) - d.mu.RUnlock() - - path, err := memoryPersistencePath(d.dataDir, changeHistoryFileName) - if err != nil { - return err - } - data, err := json.MarshalIndent(changes, "", " ") - if err != nil { - return err - } - - tmpPath := path + ".tmp" - if err := os.WriteFile(tmpPath, data, 0600); err != nil { - return err - } - - return os.Rename(tmpPath, path) -} - -// loadFromDisk loads changes from JSON file +// loadFromDisk loads the legacy change history from its JSON file func (d *ChangeDetector) loadFromDisk() error { changes, ok, err := loadMemoryHistory(d.dataDir, changeHistoryFileName, "change history", func(a, b Change) bool { return a.DetectedAt.Before(b.DetectedAt) diff --git a/internal/ai/memory/memory_coverage_test.go b/internal/ai/memory/memory_coverage_test.go index e96eb817af..50b9cba439 100644 --- a/internal/ai/memory/memory_coverage_test.go +++ b/internal/ai/memory/memory_coverage_test.go @@ -79,67 +79,6 @@ func TestNewChangeDetector_LoadError(t *testing.T) { } } -func TestChangeDetector_SaveToDisk_Scenarios(t *testing.T) { - t.Run("NoDataDir", func(t *testing.T) { - d := &ChangeDetector{} - if err := d.saveToDisk(); err != nil { - t.Fatalf("expected nil error, got %v", err) - } - }) - - t.Run("MissingDir", func(t *testing.T) { - tmpDir := t.TempDir() - missing := filepath.Join(tmpDir, "missing") - d := &ChangeDetector{ - dataDir: missing, - changes: []Change{{ID: "c1"}}, - } - if err := d.saveToDisk(); err == nil { - t.Fatal("expected error for missing directory") - } - }) - - t.Run("MarshalError", func(t *testing.T) { - tmpDir := t.TempDir() - d := &ChangeDetector{ - dataDir: tmpDir, - changes: []Change{{ID: "c1", Before: func() {}}}, - } - if err := d.saveToDisk(); err == nil { - t.Fatal("expected marshal error") - } - }) - - t.Run("RenameError", func(t *testing.T) { - tmpDir := t.TempDir() - destDir := filepath.Join(tmpDir, "ai_changes.json") - if err := os.MkdirAll(destDir, 0755); err != nil { - t.Fatalf("mkdir: %v", err) - } - d := &ChangeDetector{ - dataDir: tmpDir, - changes: []Change{{ID: "c1"}}, - } - if err := d.saveToDisk(); err == nil { - t.Fatal("expected rename error") - } - }) - - t.Run("Success", func(t *testing.T) { - tmpDir := t.TempDir() - d := &ChangeDetector{ - dataDir: tmpDir, - changes: []Change{{ID: "c1"}}, - } - if err := d.saveToDisk(); err != nil { - t.Fatalf("saveToDisk error: %v", err) - } - if _, err := os.Stat(filepath.Join(tmpDir, "ai_changes.json")); err != nil { - t.Fatalf("expected file to exist: %v", err) - } - }) -} - func TestChangeDetector_LoadFromDisk_Scenarios(t *testing.T) { t.Run("NoFile", func(t *testing.T) { d := &ChangeDetector{dataDir: t.TempDir()} @@ -232,20 +171,6 @@ func TestChangeDetector_LoadFromDisk_Scenarios(t *testing.T) { }) } -func TestChangeDetector_DetectChanges_SaveError(t *testing.T) { - tmpDir := t.TempDir() - badDir := filepath.Join(tmpDir, "not-dir") - if err := os.WriteFile(badDir, []byte("x"), 0600); err != nil { - t.Fatalf("write file: %v", err) - } - - detector := NewChangeDetector(ChangeDetectorConfig{DataDir: badDir}) - detector.DetectChanges([]ResourceSnapshot{ - {ID: "vm-1", Name: "vm-1", Type: "vm", Status: "running"}, - }) - time.Sleep(20 * time.Millisecond) -} - func TestIncidentStore_DefaultsAndSummary(t *testing.T) { store := NewIncidentStore(IncidentStoreConfig{}) if store.maxIncidents != defaultIncidentMaxIncidents { @@ -913,10 +838,10 @@ func TestRemediationLog_Log_SaveError(t *testing.T) { dataDir: badDir, maxRecords: 1, } + // Log persists before returning, so the failed save has already run here. if err := log.Log(RemediationRecord{Problem: "p", Action: "a"}); err != nil { t.Fatalf("log error: %v", err) } - time.Sleep(20 * time.Millisecond) } func TestNewRemediationLog_LoadsFromDisk(t *testing.T) { diff --git a/internal/ai/memory/memory_extended_test.go b/internal/ai/memory/memory_extended_test.go index 4e16ad8bfc..fcaafede8e 100644 --- a/internal/ai/memory/memory_extended_test.go +++ b/internal/ai/memory/memory_extended_test.go @@ -1,6 +1,7 @@ package memory import ( + "bytes" "os" "path/filepath" "testing" @@ -143,31 +144,42 @@ func TestChangeDetector_MultipleChangesAtOnce(t *testing.T) { } } -func TestChangeDetector_Persistence(t *testing.T) { - tmpDir, err := os.MkdirTemp("", "changes-test") - if err != nil { - t.Fatalf("Failed to create temp dir: %v", err) +func TestChangeDetector_LegacyHistoryIsReadOnly(t *testing.T) { + tmpDir := t.TempDir() + path := filepath.Join(tmpDir, "ai_changes.json") + legacy := []byte(`[{"id":"legacy-1","resource_id":"vm-100","resource_type":"vm","change_type":"created","detected_at":"2026-01-01T00:00:00Z"}]`) + if err := os.WriteFile(path, legacy, 0600); err != nil { + t.Fatalf("write legacy history: %v", err) } - defer os.RemoveAll(tmpDir) - // Create detector with persistence d := NewChangeDetector(ChangeDetectorConfig{ MaxChanges: 100, DataDir: tmpDir, }) - - // Create some changes d.DetectChanges([]ResourceSnapshot{ {ID: "vm-100", Name: "web-server", Type: "vm", Status: "running", Node: "node1"}, }) - // Wait a bit for async save - time.Sleep(100 * time.Millisecond) + // Detected changes join the loaded legacy history in memory. + if got := d.GetChangesForResource("vm-100", 10); len(got) != 2 { + t.Fatalf("expected legacy and detected change, got %d", len(got)) + } - // Check if file was created (persistence is async, might not exist) - filePath := filepath.Join(tmpDir, "ai_changes.json") - if _, err := os.Stat(filePath); os.IsNotExist(err) { - t.Log("Changes file not created - persistence may be async") + // Nothing writes the data directory, so no background save can outlive + // the test and race TempDir cleanup. + entries, err := os.ReadDir(tmpDir) + if err != nil { + t.Fatalf("read data dir: %v", err) + } + if len(entries) != 1 || entries[0].Name() != "ai_changes.json" { + t.Fatalf("data dir must hold only the legacy history, got %v", entries) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read legacy history: %v", err) + } + if !bytes.Equal(data, legacy) { + t.Fatalf("legacy history was rewritten: %s", data) } } @@ -304,11 +316,7 @@ func TestRemediationLog_FormatForContext_NoRecords(t *testing.T) { } func TestRemediationLog_Persistence(t *testing.T) { - tmpDir, err := os.MkdirTemp("", "remediation-test") - if err != nil { - t.Fatalf("Failed to create temp dir: %v", err) - } - defer os.RemoveAll(tmpDir) + tmpDir := t.TempDir() // Create log with persistence r := NewRemediationLog(RemediationLogConfig{ @@ -316,20 +324,19 @@ func TestRemediationLog_Persistence(t *testing.T) { DataDir: tmpDir, }) - _ = r.Log(RemediationRecord{ + if err := r.Log(RemediationRecord{ ResourceID: "vm-100", Problem: "Test problem", Action: "Test action", Outcome: OutcomeResolved, - }) + }); err != nil { + t.Fatalf("Log: %v", err) + } - // Wait a bit for async save - time.Sleep(100 * time.Millisecond) - - // Check if file was created (persistence may be async) + // Log persists before returning. filePath := filepath.Join(tmpDir, "ai_remediations.json") - if _, err := os.Stat(filePath); os.IsNotExist(err) { - t.Log("Remediation file not created - persistence may be async") + if _, err := os.Stat(filePath); err != nil { + t.Fatalf("expected remediation file after Log: %v", err) } } From 522926b0f81752633604b9f41150e926d1d0afe7 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:24:58 +0100 Subject: [PATCH 03/14] Make agent report refreshes linear and clone the registry once Every accepted agent report refreshes the canonical resource store, and two parts of that pass did more work than the estate warranted (#2199). Guest ingest resolved each guest's parent node through a fallback that walked every resource, which made each rebuild quadratic. Every agent-producing source is ingested before the VM and container loops, so the registry now buckets agents by node name for those loops, as buildChildCounts already did. It rebuilds the buckets whenever an ingest yields an agent and skips entries a merge replaced, so parent resolution matches the full scan exactly. On 8-node clusters one rebuild went from 74 to 42 ms at 2,080 resources and now grows linearly. The refresh pass then cloned the whole registry up to six times, once in each metric sync and again for the alert sync. A pass now wraps the store in resourceSnapshotStore, which clones once and forwards metrics-target resolution, so every consumer reads one generation. Together, one synthetic agent-report refresh drops from 188 to 82 ms at 2,080 resources and from 90 to 41 ms at 1,040, with allocations per report falling from 93 MB to 50 MB. Refreshes stay synchronous, as the agent-lifecycle contract requires; each pass still covers the whole estate, and coalescing reports would need that contract changed first. Refs #2199 --- .../v6/internal/subsystems/agent-lifecycle.md | 14 ++ .../v6/internal/subsystems/monitoring.md | 16 +++ .../internal/subsystems/unified-resources.md | 20 +++ internal/monitoring/monitor.go | 44 ++++++ .../monitoring/monitor_host_agents_test.go | 76 +++++++++++ internal/unifiedresources/registry.go | 67 +++++++-- internal/unifiedresources/registry_test.go | 129 ++++++++++++++++++ 7 files changed, 353 insertions(+), 13 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 776c588425..2dc2cee1e4 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -53,6 +53,20 @@ 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 + +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. + +### Canonical Patrol and Assistant continuation, 2026-09-07 + ### Canonical Patrol and Assistant continuation, 2026-09-07 Patrol planning uses the canonical action broker's plan-only boundary. Policy diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 8b37947170..314c5786fe 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -30,6 +30,22 @@ of an unchanged snapshot must not append duplicate change records. cover the publish and blocked-persistence boundaries. Synthetic performance improvement alone does not establish field CPU relief. +### One registry snapshot per store refresh pass — issue #2199 + +Each store refresh pass (`updateResourceStore`, its read-path twin, and so +every accepted agent report) ran five metric syncs and the alert sync, and each +cloned the whole registry through `GetAll`. A pass now wraps the store in +`resourceSnapshotStore`, which clones once and forwards metrics-target +resolution, so every consumer of the pass reads one generation. Stores without +metrics-target resolution pass through unwrapped, keeping the syncs' capability +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). + +### Linked Pulse agent memory for Proxmox LXC — issue #2148 (22 September 2026) + ### Linked Pulse agent memory for Proxmox LXC — issue #2148 (22 September 2026) Both LXC memory paths (the efficient `cluster/resources` builder and the diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 0717837694..71e1c5d1c7 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -38,6 +38,26 @@ compares the old and new paths, while boundary. The 1,000-resource benchmark is component evidence, not installed fleet CPU attribution or a release acceptance claim. +### Node-indexed guest parent resolution during ingest — issue #2199 + +Guest ingest resolves each Proxmox guest's parent node, and that lookup can +fall back to scanning for agent candidates by node name. Walking every resource +for every guest made each snapshot rebuild quadratic in the estate. Every +agent-producing source is ingested before the VM and container loops, so the +registry buckets agents by lowercased node name for those loops, reusing +`agentNodeScanIndex` from `buildChildCounts`. `ingestRecord` rebuilds a live +index whenever an ingest yields an agent, and the indexed walk skips entries +that are no longer the registry's current object for their ID, so resolution +matches the full scan exactly. `TestGuestIngestParentsMatchFullAgentScan` +compares every guest's ingest-time parent with the full scan, including a +cluster whose guests are reachable only through the scan, and +`TestAgentNodeScanIndexTracksAgentsIngestedWhileLive` pins the rebuild. On a +synthetic estate of 8-node clusters the rebuild moved from 74 to 42 ms at 2,080 +resources and now grows linearly; that is component evidence, not installed +fleet CPU attribution. + +### Bounded incident-history selection + Canonical frontend memory withdrawal is explicit: when a resource snapshot omits the canonical memory metric and its Proxmox memory facet marks usageUnavailable, the display merge must clear any previous metric. Plain partial omission remains compatible with richer REST state, and an incoming canonical metric (including measured zero) takes precedence over unavailable raw evidence. The adapter transition tests pin withdrawal and recovery; the hybrid-memory Chromium fixture exercises the rendered table and drawer at 1280px and 390px. Workload details remain canonical-only: withdrawal shows N/A in the table and removes the Memory section rather than manufacturing a raw-facet total; measured-zero recovery restores Total and Free, with screenshot positioning above fixed navigation. This does not change agent-only or arbitrary field-deletion semantics. **VM-linked agent memory read — issue #1962 (7 September 2026)** diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 1f33ab3d3c..76c58be0f9 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -5110,6 +5110,7 @@ func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot) { return } recordSupplementalResourceChanges(store, m.collectSupplementalChanges()) + store = newResourceSnapshotStore(store) m.syncAllUnifiedMetrics(store) m.syncUnifiedResourceAlertsToState(store.GetAll()) } @@ -5157,6 +5158,7 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) { if atomicStore, ok := store.(AtomicSnapshotResourceStore); ok { atomicStore.PopulateSnapshotAndSupplemental(snapshotForStore, recordsBySource) recordSupplementalResourceChanges(store, supplementalChanges) + store = newResourceSnapshotStore(store) m.syncAllUnifiedMetrics(store) for source, records := range recordsBySource { if len(records) == 0 { @@ -5188,10 +5190,52 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) { } recordSupplementalResourceChanges(store, supplementalChanges) + store = newResourceSnapshotStore(store) m.syncAllUnifiedMetrics(store) m.syncUnifiedResourceAlertsToState(store.GetAll()) } +// resourceSnapshotStore serves one GetAll clone to every consumer of a single +// store-refresh pass. The metric syncs and the alert sync each cloned the +// whole registry for themselves, up to six clones for every accepted agent +// report (#2199). Consumers of a pass only read, so they can share one +// generation. +type resourceSnapshotStore struct { + ResourceStoreInterface + targets MetricsTargetResourceStore + resources []unifiedresources.Resource + listed bool +} + +// newResourceSnapshotStore wraps store for one refresh pass. A store that +// cannot resolve metrics targets is returned as is, so the metric syncs' +// capability checks see exactly what they would without the wrapper. +func newResourceSnapshotStore(store ResourceStoreInterface) ResourceStoreInterface { + if store == nil { + return nil + } + if _, ok := store.(*resourceSnapshotStore); ok { + return store + } + targets, ok := store.(MetricsTargetResourceStore) + if !ok { + return store + } + return &resourceSnapshotStore{ResourceStoreInterface: store, targets: targets} +} + +func (s *resourceSnapshotStore) GetAll() []unifiedresources.Resource { + if !s.listed { + s.resources = s.ResourceStoreInterface.GetAll() + s.listed = true + } + return s.resources +} + +func (s *resourceSnapshotStore) MetricsTargetForResource(resourceID string) *unifiedresources.MetricsTarget { + return s.targets.MetricsTargetForResource(resourceID) +} + // refreshUnifiedResourceStoreAfterAgentStateChange makes accepted agent // ingest and removal immediately visible to canonical ReadState consumers. // WebSocket broadcasts may also rebuild the store for their own hydrate path, diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 909933c9be..db06f95852 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -6408,3 +6408,79 @@ func TestApplyHostReportHonoursRemovalBlockWhenIdentityWouldFork(t *testing.T) { t.Fatalf("expected removal block to be cleared, still have %+v", monitor.state.GetRemovedHostAgents()) } } + +// countingResourceStore counts registry clones handed out by the live store. +type countingResourceStore struct { + *unifiedresources.MonitorAdapter + getAll int +} + +func (c *countingResourceStore) GetAll() []unifiedresources.Resource { + c.getAll++ + return c.MonitorAdapter.GetAll() +} + +// plainResourceStore lacks metrics-target resolution. +type plainResourceStore struct { + ResourceStoreInterface +} + +func TestResourceSnapshotStoreClonesOncePerPass(t *testing.T) { + now := time.Now().UTC() + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + adapter.PopulateFromSnapshot(models.StateSnapshot{ + LastUpdate: now, + VMs: []models.VM{{ + ID: "lab:pve1:101", VMID: 101, Name: "db", Node: "pve1", Instance: "lab", + Status: "running", Type: "qemu", LastSeen: now, + }}, + }) + counting := &countingResourceStore{MonitorAdapter: adapter} + + store := newResourceSnapshotStore(counting) + first := store.GetAll() + second := store.GetAll() + if counting.getAll != 1 { + t.Fatalf("underlying GetAll calls = %d, want 1", counting.getAll) + } + 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)) + } + resolver, ok := store.(MetricsTargetResourceStore) + if !ok || resolver.MetricsTargetForResource(first[0].ID) == nil { + t.Fatal("snapshot store does not forward metrics-target resolution") + } + if again := newResourceSnapshotStore(store); again != store { + t.Fatal("wrapping a pass snapshot again must reuse it") + } + + plain := plainResourceStore{ResourceStoreInterface: adapter} + if wrapped := newResourceSnapshotStore(plain); wrapped != ResourceStoreInterface(plain) { + t.Fatal("a store without metrics-target resolution must pass through unwrapped") + } +} + +// An accepted agent report refreshes the store once. Every metric sync and the +// alert sync in that pass must share a single registry clone (#2199). +func TestAgentReportRefreshClonesRegistryOnce(t *testing.T) { + now := time.Now().UTC() + state := models.NewState() + state.UpdateNodes([]models.Node{{ID: "lab-pve1", Name: "pve1", Instance: "lab", Status: "online", Type: "node", LastSeen: now}}) + state.UpdateVMs([]models.VM{{ + ID: "lab:pve1:101", VMID: 101, Name: "db", Node: "pve1", Instance: "lab", + Status: "running", Type: "qemu", CPU: 0.2, LastSeen: now, + }}) + state.UpsertHost(models.Host{ID: "agent-1", Hostname: "pve1.example", Status: "online", CPUUsage: 3, LastSeen: now}) + + counting := &countingResourceStore{MonitorAdapter: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil))} + m := &Monitor{ + state: state, + resourceStore: counting, + metricsHistory: NewMetricsHistory(32, time.Hour), + } + + m.refreshUnifiedResourceStoreAfterAgentStateChange() + if counting.getAll != 1 { + t.Fatalf("registry clones for one agent report refresh = %d, want 1", counting.getAll) + } +} diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 69003927fd..80146cd2f7 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -126,12 +126,13 @@ type ResourceRegistry struct { // viewsDirty is false, like every other cached field above. cachedSourceTargets map[string][]SourceTarget - // agentNodeScanIndex buckets agent resources by lowercased node name for - // the duration of one buildChildCounts pass, where the agent-parent - // fallback would otherwise scan every resource once per guest. Only that - // pass sets it; nil means callers take the full scan. Bucket membership - // is fixed for the pass (Type and NodeName are never mutated mid-pass) - // while scores stay live through the shared pointers. + // agentNodeScanIndex buckets agent resources by lowercased node name + // while guests are ingested and during one buildChildCounts pass, where + // the agent-parent fallback would otherwise scan every resource once per + // guest. nil means callers take the full scan. ingestRecord rebuilds it + // whenever an ingest yields an agent, so bucket membership always matches + // what the full scan would see; scores stay live through the shared + // pointers. agentNodeScanIndex map[string][]agentNodeCandidate } @@ -268,12 +269,23 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh rr.ingestPMGInstance(instance) } var guestSuccessions []CanonicalIDSuccession + // Every guest resolves its parent node, which can fall back to a scan of + // every resource. Each agent-producing source is ingested above, so bucket + // agents by node once for the guest loops instead of walking the registry + // per guest: that walk made every agent report's rebuild quadratic in the + // estate (#2199). + rr.mu.Lock() + rr.agentNodeScanIndex = rr.buildAgentNodeScanIndexLocked() + rr.mu.Unlock() for _, vm := range snapshot.VMs { guestSuccessions = append(guestSuccessions, rr.ingestVM(vm, clusterByInstance, nodeNamesByInstance)...) } for _, ct := range snapshot.Containers { guestSuccessions = append(guestSuccessions, rr.ingestContainer(ct, clusterByInstance, nodeNamesByInstance)...) } + rr.mu.Lock() + rr.agentNodeScanIndex = nil + rr.mu.Unlock() rr.applyRecordSuccessions(guestSuccessions) for _, storage := range snapshot.Storage { rr.ingestStorage(storage) @@ -2658,9 +2670,12 @@ func (rr *ResourceRegistry) ingest(source DataSource, sourceID string, resource // normal canonical matching rules, but may only introduce absent resources. // Returning an empty ID also prevents a skipped record from attaching retired // identities or migrating operator-owned state onto the current resource. -func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, resource Resource, identity ResourceIdentity, onlyMissing bool) string { +func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, resource Resource, identity ResourceIdentity, onlyMissing bool) (ingestedID string) { rr.mu.Lock() defer rr.mu.Unlock() + if rr.agentNodeScanIndex != nil { + defer func() { rr.refreshAgentNodeScanIndexLocked(ingestedID) }() + } rr.invalidateSourceTargetsLocked() sourceID = normalizeSourceID(sourceID) if sourceID == "" { @@ -4549,6 +4564,11 @@ func (rr *ResourceRegistry) proxmoxNodeParentIDFromResourcesLocked(instance, clu // buildChildCounts is mid-pass: only this node's agent candidates // can match, so skip the full walk. for _, candidate := range rr.agentNodeScanIndex[strings.ToLower(nodeName)] { + // A merge since indexing may have replaced or removed the entry; + // the full scan would only see the registry's current object. + if rr.resources[candidate.id] != candidate.resource { + continue + } consider(candidate.id, candidate.resource) } return bestID @@ -4601,11 +4621,10 @@ func proxmoxNodeParentScopeScore(instance, clusterName string, parent *ProxmoxDa } } -func (rr *ResourceRegistry) buildChildCounts() { - // The parent-resolution loop below may hit the agent fallback scan once - // per guest. Bucketing agents by node name up front turns each of those - // scans into a lookup over that node's few candidates. - rr.agentNodeScanIndex = make(map[string][]agentNodeCandidate) +// buildAgentNodeScanIndexLocked buckets agent resources by lowercased node +// name. The caller holds rr.mu for writing. +func (rr *ResourceRegistry) buildAgentNodeScanIndexLocked() map[string][]agentNodeCandidate { + index := make(map[string][]agentNodeCandidate) for id, r := range rr.resources { if r == nil || r.Proxmox == nil || CanonicalResourceType(r.Type) != ResourceTypeAgent { continue @@ -4614,8 +4633,30 @@ func (rr *ResourceRegistry) buildChildCounts() { if nodeName == "" { continue } - rr.agentNodeScanIndex[nodeName] = append(rr.agentNodeScanIndex[nodeName], agentNodeCandidate{id: id, resource: r}) + index[nodeName] = append(index[nodeName], agentNodeCandidate{id: id, resource: r}) } + return index +} + +// refreshAgentNodeScanIndexLocked rebuilds a live index after an ingest +// yields an agent, whose node name may be new to the buckets. Guest ingests, +// the common case while the index is live, leave it untouched. +func (rr *ResourceRegistry) refreshAgentNodeScanIndexLocked(resourceID string) { + if rr.agentNodeScanIndex == nil { + return + } + resource := rr.resources[CanonicalResourceID(resourceID)] + if resource == nil || resource.Proxmox == nil || CanonicalResourceType(resource.Type) != ResourceTypeAgent { + return + } + rr.agentNodeScanIndex = rr.buildAgentNodeScanIndexLocked() +} + +func (rr *ResourceRegistry) buildChildCounts() { + // The parent-resolution loop below may hit the agent fallback scan once + // per guest. Bucketing agents by node name up front turns each of those + // scans into a lookup over that node's few candidates. + rr.agentNodeScanIndex = rr.buildAgentNodeScanIndexLocked() defer func() { rr.agentNodeScanIndex = nil }() // ChildCount and ParentName are derived fields. Clear prior values before diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index 9d05b7d728..ea5104982a 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -6650,3 +6650,132 @@ func TestIssue2076USBLinkedDiskAmbiguity(t *testing.T) { }) } } + +// Guest ingest resolves each guest's parent node, which can fall back to a +// scan of every agent. The guest pass answers that fallback from +// agentNodeScanIndex, so the index must resolve exactly what the scan would. + +// agentNodeScanFixture spans three clusters that reuse node names, so a +// by-name bucket holds candidates from every cluster and scope scoring has to +// pick the right one. Every other node runs a linked host agent. A fourth +// cluster's nodes are keyed by their endpoint instance while its guests report +// the cluster name, so no direct source mapping reaches their parent and only +// the agent scan resolves it. +func agentNodeScanFixture(now time.Time) models.StateSnapshot { + snapshot := models.StateSnapshot{LastUpdate: now} + for n := 0; n < 2; n++ { + nodeName := fmt.Sprintf("pve%d", n) + snapshot.Nodes = append(snapshot.Nodes, models.Node{ + ID: "ep-" + nodeName, Name: nodeName, Instance: "ep", ClusterName: "prod", + Status: "online", Type: "node", LastSeen: now, + }) + vmid := 900 + n + snapshot.VMs = append(snapshot.VMs, models.VM{ + ID: fmt.Sprintf("prod:%s:%d", nodeName, vmid), VMID: vmid, + Name: fmt.Sprintf("vm-%d", vmid), Node: nodeName, Instance: "prod", + Status: "running", Type: "qemu", LastSeen: now, + }) + } + for cluster := 0; cluster < 3; cluster++ { + instance := fmt.Sprintf("lab%d", cluster) + for n := 0; n < 4; n++ { + nodeName := fmt.Sprintf("pve%d", n) + nodeID := instance + "-" + nodeName + node := models.Node{ + ID: nodeID, + Name: nodeName, + Instance: instance, + Status: "online", + Type: "node", + LastSeen: now, + } + if n%2 == 0 { + hostID := fmt.Sprintf("agent-%s-%s", instance, nodeName) + node.LinkedAgentID = hostID + snapshot.Hosts = append(snapshot.Hosts, models.Host{ + ID: hostID, + Hostname: fmt.Sprintf("%s.%s.example", nodeName, instance), + Status: "online", + LastSeen: now, + LinkedNodeID: nodeID, + }) + } + snapshot.Nodes = append(snapshot.Nodes, node) + for g := 0; g < 3; g++ { + vmid := 100 + cluster*100 + n*10 + g + snapshot.VMs = append(snapshot.VMs, models.VM{ + ID: fmt.Sprintf("%s:%s:%d", instance, nodeName, vmid), VMID: vmid, + Name: fmt.Sprintf("vm-%d", vmid), Node: nodeName, Instance: instance, + Status: "running", Type: "qemu", LastSeen: now, + }) + ctid := 5000 + vmid + snapshot.Containers = append(snapshot.Containers, models.Container{ + ID: fmt.Sprintf("%s:%s:%d", instance, nodeName, ctid), VMID: ctid, + Name: fmt.Sprintf("ct-%d", ctid), Node: nodeName, Instance: instance, + Status: "running", Type: "lxc", LastSeen: now, + }) + } + } + } + return snapshot +} + +func TestGuestIngestParentsMatchFullAgentScan(t *testing.T) { + snapshot := agentNodeScanFixture(time.Now().UTC()) + rr := NewRegistry(nil) + rr.IngestSnapshot(snapshot) + + rr.mu.RLock() + defer rr.mu.RUnlock() + if rr.agentNodeScanIndex != nil { + t.Fatal("agent node index outlived the ingest pass") + } + checked := 0 + for id, resource := range rr.resources { + if resource.Type != ResourceTypeVM && resource.Type != ResourceTypeSystemContainer { + continue + } + if resource.Proxmox == nil { + t.Fatalf("guest %s has no Proxmox data", id) + } + // With the index nil this resolves through the full scan. + want := rr.proxmoxNodeParentIDLocked(resource.Proxmox.Instance, resource.Proxmox.ClusterName, resource.Proxmox.NodeName, "") + if want == "" { + t.Fatalf("full scan found no parent for guest %s on %s/%s", id, resource.Proxmox.Instance, resource.Proxmox.NodeName) + } + if got := CanonicalResourceID(resource.parentBySource[SourceProxmox]); got != want { + t.Fatalf("guest %s on %s/%s: ingest parent %q, full scan %q", id, resource.Proxmox.Instance, resource.Proxmox.NodeName, got, want) + } + checked++ + } + if want := len(snapshot.VMs) + len(snapshot.Containers); checked != want { + t.Fatalf("checked %d guests, want %d", checked, want) + } +} + +func TestAgentNodeScanIndexTracksAgentsIngestedWhileLive(t *testing.T) { + now := time.Now().UTC() + rr := NewRegistry(nil) + rr.IngestSnapshot(models.StateSnapshot{ + LastUpdate: now, + Nodes: []models.Node{{ID: "lab-pve1", Name: "pve1", Instance: "lab", Status: "online", Type: "node", LastSeen: now}}, + }) + + rr.mu.Lock() + rr.agentNodeScanIndex = rr.buildAgentNodeScanIndexLocked() + rr.mu.Unlock() + defer func() { + rr.mu.Lock() + rr.agentNodeScanIndex = nil + rr.mu.Unlock() + }() + + rr.ingestProxmoxNode(models.Node{ID: "lab-pve2", Name: "pve2", Instance: "lab", Status: "online", Type: "node", LastSeen: now}, nil) + + rr.mu.RLock() + defer rr.mu.RUnlock() + candidates := rr.agentNodeScanIndex["pve2"] + if len(candidates) != 1 || rr.resources[candidates[0].id] != candidates[0].resource { + t.Fatalf("index bucket for an agent ingested while live = %+v, want its registry entry", candidates) + } +} From 515e88e4c2651924e567df1cf9d9b464a3584b94 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:45:55 +0100 Subject: [PATCH 04/14] Keep the provider MSP install proof out of client slots Setup's summary tells every new provider to run the install proof before their first real client. Its temporary workspaces are created on the provider's own account, so they count against the client limit. On a two-client evaluation that already had a client, the proof created its first workspace, was refused the second, and returned no report, so install-proof's cleanup had nothing to remove and "Provider MSP Proof 01" stayed behind in one of the evaluation's two slots. Found on the walkthrough lab. The proof now checks for room before creating anything and says why it cannot run, and a proof that fails part-way returns the IDs it created so install-proof removes them. On the lab, with 3 of 3 slots in use, install-proof now refuses up front and the client count is unchanged. --- .../provider_msp_install_proof.go | 7 +++ .../provider_msp_install_proof_test.go | 51 +++++++++++++++++++ cmd/pulse-control-plane/provider_msp_proof.go | 47 +++++++++++++++-- .../provider_msp_proof_test.go | 47 +++++++++++++++++ .../subsystems/deployment-installability.md | 19 +++++++ 5 files changed, 168 insertions(+), 3 deletions(-) diff --git a/cmd/pulse-control-plane/provider_msp_install_proof.go b/cmd/pulse-control-plane/provider_msp_install_proof.go index 69a3395540..046c472491 100644 --- a/cmd/pulse-control-plane/provider_msp_install_proof.go +++ b/cmd/pulse-control-plane/provider_msp_install_proof.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "path/filepath" "strings" @@ -265,6 +266,12 @@ func runProviderMSPInstallProofWithDependencies(ctx context.Context, cfg *cloudc proofTenantIDs = providerMSPInstallProofTenantIDs(proof.Workspaces) } if err != nil { + // A proof that fails part-way returns no report, but it may already + // have created workspaces; remove those too. + var partial *providerMSPProofPartialError + if errors.As(err, &partial) && len(proofTenantIDs) == 0 { + proofTenantIDs = partial.CreatedTenantIDs + } addFailure("workspace proof: %v", err) return failAfterProof(err) } diff --git a/cmd/pulse-control-plane/provider_msp_install_proof_test.go b/cmd/pulse-control-plane/provider_msp_install_proof_test.go index 9e1fc07b75..a5a9e95025 100644 --- a/cmd/pulse-control-plane/provider_msp_install_proof_test.go +++ b/cmd/pulse-control-plane/provider_msp_install_proof_test.go @@ -407,3 +407,54 @@ func (f *fakeProviderMSPInstallProofRuntime) CleanupProviderMSPProofTenants(_ co func (f *fakeProviderMSPInstallProofRuntime) Close() { f.closed = true } + +// A proof that fails part-way returns no report, so the workspace list the +// install proof used to clean up from was empty and the workspaces it had +// already created stayed behind in the provider's client slots. +func TestProviderMSPInstallProofCleansUpWorkspacesFromAPartialProof(t *testing.T) { + cfg := testProviderMSPProofConfig(t) + fakeRuntime := &fakeProviderMSPInstallProofRuntime{ + proofErr: &providerMSPProofPartialError{ + err: errors.New("create proof workspace 2: workspace limit reached"), + CreatedTenantIDs: []string{"ws-proof-01"}, + }, + } + + report, err := runProviderMSPInstallProofWithDependencies(context.Background(), cfg, providerMSPInstallProofOptions{ + AccountName: "Example MSP", + OwnerEmail: "owner@example.com", + WorkspacePrefix: "Provider MSP Proof", + WorkspaceCount: 2, + InstallType: "pve", + TargetPath: "/settings/infrastructure?add=linux-host", + BackupOutput: "/tmp/provider-msp-install-proof.tar.gz", + RestoreTargetDataDir: "/tmp/provider-msp-restore-drill", + Cleanup: true, + }, providerMSPInstallProofDependencies{ + Bootstrap: func(context.Context, *cloudcp.CPConfig, cloudcp.ProviderMSPBootstrapOptions) (*cloudcp.ProviderMSPBootstrapResult, error) { + return healthyProviderMSPInstallProofBootstrap(), nil + }, + RunPreflight: func(context.Context, *cloudcp.CPConfig, providerMSPPreflightOptions) (*providerMSPPreflightReport, error) { + return healthyProviderMSPStatusPreflightReport(), nil + }, + RunStatus: func(context.Context, *cloudcp.CPConfig, providerMSPStatusOptions) (*providerMSPStatusReport, error) { + return &providerMSPStatusReport{OK: true}, nil + }, + NewProofRuntime: func(*cloudcp.CPConfig) (providerMSPInstallProofRuntime, error) { + return fakeRuntime, nil + }, + CreateBackup: func(context.Context, *cloudcp.CPConfig, string) (*cloudcp.ProviderMSPBackupCreateResult, error) { + t.Fatal("backup should not run after a failed proof") + return nil, nil + }, + }) + if err == nil || !strings.Contains(err.Error(), "workspace limit reached") { + t.Fatalf("install proof error = %v, want the proof failure", err) + } + if report == nil || report.OK || !report.CleanupOK { + t.Fatalf("report = %#v, want failed with cleanup done", report) + } + if !reflect.DeepEqual(fakeRuntime.cleanupTenantIDs, []string{"ws-proof-01"}) { + t.Fatalf("cleanup tenant ids = %#v, want the workspace the partial proof created", fakeRuntime.cleanupTenantIDs) + } +} diff --git a/cmd/pulse-control-plane/provider_msp_proof.go b/cmd/pulse-control-plane/provider_msp_proof.go index 8c4982645d..3a02cc09c1 100644 --- a/cmd/pulse-control-plane/provider_msp_proof.go +++ b/cmd/pulse-control-plane/provider_msp_proof.go @@ -260,7 +260,26 @@ func (rt *providerMSPProofRuntime) runProviderMSPProof(ctx context.Context, opts return nil, fmt.Errorf("bootstrap provider MSP account: %w", err) } + // The proof creates its workspaces on the provider's own account, so they + // count against the same client limit real clients do. Check for room + // before creating anything: on a two-client evaluation that already has a + // client, the second proof workspace would otherwise be refused after the + // first was created, leaving it behind in one of the evaluation's slots. + if bootstrap.WorkspaceLimit > 0 { + inUse, err := rt.registry.CountActiveByAccountID(bootstrap.AccountID) + if err != nil { + return nil, fmt.Errorf("count client workspaces: %w", err) + } + if free := bootstrap.WorkspaceLimit - inUse; free < opts.WorkspaceCount { + return nil, fmt.Errorf("the proof creates %d temporary client workspaces but this account has %d free (%d of %d in use); run it on a fresh platform, before adding clients", + opts.WorkspaceCount, max(free, 0), inUse, bootstrap.WorkspaceLimit) + } + } + createdTenants := make([]*registry.Tenant, 0, opts.WorkspaceCount) + partial := func(err error) error { + return &providerMSPProofPartialError{err: err, CreatedTenantIDs: providerMSPProofTenantIDs(createdTenants)} + } if opts.Cleanup { defer func() { _ = rt.cleanupProviderMSPProofTenants(context.Background(), createdTenants) @@ -292,13 +311,13 @@ func (rt *providerMSPProofRuntime) runProviderMSPProof(ctx context.Context, opts displayName := fmt.Sprintf("%s %02d", opts.WorkspacePrefix, idx+1) tenant, err := rt.createProviderMSPProofWorkspace(ctx, bootstrap.AccountID, displayName, bootstrap.OwnerEmail) if err != nil { - return nil, fmt.Errorf("create proof workspace %d: %w", idx+1, err) + return nil, partial(fmt.Errorf("create proof workspace %d: %w", idx+1, err)) } createdTenants = append(createdTenants, tenant) workspace, err := rt.proveProviderMSPWorkspace(ctx, tenant, bootstrap.OwnerUserID, opts.InstallType, opts.TargetPath) if err != nil { - return nil, fmt.Errorf("prove workspace %s: %w", tenant.ID, err) + return nil, partial(fmt.Errorf("prove workspace %s: %w", tenant.ID, err)) } if workspace.ContainerID == "" { report.RuntimeContainerVerified = false @@ -327,13 +346,35 @@ func (rt *providerMSPProofRuntime) runProviderMSPProof(ctx context.Context, opts boundaryOK, err := rt.verifyProviderMSPInstallTokenIsolation(report.Workspaces) if err != nil { - return nil, err + return nil, partial(err) } report.InstallTokenBoundaryOK = boundaryOK report.WorkspaceCount = len(report.Workspaces) return report, nil } +// providerMSPProofPartialError is a proof that failed after creating some of +// its workspaces. It carries their IDs so the caller can remove them: without +// that, a failure part-way through left proof workspaces occupying client +// slots, because no report (and so no workspace list) comes back on error. +type providerMSPProofPartialError struct { + err error + CreatedTenantIDs []string +} + +func (e *providerMSPProofPartialError) Error() string { return e.err.Error() } +func (e *providerMSPProofPartialError) Unwrap() error { return e.err } + +func providerMSPProofTenantIDs(tenants []*registry.Tenant) []string { + ids := make([]string, 0, len(tenants)) + for _, tenant := range tenants { + if tenant != nil && strings.TrimSpace(tenant.ID) != "" { + ids = append(ids, tenant.ID) + } + } + return ids +} + func normalizeProviderMSPProofOptions(opts providerMSPProofOptions) (providerMSPProofOptions, error) { opts.AccountName = strings.TrimSpace(opts.AccountName) opts.OwnerEmail = strings.ToLower(strings.TrimSpace(opts.OwnerEmail)) diff --git a/cmd/pulse-control-plane/provider_msp_proof_test.go b/cmd/pulse-control-plane/provider_msp_proof_test.go index d81786bb45..e5192fd2e1 100644 --- a/cmd/pulse-control-plane/provider_msp_proof_test.go +++ b/cmd/pulse-control-plane/provider_msp_proof_test.go @@ -6,12 +6,14 @@ import ( "crypto/rand" "encoding/base64" "encoding/json" + "fmt" "os" "strings" "testing" "time" "github.com/rcourtman/pulse-go-rewrite/internal/cloudcp" + "github.com/rcourtman/pulse-go-rewrite/internal/cloudcp/registry" pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing" ) @@ -430,3 +432,48 @@ func writeProviderMSPProofLicenseForTest(t *testing.T, licenseID, email, planVer } return path } + +// The proof's workspaces count against the provider's own client limit. When +// there is no room for all of them it must refuse before creating any: on a +// two-client evaluation with a client already added, it used to create the +// first proof workspace, get refused on the second, and leave the first +// occupying a slot. +func TestProviderMSPProofRefusesWithoutFreeSlotsAndCreatesNothing(t *testing.T) { + t.Setenv("DOCKER_HOST", "unix:///tmp/pulse-provider-msp-proof-missing-docker.sock") + t.Setenv("DOCKER_TLS_VERIFY", "") + t.Setenv("DOCKER_CERT_PATH", "") + + cfg := testProviderMSPProofConfig(t) + rt, err := newProviderMSPProofRuntimeFromConfig(cfg) + if err != nil { + t.Fatalf("newProviderMSPProofRuntimeFromConfig: %v", err) + } + defer rt.close() + + bootstrap, err := cloudcp.BootstrapProviderMSP(context.Background(), cfg, cloudcp.ProviderMSPBootstrapOptions{ + AccountName: "Acme Provider", + OwnerEmail: "owner@example.com", + }) + if err != nil { + t.Fatalf("BootstrapProviderMSP: %v", err) + } + if err := rt.registry.Create(®istry.Tenant{ID: "t-realclient", AccountID: bootstrap.AccountID, DisplayName: "Real Client", State: registry.TenantStateActive}); err != nil { + t.Fatalf("create existing client: %v", err) + } + + _, err = rt.runProviderMSPProof(context.Background(), providerMSPProofOptions{ + AccountName: "Acme Provider", + OwnerEmail: "owner@example.com", + WorkspacePrefix: "Provider MSP Proof", + WorkspaceCount: bootstrap.WorkspaceLimit, + InstallType: "pve", + TargetPath: "/settings/infrastructure?add=linux-host", + }) + want := fmt.Sprintf("has %d free (1 of %d in use)", bootstrap.WorkspaceLimit-1, bootstrap.WorkspaceLimit) + if err == nil || !strings.Contains(err.Error(), want) { + t.Fatalf("runProviderMSPProof error = %v, want refusal naming %q", err, want) + } + if count, err := rt.registry.CountActiveByAccountID(bootstrap.AccountID); err != nil || count != 1 { + t.Fatalf("active workspaces after refusal = %d, %v; want only the existing client", count, err) + } +} diff --git a/docs/release-control/v6/internal/subsystems/deployment-installability.md b/docs/release-control/v6/internal/subsystems/deployment-installability.md index 6bc7bb2a9a..d4159a18ef 100644 --- a/docs/release-control/v6/internal/subsystems/deployment-installability.md +++ b/docs/release-control/v6/internal/subsystems/deployment-installability.md @@ -2423,6 +2423,25 @@ artifact-selection behaviour. ## Current State +### Provider MSP install proof never strands a client slot + +`run-install-proof.sh`, which setup's summary recommends before the first real +client, creates its temporary proof workspaces on the provider's own account, +so they count against the same client limit. It now checks for room before +creating any and refuses with the counts ("the proof creates 2 temporary client +workspaces but this account has 0 free (3 of 3 in use)"). Previously, on a +two-client evaluation that already had a client, it created the first proof +workspace, was refused the second, and returned no report, so its cleanup had +no workspace list and the first proof workspace stayed behind in one of the +evaluation's two slots. A proof that fails part-way now returns the IDs it +created, and `install-proof` removes them. Verified on 2026-09-24 against the +walkthrough lab: refusal before any workspace was created, client count +unchanged. Regression coverage: +`TestProviderMSPProofRefusesWithoutFreeSlotsAndCreatesNothing` in +`cmd/pulse-control-plane/provider_msp_proof_test.go` and +`TestProviderMSPInstallProofCleansUpWorkspacesFromAPartialProof` in +`cmd/pulse-control-plane/provider_msp_install_proof_test.go`. + ### Provider MSP setup points buyers at the portal `deploy/provider-msp/setup.sh` no longer tells an evaluating provider to From 298660157ef47b5c3010bb23d804573396b3903a Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 10:02:28 +0100 Subject: [PATCH 05/14] Show the real Pulse mark in the branding header preview MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With no custom logo, Settings > General > Application branding previewed the header with a hand-drawn stand-in: a blue circle holding a "●" text character, which does not look like the Pulse logo. The header itself falls back to PulseBrandMark (blue disc, white ring, white centre dot), so the preview misrepresented what the header shows. Render the same PulseBrandMark at the header's size; it follows the app theme in light and dark. --- frontend-modern/browser-verification.json | 36 +++++++------------ .../Settings/BrandingSettingsCard.tsx | 7 ++-- .../__tests__/BrandingSettingsCard.test.tsx | 21 +++++++++++ 3 files changed, 36 insertions(+), 28 deletions(-) diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index 20516b3924..88e568401c 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -1,23 +1,16 @@ { "version": 1, - "base_sha": "4f11deddc728a69f937dcb04777c9e44276c3f95", - "verified_at": "2026-09-23T11:26:00Z", + "base_sha": "3b1b43293e21d06d1286124f96f78b1247443dea", + "verified_at": "2026-09-24T09:02:03Z", "result": "passed", "changed_paths": [ - "frontend-modern/src/features/proxmox/ProxmoxBackupServersTable.tsx", - "frontend-modern/src/types/resource.ts" + "frontend-modern/src/components/Settings/BrandingSettingsCard.tsx" ], "content_sha256": { - "frontend-modern/src/features/proxmox/ProxmoxBackupServersTable.tsx": "f6b50e71870ca4a3e0dcfa3e47413542e8444296bc68481c60e15d1a5d921f9f", - "frontend-modern/src/types/resource.ts": "db2509b73b2ed550ea1890aaef34e2a3dfe7d500aa193586562f9e20b95283ab" + "frontend-modern/src/components/Settings/BrandingSettingsCard.tsx": "6516a05aff71b1d96660f81189b6df8bceccf7e94d15d6c3213556e794f2ae40" }, "routes": [ - "/browser-tests/pbs-host-history-correlation.html?topology=guest&order=stable (Proxmox Backups server drawer)", - "/browser-tests/pbs-host-history-correlation.html?topology=guest&order=reordered (Proxmox Backups server drawer)", - "/browser-tests/pbs-host-history-correlation.html?topology=pbs-only&order=stable (Proxmox Backups server drawer)", - "/browser-tests/pbs-host-history-correlation.html?topology=pbs-only&order=reordered (Proxmox Backups server drawer)", - "/browser-tests/pbs-host-history-correlation.html?topology=side-by-side&order=stable (Proxmox Backups server drawer)", - "/browser-tests/pbs-host-history-correlation.html?topology=side-by-side&order=reordered (Proxmox Backups server drawer)" + "/settings/system-general (Application branding card, Header preview)" ], "viewports": [ { @@ -30,18 +23,15 @@ } ], "states": [ - "PBS connection configured by 10.0.0.5 and linked to the host agent only by the reported pbs.nodeName, across pbs-only, side-by-side and PVE-guest topologies", - "Drawer History tab selected with a populated correlated host series", - "Timer-driven snapshot replacing the datastore inventory in stable and reversed order", - "Snapshot that briefly omits the correlated host row while the PBS server row remains", - "Snapshot that restores the correlated host row on the same selection" + "No custom logo and no custom name: the preview falls back to the default Pulse lockup", + "Light colour scheme at 1280x900 and 390x844", + "Dark colour scheme at 1280x900" ], "interactions": [ - "Expand the Proxmox Backups server row, switch between datastore rows and select History; assert the chart paths render against the host metrics target even though the configured endpoint is not a machine-identity token", - "Apply timer-driven stable and reordered snapshots; assert the drawer is not remounted, the History selection is retained, the chart series survive and the Discovery/Metrics Target rows stay byte-identical", - "Apply a snapshot that omits the correlated host row; assert the Identity rows and History chart stay on the retained host target instead of falling back to the PBS service key", - "Restore the correlated host row; assert the same target is confirmed, and repeat the complete matrix at 1280x900 and 390x844" + "Open Settings > General with the white_label capability present; assert the Header preview renders exactly one svg.pulse-brand-logo (blue disc, white ring, white centre dot) and no \"\u25cf\" text glyph", + "Compare the preview mark with the page header lockup: both render PulseBrandMark at 20px", + "Check the page does not scroll horizontally at 390px wide" ], - "command": "pulse-worker-browser scripts/check-pbs-host-history-correlation.cjs", - "notes": "Chromium 141.0.7390.37 / Playwright 1.56.1. Production ProxmoxBackupServersTable mounted through the real browser fixture with synthetic API responses and an intercepted metrics-history request. Proves the reported-node-name host correlation and its retention across a transient host-row omission, not backend persistence or installed release behaviour. The recorded history requests target agent/agent-proxback and vm/proxmox:100 and never the PBS service target agent/pbs-1. The same check fails on the pre-fix source: the drawer falls back to the PBS service key when the endpoint is not a machine-identity token, and omitting the host row flips the Discovery and Metrics Target rows to agent:pbs-1." + "command": "node verify.mjs (Playwright against a private mock-mode backend on :7681 and Vite on :5199)", + "notes": "Chromium via Playwright 1.56.1 from repos/pulse/node_modules. Real app and backend in PULSE_MOCK_MODE with ALLOW_ADMIN_BYPASS; the only stub is /api/license/runtime-capabilities with white_label added, because the mock install has no white-label entitlement and the card is gated behind it. Before the fix the same preview drew a blue circle containing a \"\u25cf\" text character instead of the Pulse brand mark, as seen in a provider-hosted client workspace on the walkthrough lab. Console errors were only the mock environment's failing update check." } diff --git a/frontend-modern/src/components/Settings/BrandingSettingsCard.tsx b/frontend-modern/src/components/Settings/BrandingSettingsCard.tsx index 27415994fe..b7830c18de 100644 --- a/frontend-modern/src/components/Settings/BrandingSettingsCard.tsx +++ b/frontend-modern/src/components/Settings/BrandingSettingsCard.tsx @@ -1,6 +1,7 @@ import { Show, createSignal, type Accessor, type Component, type Setter } from 'solid-js'; import ImageIcon from 'lucide-solid/icons/image'; import Trash2 from 'lucide-solid/icons/trash-2'; +import { PulseBrandMark } from '@/components/Brand/PulseBrandMark'; import { Button } from '@/components/shared/Button'; import { FeatureGateSection } from '@/components/shared/FeatureGateSection'; import { formControl, formHelpText, formLabel } from '@/components/shared/Form'; @@ -157,11 +158,7 @@ export const BrandingSettingsCard: Component = (props
- ● - - } + fallback={} > {(logo) => ( { expect(brandingLogoPreview('YWJj', 'png')).toBe('data:image/png;base64,YWJj'); }); + // Without a custom logo the preview must show what the header shows: the + // real Pulse brand mark, not a stand-in glyph. + it('previews the default header with the Pulse brand mark', () => { + const { container } = render(() => ( + ''} + setDisplayName={() => ''} + logoBase64={() => ''} + setLogoBase64={() => ''} + logoFormat={() => ''} + setLogoFormat={() => ''} + setHasUnsavedChanges={() => false} + /> + )); + const mark = container.querySelector('svg.pulse-brand-logo'); + expect(mark).not.toBeNull(); + expect(mark?.querySelectorAll('circle')).toHaveLength(3); + expect(container.textContent).not.toContain('●'); + expect(screen.queryByTestId('branding-logo-preview')).toBeNull(); + }); + it('updates the application name and marks settings dirty', async () => { const state = renderCard(); 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 06/14] 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) } } From 8098b9782562eb477c9d65fbc3de7fbecc167436 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 10:42:33 +0100 Subject: [PATCH 07/14] Audit agent config fetches only when they carry new information Agents poll /api/agents/agent/{id}/config every minute, and every successful poll was written as a signed agent_config_fetch audit event. On a five-agent homelab that was 483,000 of 483,101 audit rows (250 MB) over 75 days, burying the 64 logins and 25 CSRF failures the log exists to show; at the default 90-day retention it is roughly 60 MB per agent. Every failed fetch is still audited. A successful fetch is audited when it is new audit information: the agent's first delivery after Pulse starts, a different token or desiredConfig hash, or an unchanged delivery last recorded 24 hours earlier. Recorded successes now carry the delivered config hash and the reason. The tracker is per handler, keyed by org and agent, and bounded to 4,096 entries. Existing rows are left to age out under retention rather than deleted, since removing entries from a security log would look like tampering. AUDIT_LOGGING.md and the API and agent-lifecycle contracts describe the new rule. --- docs/AUDIT_LOGGING.md | 2 +- .../v6/internal/subsystems/agent-lifecycle.md | 5 ++ .../v6/internal/subsystems/api-contracts.md | 10 +++ .../internal/subsystems/storage-recovery.md | 11 +++ frontend-modern/public/docs/AUDIT_LOGGING.md | 2 +- internal/api/agent_config_fetch_audit.go | 80 +++++++++++++++++++ internal/api/agent_ingest.go | 21 ++++- ...gent_removal_lifecycle_integration_test.go | 71 ++++++++++++++++ internal/api/unified_agent_handlers_test.go | 33 ++++++++ 9 files changed, 230 insertions(+), 5 deletions(-) create mode 100644 internal/api/agent_config_fetch_audit.go diff --git a/docs/AUDIT_LOGGING.md b/docs/AUDIT_LOGGING.md index 15ff2bdd32..b6d397f85f 100644 --- a/docs/AUDIT_LOGGING.md +++ b/docs/AUDIT_LOGGING.md @@ -31,7 +31,7 @@ Pulse automatically captures the following events: | `agent_profile_assigned` | Agent profile assignments | Profile `production` assigned to agent | | `agent_profile_unassigned` | Agent profile removals | Profile removed from agent | | `user_roles_updated` | RBAC role assignments changed | Updated roles for user jane: [operator] | -| `agent_config_fetch` | Agent configuration retrieval attempts | Agent config fetched successfully | +| `agent_config_fetch` | Every failed agent configuration fetch. A successful fetch is recorded on the agent's first delivery after Pulse starts, when its token or delivered configuration changes, and otherwise once a day; agents poll every minute, so repeat polls are not recorded | `agent_id=… token_id=… config=sha256:… reason=config_changed` | Each event includes: - **Timestamp** (UTC) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 2dc2cee1e4..3e8d5f9ddc 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -5226,6 +5226,11 @@ agent truth, effective enforcement, and bounded reason separately so lifecycle surfaces can show desired-disabled/applied-enabled and desired-enabled/applied-disabled as drift or attention, and no-report cases as pending or unknown rather than in-sync. + +Routine config polling is not a per-poll audit event: +`/api/agents/agent/{id}/config` audits every failed fetch, and a successful one +only on the agent's first delivery after start, a token or desired-config +change, or once a day (see the API contract). That same canonical /api/auto-register path must also complete the live post-registration contract after persistence: it must trigger discovery refresh and emit the canonical `node_auto_registered` WebSocket payload instead of diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index e79dbd405f..bfc559794f 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -8694,6 +8694,16 @@ payload shape: `agentId`, `commandsEnabled`, `settings`, `issuedAt`, and signed command decision and the signed settings payload, restricted to the agent-applied settings key schema, rather than treating `desiredConfig` itself as a direct member of that signature payload. + +Its `agent_config_fetch` audit events record every failed fetch but a +successful fetch only when it is new audit information: the agent's first +delivery after Pulse starts, a different token or `desiredConfig` hash, or an +unchanged delivery last recorded 24 hours earlier. Agents poll every minute, +and auditing each poll made this event over 99.9% of audit rows. Recorded +successes carry `config=` and +`reason=first_since_start|token_changed|config_changed|daily` in their details. +`TestAgentConfigFetchAuditsNewDeliveriesAndEveryFailure` and +`TestAgentConfigFetchAuditTrackerRecordsOnlyNewDeliveries` pin the rule. Agent profile delete and unassign clients must now also route canonical `204` success handling through shared allowed-status helpers in `frontend-modern/src/api/responseUtils.ts` instead of open-coding local diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 4aa2124c9c..bd3a2eeae3 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -2674,6 +2674,17 @@ turning inactive rows into recovery state or preloading their inventories. The surface contract pins the route query map, the inactive recovery null query, and the direct Protection query parameters. +### Agent config fetch auditing leaves recovery evidence unchanged + +`/api/agents/agent/{id}/config` now records a successful fetch in the security +audit log only when it is new audit information (the agent's first delivery +after start, a token or desired-config change, or once a day), while every +failed fetch is still recorded. This changes no storage or recovery path: +recovery actions keep their own action audits, dispatch receipts, and +verification records, and only routine successful config polls stop producing +one security-audit row each. The rule is pinned by +`internal/api/host_agent_removal_lifecycle_integration_test.go`. + ### Fresh agent command policy preserves the existing recovery authority boundary New self-hosted install-command tokens project their explicit command-policy diff --git a/frontend-modern/public/docs/AUDIT_LOGGING.md b/frontend-modern/public/docs/AUDIT_LOGGING.md index 15ff2bdd32..b6d397f85f 100644 --- a/frontend-modern/public/docs/AUDIT_LOGGING.md +++ b/frontend-modern/public/docs/AUDIT_LOGGING.md @@ -31,7 +31,7 @@ Pulse automatically captures the following events: | `agent_profile_assigned` | Agent profile assignments | Profile `production` assigned to agent | | `agent_profile_unassigned` | Agent profile removals | Profile removed from agent | | `user_roles_updated` | RBAC role assignments changed | Updated roles for user jane: [operator] | -| `agent_config_fetch` | Agent configuration retrieval attempts | Agent config fetched successfully | +| `agent_config_fetch` | Every failed agent configuration fetch. A successful fetch is recorded on the agent's first delivery after Pulse starts, when its token or delivered configuration changes, and otherwise once a day; agents poll every minute, so repeat polls are not recorded | `agent_id=… token_id=… config=sha256:… reason=config_changed` | Each event includes: - **Timestamp** (UTC) diff --git a/internal/api/agent_config_fetch_audit.go b/internal/api/agent_config_fetch_audit.go new file mode 100644 index 0000000000..3d03d4de17 --- /dev/null +++ b/internal/api/agent_config_fetch_audit.go @@ -0,0 +1,80 @@ +package api + +import ( + "sync" + "time" +) + +// agentConfigFetchAuditInterval re-records an unchanged successful config +// delivery at most this often per agent, so the audit trail still shows +// continued access. +const agentConfigFetchAuditInterval = 24 * time.Hour + +// maxAgentConfigFetchAudits bounds remembered deliveries. Beyond it, entries +// older than agentConfigFetchAuditInterval are dropped; they would be +// re-recorded on their next fetch anyway. +const maxAgentConfigFetchAudits = 4096 + +// agentConfigFetchAuditTracker decides which successful agent config fetches +// are audit events. Agents poll their config every minute, and recording each +// poll made agent_config_fetch over 99.9% of audit rows (about 1,440 a day per +// agent), burying the logins and failures the log exists to show. Failed +// fetches are always audited and never pass through here. +type agentConfigFetchAuditTracker struct { + mu sync.Mutex + last map[agentConfigFetchAuditKey]agentConfigFetchAuditEntry +} + +type agentConfigFetchAuditKey struct { + orgID string + agentID string +} + +type agentConfigFetchAuditEntry struct { + tokenID string + configHash string + auditedAt time.Time +} + +func newAgentConfigFetchAuditTracker() *agentConfigFetchAuditTracker { + return &agentConfigFetchAuditTracker{last: make(map[agentConfigFetchAuditKey]agentConfigFetchAuditEntry)} +} + +// observe records a successful delivery and reports whether it is new audit +// information, with the reason: the agent's first delivery since startup, a +// different token or delivered config, or an unchanged delivery last audited +// at least agentConfigFetchAuditInterval ago. A nil tracker audits every +// delivery. +func (t *agentConfigFetchAuditTracker) observe(orgID, agentID, tokenID, configHash string, now time.Time) (string, bool) { + if t == nil { + return "", true + } + key := agentConfigFetchAuditKey{orgID: orgID, agentID: agentID} + entry := agentConfigFetchAuditEntry{tokenID: tokenID, configHash: configHash, auditedAt: now} + + t.mu.Lock() + defer t.mu.Unlock() + previous, seen := t.last[key] + reason := "" + switch { + case !seen: + reason = "first_since_start" + case previous.tokenID != tokenID: + reason = "token_changed" + case previous.configHash != configHash: + reason = "config_changed" + case now.Sub(previous.auditedAt) >= agentConfigFetchAuditInterval: + reason = "daily" + default: + return "", false + } + if !seen && len(t.last) >= maxAgentConfigFetchAudits { + for staleKey, stale := range t.last { + if now.Sub(stale.auditedAt) >= agentConfigFetchAuditInterval { + delete(t.last, staleKey) + } + } + } + t.last[key] = entry + return reason, true +} diff --git a/internal/api/agent_ingest.go b/internal/api/agent_ingest.go index 452c41ab77..5d43fa882c 100644 --- a/internal/api/agent_ingest.go +++ b/internal/api/agent_ingest.go @@ -43,6 +43,8 @@ type UnifiedAgentHandlers struct { // server upgrade on their next report instead of their next hourly update // check. Empty (the default) omits it from acks. serverVersion string + + configFetchAudits *agentConfigFetchAuditTracker } // SetServerVersion supplies the running server version to include on report @@ -63,7 +65,10 @@ func trimUnifiedAgentRoutePath(path string) string { // NewUnifiedAgentHandlers constructs a new handler set for Pulse Unified Agent ingest. func NewUnifiedAgentHandlers(mtm *monitoring.MultiTenantMonitor, m *monitoring.Monitor, hub *websocket.Hub) *UnifiedAgentHandlers { - return &UnifiedAgentHandlers{baseAgentHandlers: newBaseAgentHandlers(mtm, m, hub)} + return &UnifiedAgentHandlers{ + baseAgentHandlers: newBaseAgentHandlers(mtm, m, hub), + configFetchAudits: newAgentConfigFetchAuditTracker(), + } } // HandleReport ingests Pulse Unified Agent reports. @@ -607,8 +612,18 @@ func (h *UnifiedAgentHandlers) handleGetConfig(w http.ResponseWriter, r *http.Re return } - LogAuditEventForTenant(GetOrgID(r.Context()), "agent_config_fetch", auth.GetUser(r.Context()), GetClientIP(r), r.URL.Path, true, - fmt.Sprintf("agent_id=%s token_id=%s", agentID, tokenID(record))) + configHash := "" + if signedConfig.DesiredConfig != nil { + configHash = signedConfig.DesiredConfig.Hash + } + orgID := GetOrgID(r.Context()) + if reason, audit := h.configFetchAudits.observe(orgID, agentID, tokenID(record), configHash, time.Now()); audit { + details := fmt.Sprintf("agent_id=%s token_id=%s config=%s", agentID, tokenID(record), configHash) + if reason != "" { + details += " reason=" + reason + } + LogAuditEventForTenant(orgID, "agent_config_fetch", auth.GetUser(r.Context()), GetClientIP(r), r.URL.Path, true, details) + } } func tokenID(record *config.APITokenRecord) string { diff --git a/internal/api/host_agent_removal_lifecycle_integration_test.go b/internal/api/host_agent_removal_lifecycle_integration_test.go index 502a80c385..c32a44cd9b 100644 --- a/internal/api/host_agent_removal_lifecycle_integration_test.go +++ b/internal/api/host_agent_removal_lifecycle_integration_test.go @@ -16,6 +16,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" + "github.com/rcourtman/pulse-go-rewrite/pkg/audit" ) type hostRemovalLifecycleHTTPRuntime struct { @@ -647,3 +648,73 @@ func TestHostAgentRenameKeepsCommandGateAlignedWithChannelAdmission(t *testing.T t.Fatal("channel admission admitted a different agent ID on a matching hostname") } } + +// Agents poll their config every minute. Only deliveries that tell an auditor +// something new are recorded, while every failed fetch still is. +func TestAgentConfigFetchAuditsNewDeliveriesAndEveryFailure(t *testing.T) { + capture := &auditCaptureLogger{} + prevLogger := audit.GetLogger() + prevManager := GetTenantAuditManager() + audit.SetLogger(capture) + SetTenantAuditManager(nil) + t.Cleanup(func() { + audit.SetLogger(prevLogger) + SetTenantAuditManager(prevManager) + }) + + handler, monitor := newUnifiedAgentHandlers(t, nil) + hostID := seedUnifiedAgentHost(t, monitor) + monitorState(t, monitor).UpsertHost(models.Host{ID: hostID, Hostname: "node-1", TokenID: "runtime-token"}) + fetch := func(scopes ...string) int { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/agents/agent/"+hostID+"/config", nil) + attachAPITokenRecord(req, &config.APITokenRecord{ID: "runtime-token", Scopes: scopes}) + rec := httptest.NewRecorder() + handler.HandleConfig(rec, req) + return rec.Code + } + + for i := 0; i < 3; i++ { + if code := fetch(config.ScopeAgentConfigRead, config.ScopeAgentReport); code != http.StatusOK { + t.Fatalf("config fetch %d status = %d, want 200", i, code) + } + } + commandsEnabled := true + if err := monitor.UpdateHostAgentConfig(hostID, &commandsEnabled); err != nil { + t.Fatalf("UpdateHostAgentConfig: %v", err) + } + for i := 0; i < 2; i++ { + if code := fetch(config.ScopeAgentConfigRead, config.ScopeAgentReport); code != http.StatusOK { + t.Fatalf("config fetch after change %d status = %d, want 200", i, code) + } + } + for i := 0; i < 2; i++ { + if code := fetch(config.ScopeMonitoringRead); code == http.StatusOK { + t.Fatalf("config fetch with only %s succeeded", config.ScopeMonitoringRead) + } + } + + capture.mu.Lock() + events := append([]audit.Event(nil), capture.events...) + capture.mu.Unlock() + var successes []string + failures := 0 + for _, event := range events { + if event.EventType != "agent_config_fetch" { + continue + } + if event.Success { + successes = append(successes, event.Details) + } else { + failures++ + } + } + if len(successes) != 2 || + !strings.Contains(successes[0], "reason=first_since_start") || + !strings.Contains(successes[1], "reason=config_changed") { + t.Fatalf("successful fetch audit events = %q, want the first delivery and the config change", successes) + } + if failures != 2 { + t.Fatalf("failed fetch audit events = %d, want every failure", failures) + } +} diff --git a/internal/api/unified_agent_handlers_test.go b/internal/api/unified_agent_handlers_test.go index ffab87690b..4abb3b8a6f 100644 --- a/internal/api/unified_agent_handlers_test.go +++ b/internal/api/unified_agent_handlers_test.go @@ -552,3 +552,36 @@ func TestUnifiedAgentHandlers_HandleLinkUnlink(t *testing.T) { t.Fatalf("unlink status = %d, want 200: %s", unlinkRec.Code, unlinkRec.Body.String()) } } + +func TestAgentConfigFetchAuditTrackerRecordsOnlyNewDeliveries(t *testing.T) { + tracker := newAgentConfigFetchAuditTracker() + start := time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC) + for _, step := range []struct { + name string + token string + hash string + at time.Duration + reason string + audit bool + }{ + {"first delivery", "tok-a", "sha256:1", 0, "first_since_start", true}, + {"unchanged poll", "tok-a", "sha256:1", time.Minute, "", false}, + {"config change", "tok-a", "sha256:2", 2 * time.Minute, "config_changed", true}, + {"unchanged after change", "tok-a", "sha256:2", 3 * time.Minute, "", false}, + {"token rotation", "tok-b", "sha256:2", 4 * time.Minute, "token_changed", true}, + {"just under a day", "tok-b", "sha256:2", 4*time.Minute + agentConfigFetchAuditInterval - time.Second, "", false}, + {"a day since last audit", "tok-b", "sha256:2", 4*time.Minute + agentConfigFetchAuditInterval, "daily", true}, + } { + reason, audit := tracker.observe("default", "agent-1", step.token, step.hash, start.Add(step.at)) + if audit != step.audit || reason != step.reason { + t.Fatalf("%s: audit=%v reason=%q, want audit=%v reason=%q", step.name, audit, reason, step.audit, step.reason) + } + } + if reason, audit := tracker.observe("other-org", "agent-1", "tok-b", "sha256:2", start.Add(5*time.Minute)); !audit || reason != "first_since_start" { + t.Fatalf("same agent ID in another org: audit=%v reason=%q, want its own first delivery", audit, reason) + } + var unset *agentConfigFetchAuditTracker + if _, audit := unset.observe("default", "agent-1", "tok-a", "sha256:1", start); !audit { + t.Fatal("a handler without a tracker must audit every delivery") + } +} From 41abdd0b2c925cd61fa09e064d9232fa96150989 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 10:43:01 +0100 Subject: [PATCH 08/14] Let provider MSP upgrade.sh move an install to a new release setup.sh resolves the image pins in .env to digests once, so upgrade.sh only ever re-pulled the release a provider first installed. Moving to a new release meant finding the new digests by hand, so none of a release's fixes, and no new feature such as the portal Plan tab, reached an existing install. Run from a newly extracted release bundle, upgrade.sh now works on the existing install: it resolves the bundle's control-plane and runtime tags to digests, runs the status gate, preflight and verified backup with the new release's control plane through a shell override while nothing on disk changes, then keeps a copy of .env, installs the bundle files and writes the new pins before starting the new images. --keep-image-pins keeps hand-set pins. It also pulls the tenant runtime image before the first status check, which otherwise failed every upgrade to a new CP_PULSE_IMAGE with "not present locally", including the documented pin-editing route. --- deploy/provider-msp/.env.example | 14 +- deploy/provider-msp/upgrade.sh | 133 +++++++++++++- .../subsystems/deployment-installability.md | 26 +++ .../installtests/provider_msp_deploy_test.go | 164 ++++++++++++++++++ 4 files changed, 331 insertions(+), 6 deletions(-) diff --git a/deploy/provider-msp/.env.example b/deploy/provider-msp/.env.example index c5f271d119..59f1688bf8 100644 --- a/deploy/provider-msp/.env.example +++ b/deploy/provider-msp/.env.example @@ -113,10 +113,16 @@ PULSE_EMAIL_REPLY_TO=support@example.com # images: # ./upgrade.sh --dry-run # -# Apply a provider control-plane upgrade after updating the image pins in this -# file. The runner creates and verifies a fresh backup, dry-runs restore into a -# separate target data dir, pulls provider images, starts Traefik/control-plane, -# and prints the tenant runtime rollout plan for CP_PULSE_IMAGE: +# Move to a new release: download and verify its provider bundle as for a fresh +# install, then run upgrade.sh from the extracted bundle directory. It installs +# the bundle files here, keeps this file (and a pre-upgrade copy), re-pins +# CONTROL_PLANE_IMAGE and CP_PULSE_IMAGE to that release, creates and verifies a +# fresh backup, dry-runs restore into a separate target data dir, pulls provider +# images, starts Traefik/control-plane, and prints the tenant runtime rollout +# plan for CP_PULSE_IMAGE (--keep-image-pins keeps pins you set by hand): +# sudo -E bash ./upgrade.sh +# +# Run from this directory, upgrade.sh applies the image pins already here: # ./upgrade.sh # # Also roll all client runtimes to the configured CP_PULSE_IMAGE line after the diff --git a/deploy/provider-msp/upgrade.sh b/deploy/provider-msp/upgrade.sh index 2544a7e294..8ef8c0c272 100755 --- a/deploy/provider-msp/upgrade.sh +++ b/deploy/provider-msp/upgrade.sh @@ -7,6 +7,14 @@ usage() { Usage: ./upgrade.sh [options] +To move to a new release, download and verify its provider bundle exactly as +for a fresh install, then run ./upgrade.sh from the extracted bundle directory +(sudo -E bash ./upgrade.sh). It installs the bundle's files into the existing +install (PULSE_PROVIDER_MSP_INSTALL_DIR, default /opt/pulse-provider-msp), +keeps .env and a pre-upgrade copy of it, re-pins CONTROL_PLANE_IMAGE and +CP_PULSE_IMAGE to that release, and then runs the flow below. Run from the +install directory, it applies the image pins already in .env. + Runs the provider-hosted MSP pre-upgrade and upgrade flow: 1. validates .env and docker-compose.yml 2. checks provider status and install preflight @@ -26,6 +34,7 @@ Options: --restore-target DIR Restore dry-run target data dir (default: /upgrade-restore-drill) --run-id ID Operator-visible tenant reconcile run id --health-timeout DURATION Tenant rollout health timeout (default: 90s) + --keep-image-pins From a bundle: install its files but keep the image pins in .env -h, --help Show this help Environment equivalents: @@ -75,6 +84,98 @@ run_control() { docker compose run --rm --no-deps control-plane "$@" } +set_env_value() { + local key="$1" value="$2" env_path="$3" tmp + tmp="$(mktemp)" + if grep -q -E "^${key}=" "${env_path}"; then + awk -v key="${key}" -v value="${value}" 'BEGIN{done=0} $0 ~ "^" key "=" && done==0 { print key "=" value; done=1; next } { print }' "${env_path}" >"${tmp}" + else + cat "${env_path}" >"${tmp}" + printf '%s=%s\n' "${key}" "${value}" >>"${tmp}" + fi + cat "${tmp}" >"${env_path}" + rm -f "${tmp}" +} + +# resolve_image_digest turns a tag into an immutable digest ref, as setup.sh +# does, reading the registry without pulling the image. +resolve_image_digest() { + local ref="$1" manifest_json digest + manifest_json="$(docker buildx imagetools inspect "${ref}" --format '{{json .Manifest}}' 2>/dev/null || true)" + digest="$(printf '%s' "${manifest_json}" | jq -r 'if type == "object" then .digest // empty else empty end' 2>/dev/null || true)" + if [[ "${digest}" != sha256:* ]]; then + digest="$(docker buildx imagetools inspect "${ref}" 2>/dev/null | awk '$1 == "Digest:" {print $2; exit}' || true)" + fi + [[ "${digest}" == sha256:* ]] || return 1 + printf '%s@%s\n' "${ref%:*}" "${digest}" +} + +# The files setup.sh installs from a bundle. A bundle upgrade installs the same +# set, so an upgraded install matches a fresh one. +bundle_install_files=(docker-compose.yml traefik.yml traefik-dynamic.yml .env.example run-install-proof.sh upgrade.sh) + +# A bundle upgrade brings an existing install up to the release bundle this +# script was extracted with. Setup resolves the image pins in .env to digests +# once, so an upgrade used to re-pull only the release the provider first +# installed; moving on meant finding four digests by hand. Nothing on disk +# changes until the platform has been checked and backed up: those steps run +# the new release's control plane as a one-off through a shell override of the +# pins (compose prefers the shell over .env), which also carries fixes to the +# checks themselves. Only then are the bundle files installed and .env +# re-pinned, just before the new images start. +bundle_dir="" +bundle_version="" +bundle_pin_keys=() +bundle_pin_values=() + +plan_bundle_upgrade() { + local install_dir="$1" env_path key target current resolved + bundle_dir="$2" + bundle_version="$(tr -d '[:space:]' <"${bundle_dir}/VERSION")" + env_path="${install_dir}/.env" + [[ -f "${env_path}" ]] || die "no provider MSP install at ${install_dir} (missing .env); run ./setup.sh for a fresh install" + [[ -w "${install_dir}" && -w "${env_path}" ]] || die "cannot write ${install_dir}; run: sudo -E bash ./upgrade.sh" + echo "provider_msp_upgrade_bundle_version=${bundle_version}" + echo "provider_msp_upgrade_install_dir=${install_dir}" + if truthy "${keep_image_pins}"; then + echo "provider_msp_upgrade_image_pins=kept" + return 0 + fi + for key in CONTROL_PLANE_IMAGE CP_PULSE_IMAGE; do + target="$(env_value "${key}" "${bundle_dir}/.env.example")" + [[ -n "${target}" ]] || die "bundle .env.example has no ${key}; download the release bundle, not the source tree" + current="$(env_value "${key}" "${env_path}")" + if [[ "${target}" == *@sha256:* ]]; then + resolved="${target}" + elif ! resolved="$(resolve_image_digest "${target}")"; then + die "could not resolve a digest for ${target}; check this host can reach the registry" + fi + echo "provider_msp_upgrade_image_pin ${key} current=${current} target=${resolved}" + bundle_pin_keys+=("${key}") + bundle_pin_values+=("${resolved}") + export "${key}=${resolved}" + done +} + +apply_bundle_upgrade() { + local install_dir="$1" env_path stamp file i + env_path="${install_dir}/.env" + stamp="$(date -u +'%Y%m%dT%H%M%SZ')" + cp -p "${env_path}" "${env_path}.pre-upgrade-${stamp}" + echo "provider_msp_upgrade_env_backup=${env_path}.pre-upgrade-${stamp}" + for file in "${bundle_install_files[@]}"; do + [[ -f "${bundle_dir}/${file}" ]] || die "bundle is missing ${file}" + case "${file}" in + *.sh) install -m 0755 "${bundle_dir}/${file}" "${install_dir}/${file}" ;; + *) install -m 0644 "${bundle_dir}/${file}" "${install_dir}/${file}" ;; + esac + done + for ((i = 0; i < ${#bundle_pin_keys[@]}; i++)); do + set_env_value "${bundle_pin_keys[$i]}" "${bundle_pin_values[$i]}" "${env_path}" + done + echo "provider_msp_upgrade_bundle_installed=true" +} + script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" cd "${script_dir}" @@ -87,7 +188,7 @@ backup_output="${PROVIDER_MSP_UPGRADE_BACKUP_OUTPUT:-}" restore_target="${PROVIDER_MSP_UPGRADE_RESTORE_TARGET:-}" run_id="${PROVIDER_MSP_UPGRADE_RUN_ID:-provider-msp-upgrade-$(date -u +'%Y%m%dT%H%M%SZ')}" health_timeout="${PROVIDER_MSP_UPGRADE_HEALTH_TIMEOUT:-90s}" - +keep_image_pins="${PROVIDER_MSP_UPGRADE_KEEP_IMAGE_PINS:-0}" while (($# > 0)); do case "$1" in --dry-run) @@ -126,6 +227,10 @@ while (($# > 0)); do health_timeout="${2:-}" shift 2 ;; + --keep-image-pins) + keep_image_pins=1 + shift + ;; -h|--help) usage exit 0 @@ -138,13 +243,22 @@ while (($# > 0)); do esac done +# Run from an extracted release bundle (it carries VERSION and no .env), work +# on the existing install and move it to that release. +install_dir="${script_dir}" +if [[ -f "${script_dir}/VERSION" && ! -f "${script_dir}/.env" ]]; then + install_dir="${PULSE_PROVIDER_MSP_INSTALL_DIR:-/opt/pulse-provider-msp}" + plan_bundle_upgrade "${install_dir}" "${script_dir}" + cd "${install_dir}" +fi + if [[ ! -f .env ]]; then die "deploy/provider-msp/.env is required; copy .env.example to .env and fill in the provider values" fi provider_data_dir="$(env_value PULSE_PROVIDER_MSP_DATA_DIR .env)" provider_data_dir="${provider_data_dir:-/data}" -tenant_runtime_image="$(env_value CP_PULSE_IMAGE .env)" +tenant_runtime_image="${CP_PULSE_IMAGE:-$(env_value CP_PULSE_IMAGE .env)}" if [[ -z "${tenant_runtime_image}" ]]; then die "CP_PULSE_IMAGE is required" fi @@ -169,11 +283,21 @@ echo "provider_msp_upgrade_run_id=${run_id}" echo "provider_msp_upgrade_restore_target=${restore_target}" echo "provider_msp_upgrade_tenant_runtime_image=${tenant_runtime_image}" +# Status checks that the tenant runtime image is already on this host, and runs +# before anything else would pull it, so a new CP_PULSE_IMAGE failed the gate on +# every upgrade. Pulling changes nothing that is running. +if ! truthy "${skip_runtime_image_pull}"; then + docker pull "${tenant_runtime_image}" >/dev/null +fi + run_control provider-msp status run_control "${preflight_args[@]}" if truthy "${dry_run}"; then run_control tenant-runtime rollout --all --image "${tenant_runtime_image}" --dry-run + if [[ -n "${bundle_dir}" ]]; then + echo "provider_msp_upgrade_bundle_installed=false" + fi echo "tenant_runtime_rollout_applied=false" echo "provider_msp_upgrade_plan_ok=true" exit 0 @@ -195,6 +319,11 @@ run_control provider-msp backup verify "${archive_path}" run_control provider-msp backup restore "${archive_path}" --target-data-dir "${restore_target}" --dry-run run_control provider-msp status --require-backup +# The platform is checked and backed up; now install the bundle and re-pin. +if [[ -n "${bundle_dir}" ]]; then + apply_bundle_upgrade "${install_dir}" +fi + if ! truthy "${skip_compose_pull}"; then docker compose pull traefik docker-socket-proxy control-plane fi diff --git a/docs/release-control/v6/internal/subsystems/deployment-installability.md b/docs/release-control/v6/internal/subsystems/deployment-installability.md index 419db400c3..a9a58c9055 100644 --- a/docs/release-control/v6/internal/subsystems/deployment-installability.md +++ b/docs/release-control/v6/internal/subsystems/deployment-installability.md @@ -2438,6 +2438,32 @@ artifact-selection behaviour. ## Current State +### Provider MSP upgrades move an install to a new release bundle + +`setup.sh` resolves the image pins in `.env` to digests once, so `upgrade.sh` +only ever re-pulled the release a provider first installed; moving to a new +release meant finding the new digests by hand, and none of that release's +fixes reached the install. Run from a newly extracted and verified release +bundle (it carries `VERSION` and no `.env`), `upgrade.sh` now works on the +existing install (`PULSE_PROVIDER_MSP_INSTALL_DIR`, default +`/opt/pulse-provider-msp`): it resolves the bundle's `CONTROL_PLANE_IMAGE` and +`CP_PULSE_IMAGE` tags (stamped by `scripts/build-release.sh`) to digests and +prints current and target pins. It runs the status gate, preflight and the +verified backup with the new release's control plane through a shell override, +which compose prefers over `.env`, so those checks carry the new release's +fixes while nothing on disk changes. Only then does it keep a +`.env.pre-upgrade-