merge upstream/main: replace AP+NN cache with unified ContainerProfileCache (#788)

Makefile

@@ -11,6 +11,10 @@ TAG?=test
 binary:
 	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o $(BINARY_NAME) ./cmd/main.go
 
+.PHONY: check-legacy-packages
+check-legacy-packages:
+	go test ./tests/containerprofilecache -run TestLegacyPackagesDeleted
+
 docker-build-only:
 	docker buildx build --platform linux/amd64 -t $(IMAGE):$(TAG) -f $(DOCKERFILE_PATH) --load .
 

cmd/main.go

@@ -45,10 +45,9 @@ import (
 	"github.com/kubescape/node-agent/pkg/nodeprofilemanager"
 	nodeprofilemanagerv1 "github.com/kubescape/node-agent/pkg/nodeprofilemanager/v1"
 	"github.com/kubescape/node-agent/pkg/objectcache"
-	"github.com/kubescape/node-agent/pkg/objectcache/applicationprofilecache"
+	"github.com/kubescape/node-agent/pkg/objectcache/containerprofilecache"
 	"github.com/kubescape/node-agent/pkg/objectcache/dnscache"
 	"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/processtree"
 	containerprocesstree "github.com/kubescape/node-agent/pkg/processtree/container"
@@ -297,16 +296,14 @@ func main() {
 		ruleBindingNotify = make(chan rulebinding.RuleBindingNotify, 100)
 		ruleBindingCache.AddNotifier(&ruleBindingNotify)
 
-		apc := applicationprofilecache.NewApplicationProfileCache(cfg, storageClient, k8sObjectCache, exporter)
-		apc.Start(ctx)
-
-		nnc := networkneighborhoodcache.NewNetworkNeighborhoodCache(cfg, storageClient, k8sObjectCache, exporter)
-		nnc.Start(ctx)
+		cpc := containerprofilecache.NewContainerProfileCache(cfg, storageClient, k8sObjectCache, prometheusExporter)
+		cpc.Start(ctx)
+		logger.L().Info("ContainerProfileCache active; legacy AP/NN caches removed")
 
 		dc := dnscache.NewDnsCache(dnsResolver)
 
 		// create object cache
-		objCache = objectcachev1.NewObjectCache(k8sObjectCache, apc, nnc, dc)
+		objCache = objectcachev1.NewObjectCache(k8sObjectCache, cpc, dc)
 
 		ruleCooldown := rulecooldown.NewRuleCooldown(cfg.RuleCoolDown)
 
@@ -328,10 +325,9 @@ func main() {
 
 	} else {
 		ruleManager = rulemanager.CreateRuleManagerMock()
-		apc := &objectcache.ApplicationProfileCacheMock{}
-		nnc := &objectcache.NetworkNeighborhoodCacheMock{}
+		cpc := &objectcache.ContainerProfileCacheMock{}
 		dc := &objectcache.DnsCacheMock{}
-		objCache = objectcachev1.NewObjectCache(k8sObjectCache, apc, nnc, dc)
+		objCache = objectcachev1.NewObjectCache(k8sObjectCache, cpc, dc)
 		ruleBindingNotify = make(chan rulebinding.RuleBindingNotify, 1)
 	}
 

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/joncrlsn/dque v0.0.0-20241024143830-7723fd131a64
 	github.com/kubescape/backend v0.0.39
 	github.com/kubescape/go-logger v0.0.28
-	github.com/kubescape/k8s-interface v0.0.206
+	github.com/kubescape/k8s-interface v0.0.208
 	github.com/kubescape/storage v0.0.258
 	github.com/kubescape/workerpool v0.0.0-20250526074519-0e4a4e7f44cf
 	github.com/moby/sys/mountinfo v0.7.2
@@ -60,6 +60,7 @@ require (
 	go.uber.org/multierr v1.11.0
 	golang.org/x/net v0.53.0
 	golang.org/x/sys v0.43.0
+	golang.org/x/tools v0.43.0
 	gonum.org/v1/plot v0.14.0
 	google.golang.org/grpc v1.80.0
 	google.golang.org/protobuf v1.36.11
@@ -473,7 +474,6 @@ require (
 	golang.org/x/term v0.42.0 // indirect
 	golang.org/x/text v0.36.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
-	golang.org/x/tools v0.43.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/api v0.271.0 // indirect
 	google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 // indirect

go.sum

@@ -1013,8 +1013,8 @@ github.com/kubescape/backend v0.0.39 h1:B1QRfKCSFlzuE+jWOnk/l7EpH71/Q3n14KKq0QSn
 github.com/kubescape/backend v0.0.39/go.mod h1:cMEGP8cXUZgY89YU4GRBGIla9HZW7grZsUtlCwvZgAE=
 github.com/kubescape/go-logger v0.0.28 h1:xulKTp9kOg3rD98sopFELQ6yZCHQoQXMDzteoSHDFKI=
 github.com/kubescape/go-logger v0.0.28/go.mod h1:YZHFjwGCDar1hP9OyBLE46oR7a0Y/Z/0FperDo8+9D0=
-github.com/kubescape/k8s-interface v0.0.206 h1:qaYu4mlLmSBePanSGq+DBCssh4O785TAT0lQGNGWyGw=
-github.com/kubescape/k8s-interface v0.0.206/go.mod h1:WNYUG93aZ5kDmuaRKFLtVhp18Yc6EfaHdD1gLYtVTN4=
+github.com/kubescape/k8s-interface v0.0.208 h1:vmZ2FVAQRsz3XRKNG/6wJAYvZJ12RtMoDTLVxFEktms=
+github.com/kubescape/k8s-interface v0.0.208/go.mod h1:WNYUG93aZ5kDmuaRKFLtVhp18Yc6EfaHdD1gLYtVTN4=
 github.com/kubescape/workerpool v0.0.0-20250526074519-0e4a4e7f44cf h1:hI0jVwrB6fT4GJWvuUjzObfci1CUknrZdRHfnRVtKM0=
 github.com/kubescape/workerpool v0.0.0-20250526074519-0e4a4e7f44cf/go.mod h1:Il5baM40PV9cTt4OGdLMeTRRAai3TMfvImu31itIeCM=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=

pkg/config/config.go

@@ -107,6 +107,7 @@ type Config struct {
 	ProcfsPidScanInterval          time.Duration                        `mapstructure:"procfsPidScanInterval"`
 	ProcfsScanInterval             time.Duration                        `mapstructure:"procfsScanInterval"`
 	ProfilesCacheRefreshRate       time.Duration                        `mapstructure:"profilesCacheRefreshRate"`
+	StorageRPCBudget               time.Duration                        `mapstructure:"storageRPCBudget"`
 	RuleCoolDown                   rulecooldown.RuleCooldownConfig      `mapstructure:"ruleCooldown"`
 	TestMode                       bool                                 `mapstructure:"testMode"`
 	UpdateDataPeriod               time.Duration                        `mapstructure:"updateDataPeriod"`

pkg/containerprofilemanager/v1/lifecycle.go

@@ -162,6 +162,7 @@ func (cpm *ContainerProfileManager) addContainer(container *containercollection.
 
 	// Setup monitoring timer
 	sniffingTime := cpm.calculateSniffingTime(container)
+	sharedData.LearningPeriod = sniffingTime
 	timer := time.AfterFunc(sniffingTime, func() {
 		cpm.handleContainerMaxTime(container)
 	})

pkg/containerwatcher/v2/container_watcher_collection.go

@@ -60,8 +60,7 @@ func (cw *ContainerWatcher) StartContainerCollection(ctx context.Context) error
 		cw.containerCallbackAsync,
 		cw.containerProcessTree.ContainerCallback,
 		cw.containerProfileManager.ContainerCallback,
-		cw.objectCache.ApplicationProfileCache().ContainerCallback,
-		cw.objectCache.NetworkNeighborhoodCache().ContainerCallback,
+		cw.objectCache.ContainerProfileCache().ContainerCallback,
 		cw.malwareManager.ContainerCallback,
 		cw.ruleManager.ContainerCallback,
 		cw.sbomManager.ContainerCallback,

pkg/hostsensormanager/sensor_kubelet.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"fmt"
 
+	logger "github.com/kubescape/go-logger"
 	"github.com/kubescape/go-logger/helpers"
 	"github.com/kubescape/k8s-interface/hostsensor"
+	"sigs.k8s.io/yaml"
 )
 
 const (
@@ -25,6 +27,32 @@ var kubeletKubeConfigDefaultPathList = []string{
 	"/var/lib/kubelet/kubeconfig",
 }
 
+var kubeletServiceFilePaths = []string{
+	"/etc/systemd/system/kubelet.service",
+	"/usr/lib/systemd/system/kubelet.service",
+	"/lib/systemd/system/kubelet.service",
+}
+
+const kubeletServiceDropInDir = "/etc/systemd/system/kubelet.service.d"
+
+// kubeletConfigYAML is a minimal subset of KubeletConfiguration for CA file extraction.
+type kubeletConfigYAML struct {
+	Authentication struct {
+		X509 struct {
+			ClientCAFile string `json:"clientCAFile"`
+		} `json:"x509"`
+	} `json:"authentication"`
+}
+
+// extractClientCAFromKubeletConfig parses kubelet config YAML and returns the clientCAFile path.
+func extractClientCAFromKubeletConfig(content []byte) (string, error) {
+	var cfg kubeletConfigYAML
+	if err := yaml.Unmarshal(content, &cfg); err != nil {
+		return "", fmt.Errorf("failed to parse kubelet config: %w", err)
+	}
+	return cfg.Authentication.X509.ClientCAFile, nil
+}
+
 // KubeletInfoSensor implements the Sensor interface for kubelet info data
 type KubeletInfoSensor struct {
 	nodeName string
@@ -73,12 +101,31 @@ func (s *KubeletInfoSensor) Sense() (interface{}, error) {
 		ret.KubeConfigFile = makeContaineredFileInfoFromListVerbose(ctx, kubeletProcess, kubeletKubeConfigDefaultPathList, true, helpers.String("in", "SenseKubeletInfo"))
 	}
 
-	// Client CA
+	// Client CA: check cmdLine first, then fall back to kubelet config YAML
 	if caFilePath, ok := kubeletProcess.GetArg(kubeletClientCAArgName); ok {
 		ret.ClientCAFile = makeContaineredFileInfoVerbose(ctx, kubeletProcess, caFilePath, false, helpers.String("in", "SenseKubeletInfo"))
+	} else if ret.ConfigFile != nil && len(ret.ConfigFile.Content) > 0 {
+		if caFilePath, err := extractClientCAFromKubeletConfig(ret.ConfigFile.Content); err != nil {
+			logger.L().Debug("failed to extract clientCAFile from kubelet config", helpers.String("in", "SenseKubeletInfo"), helpers.Error(err))
+		} else if caFilePath != "" {
+			ret.ClientCAFile = makeContaineredFileInfoVerbose(ctx, kubeletProcess, caFilePath, false, helpers.String("in", "SenseKubeletInfo"))
+		}
 	}
 
 	ret.CmdLine = kubeletProcess.RawCmd()
 
+	// Service files: main unit file and drop-in directory
+	for _, svcPath := range kubeletServiceFilePaths {
+		if fi := makeHostFileInfoVerbose(ctx, svcPath, false); fi != nil {
+			ret.ServiceFiles = append(ret.ServiceFiles, *fi)
+			break
+		}
+	}
+	if dropIns, err := makeHostDirFilesInfoVerbose(ctx, kubeletServiceDropInDir, false, 0); err == nil {
+		for _, fi := range dropIns {
+			ret.ServiceFiles = append(ret.ServiceFiles, *fi)
+		}
+	}
+
 	return &ret, nil
 }

pkg/metricsmanager/metrics_manager_interface.go

@@ -20,4 +20,9 @@ type MetricsManager interface {
 	ReportContainerStart()
 	ReportContainerStop()
 	ReportDedupEvent(eventType utils.EventType, duplicate bool)
+	ReportContainerProfileLegacyLoad(kind, completeness string)
+	SetContainerProfileCacheEntries(kind string, count float64)
+	ReportContainerProfileCacheHit(hit bool)
+	ReportContainerProfileReconcilerDuration(phase string, duration time.Duration)
+	ReportContainerProfileReconcilerEviction(reason string)
 }

pkg/metricsmanager/metrics_manager_mock.go

@@ -66,4 +66,9 @@ func (m *MetricsMock) ReportContainerStart() {}
 
 func (m *MetricsMock) ReportContainerStop() {}
 
-func (m *MetricsMock) ReportDedupEvent(eventType utils.EventType, duplicate bool) {}
+func (m *MetricsMock) ReportDedupEvent(eventType utils.EventType, duplicate bool)          {}
+func (m *MetricsMock) ReportContainerProfileLegacyLoad(_, _ string)                       {}
+func (m *MetricsMock) SetContainerProfileCacheEntries(_ string, _ float64)                {}
+func (m *MetricsMock) ReportContainerProfileCacheHit(_ bool)                              {}
+func (m *MetricsMock) ReportContainerProfileReconcilerDuration(_ string, _ time.Duration) {}
+func (m *MetricsMock) ReportContainerProfileReconcilerEviction(_ string)                  {}

pkg/metricsmanager/metrics_manager_noop.go

@@ -22,3 +22,8 @@ func (m *MetricsNoop) ReportRuleEvaluationTime(_ string, _ utils.EventType, _ ti
 func (m *MetricsNoop) ReportContainerStart()                                                {}
 func (m *MetricsNoop) ReportContainerStop()                                                 {}
 func (m *MetricsNoop) ReportDedupEvent(_ utils.EventType, _ bool)                           {}
+func (m *MetricsNoop) ReportContainerProfileLegacyLoad(_, _ string)                        {}
+func (m *MetricsNoop) SetContainerProfileCacheEntries(_ string, _ float64)                 {}
+func (m *MetricsNoop) ReportContainerProfileCacheHit(_ bool)                               {}
+func (m *MetricsNoop) ReportContainerProfileReconcilerDuration(_ string, _ time.Duration)  {}
+func (m *MetricsNoop) ReportContainerProfileReconcilerEviction(_ string)                   {}

pkg/metricsmanager/prometheus/prometheus.go

@@ -63,6 +63,13 @@ type PrometheusMetric struct {
 	// Dedup metrics
 	dedupEventCounter *prometheus.CounterVec
 
+	// ContainerProfile cache metrics
+	cpCacheLegacyLoadsCounter      *prometheus.CounterVec
+	cpCacheEntriesGauge            *prometheus.GaugeVec
+	cpCacheHitCounter              *prometheus.CounterVec
+	cpReconcilerDurationHistogram  *prometheus.HistogramVec
+	cpReconcilerEvictionsCounter   *prometheus.CounterVec
+
 	// Cache to avoid allocating Labels maps on every call
 	ruleCounterCache          map[string]prometheus.Counter
 	rulePrefilteredCounterCache map[string]prometheus.Counter
@@ -215,6 +222,29 @@ func NewPrometheusMetric() *PrometheusMetric {
 			Help: "Total number of events processed by the dedup layer",
 		}, []string{eventTypeLabel, "result"}),
 
+		// ContainerProfile cache metrics
+		cpCacheLegacyLoadsCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "node_agent_user_profile_legacy_loads_total",
+			Help: "Number of times a user-authored legacy ApplicationProfile or NetworkNeighborhood was loaded into the ContainerProfileCache; will be removed in a future release.",
+		}, []string{"kind", "completeness"}),
+		cpCacheEntriesGauge: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "node_agent_containerprofile_cache_entries",
+			Help: "Current number of cached ContainerProfile entries per kind.",
+		}, []string{"kind"}),
+		cpCacheHitCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "node_agent_containerprofile_cache_hit_total",
+			Help: "Total number of ContainerProfile cache lookups by result.",
+		}, []string{"result"}),
+		cpReconcilerDurationHistogram: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "node_agent_containerprofile_reconciler_duration_seconds",
+			Help:    "Duration of ContainerProfile reconciler phases in seconds.",
+			Buckets: prometheus.DefBuckets,
+		}, []string{"phase"}),
+		cpReconcilerEvictionsCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "node_agent_containerprofile_reconciler_evictions_total",
+			Help: "Total number of ContainerProfile cache evictions by reason.",
+		}, []string{"reason"}),
+
 		// Initialize counter caches
 		ruleCounterCache:            make(map[string]prometheus.Counter),
 		rulePrefilteredCounterCache: make(map[string]prometheus.Counter),
@@ -256,6 +286,11 @@ func (p *PrometheusMetric) Destroy() {
 	prometheus.Unregister(p.containerStartCounter)
 	prometheus.Unregister(p.containerStopCounter)
 	prometheus.Unregister(p.dedupEventCounter)
+	prometheus.Unregister(p.cpCacheLegacyLoadsCounter)
+	prometheus.Unregister(p.cpCacheEntriesGauge)
+	prometheus.Unregister(p.cpCacheHitCounter)
+	prometheus.Unregister(p.cpReconcilerDurationHistogram)
+	prometheus.Unregister(p.cpReconcilerEvictionsCounter)
 	// Unregister program ID metrics
 	prometheus.Unregister(p.programRuntimeGauge)
 	prometheus.Unregister(p.programRunCountGauge)
@@ -432,3 +467,27 @@ func (p *PrometheusMetric) ReportDedupEvent(eventType utils.EventType, duplicate
 	}
 	p.dedupEventCounter.WithLabelValues(string(eventType), result).Inc()
 }
+
+func (p *PrometheusMetric) ReportContainerProfileLegacyLoad(kind, completeness string) {
+	p.cpCacheLegacyLoadsCounter.WithLabelValues(kind, completeness).Inc()
+}
+
+func (p *PrometheusMetric) SetContainerProfileCacheEntries(kind string, count float64) {
+	p.cpCacheEntriesGauge.WithLabelValues(kind).Set(count)
+}
+
+func (p *PrometheusMetric) ReportContainerProfileCacheHit(hit bool) {
+	result := "hit"
+	if !hit {
+		result = "miss"
+	}
+	p.cpCacheHitCounter.WithLabelValues(result).Inc()
+}
+
+func (p *PrometheusMetric) ReportContainerProfileReconcilerDuration(phase string, duration time.Duration) {
+	p.cpReconcilerDurationHistogram.WithLabelValues(phase).Observe(duration.Seconds())
+}
+
+func (p *PrometheusMetric) ReportContainerProfileReconcilerEviction(reason string) {
+	p.cpReconcilerEvictionsCounter.WithLabelValues(reason).Inc()
+}