Integrate reviewed same-generation broadcast resource and History capture

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-05 13:03:31 +01:00
11 changed files with 698 additions and 28 deletions
@@ -17,6 +17,36 @@
## 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.
Complete-content comparisons use one health-evaluation instant through a
private clock seam. Production still samples real UTC at the same post-read
boundary. The minute-boundary control requires age labels to advance without
renewing LastSeen; comparisons retain every health field instead of stripping
the reproduced `1m`/`2m` clock difference from JSON.
### 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,320 @@
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()
evaluationTime := time.Now().UTC()
before, err := json.Marshal(adapter.GetAll())
if err != nil {
t.Fatal(err)
}
want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, evaluationTime)
// The parent's unified view fills a missing store watermark from the
// listed resources. Preserve that full-state header in the oracle too.
if freshness := latestUnifiedResourceLastSeen(adapter.GetAll()); !freshness.IsZero() {
want.LastUpdate = freshness.UnixMilli()
}
got := m.buildBroadcastFrontendStateFromSnapshotWithClock(graph.State, func() time.Time { return evaluationTime })
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) {
first := 0
for first < len(wantJSON) && first < len(gotJSON) && wantJSON[first] == gotJSON[first] {
first++
}
start := max(0, first-80)
t.Fatalf("%s: complete frontend/catalogue/infrastructure JSON differs at byte %d: want %q, got %q (bytes %d/%d)",
stage, first, wantJSON[start:min(len(wantJSON), first+160)], gotJSON[start:min(len(gotJSON), first+160)], len(wantJSON), len(gotJSON))
}
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 TestBroadcastMetricsSnapshotHealthClockBoundary(t *testing.T) {
m, adapter, id := newReadStateCloneTestMonitor(t, 1)
rows := adapter.GetAll()
original := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
rows[0].LastSeen = original
rows[0].SourceStatus = map[unifiedresources.DataSource]unifiedresources.SourceStatus{
unifiedresources.SourceProxmox: {Status: "stale", LastSeen: original},
}
rr := unifiedresources.NewRegistry(nil)
rr.IngestResources(rows)
m.resourceStore = &broadcastProjectionCountingStore{MonitorAdapter: unifiedresources.NewMonitorAdapter(rr)}
for _, tc := range []struct {
age time.Duration
detail string
}{{59 * time.Second, "0m"}, {61 * time.Second, "1m"}} {
calls := 0
got := m.buildBroadcastFrontendStateFromSnapshotWithClock(models.EmptyStateSnapshot(), func() time.Time {
calls++
return original.Add(tc.age)
})
if calls != 1 || len(got.Resources) != 1 || got.Resources[0].ID != id || got.Resources[0].LastSeen != original.UnixMilli() {
t.Fatal("health clock changed capture identity/original time or was not sampled once")
}
var health unifiedresources.ResourceHealth
if err := json.Unmarshal(got.Resources[0].Health, &health); err != nil {
t.Fatal(err)
}
if health.Verdict != unifiedresources.HealthStale || len(health.Reasons) != 1 || health.Reasons[0].Code != "telemetry_stale" || health.Reasons[0].Detail != tc.detail {
t.Fatalf("health age must advance across the minute boundary: %+v", health)
}
}
}
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")
}
}
}
for _, r := range view.resources {
if r.Type != unifiedresources.ResourceTypeVM {
continue
}
baseline := unifiedresources.EvaluateResourceHealth(r, nil, now)
critical := unifiedresources.EvaluateResourceHealth(r, []unifiedresources.ResourceHealthAlert{{ResourceID: r.ID, Level: "critical", Type: "cpu"}}, now)
cleared := unifiedresources.EvaluateResourceHealth(r, nil, now)
if baseline.Verdict != unifiedresources.HealthOK || critical.Verdict != unifiedresources.HealthCritical || !reflect.DeepEqual(cleared, baseline) {
t.Fatal("live alert/clear 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)
healthAlerts := resourceHealthAlerts(snapshot.ActiveAlerts)
for i := range rows {
m.applyPersistedMetadataToUnifiedResource(&rows[i])
health := unifiedresources.EvaluateResourceHealth(rows[i], healthAlerts, 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)
if freshness := latestUnifiedResourceLastSeen(resources); !freshness.IsZero() {
out.LastUpdate = freshness.UnixMilli()
}
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",
@@ -2957,7 +2958,7 @@ func TestBroadcastProjectionMatchesPreviousPipeline(t *testing.T) {
if !view.freshness.IsZero() {
want.LastUpdate = view.freshness.UnixMilli()
}
got := m.buildBroadcastFrontendStateFromSnapshot(snapshot)
got := m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, func() time.Time { return now })
wantJSON, err := json.Marshal(want)
if err != nil {
t.Fatal(err)
+69 -23
View File
@@ -4333,6 +4333,13 @@ func buildFrontendStateFromSnapshot(snapshot models.StateSnapshot) models.StateF
}
func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateSnapshot) models.StateFrontend {
return m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, time.Now)
}
// The clock is sampled at the same post-read health-evaluation boundary as
// ordinary broadcasts. Tests can compare complete projections at one instant
// without mistaking a naturally advancing health-age label for lost content.
func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot models.StateSnapshot, healthClock func() time.Time) models.StateFrontend {
frontendState := buildFrontendStateFromSnapshot(snapshot)
m.updateResourceStoreForRead(snapshot)
if m != nil && m.alertManager != nil {
@@ -4341,20 +4348,22 @@ 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.
healthAlerts := resourceHealthAlerts(frontendState.ActiveAlerts)
now := time.Now().UTC()
now := healthClock().UTC()
for i := range broadcastResources {
m.applyPersistedMetadataToUnifiedResource(&broadcastResources[i])
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 +4880,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 +4931,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 +4953,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 +6518,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
}
@@ -6759,6 +6763,10 @@ func TestBroadcastProjectionListsRegistryOnceAndKeepsLiveChanges(t *testing.T) {
if row == nil || row.CPU.Current != 77 || len(row.Tags) != 1 || row.Tags[0] != "new-tag" {
t.Fatalf("mutable row was stale: %#v", row)
}
var health unifiedresources.ResourceHealth
if err := json.Unmarshal(row.Health, &health); err != nil || health.Verdict != unifiedresources.HealthCritical {
t.Fatalf("live critical alert did not reach the freshly projected row: %s (%v)", row.Health, err)
}
if first.Resources[0].CPU.Current != oldCPU {
t.Fatal("later projection mutated an accepted baseline")
}
@@ -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")
}
}