diff --git a/pkg/containerwatcher/v2/containercallback.go b/pkg/containerwatcher/v2/containercallback.go index bdb115255a..afd110ae73 100644 --- a/pkg/containerwatcher/v2/containercallback.go +++ b/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, diff --git a/pkg/containerwatcher/v2/containercallback_standalone_test.go b/pkg/containerwatcher/v2/containercallback_standalone_test.go new file mode 100644 index 0000000000..08b765da00 --- /dev/null +++ b/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") + } + }) + } +} diff --git a/pkg/networkstream/v1/network_stream.go b/pkg/networkstream/v1/network_stream.go index f9db7d3e9b..fa5bbc02fb 100644 --- a/pkg/networkstream/v1/network_stream.go +++ b/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: diff --git a/pkg/networkstream/v1/network_stream_standalone_test.go b/pkg/networkstream/v1/network_stream_standalone_test.go new file mode 100644 index 0000000000..a4586cc6a5 --- /dev/null +++ b/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") + }) + } +} diff --git a/pkg/rulemanager/containercallbacks.go b/pkg/rulemanager/containercallbacks.go index ebce6fda8e..54a78874cc 100644 --- a/pkg/rulemanager/containercallbacks.go +++ b/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) diff --git a/pkg/rulemanager/containercallbacks_standalone_test.go b/pkg/rulemanager/containercallbacks_standalone_test.go new file mode 100644 index 0000000000..5ad0048017 --- /dev/null +++ b/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") +} diff --git a/pkg/rulemanager/rule_manager.go b/pkg/rulemanager/rule_manager.go index cd6a1511a5..e124cafda7 100644 --- a/pkg/rulemanager/rule_manager.go +++ b/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)) diff --git a/pkg/utils/container.go b/pkg/utils/container.go index b192fed038..fe17fcd6ba 100644 --- a/pkg/utils/container.go +++ b/pkg/utils/container.go @@ -17,6 +17,16 @@ func IsHostContainer(container *containercollection.Container) bool { return container.Runtime.ContainerPID == 1 || container.Runtime.ContainerID == armotypes.HostContainerID } +// HasKubernetesPodIdentity reports whether a container identifies an addressable +// Kubernetes Pod. Other runtime metadata cannot be used for a Pod lookup. +// Container-name enrichment is not required to acquire the Pod's shared data. +func HasKubernetesPodIdentity(container *containercollection.Container) bool { + if container == nil { + return false + } + return container.K8s.Namespace != "" && container.K8s.PodName != "" +} + // IsHost reports whether containerID identifies the virtual host // pseudo-container (armotypes.HostContainerID). This is the single choke // point for the "is this the host" check wherever only a containerID string