From 93156c8dbc3844fcb77396d7eba6e4f0fecda54c Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 12:19:32 +0200 Subject: [PATCH 1/8] fix: cache only completed container profiles to ensure data integrity Signed-off-by: Matthias Bertschy --- .../containerprofilecache.go | 20 ++-- .../containerprofilecache_test.go | 17 +++- .../containerprofilecache/reconciler.go | 36 ++++---- .../containerprofilecache/reconciler_test.go | 91 ++++++++++--------- .../t8_overlay_refresh_test.go | 2 + 5 files changed, 95 insertions(+), 71 deletions(-) diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index e85f693c35..3c80f6f87e 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -362,16 +362,16 @@ func (c *ContainerProfileCacheImpl) tryPopulateEntry( } } - // Fix (reviewer #3): if the consolidated CP is still Partial and this - // container is not PreRunning (i.e. we saw it start fresh after the - // agent was already up), the partial view belongs to a PREVIOUS container - // incarnation. Legacy caches explicitly deleted such partials on restart - // so rule evaluation fell through to "no profile" until a new Full - // profile arrived. Mirror that: keep pending, retry each tick. - if !sharedData.PreRunningContainer { - if cp != nil && cp.Annotations[helpersv1.CompletionMetadataKey] == helpersv1.Partial { - cp = nil - } + // Only cache profiles whose status is Completed. Completion (Partial/Full) + // is about data coverage and does not affect caching eligibility. + // Return false (not nil) so the synthetic-CP fallback below does not run + // and inadvertently cache via an overlay when the real CP is not yet done. + if cp != nil && cp.Annotations[helpersv1.StatusMetadataKey] != helpersv1.Completed { + logger.L().Debug("tryPopulateEntry: CP status not completed; keeping pending", + helpers.String("containerID", containerID), + helpers.String("namespace", ns), + helpers.String("status", cp.Annotations[helpersv1.StatusMetadataKey])) + return false } // Fetch user-authored legacy CRDs when the pod carries the diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache_test.go b/pkg/objectcache/containerprofilecache/containerprofilecache_test.go index f828d37643..66dbbcaf43 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache_test.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache_test.go @@ -172,8 +172,11 @@ func TestSharedFastPath_NoOverlay(t *testing.T) { // is merged into the projected profile. func TestOverlayPath_DeepCopies(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp-1", Namespace: "default", ResourceVersion: "1"}, - Spec: v1beta1.ContainerProfileSpec{Capabilities: []string{"SYS_PTRACE"}}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-1", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, + Spec: v1beta1.ContainerProfileSpec{Capabilities: []string{"SYS_PTRACE"}}, } userAP := &v1beta1.ApplicationProfile{ ObjectMeta: metav1.ObjectMeta{Name: "override", Namespace: "default", ResourceVersion: "u1"}, @@ -207,7 +210,10 @@ func TestOverlayPath_DeepCopies(t *testing.T) { // fresh mutex. func TestDeleteContainer_LockAndCleanup(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp-delete", Namespace: "default", ResourceVersion: "1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-delete", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, } client := &fakeProfileClient{cp: cp} c, k8s := newTestCache(t, client) @@ -283,7 +289,10 @@ func TestContainerCallback_HostContainer(t *testing.T) { // GetCallStackSearchTree. func TestCallStackIndexBuiltFromProfile(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp-stack", Namespace: "default", ResourceVersion: "1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-stack", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, Spec: v1beta1.ContainerProfileSpec{ IdentifiedCallStacks: []v1beta1.IdentifiedCallStack{ { diff --git a/pkg/objectcache/containerprofilecache/reconciler.go b/pkg/objectcache/containerprofilecache/reconciler.go index 14be22eaa7..aa1030c5ca 100644 --- a/pkg/objectcache/containerprofilecache/reconciler.go +++ b/pkg/objectcache/containerprofilecache/reconciler.go @@ -2,13 +2,13 @@ // // The reconciler is the safety-net eviction path AND the freshness refresh // loop. Each tick it: -// 1. reconcileOnce: evicts cache entries whose pod is gone or whose -// container is no longer Running. -// 2. refreshAllEntries (single-flight via atomic flag): re-fetches the -// consolidated CP, the workload-level AP+NN, the user-managed -// "ug-" AP+NN, and any label-referenced user AP/NN overlay, -// then rebuilds the projection iff any resourceVersion changed. Fast-skip -// when every RV matches what's already cached. +// 1. reconcileOnce: evicts cache entries whose pod is gone or whose +// container is no longer Running. +// 2. refreshAllEntries (single-flight via atomic flag): re-fetches the +// consolidated CP, the workload-level AP+NN, the user-managed +// "ug-" AP+NN, and any label-referenced user AP/NN overlay, +// then rebuilds the projection iff any resourceVersion changed. Fast-skip +// when every RV matches what's already cached. // // RPC cost @ 300 containers / 30s cadence steady-state: up to 7 gets per // entry per tick (CP + 3×AP + 3×NN). At 300 entries that's 70 RPC/s in the @@ -273,12 +273,9 @@ func (c *ContainerProfileCacheImpl) refreshAllEntries(ctx context.Context) { // // base CP → workload AP+NN → user-managed (ug-) AP+NN → user overlay AP+NN. // -// We intentionally DO NOT re-apply the partial-on-non-PreRunning gate here: -// any entry that survived addContainer already passed that gate (or was -// PreRunning), so refresh can accept partial profiles freely. (Fix B for -// Test_17 / Test_19: the workload AP/NN must be re-fetched each tick so a -// "ready" -> "completed" transition propagates to ProfileState.Status, which -// in turn promotes fail_on_profile from false to true.) +// The completed-only gate is re-applied here: if the CP regresses to a +// non-Completed status we keep the existing cached entry rather than +// projecting stale/incomplete data. func (c *ContainerProfileCacheImpl) refreshOneEntry(ctx context.Context, id string, e *CachedContainerProfile) { // Resurrection guard (reviewer #1): refreshAllEntries snapshots entries // without holding containerLocks, so a concurrent deleteContainer / @@ -322,6 +319,13 @@ func (c *ContainerProfileCacheImpl) refreshOneEntry(ctx context.Context, id stri helpers.Error(cpErr)) cp = nil } + if cp != nil && cp.Annotations[helpersv1.StatusMetadataKey] != helpersv1.Completed { + logger.L().Debug("refreshOneEntry: CP status not completed; keeping cached entry", + helpers.String("containerID", id), + helpers.String("cpName", e.CPName), + helpers.String("status", cp.Annotations[helpersv1.StatusMetadataKey])) + return + } var userManagedAP *v1beta1.ApplicationProfile var userManagedNN *v1beta1.NetworkNeighborhood if e.WorkloadName != "" { @@ -519,9 +523,9 @@ func (c *ContainerProfileCacheImpl) rebuildEntryFromSources( } newEntry := &CachedContainerProfile{ - Projected: projectedCP, - SpecHash: projectedCP.SpecHash, - State: &objectcache.ProfileState{Completion: effectiveCP.Annotations[helpersv1.CompletionMetadataKey], Status: effectiveCP.Annotations[helpersv1.StatusMetadataKey], Name: effectiveCP.Name}, + Projected: projectedCP, + SpecHash: projectedCP.SpecHash, + State: &objectcache.ProfileState{Completion: effectiveCP.Annotations[helpersv1.CompletionMetadataKey], Status: effectiveCP.Annotations[helpersv1.StatusMetadataKey], Name: effectiveCP.Name}, CallStackTree: tree, ContainerName: prev.ContainerName, PodName: prev.PodName, diff --git a/pkg/objectcache/containerprofilecache/reconciler_test.go b/pkg/objectcache/containerprofilecache/reconciler_test.go index 3b572dc9c0..49b3eeeb68 100644 --- a/pkg/objectcache/containerprofilecache/reconciler_test.go +++ b/pkg/objectcache/containerprofilecache/reconciler_test.go @@ -344,7 +344,10 @@ func TestReconcilerExitsOnCtxCancel(t *testing.T) { // TestRefreshFastSkipWhenAllRVsMatch — delta #4. When CP RV and both overlay // RVs match the cached values, refreshOneEntry returns without rebuilding. func TestRefreshFastSkipWhenAllRVsMatch(t *testing.T) { - cp := &v1beta1.ContainerProfile{ObjectMeta: metav1.ObjectMeta{Name: "cp", Namespace: "default", ResourceVersion: "100"}} + cp := &v1beta1.ContainerProfile{ObjectMeta: metav1.ObjectMeta{ + Name: "cp", Namespace: "default", ResourceVersion: "100", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }} ap := &v1beta1.ApplicationProfile{ObjectMeta: metav1.ObjectMeta{Name: "override", Namespace: "default", ResourceVersion: "50"}} nn := &v1beta1.NetworkNeighborhood{ObjectMeta: metav1.ObjectMeta{Name: "override", Namespace: "default", ResourceVersion: "60"}} client := &countingProfileClient{cp: cp, ap: ap, nn: nn} @@ -389,8 +392,11 @@ func TestRefreshFastSkipWhenAllRVsMatch(t *testing.T) { // a newer AP RV and rebuilds. func TestRefreshRebuildsOnUserAPChange(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp", Namespace: "default", ResourceVersion: "100"}, - Spec: v1beta1.ContainerProfileSpec{Capabilities: []string{"SYS_PTRACE"}}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp", Namespace: "default", ResourceVersion: "100", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, + Spec: v1beta1.ContainerProfileSpec{Capabilities: []string{"SYS_PTRACE"}}, } ap := &v1beta1.ApplicationProfile{ ObjectMeta: metav1.ObjectMeta{Name: "override", Namespace: "default", ResourceVersion: "51"}, @@ -442,8 +448,11 @@ func TestRefreshRebuildsOnUserAPChange(t *testing.T) { // TestRefreshRebuildsOnCPChange — CP RV changed; entry rebuilds with fresh CP. func TestRefreshRebuildsOnCPChange(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp", Namespace: "default", ResourceVersion: "101"}, - Spec: v1beta1.ContainerProfileSpec{Capabilities: []string{"SYS_ADMIN"}}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp", Namespace: "default", ResourceVersion: "101", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, + Spec: v1beta1.ContainerProfileSpec{Capabilities: []string{"SYS_ADMIN"}}, } client := &countingProfileClient{cp: cp} k8s := newControllableK8sCache() @@ -469,7 +478,10 @@ func TestRefreshRebuildsOnCPChange(t *testing.T) { // reflects the new execs AND that the legacy-load metric was re-emitted. func TestT8_EndToEndRefreshUpdatesProjection(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp", Namespace: "default", ResourceVersion: "100"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp", Namespace: "default", ResourceVersion: "100", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, Spec: v1beta1.ContainerProfileSpec{ Execs: []v1beta1.ExecCalls{{Path: "/bin/base", Args: []string{"a"}}}, }, @@ -850,6 +862,7 @@ func TestRetryPendingEntries_CPCreatedAfterAdd(t *testing.T) { Name: "cp-pending", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, }, } @@ -918,11 +931,10 @@ type testNotFoundErr struct{ name string } func (e *testNotFoundErr) Error() string { return "container profile " + e.name + ": not found" } -// TestPartialCP_NonPreRunning_StaysPending verifies that a CP marked partial -// is NOT cached when the container is not PreRunning (i.e. started after the -// agent was up). Legacy caches explicitly deleted partials on restart; we -// mirror that by staying pending until the CP becomes Full. -func TestPartialCP_NonPreRunning_StaysPending(t *testing.T) { +// TestPartialCP_Accepted verifies that a CP with Status=Completed is cached +// regardless of its Completion value (Partial or Full). Completion describes +// data coverage, not caching eligibility — only Status matters. +func TestPartialCP_Accepted(t *testing.T) { cp := &v1beta1.ContainerProfile{ ObjectMeta: metav1.ObjectMeta{ Name: "cp-partial", @@ -937,28 +949,16 @@ func TestPartialCP_NonPreRunning_StaysPending(t *testing.T) { client := &fakeProfileClient{cp: cp} c, k8s := newTestCache(t, client) - id := "container-partial-restart" + id := "container-partial" primeSharedData(t, k8s, id, "wlid://cluster-a/namespace-default/deployment-nginx") - // sharedData.PreRunningContainer is false by default → this simulates a - // fresh container start observed by a running agent. require.NoError(t, c.addContainer(eventContainer(id), context.Background())) - assert.Nil(t, c.GetProjectedContainerProfile(id), "partial CP must not populate cache on fresh container") - assert.Equal(t, 1, c.pending.Len(), "partial-on-restart stays pending") - - // Simulate the CP becoming Full (new agent-side aggregation round). - cp.Annotations[helpersv1.CompletionMetadataKey] = helpersv1.Full - cp.ResourceVersion = "2" - c.retryPendingEntries(context.Background()) - - assert.NotNil(t, c.GetProjectedContainerProfile(id), "Full CP promotes pending entry") - assert.Equal(t, 0, c.pending.Len(), "pending drained on Full") + assert.NotNil(t, c.GetProjectedContainerProfile(id), "Partial+Completed CP must be accepted into cache") + assert.Equal(t, 0, c.pending.Len(), "not pending when Status=Completed") } -// TestPartialCP_PreRunning_Accepted verifies the inverse: when the agent -// restarts (all containers become PreRunning), we accept even a partial CP so -// rule evaluation can still alert on out-of-profile behavior (Test_19 -// semantics). +// TestPartialCP_PreRunning_Accepted verifies that PreRunning containers also +// accept a partial CP when Status=Completed (same rule as non-PreRunning). func TestPartialCP_PreRunning_Accepted(t *testing.T) { cp := &v1beta1.ContainerProfile{ ObjectMeta: metav1.ObjectMeta{ @@ -989,7 +989,10 @@ func TestPartialCP_PreRunning_Accepted(t *testing.T) { // on subsequent ticks instead of permanently dropping the overlay. func TestOverlayLabel_TransientFetchFailure_RefsRetained(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp-with-overlay", Namespace: "default", ResourceVersion: "1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-with-overlay", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, } // Overlay fetch returns an error; the base CP is fine. client := &fakeProfileClient{cp: cp, apErr: assertErrNotFound("override"), nnErr: assertErrNotFound("override")} @@ -1018,7 +1021,10 @@ func TestOverlayLabel_TransientFetchFailure_RefsRetained(t *testing.T) { // NOT re-insert it. func TestRefreshDoesNotResurrectDeletedEntry(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp-resurrect", Namespace: "default", ResourceVersion: "1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-resurrect", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, } client := &fakeProfileClient{cp: cp} c, k8s := newTestCache(t, client) @@ -1102,7 +1108,7 @@ func TestRefreshUpdatesCPStatus(t *testing.T) { ResourceVersion: "1", Annotations: map[string]string{ helpersv1.CompletionMetadataKey: helpersv1.Full, - helpersv1.StatusMetadataKey: helpersv1.Learning, // "ready" + helpersv1.StatusMetadataKey: helpersv1.Learning, // not yet completed }, }, } @@ -1113,11 +1119,11 @@ func TestRefreshUpdatesCPStatus(t *testing.T) { primeSharedData(t, k8s, id, "wlid://cluster-a/namespace-default/deployment-nginx") require.NoError(t, c.addContainer(eventContainer(id), context.Background())) - entry, ok := c.entries.Load(id) - require.True(t, ok, "entry populated from CP") - require.NotNil(t, entry.State) - assert.Equal(t, helpersv1.Learning, entry.State.Status, - "Status reflects the CP at add time (ready / learning)") + // A CP with a non-Completed status must not be accepted into the cache; + // the container stays pending until the CP transitions to Completed. + _, ok := c.entries.Load(id) + assert.False(t, ok, "non-completed CP must not populate cache entry") + assert.Equal(t, 1, c.pending.Len(), "container stays pending while CP status is learning") // Storage transitions CP to Status=completed. client.cp = &v1beta1.ContainerProfile{ @@ -1132,14 +1138,14 @@ func TestRefreshUpdatesCPStatus(t *testing.T) { }, } - c.refreshAllEntries(context.Background()) + c.retryPendingEntries(context.Background()) stored, ok := c.entries.Load(id) - require.True(t, ok) + require.True(t, ok, "entry populated after CP becomes completed") require.NotNil(t, stored.State) assert.Equal(t, helpersv1.Completed, stored.State.Status, - "refresh propagates CP Status=completed into ProfileState") - assert.Equal(t, "2", stored.RV, "refresh records the new CP RV") + "ProfileState reflects Completed status") + assert.Equal(t, "2", stored.RV, "RV recorded from the completed CP") } // TestUserManagedProfileMerged exercises the user-managed merge path @@ -1216,7 +1222,10 @@ func TestUserManagedProfileMerged(t *testing.T) { // tests cannot wait for the background goroutine, so we drive it explicitly. func TestSpecChange_TriggersReprojection(t *testing.T) { cp := &v1beta1.ContainerProfile{ - ObjectMeta: metav1.ObjectMeta{Name: "cp", Namespace: "default", ResourceVersion: "1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "cp", Namespace: "default", ResourceVersion: "1", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, + }, Spec: v1beta1.ContainerProfileSpec{ Capabilities: []string{"SYS_PTRACE", "NET_ADMIN"}, }, diff --git a/pkg/objectcache/containerprofilecache/t8_overlay_refresh_test.go b/pkg/objectcache/containerprofilecache/t8_overlay_refresh_test.go index 3802e52b3e..4bf4496ef6 100644 --- a/pkg/objectcache/containerprofilecache/t8_overlay_refresh_test.go +++ b/pkg/objectcache/containerprofilecache/t8_overlay_refresh_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + helpersv1 "github.com/kubescape/k8s-interface/instanceidhandler/v1/helpers" "github.com/kubescape/node-agent/pkg/config" "github.com/kubescape/node-agent/pkg/objectcache" cpc "github.com/kubescape/node-agent/pkg/objectcache/containerprofilecache" @@ -29,6 +30,7 @@ func TestT8_EndToEndRefreshUpdatesProjection(t *testing.T) { Name: "cp", Namespace: "default", ResourceVersion: "100", + Annotations: map[string]string{helpersv1.StatusMetadataKey: helpersv1.Completed}, }, Spec: v1beta1.ContainerProfileSpec{ Execs: []v1beta1.ExecCalls{{Path: "/bin/base", Args: []string{"a"}}}, From 8e0afcc0d91af8af417c7c88583052f7faed6797 Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 13:03:46 +0200 Subject: [PATCH 2/8] fix: update k8s-interface dependency to v0.0.210 Signed-off-by: Matthias Bertschy --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index b45a7d7ed9..2a1b73be13 100644 --- a/go.mod +++ b/go.mod @@ -34,7 +34,7 @@ require ( github.com/joncrlsn/dque v0.0.0-20241024143830-7723fd131a64 github.com/kubescape/backend v0.0.39 github.com/kubescape/go-logger v0.0.28 - github.com/kubescape/k8s-interface v0.0.207 + github.com/kubescape/k8s-interface v0.0.210 github.com/kubescape/storage v0.0.258 github.com/kubescape/workerpool v0.0.0-20250526074519-0e4a4e7f44cf github.com/moby/sys/mountinfo v0.7.2 diff --git a/go.sum b/go.sum index d076c623ac..f00bff5efb 100644 --- a/go.sum +++ b/go.sum @@ -883,8 +883,8 @@ github.com/kubescape/backend v0.0.39 h1:B1QRfKCSFlzuE+jWOnk/l7EpH71/Q3n14KKq0QSn github.com/kubescape/backend v0.0.39/go.mod h1:cMEGP8cXUZgY89YU4GRBGIla9HZW7grZsUtlCwvZgAE= github.com/kubescape/go-logger v0.0.28 h1:xulKTp9kOg3rD98sopFELQ6yZCHQoQXMDzteoSHDFKI= github.com/kubescape/go-logger v0.0.28/go.mod h1:YZHFjwGCDar1hP9OyBLE46oR7a0Y/Z/0FperDo8+9D0= -github.com/kubescape/k8s-interface v0.0.207 h1:jX+EqZLjSArw4xa+XMvjnnoK0Q8IxdD2tvihwLa/WGg= -github.com/kubescape/k8s-interface v0.0.207/go.mod h1:WNYUG93aZ5kDmuaRKFLtVhp18Yc6EfaHdD1gLYtVTN4= +github.com/kubescape/k8s-interface v0.0.210 h1:3TiO3lYxdIHncoBRAMAMFdwanHmllUpYKFy5cG0h97o= +github.com/kubescape/k8s-interface v0.0.210/go.mod h1:WNYUG93aZ5kDmuaRKFLtVhp18Yc6EfaHdD1gLYtVTN4= github.com/kubescape/storage v0.0.258 h1:0mL0z3dAmtP1qup7VgoEgwLgbBSROu5oOusBAPeMmus= github.com/kubescape/storage v0.0.258/go.mod h1:VHs+xQzvZKE2lJDN8rR1sFmTa43N6XJAcatZ249gviU= github.com/kubescape/syft v1.32.0-ks.2 h1:xdUksUmKEyyVKsTfJDYW8Z5HawVJtelsUolPOsWtDx0= From 9b767e317e5bb28a765b65dbdb4f580515eab2a3 Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 14:05:05 +0200 Subject: [PATCH 3/8] fix: replace 5s pending-ticker with completion notification callback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instead of polling pending containers every 5s (6× storage pressure), wire a CompletionNotifier from containerprofilemanager to the CP cache. NotifyContainerCompleted is called on ContainerReachedMaxTime and ObjectCompletedError, launching a bounded retry goroutine (5 × 3s) that promotes the pending entry as soon as storage consolidation completes. The 30s reconciler tick retains retryPendingEntries as a safety net. Signed-off-by: Matthias Bertschy Co-Authored-By: Claude Sonnet 4.6 --- cmd/main.go | 3 ++ .../v1/containerprofile_manager.go | 12 +++++ pkg/containerprofilemanager/v1/monitoring.go | 2 + pkg/objectcache/completion_notifier.go | 9 ++++ .../containerprofilecache.go | 49 ++++++++++++++++--- .../containerprofilecache/reconciler.go | 8 ++- 6 files changed, 73 insertions(+), 10 deletions(-) create mode 100644 pkg/objectcache/completion_notifier.go diff --git a/cmd/main.go b/cmd/main.go index b81d6d15a5..a0b8f36029 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -298,6 +298,9 @@ func main() { cpc := containerprofilecache.NewContainerProfileCache(cfg, storageClient, k8sObjectCache, prometheusExporter) cpc.Start(ctx) + if cpm, ok := containerProfileManager.(*containerprofilemanagerv1.ContainerProfileManager); ok { + cpm.SetCompletionNotifier(cpc) + } logger.L().Info("ContainerProfileCache active; legacy AP/NN caches removed") dc := dnscache.NewDnsCache(dnsResolver) diff --git a/pkg/containerprofilemanager/v1/containerprofile_manager.go b/pkg/containerprofilemanager/v1/containerprofile_manager.go index f8455d61c1..7b20f44719 100644 --- a/pkg/containerprofilemanager/v1/containerprofile_manager.go +++ b/pkg/containerprofilemanager/v1/containerprofile_manager.go @@ -92,6 +92,18 @@ type ContainerProfileManager struct { hostProfile *v1beta1.ContainerProfile hostProfileMu sync.RWMutex hostID string + + completionNotifier objectcache.CompletionNotifier +} + +func (cpm *ContainerProfileManager) SetCompletionNotifier(n objectcache.CompletionNotifier) { + cpm.completionNotifier = n +} + +func (cpm *ContainerProfileManager) notifyCompleted(containerID string) { + if cpm.completionNotifier != nil { + cpm.completionNotifier.NotifyContainerCompleted(containerID) + } } // NewContainerProfileManager creates a new container profile manager diff --git a/pkg/containerprofilemanager/v1/monitoring.go b/pkg/containerprofilemanager/v1/monitoring.go index 4832327db1..b8bb89bccd 100644 --- a/pkg/containerprofilemanager/v1/monitoring.go +++ b/pkg/containerprofilemanager/v1/monitoring.go @@ -63,6 +63,7 @@ func (cpm *ContainerProfileManager) monitorContainer(container *containercollect helpers.String("status", string(watchedContainer.GetStatus())), helpers.String("completionStatus", string(watchedContainer.GetCompletionStatus()))) } + cpm.notifyCompleted(watchedContainer.ContainerID) // Signal ack to lifecycle goroutine if watchedContainer.AckChan != nil { watchedContainer.AckChan <- struct{}{} @@ -97,6 +98,7 @@ func (cpm *ContainerProfileManager) handleSaveProfileError(err error, watchedCon watchedContainer.SetStatus(objectcache.WatchedContainerStatusCompleted) cpm.deleteContainer(container) cpm.notifyContainerEndOfLife(container) + cpm.notifyCompleted(watchedContainer.ContainerID) return file.ObjectCompletedError } else { logger.L().Error("failed to save container profile", helpers.Error(err), diff --git a/pkg/objectcache/completion_notifier.go b/pkg/objectcache/completion_notifier.go new file mode 100644 index 0000000000..2fa519229a --- /dev/null +++ b/pkg/objectcache/completion_notifier.go @@ -0,0 +1,9 @@ +package objectcache + +// CompletionNotifier is implemented by ContainerProfileCacheImpl. The +// containerprofilemanager calls NotifyContainerCompleted when it writes a +// container profile with status="completed" to storage, allowing the CP cache +// to promote any pending entry without waiting for the next reconciler tick. +type CompletionNotifier interface { + NotifyContainerCompleted(containerID string) +} diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index 3c80f6f87e..0c9c6113f7 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -108,8 +108,8 @@ type ContainerProfileCacheImpl struct { k8sObjectCache objectcache.K8sObjectCache metricsManager metricsmanager.MetricsManager - reconcileEvery time.Duration - rpcBudget time.Duration + reconcileEvery time.Duration + rpcBudget time.Duration refreshInProgress atomic.Bool // deprecationDedup tracks (kind|ns/name@rv) keys to emit one WARN log @@ -140,14 +140,14 @@ func NewContainerProfileCache(cfg config.Config, storageClient storage.ProfileCl rpcBudget = defaultStorageRPCBudget } return &ContainerProfileCacheImpl{ - cfg: cfg, - containerLocks: resourcelocks.New(), - storageClient: storageClient, - k8sObjectCache: k8sObjectCache, - metricsManager: metricsManager, + cfg: cfg, + containerLocks: resourcelocks.New(), + storageClient: storageClient, + k8sObjectCache: k8sObjectCache, + metricsManager: metricsManager, reconcileEvery: reconcileEvery, rpcBudget: rpcBudget, - nudge: make(chan struct{}, 1), + nudge: make(chan struct{}, 1), } } @@ -659,5 +659,38 @@ func (c *ContainerProfileCacheImpl) waitForSharedContainerData(containerID strin }, backoff.WithBackOff(backoff.NewExponentialBackOff())) } +// NotifyContainerCompleted is called by containerprofilemanager when it writes a +// CP with status="completed". If the container is still pending it launches a +// bounded retry goroutine (up to 5 attempts × 3 s) so the cache entry is +// promoted within seconds of the consolidation cycle completing, without waiting +// for the next 30 s reconciler tick. +func (c *ContainerProfileCacheImpl) NotifyContainerCompleted(containerID string) { + p, pending := c.pending.Load(containerID) + if !pending { + return + } + go func() { + for i := 0; i < 5; i++ { + if i > 0 { + time.Sleep(3 * time.Second) + } + if _, still := c.pending.Load(containerID); !still { + return + } + ctx, cancel := context.WithTimeout(context.Background(), c.rpcBudget) + var promoted bool + c.containerLocks.WithLock(containerID, func() { + if _, still := c.pending.Load(containerID); still { + promoted = c.tryPopulateEntry(ctx, containerID, p.container, p.sharedData, p.cpName, p.workloadName) + } + }) + cancel() + if promoted { + return + } + } + }() +} + // Ensure ContainerProfileCacheImpl implements the ContainerProfileCache interface. var _ objectcache.ContainerProfileCache = (*ContainerProfileCacheImpl)(nil) diff --git a/pkg/objectcache/containerprofilecache/reconciler.go b/pkg/objectcache/containerprofilecache/reconciler.go index aa1030c5ca..da0a997172 100644 --- a/pkg/objectcache/containerprofilecache/reconciler.go +++ b/pkg/objectcache/containerprofilecache/reconciler.go @@ -32,8 +32,12 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -// tickLoop drives the reconciler. Evict runs synchronously on the tick; -// refresh runs on a single-flight goroutine guarded by refreshInProgress so a +// tickLoop drives the reconciler. Each tick it evicts terminated containers, +// retries pending entries, and refreshes all cached entries. Pending-entry +// retries are also triggered immediately via NotifyContainerCompleted when the +// containerprofilemanager writes a CP with status="completed". +// +// Refresh runs on a single-flight goroutine guarded by refreshInProgress so a // slow refresh never stacks. func (c *ContainerProfileCacheImpl) tickLoop(ctx context.Context) { if c.reconcileEvery == 0 { From 2cfbbf429ac9519b40e26feb01d384be0bd85a61 Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 14:28:44 +0200 Subject: [PATCH 4/8] fix: extend NotifyContainerCompleted retries to 60s to span consolidation window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Storage consolidation runs every 30s; the original 5×3s=15s retry window expired before consolidation completed, leaving the container in pending until the next 30s reconciler tick. Increasing to 20×3s=60s ensures at least one retry fires after consolidation, closing the race with Test_12_MergingProfilesTest's 10s post-completion sleep. Signed-off-by: Matthias Bertschy Co-Authored-By: Claude Sonnet 4.6 --- pkg/objectcache/containerprofilecache/containerprofilecache.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index 0c9c6113f7..c96240d68b 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -670,7 +670,7 @@ func (c *ContainerProfileCacheImpl) NotifyContainerCompleted(containerID string) return } go func() { - for i := 0; i < 5; i++ { + for i := 0; i < 20; i++ { if i > 0 { time.Sleep(3 * time.Second) } From d5dd0115346592a91ec0dbaa9207be407d355e8a Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 15:17:33 +0200 Subject: [PATCH 5/8] fix: accept TooLarge as terminal CP status in cache gate TooLarge is a terminal state (manager stopped collecting; data is truncated but valid for detection). The Completed-only gate in tryPopulateEntry and refreshOneEntry incorrectly kept TooLarge containers pending indefinitely since they never transition to Completed. Introduce isTerminalCPStatus(Completed|TooLarge) and apply it in both code paths. Add TestTooLargeCP_Accepted covering the addContainer and refresh paths. Signed-off-by: Matthias Bertschy Co-Authored-By: Claude Sonnet 4.6 --- .../containerprofilecache.go | 24 +++++++--- .../containerprofilecache/reconciler.go | 4 +- .../containerprofilecache/reconciler_test.go | 46 +++++++++++++++++++ 3 files changed, 66 insertions(+), 8 deletions(-) diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index c96240d68b..3ee210da2b 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -362,12 +362,13 @@ func (c *ContainerProfileCacheImpl) tryPopulateEntry( } } - // Only cache profiles whose status is Completed. Completion (Partial/Full) - // is about data coverage and does not affect caching eligibility. - // Return false (not nil) so the synthetic-CP fallback below does not run - // and inadvertently cache via an overlay when the real CP is not yet done. - if cp != nil && cp.Annotations[helpersv1.StatusMetadataKey] != helpersv1.Completed { - logger.L().Debug("tryPopulateEntry: CP status not completed; keeping pending", + // Only cache profiles whose status is terminal (Completed or TooLarge). + // Learning/ready profiles are still being written; caching them would let + // rules fire against incomplete data. TooLarge is terminal: the manager + // stopped collecting but the truncated data is still valid for detection. + // Return false so the synthetic-CP fallback below does not bypass the gate. + if cp != nil && !isTerminalCPStatus(cp.Annotations[helpersv1.StatusMetadataKey]) { + logger.L().Debug("tryPopulateEntry: CP status not terminal; keeping pending", helpers.String("containerID", containerID), helpers.String("namespace", ns), helpers.String("status", cp.Annotations[helpersv1.StatusMetadataKey])) @@ -692,5 +693,16 @@ func (c *ContainerProfileCacheImpl) NotifyContainerCompleted(containerID string) }() } +// isTerminalCPStatus reports whether the CP status annotation value represents +// a terminal state that the cache should accept. Terminal states are: +// - Completed: learning period finished normally. +// - TooLarge: manager stopped collecting because the profile grew too large; +// the truncated data is still valid for rule evaluation. +// +// Learning ("ready") is not terminal — the CP is still being written. +func isTerminalCPStatus(status string) bool { + return status == helpersv1.Completed || status == helpersv1.TooLarge +} + // Ensure ContainerProfileCacheImpl implements the ContainerProfileCache interface. var _ objectcache.ContainerProfileCache = (*ContainerProfileCacheImpl)(nil) diff --git a/pkg/objectcache/containerprofilecache/reconciler.go b/pkg/objectcache/containerprofilecache/reconciler.go index da0a997172..0af5c8ee49 100644 --- a/pkg/objectcache/containerprofilecache/reconciler.go +++ b/pkg/objectcache/containerprofilecache/reconciler.go @@ -323,8 +323,8 @@ func (c *ContainerProfileCacheImpl) refreshOneEntry(ctx context.Context, id stri helpers.Error(cpErr)) cp = nil } - if cp != nil && cp.Annotations[helpersv1.StatusMetadataKey] != helpersv1.Completed { - logger.L().Debug("refreshOneEntry: CP status not completed; keeping cached entry", + if cp != nil && !isTerminalCPStatus(cp.Annotations[helpersv1.StatusMetadataKey]) { + logger.L().Debug("refreshOneEntry: CP status not terminal; keeping cached entry", helpers.String("containerID", id), helpers.String("cpName", e.CPName), helpers.String("status", cp.Annotations[helpersv1.StatusMetadataKey])) diff --git a/pkg/objectcache/containerprofilecache/reconciler_test.go b/pkg/objectcache/containerprofilecache/reconciler_test.go index 49b3eeeb68..7f37810617 100644 --- a/pkg/objectcache/containerprofilecache/reconciler_test.go +++ b/pkg/objectcache/containerprofilecache/reconciler_test.go @@ -1148,6 +1148,52 @@ func TestRefreshUpdatesCPStatus(t *testing.T) { assert.Equal(t, "2", stored.RV, "RV recorded from the completed CP") } +// TestTooLargeCP_Accepted verifies that a CP with Status=TooLarge is treated +// as a terminal state and cached immediately, not kept in the pending queue. +// TooLarge profiles have truncated but valid data that the rule engine should +// use for detection; rejecting them would leave the container permanently +// pending since the manager never transitions TooLarge → Completed. +func TestTooLargeCP_Accepted(t *testing.T) { + cp := &v1beta1.ContainerProfile{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-too-large", + Namespace: "default", + ResourceVersion: "1", + Annotations: map[string]string{ + helpersv1.CompletionMetadataKey: helpersv1.Partial, + helpersv1.StatusMetadataKey: helpersv1.TooLarge, + }, + }, + Spec: v1beta1.ContainerProfileSpec{ + Execs: []v1beta1.ExecCalls{{Path: "/bin/sh"}}, + }, + } + client := &fakeProfileClient{cp: cp} + c, k8s := newTestCache(t, client) + + id := "container-too-large" + primeSharedData(t, k8s, id, "wlid://cluster-a/namespace-default/deployment-nginx") + require.NoError(t, c.addContainer(eventContainer(id), context.Background())) + + stored, ok := c.entries.Load(id) + require.True(t, ok, "TooLarge CP must populate cache entry immediately") + assert.Equal(t, 0, c.pending.Len(), "no pending entry when Status=TooLarge") + require.NotNil(t, stored.State) + assert.Equal(t, helpersv1.TooLarge, stored.State.Status, "ProfileState reflects TooLarge status") + assert.Equal(t, "1", stored.RV, "RV recorded from the too-large CP") + + // refreshOneEntry must also accept TooLarge on subsequent ticks. + cp2 := cp.DeepCopy() + cp2.ResourceVersion = "2" + cp2.Spec.Execs = append(cp2.Spec.Execs, v1beta1.ExecCalls{Path: "/usr/bin/id"}) + client.cp = cp2 + c.refreshAllEntries(context.Background()) + + stored2, ok := c.entries.Load(id) + require.True(t, ok, "entry survives refresh with TooLarge status") + assert.Equal(t, "2", stored2.RV, "RV updated on refresh") +} + // TestUserManagedProfileMerged exercises the user-managed merge path // (Test_12_MergingProfilesTest / Test_13_MergingNetworkNeighborhoodTest): // a user-managed AP published at "ug-" is merged on top of From d5473b6483b7f5849c30e9e2fda63f04a9c53afd Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 15:37:04 +0200 Subject: [PATCH 6/8] fix: fire terminal notification for TooLarge containers Wire notifyCompleted into the ObjectTooLargeError path in handleSaveProfileError so pending containers promoted to TooLarge get the same immediate cache-promotion goroutine as Completed ones. Without this, a TooLarge container stayed in the pending map for up to 30s (next reconciler tick) even though a valid truncated CP was already in storage. Add TestNotifyContainerTerminal_TooLarge to assert the fast-path promotion fires without waiting for the tick. Signed-off-by: Matthias Bertschy Co-Authored-By: Claude Sonnet 4.6 --- pkg/containerprofilemanager/v1/monitoring.go | 1 + .../containerprofilecache/reconciler_test.go | 50 +++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/pkg/containerprofilemanager/v1/monitoring.go b/pkg/containerprofilemanager/v1/monitoring.go index b8bb89bccd..06162cac05 100644 --- a/pkg/containerprofilemanager/v1/monitoring.go +++ b/pkg/containerprofilemanager/v1/monitoring.go @@ -93,6 +93,7 @@ func (cpm *ContainerProfileManager) handleSaveProfileError(err error, watchedCon watchedContainer.SetStatus(objectcache.WatchedContainerStatusTooLarge) cpm.deleteContainer(container) cpm.notifyContainerEndOfLife(container) + cpm.notifyCompleted(watchedContainer.ContainerID) return file.ObjectTooLargeError } else if err.Error() == file.ObjectCompletedError.Error() { watchedContainer.SetStatus(objectcache.WatchedContainerStatusCompleted) diff --git a/pkg/objectcache/containerprofilecache/reconciler_test.go b/pkg/objectcache/containerprofilecache/reconciler_test.go index 7f37810617..ca1d4de323 100644 --- a/pkg/objectcache/containerprofilecache/reconciler_test.go +++ b/pkg/objectcache/containerprofilecache/reconciler_test.go @@ -1194,6 +1194,56 @@ func TestTooLargeCP_Accepted(t *testing.T) { assert.Equal(t, "2", stored2.RV, "RV updated on refresh") } +// TestNotifyContainerTerminal_TooLarge verifies the fast-path promotion for a +// container whose CP becomes TooLarge. Without waiting for the 30s reconciler +// tick, NotifyContainerCompleted must promote the pending entry immediately once +// the TooLarge CP is visible in storage. +func TestNotifyContainerTerminal_TooLarge(t *testing.T) { + // Start with no CP in storage so the container stays pending. + client := &fakeProfileClient{cp: nil} + c, k8s := newTestCache(t, client) + + id := "container-too-large-notify" + primeSharedData(t, k8s, id, "wlid://cluster-a/namespace-default/deployment-nginx") + require.NoError(t, c.addContainer(eventContainer(id), context.Background())) + + assert.Equal(t, 1, c.pending.Len(), "container is pending while CP absent") + _, ok := c.entries.Load(id) + assert.False(t, ok, "no entry while CP absent") + + // Storage now has a TooLarge terminal CP. + client.cp = &v1beta1.ContainerProfile{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-too-large", + Namespace: "default", + ResourceVersion: "1", + Annotations: map[string]string{ + helpersv1.CompletionMetadataKey: helpersv1.Partial, + helpersv1.StatusMetadataKey: helpersv1.TooLarge, + }, + }, + Spec: v1beta1.ContainerProfileSpec{ + Execs: []v1beta1.ExecCalls{{Path: "/bin/sh"}}, + }, + } + + // Simulate the notification fired by containerprofilemanager on TooLarge. + // The goroutine launched by NotifyContainerCompleted must promote the entry + // without the caller needing to wait for the 30s periodic tick. + c.NotifyContainerCompleted(id) + + // Allow the notification goroutine to run (first attempt is immediate). + assert.Eventually(t, func() bool { + _, ok := c.entries.Load(id) + return ok + }, 2*time.Second, 10*time.Millisecond, "entry promoted via notification, no 30s tick needed") + + assert.Equal(t, 0, c.pending.Len(), "pending cleared after TooLarge promotion") + stored, ok := c.entries.Load(id) + require.True(t, ok) + assert.Equal(t, helpersv1.TooLarge, stored.State.Status) +} + // TestUserManagedProfileMerged exercises the user-managed merge path // (Test_12_MergingProfilesTest / Test_13_MergingNetworkNeighborhoodTest): // a user-managed AP published at "ug-" is merged on top of From ba0b0625cfd16f80d94f296b123f06e54b7f9a90 Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 15:49:52 +0200 Subject: [PATCH 7/8] fix: notify cache on normal container exit (ContainerHasTerminatedError) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a container exits normally the monitorContainer loop handles ContainerHasTerminatedError: it saves the final CP (Status=Completed set by lifecycle.go) but previously never called notifyCompleted, so pending cache entries were only promoted on the next 30s reconciler tick. Fire notifyCompleted when the terminal status is Completed so the fast-path promotion goroutine runs immediately. Add TestNotifyContainerTerminal_Completed: pending → Completed via notification, verified without waiting for the periodic tick. Signed-off-by: Matthias Bertschy Co-Authored-By: Claude Sonnet 4.6 --- pkg/containerprofilemanager/v1/monitoring.go | 3 ++ .../containerprofilecache/reconciler_test.go | 42 +++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/pkg/containerprofilemanager/v1/monitoring.go b/pkg/containerprofilemanager/v1/monitoring.go index 06162cac05..c7e7f9f051 100644 --- a/pkg/containerprofilemanager/v1/monitoring.go +++ b/pkg/containerprofilemanager/v1/monitoring.go @@ -47,6 +47,9 @@ func (cpm *ContainerProfileManager) monitorContainer(container *containercollect helpers.String("status", string(watchedContainer.GetStatus())), helpers.String("completionStatus", string(watchedContainer.GetCompletionStatus()))) } + if watchedContainer.GetStatus() == objectcache.WatchedContainerStatusCompleted { + cpm.notifyCompleted(watchedContainer.ContainerID) + } // Signal ack to lifecycle goroutine if watchedContainer.AckChan != nil { watchedContainer.AckChan <- struct{}{} diff --git a/pkg/objectcache/containerprofilecache/reconciler_test.go b/pkg/objectcache/containerprofilecache/reconciler_test.go index ca1d4de323..e76c384d6a 100644 --- a/pkg/objectcache/containerprofilecache/reconciler_test.go +++ b/pkg/objectcache/containerprofilecache/reconciler_test.go @@ -1244,6 +1244,48 @@ func TestNotifyContainerTerminal_TooLarge(t *testing.T) { assert.Equal(t, helpersv1.TooLarge, stored.State.Status) } +// TestNotifyContainerTerminal_Completed verifies the fast-path promotion for a +// container that exits normally (ContainerHasTerminatedError with +// Status=Completed). The notification goroutine must promote the pending entry +// without waiting for the 30s reconciler tick. +func TestNotifyContainerTerminal_Completed(t *testing.T) { + // No CP yet — container stays pending. + client := &fakeProfileClient{cp: nil} + c, k8s := newTestCache(t, client) + + id := "container-normal-exit" + primeSharedData(t, k8s, id, "wlid://cluster-a/namespace-default/deployment-nginx") + require.NoError(t, c.addContainer(eventContainer(id), context.Background())) + + assert.Equal(t, 1, c.pending.Len(), "container pending while CP absent") + + // Simulate the lifecycle: container exits → CP written with Status=Completed. + client.cp = &v1beta1.ContainerProfile{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cp-exited", + Namespace: "default", + ResourceVersion: "1", + Annotations: map[string]string{ + helpersv1.CompletionMetadataKey: helpersv1.Full, + helpersv1.StatusMetadataKey: helpersv1.Completed, + }, + }, + } + + // Simulate what monitorContainer now does on ContainerHasTerminatedError. + c.NotifyContainerCompleted(id) + + assert.Eventually(t, func() bool { + _, ok := c.entries.Load(id) + return ok + }, 2*time.Second, 10*time.Millisecond, "entry promoted via notification without 30s tick") + + assert.Equal(t, 0, c.pending.Len(), "pending cleared after normal-exit promotion") + stored, ok := c.entries.Load(id) + require.True(t, ok) + assert.Equal(t, helpersv1.Completed, stored.State.Status) +} + // TestUserManagedProfileMerged exercises the user-managed merge path // (Test_12_MergingProfilesTest / Test_13_MergingNetworkNeighborhoodTest): // a user-managed AP published at "ug-" is merged on top of From 97670970d37f660dc0639a74d2bc63b8bb7ffef8 Mon Sep 17 00:00:00 2001 From: Matthias Bertschy Date: Mon, 18 May 2026 16:11:20 +0200 Subject: [PATCH 8/8] fix: pre-initialize SafeMap fields to eliminate Load/Set data race goradd/maps.SafeMap.Load() reads m.items == nil without a lock while Set() initializes m.items under a write lock, causing a data race on the first concurrent access to a zero-value SafeMap. Pre-initialize entries and pending in NewContainerProfileCache so m.items is non-nil before any goroutine can call Load concurrently. Signed-off-by: Matthias Bertschy Co-Authored-By: Claude Sonnet 4.6 --- .../containerprofilecache.go | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index 3ee210da2b..3c2535ab8c 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -139,16 +139,24 @@ func NewContainerProfileCache(cfg config.Config, storageClient storage.ProfileCl if rpcBudget <= 0 { rpcBudget = defaultStorageRPCBudget } - return &ContainerProfileCacheImpl{ - cfg: cfg, - containerLocks: resourcelocks.New(), - storageClient: storageClient, - k8sObjectCache: k8sObjectCache, - metricsManager: metricsManager, + c := &ContainerProfileCacheImpl{ + cfg: cfg, + containerLocks: resourcelocks.New(), + storageClient: storageClient, + k8sObjectCache: k8sObjectCache, + metricsManager: metricsManager, reconcileEvery: reconcileEvery, rpcBudget: rpcBudget, - nudge: make(chan struct{}, 1), + nudge: make(chan struct{}, 1), } + // Pre-initialize SafeMap internal maps: Load() reads m.items == nil without + // a lock while Set() writes m.items under a write lock, causing a data race + // on the first concurrent access to a zero-value SafeMap. + c.entries.Set("", nil) + c.entries.Delete("") + c.pending.Set("", nil) + c.pending.Delete("") + return c } func shouldLogOptionalUserManagedFetchError(err error) bool {