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/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= 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..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{}{} @@ -63,6 +66,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{}{} @@ -92,11 +96,13 @@ 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) 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 e85f693c35..3c2535ab8c 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 @@ -139,7 +139,7 @@ func NewContainerProfileCache(cfg config.Config, storageClient storage.ProfileCl if rpcBudget <= 0 { rpcBudget = defaultStorageRPCBudget } - return &ContainerProfileCacheImpl{ + c := &ContainerProfileCacheImpl{ cfg: cfg, containerLocks: resourcelocks.New(), storageClient: storageClient, @@ -149,6 +149,14 @@ func NewContainerProfileCache(cfg config.Config, storageClient storage.ProfileCl rpcBudget: rpcBudget, 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 { @@ -362,16 +370,17 @@ 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 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])) + return false } // Fetch user-authored legacy CRDs when the pod carries the @@ -659,5 +668,49 @@ 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 < 20; 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 + } + } + }() +} + +// 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/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..0af5c8ee49 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 @@ -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 { @@ -273,12 +277,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 +323,13 @@ func (c *ContainerProfileCacheImpl) refreshOneEntry(ctx context.Context, id stri helpers.Error(cpErr)) cp = nil } + 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])) + return + } var userManagedAP *v1beta1.ApplicationProfile var userManagedNN *v1beta1.NetworkNeighborhood if e.WorkloadName != "" { @@ -519,9 +527,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..e76c384d6a 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,152 @@ 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") +} + +// 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") +} + +// 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) +} + +// 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 @@ -1216,7 +1360,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"}}},