Capture fresh broadcast resources and history targets together

Resolve canonical source mappings once per complete projection while preserving detached resource content and target selection after host coalescing. Bind resources and targets to the same locked generation, including absence and continuity overlays, and avoid another owned outer-slice copy. No timestamp cache or native CPU/RSS claim. Retain cold, clean and dirty full-content controls and paired complete-projection measurements.

Contract-Neutral: Internal bulk-read and copy-ownership repair preserves existing public resource/history fields, source precedence, agent identity/lifecycle, tenant/revocation and safety contracts; no wire or permission delta.
Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-05 11:22:07 +01:00
parent d7a305922a
commit 54de7bdca7
11 changed files with 623 additions and 26 deletions
@@ -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
@@ -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
@@ -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)
}
}
}
})
}
}
}
}
@@ -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",
+61 -22
View File
@@ -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 {
@@ -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
}
@@ -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 {
@@ -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")
}
}
+30 -1
View File
@@ -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
}
}
@@ -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
}
}
}
}
})
}
}
}
}
@@ -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")
}
}