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

@@ -16,6 +16,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"
 )
 
@@ -54,6 +55,7 @@ type AlertDeduplicationConfig struct {
 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"`
@@ -209,6 +211,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("eventDedup::enabled", true)

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"
@@ -101,6 +102,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

@@ -346,6 +346,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

@@ -65,4 +65,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

@@ -99,3 +99,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

@@ -55,3 +55,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

@@ -77,6 +77,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
@@ -247,6 +254,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)
 
@@ -557,3 +574,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/metricsmanager/prometheus/prometheus.go

@@ -107,6 +107,13 @@ type PrometheusMetric struct {
 	// Alert suppression funnel
 	alertSuppressedCounter *prometheus.CounterVec
 
+	// CEL rule state store
+	stateWritesCounter        *prometheus.CounterVec
+	stateWriteRejectedCounter *prometheus.CounterVec
+	stateExpiredCounter       prometheus.Counter
+	statePurgedCounter        prometheus.Counter
+	stateEntriesGauge         *prometheus.GaugeVec
+
 	// Cache to avoid allocating Labels maps on every call
 	ruleCounterCache            map[string]prometheus.Counter
 	rulePrefilteredCounterCache map[string]prometheus.Counter
@@ -387,6 +394,26 @@ func NewPrometheusMetric() *PrometheusMetric {
 			Name: "node_agent_alert_suppressed_total",
 			Help: "Total alerts suppressed before delivery, labeled by rule_id and reason",
 		}, []string{prometheusRuleIdLabel, "reason"}),
+		stateWritesCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "node_agent_state_writes_total",
+			Help: "Total CEL rule state entries written, labeled by rule_id",
+		}, []string{prometheusRuleIdLabel, "result"}),
+		stateWriteRejectedCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "node_agent_state_write_rejected_total",
+			Help: "Total CEL rule state writes rejected; a rule is being starved of the state it needs to correlate",
+		}, []string{prometheusRuleIdLabel, "reason"}),
+		stateExpiredCounter: promauto.NewCounter(prometheus.CounterOpts{
+			Name: "node_agent_state_expired_total",
+			Help: "Total CEL rule state entries reclaimed by TTL expiry",
+		}),
+		statePurgedCounter: promauto.NewCounter(prometheus.CounterOpts{
+			Name: "node_agent_state_purged_total",
+			Help: "Total CEL rule state entries dropped by scope purge, e.g. container removal",
+		}),
+		stateEntriesGauge: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "node_agent_state_entries",
+			Help: "Current CEL rule state entries, labeled by scope",
+		}, []string{"scope"}),
 		sbomScanDuration: promauto.NewHistogramVec(prometheus.HistogramOpts{
 			Name:    "sbom_scan_duration_seconds",
 			Help:    "SBOM scan duration in seconds",
@@ -479,6 +506,11 @@ func (p *PrometheusMetric) Destroy() {
 	prometheus.Unregister(p.programPerCpuUsageGauge)
 	prometheus.Unregister(p.sbomScanCounter)
 	prometheus.Unregister(p.alertSuppressedCounter)
+	prometheus.Unregister(p.stateWritesCounter)
+	prometheus.Unregister(p.stateWriteRejectedCounter)
+	prometheus.Unregister(p.stateExpiredCounter)
+	prometheus.Unregister(p.statePurgedCounter)
+	prometheus.Unregister(p.stateEntriesGauge)
 	prometheus.Unregister(p.sbomScanDuration)
 	prometheus.Unregister(p.sbomRestarts)
 	prometheus.Unregister(p.sbomReady)
@@ -772,3 +804,23 @@ func (p *PrometheusMetric) SetSBOMScannerReady(ready bool) {
 func (p *PrometheusMetric) ReportAlertSuppressed(ruleID, reason string) {
 	p.alertSuppressedCounter.WithLabelValues(ruleID, reason).Inc()
 }
+
+func (p *PrometheusMetric) ReportStateWrite(ruleID, result string) {
+	p.stateWritesCounter.WithLabelValues(ruleID, result).Inc()
+}
+
+func (p *PrometheusMetric) ReportStateWriteRejected(ruleID, reason string) {
+	p.stateWriteRejectedCounter.WithLabelValues(ruleID, reason).Inc()
+}
+
+func (p *PrometheusMetric) ReportStateExpired(n int) {
+	p.stateExpiredCounter.Add(float64(n))
+}
+
+func (p *PrometheusMetric) ReportStatePurged(n int) {
+	p.statePurgedCounter.Add(float64(n))
+}
+
+func (p *PrometheusMetric) ReportStateEntries(scope string, n int) {
+	p.stateEntriesGauge.WithLabelValues(scope).Set(float64(n))
+}

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,7 +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",
+		RuntimeRule:         armotypes.RuntimeRule{ID: "test-rule"},
 		ProfileDataRequired: pdr,
 	}
 }
@@ -63,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)
 
@@ -90,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

@@ -7,6 +7,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"
@@ -257,8 +259,7 @@ func networkProfile() *v1beta1.ContainerProfile {
 func mixedFilterRules() []typesv1.Rule {
 	return []typesv1.Rule{
 		{
-			ID:   "RULE-A",
-			Name: "mixed-file-and-net",
+			RuntimeRule: armotypes.RuntimeRule{ID: "RULE-A", Name: "mixed-file-and-net"},
 			ProfileDataRequired: &typesv1.ProfileDataRequired{
 				Opens: declaredPatterns(
 					typesv1.PatternObject{Exact: "/etc/passwd"},
@@ -279,8 +280,7 @@ func mixedFilterRules() []typesv1.Rule {
 		{
 			// A second rule narrows capabilities further and adds an opens
 			// prefix, proving the union merge across rules.
-			ID:   "RULE-B",
-			Name: "extra-caps",
+			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/"}),
@@ -294,8 +294,7 @@ func mixedFilterRules() []typesv1.Rule {
 func netAllRules() []typesv1.Rule {
 	return []typesv1.Rule{
 		{
-			ID:   "RULE-NET",
-			Name: "net-all",
+			RuntimeRule: armotypes.RuntimeRule{ID: "RULE-NET", Name: "net-all"},
 			ProfileDataRequired: &typesv1.ProfileDataRequired{
 				EgressDomains:    declaredAll(),
 				EgressAddresses:  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
+}