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
This commit is contained in:
pulse-triage[bot]
2026-10-05 01:17:08 +01:00
parent bedb245c72
commit 187e5daaa8
11 changed files with 374 additions and 45 deletions
@@ -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
@@ -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`,
@@ -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
+44
View File
@@ -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
}
+156
View File
@@ -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()
}
+3
View File
@@ -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()
}
+28
View File
@@ -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()
+6 -2
View File
@@ -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...),
+60 -4
View File
@@ -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")
+13 -37
View File
@@ -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)
@@ -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")
}