feat(enricher): make ring buffer capacity configurable

Signed-off-by: Matthew McKeen <matthew.mckeen@fastly.com>

Author: mmckeen

cmd/standard/daemon.go

@@ -245,7 +245,11 @@ func (d *Daemon) Start() error {
 	if daemonConfig.EnablePodLevel {
 		pubSub := pubsub.New()
 		controllerCache := controllercache.New(pubSub)
-		enrich := enricher.New(ctx, controllerCache)
+		ringCap, capErr := enricher.RingCapacityOrDefault(daemonConfig.EnricherRingCapacity)
+		if capErr != nil {
+			mainLogger.Fatal("invalid enricher ring capacity", zap.Error(capErr))
+		}
+		enrich := enricher.New(ctx, controllerCache, ringCap)
 		//nolint:govet // shadowing this err is fine
 		fm, err := filtermanager.Init(5, daemonConfig.FilterMapMaxEntries) //nolint:gomnd // defaults
 		if err != nil {

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

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

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

@@ -72,6 +72,9 @@ packetParserRingBufferSize: 8388608
 # This map tracks IP addresses of pods of interest for network observability.
 # Default: 255. Increase for large clusters with many tracked pods.
 filterMapMaxEntries: 255
+# Capacity of the enricher flow ring buffer. Must be one less than a power of two
+# (e.g. 1023, 4095, 65535). Default: 1023. Increase to absorb bursts and reduce enricher_ring drops.
+enricherRingCapacity: 1023
 
 imagePullSecrets: []
 nameOverride: "retina"

docs/02-Installation/03-Config.md

@@ -56,6 +56,7 @@ Apply to both Agent and Operator.
 * `dataSamplingRate`: Defines the data sampling rate for `packetparser`.  See [Sampling](../03-Metrics/plugins/Linux/packetparser.md#sampling) for more details.
 * `packetParserRingBuffer`: Selects the kernel-to-userspace transport for `packetparser`. Accepted values: `enabled` (ring buffer) or `disabled` (perf event array). `auto` is reserved for future use.
 * `packetParserRingBufferSize`: Ring buffer size in bytes when `packetParserRingBuffer=enabled`. Must be a power of two between the kernel page size and 1GiB (inclusive); invalid values cause startup to fail.
+* `enricherRingCapacity`: Capacity of the enricher flow ring buffer (default `1023`). Must be one less than a power of two (e.g. `1023`, `4095`, `65535`); invalid values cause startup to fail. Increase to absorb bursts and reduce `enricher_ring` lost events. Takes effect on agent restart.
 
 ## Operator Configuration
 

pkg/config/config.go

@@ -135,6 +135,7 @@ type Config struct {
 	PacketParserRingBufferSize uint32                     `yaml:"packetParserRingBufferSize"`
 	FilterMapMaxEntries        uint32                     `yaml:"filterMapMaxEntries"`
 	EnableTCX                  TCXMode                    `yaml:"enableTCX"`
+	EnricherRingCapacity       uint32                     `yaml:"enricherRingCapacity"`
 }
 
 func GetConfig(cfgFilename string) (*Config, error) {

pkg/enricher/enricher.go

@@ -5,6 +5,7 @@ package enricher
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 	"sync"
 
@@ -41,28 +42,41 @@ type Enricher struct {
 	outputRing *container.Ring
 }
 
-func New(ctx context.Context, c cache.CacheInterface) *Enricher {
+func New(ctx context.Context, c cache.CacheInterface, ringCapacity container.Capacity) *Enricher {
 	once.Do(func() {
-		e = newEnricher(ctx, c)
+		e = newEnricher(ctx, c, ringCapacity)
 	})
 
 	return e
 }
 
-func newEnricher(ctx context.Context, c cache.CacheInterface) *Enricher {
-	ir := container.NewRing(container.Capacity1023)
+func newEnricher(ctx context.Context, c cache.CacheInterface, ringCapacity container.Capacity) *Enricher {
+	ir := container.NewRing(ringCapacity)
 	enricher := &Enricher{
 		ctx:        ctx,
 		l:          log.Logger().Named("enricher"),
 		cache:      c,
 		inputRing:  ir,
 		Reader:     container.NewRingReader(ir, ir.OldestWrite()),
-		outputRing: container.NewRing(container.Capacity1023),
+		outputRing: container.NewRing(ringCapacity),
 	}
 	initialized = true
 	return enricher
 }
 
+// RingCapacityOrDefault converts a configured ring capacity to a container.Capacity,
+// using the default when n is 0. n must be one less than a power of two.
+func RingCapacityOrDefault(n uint32) (container.Capacity, error) {
+	if n == 0 {
+		return container.Capacity1023, nil
+	}
+	c, err := container.NewCapacity(int(n))
+	if err != nil {
+		return nil, fmt.Errorf("invalid ring capacity %d: %w", n, err)
+	}
+	return c, nil
+}
+
 func Instance() *Enricher {
 	return e
 }

pkg/enricher/enricher_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
+	"github.com/cilium/cilium/pkg/hubble/container"
 	retinav1alpha1 "github.com/microsoft/retina/crd/api/v1alpha1"
 	"github.com/microsoft/retina/pkg/common"
 	"github.com/microsoft/retina/pkg/controllers/cache"
@@ -102,7 +103,7 @@ func TestEnricher(t *testing.T) {
 	err = c.UpdateRetinaEndpoint(destPod)
 	require.NoError(t, err)
 
-	e := newEnricher(context.Background(), c)
+	e := newEnricher(context.Background(), c, container.Capacity1023)
 
 	var wg sync.WaitGroup
 	defer wg.Wait()
@@ -159,7 +160,7 @@ func TestEnricherSecondaryIPs(t *testing.T) {
 	require.NoError(t, err)
 
 	// create new enricher (not using singleton here)
-	e := newEnricher(ctx, c)
+	e := newEnricher(ctx, c, container.Capacity1023)
 	var wg sync.WaitGroup
 
 	wg.Add(1)
@@ -279,7 +280,7 @@ func TestEnricherZoneResolution(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	e := newEnricher(ctx, c)
+	e := newEnricher(ctx, c, container.Capacity1023)
 
 	// Get the export reader before running and writing, so we don't miss events.
 	oreader := e.ExportReader()
@@ -330,7 +331,7 @@ func TestEnricherZoneResolution_NoNode(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	e := newEnricher(ctx, c)
+	e := newEnricher(ctx, c, container.Capacity1023)
 
 	oreader := e.ExportReader()
 	e.Run()
@@ -357,3 +358,30 @@ func TestEnricherZoneResolution_NoNode(t *testing.T) {
 	assert.Equal(t, "unknown", utils.SourceZone(enrichedFlow))
 	assert.Equal(t, "unknown", utils.DestinationZone(enrichedFlow))
 }
+
+func TestRingCapacityOrDefault(t *testing.T) {
+	tests := []struct {
+		name    string
+		n       uint32
+		want    container.Capacity
+		wantErr bool
+	}{
+		{"zero uses default", 0, container.Capacity1023, false},
+		{"valid 1023", 1023, container.Capacity1023, false},
+		{"valid 4095", 4095, container.Capacity4095, false},
+		{"valid max 65535", 65535, container.Capacity65535, false},
+		{"invalid not power-of-two-minus-one", 4096, nil, true},
+		{"invalid arbitrary", 5000, nil, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := RingCapacityOrDefault(tt.n)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

pkg/managers/controllermanager/controllermanager.go

@@ -4,6 +4,7 @@ package controllermanager
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"time"
 
@@ -85,7 +86,11 @@ func (m *Controller) Init(ctx context.Context) error {
 		m.cache = cache.New(m.pubsub)
 
 		// create enricher instance
-		m.enricher = enricher.New(ctx, m.cache)
+		ringCap, err := enricher.RingCapacityOrDefault(m.conf.EnricherRingCapacity)
+		if err != nil {
+			return fmt.Errorf("failed to resolve enricher ring capacity: %w", err)
+		}
+		m.enricher = enricher.New(ctx, m.cache, ringCap)
 	}
 
 	return nil

pkg/plugin/ciliumeventobserver/ciliumeventobserver_linux_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
+	"github.com/cilium/cilium/pkg/hubble/container"
 	"github.com/cilium/cilium/pkg/hubble/testutils"
 	"github.com/cilium/cilium/pkg/monitor"
 	monitorAPI "github.com/cilium/cilium/pkg/monitor/api"
@@ -31,7 +32,7 @@ func TestStartError(t *testing.T) {
 	_, _ = log.SetupZapLogger(log.GetDefaultLogOpts())
 
 	c := cache.New(pubsub.New())
-	e := enricher.New(ctxTimeout, c)
+	e := enricher.New(ctxTimeout, c, container.Capacity1023)
 	e.Run()
 	defer e.Reader.Close()
 

test/enricher/main_linux.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
+	"github.com/cilium/cilium/pkg/hubble/container"
 	"github.com/microsoft/retina/pkg/controllers/cache"
 	"github.com/microsoft/retina/pkg/enricher"
 	"github.com/microsoft/retina/pkg/log"
@@ -26,7 +27,7 @@ func main() {
 	ctx := context.Background()
 	c := cache.New(pubsub.New())
 
-	e := enricher.New(ctx, c)
+	e := enricher.New(ctx, c, container.Capacity1023)
 
 	e.Run()