diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 354ad8d5e2..fd23f74cc1 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,30 @@ ## Purpose +### Fresh broadcast projection owns one resource/target capture + +Canonical live, fallback and standalone-host continuity broadcast reads prefer +the adapter's detached bulk resource/metrics-target capture. After host +coalescing selects presentation identity, missing history targets are attached +from that same capture, including authoritative absence. Stores without this +optional capability retain the previous point-resolver fallback. Only the owned +outer coalesced slice is decorated in place; nested source payloads stay read-only. + +Every read still observes complete current resource data and derives live alert +health, metadata URL edits/clears and connected/ignored infrastructure. No +timestamp-only cache, telemetry omission, policy change or new wire field is +introduced. All existing catalogs, discovery/history coordinates, original +observation times, LastSeen, identity and tenant/revocation boundaries remain. + +`broadcast_metrics_snapshot_test.go` compares the complete served JSON with an +independent earlier projection across the metadata-heavy nine-provider demo, +typed-view and same-time dirty cases; it also covers captured target replacement +and continuity/metadata/live-alert boundaries. `monitor_host_agents_test.go` +retains its one-materialization and live-change regression. Benchmarks include +the exact parent's complete conversion and final JSON encoding, with cold, +list-only, typed-clean and dirty inputs; they do not establish installed CPU/RSS +relief, native identity refresh throughput or completion of #2199. + ### Report identity lookups do not rebuild unrelated Kubernetes metadata The producer's two temporary host/Docker-host admission registries use a diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 829cedf05e..1bd25916bf 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -23,6 +23,29 @@ and sort the complete canonical change table while startup and ingestion wait. ## Purpose +### Broadcast bulk reads capture resources and history coordinates together + +`ListWithMetricsTargets` and the monitor adapter's matching optional read return +detached resources and a detached target map from one locked registry generation. +The existing source-mapping inversion runs once for this read when typed views +are dirty; it does not build all typed views or add a retained generation cache. +Clean typed views may supply their already-current inverse mapping. Legacy List, +GetAll and individual target reads preserve their shapes and semantics. + +The target map stays separate from resources until presentation host coalescing +selects the row identity. Missing coordinates remain missing in that capture; +they never fall through to a later live registry. Mapping changes with unchanged +LastSeen take effect on the next read. Returned resources, policy/identity blocks +and target values remain caller-owned across mutations and adapter replacement. +Projection cannot pair an earlier resource row with a later history identity. + +`registry_test.go`, `registry_projection_test.go` and +`monitor_adapter_read_state_test.go` exercise full resource/point-target parity, +same-time edits, concurrent mapping writes, detachments, empty/nil reads and +tenant separation. Connected monitoring proof covers all nine demo providers, +continuity overlays and complete served JSON. Paired cold/clean/dirty benchmarks +measure whole materialization, not a new point-lookup envelope or native relief. + Canonical resource History honours selected memory observation provenance and Proxmox VM filesystem deferrals instead of treating numeric carriers as fresh measurements. ### Canonical resource History cannot promote retained guest readings diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go new file mode 100644 index 0000000000..4e61cc2f3d --- /dev/null +++ b/internal/monitoring/broadcast_metrics_snapshot_test.go @@ -0,0 +1,264 @@ +package monitoring + +import ( + "encoding/json" + "fmt" + "reflect" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +func projectionFixtureGraph(t testing.TB) mock.FixtureGraph { + t.Helper() + previousEnabled, previousConfig := mock.IsMockEnabled(), mock.GetConfig() + t.Cleanup(func() { + mustSetMockEnabled(t, false) + mock.SetMockConfig(previousConfig) + mustSetMockEnabled(t, previousEnabled) + }) + mustSetMockEnabled(t, false) + cfg := mock.DefaultConfig + cfg.UpdateInterval = 5 * time.Minute + mock.SetMockConfig(cfg) + mustSetMockEnabled(t, true) + graph := mock.CurrentFixtureGraph() + mustSetMockEnabled(t, false) + return graph +} + +func projectionFixtureRegistry(graph mock.FixtureGraph) *unifiedresources.ResourceRegistry { + rr := unifiedresources.NewRegistry(nil) + rr.IngestSnapshot(graph.State) + for _, source := range []unifiedresources.DataSource{unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} { + rr.IngestRecords(source, graph.SupplementalRecords(source)) + } + return rr +} + +// The original full projection, including target resolution after host +// coalescing. Compare all wire fields/catalogues, not selected metric values. +func legacyFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend { + rows := unifiedresources.CoalescePresentationHostResources(resources) + rows = m.applyPersistedMetadataToUnifiedResources(rows) + rows = unifiedresources.AttachResourceHealth(rows, resourceHealthAlerts(snapshot.ActiveAlerts), now) + projected, catalogs := convertResourcesForBroadcastReference(rows, resolver) + out := snapshot.ToFrontend() + out.Resources = projected + out.CapabilityCatalog = catalogs.capabilities + out.PolicyCatalog = catalogs.policies + out.AISafeSummaryCatalog = catalogs.aiSafeSummaries + out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot) + return out +} + +func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) { + graph := projectionFixtureGraph(t) + rr := projectionFixtureRegistry(graph) + adapter := unifiedresources.NewMonitorAdapter(rr) + store := &broadcastProjectionCountingStore{MonitorAdapter: adapter} + m := &Monitor{resourceStore: store} + resources := adapter.GetAll() + if len(resources) < 1000 { + t.Fatalf("demo fixture too small: %d", len(resources)) + } + sources := map[unifiedresources.DataSource]bool{} + for _, r := range resources { + for _, source := range r.Sources { + sources[source] = true + } + } + for _, source := range []unifiedresources.DataSource{unifiedresources.SourceProxmox, unifiedresources.SourceAgent, unifiedresources.SourceDocker, unifiedresources.SourcePBS, unifiedresources.SourcePMG, unifiedresources.SourceK8s, unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} { + if !sources[source] { + t.Fatalf("demo omits provider %s", source) + } + } + assertEqual := func(stage string) { + t.Helper() + before, err := json.Marshal(adapter.GetAll()) + if err != nil { + t.Fatal(err) + } + want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC()) + got := m.buildBroadcastFrontendStateFromSnapshot(graph.State) + wantJSON, err := json.Marshal(want) + if err != nil { + t.Fatal(err) + } + gotJSON, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + if string(wantJSON) != string(gotJSON) { + t.Fatalf("%s: complete frontend/catalogue/infrastructure JSON differs", stage) + } + after, _ := json.Marshal(adapter.GetAll()) + if string(before) != string(after) { + t.Fatalf("%s: projection changed source data", stage) + } + t.Logf("%s: %d complete resources, %d frontend rows, %d wire bytes equal legacy projection", stage, len(adapter.GetAll()), len(got.Resources), len(gotJSON)) + } + assertEqual("cold-list-only") + rr.Workloads() + assertEqual("typed-views-clean") + // Same timestamps, new payload, policy and URL. No time-only invalidation. + resources = adapter.GetAll() + resources[0].Name = "changed-resource" + resources[0].Tags = []string{"customer-data"} + resources[0].CustomURL = "https://example.invalid/changed" + rr.IngestResources(resources) + assertEqual("dirty-same-time") +} + +func TestBroadcastMetricsSnapshotOwnsTargetsBeforeLiveReplacement(t *testing.T) { + m, adapter, _ := newReadStateCloneTestMonitor(t, 4) + view := m.currentUnifiedStateView() + if view.metricsTargets == nil { + t.Fatal("broadcast did not capture a bulk resolver") + } + old := view.resources[0] + want := adapter.MetricsTargetForResource(old.ID) + adapter.PopulateFromSnapshot(models.EmptyStateSnapshot()) + if got := view.metricsTargets.MetricsTargetForResource(old.ID); !reflect.DeepEqual(got, want) { + t.Fatal("captured resources were paired with a later generation's target") + } + if got := view.metricsTargets.MetricsTargetForResource("unknown"); got != nil { + t.Fatal("unknown target fell through to live data") + } + copyTarget := view.metricsTargets.MetricsTargetForResource(old.ID) + copyTarget.ResourceID = "caller-edit" + if got := view.metricsTargets.MetricsTargetForResource(old.ID); got.ResourceID == "caller-edit" { + t.Fatal("caller mutated the captured resolver") + } +} + +func TestBroadcastMetricsSnapshotContinuityAndLiveOverlays(t *testing.T) { + m, adapter, _ := newReadStateCloneTestMonitor(t, 2) + now := time.Now().UTC() + m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil) + entry := config.HostContinuityEntry{HostID: "absent-host", Hostname: "absent.example", MachineID: "absent-machine", LastSeen: now.Add(-time.Hour)} + if err := m.hostContinuityStore.Upsert(entry); err != nil { + t.Fatal(err) + } + view := m.unifiedStateViewWithStandaloneHostContinuity(monitorUnifiedStateView{readState: adapter}) + if len(view.resources) != 3 || view.metricsTargets == nil { + t.Fatal("continuity projection omitted its resources or captured resolver") + } + var host unifiedresources.Resource + for _, r := range view.resources { + if r.Agent != nil && r.Agent.AgentID == entry.HostID { + host = r + } + if got, want := view.metricsTargets.MetricsTargetForResource(r.ID), broadcastMetricsTargetResolver(view.readState).MetricsTargetForResource(r.ID); !reflect.DeepEqual(got, want) { + t.Fatal("overlay target differs from overlay registry") + } + } + if host.ID == "" || host.Status == unifiedresources.StatusOnline || !host.LastSeen.Equal(entry.LastSeen) { + t.Fatal("continuity history was omitted or promoted to current") + } + m.hostMetadataStore = config.NewHostMetadataStore(t.TempDir(), nil) + for _, url := range []string{"https://example.invalid/operator", ""} { + if err := m.hostMetadataStore.Set(entry.HostID, &config.HostMetadata{CustomURL: url}); err != nil { + t.Fatal(err) + } + decorated := m.applyPersistedMetadataToUnifiedResources(view.resources) + for _, r := range decorated { + if r.ID == host.ID && r.CustomURL != url { + t.Fatal("operator URL change/clear was hidden by captured data") + } + } + } + critical := unifiedresources.EvaluateResourceHealth(host, []unifiedresources.ResourceHealthAlert{{ResourceID: host.ID, Level: "critical", Type: "cpu"}}, now) + if critical.Verdict != unifiedresources.HealthCritical { + t.Fatal("live alert was hidden by captured data") + } + if len(adapter.GetAll()) != 2 { + t.Fatal("continuity overlay wrote back to the live registry") + } +} + +var completeProjectionSink models.StateFrontend +var completeProjectionJSONSink []byte + +// Exact parent broadcast shape (one coalesce, copy-to-attach, then conversion), +// not the older two-coalesce content oracle above. The unchanged conversion, +// metadata/health decoration and snapshot projection are part of both costs. +func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend { + rows := unifiedresources.CoalescePresentationHostResources(resources) + for i := range rows { + m.applyPersistedMetadataToUnifiedResource(&rows[i]) + health := unifiedresources.EvaluateResourceHealth(rows[i], resourceHealthAlerts(snapshot.ActiveAlerts), now) + rows[i].Health = &health + } + projected, catalogs := convertPresentationResourcesForBroadcast(attachBroadcastMetricsTargets(rows, resolver)) + out := snapshot.ToFrontend() + out.Resources = projected + out.CapabilityCatalog = catalogs.capabilities + out.PolicyCatalog = catalogs.policies + out.AISafeSummaryCatalog = catalogs.aiSafeSummaries + out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot) + return out +} + +func BenchmarkBroadcastMetricsSnapshot(b *testing.B) { + graph := projectionFixtureGraph(b) + for _, mode := range []string{"cold", "list-clean", "views-clean", "dirty"} { + for _, encode := range []bool{false, true} { + stage := "projection" + if encode { + stage = "json-frame" + } + for _, bulk := range []bool{false, true} { + path := "point-targets" + if bulk { + path = "bulk-targets" + } + b.Run(fmt.Sprintf("%s/%s/%s", mode, stage, path), func(b *testing.B) { + rr := projectionFixtureRegistry(graph) + adapter := unifiedresources.NewMonitorAdapter(rr) + store := &broadcastProjectionCountingStore{MonitorAdapter: adapter} + m := &Monitor{resourceStore: store} + if mode != "cold" { + rr.List() + } + if mode == "views-clean" { + rr.Workloads() + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if mode == "cold" || mode == "dirty" { + b.StopTimer() + if mode == "cold" { + rr = projectionFixtureRegistry(graph) + adapter = unifiedresources.NewMonitorAdapter(rr) + store.MonitorAdapter = adapter + } else { + changed := rr.List()[0] + changed.Tags = []string{fmt.Sprintf("edit-%d", i)} + rr.IngestResources([]unifiedresources.Resource{changed}) + } + b.StartTimer() + } + if bulk { + completeProjectionSink = m.buildBroadcastFrontendStateFromSnapshot(graph.State) + } else { + completeProjectionSink = parentFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC()) + } + if encode { + var err error + completeProjectionJSONSink, err = json.Marshal(completeProjectionSink) + if err != nil { + b.Fatal(err) + } + } + } + }) + } + } + } +} diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 165f94a540..9e11cb4f7e 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -607,10 +607,11 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) { source := string(data) for _, snippet := range []string{ - "metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)", + "metricsTargetResolver := unifiedView.metricsTargets", + "metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)", "broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)", - "broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(", - "attachBroadcastMetricsTargets(broadcastResources, metricsTargetResolver)", + "broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)", + "attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)", "frontendState.CapabilityCatalog = broadcastCatalogs.capabilities", "frontendState.PolicyCatalog = broadcastCatalogs.policies", "frontendState.AISafeSummaryCatalog = broadcastCatalogs.aiSafeSummaries", diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 6bab63084c..162d189642 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -4341,7 +4341,10 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS } } unifiedView := m.currentUnifiedStateView() - metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState) + metricsTargetResolver := unifiedView.metricsTargets + if metricsTargetResolver == nil { + metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState) + } broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources) // Coalescing owns the outer slice. Decorate that one projection in place, // not three full-resource copies; nested store data is still read-only. @@ -4352,9 +4355,8 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS health := unifiedresources.EvaluateResourceHealth(broadcastResources[i], healthAlerts, now) broadcastResources[i].Health = &health } - broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast( - attachBroadcastMetricsTargets(broadcastResources, metricsTargetResolver), - ) + attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver) + broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources) frontendState.Resources = broadcastFrontendResources frontendState.CapabilityCatalog = broadcastCatalogs.capabilities frontendState.PolicyCatalog = broadcastCatalogs.policies @@ -4871,23 +4873,50 @@ func (m *Monitor) RecentResourceChanges(resourceID string, since time.Time, limi } type monitorUnifiedStateView struct { - resources []unifiedresources.Resource - readState unifiedresources.ReadState - freshness time.Time + resources []unifiedresources.Resource + readState unifiedresources.ReadState + freshness time.Time + metricsTargets MetricsTargetResourceStore } type unifiedResourceReadStateLister interface { GetAll() []unifiedresources.Resource } +type unifiedResourceProjectionLister interface { + GetAllWithMetricsTargets() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget) +} + +// This resolver never falls through to the live store, including for an +// unknown ID. A missing target belongs to this captured generation too. +type projectionMetricsTargets map[string]unifiedresources.MetricsTarget + +func (targets projectionMetricsTargets) MetricsTargetForResource(id string) *unifiedresources.MetricsTarget { + target, ok := targets[unifiedresources.CanonicalResourceID(id)] + if !ok { + return nil + } + return &target +} + +func unifiedProjectionResources(lister unifiedResourceReadStateLister) ([]unifiedresources.Resource, MetricsTargetResourceStore) { + if projection, ok := lister.(unifiedResourceProjectionLister); ok { + resources, targets := projection.GetAllWithMetricsTargets() + return resources, projectionMetricsTargets(targets) + } + return lister.GetAll(), nil +} + func monitorUnifiedStateViewFromSnapshot(snapshot models.StateSnapshot) monitorUnifiedStateView { registry := unifiedresources.NewRegistry(nil) registry.IngestSnapshot(snapshot) adapter := unifiedresources.NewMonitorAdapter(registry) + resources, targets := unifiedProjectionResources(adapter) return monitorUnifiedStateView{ - resources: registry.List(), - readState: adapter, - freshness: snapshot.LastUpdate, + resources: resources, + readState: adapter, + freshness: snapshot.LastUpdate, + metricsTargets: targets, } } @@ -4895,10 +4924,12 @@ func monitorUnifiedStateViewFromResources(resources []unifiedresources.Resource, registry := unifiedresources.NewRegistry(nil) registry.IngestResources(resources) adapter := unifiedresources.NewMonitorAdapter(registry) + resources, targets := unifiedProjectionResources(adapter) return monitorUnifiedStateView{ - resources: registry.List(), - readState: adapter, - freshness: freshness, + resources: resources, + readState: adapter, + freshness: freshness, + metricsTargets: targets, } } @@ -4915,8 +4946,9 @@ func (m *Monitor) unifiedStateViewWithStandaloneHostContinuity(view monitorUnifi return view } - resources := lister.GetAll() + resources, targets := unifiedProjectionResources(lister) view.resources = resources + view.metricsTargets = targets if view.freshness.IsZero() { view.freshness = latestUnifiedResourceLastSeen(resources) } @@ -6479,17 +6511,24 @@ func attachBroadcastMetricsTargets( } out := make([]unifiedresources.Resource, len(resources)) - for i, resource := range resources { - if resource.MetricsTarget == nil { - resource.MetricsTarget = validBroadcastMetricsTarget( - resolver.MetricsTargetForResource(resource.ID), - ) - } - out[i] = resource - } + copy(out, resources) + attachBroadcastMetricsTargetsInPlace(out, resolver) return out } +// Only call with the owned outer slice of a coalesced broadcast projection. +// Nested resource data stays read-only, and the resolver returns owned targets. +func attachBroadcastMetricsTargetsInPlace(resources []unifiedresources.Resource, resolver MetricsTargetResourceStore) { + if resolver == nil { + return + } + for i := range resources { + if resources[i].MetricsTarget == nil { + resources[i].MetricsTarget = validBroadcastMetricsTarget(resolver.MetricsTargetForResource(resources[i].ID)) + } + } +} + func validBroadcastMetricsTarget( target *unifiedresources.MetricsTarget, ) *unifiedresources.MetricsTarget { diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 12e934c142..695d475395 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -6718,6 +6718,10 @@ func (s *broadcastProjectionCountingStore) GetAll() []unifiedresources.Resource s.reads++ return s.MonitorAdapter.GetAll() } +func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargets() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget) { + s.reads++ + return s.MonitorAdapter.GetAllWithMetricsTargets() +} func (*broadcastProjectionCountingStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool { return false } diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index 28eda9d110..8bac41ceaa 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -496,6 +496,18 @@ func (a *MonitorAdapter) GetAll() []Resource { return registry.List() } +// GetAllWithMetricsTargets materializes one complete broadcast read, binding +// its detached target map to the same registry as its detached resource list. +// The adapter may publish a replacement immediately after this capture; neither +// half of the returned read then consults that newer generation. +func (a *MonitorAdapter) GetAllWithMetricsTargets() ([]Resource, map[string]MetricsTarget) { + registry := a.currentRegistry() + if registry == nil { + return nil, nil + } + return registry.ListWithMetricsTargets() +} + func (a *MonitorAdapter) unifiedAIAdapter() *UnifiedAIAdapter { registry := a.currentRegistry() if registry == nil { diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go index 5eb791f3cf..7eefcb5b99 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -971,3 +971,40 @@ func TestHostContinuityOverlayReusedWithinRegistryGeneration(t *testing.T) { t.Fatal("overlay was reused past its maximum age") } } + +func TestMonitorAdapterProjectionSnapshotKeepsReplacedGeneration(t *testing.T) { + var absent *MonitorAdapter + rows, targets := absent.GetAllWithMetricsTargets() + if rows != nil || targets != nil { + t.Fatal("nil adapter invented a projection") + } + adapter := NewMonitorAdapter(NewRegistry(nil)) + snapshot := readRefreshSnapshot("before") + adapter.PopulateFromSnapshot(snapshot) + first, firstTargets := adapter.GetAllWithMetricsTargets() + snapshot.VMs[0].Name = "after" + snapshot.VMs[0].ID = "new-node:201" + snapshot.VMs[0].VMID = 201 + // Observation time is deliberately unchanged. + adapter.PopulateFromSnapshot(snapshot) + second, secondTargets := adapter.GetAllWithMetricsTargets() + if len(first) != 1 || len(second) != 1 || first[0].Name != "before" || second[0].Name != "after" { + t.Fatal("replacement lost complete fresh resource content") + } + if firstTargets[first[0].ID].ResourceID == secondTargets[second[0].ID].ResourceID { + t.Fatal("replacement retained old history coordinates") + } + for _, r := range second { + if want := adapter.MetricsTargetForResource(r.ID); want == nil || *want != secondTargets[r.ID] { + t.Fatal("target differs from the generation's point resolver") + } + } + other := NewMonitorAdapter(NewRegistry(nil)) + snapshot.VMs[0].ID = "other-tenant:501" + snapshot.VMs[0].Name = "other-tenant" + other.PopulateFromSnapshot(snapshot) + otherRows, otherTargets := other.GetAllWithMetricsTargets() + if reflect.DeepEqual(otherRows, second) || reflect.DeepEqual(otherTargets, secondTargets) { + t.Fatal("separate adapters shared a tenant projection") + } +} diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 774f2d8930..34992538c1 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -1321,6 +1321,21 @@ func seededVMwareSourceID(resource *Resource) string { // the read lock and copy it without repeating identity/scope/policy derivation. // The typed views remain lazy, so a List-only consumer pays for no extra copy. func (rr *ResourceRegistry) List() []Resource { + resources, _ := rr.listMaterialized(false) + return resources +} + +// ListWithMetricsTargets captures detached resources and their history targets +// from the same registry generation. Resolve all source mappings in one pass, +// without forcing the unrelated typed views to be built. In particular a +// freshly ingested registry must not scan every mapping once per broadcast row. +// Targets are separate: host coalescing must select the presentation identity +// before a missing target is attached, just as with individual target lookups. +func (rr *ResourceRegistry) ListWithMetricsTargets() ([]Resource, map[string]MetricsTarget) { + return rr.listMaterialized(true) +} + +func (rr *ResourceRegistry) listMaterialized(withTargets bool) ([]Resource, map[string]MetricsTarget) { for { rr.mu.RLock() if rr.canonicalMetadataDirty { @@ -1332,13 +1347,27 @@ func (rr *ResourceRegistry) List() []Resource { // rather than copy a generation that is no longer materialized. continue } + var sourceTargets map[string][]SourceTarget + var targets map[string]MetricsTarget + if withTargets { + sourceTargets = rr.cachedSourceTargets + if rr.viewsDirty || sourceTargets == nil { + sourceTargets = rr.buildSourceTargetsIndexLocked() + } + targets = make(map[string]MetricsTarget, len(rr.resources)) + } out := make([]Resource, 0, len(rr.resources)) for _, r := range rr.resources { out = append(out, cloneMaterializedResource(r)) + if withTargets { + if target := rr.metricsTargetFromSourceTargets(r, sourceTargets[r.ID]); target != nil { + targets[r.ID] = *target + } + } } rr.mu.RUnlock() sortResourcesByName(out) - return out + return out, targets } } diff --git a/internal/unifiedresources/registry_projection_test.go b/internal/unifiedresources/registry_projection_test.go new file mode 100644 index 0000000000..ca9dda8383 --- /dev/null +++ b/internal/unifiedresources/registry_projection_test.go @@ -0,0 +1,99 @@ +package unifiedresources + +import ( + "fmt" + "sync" + "testing" +) + +// Exercise the capture while a writer changes the mapping and row together. +// Separate List/point reads can legitimately observe different generations; +// this paired bulk read cannot. The test never touches persistence or drivers. +func TestRegistryProjectionSnapshotConcurrentMappingEdits(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshot(benchmarkVMState(1)) + id := rr.List()[0].ID + set := func(version int) { + rr.mu.Lock() + defer rr.mu.Unlock() + name := fmt.Sprintf("coordinate-%d", version) + rr.resources[id].Name = name + rr.bySource[SourceProxmox] = map[string]string{name: id} + rr.invalidateViewsLocked() + } + set(0) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + set(i % 2) + } + }() + defer wg.Wait() + for i := 0; i < 200; i++ { + rows, targets := rr.ListWithMetricsTargets() + if len(rows) != 1 || rows[0].Name != targets[id].ResourceID { + t.Fatalf("mixed row/target generation: row=%q target=%q", rows[0].Name, targets[id].ResourceID) + } + } +} + +var projectionResourceSink []Resource +var projectionTargetSink map[string]MetricsTarget + +// Include the entire owned List and every history coordinate. These are not +// point-lookup timings, nor a replacement for the old clone-allocation envelope. +func BenchmarkRegistryProjectionSnapshot(b *testing.B) { + for _, count := range []int{0, 1, 1000} { + for _, mode := range []string{"cold", "list-clean", "views-clean", "dirty"} { + for _, bulk := range []bool{false, true} { + name := "point-targets" + if bulk { + name = "bulk-targets" + } + b.Run(fmt.Sprintf("%d/%s/%s", count, mode, name), func(b *testing.B) { + state := benchmarkVMState(count) + rr := NewRegistry(nil) + rr.IngestSnapshot(state) + if mode != "cold" { + rr.List() + } + if mode == "views-clean" { + rr.VMs() + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if mode == "cold" { + b.StopTimer() + rr = NewRegistry(nil) + rr.IngestSnapshot(state) + b.StartTimer() + } + if mode == "dirty" { + rr.mu.Lock() + for _, r := range rr.resources { + r.Tags = []string{fmt.Sprintf("version-%d", i)} + break + } + rr.invalidateViewsLocked() + rr.mu.Unlock() + } + if bulk { + projectionResourceSink, projectionTargetSink = rr.ListWithMetricsTargets() + } else { + projectionResourceSink = rr.List() + projectionTargetSink = make(map[string]MetricsTarget, len(projectionResourceSink)) + for _, row := range projectionResourceSink { + if target := rr.MetricsTarget(row.ID); target != nil { + projectionTargetSink[row.ID] = *target + } + } + } + } + }) + } + } + } +} diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index e7490bef07..8eacc8f37b 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -6968,3 +6968,68 @@ func TestEvaluateResourceHealthKeepsStaleTelemetryBehindAWinningAlert(t *testing t.Fatalf("expected only the alert reason for a fresh source, got %+v", health.Reasons) } } + +// A List-only broadcast must capture history coordinates without constructing +// every typed view, and without trusting freshness timestamps as mutation keys. +func TestRegistryProjectionSnapshotDetachesAndTracksMappings(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshot(benchmarkVMState(3)) + resources, targets := rr.ListWithMetricsTargets() + if !reflect.DeepEqual(resources, rr.List()) { + t.Fatal("bulk read changed resource content") + } + if !rr.viewsDirty || rr.cachedVMs != nil { + t.Fatal("projection forced unrelated typed views to materialize") + } + if len(targets) != len(resources) { + t.Fatalf("targets=%d resources=%d", len(targets), len(resources)) + } + for _, r := range resources { + if want := rr.MetricsTarget(r.ID); want == nil || *want != targets[r.ID] { + t.Fatalf("target for %s differs from point resolver", r.ID) + } + } + id := resources[0].ID + observedAt := resources[0].LastSeen + rr.mu.Lock() + for key, mapped := range rr.bySource[SourceProxmox] { + if mapped == id { + delete(rr.bySource[SourceProxmox], key) + } + } + rr.bySource[SourceProxmox]["new-history-coordinate"] = id + rr.resources[id].Tags = []string{"customer-data"} + rr.invalidateViewsLocked() + rr.mu.Unlock() + changed, updatedTargets := rr.ListWithMetricsTargets() + if updatedTargets[id].ResourceID != "new-history-coordinate" { + t.Fatal("same-time mapping change left an old target") + } + if targets[id].ResourceID == "new-history-coordinate" { + t.Fatal("later mapping edit mutated a prior target snapshot") + } + for _, r := range changed { + if r.ID == id { + if !r.LastSeen.Equal(observedAt) || r.Policy.Sensitivity != ResourceSensitivityRestricted { + t.Fatal("same-time metadata edit was stale") + } + r.Canonical.Aliases[0] = "caller-change" + r.Tags[0] = "caller-change" + r.Policy.Routing.Redact = append(r.Policy.Routing.Redact, ResourceRedactionPath) + } + } + again, againTargets := rr.ListWithMetricsTargets() + if !reflect.DeepEqual(againTargets, updatedTargets) { + t.Fatal("detached map mutation escaped") + } + for _, r := range again { + if r.ID == id && (r.Tags[0] != "customer-data" || r.Canonical.Aliases[0] == "caller-change") { + t.Fatal("caller changed live registry") + } + } + empty := NewRegistry(nil) + rows, emptyTargets := empty.ListWithMetricsTargets() + if len(rows) != 0 || len(emptyTargets) != 0 { + t.Fatal("empty projection invented resources or targets") + } +}