Derive canonical bulk-read metadata once per registry mutation

Keep full inventory and arbitrary clone refresh. Detach materialized identity and policy under the existing registry lock, invalidate between batch records and on stale-source transitions, and keep List-only reads independent of typed view construction. Compare every field and wire payload with an independent fresh-clone oracle across all demo providers.

Refs: #2199
Change-source: pulse-maintainer
Contract-Neutral: Registry-internal copying and invalidation repair existing freshness and ownership guarantees; canonical resource fields, wire schemas, identity rules, policy derivation, telemetry and alert evaluation are unchanged.
This commit is contained in:
pulse-triage[bot]
2026-10-05 02:57:02 +01:00
parent d79f7e545b
commit 51ef7eddda
7 changed files with 300 additions and 28 deletions
@@ -15,6 +15,25 @@
## Purpose
### Bulk resource reads do not repeatedly derive canonical metadata
Registry List and typed-view copies materialize identity/scopes/policy once
per observed mutation, then detach those blocks. Separate dirtiness lets List
avoid constructing an unused estate-sized typed view. Mutation-time invalidation
includes incremental batch records and stale-source transitions; no timestamp,
last-update or generation alone is used to declare a broadcast unchanged.
`cloneResource` retains arbitrary-input refresh semantics. Full inventory,
policy, observation time and health/alert evaluation remain unchanged.
The independent pre-change List oracle checks every resource field and wire
content, and paired one/one-thousand-resource controls expose repeated canonical
allocation work without wall-clock thresholds. Include dirty reads and typed
rebuilds when measuring: a cold generation must still derive metadata, and
moving that work into the registry is not proof that total ingest or broadcast
CPU improved. Native #2199 CPU/RSS, trailing refresh/duplicate ingest identity,
owner-set grouping, broadcast conversion and the sawtooth remain separate.
### Agent identity lookup cost follows identity-relevant inventory
Report admission no longer clones/ingests twenty explicitly listed Kubernetes
@@ -23,6 +23,33 @@ and sort the complete canonical change table while startup and ingestion wait.
## Purpose
### Bulk registry copies reuse mutation-bound canonical metadata
`ResourceRegistry.List` and typed-view rebuilding share canonical identity,
platform-scope and policy materialization under the registry write lock. Clean
bulk readers take the shared lock and detach the derived blocks rather than
re-derive them for every copied resource. A List-only consumer does not build
all typed views. Arbitrary `cloneResource` callers and point reads still refresh
metadata from their own current input; no caller-supplied metadata is trusted.
Record/mapping changes invalidate at the locked mutation, not just the batch
epilogue. Resource seeding, superseded-ID retention, Docker network relationship
refresh, PBS host association and final parent/rollup assembly invalidate too.
A real stale-source/status transition invalidates metadata and typed views; an
unchanged stale pass does not. A reader rechecks dirtiness after lock upgrade,
so an intervening mutation cannot leak a previous generation. Returned canonical
aliases, superseded IDs, policy redaction hints and platform scopes are detached
from the stored entries and retained views. This does not cache health, alerts,
broadcast conversion, observation times or telemetry, and prunes no inventory.
Verification: `registry_materialization_test.go` compares complete resources
and wire content against the original always-refresh List implementation,
including mid-batch changes, policy edits, supersession, staleness, ownership,
lazy views, races and allocation cost. The external connected demo test includes
all nine providers and its metadata-heavy estate. These are source invariants,
not native fleet CPU/RSS relief, backup recovery or release availability.
### Container CPU presentation explains its existing scale
The existing Docker/Podman container table and drawer visibly explain that CPU
+36 -1
View File
@@ -20,6 +20,42 @@ func cloneResource(in *Resource) Resource {
if in == nil {
return Resource{}
}
out := cloneResourceData(in)
RefreshCanonicalMetadata(&out)
return out
}
// cloneMaterializedResource is only for registry bulk reads after canonical
// metadata has been refreshed under the registry write lock. Arbitrary callers
// must use cloneResource, which still derives metadata from the current input.
// Copy the derived blocks too: a List consumer must not mutate registry policy,
// identity aliases or the detached typed view through its returned resource.
func cloneMaterializedResource(in *Resource) Resource {
if in == nil {
return Resource{}
}
out := cloneResourceData(in)
if in.Canonical != nil {
canonical := *in.Canonical
canonical.Aliases = cloneStringSlice(in.Canonical.Aliases)
canonical.SupersededIDs = cloneStringSlice(in.Canonical.SupersededIDs)
out.Canonical = &canonical
}
out.Policy = CloneResourcePolicy(in.Policy)
return out
}
func cloneMaterializedResourcePtr(in *Resource) *Resource {
if in == nil {
return nil
}
out := cloneMaterializedResource(in)
return &out
}
// cloneResourceData owns the same native payload detachments for refreshed and
// already-materialized clones. It does not decide whether metadata is current.
func cloneResourceData(in *Resource) Resource {
out := *in
out.DiscoveryTarget = cloneDiscoveryTarget(in.DiscoveryTarget)
@@ -54,7 +90,6 @@ func cloneResource(in *Resource) Resource {
out.Availability = cloneAvailabilityData(in.Availability)
out.AvailabilityChecks = cloneAvailabilityDataSlice(in.AvailabilityChecks)
out.FacetCounts = resourceFacetCounts(out)
RefreshCanonicalMetadata(&out)
return out
}
+1 -1
View File
@@ -88,7 +88,7 @@ func (rr *ResourceRegistry) associatePBSHostAgentResources(
pbsParentID,
)
}
rr.viewsDirty = true
rr.invalidateViewsLocked()
}
func uniquePBSHostAgent(instance models.PBSInstance, hosts []models.Host, vms []models.VM) *models.Host {
+70 -24
View File
@@ -108,6 +108,10 @@ type ResourceRegistry struct {
// Cached typed view indexes. Invalidated on ingest, rebuilt lazily on
// first access. Protected by mu — callers hold RLock to read, and the
// rebuild path upgrades to a write lock only when needed.
// Canonical metadata has its own dirty flag: List must not build every
// typed view just to refresh it. Both are invalidated at mutation time,
// including the intervals between records in a batch.
canonicalMetadataDirty bool
viewsDirty bool
cachedVMs []*VMView
cachedLXC []*ContainerView
@@ -148,24 +152,42 @@ type agentNodeCandidate struct {
resource *Resource
}
// invalidateSourceTargetsLocked drops the inverse source-target index the
// moment any bySource mapping mutates. Batch ingests release the registry
// lock between records and only mark viewsDirty in their epilogue, so
// without this a reader interleaving mid-batch would be served pre-batch
// targets from the cache; dropping it sends those readers to the legacy
// live scan, exactly the freshness the pre-index code had.
// invalidateSourceTargetsLocked drops the inverse source-target index and
// derived views at each mapping mutation. Batch ingests release the lock
// between records, so invalidating only at their epilogue would leave an
// interleaving bulk reader with pre-batch canonical metadata or typed views.
func (rr *ResourceRegistry) invalidateSourceTargetsLocked() {
rr.cachedSourceTargets = nil
rr.invalidateViewsLocked()
}
// Caller holds rr.mu for writing. No timestamp or caller-supplied derived
// metadata is evidence that a resource has survived a mutation unchanged.
func (rr *ResourceRegistry) invalidateViewsLocked() {
rr.viewsDirty = true
rr.canonicalMetadataDirty = true
}
func (rr *ResourceRegistry) ensureCanonicalMetadataLocked() {
if !rr.canonicalMetadataDirty {
return
}
for _, resource := range rr.resources {
RefreshCanonicalMetadata(resource)
}
rr.canonicalIdentityIndex = nil
rr.canonicalMetadataDirty = false
}
// NewRegistry creates a new registry using the provided store for overrides.
func NewRegistry(store ResourceStore) *ResourceRegistry {
rr := &ResourceRegistry{
resources: make(map[string]*Resource),
bySource: make(map[DataSource]map[string]string),
matcher: NewIdentityMatcher(),
store: store,
exclusions: make(map[string]struct{}),
resources: make(map[string]*Resource),
bySource: make(map[DataSource]map[string]string),
matcher: NewIdentityMatcher(),
store: store,
exclusions: make(map[string]struct{}),
canonicalMetadataDirty: true,
}
rr.bySource[SourceProxmox] = make(map[string]string)
@@ -577,7 +599,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
rr.buildChildCounts()
rr.refreshCanonicalIdentitiesLocked()
rr.markStaleLocked(time.Now().UTC(), thresholds)
rr.viewsDirty = true
rr.invalidateViewsLocked()
rr.mu.Unlock()
}
@@ -627,7 +649,7 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec
rr.refreshIncidentRollupsLocked()
rr.buildChildCounts()
rr.refreshCanonicalIdentitiesLocked()
rr.viewsDirty = true
rr.invalidateViewsLocked()
rr.mu.Unlock()
}
@@ -658,6 +680,7 @@ func (rr *ResourceRegistry) retainSupersededCanonicalIDs(resourceID string, supe
ids = append(ids, supersededID)
}
resource.SupersededCanonicalIDs = uniqueTrimmed(ids...)
rr.invalidateViewsLocked()
rr.indexSupersededCanonicalIDsLocked(resourceID, resource.SupersededCanonicalIDs)
}
@@ -791,7 +814,7 @@ func (rr *ResourceRegistry) ingestResources(resources []Resource, thresholds map
rr.canonicalIdentityIndex = nil
rr.resources[resource.ID] = resource
rr.matcher.Add(resource.ID, resource.Identity)
rr.viewsDirty = true
rr.invalidateViewsLocked()
rr.mu.Unlock()
seededIDs = append(seededIDs, resource.ID)
@@ -812,7 +835,7 @@ func (rr *ResourceRegistry) ingestResources(resources []Resource, thresholds map
rr.buildChildCounts()
rr.refreshCanonicalIdentitiesLocked()
rr.markStaleLocked(time.Now().UTC(), thresholds)
rr.viewsDirty = true
rr.invalidateViewsLocked()
rr.mu.Unlock()
}
@@ -1293,16 +1316,30 @@ func seededVMwareSourceID(resource *Resource) string {
return strings.Join(parts, ":")
}
// List returns all resources.
// List returns detached resources with current canonical metadata. A dirty
// generation derives metadata once under the write lock; clean readers share
// 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 {
rr.mu.RLock()
defer rr.mu.RUnlock()
out := make([]Resource, 0, len(rr.resources))
for _, r := range rr.resources {
out = append(out, cloneResource(r))
for {
rr.mu.RLock()
if rr.canonicalMetadataDirty {
rr.mu.RUnlock()
rr.mu.Lock()
rr.ensureCanonicalMetadataLocked()
rr.mu.Unlock()
// A mutation can occur after the lock is released. Check again
// rather than copy a generation that is no longer materialized.
continue
}
out := make([]Resource, 0, len(rr.resources))
for _, r := range rr.resources {
out = append(out, cloneMaterializedResource(r))
}
rr.mu.RUnlock()
sortResourcesByName(out)
return out
}
sortResourcesByName(out)
return out
}
// ListForPresentation returns resources in the canonical API/broadcast
@@ -1693,7 +1730,9 @@ func sourceSightingStatus(lastSeen time.Time) string {
func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSource]time.Duration) {
thresholds = effectiveStaleThresholds(thresholds)
changed := false
for _, resource := range rr.resources {
previousStatus := resource.Status
staleFound := false
for source, status := range resource.SourceStatus {
threshold, ok := thresholds[source]
@@ -1709,6 +1748,7 @@ func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSo
continue
}
if now.Sub(status.LastSeen) > threshold {
changed = changed || status.Status != "stale"
status.Status = "stale"
resource.SourceStatus[source] = status
staleFound = true
@@ -1722,6 +1762,10 @@ func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSo
resource.Status = StatusWarning
}
}
changed = changed || resource.Status != previousStatus
}
if changed {
rr.invalidateViewsLocked()
}
}
@@ -2171,6 +2215,7 @@ func (rr *ResourceRegistry) refreshDockerNetworkAttachmentRelationships(host mod
rr.mu.Lock()
defer rr.mu.Unlock()
rr.invalidateViewsLocked()
dockerSource := rr.bySource[SourceDocker]
networkIDByName := make(map[string]string, len(host.Networks))
@@ -5803,6 +5848,7 @@ func readCleanViewCache[T any](rr *ResourceRegistry, fn func() T) (T, bool) {
// rebuildViews recomputes all cached view slices from the current resource map.
// Caller must hold rr.mu for writing.
func (rr *ResourceRegistry) rebuildViews() {
rr.ensureCanonicalMetadataLocked()
rr.cachedVMs = nil
rr.cachedLXC = nil
rr.cachedNodes = nil
@@ -5827,7 +5873,7 @@ func (rr *ResourceRegistry) rebuildViews() {
rr.cachedSourceTargets = sourceTargetsIndex
for _, r := range rr.resources {
viewResource := cloneResourcePtr(r)
viewResource := cloneMaterializedResourcePtr(r)
viewResource.MetricsTarget = rr.metricsTargetFromSourceTargets(r, sourceTargetsIndex[r.ID])
switch r.Type {
case ResourceTypeVM:
@@ -0,0 +1,91 @@
package unifiedresources_test
import (
"encoding/json"
"reflect"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/mock"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
// Use the real multi-provider demo graph, including its metadata-heavy
// Kubernetes estate, rather than accepting a VM-only copying oracle.
func TestRegistryMaterializedMetadataConnectedDemo(t *testing.T) {
previousEnabled := mock.IsMockEnabled()
previousConfig := mock.GetConfig()
t.Cleanup(func() {
_ = mock.SetEnabled(false)
mock.SetMockConfig(previousConfig)
if previousEnabled {
if err := mock.SetEnabled(true); err != nil {
t.Error(err)
}
}
})
if err := mock.SetEnabled(false); err != nil {
t.Fatal(err)
}
cfg := mock.DefaultConfig
cfg.UpdateInterval = 24 * time.Hour
mock.SetMockConfig(cfg)
if err := mock.SetEnabled(true); err != nil {
t.Fatal(err)
}
graph := mock.CurrentFixtureGraph()
registry := unifiedresources.NewRegistry(nil)
registry.IngestSnapshot(graph.State)
for _, source := range []unifiedresources.DataSource{unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} {
registry.IngestRecords(source, graph.SupplementalRecords(source))
}
assertEqual := func(stage string) {
t.Helper()
actual := registry.List()
expected := unifiedresources.FreshRegistryMetadataListForTest(registry)
if len(actual) < 1000 {
t.Fatalf("%s: fixture is not a large estate (%d resources)", stage, len(actual))
}
if !reflect.DeepEqual(actual, expected) {
t.Fatalf("%s: complete demo resources differ from fresh clone", stage)
}
a, err := json.Marshal(actual)
if err != nil {
t.Fatal(err)
}
e, err := json.Marshal(expected)
if err != nil {
t.Fatal(err)
}
if string(a) != string(e) {
t.Fatalf("%s: demo wire content differs", stage)
}
sources := map[unifiedresources.DataSource]bool{}
for _, r := range actual {
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("%s: fixture omits provider %s", stage, source)
}
}
t.Logf("%s: %d complete resources equal the independent fresh-clone oracle", stage, len(actual))
}
assertEqual("initial")
registry.Workloads()
registry.IngestSnapshot(graph.State)
assertEqual("snapshot-refresh")
registry.MarkStale(time.Now().UTC().Add(7*24*time.Hour), nil)
assertEqual("stale-transition")
// A normalized re-seed must keep identity, supersession, policy and every
// native facet, rather than assuming old derived fields are trustworthy.
resources := registry.List()
resources[0].Name = "changed-name"
resources[0].Tags = []string{"customer-data"}
resources[0].PlatformScopes = []string{"old"}
resources[0].AISafeSummary = "old"
registry.IngestResources(resources)
assertEqual("resource-reseed")
}
@@ -63,8 +63,8 @@ func metadataOracleFixture() []Resource {
{ID: "k8s:node", Type: ResourceTypeK8sNode, Name: "node", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}},
{ID: "k8s:pod", Type: ResourceTypePod, Name: "pod", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}},
{ID: "k8s:deploy", Type: ResourceTypeK8sDeployment, Name: "deploy", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}},
{ID: "k8s:secret", Type: ResourceTypeK8sSecret, Name: "synthetic-secret-metadata", LastSeen: now, Sources: []DataSource{SourceK8s}},
{ID: "k8s:role", Type: ResourceTypeK8sRole, Name: "role", LastSeen: now, Sources: []DataSource{SourceK8s}},
{ID: "k8s:secret", Type: ResourceTypeK8sSecret, Name: "synthetic-secret-metadata", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab", SecretUID: "secret-lab"}},
{ID: "k8s:role", Type: ResourceTypeK8sRole, Name: "role", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab", RoleUID: "role-lab"}},
{ID: "tn:lab", Type: ResourceTypeAgent, Name: "nas", LastSeen: now, Sources: []DataSource{SourceTrueNAS}, TrueNAS: &TrueNASData{Hostname: "nas.example.test"}},
{ID: "vmware:lab", Type: ResourceTypeAgent, Name: "esxi", LastSeen: now, Sources: []DataSource{SourceVMware}, VMware: &VMwareData{ConnectionID: "vsphere-lab", ManagedObjectID: "host-1"}},
{ID: "disk:lab", Type: ResourceTypePhysicalDisk, Name: "disk", LastSeen: now, Sources: []DataSource{SourceAgent}, PhysicalDisk: &PhysicalDiskMeta{}},
@@ -190,3 +190,57 @@ func BenchmarkRegistryMetadataBulkReads(b *testing.B) {
}
}
}
// FreshRegistryMetadataListForTest exposes the independent clone oracle only
// to external-package connected fixture tests, never production consumers.
func FreshRegistryMetadataListForTest(rr *ResourceRegistry) []Resource {
return freshRegistryListForMetadataTest(rr)
}
func TestRegistryMaterializedMetadataViewsKeepFreshness(t *testing.T) {
rr := NewRegistry(nil)
rr.IngestSnapshot(benchmarkVMState(1))
before := rr.VMs()[0]
if before.Status() != StatusOnline {
t.Fatalf("fresh VM status=%s", before.Status())
}
rr.MarkStale(time.Now().UTC().Add(24*time.Hour), nil)
after := rr.VMs()[0]
if after.r.SourceStatus[SourceProxmox].Status != "stale" || after.Status() == StatusOnline {
t.Fatal("stale marking left the typed view online")
}
if before.Status() != StatusOnline {
t.Fatal("stale marking mutated a retained detached view")
}
// A no-change stale pass must not throw away the already fresh view.
rr.MarkStale(time.Now().UTC().Add(24*time.Hour), nil)
if rr.VMs()[0] != after {
t.Fatal("unchanged staleness rebuilt the typed view")
}
rr.ingestRecord(SourceProxmox, "lab:node-1:100", Resource{Type: ResourceTypeVM, Name: "renamed-vm", Status: StatusOnline, LastSeen: time.Now().UTC(), Proxmox: &ProxmoxData{VMID: 100, NodeName: "node-1"}}, ResourceIdentity{}, false)
// No batch epilogue has run here; source mapping and typed view must already
// reflect the resource visible to List and point reads.
current := rr.VMs()[0]
if current.Status() != StatusOnline || current.Name() != "renamed-vm" {
t.Fatal("mid-batch record retained the previous typed view")
}
assertRegistryMetadataOracle(t, rr)
}
func TestRegistryMaterializedMetadataListKeepsViewsLazy(t *testing.T) {
rr := NewRegistry(nil)
rr.IngestSnapshot(benchmarkVMState(1000))
rr.List()
if rr.cachedVMs != nil || !rr.viewsDirty {
t.Fatal("a List-only read built unnecessary typed copies")
}
rr.VMs()
if rr.canonicalMetadataDirty || rr.viewsDirty {
t.Fatal("views were not materialized from current canonical metadata")
}
before := rr.cachedVMs[0]
rr.List()
if rr.cachedVMs[0] != before {
t.Fatal("clean bulk read discarded detached typed views")
}
}