split ContainerCallback into sync and async parts

pkg/applicationprofilemanager/v1/applicationprofile_manager.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/cenkalti/backoff/v5"
 	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/gammazero/workerpool"
 	"github.com/goradd/maps"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	"github.com/kubescape/go-logger"
@@ -72,6 +73,7 @@ type ApplicationProfileManager struct {
 	seccompManager           seccompmanager.SeccompManagerClient
 	enricher                 applicationprofilemanager.Enricher
 	ruleCache                rulebindingmanager.RuleBindingCache
+	pool                     *workerpool.WorkerPool
 }
 
 var _ applicationprofilemanager.ApplicationProfileManagerClient = (*ApplicationProfileManager)(nil)
@@ -91,6 +93,7 @@ func CreateApplicationProfileManager(ctx context.Context, cfg config.Config, clu
 		seccompManager:          seccompManager,
 		enricher:                enricher,
 		ruleCache:               ruleCache,
+		pool:                    workerpool.New(1),
 	}, nil
 }
 
@@ -698,7 +701,12 @@ func (am *ApplicationProfileManager) ContainerCallback(notif containercollection
 	if am.cfg.SkipNamespace(notif.Container.K8s.Namespace) {
 		return
 	}
+	am.pool.Submit(func() {
+		am.containerCallbackAsync(notif)
+	})
+}
 
+func (am *ApplicationProfileManager) containerCallbackAsync(notif containercollection.PubSubEvent) {
 	k8sContainerID := utils.CreateK8sContainerID(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.ContainerName)
 	ctx, span := otel.Tracer("").Start(am.ctx, "ApplicationProfileManager.ContainerCallback", trace.WithAttributes(attribute.String("containerID", notif.Container.Runtime.ContainerID), attribute.String("k8s workload", k8sContainerID)))
 	defer span.End()

pkg/containerwatcher/v1/container_watcher.go

@@ -6,14 +6,14 @@ import (
 	"reflect"
 	"time"
 
-	toptracer "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/top/ebpf/tracer"
-
 	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/gammazero/workerpool"
 	"github.com/goradd/maps"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	containerutilsTypes "github.com/inspektor-gadget/inspektor-gadget/pkg/container-utils/types"
 	tracerseccomp "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/advise/seccomp/tracer"
 	"github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/top"
+	toptracer "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/top/ebpf/tracer"
 	toptypes "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/top/ebpf/types"
 	tracercapabilities "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/trace/capabilities/tracer"
 	tracercapabilitiestype "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/trace/capabilities/types"
@@ -167,6 +167,7 @@ type IGContainerWatcher struct {
 	httpWorkerChan         chan *tracerhttptype.Event
 	iouringWorkerChan      chan *traceriouringtype.Event
 
+	pool            *workerpool.WorkerPool
 	objectCache     objectcache.ObjectCache
 	ruleManagedPods mapset.Set[string] // list of pods to track based on rules
 	metrics         metricsmanager.MetricsManager
@@ -537,6 +538,7 @@ func CreateIGContainerWatcher(cfg config.Config,
 		thirdPartyContainerReceivers: mapset.NewSet[containerwatcher.ContainerReceiver](),
 		thirdPartyEnricher:           thirdPartyEnricher,
 		processManager:               processManager,
+		pool:                         workerpool.New(1),
 		objectCache:                  objectCache,
 	}, nil
 }

pkg/containerwatcher/v1/container_watcher_private.go

@@ -39,7 +39,12 @@ func (ch *IGContainerWatcher) containerCallback(notif containercollection.PubSub
 		}
 		return
 	}
+	ch.pool.Submit(func() {
+		ch.containerCallbackAsync(notif)
+	})
+}
 
+func (ch *IGContainerWatcher) containerCallbackAsync(notif containercollection.PubSubEvent) {
 	k8sContainerID := utils.CreateK8sContainerID(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.ContainerName)
 
 	switch notif.Type {

pkg/malwaremanager/v1/malware_manager.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cenkalti/backoff/v5"
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/dustin/go-humanize"
+	"github.com/gammazero/workerpool"
 	"github.com/goradd/maps"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	tracerexectype "github.com/inspektor-gadget/inspektor-gadget/pkg/gadgets/trace/exec/types"
@@ -45,6 +46,7 @@ type MalwareManager struct {
 	cfg                  config.Config
 	containerIdToShimPid maps.SafeMap[string, uint32]
 	k8sObjectCache       objectcache.K8sObjectCache
+	pool                 *workerpool.WorkerPool
 }
 
 var _ malwaremanager.MalwareManagerClient = (*MalwareManager)(nil)
@@ -72,6 +74,7 @@ func CreateMalwareManager(cfg config.Config, k8sClient k8sclient.K8sClientInterf
 		clusterName:     clusterName,
 		metrics:         prometheusExporter,
 		k8sObjectCache:  k8sObjectCache,
+		pool:            workerpool.New(1),
 	}, nil
 }
 
@@ -80,7 +83,12 @@ func (mm *MalwareManager) ContainerCallback(notif containercollection.PubSubEven
 	if mm.cfg.SkipNamespace(notif.Container.K8s.Namespace) {
 		return
 	}
+	mm.pool.Submit(func() {
+		mm.containerCallbackAsync(notif)
+	})
+}
 
+func (mm *MalwareManager) containerCallbackAsync(notif containercollection.PubSubEvent) {
 	t := time.NewTicker(utils.AddJitter(mm.cfg.InitialDelay, mm.cfg.MaxJitterPercentage))
 
 	switch notif.Type {

pkg/networkmanager/v2/network_manager.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/cenkalti/backoff/v5"
 	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/gammazero/workerpool"
 	"github.com/google/uuid"
 	"github.com/goradd/maps"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
@@ -48,6 +49,7 @@ type NetworkManager struct {
 	k8sObjectCache           objectcache.K8sObjectCache
 	storageClient            storage.StorageClient
 	dnsResolverClient        dnsmanager.DNSResolver
+	pool                     *workerpool.WorkerPool
 }
 
 var _ networkmanager.NetworkManagerClient = (*NetworkManager)(nil)
@@ -65,6 +67,7 @@ func CreateNetworkManager(ctx context.Context, cfg config.Config, clusterName st
 		trackedContainers:       mapset.NewSet[string](),
 		removedContainers:       mapset.NewSet[string](),
 		droppedEventsContainers: mapset.NewSet[string](),
+		pool:                    workerpool.New(1),
 	}
 }
 
@@ -464,7 +467,12 @@ func (nm *NetworkManager) ContainerCallback(notif containercollection.PubSubEven
 	if nm.cfg.SkipNamespace(notif.Container.K8s.Namespace) {
 		return
 	}
+	nm.pool.Submit(func() {
+		nm.containerCallbackAsync(notif)
+	})
+}
 
+func (nm *NetworkManager) containerCallbackAsync(notif containercollection.PubSubEvent) {
 	k8sContainerID := utils.CreateK8sContainerID(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.ContainerName)
 	ctx, span := otel.Tracer("").Start(nm.ctx, "NetworkManager.ContainerCallback", trace.WithAttributes(attribute.String("containerID", notif.Container.Runtime.ContainerID), attribute.String("k8s workload", k8sContainerID)))
 	defer span.End()

pkg/processmanager/v1/process_manager.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
+	"github.com/gammazero/workerpool"
 	"github.com/goradd/maps"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	"github.com/prometheus/procfs"
@@ -29,11 +30,13 @@ type ProcessManager struct {
 	processTree          maps.SafeMap[uint32, apitypes.Process]
 	// For testing purposes we allow to override the function that gets process info from /proc.
 	getProcessFromProc func(pid int) (apitypes.Process, error)
+	pool               *workerpool.WorkerPool
 }
 
 func CreateProcessManager(ctx context.Context) *ProcessManager {
 	pm := &ProcessManager{
 		getProcessFromProc: getProcessFromProc,
+		pool:               workerpool.New(1),
 	}
 	go pm.startCleanupRoutine(ctx)
 	return pm
@@ -116,6 +119,12 @@ func (p *ProcessManager) isDescendantOfShim(pid uint32, ppid uint32, shimPIDs ma
 // For new containers, it identifies the container's shim process and adds it to the tracking system.
 // For removed containers, it cleans up the associated processes from the process tree.
 func (p *ProcessManager) ContainerCallback(notif containercollection.PubSubEvent) {
+	p.pool.Submit(func() {
+		p.containerCallbackAsync(notif)
+	})
+}
+
+func (p *ProcessManager) containerCallbackAsync(notif containercollection.PubSubEvent) {
 	containerID := notif.Container.Runtime.BasicRuntimeMetadata.ContainerID
 
 	switch notif.Type {

pkg/processmanager/v1/process_manager_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sync"
 	"testing"
+	"time"
 
 	apitypes "github.com/armosec/armoapi-go/armotypes"
 	"github.com/stretchr/testify/assert"
@@ -82,6 +83,7 @@ func TestProcessManagerBasics(t *testing.T) {
 			},
 		},
 	})
+	time.Sleep(time.Second)
 
 	// Verify shim was recorded
 	assert.True(t, pm.containerIdToShimPid.Has(containerID))
@@ -322,6 +324,7 @@ func TestContainerRemoval(t *testing.T) {
 			},
 		},
 	})
+	time.Sleep(time.Second)
 
 	// Verify all processes were removed
 	for _, proc := range processes {
@@ -379,6 +382,7 @@ func TestMultipleContainers(t *testing.T) {
 		pm.ReportEvent(utils.ExecveEventType, event1)
 		pm.ReportEvent(utils.ExecveEventType, event2)
 	}
+	time.Sleep(time.Second)
 
 	// Verify each container's processes
 	for _, c := range containers {
@@ -410,6 +414,7 @@ func TestMultipleContainers(t *testing.T) {
 			},
 		},
 	})
+	time.Sleep(time.Second)
 
 	// Verify first container's processes are gone
 	assert.False(t, pm.processTree.Has(containers[0].containerPID))

pkg/rulemanager/v1/rule_manager.go

@@ -12,6 +12,7 @@ import (
 	backoffv5 "github.com/cenkalti/backoff/v5"
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/dustin/go-humanize"
+	"github.com/gammazero/workerpool"
 	"github.com/goradd/maps"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	eventtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types"
@@ -56,6 +57,7 @@ type RuleManager struct {
 	enricher             ruleenginetypes.Enricher
 	processManager       processmanager.ProcessManagerClient
 	dnsManager           dnsmanager.DNSResolver
+	pool                 *workerpool.WorkerPool
 }
 
 var _ rulemanager.RuleManagerClient = (*RuleManager)(nil)
@@ -75,6 +77,7 @@ func CreateRuleManager(ctx context.Context, cfg config.Config, k8sClient k8sclie
 		enricher:          enricher,
 		processManager:    processManager,
 		dnsManager:        dnsManager,
+		pool:              workerpool.New(1),
 	}, nil
 }
 
@@ -184,7 +187,12 @@ func (rm *RuleManager) ContainerCallback(notif containercollection.PubSubEvent)
 	if rm.cfg.SkipNamespace(notif.Container.K8s.Namespace) {
 		return
 	}
+	rm.pool.Submit(func() {
+		rm.containerCallbackAsync(notif)
+	})
+}
 
+func (rm *RuleManager) containerCallbackAsync(notif containercollection.PubSubEvent) {
 	k8sContainerID := utils.CreateK8sContainerID(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.ContainerName)
 
 	switch notif.Type {