fix: pod IP deletion leak, namespace filtering, and configurable filter map size

cmd/standard/daemon.go

@@ -241,7 +241,7 @@ func (d *Daemon) Start() error {
 		controllerCache := controllercache.New(pubSub)
 		enrich := enricher.New(ctx, controllerCache)
 		//nolint:govet // shadowing this err is fine
-		fm, err := filtermanager.Init(5) //nolint:gomnd // defaults
+		fm, err := filtermanager.Init(5, daemonConfig.FilterMapMaxEntries) //nolint:gomnd // defaults
 		if err != nil {
 			mainLogger.Fatal("unable to create filter manager", zap.Error(err))
 		}

deploy/standard/manifests/controller/helm/retina/templates/configmap.yaml

@@ -28,6 +28,7 @@ data:
     dataSamplingRate: {{ .Values.dataSamplingRate }}
     packetParserRingBuffer: {{ .Values.packetParserRingBuffer }}
     packetParserRingBufferSize: {{ .Values.packetParserRingBufferSize }}
+    filterMapMaxEntries: {{ .Values.filterMapMaxEntries }}
 {{- end}}
 ---
 {{- if .Values.os.windows}}

deploy/standard/manifests/controller/helm/retina/values.yaml

@@ -68,6 +68,10 @@ packetParserRingBuffer: "disabled"
 # Invalid values cause startup to fail with a validation error.
 # This value is not applied when ring buffers are disabled.
 packetParserRingBufferSize: 8388608
+# Maximum number of entries in the eBPF filter map (retina_filter).
+# This map tracks IP addresses of pods of interest for network observability.
+# Default: 255. Increase for large clusters with many tracked pods.
+filterMapMaxEntries: 255
 
 imagePullSecrets: []
 nameOverride: "retina"

init/retina/main_linux.go

@@ -58,7 +58,7 @@ func run(args ...string) error {
 	}
 
 	// Setup BPF
-	err = bpf.Setup(l)
+	err = bpf.Setup(l, cfg.FilterMapMaxEntries)
 	if err != nil {
 		return errors.Wrap(err, "failed to setup Retina bpf filesystem")
 	}

pkg/bpf/setup_linux.go

@@ -55,7 +55,7 @@ func mountBpfFs() error {
 	return nil
 }
 
-func Setup(l *zap.Logger) error {
+func Setup(l *zap.Logger, filterMapMaxEntries uint32) error {
 	err := mountBpfFs()
 	if err != nil {
 		return errors.Wrap(err, "failed to mount BPF filesystem")
@@ -71,7 +71,7 @@ func Setup(l *zap.Logger) error {
 
 	// Initialize the filter map.
 	// This will create the filter map in kernel and pin it to /sys/fs/bpf.
-	_, err = filter.Init()
+	_, err = filter.Init(filterMapMaxEntries)
 	if err != nil {
 		return errors.Wrap(err, "failed to initialize filter map")
 	}

pkg/config/config.go

@@ -121,6 +121,7 @@ type Config struct {
 	DataSamplingRate           uint32                     `yaml:"dataSamplingRate"`
 	PacketParserRingBuffer     PacketParserRingBufferMode `yaml:"packetParserRingBuffer"`
 	PacketParserRingBufferSize uint32                     `yaml:"packetParserRingBufferSize"`
+	FilterMapMaxEntries        uint32                     `yaml:"filterMapMaxEntries"`
 }
 
 func GetConfig(cfgFilename string) (*Config, error) {
@@ -175,6 +176,11 @@ func GetConfig(cfgFilename string) (*Config, error) {
 		config.DataSamplingRate = DefaultSamplingRate
 	}
 
+	// Default filter map max entries to 255 if not set.
+	if config.FilterMapMaxEntries == 0 {
+		config.FilterMapMaxEntries = 255
+	}
+
 	switch config.PacketParserRingBuffer { //nolint:exhaustive // we only care about Auto and empty (default) here
 	case "":
 		config.PacketParserRingBuffer = PacketParserRingBufferDisabled

pkg/controllers/cache/cache.go

@@ -493,3 +493,14 @@ func (c *Cache) GetAnnotatedNamespaces() []string {
 	sort.Strings(ns)
 	return ns
 }
+
+// GetAllNamespaces returns all namespaces that have endpoints in the cache
+func (c *Cache) GetAllNamespaces() []string {
+	c.RLock()
+	defer c.RUnlock()
+	namespaces := make([]string, 0, len(c.nsMap))
+	for ns := range c.nsMap {
+		namespaces = append(namespaces, ns)
+	}
+	return namespaces
+}

pkg/controllers/cache/types.go

@@ -22,6 +22,8 @@ type CacheInterface interface {
 	GetIPsByNamespace(ns string) []net.IP
 	// GetAnnotatedNamespaces returns list of namespaces that are annotated with retina to observe.
 	GetAnnotatedNamespaces() []string
+	// GetAllNamespaces returns all namespaces that have endpoints in the cache.
+	GetAllNamespaces() []string
 
 	// UpdateRetinaEndpoint updates the retina endpoint in the cache.
 	UpdateRetinaEndpoint(ep *common.RetinaEndpoint) error

pkg/managers/filtermanager/cache.go

@@ -6,6 +6,9 @@ package filtermanager
 import (
 	"net"
 	"sync"
+
+	"github.com/microsoft/retina/pkg/log"
+	"go.uber.org/zap"
 )
 
 var fc = &filterCache{data: make(map[string]requests)}
@@ -61,13 +64,15 @@ func (f *filterCache) addIP(ip net.IP, r Requestor, m RequestMetadata) {
 	defer f.mu.Unlock()
 
 	key := ip.String()
+
 	if _, ok := f.data[key]; !ok {
 		f.data[key] = make(requests)
 	}
 	if _, ok := f.data[key][r]; !ok {
 		f.data[key][r] = make(map[RequestMetadata]bool)
 	}
 	f.data[key][r][m] = true
+
 }
 
 // Return true if the IP was deleted from the cache.
@@ -91,6 +96,21 @@ func (f *filterCache) deleteIP(ip net.IP, r Requestor, m RequestMetadata) bool {
 				delete(f.data, key)
 				ipDeleted = true
 			}
+
+		} else {
+			if log.Logger() != nil {
+				log.Logger().Warn("deleteIP - requestor not found",
+					zap.String("ip", key),
+					zap.String("requestor", string(r)),
+					zap.String("metadata_ruleID", m.RuleID))
+			}
+		}
+	} else {
+		if log.Logger() != nil {
+			log.Logger().Warn("deleteIP - IP not found in cache",
+				zap.String("ip", key),
+				zap.String("requestor", string(r)),
+				zap.String("metadata_ruleID", m.RuleID))
 		}
 	}
 	return ipDeleted

pkg/managers/filtermanager/manager_linux.go

@@ -38,7 +38,7 @@ type FilterManager struct {
 // Total time: 7 seconds
 // The manager locks the cache during retry.
 // Suggest to keep retry to a small number (not more than 3).
-func Init(retry int) (*FilterManager, error) {
+func Init(retry int, filterMapMaxEntries uint32) (*FilterManager, error) {
 	var err error
 	if retry < 1 {
 		return nil, errors.New("retry should be greater than 0")
@@ -55,7 +55,7 @@ func Init(retry int) (*FilterManager, error) {
 	if f.c == nil {
 		f.c = getCache()
 	}
-	f.fm, err = filter.Init()
+	f.fm, err = filter.Init(filterMapMaxEntries)
 	return f, errors.Wrapf(err, "failed to initialize filter map")
 }
 

pkg/managers/filtermanager/manager_windows.go

@@ -11,7 +11,7 @@ import (
 
 type FilterManager struct{}
 
-func Init(_ int) (*FilterManager, error) {
+func Init(_ int, _ uint32) (*FilterManager, error) {
 	return nil, nil
 }