From 7183fa0c02da2f3b38fd3d7115efd6ab51635d26 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Wed, 8 Jul 2026 15:28:54 -0700 Subject: [PATCH] feat(enricher): make ring buffer capacity configurable Signed-off-by: Matthew McKeen --- cmd/standard/daemon.go | 6 +++- .../helm/retina/templates/configmap.yaml | 1 + .../controller/helm/retina/values.yaml | 3 ++ docs/02-Installation/03-Config.md | 1 + pkg/config/config.go | 1 + pkg/enricher/enricher.go | 24 ++++++++++--- pkg/enricher/enricher_test.go | 36 ++++++++++++++++--- .../controllermanager/controllermanager.go | 7 +++- .../ciliumeventobserver_linux_test.go | 3 +- test/enricher/main_linux.go | 3 +- test/plugin/dns/main_linux.go | 3 +- 11 files changed, 74 insertions(+), 14 deletions(-) diff --git a/cmd/standard/daemon.go b/cmd/standard/daemon.go index b93763054e..ffc9dddad5 100644 --- a/cmd/standard/daemon.go +++ b/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 { diff --git a/deploy/standard/manifests/controller/helm/retina/templates/configmap.yaml b/deploy/standard/manifests/controller/helm/retina/templates/configmap.yaml index 61974ff10a..0d1fbf0e6a 100644 --- a/deploy/standard/manifests/controller/helm/retina/templates/configmap.yaml +++ b/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}} diff --git a/deploy/standard/manifests/controller/helm/retina/values.yaml b/deploy/standard/manifests/controller/helm/retina/values.yaml index 063fb02ecd..43c9b4901e 100644 --- a/deploy/standard/manifests/controller/helm/retina/values.yaml +++ b/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" diff --git a/docs/02-Installation/03-Config.md b/docs/02-Installation/03-Config.md index dc622e1227..add702ef0c 100644 --- a/docs/02-Installation/03-Config.md +++ b/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 diff --git a/pkg/config/config.go b/pkg/config/config.go index 6f26f50cc3..b6431f3f24 100644 --- a/pkg/config/config.go +++ b/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) { diff --git a/pkg/enricher/enricher.go b/pkg/enricher/enricher.go index 408e4dfed2..5bea918f7c 100644 --- a/pkg/enricher/enricher.go +++ b/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 } diff --git a/pkg/enricher/enricher_test.go b/pkg/enricher/enricher_test.go index 7f818923fd..be4869c0e1 100644 --- a/pkg/enricher/enricher_test.go +++ b/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) + }) + } +} diff --git a/pkg/managers/controllermanager/controllermanager.go b/pkg/managers/controllermanager/controllermanager.go index b5fcd86380..5017011094 100644 --- a/pkg/managers/controllermanager/controllermanager.go +++ b/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 diff --git a/pkg/plugin/ciliumeventobserver/ciliumeventobserver_linux_test.go b/pkg/plugin/ciliumeventobserver/ciliumeventobserver_linux_test.go index b0b4f53053..f64db9b37d 100644 --- a/pkg/plugin/ciliumeventobserver/ciliumeventobserver_linux_test.go +++ b/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() diff --git a/test/enricher/main_linux.go b/test/enricher/main_linux.go index acb700be85..f003885aee 100644 --- a/test/enricher/main_linux.go +++ b/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() diff --git a/test/plugin/dns/main_linux.go b/test/plugin/dns/main_linux.go index d225de580b..6347d06fca 100644 --- a/test/plugin/dns/main_linux.go +++ b/test/plugin/dns/main_linux.go @@ -20,6 +20,7 @@ import ( "log/slog" "time" + "github.com/cilium/cilium/pkg/hubble/container" "github.com/microsoft/retina/pkg/config" "github.com/microsoft/retina/pkg/controllers/cache" "github.com/microsoft/retina/pkg/enricher" @@ -53,7 +54,7 @@ func main() { defer cancel() c := cache.New(pubsub.New()) - e := enricher.New(ctx, c) + e := enricher.New(ctx, c, container.Capacity1023) e.Run() err = tt.Generate(ctxTimeout)