fix: skip Pod metadata retries for standalone containers

pkg/containerwatcher/v2/containercallback.go

@@ -112,6 +112,12 @@ func (cw *ContainerWatcher) containerCallbackAsync(notif containercollection.Pub
 			return
 		}
 
+		if !utils.HasKubernetesPodIdentity(notif.Container) {
+			// Standalone runtime containers have no Pod to query. Other callback
+			// receivers still receive their lifecycle events and monitor them.
+			return
+		}
+
 		// Set shared watched container data
 		if cw.cfg.NamespaceFilterFile != "" {
 			// The per-receiver lifecycle queue orders this write before removal,

pkg/containerwatcher/v2/containercallback_standalone_test.go

@@ -0,0 +1,62 @@
+package containerwatcher
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	"github.com/kubescape/go-logger"
+	"github.com/kubescape/node-agent/pkg/config"
+	"github.com/kubescape/node-agent/pkg/metricsmanager"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestContainerCallbackAsync_StandaloneDoesNotAcquirePodData(t *testing.T) {
+	// Cancellation makes the synchronous metadata acquisition fail immediately;
+	// it must still report an error for an addressable Pod, even before container
+	// name enrichment. Names, labels or sandbox IDs alone cannot address a Pod.
+	// Standalone registration must never enter that acquisition path.
+	cases := []struct {
+		name      string
+		metadata  func(*containercollection.Container)
+		wantError bool
+	}{
+		{"standalone", func(*containercollection.Container) {}, false},
+		{"namespace", func(c *containercollection.Container) { c.K8s.Namespace = "test" }, false},
+		{"pod", func(c *containercollection.Container) { c.K8s.PodName = "test" }, false},
+		{"container name", func(c *containercollection.Container) { c.K8s.ContainerName = "test" }, false},
+		{"UID", func(c *containercollection.Container) { c.K8s.PodUID = "test-uid" }, false},
+		{"labels", func(c *containercollection.Container) { c.K8s.PodLabels = map[string]string{"app": "test"} }, false},
+		{"sandbox", func(c *containercollection.Container) { c.SandboxId = "test-sandbox" }, false},
+		{"addressable Pod", func(c *containercollection.Container) { c.K8s.Namespace = "test"; c.K8s.PodName = "test" }, true},
+	}
+	log := logger.L()
+	level := log.GetLevel()
+	require.NoError(t, log.SetLevel("error"))
+	t.Cleanup(func() { log.SetWriter(os.Stderr); _ = log.SetLevel(level) })
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			output, err := os.CreateTemp(t.TempDir(), "log")
+			require.NoError(t, err)
+			t.Cleanup(func() { _ = output.Close() })
+			log.SetWriter(output)
+			ctx, cancel := context.WithCancel(t.Context())
+			cancel()
+			cw := &ContainerWatcher{ctx: ctx, cfg: config.Config{NamespaceFilterFile: "namespaces"}, metrics: metricsmanager.NewMetricsMock()}
+			c := &containercollection.Container{}
+			c.Runtime.ContainerID = "runtime-test"
+			c.Runtime.ContainerPID = uint32(os.Getpid())
+			tc.metadata(c)
+			cw.containerCallbackAsync(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: c})
+			data, err := os.ReadFile(output.Name())
+			require.NoError(t, err)
+			if tc.wantError {
+				assert.Contains(t, string(data), "error getting shared watched container data")
+			} else {
+				assert.Empty(t, string(data), "standalone registration must not try to acquire Pod metadata")
+			}
+		})
+	}
+}

pkg/networkstream/v1/network_stream.go

@@ -144,7 +144,7 @@ func (ns *NetworkStream) ContainerCallback(notif containercollection.PubSubEvent
 		// The container is announced now, so silence no longer means it is gone.
 		delete(ns.unannouncedEntities, containerID)
 		ns.eventsStorageMutex.Unlock()
-		if ns.k8sObjectCache != nil && !utils.IsHostContainer(notif.Container) {
+		if ns.k8sObjectCache != nil && !utils.IsHostContainer(notif.Container) && utils.HasKubernetesPodIdentity(notif.Container) {
 			go ns.enrichWorkloadDetails(notif.Container.Runtime.ContainerID)
 		}
 	case containercollection.EventTypeRemoveContainer:

pkg/networkstream/v1/network_stream_standalone_test.go

@@ -0,0 +1,53 @@
+package networkstream
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	"github.com/kubescape/node-agent/pkg/hostidentity"
+	"github.com/kubescape/node-agent/pkg/objectcache"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type standaloneSharedDataProbe struct {
+	objectcache.K8sObjectCacheMock
+	reads atomic.Int32
+}
+
+func (p *standaloneSharedDataProbe) GetSharedContainerData(string) *objectcache.WatchedContainerData {
+	p.reads.Add(1)
+	// A ready sentinel prevents a broken bypass from leaving a retry goroutine
+	// behind and proves that enrichment was actually invoked.
+	return hostidentity.BuildHostWatchedContainerData("test-node")
+}
+
+func TestContainerCallback_StandaloneNetworkIdentityDoesNotWaitForPodData(t *testing.T) {
+	for _, tc := range []struct{ name, namespace, pod string }{
+		{"standalone", "", ""}, {"namespace-only", "test", ""}, {"pod-only", "", "test"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ns := newHostRegressionStream(t, "test-node")
+			probe := &standaloneSharedDataProbe{}
+			ns.k8sObjectCache = probe
+			event := addContainerEvent("runtime-test", "runtime-name", tc.namespace, tc.pod)
+			ns.ContainerCallback(event)
+			time.Sleep(100 * time.Millisecond)
+			assert.Zero(t, probe.reads.Load(), "standalone traffic must not start a Pod-data wait")
+			ns.eventsStorageMutex.RLock()
+			entity, exists := ns.networkEventsStorage.Entities["runtime-test"]
+			ns.eventsStorageMutex.RUnlock()
+			require.True(t, exists, "standalone traffic remains visible")
+			assert.Equal(t, "runtime-name", entity.ContainerName)
+			assert.Empty(t, entity.WorkloadName)
+			event.Type = containercollection.EventTypeRemoveContainer
+			ns.ContainerCallback(event)
+			ns.eventsStorageMutex.RLock()
+			_, exists = ns.networkEventsStorage.Entities["runtime-test"]
+			ns.eventsStorageMutex.RUnlock()
+			assert.False(t, exists, "standalone removal remains visible")
+		})
+	}
+}

pkg/rulemanager/containercallbacks.go

@@ -55,6 +55,14 @@ func (rm *RuleManager) ContainerCallback(notif containercollection.PubSubEvent)
 	}
 
 	k8sContainerID := utils.CreateK8sContainerID(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.ContainerName)
+	if !utils.IsHostContainer(notif.Container) && !utils.HasKubernetesPodIdentity(notif.Container) {
+		// Empty Kubernetes identities must not merge independent runtime
+		// registrations or let one removal stop another container's monitor.
+		k8sContainerID = "standalone:" + notif.Container.Runtime.ContainerID
+	}
+	if original, ok := rm.standaloneRuntimeKeys.Load(notif.Container.Runtime.ContainerID); ok {
+		k8sContainerID = original
+	}
 
 	switch notif.Type {
 	case containercollection.EventTypeAddContainer:
@@ -90,6 +98,9 @@ func (rm *RuleManager) ContainerCallback(notif containercollection.PubSubEvent)
 		}
 
 		rm.trackedContainers.Add(k8sContainerID)
+		if !utils.IsHostContainer(notif.Container) && !utils.HasKubernetesPodIdentity(notif.Container) {
+			rm.standaloneRuntimeKeys.Set(notif.Container.Runtime.ContainerID, k8sContainerID)
+		}
 		done := make(chan struct{})
 		rm.trackedContainerDone.Set(k8sContainerID, done)
 		shim, err := utils.GetProcessStat(int(notif.Container.ContainerPid()))
@@ -110,6 +121,7 @@ func (rm *RuleManager) ContainerCallback(notif containercollection.PubSubEvent)
 		}
 
 		rm.trackedContainers.Remove(k8sContainerID)
+		rm.standaloneRuntimeKeys.Delete(notif.Container.Runtime.ContainerID)
 		if done, ok := rm.trackedContainerDone.Load(k8sContainerID); ok {
 			close(done)
 			rm.trackedContainerDone.Delete(k8sContainerID)

pkg/rulemanager/containercallbacks_standalone_test.go

@@ -0,0 +1,155 @@
+package rulemanager
+
+import (
+	"context"
+	"os"
+	"sync/atomic"
+	"testing"
+
+	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	"github.com/kubescape/node-agent/pkg/contextdetection"
+	"github.com/kubescape/node-agent/pkg/objectcache"
+	objectcachev1 "github.com/kubescape/node-agent/pkg/objectcache/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type sharedDataProbe struct {
+	objectcachev1.RuleObjectCacheMock
+	reads atomic.Int32
+}
+
+func (c *sharedDataProbe) K8sObjectCache() objectcache.K8sObjectCache { return c }
+
+func (c *sharedDataProbe) GetSharedContainerData(string) *objectcache.WatchedContainerData {
+	c.reads.Add(1)
+	return &objectcache.WatchedContainerData{Wlid: "wlid://cluster-test/namespace-test/pod-test"}
+}
+
+func TestStartRuleManager_StandaloneSkipsPodData(t *testing.T) {
+	cases := []struct {
+		name      string
+		metadata  func(*containercollection.Container)
+		wantReads int32
+	}{
+		{"standalone", func(*containercollection.Container) {}, 0},
+		{"namespace only", func(c *containercollection.Container) { c.K8s.Namespace = "test" }, 0},
+		{"pod only", func(c *containercollection.Container) { c.K8s.PodName = "test" }, 0},
+		{"container name only", func(c *containercollection.Container) { c.K8s.ContainerName = "test" }, 0},
+		{"pod UID only", func(c *containercollection.Container) { c.K8s.PodUID = "test-uid" }, 0},
+		{"pod labels only", func(c *containercollection.Container) { c.K8s.PodLabels = map[string]string{"app": "test"} }, 0},
+		{"sandbox only", func(c *containercollection.Container) { c.SandboxId = "test-sandbox" }, 0},
+		{"addressable Pod without container name", func(c *containercollection.Container) { c.K8s.Namespace = "test"; c.K8s.PodName = "test" }, 1},
+		{"pod", func(c *containercollection.Container) {
+			c.K8s.Namespace = "test"
+			c.K8s.PodName = "test"
+			c.K8s.ContainerName = "app"
+		}, 1},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			cache := &sharedDataProbe{}
+			rm := newTestRuleManager(t.Context())
+			rm.objectCache = cache
+			container := &containercollection.Container{}
+			container.Runtime.ContainerID = "standalone-test"
+			container.Runtime.ContainerPID = uint32(os.Getpid())
+			tc.metadata(container)
+			done := make(chan struct{})
+			close(done)
+			rm.startRuleManager(container, "test-key", done)
+			assert.Equal(t, tc.wantReads, cache.reads.Load(), "only Kubernetes containers need Pod metadata")
+			wlid, bound := rm.podToWlid.Load(container.K8s.Namespace + "/" + container.K8s.PodName)
+			assert.Equal(t, tc.wantReads != 0, bound, "standalone containers must not create a synthetic Pod binding")
+			if bound {
+				assert.Equal(t, "wlid://cluster-test/namespace-test/pod-test", wlid)
+			}
+		})
+	}
+}
+
+func TestContainerCallback_StandaloneRegistrationsAreIndependent(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	t.Cleanup(cancel)
+	rm := newTestRuleManager(ctx)
+	rm.cfg.NamespaceName = "kubescape"
+	rm.objectCache = &sharedDataProbe{}
+	rm.mntnsRegistry = contextdetection.NewMntnsRegistry()
+	containers := []*containercollection.Container{{Mntns: 101}, {Mntns: 102}}
+	for i, c := range containers {
+		c.Runtime.ContainerID = []string{"runtime-a", "runtime-b"}[i]
+		c.Runtime.ContainerPID = uint32(os.Getpid())
+		rm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: c})
+	}
+	require.Equal(t, 2, rm.trackedContainers.Cardinality(), "empty Kubernetes fields must not merge independent runtime registrations")
+	for _, c := range containers {
+		info, ok := rm.mntnsRegistry.Lookup(c.Mntns)
+		require.True(t, ok)
+		assert.Equal(t, contextdetection.Standalone, info.Context())
+		pid, ok := rm.containerIdToPid.Load(c.Runtime.ContainerID)
+		require.True(t, ok)
+		assert.Equal(t, c.ContainerPid(), pid)
+	}
+	firstKey := "standalone:" + containers[0].Runtime.ContainerID
+	secondKey := "standalone:" + containers[1].Runtime.ContainerID
+	firstDone, ok := rm.trackedContainerDone.Load(firstKey)
+	require.True(t, ok)
+	secondDone, ok := rm.trackedContainerDone.Load(secondKey)
+	require.True(t, ok)
+	rm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeRemoveContainer, Container: containers[0]})
+	select {
+	case <-firstDone:
+	default:
+		t.Fatal("removed registration still running")
+	}
+	select {
+	case <-secondDone:
+		t.Fatal("sibling registration was stopped")
+	default:
+	}
+	_, ok = rm.mntnsRegistry.Lookup(containers[0].Mntns)
+	assert.False(t, ok)
+	assert.True(t, rm.trackedContainers.Contains(secondKey))
+	rm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: containers[0]})
+	newDone, ok := rm.trackedContainerDone.Load(firstKey)
+	require.True(t, ok)
+	assert.NotEqual(t, firstDone, newDone)
+	select {
+	case <-newDone:
+		t.Fatal("new registration inherits a closed channel")
+	default:
+	}
+	for _, c := range containers {
+		rm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeRemoveContainer, Container: c})
+	}
+	assert.Zero(t, rm.trackedContainers.Cardinality())
+}
+
+func TestContainerCallback_StandaloneRemovalRetainsRegistrationIdentity(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	t.Cleanup(cancel)
+	rm := newTestRuleManager(ctx)
+	rm.cfg.NamespaceName = "kubescape"
+	rm.objectCache = &sharedDataProbe{}
+	rm.mntnsRegistry = contextdetection.NewMntnsRegistry()
+	c := &containercollection.Container{Mntns: 103}
+	c.Runtime.ContainerID = "runtime-late-enrichment"
+	c.Runtime.ContainerPID = uint32(os.Getpid())
+	rm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeAddContainer, Container: c})
+	key := "standalone:" + c.Runtime.ContainerID
+	done, ok := rm.trackedContainerDone.Load(key)
+	require.True(t, ok)
+	// Runtime metadata may be enriched after the original registration.
+	removed := &containercollection.Container{Mntns: c.Mntns}
+	removed.Runtime = c.Runtime
+	removed.K8s.Namespace, removed.K8s.PodName, removed.K8s.ContainerName = "test", "test-pod", "app"
+	rm.ContainerCallback(containercollection.PubSubEvent{Type: containercollection.EventTypeRemoveContainer, Container: removed})
+	select {
+	case <-done:
+	default:
+		t.Fatal("late enrichment stranded the original runtime registration")
+	}
+	assert.Zero(t, rm.trackedContainers.Cardinality())
+	_, retained := rm.standaloneRuntimeKeys.Load(c.Runtime.ContainerID)
+	assert.False(t, retained, "removed runtime key must not leak")
+}

pkg/rulemanager/rule_manager.go

@@ -46,29 +46,30 @@ import (
 )
 
 type RuleManager struct {
-	cfg                  config.Config
-	ruleBindingCache     bindingcache.RuleBindingCache
-	trackedContainers    mapset.Set[string]                  // key is k8sContainerID
-	trackedContainerDone maps.SafeMap[string, chan struct{}] // key is k8sContainerID; closed when that specific registration is removed
-	k8sClient            k8sclient.K8sClientInterface
-	ctx                  context.Context
-	objectCache          objectcache.ObjectCache
-	exporter             exporters.Exporter
-	metrics              metricsmanager.MetricsManager
-	podToWlid            maps.SafeMap[string, string] // key is namespace/podName
-	containerIdToShimPid maps.SafeMap[string, uint32]
-	containerIdToPid     maps.SafeMap[string, uint32]
-	enricher             types.Enricher
-	processManager       processtree.ProcessTreeManager
-	celEvaluator         cel.RuleEvaluator
-	ruleCooldown         *rulecooldown.RuleCooldown
-	adapterFactory       *ruleadapters.EventRuleAdapterFactory
-	ruleFailureCreator   ruleadapters.RuleFailureCreatorInterface
-	rulePolicyValidator  *RulePolicyValidator
-	mntnsRegistry        contextdetection.Registry
-	detectorManager      *detectors.DetectorManager
-	alertLogDedup        *expirable.LRU[string, struct{}]
-	alertLogDedupMu      sync.Mutex
+	cfg                   config.Config
+	ruleBindingCache      bindingcache.RuleBindingCache
+	trackedContainers     mapset.Set[string]                  // key is k8sContainerID
+	trackedContainerDone  maps.SafeMap[string, chan struct{}] // key is k8sContainerID; closed when that specific registration is removed
+	standaloneRuntimeKeys maps.SafeMap[string, string]        // preserves registration identity across later metadata enrichment
+	k8sClient             k8sclient.K8sClientInterface
+	ctx                   context.Context
+	objectCache           objectcache.ObjectCache
+	exporter              exporters.Exporter
+	metrics               metricsmanager.MetricsManager
+	podToWlid             maps.SafeMap[string, string] // key is namespace/podName
+	containerIdToShimPid  maps.SafeMap[string, uint32]
+	containerIdToPid      maps.SafeMap[string, uint32]
+	enricher              types.Enricher
+	processManager        processtree.ProcessTreeManager
+	celEvaluator          cel.RuleEvaluator
+	ruleCooldown          *rulecooldown.RuleCooldown
+	adapterFactory        *ruleadapters.EventRuleAdapterFactory
+	ruleFailureCreator    ruleadapters.RuleFailureCreatorInterface
+	rulePolicyValidator   *RulePolicyValidator
+	mntnsRegistry         contextdetection.Registry
+	detectorManager       *detectors.DetectorManager
+	alertLogDedup         *expirable.LRU[string, struct{}]
+	alertLogDedupMu       sync.Mutex
 }
 
 var _ RuleManagerClient = (*RuleManager)(nil)
@@ -218,12 +219,12 @@ func (rm *RuleManager) recompileProjectionSpec() {
 }
 
 func (rm *RuleManager) startRuleManager(container *containercollection.Container, k8sContainerID string, done <-chan struct{}) {
-	if utils.IsHostContainer(container) {
-		logger.L().Debug("RuleManager - skipping shared data wait for host container",
+	if utils.IsHostContainer(container) || !utils.HasKubernetesPodIdentity(container) {
+		logger.L().Debug("RuleManager - skipping Pod data wait for non-Kubernetes container",
 			helpers.String("container ID", container.Runtime.ContainerID))
-		// Skip podToWlid and shim PID setup for host containers as they don't have K8s metadata
+		// Keep host and standalone monitoring without inventing a Pod binding.
 		if err := rm.monitorContainer(container, k8sContainerID, done); err != nil {
-			logger.L().Debug("RuleManager - stop monitor on host container",
+			logger.L().Debug("RuleManager - stop monitor on non-Kubernetes container",
 				helpers.String("reason", err.Error()),
 				helpers.String("container ID", container.Runtime.ContainerID),
 				helpers.String("k8s container id", k8sContainerID))