fix: log short-lived container shared-data timeouts at debug

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)
+	})
+}

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)

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

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 {

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

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))
+	})
+}