Feature/refactor container watcher

.github/workflows/component-tests.yaml

@@ -61,7 +61,8 @@ jobs:
           Test_20_AlertOnPartialThenLearnProcessTest,
           Test_21_AlertOnPartialThenLearnNetworkTest,
           Test_22_AlertOnPartialNetworkProfileTest,
-          Test_23_RuleCooldownTest
+          Test_23_RuleCooldownTest,
+          Test_24_ProcessTreeDepthTest
         ]
     steps:
       - name: Checkout code

Makefile

@@ -12,4 +12,4 @@ docker-build:
 	docker buildx build --platform linux/amd64 -t $(IMAGE):$(TAG) -f $(DOCKERFILE_PATH) --load .
 
 docker-push:
-	docker push $(IMAGE):$(TAG)
+	docker push $(IMAGE):$(TAG)

cmd/main.go

@@ -24,7 +24,7 @@ import (
 	"github.com/kubescape/node-agent/pkg/config"
 	"github.com/kubescape/node-agent/pkg/containerprofilemanager"
 	containerprofilemanagerv1 "github.com/kubescape/node-agent/pkg/containerprofilemanager/v1"
-	"github.com/kubescape/node-agent/pkg/containerwatcher/v1"
+	containerwatcherv2 "github.com/kubescape/node-agent/pkg/containerwatcher/v2"
 	"github.com/kubescape/node-agent/pkg/dnsmanager"
 	"github.com/kubescape/node-agent/pkg/exporters"
 	"github.com/kubescape/node-agent/pkg/healthmanager"
@@ -42,8 +42,9 @@ import (
 	"github.com/kubescape/node-agent/pkg/objectcache/k8scache"
 	"github.com/kubescape/node-agent/pkg/objectcache/networkneighborhoodcache"
 	objectcachev1 "github.com/kubescape/node-agent/pkg/objectcache/v1"
-	"github.com/kubescape/node-agent/pkg/processmanager"
-	processmanagerv1 "github.com/kubescape/node-agent/pkg/processmanager/v1"
+	processtree "github.com/kubescape/node-agent/pkg/processtree"
+	containerprocesstree "github.com/kubescape/node-agent/pkg/processtree/container"
+	processtreecreator "github.com/kubescape/node-agent/pkg/processtree/creator"
 	rulebinding "github.com/kubescape/node-agent/pkg/rulebindingmanager"
 	rulebindingcachev1 "github.com/kubescape/node-agent/pkg/rulebindingmanager/cache"
 	"github.com/kubescape/node-agent/pkg/rulemanager"
@@ -218,11 +219,27 @@ func main() {
 	}
 
 	var ruleManager rulemanager.RuleManagerClient
-	var processManager processmanager.ProcessManagerClient
+	var processTreeManager processtree.ProcessTreeManager
 	var objCache objectcache.ObjectCache
 	var ruleBindingNotify chan rulebinding.RuleBindingNotify
 	var cloudMetadata *apitypes.CloudMetadata
 
+	// Create the container process tree
+	containerProcessTree := containerprocesstree.NewContainerProcessTree()
+
+	// Create the process tree creator
+	processTreeCreator := processtreecreator.NewProcessTreeCreator(containerProcessTree, cfg)
+
+	// Create the process tree manager
+	processTreeManager = processtree.NewProcessTreeManager(
+		processTreeCreator,
+		containerProcessTree,
+		cfg,
+	)
+
+	// Start the process tree manager to activate the exit cleanup manager
+	processTreeManager.Start()
+
 	if cfg.EnableRuntimeDetection || cfg.EnableMalwareDetection {
 		cloudMetadata, err = cloudmetadata.GetCloudMetadata(ctx, k8sClient, cfg.NodeName)
 		if err != nil {
@@ -231,9 +248,6 @@ func main() {
 	}
 
 	if cfg.EnableRuntimeDetection {
-		// create the process manager
-		processManager = processmanagerv1.CreateProcessManager(ctx)
-
 		// create exporter
 		exporter := exporters.InitExporters(cfg.Exporters, clusterData.ClusterName, cfg.NodeName, cloudMetadata)
 		dWatcher.AddAdaptor(ruleBindingCache)
@@ -255,7 +269,7 @@ func main() {
 		ruleCooldown := rulecooldown.NewRuleCooldown(cfg.RuleCoolDown)
 
 		// create runtimeDetection managers
-		ruleManager, err = rulemanagerv1.CreateRuleManager(ctx, cfg, k8sClient, ruleBindingCache, objCache, exporter, prometheusExporter, cfg.NodeName, clusterData.ClusterName, processManager, dnsResolver, nil, ruleCooldown)
+		ruleManager, err = rulemanagerv1.CreateRuleManager(ctx, cfg, k8sClient, ruleBindingCache, objCache, exporter, prometheusExporter, cfg.NodeName, clusterData.ClusterName, processTreeManager, dnsResolver, nil, ruleCooldown)
 		if err != nil {
 			logger.L().Ctx(ctx).Fatal("error creating RuleManager", helpers.Error(err))
 		}
@@ -267,7 +281,6 @@ func main() {
 		dc := &objectcache.DnsCacheMock{}
 		objCache = objectcachev1.NewObjectCache(k8sObjectCache, apc, nnc, dc)
 		ruleBindingNotify = make(chan rulebinding.RuleBindingNotify, 1)
-		processManager = processmanager.CreateProcessManagerMock()
 	}
 
 	// Create the node profile manager
@@ -282,7 +295,7 @@ func main() {
 	// Create the network streaming manager
 	var networkStreamClient networkstream.NetworkStreamClient
 	if cfg.EnableNetworkStreaming {
-		networkStreamClient, err = networkstreamv1.NewNetworkStream(ctx, cfg, k8sObjectCache, dnsResolver, cfg.NodeName, nil, false, processManager)
+		networkStreamClient, err = networkstreamv1.NewNetworkStream(ctx, cfg, k8sObjectCache, dnsResolver, cfg.NodeName, nil, false, processTreeManager)
 		if err != nil {
 			logger.L().Ctx(ctx).Fatal("error creating NetworkManager", helpers.Error(err))
 		}
@@ -327,10 +340,10 @@ func main() {
 	}
 
 	// Create the container handler
-	mainHandler, err := containerwatcher.CreateIGContainerWatcher(cfg, containerProfileManager, k8sClient,
+	mainHandler, err := containerwatcherv2.CreateIGContainerWatcher(cfg, containerProfileManager, k8sClient,
 		igK8sClient, dnsManagerClient, prometheusExporter, ruleManager,
 		malwareManager, sbomManager, &ruleBindingNotify, igK8sClient.RuntimeConfig, nil, nil,
-		processManager, clusterData.ClusterName, objCache, networkStreamClient)
+		processTreeManager, clusterData.ClusterName, objCache, networkStreamClient, containerProcessTree)
 	if err != nil {
 		logger.L().Ctx(ctx).Fatal("error creating the container watcher", helpers.Error(err))
 	}

go.mod

@@ -245,6 +245,7 @@ require (
 	github.com/ncw/directio v1.0.5 // indirect
 	github.com/nwaples/rardecode v1.1.3 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
+	github.com/oleiade/lane/v2 v2.0.0 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/olvrng/ujson v1.1.0 // indirect
 	github.com/opencontainers/runtime-spec v1.2.1 // indirect
@@ -283,6 +284,7 @@ require (
 	github.com/spf13/cast v1.7.1 // indirect
 	github.com/spf13/cobra v1.9.1 // indirect
 	github.com/spf13/pflag v1.0.6 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/stripe/stripe-go/v74 v74.30.0 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/sylabs/sif/v2 v2.21.1 // indirect

go.sum

@@ -835,6 +835,8 @@ github.com/nwaples/rardecode v1.1.3 h1:cWCaZwfM5H7nAD6PyEdcVnczzV8i/JtotnyW/dD9l
 github.com/nwaples/rardecode v1.1.3/go.mod h1:5DzqNKiOdpKKBH87u8VlvAnPZMXcGRhxWkRpHbbfGS0=
 github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
+github.com/oleiade/lane/v2 v2.0.0 h1:XW/ex/Inr+bPkLd3O240xrFOhUkTd4Wy176+Gv0E3Qw=
+github.com/oleiade/lane/v2 v2.0.0/go.mod h1:i5FBPFAYSWCgLh58UkUGCChjcCzef/MI7PlQm2TKCeg=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
 github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
 github.com/olvrng/ujson v1.1.0 h1:8xVUzVlqwdMVWh5d1UHBtLQ1D50nxoPuPEq9Wozs8oA=

pkg/config/config.go

@@ -49,11 +49,13 @@ type Config struct {
 	KubernetesMode                 bool                            `mapstructure:"kubernetesMode"`
 	NetworkStreamingInterval       time.Duration                   `mapstructure:"networkStreamingInterval"`
 	WorkerPoolSize                 int                             `mapstructure:"workerPoolSize"`
+	EventBatchSize                 int                             `mapstructure:"eventBatchSize"`
 	TestMode                       bool                            `mapstructure:"testMode"`
 	ExcludeJsonPaths               []string                        `mapstructure:"excludeJsonPaths"`
 	ProfilesCacheRefreshRate       time.Duration                   `mapstructure:"profilesCacheRefreshRate"`
 	RuleCoolDown                   rulecooldown.RuleCooldownConfig `mapstructure:"ruleCooldown"`
 	EnablePartialProfileGeneration bool                            `mapstructure:"partialProfileGenerationEnabled"`
+	ProcfsScanInterval             time.Duration                   `mapstructure:"procfsScanInterval"`
 }
 
 // LoadConfig reads configuration from file or environment variables.
@@ -78,7 +80,8 @@ func LoadConfig(path string) (Config, error) {
 	viper.SetDefault("networkStreamingEnabled", false)
 	viper.SetDefault("kubernetesMode", true)
 	viper.SetDefault("networkStreamingInterval", 2*time.Minute)
-	viper.SetDefault("workerPoolSize", 10)
+	viper.SetDefault("workerPoolSize", 2000) // Increased from 50 to handle higher event volumes
+	viper.SetDefault("eventBatchSize", 1000)
 	viper.SetDefault("testMode", false)
 	viper.SetDefault("enableEmbeddedSBOMs", false)
 	viper.SetDefault("profilesCacheRefreshRate", 1*time.Minute)
@@ -87,6 +90,7 @@ func LoadConfig(path string) (Config, error) {
 	viper.SetDefault("ruleCooldown.ruleCooldownOnProfileFailure", true)
 	viper.SetDefault("ruleCooldown.ruleCooldownMaxSize", 10000)
 	viper.SetDefault("partialProfileGenerationEnabled", true)
+	viper.SetDefault("procfsScanInterval", 5*time.Second)
 	viper.AutomaticEnv()
 
 	err := viper.ReadInConfig()

pkg/containerprofilemanager/v1/event_reporting.go

@@ -46,7 +46,6 @@ func (cpm *ContainerProfileManager) ReportFileExec(containerID string, event eve
 		if data.execs == nil {
 			data.execs = &maps.SafeMap[string, []string]{}
 		}
-
 		path := event.Comm
 		if len(event.Args) > 0 {
 			path = event.Args[0]

pkg/containerwatcher/container_watcher_interface.go

@@ -6,10 +6,11 @@ import (
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	"github.com/inspektor-gadget/inspektor-gadget/pkg/socketenricher"
 	tracercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/tracer-collection"
+	"github.com/kubescape/node-agent/pkg/ebpf/events"
 	"github.com/kubescape/node-agent/pkg/utils"
 )
 
-type ResultCallback func(utils.EnrichEvent)
+type ResultCallback func(utils.K8sEvent, string, uint32)
 
 type ContainerWatcher interface {
 	Ready() bool
@@ -35,10 +36,14 @@ type EventReceiver interface {
 	ReportEvent(eventType utils.EventType, event utils.K8sEvent)
 }
 
+type EnrichedEventReceiver interface {
+	ReportEnrichedEvent(enrichedEvent *events.EnrichedEvent)
+}
+
 type ContainerReceiver interface {
 	ContainerCallback(notif containercollection.PubSubEvent)
 }
 
 type TaskBasedEnricher interface {
-	SubmitEnrichmentTask(event utils.EnrichEvent, syscalls []uint64, callback ResultCallback)
+	SubmitEnrichmentTask(event utils.EnrichEvent, syscalls []uint64, callback ResultCallback, containerID string, processID uint32)
 }

pkg/containerwatcher/tracer_factory_interface.go

@@ -0,0 +1,12 @@
+package containerwatcher
+
+// TracerRegistrer defines the interface for registering tracers
+type TracerRegistrer interface {
+	RegisterTracer(tracer TracerInterface)
+}
+
+// TracerFactoryInterface defines the interface for creating tracers
+type TracerFactoryInterface interface {
+	CreateAllTracers(manager TracerRegistrer)
+	GetThirdPartyTracers() []CustomTracer
+}

pkg/containerwatcher/tracer_interface.go

@@ -0,0 +1,25 @@
+package containerwatcher
+
+import (
+	"context"
+
+	"github.com/kubescape/node-agent/pkg/utils"
+)
+
+// TracerInterface defines the common interface for all eBPF tracers
+type TracerInterface interface {
+	// Start initializes and starts the tracer
+	Start(ctx context.Context) error
+
+	// Stop gracefully stops the tracer
+	Stop() error
+
+	// GetName returns the unique name of the tracer
+	GetName() string
+
+	// GetEventType returns the event type this tracer produces
+	GetEventType() utils.EventType
+
+	// IsEnabled checks if this tracer should be enabled based on configuration
+	IsEnabled(cfg interface{}) bool
+}