diff --git a/pkg/containerprofilemanager/v1/container_operations.go b/pkg/containerprofilemanager/v1/container_operations.go index 9dfff18d43..95aad429ba 100644 --- a/pkg/containerprofilemanager/v1/container_operations.go +++ b/pkg/containerprofilemanager/v1/container_operations.go @@ -141,3 +141,12 @@ func (cpm *ContainerProfileManager) removeContainerEntryIfMatch(containerID stri delete(cpm.containers, containerID) return true } + +// abandonEntry untracks a failed or skipped registration, then closes ready. +// Removing first lets a waiting replayed add see the entry gone and retry. +func (cpm *ContainerProfileManager) abandonEntry(containerID string, entry *ContainerEntry) { + cpm.removeContainerEntryIfMatch(containerID, entry) + entry.readyOnce.Do(func() { + close(entry.ready) + }) +} diff --git a/pkg/containerprofilemanager/v1/container_operations_test.go b/pkg/containerprofilemanager/v1/container_operations_test.go index d21bdae73a..4068c66ec2 100644 --- a/pkg/containerprofilemanager/v1/container_operations_test.go +++ b/pkg/containerprofilemanager/v1/container_operations_test.go @@ -88,7 +88,7 @@ func TestAddContainerWithTimeout_StaleFailureCleanupDoesNotDeleteNewerEntry(t *t // already removes failingEntry from the map before addContainer returns. expiredCtx, cancel := context.WithCancel(context.Background()) cancel() - err = cpm.addContainer(container, expiredCtx) + err = cpm.addContainer(container, failingEntry, expiredCtx) require.Error(t, err, "addContainer must fail against an already-cancelled context") _, stillTracked := cpm.getContainerEntry(containerID) diff --git a/pkg/containerprofilemanager/v1/containerprofile_manager.go b/pkg/containerprofilemanager/v1/containerprofile_manager.go index cc7c867bd5..1a8c2a8656 100644 --- a/pkg/containerprofilemanager/v1/containerprofile_manager.go +++ b/pkg/containerprofilemanager/v1/containerprofile_manager.go @@ -72,6 +72,7 @@ type containerData struct { // ContainerProfileManager manages container profiles and their lifecycle type ContainerProfileManager struct { lifecycleQueue utils.LifecycleQueue + pendingAdds utils.PendingAdds ctx context.Context cfg config.Config k8sClient k8sclient.K8sClientInterface diff --git a/pkg/containerprofilemanager/v1/lifecycle.go b/pkg/containerprofilemanager/v1/lifecycle.go index fd540ff7fe..3d4cac5fab 100644 --- a/pkg/containerprofilemanager/v1/lifecycle.go +++ b/pkg/containerprofilemanager/v1/lifecycle.go @@ -23,13 +23,24 @@ func (cpm *ContainerProfileManager) ContainerCallback(notif containercollection. isHost := utils.IsHostContainer(notif.Container) if cpm.cfg.NamespaceFilterFile != "" { excluded := !isHost && cpm.cfg.IgnoreContainer(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.PodLabels) + // Track or abort before queueing. + // Otherwise a queued removal waits behind a stuck registration. + var parent context.Context + var release func() + switch notif.Type { + case containercollection.EventTypeAddContainer: + parent, release = cpm.pendingAdds.Track(notif.Container.Runtime.ContainerID) + case containercollection.EventTypeRemoveContainer: + cpm.pendingAdds.Cancel(notif.Container.Runtime.ContainerID) + } // Finish deletion before readmission of the same running container. // The regular callbacks deliberately launch asynchronous work. cpm.lifecycleQueue.Submit(notif.Container.Runtime.ContainerID, func() { switch notif.Type { case containercollection.EventTypeAddContainer: + defer release() if isHost || !cpm.cfg.IgnoreContainer(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.PodLabels) { - cpm.addContainerWithTimeout(notif.Container) + cpm.addContainerWithTimeout(parent, notif.Container) } case containercollection.EventTypeRemoveContainer: cpm.deleteContainerWithReason(notif.Container, excluded) @@ -46,15 +57,21 @@ func (cpm *ContainerProfileManager) ContainerCallback(notif containercollection. if !isHost && cpm.cfg.IgnoreContainer(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.PodLabels) { return } - go cpm.addContainerWithTimeout(notif.Container) + parent, release := cpm.pendingAdds.Track(notif.Container.Runtime.ContainerID) + go func() { + defer release() + cpm.addContainerWithTimeout(parent, notif.Container) + }() case containercollection.EventTypeRemoveContainer: + cpm.pendingAdds.Cancel(notif.Container.Runtime.ContainerID) // A namespace may have been excluded since admission; always clean up. go cpm.deleteContainer(notif.Container) } } -// addContainerWithTimeout handles adding a container with a timeout to prevent hanging -func (cpm *ContainerProfileManager) addContainerWithTimeout(container *containercollection.Container) { +// addContainerWithTimeout adds a container with a timeout. +// parent is canceled if the container is removed meanwhile. +func (cpm *ContainerProfileManager) addContainerWithTimeout(parent context.Context, container *containercollection.Container) { containerID := container.Runtime.ContainerID // Create container entry early with nil watchedContainerData @@ -65,7 +82,7 @@ func (cpm *ContainerProfileManager) addContainerWithTimeout(container *container for !cpm.addContainerEntryIfAbsent(containerID, entry) { // Another goroutine is already registering (or has registered) this // container. entry.ready closes on BOTH success and failure of that - // attempt (see the error/timeout branches below), so closure alone + // attempt (see abandonEntry), so closure alone // doesn't mean the container ended up tracked. Wait for that attempt // to settle, then check whether its entry is still in the map: if it // failed and cleaned up, this replayed add must retry the @@ -89,41 +106,19 @@ func (cpm *ContainerProfileManager) addContainerWithTimeout(container *container // the get-or-insert with the same entry. } - ctx, cancel := context.WithTimeout(context.Background(), MaxWaitForSharedContainerData) + ctx, cancel := context.WithTimeout(parent, MaxWaitForSharedContainerData) defer cancel() - done := make(chan error, 1) - go func() { - done <- cpm.addContainer(container, ctx) - }() - - select { - case err := <-done: - if err != nil { - logger.L().Error("failed to add container to the container profile manager", helpers.Error(err)) - // Close ready channel and remove entry on error. Conditional on - // this still being the entry this goroutine registered: a - // duplicate-registration retry (the loop above) may have already - // installed a newer, successfully-registered entry for the same - // containerID by the time this failure is observed here, and an - // unconditional removal would delete that newer entry instead. - entry.readyOnce.Do(func() { - close(entry.ready) - }) - cpm.removeContainerEntryIfMatch(containerID, entry) - } - case <-ctx.Done(): - logger.L().Error("timeout while adding container to the container profile manager", - helpers.String("containerID", container.Runtime.ContainerID), + // Run inline: addContainer only blocks in the ctx-bound wait. + // Closing ready earlier lets a deletion hit a nil SyncChannel. + if err := cpm.addContainer(container, entry, ctx); err != nil { + utils.AddFailureLogger(ctx)("failed to add container to the container profile manager", + helpers.String("containerID", containerID), helpers.String("containerName", container.Runtime.ContainerName), helpers.String("podName", container.K8s.PodName), - helpers.String("namespace", container.K8s.Namespace)) - // Close ready channel and remove entry on timeout (see the error - // branch above for why this must be conditional on entry match). - entry.readyOnce.Do(func() { - close(entry.ready) - }) - cpm.removeContainerEntryIfMatch(containerID, entry) + helpers.String("namespace", container.K8s.Namespace), + helpers.Error(err)) + cpm.abandonEntry(containerID, entry) } } @@ -158,20 +153,15 @@ func hostContainerWithIdentity(container *containercollection.Container, sharedD return &hostContainer } -// addContainer adds a container to the container profile manager -func (cpm *ContainerProfileManager) addContainer(container *containercollection.Container, ctx context.Context) error { +// addContainer completes the registration of entry. +// The caller has already inserted entry. +func (cpm *ContainerProfileManager) addContainer(container *containercollection.Container, entry *ContainerEntry, ctx context.Context) error { containerID := container.Runtime.ContainerID // Wait for shared container data with timeout sharedData, err := cpm.waitForSharedContainerData(containerID, ctx) if err != nil { - // Close ready channel and remove the container entry if we fail - if entry, exists := cpm.getContainerEntry(containerID); exists { - entry.readyOnce.Do(func() { - close(entry.ready) - }) - cpm.removeContainerEntryIfMatch(containerID, entry) - } + cpm.abandonEntry(containerID, entry) return fmt.Errorf("failed to get shared data for container %s: %w", containerID, err) } @@ -185,13 +175,7 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection. helpers.String("podName", container.K8s.PodName), helpers.String("namespace", container.K8s.Namespace), helpers.String("userDefinedProfile", sharedData.UserDefinedProfile)) - // Close ready channel before removing entry - if entry, exists := cpm.getContainerEntry(containerID); exists { - entry.readyOnce.Do(func() { - close(entry.ready) - }) - cpm.removeContainerEntryIfMatch(containerID, entry) - } + cpm.abandonEntry(containerID, entry) return nil } @@ -201,19 +185,12 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection. helpers.String("containerName", container.Runtime.ContainerName), helpers.String("podName", container.K8s.PodName), helpers.String("namespace", container.K8s.Namespace)) - // Close ready channel before removing entry - if entry, exists := cpm.getContainerEntry(containerID); exists { - entry.readyOnce.Do(func() { - close(entry.ready) - }) - cpm.removeContainerEntryIfMatch(containerID, entry) - } + cpm.abandonEntry(containerID, entry) return nil } - // Update the existing container entry with watchedContainerData - entry, exists := cpm.getContainerEntry(containerID) - if !exists || entry.data == nil { + // Update the container entry with watchedContainerData + if current, exists := cpm.getContainerEntry(containerID); !exists || current != entry || entry.data == nil { // Should not happen, but guard just in case return fmt.Errorf("container entry missing for %s after shared data ready", containerID) } @@ -421,7 +398,7 @@ func (cpm *ContainerProfileManager) deleteContainerWithReason(container *contain } // Remove the container entry from the map - cpm.removeContainerEntry(containerID) + cpm.removeContainerEntryIfMatch(containerID, entry) entry.mu.Lock() if entry.data != nil { diff --git a/pkg/containerprofilemanager/v1/lifecycle_host_timer_test.go b/pkg/containerprofilemanager/v1/lifecycle_host_timer_test.go index efa1579412..9c779023c0 100644 --- a/pkg/containerprofilemanager/v1/lifecycle_host_timer_test.go +++ b/pkg/containerprofilemanager/v1/lifecycle_host_timer_test.go @@ -351,7 +351,7 @@ func TestContainerCallback_ReplayDuringFailingRegistrationRetries(t *testing.T) done := make(chan struct{}) go func() { defer close(done) - cpm.addContainerWithTimeout(newHostPseudoContainer()) + cpm.addContainerWithTimeout(context.Background(), newHostPseudoContainer()) }() // Give the replay goroutine time to observe the existing entry and start diff --git a/pkg/containerprofilemanager/v1/lifecycle_removal_test.go b/pkg/containerprofilemanager/v1/lifecycle_removal_test.go new file mode 100644 index 0000000000..60103e2691 --- /dev/null +++ b/pkg/containerprofilemanager/v1/lifecycle_removal_test.go @@ -0,0 +1,166 @@ +package containerprofilemanager + +import ( + "context" + "testing" + "time" + + containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection" + eventtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types" + "github.com/kubescape/node-agent/pkg/config" + "github.com/kubescape/node-agent/pkg/dnsmanager" + "github.com/kubescape/node-agent/pkg/k8sclient" + "github.com/kubescape/node-agent/pkg/objectcache" + "github.com/kubescape/node-agent/pkg/seccompmanager" + "github.com/kubescape/node-agent/pkg/storage" + "github.com/kubescape/node-agent/pkg/utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newRemovalTestManager(t *testing.T, cfg config.Config, k8s objectcache.K8sObjectCache) *ContainerProfileManager { + t.Helper() + t.Setenv("QUEUE_DIR", t.TempDir()) + cpm, err := NewContainerProfileManager(context.Background(), cfg, &k8sclient.K8sClientMock{}, k8s, + &storage.StorageHttpClientMock{}, &dnsmanager.DNSManagerMock{}, &seccompmanager.SeccompManagerMock{}, nil, nil, nil) + require.NoError(t, err) + t.Cleanup(cpm.Close) + return cpm +} + +func removalTestContainer(id string) *containercollection.Container { + return &containercollection.Container{ + Runtime: containercollection.RuntimeMetadata{BasicRuntimeMetadata: eventtypes.BasicRuntimeMetadata{ + ContainerID: id, + ContainerName: "init", + ContainerPID: 42, + }}, + K8s: containercollection.K8sMetadata{BasicK8sMetadata: eventtypes.BasicK8sMetadata{ + Namespace: "default", + PodName: "pod", + }}, + } +} + +func newEntry() *ContainerEntry { + return &ContainerEntry{data: &containerData{}, ready: make(chan struct{})} +} + +// fetchHookK8sCache runs onGet before each shared-data lookup. +type fetchHookK8sCache struct { + *objectcache.K8sObjectCacheMock + onGet func() +} + +func (k *fetchHookK8sCache) GetSharedContainerData(containerID string) *objectcache.WatchedContainerData { + k.onGet() + return k.K8sObjectCacheMock.GetSharedContainerData(containerID) +} + +// Removal before shared data exists aborts the registration at once. +// Covers the regular and namespace-filter paths. +func TestContainerCallback_RemovalAbortsPendingRegistration(t *testing.T) { + for name, cfg := range map[string]config.Config{ + "regular": {MaxSniffingTime: time.Hour}, + "namespace filter": {MaxSniffingTime: time.Hour, NamespaceFilterFile: "filter.yaml"}, + } { + t.Run(name, func(t *testing.T) { + cpm := newRemovalTestManager(t, cfg, &objectcache.K8sObjectCacheMock{}) + container := removalTestContainer("short-lived") + + cpm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: container}) + require.Eventually(t, func() bool { + _, ok := cpm.getContainerEntry("short-lived") + return ok + }, 2*time.Second, 5*time.Millisecond, "registration must be waiting for shared data") + + cpm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeRemoveContainer, Container: container}) + require.Eventually(t, func() bool { + _, ok := cpm.getContainerEntry("short-lived") + return !ok && cpm.pendingAdds.Len("short-lived") == 0 + }, 2*time.Second, 5*time.Millisecond, "removal must abort the registration and clean up") + }) + } +} + +// Removal after shared data arrived must not close ready early. +// A deletion would then hit a nil SyncChannel. +func TestAddContainerWithTimeout_RemovalAfterSharedDataWaitsForRegistration(t *testing.T) { + mock := &objectcache.K8sObjectCacheMock{} + mock.SetSharedContainerData("racing", &objectcache.WatchedContainerData{ContainerID: "racing"}) + var cpm *ContainerProfileManager + k8s := &fetchHookK8sCache{K8sObjectCacheMock: mock, onGet: func() { cpm.pendingAdds.Cancel("racing") }} + cpm = newRemovalTestManager(t, config.Config{MaxSniffingTime: time.Hour, InitialDelay: time.Minute, UpdateDataPeriod: time.Minute}, k8s) + + parent, release := cpm.pendingAdds.Track("racing") + defer release() + cpm.addContainerWithTimeout(parent, removalTestContainer("racing")) + + entry, ok := cpm.getContainerEntry("racing") + require.True(t, ok, "registration past the shared-data wait must complete") + select { + case <-entry.ready: + default: + t.Fatal("ready must be closed once registration completes") + } + entry.mu.RLock() + defer entry.mu.RUnlock() + require.NotNil(t, entry.data.watchedContainerData) + assert.NotNil(t, entry.data.watchedContainerData.SyncChannel, "ready must not be released before SyncChannel exists") +} + +// A replayed add woken by ready must find the failed entry gone. +func TestAbandonEntry_UntracksBeforeReleasingWaiters(t *testing.T) { + cpm := &ContainerProfileManager{containers: make(map[string]*ContainerEntry)} + entry := newEntry() + require.True(t, cpm.addContainerEntryIfAbsent("c", entry)) + + stillTracked := make(chan bool, 1) + go func() { + <-entry.ready + _, ok := cpm.getContainerEntry("c") + stillTracked <- ok + }() + cpm.abandonEntry("c", entry) + assert.False(t, <-stillTracked) +} + +// Only a removal downgrades the failure. +// Live container at the deadline, or shared data present: still an error. +func TestAddContainer_FailureClassification(t *testing.T) { + k8s := &objectcache.K8sObjectCacheMock{} + cpm := newRemovalTestManager(t, config.Config{MaxSniffingTime: time.Hour, InitialDelay: time.Minute, UpdateDataPeriod: time.Minute}, k8s) + + t.Run("removed while waiting for shared data", func(t *testing.T) { + parent, release := cpm.pendingAdds.Track("removed") + defer release() + ctx, cancel := context.WithTimeout(parent, time.Minute) + defer cancel() + time.AfterFunc(20*time.Millisecond, func() { cpm.pendingAdds.Cancel("removed") }) + + require.Error(t, cpm.addContainer(removalTestContainer("removed"), newEntry(), ctx)) + assert.True(t, utils.RemovedDuringAdd(ctx)) + }) + + t.Run("live container without shared data", func(t *testing.T) { + parent, release := cpm.pendingAdds.Track("live") + defer release() + ctx, cancel := context.WithTimeout(parent, 50*time.Millisecond) + defer cancel() + + require.Error(t, cpm.addContainer(removalTestContainer("live"), newEntry(), ctx)) + assert.False(t, utils.RemovedDuringAdd(ctx)) + }) + + t.Run("failure with shared data present", func(t *testing.T) { + k8s.SetSharedContainerData("present", &objectcache.WatchedContainerData{ContainerID: "present"}) + parent, release := cpm.pendingAdds.Track("present") + defer release() + ctx, cancel := context.WithTimeout(parent, time.Minute) + defer cancel() + + // Untracked entry: addContainer fails after the shared-data wait. + require.Error(t, cpm.addContainer(removalTestContainer("present"), newEntry(), ctx)) + assert.False(t, utils.RemovedDuringAdd(ctx)) + }) +} diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index 45f9bfc838..eecc4ae86a 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -124,6 +124,7 @@ type ContainerProfileCacheImpl struct { entries maps.SafeMap[string, *CachedContainerProfile] pending maps.SafeMap[string, *pendingContainer] containerLocks *resourcelocks.ResourceLocks + pendingAdds utils.PendingAdds storageClient storage.ProfileClient k8sObjectCache objectcache.K8sObjectCache metricsManager metricsmanager.MetricsManager @@ -242,8 +243,13 @@ func (c *ContainerProfileCacheImpl) ContainerCallback(notif containercollection. containerCopy.K8s.Namespace = namespace container = &containerCopy } - go c.addContainerWithTimeout(container) + parent, release := c.pendingAdds.Track(container.Runtime.ContainerID) + go func() { + defer release() + c.addContainerWithTimeout(parent, container) + }() case containercollection.EventTypeRemoveContainer: + c.pendingAdds.Cancel(notif.Container.Runtime.ContainerID) // Skip the ignore check on Remove: a container added before its pod // labels matched the ignore filter would otherwise leak in the cache. // The reconciler eviction path is the safety net, but a Remove event @@ -282,9 +288,13 @@ func (c *ContainerProfileCacheImpl) ContainerCallback(notif containercollection. // addContainerWithTimeout runs addContainer with a 10-minute cap to prevent // a stuck storage client from wedging the callback goroutine. -func (c *ContainerProfileCacheImpl) addContainerWithTimeout(container *containercollection.Container) { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) +// parent is canceled if the container is removed meanwhile. +func (c *ContainerProfileCacheImpl) addContainerWithTimeout(parent context.Context, container *containercollection.Container) { + ctx, cancel := context.WithTimeout(parent, 10*time.Minute) defer cancel() + // Removal does not abort the profile fetch: wait for it, up to the deadline. + deadline, stop := withoutRemoval(ctx) + defer stop() done := make(chan error, 1) go func() { @@ -294,10 +304,10 @@ func (c *ContainerProfileCacheImpl) addContainerWithTimeout(container *container select { case err := <-done: if err != nil { - logger.L().Error("failed to add container to the container-profile cache", helpers.Error(err)) + utils.AddFailureLogger(ctx)("failed to add container to the container-profile cache", helpers.Error(err)) } - case <-ctx.Done(): - logger.L().Error("timeout while adding container to the container-profile cache", + case <-deadline.Done(): + utils.AddFailureLogger(ctx)("timeout while adding container to the container-profile cache", helpers.String("containerID", container.Runtime.ContainerID), helpers.String("containerName", container.Runtime.ContainerName), helpers.String("podName", container.K8s.PodName), @@ -315,12 +325,18 @@ func (c *ContainerProfileCacheImpl) addContainer(container *containercollection. return c.containerLocks.WithLockAndError(containerID, func() error { sharedData, err := c.waitForSharedContainerData(containerID, ctx) if err != nil { - logger.L().Error("failed to get shared data for container", + // Debug only when the container was removed first (#848). + utils.AddFailureLogger(ctx)("failed to get shared data for container", helpers.String("containerID", containerID), helpers.Error(err)) return err } + // Removal cancels only the wait above. The profile must still load: + // terminal events need it during the removal grace period. + fetchCtx, cancel := withoutRemoval(ctx) + defer cancel() + // Names we need: // cpName = per-container stable slug, for the consolidated CP. // Kept for forward-compat; current storage does not @@ -345,7 +361,7 @@ func (c *ContainerProfileCacheImpl) addContainer(container *containercollection. return err } - if populated := c.tryPopulateEntry(ctx, containerID, container, sharedData, cpName, workloadName); !populated { + if populated := c.tryPopulateEntry(fetchCtx, containerID, container, sharedData, cpName, workloadName); !populated { // No profile data available yet (no consolidated CP in storage, // and no authored CP resolved). Record a pending entry; // the reconciler will retry each tick until data shows up or the @@ -364,6 +380,15 @@ func (c *ContainerProfileCacheImpl) addContainer(container *containercollection. }) } +// withoutRemoval returns a context with ctx's deadline but none of its +// cancellation, so a removal of the container no longer aborts it. +func withoutRemoval(ctx context.Context) (context.Context, context.CancelFunc) { + if d, ok := ctx.Deadline(); ok { + return context.WithDeadline(context.WithoutCancel(ctx), d) + } + return context.WithCancel(context.WithoutCancel(ctx)) +} + // tryPopulateEntry issues the CP GET (plus the authored CP when the pod // label names one) and installs the cache entry on success. Returns true iff // an entry was installed. Must be called while holding diff --git a/pkg/objectcache/containerprofilecache/removal_abort_test.go b/pkg/objectcache/containerprofilecache/removal_abort_test.go new file mode 100644 index 0000000000..66163a1024 --- /dev/null +++ b/pkg/objectcache/containerprofilecache/removal_abort_test.go @@ -0,0 +1,138 @@ +package containerprofilecache + +import ( + "context" + "sync" + "testing" + "time" + + containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection" + helpersv1 "github.com/kubescape/k8s-interface/instanceidhandler/v1/helpers" + "github.com/kubescape/node-agent/pkg/config" + "github.com/kubescape/node-agent/pkg/objectcache" + "github.com/kubescape/node-agent/pkg/utils" + "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// waitSignalK8sCache closes waiting on the first shared-data lookup. +// The cache does that lookup under the container lock. +type waitSignalK8sCache struct { + *objectcache.K8sObjectCacheMock + once sync.Once + waiting chan struct{} +} + +func (k *waitSignalK8sCache) GetSharedContainerData(containerID string) *objectcache.WatchedContainerData { + k.once.Do(func() { close(k.waiting) }) + return k.K8sObjectCacheMock.GetSharedContainerData(containerID) +} + +// Removal before shared data exists aborts the registration at once. +func TestContainerCallback_RemovalAbortsPendingRegistration(t *testing.T) { + k8s := &waitSignalK8sCache{K8sObjectCacheMock: &objectcache.K8sObjectCacheMock{}, waiting: make(chan struct{})} + c := NewContainerProfileCache(config.Config{ProfilesCacheRefreshRate: 30 * time.Second}, &fakeProfileClient{}, k8s, nil) + container := eventContainer("short-lived") + + c.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: container}) + select { + case <-k8s.waiting: + case <-time.After(2 * time.Second): + t.Fatal("registration never started waiting for shared data") + } + + c.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeRemoveContainer, Container: container}) + + // The wait holds the container lock. + // Taking it proves the aborted wait returned. + acquired := make(chan struct{}) + go func() { + c.containerLocks.WithLock("short-lived", func() {}) + close(acquired) + }() + select { + case <-acquired: + case <-time.After(2 * time.Second): + t.Fatal("removal did not abort the pending registration") + } +} + +// Only a removal downgrades the failure. +// Live container at the deadline: still an error. +func TestAddContainer_FailureClassification(t *testing.T) { + c, _ := newTestCache(t, &fakeProfileClient{}) + + t.Run("removed while waiting for shared data", func(t *testing.T) { + parent, release := c.pendingAdds.Track("removed") + defer release() + ctx, cancel := context.WithTimeout(parent, time.Minute) + defer cancel() + time.AfterFunc(20*time.Millisecond, func() { c.pendingAdds.Cancel("removed") }) + + require.Error(t, c.addContainer(eventContainer("removed"), ctx)) + assert.True(t, utils.RemovedDuringAdd(ctx)) + }) + + t.Run("live container without shared data", func(t *testing.T) { + parent, release := c.pendingAdds.Track("live") + defer release() + ctx, cancel := context.WithTimeout(parent, 50*time.Millisecond) + defer cancel() + + require.Error(t, c.addContainer(eventContainer("live"), ctx)) + assert.False(t, utils.RemovedDuringAdd(ctx)) + }) +} + +// delayedProfileClient holds the profile GET until released, honoring ctx like a real client. +type delayedProfileClient struct { + *fakeProfileClient + once sync.Once + inFlight chan struct{} + release chan struct{} +} + +func (b *delayedProfileClient) GetContainerProfile(ctx context.Context, ns, name string) (*v1beta1.ContainerProfile, error) { + b.once.Do(func() { close(b.inFlight) }) + select { + case <-b.release: + return b.fakeProfileClient.GetContainerProfile(ctx, ns, name) + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// Removal after shared data exists must not abort the profile GET: +// terminal events resolve the profile during the removal grace period. +func TestContainerCallback_RemovalKeepsProfileFetch(t *testing.T) { + client := &delayedProfileClient{ + fakeProfileClient: &fakeProfileClient{cp: &v1beta1.ContainerProfile{ + Name: "cp-init", + Namespace: "default", + Annotations: map[string]string{ + helpersv1.CompletionMetadataKey: helpersv1.Full, + helpersv1.StatusMetadataKey: helpersv1.Completed, + }, + }}, + inFlight: make(chan struct{}), + release: make(chan struct{}), + } + c, k8s := newTestCache(t, client) + primeSharedData(t, k8s, "init", "wlid://cluster-a/namespace-default/deployment-nginx") + container := eventContainer("init") + + c.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: container}) + select { + case <-client.inFlight: + case <-time.After(2 * time.Second): + t.Fatal("registration never issued the profile GET") + } + + c.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeRemoveContainer, Container: container}) + close(client.release) + + require.Eventually(t, func() bool { + return c.GetProjectedContainerProfile("init") != nil + }, 2*time.Second, 5*time.Millisecond, "profile must be available during the removal grace period") +} diff --git a/pkg/utils/pending_adds.go b/pkg/utils/pending_adds.go new file mode 100644 index 0000000000..f41473cc36 --- /dev/null +++ b/pkg/utils/pending_adds.go @@ -0,0 +1,88 @@ +package utils + +import ( + "context" + "errors" + "sync" + + "github.com/kubescape/go-logger" + "github.com/kubescape/go-logger/helpers" +) + +// ErrContainerRemoved is the cancel cause when a remove event aborts a registration. +var ErrContainerRemoved = errors.New("container removed") + +// PendingAdds tracks in-flight registrations so a remove event can abort them. +// A failure from that abort is expected: the container exited first (#848). +// Any other failure is not. +// The zero value is ready for use. +type PendingAdds struct { + mu sync.Mutex + adds map[string]map[*pendingAdd]struct{} +} + +type pendingAdd struct { + cancel context.CancelCauseFunc +} + +// Track registers a registration for key. +// A later Cancel(key) cancels the returned context with ErrContainerRemoved. +// Call release once the registration settles. +// Call Track in the add callback itself, not in a spawned goroutine, +// so the following remove callback always sees it. +func (p *PendingAdds) Track(key string) (context.Context, func()) { + ctx, cancel := context.WithCancelCause(context.Background()) + add := &pendingAdd{cancel: cancel} + p.mu.Lock() + if p.adds == nil { + p.adds = make(map[string]map[*pendingAdd]struct{}) + } + if p.adds[key] == nil { + p.adds[key] = make(map[*pendingAdd]struct{}) + } + p.adds[key][add] = struct{}{} + p.mu.Unlock() + return ctx, func() { + p.mu.Lock() + delete(p.adds[key], add) + if len(p.adds[key]) == 0 { + delete(p.adds, key) + } + p.mu.Unlock() + cancel(nil) + } +} + +// Cancel aborts every registration tracked for key. +// Later Track calls are not affected, so readmission works. +func (p *PendingAdds) Cancel(key string) { + p.mu.Lock() + adds := p.adds[key] + delete(p.adds, key) + p.mu.Unlock() + for add := range adds { + add.cancel(ErrContainerRemoved) + } +} + +// Len returns the number of tracked registrations for key. +func (p *PendingAdds) Len(key string) int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.adds[key]) +} + +// RemovedDuringAdd reports whether PendingAdds.Cancel canceled ctx or a parent. +func RemovedDuringAdd(ctx context.Context) bool { + return errors.Is(context.Cause(ctx), ErrContainerRemoved) +} + +// AddFailureLogger returns Debug if removal aborted the registration, else Error. +// Missing shared data alone is no proof of removal: +// its producer may still be retrying for a live container. +func AddFailureLogger(ctx context.Context) func(string, ...helpers.IDetails) { + if RemovedDuringAdd(ctx) { + return logger.L().Debug + } + return logger.L().Error +} diff --git a/pkg/utils/pending_adds_test.go b/pkg/utils/pending_adds_test.go new file mode 100644 index 0000000000..2b7b331c41 --- /dev/null +++ b/pkg/utils/pending_adds_test.go @@ -0,0 +1,70 @@ +package utils + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPendingAdds_CancelAbortsTrackedRegistrations(t *testing.T) { + var p PendingAdds + first, releaseFirst := p.Track("c1") + defer releaseFirst() + second, releaseSecond := p.Track("c1") + defer releaseSecond() + other, releaseOther := p.Track("c2") + defer releaseOther() + + child, cancel := context.WithTimeout(first, time.Hour) + defer cancel() + + p.Cancel("c1") + + require.ErrorIs(t, first.Err(), context.Canceled) + assert.True(t, RemovedDuringAdd(first)) + assert.True(t, RemovedDuringAdd(second)) + assert.True(t, RemovedDuringAdd(child), "a timeout context derived from the tracked one must inherit the removal cause") + assert.NoError(t, other.Err(), "other containers must not be affected") + assert.Zero(t, p.Len("c1")) + assert.Equal(t, 1, p.Len("c2")) +} + +func TestPendingAdds_ReadmissionAfterCancelIsNotAborted(t *testing.T) { + var p PendingAdds + _, releaseOld := p.Track("c1") + p.Cancel("c1") + releaseOld() + + readmitted, release := p.Track("c1") + defer release() + assert.NoError(t, readmitted.Err()) + assert.Equal(t, 1, p.Len("c1")) +} + +func TestPendingAdds_ReleaseKeepsCauseAndUntracks(t *testing.T) { + var p PendingAdds + ctx, release := p.Track("c1") + p.Cancel("c1") + release() + assert.True(t, RemovedDuringAdd(ctx), "release after removal must not overwrite the cause") + + ctx, release = p.Track("c1") + release() + assert.Zero(t, p.Len("c1")) + assert.False(t, RemovedDuringAdd(ctx), "a settled registration is not a removal") + p.Cancel("c1") // no tracked registration: no-op +} + +func TestRemovedDuringAdd_OnlyRemovalCounts(t *testing.T) { + var p PendingAdds + parent, release := p.Track("c1") + defer release() + ctx, cancel := context.WithTimeout(parent, time.Millisecond) + defer cancel() + <-ctx.Done() + assert.False(t, RemovedDuringAdd(ctx), "a deadline for a live container must stay an error") + assert.False(t, RemovedDuringAdd(context.Background()), "a failure without cancellation must stay an error") +}