Merge reviewed scoped agent identity lookups

Preserve exact Core candidate history and complete published monitoring inventory while removing unrelated temporary metadata copies.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-05 02:32:33 +01:00
11 changed files with 579 additions and 41 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
}
+164
View File
@@ -0,0 +1,164 @@
package models
import (
"bytes"
"encoding/json"
"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)
}
}
// Freeze the expectation independently of the clone under test. A
// defective clone must not mutate both the source and its reference.
before, err := json.Marshal(s.GetSnapshot())
if err != nil {
t.Fatal(err)
}
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"
after, err := json.Marshal(s.GetSnapshot())
if err != nil || !bytes.Equal(before, after) {
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...),
+257
View File
@@ -0,0 +1,257 @@
package monitoring
import (
"encoding/json"
"fmt"
"reflect"
"testing"
"time"
"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
// nodes and all non-Kubernetes sources deliberately remain in the lookup.
var agentLookupMetadataCollections = []string{
"Namespaces", "Services", "Ingresses", "EndpointSlices", "NetworkPolicies",
"PersistentVolumes", "PersistentVolumeClaims", "StorageClasses", "ConfigMaps",
"Secrets", "ServiceAccounts", "Roles", "ClusterRoles", "RoleBindings",
"ClusterRoleBindings", "ResourceQuotas", "LimitRanges", "PodDisruptionBudgets",
"HorizontalPodAutoscalers", "Events",
}
func agentLookupFixture(metadataCount int) *Monitor {
now := time.Now().UTC()
state := models.NewState()
state.UpsertHost(models.Host{
ID: "host", Hostname: "worker", MachineID: "machine", Status: "online", LastSeen: now,
TokenID: "host-token", Tags: []string{"host-tag"},
NetworkInterfaces: []models.HostNetworkInterface{{Name: "eth0", Addresses: []string{"192.0.2.1"}}},
})
state.UpsertDockerHost(models.DockerHost{
ID: "docker-host", AgentID: "host", Hostname: "worker", MachineID: "machine",
Status: "online", LastSeen: now, TokenID: "docker-token", CPUs: 4,
Containers: []models.DockerContainer{{ID: "app", Name: "app", State: "running", Labels: map[string]string{"app": "web"}}},
})
state.UpdateNodes([]models.Node{{ID: "pve:worker", Name: "worker", Instance: "pve", LinkedAgentID: "host", Status: "online", LastSeen: now}})
state.UpdateVMs([]models.VM{{ID: "pve:worker:100", Name: "vm", VMID: 100, Node: "worker", Instance: "pve", Status: "running", LastSeen: now}})
state.UpdateContainers([]models.Container{{ID: "pve:worker:101", Name: "lxc", VMID: 101, Node: "worker", Instance: "pve", Status: "running", LastSeen: now}})
cluster := models.KubernetesCluster{
ID: "cluster", AgentID: "host", Name: "cluster", Status: "online", LastSeen: now,
Nodes: []models.KubernetesNode{{UID: "node", Name: "worker", MachineID: "machine", Ready: true}},
Pods: []models.KubernetesPod{{UID: "pod", Name: "pod", NodeName: "worker", Namespace: "ns", Phase: "Running"}},
Deployments: []models.KubernetesDeployment{{UID: "deployment", Name: "deployment", Namespace: "ns"}},
}
value := reflect.ValueOf(&cluster).Elem()
for i := 0; i < metadataCount; i++ {
field := value.FieldByName(agentLookupMetadataCollections[i%len(agentLookupMetadataCollections)])
entry := reflect.New(field.Type().Elem()).Elem()
// Deliberately overlap metadata names/UIDs with host identities. Type,
// not a friendly naming convention, must keep these out of admission.
for key, text := range map[string]string{"UID": fmt.Sprintf("host-%d", i), "Name": "worker", "Namespace": "ns"} {
if f := entry.FieldByName(key); f.IsValid() && f.Kind() == reflect.String {
f.SetString(text)
}
}
field.Set(reflect.Append(field, entry))
}
state.UpsertKubernetesCluster(cluster)
return &Monitor{state: state, config: &config.Config{}}
}
func fullAgentLookupReference(m *Monitor) unifiedresources.ReadState {
registry := unifiedresources.NewRegistry(nil)
thresholds := m.resourceStaleThresholds()
registry.IngestSnapshotWithStaleThresholds(m.state.GetSnapshot(), thresholds)
return unifiedresources.NewMonitorAdapterWithStaleThresholds(registry, thresholds)
}
// Adapters timestamp a newly built registry with the wall clock. Compare
// every canonical host field except that construction timestamp, retaining
// source LastSeen, metric/source observation times and all identity payloads.
func agentLookupViewResources(t *testing.T, read unifiedresources.ReadState) [][]unifiedresources.Resource {
t.Helper()
adapter, ok := read.(*unifiedresources.MonitorAdapter)
if !ok {
t.Fatal("lookup must return the real monitor adapter")
}
byID := make(map[string]unifiedresources.Resource)
for _, resource := range adapter.GetAll() {
if resource.UpdatedAt.IsZero() {
t.Fatal("registry construction timestamp is missing")
}
resource.UpdatedAt = time.Time{}
byID[resource.ID] = resource
}
result := make([][]unifiedresources.Resource, 2)
for _, host := range read.Hosts() {
resource, found := byID[host.ID()]
if !found {
t.Fatalf("host view %q is absent from the registry", host.ID())
}
result[0] = append(result[0], resource)
}
for _, host := range read.DockerHosts() {
resource, found := byID[host.ID()]
if !found {
t.Fatalf("Docker host view %q is absent from the registry", host.ID())
}
result[1] = append(result[1], resource)
}
return result
}
func TestAgentLookupMatchesWholeSnapshotHostViews(t *testing.T) {
for _, count := range []int{0, 20, 1000} {
t.Run(fmt.Sprint(count), func(t *testing.T) {
m := agentLookupFixture(count)
before := m.state.GetSnapshot()
want := fullAgentLookupReference(m)
got := m.snapshotBackedAgentLookupReadState()
if len(want.Hosts()) == 0 || len(want.DockerHosts()) == 0 {
t.Fatal("fixture must exercise both host facets")
}
if !reflect.DeepEqual(agentLookupViewResources(t, got), agentLookupViewResources(t, want)) {
t.Fatal("agent admission changed a canonical host or Docker host view")
}
if !reflect.DeepEqual(before, m.state.GetSnapshot()) {
t.Fatal("agent lookup mutated the full inventory")
}
// A second report must see current source data without a cache or
// freshness grace period, including explicit clears and removal.
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.snapshotBackedAgentLookupReadState()
if !reflect.DeepEqual(agentLookupViewResources(t, got), agentLookupViewResources(t, want)) {
t.Fatal("agent lookup retained a removed host or old metadata")
}
})
}
}
func TestAgentLookupAllocationIgnoresUnrelatedKubernetesMetadata(t *testing.T) {
measure := func(count int) float64 {
m := agentLookupFixture(count)
return testing.AllocsPerRun(3, func() {
read := m.snapshotBackedAgentLookupReadState()
if len(read.Hosts()) == 0 || len(read.DockerHosts()) == 0 {
panic("missing fixture host")
}
})
}
small, large := measure(0), measure(1000)
t.Logf("agent identity lookup allocations: no metadata=%.0f, 1000 metadata objects=%.0f", small, large)
if large > small+10 {
t.Fatalf("unrelated Kubernetes metadata still adds identity lookup allocations: %.0f -> %.0f", small, large)
}
}
func TestAgentPointLookupAllocationIgnoresOtherHosts(t *testing.T) {
measure := func(count int) float64 {
m := &Monitor{state: models.NewState()}
for i := 0; i < count; i++ {
m.state.UpsertHost(models.Host{ID: fmt.Sprint(i), Hostname: fmt.Sprint(i), Tags: []string{"tag"}, Sensors: models.HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu": 40}}})
}
return testing.AllocsPerRun(10, func() {
if _, ok := m.hostByID("0"); !ok {
panic("missing fixture host")
}
})
}
small, large := measure(1), measure(17)
t.Logf("native host point lookup allocations: one host=%.0f, 17 hosts=%.0f", small, large)
if large > small+1 {
t.Fatalf("point lookup still clones unrelated hosts: %.0f -> %.0f", small, large)
}
}
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())
got, expected := registry.GetAll(), want.List()
for i := range got {
got[i].UpdatedAt = time.Time{}
}
for i := range expected {
expected[i].UpdatedAt = time.Time{}
}
if !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)
for _, reference := range []bool{true, false} {
name := "scoped"
if reference {
name = "whole-snapshot-reference"
}
b.Run(fmt.Sprintf("%s/metadata-%d", name, count), func(b *testing.B) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
var read unifiedresources.ReadState
if reference {
read = fullAgentLookupReference(m)
} else {
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")
}