Fix: Add ContainerID to virtual host container

cmd/main.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 	"syscall"
 
-	apitypes "github.com/armosec/armoapi-go/armotypes"
+	"github.com/armosec/armoapi-go/armotypes"
 	utilsmetadata "github.com/armosec/utils-k8s-go/armometadata"
 	"github.com/cilium/ebpf/rlimit"
 	mapset "github.com/deckarep/golang-set/v2"
@@ -35,7 +35,7 @@ import (
 	"github.com/kubescape/node-agent/pkg/exporters"
 	"github.com/kubescape/node-agent/pkg/fimmanager"
 	"github.com/kubescape/node-agent/pkg/healthmanager"
-	hostsensormanager "github.com/kubescape/node-agent/pkg/hostsensormanager"
+	"github.com/kubescape/node-agent/pkg/hostsensormanager"
 	"github.com/kubescape/node-agent/pkg/malwaremanager"
 	malwaremanagerv1 "github.com/kubescape/node-agent/pkg/malwaremanager/v1"
 	"github.com/kubescape/node-agent/pkg/metricsmanager"
@@ -260,7 +260,7 @@ func main() {
 	var processTreeManager processtree.ProcessTreeManager
 	var objCache objectcache.ObjectCache
 	var ruleBindingNotify chan rulebinding.RuleBindingNotify
-	var cloudMetadata *apitypes.CloudMetadata
+	var cloudMetadata *armotypes.CloudMetadata
 
 	// Create the container process tree
 	containerProcessTree := containerprocesstree.NewContainerProcessTree()
@@ -287,7 +287,7 @@ func main() {
 
 	if cfg.EnableRuntimeDetection {
 		// create exporter
-		exporter := exporters.InitExporters(cfg.Exporters, clusterData.ClusterName, cfg.NodeName, cloudMetadata, clusterUID)
+		exporter := exporters.InitExporters(cfg.Exporters, clusterData.ClusterName, cfg.NodeName, cloudMetadata, clusterUID, armotypes.AlertSourcePlatformK8sAgent)
 		dWatcher.AddAdaptor(ruleBindingCache)
 
 		ruleBindingNotify = make(chan rulebinding.RuleBindingNotify, 100)
@@ -353,7 +353,7 @@ func main() {
 	var malwareManager malwaremanager.MalwareManagerClient
 	if cfg.EnableMalwareDetection {
 		// create exporter
-		exporter := exporters.InitExporters(cfg.Exporters, clusterData.ClusterName, cfg.NodeName, cloudMetadata, clusterUID)
+		exporter := exporters.InitExporters(cfg.Exporters, clusterData.ClusterName, cfg.NodeName, cloudMetadata, clusterUID, armotypes.AlertSourcePlatformK8sAgent)
 		malwareManager, err = malwaremanagerv1.CreateMalwareManager(cfg, k8sClient, cfg.NodeName, clusterData.ClusterName, exporter, prometheusExporter, k8sObjectCache)
 		if err != nil {
 			logger.L().Ctx(ctx).Fatal("error creating MalwareManager", helpers.Error(err))
@@ -395,7 +395,7 @@ func main() {
 	if cfg.EnableFIM {
 		// Initialize FIM-specific exporters
 		fimExportersConfig := cfg.FIM.GetFIMExportersConfig()
-		fimExporter := exporters.InitExporters(fimExportersConfig, clusterData.ClusterName, cfg.NodeName, cloudMetadata, clusterUID)
+		fimExporter := exporters.InitExporters(fimExportersConfig, clusterData.ClusterName, cfg.NodeName, cloudMetadata, clusterUID, armotypes.AlertSourcePlatformK8sAgent)
 
 		fimManager, err = fimmanager.NewFIMManager(cfg, clusterData.ClusterName, fimExporter, cloudMetadata)
 		if err != nil {

docs/ALERT_BULKING.md

@@ -102,9 +102,9 @@ type AlertBulkManager struct {
 type containerBulk struct {
     sync.Mutex
     containerID       string
-    alerts            []apitypes.RuntimeAlert
-    processMap        map[uint32]*apitypes.Process  // PID → Process
-    rootProcess       *apitypes.Process             // Container init
+    alerts            []armotypes.RuntimeAlert
+    processMap        map[uint32]*armotypes.Process  // PID → Process
+    rootProcess       *armotypes.Process             // Container init
     cloudServices     []string
     firstAlertTime    time.Time
     maxSize           int
@@ -113,8 +113,8 @@ type containerBulk struct {
 
 type bulkQueueItem struct {
     containerID   string
-    alerts        []apitypes.RuntimeAlert
-    processTree   apitypes.ProcessTree
+    alerts        []armotypes.RuntimeAlert
+    processTree   armotypes.ProcessTree
     cloudServices []string
     retryCount    int
     enqueuedAt    time.Time

docs/PROCESS_TREE_CHAIN_OPTIMIZATION.md

@@ -53,8 +53,8 @@ type containerBulk struct {
     // ... other fields ...
 
     // NEW: Maintain incrementally instead of rebuilding
-    processMap  map[uint32]*apitypes.Process  // PID -> Process for O(1) lookup
-    rootProcess *apitypes.Process             // Root of merged tree
+    processMap  map[uint32]*armotypes.Process  // PID -> Process for O(1) lookup
+    rootProcess *armotypes.Process             // Root of merged tree
 }
 ```
 
@@ -69,9 +69,9 @@ Instead of recursively merging arbitrary trees, we:
    - If new: **create and link** to parent in tree
 
 ```go
-func (cb *containerBulk) mergeProcessChain(chain *apitypes.Process) {
+func (cb *containerBulk) mergeProcessChain(chain *armotypes.Process) {
     if cb.processMap == nil {
-        cb.processMap = make(map[uint32]*apitypes.Process)
+        cb.processMap = make(map[uint32]*armotypes.Process)
     }
 
     // Flatten chain: O(k) where k = chain length
@@ -92,7 +92,7 @@ func (cb *containerBulk) mergeProcessChain(chain *apitypes.Process) {
             }
 
             if parent, ok := cb.processMap[newNode.PPID]; ok {
-                parent.ChildrenMap[apitypes.CommPID{PID: newNode.PID}] = newNode
+                parent.ChildrenMap[armotypes.CommPID{PID: newNode.PID}] = newNode
             }
         }
     }
@@ -207,4 +207,3 @@ By recognizing that runtime alerts provide **chains, not arbitrary trees**, we a
 - ✅ **Thread-safe** (race detector clean)
 
 This optimization addresses matthyx's performance concern and makes the alert bulking feature production-ready.
-

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/Masterminds/semver/v3 v3.4.0
 	github.com/anchore/syft v1.32.0
 	github.com/aquilax/truncate v1.0.0
-	github.com/armosec/armoapi-go v0.0.678
+	github.com/armosec/armoapi-go v0.0.682
 	github.com/armosec/utils-k8s-go v0.0.35
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cenkalti/backoff/v4 v4.3.0

go.sum

@@ -761,8 +761,8 @@ github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
-github.com/armosec/armoapi-go v0.0.678 h1:trltW2KxO5OqrB2zHIPx8jqe82AFvdRryo2XgBEyJw8=
-github.com/armosec/armoapi-go v0.0.678/go.mod h1:9jAH0g8ZsryhiBDd/aNMX4+n10bGwTx/doWCyyjSxts=
+github.com/armosec/armoapi-go v0.0.682 h1:H/bMUS3ESNrcun16iS9ficCkE1mWyOIkZJXokauuI6U=
+github.com/armosec/armoapi-go v0.0.682/go.mod h1:9jAH0g8ZsryhiBDd/aNMX4+n10bGwTx/doWCyyjSxts=
 github.com/armosec/gojay v1.2.17 h1:VSkLBQzD1c2V+FMtlGFKqWXNsdNvIKygTKJI9ysY8eM=
 github.com/armosec/gojay v1.2.17/go.mod h1:vuvX3DlY0nbVrJ0qCklSS733AWMoQboq3cFyuQW9ybc=
 github.com/armosec/utils-go v0.0.58 h1:g9RnRkxZAmzTfPe2ruMo2OXSYLwVSegQSkSavOfmaIE=

pkg/containerprofilemanager/v1/containerprofile_manager.go

@@ -32,6 +32,8 @@ type ContainerEntry struct {
 	mu   sync.RWMutex
 	// ready channel is used to signal when the container entry is fully initialized
 	ready chan struct{}
+	// readyOnce ensures the ready channel is closed exactly once
+	readyOnce sync.Once
 }
 
 // containerData contains all the monitored data for a single container

pkg/containerprofilemanager/v1/lifecycle.go

@@ -17,6 +17,9 @@ import (
 func (cpm *ContainerProfileManager) ContainerCallback(notif containercollection.PubSubEvent) {
 	switch notif.Type {
 	case containercollection.EventTypeAddContainer:
+		if utils.IsHostContainer(notif.Container) {
+			return
+		}
 		if cpm.cfg.IgnoreContainer(notif.Container.K8s.Namespace, notif.Container.K8s.PodName, notif.Container.K8s.PodLabels) {
 			return
 		}
@@ -52,13 +55,23 @@ func (cpm *ContainerProfileManager) addContainerWithTimeout(container *container
 	case err := <-done:
 		if err != nil {
 			logger.L().Error("failed to add container to the container profile manager", helpers.Error(err))
+			// Close ready channel and remove entry on error
+			entry.readyOnce.Do(func() {
+				close(entry.ready)
+			})
+			cpm.removeContainerEntry(containerID)
 		}
 	case <-ctx.Done():
 		logger.L().Error("timeout while adding container to the container profile manager",
 			helpers.String("containerID", container.Runtime.ContainerID),
 			helpers.String("containerName", container.Runtime.ContainerName),
 			helpers.String("podName", container.K8s.PodName),
 			helpers.String("namespace", container.K8s.Namespace))
+		// Close ready channel and remove entry on timeout
+		entry.readyOnce.Do(func() {
+			close(entry.ready)
+		})
+		cpm.removeContainerEntry(containerID)
 	}
 }
 
@@ -69,7 +82,12 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection.
 	// Wait for shared container data with timeout
 	sharedData, err := cpm.waitForSharedContainerData(containerID, ctx)
 	if err != nil {
-		// Remove the container entry and all stacked events if we fail
+		// Close ready channel and remove the container entry if we fail
+		if entry, exists := cpm.getContainerEntry(containerID); exists {
+			entry.readyOnce.Do(func() {
+				close(entry.ready)
+			})
+		}
 		cpm.removeContainerEntry(containerID)
 		return fmt.Errorf("failed to get shared data for container %s: %w", containerID, err)
 	}
@@ -82,6 +100,12 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection.
 			helpers.String("podName", container.K8s.PodName),
 			helpers.String("namespace", container.K8s.Namespace),
 			helpers.String("userDefinedProfile", sharedData.UserDefinedProfile))
+		// Close ready channel before removing entry
+		if entry, exists := cpm.getContainerEntry(containerID); exists {
+			entry.readyOnce.Do(func() {
+				close(entry.ready)
+			})
+		}
 		cpm.removeContainerEntry(containerID)
 		return nil
 	}
@@ -93,6 +117,12 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection.
 			helpers.String("containerName", container.Runtime.ContainerName),
 			helpers.String("podName", container.K8s.PodName),
 			helpers.String("namespace", container.K8s.Namespace))
+		// Close ready channel before removing entry
+		if entry, exists := cpm.getContainerEntry(containerID); exists {
+			entry.readyOnce.Do(func() {
+				close(entry.ready)
+			})
+		}
 		cpm.removeContainerEntry(containerID)
 		return nil
 	}
@@ -103,6 +133,12 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection.
 			helpers.String("containerName", container.Runtime.ContainerName),
 			helpers.String("podName", container.K8s.PodName),
 			helpers.String("namespace", container.K8s.Namespace))
+		// Close ready channel before removing entry
+		if entry, exists := cpm.getContainerEntry(containerID); exists {
+			entry.readyOnce.Do(func() {
+				close(entry.ready)
+			})
+		}
 		cpm.removeContainerEntry(containerID)
 		return nil
 	}
@@ -135,7 +171,9 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection.
 	go cpm.startContainerMonitoring(container, sharedData)
 
 	// Signal that the container entry is ready
-	close(entry.ready)
+	entry.readyOnce.Do(func() {
+		close(entry.ready)
+	})
 
 	logger.L().Debug("container added to container profile manager",
 		helpers.String("containerID", containerID),

pkg/containerwatcher/v2/container_watcher_collection.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/armosec/armoapi-go/armotypes"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	containerutils "github.com/inspektor-gadget/inspektor-gadget/pkg/container-utils"
 	"github.com/inspektor-gadget/inspektor-gadget/pkg/operators/socketenricher"
@@ -186,6 +187,7 @@ func GetHostAsContainer() (*containercollection.Container, error) {
 	return &containercollection.Container{
 		Runtime: containercollection.RuntimeMetadata{
 			BasicRuntimeMetadata: eventtypes.BasicRuntimeMetadata{
+				ContainerID:  armotypes.HostContainerID,
 				ContainerPID: uint32(hostInitPID),
 			},
 		},

pkg/containerwatcher/v2/containercallback.go

@@ -60,8 +60,7 @@ func (cw *ContainerWatcher) containerCallbackAsync(notif containercollection.Pub
 			helpers.String("ContainerImageName", notif.Container.Runtime.ContainerImageName))
 		cw.metrics.ReportContainerStart()
 
-		// Skip shared data setup for virtual host container (identified by ContainerPID == 1)
-		if notif.Container.Runtime.ContainerPID == 1 {
+		if utils.IsHostContainer(notif.Container) {
 			logger.L().Debug("ContainerWatcher.containerCallback - skipping shared data setup for virtual host container")
 			return
 		}

pkg/containerwatcher/v2/event_enricher.go

@@ -3,7 +3,7 @@ package containerwatcher
 import (
 	"fmt"
 
-	apitypes "github.com/armosec/armoapi-go/armotypes"
+	"github.com/armosec/armoapi-go/armotypes"
 	"github.com/kubescape/go-logger"
 	"github.com/kubescape/go-logger/helpers"
 	ebpfevents "github.com/kubescape/node-agent/pkg/ebpf/events"
@@ -26,7 +26,7 @@ func NewEventEnricher(
 }
 
 func (ee *EventEnricher) EnrichEvents(entry EventEntry) *ebpfevents.EnrichedEvent {
-	var processTree apitypes.Process
+	var processTree armotypes.Process
 
 	eventType := entry.EventType
 	event := entry.Event

pkg/containerwatcher/v2/tracers/procfs.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/armosec/armoapi-go/armotypes"
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
 	tracercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/tracer-collection"
 	"github.com/inspektor-gadget/inspektor-gadget/pkg/types"
@@ -183,7 +184,7 @@ func (pt *ProcfsTracer) handleProcfsEvent(event conversion.ProcessEvent) {
 	if container != nil {
 		procfsEvent.ContainerID = container.Runtime.ContainerID
 	} else {
-		procfsEvent.ContainerID = "host"
+		procfsEvent.ContainerID = armotypes.HostContainerID
 	}
 	processID := event.PID
 	containerID := procfsEvent.ContainerID

pkg/ebpf/events/enriched_event.go

@@ -3,13 +3,13 @@ package events
 import (
 	"time"
 
-	apitypes "github.com/armosec/armoapi-go/armotypes"
+	"github.com/armosec/armoapi-go/armotypes"
 	"github.com/kubescape/node-agent/pkg/contextdetection"
 	"github.com/kubescape/node-agent/pkg/utils"
 )
 
 // NewEnrichedEvent creates a new enriched event
-func NewEnrichedEvent(event utils.K8sEvent, timestamp time.Time, containerID string, processTree apitypes.Process) *EnrichedEvent {
+func NewEnrichedEvent(event utils.K8sEvent, timestamp time.Time, containerID string, processTree armotypes.Process) *EnrichedEvent {
 	return &EnrichedEvent{
 		Event:       event,
 		Timestamp:   timestamp,
@@ -19,18 +19,18 @@ func NewEnrichedEvent(event utils.K8sEvent, timestamp time.Time, containerID str
 }
 
 type EnrichedEvent struct {
-	Event              utils.K8sEvent
-	Timestamp          time.Time
-	ContainerID        string
-	ProcessTree        apitypes.Process
-	PID                uint32
-	PPID               uint32
+	Event       utils.K8sEvent
+	Timestamp   time.Time
+	ContainerID string
+	ProcessTree armotypes.Process
+	PID         uint32
+	PPID        uint32
 	// SourceContext holds the context information for this event (K8s, Host, or Standalone).
 	// This is populated during event enrichment if the feature is enabled.
 	// May be nil for legacy K8s-only events or when feature is disabled.
-	SourceContext      contextdetection.ContextInfo
+	SourceContext contextdetection.ContextInfo
 	// MountNamespaceID is the mount namespace ID from the container.
 	// This uniquely identifies the container/host and is used for context lookup.
 	// May be 0 if unavailable.
-	MountNamespaceID   uint64
+	MountNamespaceID uint64
 }