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") }