From 187e5daaa86408708316b39f958ea00a87e4d601 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 01:17:08 +0100 Subject: [PATCH] Keep agent identity lookups scoped to relevant inventory Avoid cloning and ingesting twenty known non-host Kubernetes metadata kinds in temporary per-report matching registries, while preserving all publication and canonical matching. Clone only the selected native record for exact host reads and detach mutable host update, module and config state. Pin reference equivalence, live report publication, ownership and allocation scaling with connected controls. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 19 +++ .../v6/internal/subsystems/monitoring.md | 25 +++ .../subsystems/performance-and-scalability.md | 18 ++ internal/models/agent_lookup.go | 44 +++++ internal/models/agent_lookup_test.go | 156 ++++++++++++++++++ internal/models/deepcopy.go | 3 + internal/models/models.go | 28 ++++ internal/models/state_snapshot.go | 8 +- internal/monitoring/agent_lookup_test.go | 64 ++++++- internal/monitoring/monitor_agents.go | 50 ++---- .../monitoring/monitor_host_agents_test.go | 4 +- 11 files changed, 374 insertions(+), 45 deletions(-) create mode 100644 internal/models/agent_lookup.go create mode 100644 internal/models/agent_lookup_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 63d996419a..469e7d2737 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -15,6 +15,25 @@ ## Purpose +### Source-record lookup preserves report admission + +Host and Docker report admission still performs canonical host-view matching +before and after taking its per-source lock, with the same identity, source +freshness, token binding, removal/re-enrolment and stream ordering policy. +Only its private temporary snapshot omits twenty enumerated non-host Kubernetes +metadata collections; normal inventory and accepted-report publication stay +complete. Exact native host reads deep-copy the selected record under State's +read lock, rather than copying every peer. Returned agent modules, update +status/timestamps and applied config are detached for both single and list +reads. No cached identity, new alias, heartbeat or timeout rule is introduced. + +`internal/models/agent_lookup_test.go` pins exact-ID/missing/ownership semantics +and concurrent source replacement. `internal/monitoring/agent_lookup_test.go` +pins whole-registry host-view equality, live replacement/removal, both serialized +report/publication paths and canonical Docker-ID resolution. Existing identity, +removal, credential and report-order suites remain applicable. + + ### Memory annotation is not agent admission The shared Memory model's optional guest `observation` records selected metric diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 743401a8ff..2f746480d2 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,31 @@ ## Purpose +### Report identity lookups do not rebuild unrelated Kubernetes metadata + +The producer's two temporary host/Docker-host admission registries use a +source-locked detached lookup snapshot. It omits only twenty explicitly named +Kubernetes metadata collections before cloning; clusters, nodes, pods and +controller workloads, every non-Kubernetes source and all other snapshot +fields remain. The ordinary registry resolves identity and freshness on each +lookup, including the post-per-host-lock read; there is no generation/time +cache, ID-hint shortcut or changed report watermark/credential policy. + +These temporary registries are never publication or alert inventories. +Accepted-report refresh, normal GetSnapshot/GetState, canonical reads and +broadcasts still ingest every collection. Exact source-record point reads +clone only the selected native host, after the existing canonical Docker ID +resolution. Missing native records allocate nothing. Host clone ownership also +includes mutable agent modules, update status/times and config fingerprints. + +`agent_lookup_test.go` compares every field of both canonical host views with +the previous whole-snapshot registry, exercises current replacement/removal, +and carries serialized host/Docker reports through complete native and +canonical publication. Allocation controls pin independence from unrelated +metadata/peer-host payloads; paired unchanged-reference benchmarks measure +this lookup, not fleet CPU/RSS relief or all of #2199's remaining mechanisms. + + ### Guest memory observations are source-owned QEMU and LXC memory carry an optional `observation` with `state` (`current`, diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 373999135f..e7775e89f8 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -15,6 +15,24 @@ ## Purpose +### Agent identity lookup cost follows identity-relevant inventory + +Report admission no longer clones/ingests twenty explicitly listed Kubernetes +metadata collections that cannot supply a host facet. Clusters, nodes and +workloads, every other source, canonical matching and fresh post-lock reads +remain. No complete inventory, graph, alert or publication is pruned. Native +point lookups retain an ID scan but clone just one selected record; peer sensor +and container payloads no longer multiply the allocation cost of that read. + +The allocation controls compare zero/one-thousand metadata objects and one/ +seventeen hosts without wall-clock thresholds. The paired benchmark retains +an executable copy of the old full-snapshot lookup in the same process, with +identical fixtures and host-view reads, so lookup time/bytes/allocations can be +compared without selecting a favourable rerun. This is a bounded source cost +improvement, not attribution of the reporter's native CPU, RSS, sawtooth or +remaining refresh/grouping/broadcast mechanisms. + + ### Memory presentation does not collect or renew observations Guest-memory provenance is projected from the already-owned snapshot in the diff --git a/internal/models/agent_lookup.go b/internal/models/agent_lookup.go new file mode 100644 index 0000000000..1cfaee2260 --- /dev/null +++ b/internal/models/agent_lookup.go @@ -0,0 +1,44 @@ +package models + +// GetAgentLookupSnapshot is only for the producer's temporary host-identity +// lookup registry, never for publication, alerting or a complete inventory. +// Keep the ordinary snapshot's lock, ownership and all other source fields; +// omit only these known Kubernetes metadata kinds before deep copying them. +// They cannot merge into a host. Nodes and workloads still contribute the +// same identity, source selection, relationships and freshness evidence. +func (s *State) GetAgentLookupSnapshot() StateSnapshot { + return s.getSnapshot(cloneKubernetesClustersForAgentLookup) +} + +func cloneKubernetesClustersForAgentLookup(src []KubernetesCluster) []KubernetesCluster { + if len(src) == 0 { + return nil + } + dest := make([]KubernetesCluster, len(src)) + for i, cluster := range src { + // The struct is a value copy. Do not normalize or mutate its remaining + // slice members until cloneKubernetesCluster has detached them. + cluster.Namespaces = nil + cluster.Services = nil + cluster.Ingresses = nil + cluster.EndpointSlices = nil + cluster.NetworkPolicies = nil + cluster.PersistentVolumes = nil + cluster.PersistentVolumeClaims = nil + cluster.StorageClasses = nil + cluster.ConfigMaps = nil + cluster.Secrets = nil + cluster.ServiceAccounts = nil + cluster.Roles = nil + cluster.ClusterRoles = nil + cluster.RoleBindings = nil + cluster.ClusterRoleBindings = nil + cluster.ResourceQuotas = nil + cluster.LimitRanges = nil + cluster.PodDisruptionBudgets = nil + cluster.HorizontalPodAutoscalers = nil + cluster.Events = nil + dest[i] = cloneKubernetesCluster(cluster) + } + return dest +} diff --git a/internal/models/agent_lookup_test.go b/internal/models/agent_lookup_test.go new file mode 100644 index 0000000000..e52c132272 --- /dev/null +++ b/internal/models/agent_lookup_test.go @@ -0,0 +1,156 @@ +package models + +import ( + "fmt" + "reflect" + "sync" + "testing" + "time" +) + +var agentLookupOmittedCollections = []string{ + "Namespaces", "Services", "Ingresses", "EndpointSlices", "NetworkPolicies", + "PersistentVolumes", "PersistentVolumeClaims", "StorageClasses", "ConfigMaps", + "Secrets", "ServiceAccounts", "Roles", "ClusterRoles", "RoleBindings", + "ClusterRoleBindings", "ResourceQuotas", "LimitRanges", "PodDisruptionBudgets", + "HorizontalPodAutoscalers", "Events", +} + +func TestAgentLookupSnapshotPreservesEveryOtherField(t *testing.T) { + cluster := KubernetesCluster{ID: "cluster", AgentID: "agent", Name: "cluster", TokenID: "token", LastSeen: time.Now().UTC()} + // Populate every collection, not just the fields currently omitted. New + // fields must stay on the ordinary clone path unless explicitly reviewed. + value := reflect.ValueOf(&cluster).Elem() + for i := 0; i < value.NumField(); i++ { + field := value.Field(i) + if field.Kind() == reflect.Slice { + field.Set(reflect.MakeSlice(field.Type(), 1, 1)) + if uid := field.Index(0).FieldByName("UID"); uid.IsValid() { + uid.SetString(value.Type().Field(i).Name) + } + } + } + s := NewState() + s.UpsertKubernetesCluster(cluster) + s.UpsertHost(Host{ID: "host", Tags: []string{"tag"}, LastSeen: cluster.LastSeen}) + s.UpsertDockerHost(DockerHost{ID: "docker", Containers: []DockerContainer{{ID: "app", Labels: map[string]string{"app": "web"}}}}) + s.PVEBackups = PVEBackups{GuestSnapshots: []GuestSnapshot{{VMID: 100}}} + s.ConnectionHealth["cluster"] = true + before := s.GetSnapshot() + want := before.Clone() + wantCluster := reflect.ValueOf(&want.KubernetesClusters[0]).Elem() + for _, name := range agentLookupOmittedCollections { + field := wantCluster.FieldByName(name) + field.Set(reflect.MakeSlice(field.Type(), 0, 0)) + } + got := s.GetAgentLookupSnapshot() + if !reflect.DeepEqual(got, want) { + t.Fatal("agent lookup projection changed more than the explicitly omitted metadata collections") + } + got.KubernetesClusters[0].Nodes[0].Roles = append(got.KubernetesClusters[0].Nodes[0].Roles, "changed") + got.Hosts[0].Tags[0] = "changed" + got.DockerHosts[0].Containers[0].Labels["app"] = "changed" + got.ConnectionHealth["cluster"] = false + if !reflect.DeepEqual(before, s.GetSnapshot()) { + t.Fatal("lookup projection aliases the full source inventory") + } +} + +func TestStatePointLookupsMatchWholeLists(t *testing.T) { + now := time.Now().UTC() + s := NewState() + s.UpsertHost(Host{ + ID: "host", Tags: []string{"tag"}, NetworkInterfaces: []HostNetworkInterface{{Addresses: []string{"192.0.2.1"}}}, + Sensors: HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu": 40}}, + AppliedConfig: &AgentConfigFingerprint{Version: "1", Hash: "hash"}, + AgentModules: []AgentModuleStatus{{Name: "host", State: "running"}}, + AgentUpdate: &AgentUpdateStatus{State: "idle", LastCheckedAt: &now, LastAttemptAt: &now, LastSuccessAt: &now}, + }) + s.UpsertDockerHost(DockerHost{ + ID: "docker", AgentID: "collector", Containers: []DockerContainer{{ID: "app", Labels: map[string]string{"app": "web"}}}, + NetworkInterfaces: []HostNetworkInterface{{Addresses: []string{"192.0.2.1"}}}, + AgentModules: []AgentModuleStatus{{Name: "docker", State: "running"}}, + }) + for _, id := range []string{"", "missing", " host ", "host", "docker", "collector"} { + wantHost, wantDocker := Host{}, DockerHost{} + hostFound, dockerFound := false, false + for _, host := range s.GetHosts() { + if host.ID == id { + wantHost, hostFound = host, true + break + } + } + for _, host := range s.GetDockerHosts() { + if host.ID == id { + wantDocker, dockerFound = host, true + break + } + } + gotHost, gotHostFound := s.GetHost(id) + gotDocker, gotDockerFound := s.GetDockerHost(id) + if gotHostFound != hostFound || gotDockerFound != dockerFound || !reflect.DeepEqual(gotHost, wantHost) || !reflect.DeepEqual(gotDocker, wantDocker) { + t.Fatalf("exact source lookup %q differs from the existing whole-list lookup", id) + } + } + before := s.GetSnapshot() + for _, fromList := range []bool{false, true} { + host, _ := s.GetHost("host") + docker, _ := s.GetDockerHost("docker") + if fromList { + host, docker = s.GetHosts()[0], s.GetDockerHosts()[0] + } + host.Tags[0] = "changed" + host.NetworkInterfaces[0].Addresses[0] = "changed" + host.Sensors.TemperatureCelsius["cpu"] = 99 + host.AppliedConfig.Hash = "changed" + host.AgentModules[0].State = "changed" + host.AgentUpdate.State = "changed" + *host.AgentUpdate.LastCheckedAt = time.Time{} + *host.AgentUpdate.LastAttemptAt = time.Time{} + *host.AgentUpdate.LastSuccessAt = time.Time{} + docker.Containers[0].Labels["app"] = "changed" + docker.NetworkInterfaces[0].Addresses[0] = "changed" + docker.AgentModules[0].State = "changed" + if !reflect.DeepEqual(before, s.GetSnapshot()) { + t.Fatal("point or list result aliases mutable source fields") + } + } +} + +func TestStatePointLookupAllocationsAndConcurrentOwnership(t *testing.T) { + s := NewState() + for i := 0; i < 17; i++ { + id := fmt.Sprint(i) + s.UpsertHost(Host{ID: id, Tags: []string{"tag"}}) + s.UpsertDockerHost(DockerHost{ID: id, Containers: []DockerContainer{{ID: "app", Labels: map[string]string{"key": "value"}}}}) + } + for _, lookup := range []func(string){ + func(id string) { _, _ = s.GetHost(id) }, + func(id string) { _, _ = s.GetDockerHost(id) }, + } { + if allocations := testing.AllocsPerRun(10, func() { lookup("missing") }); allocations != 0 { + t.Fatalf("missing exact record allocated %.0f times", allocations) + } + } + var workers sync.WaitGroup + for i := 0; i < 3; i++ { + workers.Add(1) + go func(i int) { + defer workers.Done() + for j := 0; j < 20; j++ { + id := fmt.Sprint(i) + if i == 0 { + s.UpsertHost(Host{ID: id, Tags: []string{"tag"}}) + s.UpsertDockerHost(DockerHost{ID: id, Containers: []DockerContainer{{ID: "app", Labels: map[string]string{"key": "value"}}}}) + } else { + host, _ := s.GetHost("0") + docker, _ := s.GetDockerHost("0") + host.Tags[0] = "caller-owned" + docker.Containers[0].Labels["key"] = "caller-owned" + _ = s.GetAgentLookupSnapshot() + } + } + }(i) + } + workers.Wait() +} diff --git a/internal/models/deepcopy.go b/internal/models/deepcopy.go index 41f70ee637..15d487f630 100644 --- a/internal/models/deepcopy.go +++ b/internal/models/deepcopy.go @@ -413,6 +413,9 @@ func cloneHost(src Host) Host { dest.DiskExclude = append([]string(nil), src.DiskExclude...) dest.IdentityConflict = cloneHostIdentityConflict(src.IdentityConflict) dest.AgentPrivilege = cloneAgentPrivilegeStatus(src.AgentPrivilege) + dest.AgentModules = cloneAgentModuleStatuses(src.AgentModules) + dest.AgentUpdate = cloneAgentUpdateStatus(src.AgentUpdate) + dest.AppliedConfig = cloneAgentConfigFingerprint(src.AppliedConfig) return dest.NormalizeCollections() } diff --git a/internal/models/models.go b/internal/models/models.go index aff44ce977..5f1f2af5b8 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -5087,6 +5087,20 @@ func (s *State) RemoveStaleDockerHosts(cutoff time.Time) []DockerHost { return cloneDockerHosts(removed) } +// GetDockerHost returns one detached source record. Identity resolution belongs +// to the caller; an exact source ID lookup must not clone every peer inventory. +func (s *State) GetDockerHost(hostID string) (DockerHost, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + + for _, host := range s.DockerHosts { + if host.ID == hostID { + return cloneDockerHost(host), true + } + } + return DockerHost{}, false +} + // GetDockerHosts returns a copy of docker hosts. func (s *State) GetDockerHosts() []DockerHost { s.mu.RLock() @@ -5400,6 +5414,20 @@ func (s *State) UpsertHost(host Host) { s.LastUpdate = time.Now() } +// GetHost returns one detached source record under the same lock and clone +// policy as GetHosts, without copying unrelated sensors or workload inventory. +func (s *State) GetHost(hostID string) (Host, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + + for _, host := range s.Hosts { + if host.ID == hostID { + return cloneHost(host), true + } + } + return Host{}, false +} + // GetHosts returns a copy of all generic hosts. func (s *State) GetHosts() []Host { s.mu.RLock() diff --git a/internal/models/state_snapshot.go b/internal/models/state_snapshot.go index ec71aaa53b..7c7c26ad70 100644 --- a/internal/models/state_snapshot.go +++ b/internal/models/state_snapshot.go @@ -205,8 +205,12 @@ func (s StateSnapshot) Clone() StateSnapshot { return snapshot } -// GetSnapshot returns a snapshot of the current state without mutex +// GetSnapshot returns a detached snapshot of the complete current state. func (s *State) GetSnapshot() StateSnapshot { + return s.getSnapshot(cloneKubernetesClusters) +} + +func (s *State) getSnapshot(cloneClusters func([]KubernetesCluster) []KubernetesCluster) StateSnapshot { s.mu.RLock() defer s.mu.RUnlock() @@ -221,7 +225,7 @@ func (s *State) GetSnapshot() StateSnapshot { Containers: cloneContainers(s.Containers), DockerHosts: cloneDockerHosts(s.DockerHosts), RemovedDockerHosts: append([]RemovedDockerHost(nil), s.RemovedDockerHosts...), - KubernetesClusters: cloneKubernetesClusters(s.KubernetesClusters), + KubernetesClusters: cloneClusters(s.KubernetesClusters), RemovedKubernetesClusters: append([]RemovedKubernetesCluster(nil), s.RemovedKubernetesClusters...), Hosts: cloneHosts(s.Hosts), RemovedHostAgents: append([]RemovedHostAgent(nil), s.RemovedHostAgents...), diff --git a/internal/monitoring/agent_lookup_test.go b/internal/monitoring/agent_lookup_test.go index 47880eaea6..de5e2d2113 100644 --- a/internal/monitoring/agent_lookup_test.go +++ b/internal/monitoring/agent_lookup_test.go @@ -1,6 +1,7 @@ package monitoring import ( + "encoding/json" "fmt" "reflect" "testing" @@ -9,6 +10,8 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" ) // These collections cannot contribute a host facet. Workloads, Kubernetes @@ -73,7 +76,7 @@ func TestAgentLookupMatchesWholeSnapshotHostViews(t *testing.T) { m := agentLookupFixture(count) before := m.state.GetSnapshot() want := fullAgentLookupReference(m) - got := m.snapshotBackedUnifiedReadState() + got := m.snapshotBackedAgentLookupReadState() if len(want.Hosts()) == 0 || len(want.DockerHosts()) == 0 { t.Fatal("fixture must exercise both host facets") } @@ -88,7 +91,7 @@ func TestAgentLookupMatchesWholeSnapshotHostViews(t *testing.T) { m.state.UpsertHost(models.Host{ID: "host", Hostname: "renamed", MachineID: "machine", Status: "online", LastSeen: time.Now().UTC()}) m.state.RemoveDockerHost("docker-host") want = fullAgentLookupReference(m) - got = m.snapshotBackedUnifiedReadState() + got = m.snapshotBackedAgentLookupReadState() if !reflect.DeepEqual(got.Hosts(), want.Hosts()) || !reflect.DeepEqual(got.DockerHosts(), want.DockerHosts()) { t.Fatal("agent lookup retained a removed host or old metadata") } @@ -100,7 +103,7 @@ func TestAgentLookupAllocationIgnoresUnrelatedKubernetesMetadata(t *testing.T) { measure := func(count int) float64 { m := agentLookupFixture(count) return testing.AllocsPerRun(3, func() { - read := m.snapshotBackedUnifiedReadState() + read := m.snapshotBackedAgentLookupReadState() if len(read.Hosts()) == 0 || len(read.DockerHosts()) == 0 { panic("missing fixture host") } @@ -132,6 +135,59 @@ func TestAgentPointLookupAllocationIgnoresOtherHosts(t *testing.T) { } } +func TestAgentLookupReportsKeepCompleteKubernetesInventory(t *testing.T) { + m := newTestMonitor(t) + m.state = agentLookupFixture(1000).state + before := m.state.GetSnapshot().KubernetesClusters + registry := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + m.SetResourceStore(registry) + + // Both serialized report paths use the new admission lookup, but their + // normal accepted-report publication must retain the complete inventory. + hostReport := agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "host", Version: "test", IntervalSeconds: 30}, + Host: agentshost.HostInfo{ID: "host", MachineID: "machine", Hostname: "worker", Platform: "linux"}, + Timestamp: time.Now().UTC(), + } + payload, err := json.Marshal(hostReport) + if err != nil || json.Unmarshal(payload, &hostReport) != nil { + t.Fatal("host report round trip failed") + } + if host, err := m.ApplyHostReport(hostReport, nil); err != nil || host.ID != "host" { + t.Fatalf("host report identity changed: %q, %v", host.ID, err) + } + dockerReport := agentsdocker.Report{ + Agent: agentsdocker.AgentInfo{ID: "host", Version: "test", IntervalSeconds: 30}, + Host: agentsdocker.HostInfo{MachineID: "machine", Hostname: "worker", Runtime: "podman", TotalCPU: 4}, + Containers: []agentsdocker.Container{{ID: "app", Name: "app", State: "running", CPUPercent: 9.4}}, + Timestamp: time.Now().UTC(), + } + payload, err = json.Marshal(dockerReport) + if err != nil || json.Unmarshal(payload, &dockerReport) != nil { + t.Fatal("Docker report round trip failed") + } + if host, err := m.ApplyDockerReport(dockerReport, nil); err != nil || host.ID != "docker-host" { + t.Fatalf("Docker report identity changed: %q, %v", host.ID, err) + } + if !reflect.DeepEqual(before, m.GetState().KubernetesClusters) { + t.Fatal("report lookup discarded or mutated published Kubernetes metadata") + } + // Compare all canonical resources as well as the native source snapshot. + want := unifiedresources.NewRegistry(nil) + want.IngestSnapshotWithStaleThresholds(m.GetState(), m.resourceStaleThresholds()) + if got, expected := registry.GetAll(), want.List(); !reflect.DeepEqual(got, expected) { + t.Fatal("report publication no longer contains the full canonical inventory") + } + for _, host := range m.GetUnifiedReadState().DockerHosts() { + if host == nil { + continue + } + if got, found := m.GetDockerHost(host.ID()); !found || got.ID != "docker-host" { + t.Fatal("point lookup bypassed canonical Docker host identity resolution") + } + } +} + func BenchmarkAgentIdentityLookup(b *testing.B) { for _, count := range []int{0, 1000} { m := agentLookupFixture(count) @@ -147,7 +203,7 @@ func BenchmarkAgentIdentityLookup(b *testing.B) { if reference { read = fullAgentLookupReference(m) } else { - read = m.snapshotBackedUnifiedReadState() + read = m.snapshotBackedAgentLookupReadState() } if len(read.Hosts()) == 0 || len(read.DockerHosts()) == 0 { b.Fatal("missing fixture host") diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 2a1d74ee88..4ab39a48ea 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -517,14 +517,7 @@ func (m *Monitor) removeHostAgentLocked(hostID, requiredTokenID string, requireD continuity, hasContinuity = m.hostContinuityStore.Get(hostID) } - host, present := models.Host{}, false - for _, candidate := range m.state.GetHosts() { - if candidate.ID == hostID { - host = candidate - present = true - break - } - } + host, present := m.hostByID(hostID) if !present { if logging.IsLevelEnabled(zerolog.DebugLevel) { log.Debug().Str("hostID", hostID).Msg("host not present in state during removal") @@ -1380,13 +1373,7 @@ func (m *Monitor) GetDockerHost(hostID string) (models.DockerHost, bool) { } hostID = m.canonicalDockerHostID(hostID) - hosts := m.state.GetDockerHosts() - for _, host := range hosts { - if host.ID == hostID { - return host, true - } - } - return models.DockerHost{}, false + return m.state.GetDockerHost(hostID) } // GetDockerHosts returns a point-in-time snapshot of all Docker hosts Pulse knows about. @@ -1442,14 +1429,16 @@ func (m *Monitor) resolveDockerHostView(hostID string) (*unifiedresources.Docker return nil, "", false } -func (m *Monitor) snapshotBackedUnifiedReadState() unifiedresources.ReadState { +// Only report admission uses this temporary registry. Full publication and +// consumer read paths keep the complete snapshot and canonical inventory. +func (m *Monitor) snapshotBackedAgentLookupReadState() unifiedresources.ReadState { if m == nil || m.state == nil { return nil } registry := unifiedresources.NewRegistry(nil) thresholds := m.resourceStaleThresholds() - registry.IngestSnapshotWithStaleThresholds(m.state.GetSnapshot(), thresholds) + registry.IngestSnapshotWithStaleThresholds(m.state.GetAgentLookupSnapshot(), thresholds) return unifiedresources.NewMonitorAdapterWithStaleThresholds(registry, thresholds) } @@ -1697,12 +1686,7 @@ func (m *Monitor) persistHostContinuity(host models.Host, report agentshost.Repo } func (m *Monitor) hostByID(hostID string) (models.Host, bool) { - for _, host := range m.state.GetHosts() { - if host.ID == hostID { - return host, true - } - } - return models.Host{}, false + return m.state.GetHost(hostID) } func (m *Monitor) applyRejectedHostReportLiveness( @@ -2052,7 +2036,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con }, nil } - readState := m.snapshotBackedUnifiedReadState() + readState := m.snapshotBackedAgentLookupReadState() var dockerHosts []*unifiedresources.DockerHostView if readState != nil { dockerHosts = readState.DockerHosts() @@ -2107,7 +2091,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con defer unlockDockerReport() receivedAt := time.Now() previousState, _ := m.GetDockerHost(identifier) - readState = m.snapshotBackedUnifiedReadState() + readState = m.snapshotBackedAgentLookupReadState() dockerHosts = nil previous = nil hasPrevious = false @@ -3074,7 +3058,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. baseIdentifier = strings.TrimSpace(persisted.HostID) } - readState := m.snapshotBackedUnifiedReadState() + readState := m.snapshotBackedAgentLookupReadState() var existingHosts []*unifiedresources.HostView if readState != nil { existingHosts = readState.Hosts() @@ -3263,7 +3247,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. // Identity resolution happens before taking the per-host lock. Refresh the // shared read model after waiting so token inheritance and previous-state // comparisons use the latest accepted report, not the pre-wait snapshot. - readState = m.snapshotBackedUnifiedReadState() + readState = m.snapshotBackedAgentLookupReadState() existingHosts = nil if readState != nil { existingHosts = readState.Hosts() @@ -3464,8 +3448,9 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. ) xcpngData := normalizeAgentXCPNGInventory(report.XCPNG, previousHostModel, observedAt) + previousHost, _ := m.hostByID(identifier) agentUpdate := mergeAgentUpdateStatus( - previousHostAgentUpdate(m.state.GetHosts(), identifier), + cloneModelAgentUpdateStatus(previousHost.AgentUpdate), convertAgentUpdateStatus(report.Agent.Update), strings.TrimSpace(report.Agent.UpdatedFrom), observedAt, @@ -3956,15 +3941,6 @@ func convertAgentPrivilegeStatus(value *agentshost.PrivilegeStatus) *models.Agen } } -func previousHostAgentUpdate(hosts []models.Host, identifier string) *models.AgentUpdateStatus { - for i := range hosts { - if hosts[i].ID == identifier { - return cloneModelAgentUpdateStatus(hosts[i].AgentUpdate) - } - } - return nil -} - func mergeAgentUpdateStatus(previous, reported *models.AgentUpdateStatus, updatedFrom string, observedAt time.Time) *models.AgentUpdateStatus { if reported == nil { reported = cloneModelAgentUpdateStatus(previous) diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 97e335410a..12e934c142 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -645,7 +645,7 @@ func TestSnapshotBackedUnifiedReadStateUsesConfiguredStaleThreshold(t *testing.T state: state, } - readState := monitor.snapshotBackedUnifiedReadState() + readState := monitor.snapshotBackedAgentLookupReadState() if readState == nil { t.Fatal("expected snapshot-backed read state") } @@ -1731,7 +1731,7 @@ func TestApplyHostReportUsesReceiptTimeForSkewedAgentClockLiveness(t *testing.T) t.Fatalf("host LastSeen followed skewed agent timestamp %s: got %s", agentClock, host.LastSeen) } - readState := monitor.snapshotBackedUnifiedReadState() + readState := monitor.snapshotBackedAgentLookupReadState() if readState == nil { t.Fatal("expected snapshot-backed read state") }