feat(rulemanager): CEL rule state store for cross-event correlation

.github/workflows/component-tests.yaml

@@ -105,6 +105,7 @@ jobs:
           Test_34_NetworkNeighborsCIDRCollapse,
           Test_35_ExecTTYFieldTest,
           Test_36_MultiContainerPerContainerBinding,
+          Test_37_CelStateStoreCorrelation,
           Test_43_RelativeOpenPathResolution,
           Test_48_MultiSubtypeGroupedProfileDocument,
           Test_49_EphemeralContainerFullTreatment

pkg/config/config.go

@@ -17,6 +17,7 @@ import (
 	processtreecreator "github.com/kubescape/node-agent/pkg/processtree/config"
 	"github.com/kubescape/node-agent/pkg/rulemanager/cel/libraries/cache"
 	"github.com/kubescape/node-agent/pkg/rulemanager/rulecooldown"
+	"github.com/kubescape/node-agent/pkg/rulestate"
 	"github.com/spf13/viper"
 )
 
@@ -87,6 +88,7 @@ func NormalizeDNSSuffix(s string) string {
 type Config struct {
 	BlockEvents                    bool                                 `mapstructure:"blockEvents"`
 	CelConfigCache                 cache.FunctionCacheConfig            `mapstructure:"celConfigCache"`
+	CelStateStore                  rulestate.Config                     `mapstructure:"celStateStore"`
 	ContainerEolNotificationBuffer int                                  `mapstructure:"containerEolNotificationBuffer"`
 	DBpf                           bool                                 `mapstructure:"dBpf"`
 	DCapSys                        bool                                 `mapstructure:"dCapSys"`
@@ -258,6 +260,17 @@ func LoadConfigOptional(path string, errNotFound bool) (Config, error) {
 	viper.SetDefault("blockEvents", false)
 	viper.SetDefault("celConfigCache::maxSize", 100000)
 	viper.SetDefault("celConfigCache::ttl", 1*time.Minute)
+
+	// CEL rule state store. maxEntriesForHost is larger than the per-container cap
+	// because the host bucket holds the whole node's process space rather than one
+	// workload, and never receives a container-removal purge -- it relies on TTL.
+	viper.SetDefault("celStateStore::enabled", true)
+	viper.SetDefault("celStateStore::maxSize", 100000)
+	viper.SetDefault("celStateStore::maxEntriesPerContainer", 256)
+	viper.SetDefault("celStateStore::maxEntriesForHost", 4096)
+	viper.SetDefault("celStateStore::maxTtl", 30*time.Minute)
+	viper.SetDefault("celStateStore::sweepInterval", 30*time.Second)
+	viper.SetDefault("celStateStore::ancestorMaxDepth", 8)
 	viper.SetDefault("ignoreRuleBindings", false)
 	viper.SetDefault("namespaceFilterFile", "")
 

pkg/config/config_test.go

@@ -10,6 +10,7 @@ import (
 	processtreecreator "github.com/kubescape/node-agent/pkg/processtree/config"
 	"github.com/kubescape/node-agent/pkg/rulemanager/cel/libraries/cache"
 	"github.com/kubescape/node-agent/pkg/rulemanager/rulecooldown"
+	"github.com/kubescape/node-agent/pkg/rulestate"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -105,6 +106,15 @@ func TestLoadConfig(t *testing.T) {
 					MaxSize: 100000,
 					TTL:     1 * time.Minute,
 				},
+				CelStateStore: rulestate.Config{
+					Enabled:                true,
+					MaxSize:                100000,
+					MaxEntriesPerContainer: 256,
+					MaxEntriesForHost:      4096,
+					MaxTTL:                 30 * time.Minute,
+					SweepInterval:          30 * time.Second,
+					AncestorMaxDepth:       8,
+				},
 				DNSCacheSize:                   50000,
 				ContainerEolNotificationBuffer: 100,
 				FIM: FIMConfig{

pkg/exporters/http_exporter.go

@@ -356,6 +356,7 @@ func (e *HTTPExporter) createRuleAlert(failedRule types.RuleFailure) armotypes.R
 		RuleID:                 failedRule.GetRuleId(),
 		IsTriggerAlert:         failedRule.GetIsTriggerAlert(),
 		HttpRuleAlert:          httpDetails,
+		CorrelationAlert:       failedRule.GetCorrelationAlert(),
 	}
 }
 

pkg/metricsmanager/metrics_manager_interface.go

@@ -86,4 +86,15 @@ type MetricsManager interface {
 
 	// Alert suppression funnel — counts how many alerts were dropped and why.
 	ReportAlertSuppressed(ruleID, reason string)
+
+	// CEL rule state store. Labelled by ruleID only — never by state key, which is
+	// unbounded cardinality.
+	//
+	// ReportStateWriteRejected is the alert-worthy one: it means a rule is being
+	// silently starved of the state it needs to correlate.
+	ReportStateWrite(ruleID, result string)
+	ReportStateWriteRejected(ruleID, reason string)
+	ReportStateExpired(n int)
+	ReportStatePurged(n int)
+	ReportStateEntries(scope string, n int)
 }

pkg/metricsmanager/metrics_manager_mock.go

@@ -109,3 +109,9 @@ func (m *MetricsMock) ObserveSBOMScanDuration(_, _ string, _ time.Duration) {}
 func (m *MetricsMock) ReportSBOMScannerRestart()                            {}
 func (m *MetricsMock) SetSBOMScannerReady(_ bool)                           {}
 func (m *MetricsMock) ReportAlertSuppressed(_, _ string)                    {}
+
+func (m *MetricsMock) ReportStateWrite(_, _ string)         {}
+func (m *MetricsMock) ReportStateWriteRejected(_, _ string) {}
+func (m *MetricsMock) ReportStateExpired(_ int)             {}
+func (m *MetricsMock) ReportStatePurged(_ int)              {}
+func (m *MetricsMock) ReportStateEntries(_ string, _ int)   {}

pkg/metricsmanager/metrics_manager_noop.go

@@ -57,3 +57,9 @@ func (m *MetricsNoop) ObserveSBOMScanDuration(_, _ string, _ time.Duration)
 func (m *MetricsNoop) ReportSBOMScannerRestart()                                          {}
 func (m *MetricsNoop) SetSBOMScannerReady(_ bool)                                         {}
 func (m *MetricsNoop) ReportAlertSuppressed(_, _ string)                                  {}
+
+func (m *MetricsNoop) ReportStateWrite(_, _ string)         {}
+func (m *MetricsNoop) ReportStateWriteRejected(_, _ string) {}
+func (m *MetricsNoop) ReportStateExpired(_ int)             {}
+func (m *MetricsNoop) ReportStatePurged(_ int)              {}
+func (m *MetricsNoop) ReportStateEntries(_ string, _ int)   {}

pkg/metricsmanager/otel/otel_metrics_manager.go

@@ -79,6 +79,13 @@ type OTELMetricsManager struct {
 	// Alert suppression funnel
 	alertSuppressedTotal metric.Int64Counter
 
+	// CEL rule state store
+	stateWritesTotal        metric.Int64Counter
+	stateWriteRejectedTotal metric.Int64Counter
+	stateExpiredTotal       metric.Int64Counter
+	statePurgedTotal        metric.Int64Counter
+	stateEntries            metric.Float64Gauge
+
 	// Live container count — incremented on start, decremented on stop.
 	// Exposed as node_agent.container.count observable gauge.
 	containerCount atomic.Int64
@@ -253,6 +260,16 @@ func NewOTELMetricsManager(ownContainerID, ownPodUID string, hostCgroupMounted b
 
 	m.alertSuppressedTotal = mustCounter("node_agent.alert.suppressed.total",
 		"Total alerts suppressed before delivery, labeled by rule_id and reason")
+	m.stateWritesTotal = mustCounter("node_agent.state.writes.total",
+		"Total CEL rule state entries written, labeled by rule_id")
+	m.stateWriteRejectedTotal = mustCounter("node_agent.state.write.rejected.total",
+		"Total CEL rule state writes rejected; a rule is being starved of the state it needs to correlate")
+	m.stateExpiredTotal = mustCounter("node_agent.state.expired.total",
+		"Total CEL rule state entries reclaimed by TTL expiry")
+	m.statePurgedTotal = mustCounter("node_agent.state.purged.total",
+		"Total CEL rule state entries dropped by scope purge, e.g. container removal")
+	m.stateEntries = mustGauge("node_agent.state.entries",
+		"Current CEL rule state entries, labeled by scope")
 
 	registerResourceMetrics(meter, &m.containerCount, ownContainerID, ownPodUID, hostCgroupMounted)
 
@@ -581,3 +598,27 @@ func (m *OTELMetricsManager) suppressedOption(ruleID, reason string) metric.Meas
 func (m *OTELMetricsManager) ReportAlertSuppressed(ruleID, reason string) {
 	m.alertSuppressedTotal.Add(context.Background(), 1, m.suppressedOption(ruleID, reason))
 }
+
+// The state counters reuse suppressedOption: it caches a (ruleID, reason)
+// attribute set, which is exactly the label pair these need. Labelling by ruleID
+// only is deliberate -- a state key is unbounded cardinality.
+func (m *OTELMetricsManager) ReportStateWrite(ruleID, result string) {
+	m.stateWritesTotal.Add(context.Background(), 1, m.suppressedOption(ruleID, result))
+}
+
+func (m *OTELMetricsManager) ReportStateWriteRejected(ruleID, reason string) {
+	m.stateWriteRejectedTotal.Add(context.Background(), 1, m.suppressedOption(ruleID, reason))
+}
+
+func (m *OTELMetricsManager) ReportStateExpired(n int) {
+	m.stateExpiredTotal.Add(context.Background(), int64(n))
+}
+
+func (m *OTELMetricsManager) ReportStatePurged(n int) {
+	m.statePurgedTotal.Add(context.Background(), int64(n))
+}
+
+func (m *OTELMetricsManager) ReportStateEntries(scope string, n int) {
+	m.stateEntries.Record(context.Background(), float64(n),
+		metric.WithAttributes(attribute.String("scope", scope)))
+}

pkg/objectcache/containerprofilecache/projection_compile_test.go

@@ -3,6 +3,8 @@ package containerprofilecache
 import (
 	"testing"
 
+	"github.com/armosec/armoapi-go/armotypes"
+
 	"github.com/kubescape/node-agent/pkg/objectcache"
 	typesv1 "github.com/kubescape/node-agent/pkg/rulemanager/types/v1"
 	"github.com/stretchr/testify/assert"
@@ -12,8 +14,7 @@ import (
 // makeRule is a helper that builds a Rule with a ProfileDataRequired.
 func makeRule(pdr *typesv1.ProfileDataRequired) typesv1.Rule {
 	return typesv1.Rule{
-		ID:                  "test-rule",
-		ProfileDataRequired: pdr,
+		RuntimeRule: armotypes.RuntimeRule{ID: "test-rule", ProfileDataRequired: pdr},
 	}
 }
 
@@ -64,8 +65,8 @@ func TestCompileSpec_Empty(t *testing.T) {
 // ProfileDataRequired do not contribute to the spec.
 func TestCompileSpec_NilProfileDataRequiredSkipped(t *testing.T) {
 	rules := []typesv1.Rule{
-		{ID: "no-pdr", ProfileDataRequired: nil},
-		{ID: "also-no-pdr", ProfileDataRequired: nil},
+		{RuntimeRule: armotypes.RuntimeRule{ID: "no-pdr", ProfileDataRequired: nil}},
+		{RuntimeRule: armotypes.RuntimeRule{ID: "also-no-pdr", ProfileDataRequired: nil}},
 	}
 	spec := CompileSpec(rules)
 
@@ -91,7 +92,7 @@ func TestCompileSpec_DeterministicHash(t *testing.T) {
 	pdr2 := &typesv1.ProfileDataRequired{
 		Execs: fieldReqAll(),
 	}
-	rule2 := typesv1.Rule{ID: "r2", ProfileDataRequired: pdr2}
+	rule2 := typesv1.Rule{RuntimeRule: armotypes.RuntimeRule{ID: "r2", ProfileDataRequired: pdr2}}
 
 	specAB := CompileSpec([]typesv1.Rule{rule, rule2})
 	specBA := CompileSpec([]typesv1.Rule{rule2, rule})

pkg/objectcache/containerprofilecache/projection_golden_test.go

@@ -8,6 +8,8 @@ import (
 	"sort"
 	"testing"
 
+	"github.com/armosec/armoapi-go/armotypes"
+
 	"github.com/kubescape/node-agent/pkg/objectcache"
 	"github.com/kubescape/node-agent/pkg/objectcache/callstackcache"
 	typesv1 "github.com/kubescape/node-agent/pkg/rulemanager/types/v1"
@@ -255,9 +257,7 @@ func networkProfile() *v1beta1.ContainerProfile {
 func mixedFilterRules() []typesv1.Rule {
 	return []typesv1.Rule{
 		{
-			ID:   "RULE-A",
-			Name: "mixed-file-and-net",
-			ProfileDataRequired: &typesv1.ProfileDataRequired{
+			RuntimeRule: armotypes.RuntimeRule{ID: "RULE-A", Name: "mixed-file-and-net", ProfileDataRequired: &typesv1.ProfileDataRequired{
 				Opens: declaredPatterns(
 					typesv1.PatternObject{Exact: "/etc/passwd"},
 					typesv1.PatternObject{Prefix: "/var/"},
@@ -272,17 +272,15 @@ func mixedFilterRules() []typesv1.Rule {
 				EgressAddresses:  declaredAll(),
 				IngressDomains:   declaredAll(),
 				IngressAddresses: declaredAll(),
-			},
+			}},
 		},
 		{
 			// A second rule narrows capabilities further and adds an opens
 			// prefix, proving the union merge across rules.
-			ID:   "RULE-B",
-			Name: "extra-caps",
-			ProfileDataRequired: &typesv1.ProfileDataRequired{
+			RuntimeRule: armotypes.RuntimeRule{ID: "RULE-B", Name: "extra-caps", ProfileDataRequired: &typesv1.ProfileDataRequired{
 				Capabilities: declaredPatterns(typesv1.PatternObject{Exact: "SYS_PTRACE"}),
 				Opens:        declaredPatterns(typesv1.PatternObject{Prefix: "/data/"}),
-			},
+			}},
 		},
 	}
 }
@@ -292,14 +290,12 @@ func mixedFilterRules() []typesv1.Rule {
 func netAllRules() []typesv1.Rule {
 	return []typesv1.Rule{
 		{
-			ID:   "RULE-NET",
-			Name: "net-all",
-			ProfileDataRequired: &typesv1.ProfileDataRequired{
+			RuntimeRule: armotypes.RuntimeRule{ID: "RULE-NET", Name: "net-all", ProfileDataRequired: &typesv1.ProfileDataRequired{
 				EgressDomains:    declaredAll(),
 				EgressAddresses:  declaredAll(),
 				IngressDomains:   declaredAll(),
 				IngressAddresses: declaredAll(),
-			},
+			}},
 		},
 	}
 }

pkg/processtree/ancestors.go

@@ -0,0 +1,49 @@
+package processtree
+
+import (
+	"github.com/armosec/armoapi-go/armotypes"
+)
+
+// GetAncestorPIDs returns pid's ancestors, nearest first, up to maxDepth entries.
+// pid itself is excluded.
+//
+// This walks the creator's global process map rather than
+// containerTree.GetPidBranch, because GetPidBranch resolves a container shim and
+// errors out when there is none -- which is every host / cgroup-0 process. Walking
+// the map works identically for containerised and host processes.
+//
+// maxDepth also bounds the walk defensively: a reparenting race could in
+// principle produce a parent cycle, and the evaluator must not hang.
+func (ptm *ProcessTreeManagerImpl) GetAncestorPIDs(pid uint32, maxDepth int) []uint32 {
+	if maxDepth <= 0 {
+		return nil
+	}
+
+	var out []uint32
+	seen := make(map[uint32]struct{}, maxDepth)
+	current := pid
+
+	for len(out) < maxDepth {
+		var node *armotypes.Process
+		func() {
+			ptm.mutex.RLock()
+			defer ptm.mutex.RUnlock()
+			node, _ = ptm.creator.GetProcessNode(int(current))
+		}()
+		// PPID 0 means "parent unknown", not "parent is pid 0" -- recording it
+		// would add a key no state entry can ever be stored under.
+		if node == nil || node.PPID == 0 {
+			break
+		}
+		if _, dup := seen[node.PPID]; dup {
+			break
+		}
+		seen[node.PPID] = struct{}{}
+		out = append(out, node.PPID)
+		if node.PPID == 1 {
+			break
+		}
+		current = node.PPID
+	}
+	return out
+}