fix(rulemanager): scope monitorContainer exit to its own registration

docs/features/rulemanager-monitor-container-lifecycle.md

@@ -0,0 +1,92 @@
+# RuleManager per-container monitor goroutine lifecycle
+
+`RuleManager.monitorContainer` (`pkg/rulemanager/containercallbacks.go`) is the
+goroutine `ContainerCallback` spawns (via `startRuleManager`) for every
+`EventTypeAddContainer` notification. It used to have no scoped way to know
+when *its own* container registration ended.
+
+## The bug
+
+The old exit condition polled shared mutable state on a 5s ticker instead of
+listening for its own registration's removal:
+
+```go
+case <-syscallTicker.C:
+    if !rm.trackedContainers.Contains(k8sContainerID) {
+        return nil
+    }
+```
+
+`ContainerCallback` dedups `EventTypeAddContainer` by checking
+`trackedContainers.Contains(k8sContainerID)` and spawns a new
+`monitorContainer` goroutine whenever that check is false — including right
+after a `EventTypeRemoveContainer` for the same ID clears it. If a container
+is removed and re-added faster than the 5s ticker, the sequence is:
+
+1. Add(A) → tracked, goroutine **G1** spawned.
+2. Remove(A) → untracked.
+3. Add(A) again, before G1's next tick → tracked again, a **second**
+   goroutine **G2** spawned for the same ID.
+4. G1's ticker fires and finds `trackedContainers.Contains("A") == true`
+   (re-added at step 3) — indistinguishable from "still my container" — so it
+   loops forever instead of exiting.
+
+Every such remove→add cycle inside one tick period leaked one goroutine
+permanently (only full process shutdown, via `rm.ctx.Done()`, ever collected
+them).
+
+## The fix
+
+Each `ContainerCallback` `EventTypeAddContainer` now creates its own
+`chan struct{}` ("done") and stores it in a new
+`RuleManager.trackedContainerDone` map keyed by `k8sContainerID`, alongside
+`trackedContainers`. That channel is passed directly into `startRuleManager`
+→ `monitorContainer` as a parameter, not looked up by ID:
+
+```go
+select {
+case <-rm.ctx.Done():
+    return nil
+case <-done:
+    return nil
+}
+```
+
+`EventTypeRemoveContainer` closes and deletes the current registration's
+channel. Because `monitorContainer` holds its `done` channel via closure
+rather than re-reading a shared map/set by ID at wake time, only the removal
+that closes *that specific channel object* can stop it — a sibling
+registration's channel (or a later re-Add's fresh channel) has no effect on
+it. This closes the race structurally instead of narrowing the polling
+window.
+
+One side effect: `monitorContainer` no longer polls on a timer at all, so the
+now-unused `syscallPeriod` ticker and its `container.Mntns == 0` debug check
+were removed along with it.
+
+## Observable changes
+
+- A container remove→add cycle no longer leaks a `monitorContainer`
+  goroutine, regardless of how fast the cycle repeats.
+- `monitorContainer` exits immediately when its container is removed, instead
+  of up to `syscallPeriod` (5s) later.
+
+## Testing
+
+`pkg/rulemanager/containercallbacks_test.go` pins:
+
+- `TestMonitorContainer_ExitsOnDoneChannel` — exits when its own channel
+  closes.
+- `TestMonitorContainer_IgnoresSiblingDoneChannel` — does **not** exit when
+  an unrelated channel closes.
+- `TestContainerCallback_FastRemoveAddDoesNotLeakGoroutine` — reproduces the
+  Add→Remove→Add race end-to-end through `ContainerCallback` and asserts the
+  first registration's channel is closed (so its goroutine can exit) while
+  the second (current) registration's channel stays open and distinct from
+  the first.
+
+Run with the race detector:
+
+```bash
+go test -race ./pkg/rulemanager/... -run 'TestMonitorContainer|TestContainerCallback'
+```

pkg/rulemanager/containercallbacks.go

@@ -15,30 +15,31 @@ import (
 	"github.com/kubescape/node-agent/pkg/utils"
 )
 
-func (rm *RuleManager) monitorContainer(container *containercollection.Container, k8sContainerID string) error {
+// monitorContainer blocks for the lifetime of one container registration. It
+// exits only in response to rm.ctx being cancelled (process shutdown) or to
+// done being closed by the specific ContainerCallback removal that created
+// it. It intentionally does not re-derive liveness by polling shared mutable
+// state (e.g. rm.trackedContainers) on a timer: a container ID that is
+// removed and re-added faster than a poll interval would look "tracked
+// again" to a stale goroutine from the earlier registration, letting it
+// survive alongside the new one. done is scoped per-registration so only the
+// registration that owns it is ever told to stop.
+func (rm *RuleManager) monitorContainer(container *containercollection.Container, k8sContainerID string, done <-chan struct{}) error {
 	logger.L().Debug("RuleManager - start monitor on container",
 		helpers.String("container ID", container.Runtime.ContainerID),
 		helpers.String("k8s container id", k8sContainerID))
 
-	syscallTicker := time.NewTicker(syscallPeriod)
-
-	for {
-		select {
-		case <-rm.ctx.Done():
-			logger.L().Debug("RuleManager - stop monitor on container",
-				helpers.String("container ID", container.Runtime.ContainerID),
-				helpers.String("k8s container id", k8sContainerID))
-			return nil
-		case <-syscallTicker.C:
-			if container.Mntns == 0 {
-				logger.L().Debug("RuleManager - mount namespace ID is not set", helpers.String("container ID", container.Runtime.ContainerID))
-			}
-
-			if !rm.trackedContainers.Contains(k8sContainerID) {
-				logger.L().Debug("RuleManager - container is not tracked", helpers.String("container ID", container.Runtime.ContainerID))
-				return nil
-			}
-		}
+	select {
+	case <-rm.ctx.Done():
+		logger.L().Debug("RuleManager - stop monitor on container",
+			helpers.String("container ID", container.Runtime.ContainerID),
+			helpers.String("k8s container id", k8sContainerID))
+		return nil
+	case <-done:
+		logger.L().Debug("RuleManager - container is not tracked",
+			helpers.String("container ID", container.Runtime.ContainerID),
+			helpers.String("k8s container id", k8sContainerID))
+		return nil
 	}
 }
 
@@ -84,14 +85,16 @@ func (rm *RuleManager) ContainerCallback(notif containercollection.PubSubEvent)
 		}
 
 		rm.trackedContainers.Add(k8sContainerID)
+		done := make(chan struct{})
+		rm.trackedContainerDone.Set(k8sContainerID, done)
 		shim, err := utils.GetProcessStat(int(notif.Container.ContainerPid()))
 		if err != nil {
 			logger.L().Warning("RuleManager - failed to get shim process", helpers.Error(err))
 		} else {
 			rm.containerIdToShimPid.Set(notif.Container.Runtime.ContainerID, uint32(shim.PPID))
 		}
 		rm.containerIdToPid.Set(notif.Container.Runtime.ContainerID, notif.Container.ContainerPid())
-		go rm.startRuleManager(notif.Container, k8sContainerID)
+		go rm.startRuleManager(notif.Container, k8sContainerID, done)
 	case containercollection.EventTypeRemoveContainer:
 		logger.L().Debug("RuleManager - remove container",
 			helpers.String("container ID", notif.Container.Runtime.ContainerID),
@@ -102,6 +105,10 @@ func (rm *RuleManager) ContainerCallback(notif containercollection.PubSubEvent)
 		}
 
 		rm.trackedContainers.Remove(k8sContainerID)
+		if done, ok := rm.trackedContainerDone.Load(k8sContainerID); ok {
+			close(done)
+			rm.trackedContainerDone.Delete(k8sContainerID)
+		}
 		namespace := notif.Container.K8s.Namespace
 		podName := notif.Container.K8s.PodName
 		podID := utils.CreateK8sPodID(namespace, podName)

pkg/rulemanager/containercallbacks_test.go

@@ -0,0 +1,173 @@
+package rulemanager
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/armosec/armoapi-go/armotypes"
+	mapset "github.com/deckarep/golang-set/v2"
+	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	"github.com/kubescape/node-agent/pkg/config"
+	"github.com/kubescape/node-agent/pkg/contextdetection/detectors"
+	"github.com/kubescape/node-agent/pkg/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newTestRuleManager builds a minimal RuleManager sufficient to exercise
+// ContainerCallback/monitorContainer without a cluster.
+func newTestRuleManager(ctx context.Context) *RuleManager {
+	return &RuleManager{
+		ctx:               ctx,
+		cfg:               config.Config{},
+		trackedContainers: mapset.NewSet[string](),
+		detectorManager:   detectors.NewDetectorManager(nil),
+	}
+}
+
+// hostContainer builds a container that IsHostContainer treats as a host
+// container, so startRuleManager takes the monitorContainer-only path and
+// never touches objectCache/podToWlid, keeping the test self-contained.
+func hostContainer(namespace, podName string) *containercollection.Container {
+	c := &containercollection.Container{}
+	c.Runtime.ContainerID = armotypes.HostContainerID
+	c.K8s.Namespace = namespace
+	c.K8s.PodName = podName
+	c.K8s.ContainerName = podName
+	return c
+}
+
+// waitForCondition polls until cond returns true or the timeout elapses.
+func waitForCondition(t *testing.T, timeout time.Duration, cond func() bool) bool {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return true
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return cond()
+}
+
+// TestMonitorContainer_ExitsOnDoneChannel pins that monitorContainer exits
+// promptly when its own done channel is closed.
+func TestMonitorContainer_ExitsOnDoneChannel(t *testing.T) {
+	rm := newTestRuleManager(context.Background())
+	container := hostContainer("ns", "pod-a")
+	done := make(chan struct{})
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- rm.monitorContainer(container, "ns/pod-a/pod-a", done)
+	}()
+
+	close(done)
+
+	select {
+	case err := <-errCh:
+		assert.NoError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("monitorContainer did not exit after its done channel was closed")
+	}
+}
+
+// TestMonitorContainer_IgnoresSiblingDoneChannel pins the fix for
+// https://github.com/kubescape/node-agent/issues/919: a monitorContainer
+// goroutine must exit only in response to ITS OWN done channel, never a
+// different registration's channel for the same container ID.
+func TestMonitorContainer_IgnoresSiblingDoneChannel(t *testing.T) {
+	rm := newTestRuleManager(context.Background())
+	container := hostContainer("ns", "pod-a")
+	ownDone := make(chan struct{})
+	siblingDone := make(chan struct{})
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- rm.monitorContainer(container, "ns/pod-a/pod-a", ownDone)
+	}()
+
+	// Closing an unrelated channel must not affect this goroutine.
+	close(siblingDone)
+
+	select {
+	case <-errCh:
+		t.Fatal("monitorContainer exited in response to a sibling's done channel")
+	case <-time.After(100 * time.Millisecond):
+		// still running, as expected
+	}
+
+	close(ownDone)
+
+	select {
+	case err := <-errCh:
+		assert.NoError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("monitorContainer did not exit after its own done channel was closed")
+	}
+}
+
+// TestContainerCallback_FastRemoveAddDoesNotLeakGoroutine reproduces the
+// race from issue #919: Add -> Remove -> Add for the same k8sContainerID
+// happening faster than any polling interval. Before the fix, the first
+// monitorContainer goroutine re-derived liveness from shared mutable state
+// (rm.trackedContainers) on a timer, so it would see the ID "tracked again"
+// after the second Add and never exit, permanently leaking one goroutine per
+// such cycle. After the fix, each monitorContainer is scoped to its own
+// done channel: only the registration that owns a given goroutine can stop
+// it, so the first goroutine exits on the Remove that closed its channel
+// and does not resurrect on the following Add.
+func TestContainerCallback_FastRemoveAddDoesNotLeakGoroutine(t *testing.T) {
+	rm := newTestRuleManager(context.Background())
+	container := hostContainer("ns", "pod-a")
+	k8sContainerID := utils.CreateK8sContainerID("ns", "pod-a", "pod-a")
+
+	rm.ContainerCallback(containercollection.PubSubEvent{
+		Type:      containercollection.EventTypeAddContainer,
+		Container: container,
+	})
+	firstDone, ok := rm.trackedContainerDone.Load(k8sContainerID)
+	require.True(t, ok, "expected a done channel to be registered after Add")
+
+	rm.ContainerCallback(containercollection.PubSubEvent{
+		Type:      containercollection.EventTypeRemoveContainer,
+		Container: container,
+	})
+
+	rm.ContainerCallback(containercollection.PubSubEvent{
+		Type:      containercollection.EventTypeAddContainer,
+		Container: container,
+	})
+	secondDone, ok := rm.trackedContainerDone.Load(k8sContainerID)
+	require.True(t, ok, "expected a done channel to be registered after re-Add")
+	require.NotEqual(t, firstDone, secondDone, "re-Add must create a fresh, distinct done channel")
+
+	// The first registration's channel must already be closed by the Remove
+	// that preceded the re-Add -- this is what lets its monitorContainer
+	// goroutine exit instead of leaking.
+	select {
+	case _, open := <-firstDone:
+		assert.False(t, open, "first registration's done channel should be closed")
+	default:
+		t.Fatal("first registration's done channel is not closed")
+	}
+
+	// The second (current) registration's channel must still be open.
+	ok = waitForCondition(t, time.Second, func() bool {
+		select {
+		case <-secondDone:
+			return false
+		default:
+			return true
+		}
+	})
+	assert.True(t, ok, "current registration's done channel must remain open")
+
+	rm.ContainerCallback(containercollection.PubSubEvent{
+		Type:      containercollection.EventTypeRemoveContainer,
+		Container: container,
+	})
+	_, ok = rm.trackedContainerDone.Load(k8sContainerID)
+	assert.False(t, ok, "done channel entry should be removed after the matching Remove")
+}

pkg/rulemanager/rule_manager.go

@@ -45,14 +45,11 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-const (
-	syscallPeriod = 5 * time.Second
-)
-
 type RuleManager struct {
 	cfg                  config.Config
 	ruleBindingCache     bindingcache.RuleBindingCache
-	trackedContainers    mapset.Set[string] // key is k8sContainerID
+	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
@@ -220,12 +217,12 @@ func (rm *RuleManager) recompileProjectionSpec() {
 	rm.objectCache.ContainerProfileCache().SetProjectionSpec(spec)
 }
 
-func (rm *RuleManager) startRuleManager(container *containercollection.Container, k8sContainerID string) {
+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",
 			helpers.String("container ID", container.Runtime.ContainerID))
 		// Skip podToWlid and shim PID setup for host containers as they don't have K8s metadata
-		if err := rm.monitorContainer(container, k8sContainerID); err != nil {
+		if err := rm.monitorContainer(container, k8sContainerID, done); err != nil {
 			logger.L().Debug("RuleManager - stop monitor on host container",
 				helpers.String("reason", err.Error()),
 				helpers.String("container ID", container.Runtime.ContainerID),
@@ -250,7 +247,7 @@ func (rm *RuleManager) startRuleManager(container *containercollection.Container
 		}
 	}
 
-	if err := rm.monitorContainer(container, k8sContainerID); err != nil {
+	if err := rm.monitorContainer(container, k8sContainerID, done); err != nil {
 		logger.L().Debug("RuleManager - stop monitor on container", helpers.String("reason", err.Error()),
 			helpers.String("container ID", container.Runtime.ContainerID),
 			helpers.String("k8s container id", k8sContainerID))