[datapath] increase netlink receive buffer for neighbor subscription

Signed-off-by: Jared Ledvina <jared.ledvina@datadoghq.com>

Author: jaredledvina

Documentation/cmdref/cilium-agent.md

@@ -7,12 +7,14 @@ package linux
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
 	"net/netip"
 	"slices"
 	"strings"
+	"sync/atomic"
 
 	"github.com/cilium/hive/cell"
 	"github.com/cilium/statedb"
@@ -66,6 +68,8 @@ func (c DevicesConfig) Flags(flags *pflag.FlagSet) {
 	flags.StringSlice(option.Devices, []string{}, "List of devices facing cluster/external network (used for BPF NodePort, BPF masquerading and host firewall); supports '+' as wildcard in device name, e.g. 'eth+'; support '!' to exclude devices, e.g. '!eth+' excludes any device with prefix 'eth'. Note '!' says nothing about which ones to include. A device must match other criteria to be selected; The filters are matched in order and whatever matched first wins.")
 
 	flags.Bool(option.ForceDeviceDetection, false, "Forces the auto-detection of devices, even if specific devices are explicitly listed")
+
+	flags.Int(option.NeighborNetlinkBufferSize, defaults.NeighborNetlinkBufferSize, "Size (in bytes) of the netlink socket receive buffer used to subscribe to neighbor updates; a larger buffer reduces the chance of ENOBUFS errors and subscription restarts under high neighbor churn")
 }
 
 var (
@@ -92,7 +96,8 @@ type DevicesConfig struct {
 	Devices []string
 	// ForceDeviceDetection forces the auto-detection of devices,
 	// even if user-specific devices are explicitly listed.
-	ForceDeviceDetection bool
+	ForceDeviceDetection      bool
+	NeighborNetlinkBufferSize int
 }
 
 type devicesControllerParams struct {
@@ -141,7 +146,7 @@ func newDevicesController(lc cell.Lifecycle, p devicesControllerParams) (*device
 func (dc *devicesController) Start(startCtx cell.HookContext) error {
 	if dc.params.NetlinkFuncs == nil {
 		var err error
-		dc.params.NetlinkFuncs, err = makeNetlinkFuncs()
+		dc.params.NetlinkFuncs, err = makeNetlinkFuncs(dc.log, dc.params.Config.NeighborNetlinkBufferSize)
 		if err != nil {
 			return err
 		}
@@ -741,9 +746,57 @@ type netlinkFuncs struct {
 	NeighList         func(linkIndex, family int) ([]netlink.Neigh, error)
 }
 
+// neighSubscribeOptions normalizes negative sizes to 0: the underlying library
+// skips setsockopt only when the size is exactly 0, so a negative size would
+// otherwise reach setsockopt(SO_RCVBUF) and be silently clamped by the kernel.
+func neighSubscribeOptions(receiveBufferSize int, ns *vns.NsHandle, errorCallback func(error)) netlink.NeighSubscribeOptions {
+	if receiveBufferSize < 0 {
+		receiveBufferSize = 0
+	}
+	return netlink.NeighSubscribeOptions{
+		ListExisting:           false,
+		ErrorCallback:          errorCallback,
+		Namespace:              ns,
+		ReceiveBufferSize:      receiveBufferSize,
+		ReceiveBufferForceSize: receiveBufferSize > 0,
+	}
+}
+
+// subscribeNeighWithBufferFallback forces the receive buffer via
+// SO_RCVBUFFORCE, which needs CAP_NET_ADMIN and otherwise fails with EPERM. On
+// EPERM it falls back once to the kernel default and latches forceFailed so
+// later restarts skip forcing: the vendored library orphans its netlink socket
+// on the setsockopt-failure path, so re-attempting every restart would leak
+// sockets.
+func subscribeNeighWithBufferFallback(
+	subscribe func(netlink.NeighSubscribeOptions) error,
+	bufferSize int,
+	ns *vns.NsHandle,
+	errorCallback func(error),
+	forceFailed *atomic.Bool,
+	log *slog.Logger,
+) error {
+	forceBufferSize := bufferSize
+	if forceFailed.Load() {
+		forceBufferSize = 0
+	}
+
+	err := subscribe(neighSubscribeOptions(forceBufferSize, ns, errorCallback))
+	if err == nil || forceBufferSize <= 0 || !errors.Is(err, unix.EPERM) {
+		return err
+	}
+
+	forceFailed.Store(true)
+	log.Warn("Failed to subscribe to neighbor updates with a forced netlink receive buffer (the agent likely lacks CAP_NET_ADMIN, which SO_RCVBUFFORCE requires); retrying without forcing it. Neighbor updates may be dropped more often under high neighbor churn.",
+		logfields.Error, err,
+		logfields.BufferSize, bufferSize,
+	)
+	return subscribe(neighSubscribeOptions(0, ns, errorCallback))
+}
+
 // makeNetlinkFuncs returns a *netlinkFuncs containing netlink accessors to the
 // network namespace of the calling goroutine's OS thread.
-func makeNetlinkFuncs() (*netlinkFuncs, error) {
+func makeNetlinkFuncs(log *slog.Logger, neighborReceiveBufferSize int) (*netlinkFuncs, error) {
 	netlinkHandle, err := safenetlink.NewHandle(&safenetlink.HandleConfig{NLFamilies: []int{unix.NETLINK_ROUTE}})
 	if err != nil {
 		return nil, fmt.Errorf("creating netlink handle: %w", err)
@@ -754,6 +807,9 @@ func makeNetlinkFuncs() (*netlinkFuncs, error) {
 		return nil, fmt.Errorf("getting current netns: %w", err)
 	}
 
+	// Latches after an EPERM so restarts stop re-attempting the forced buffer.
+	var neighForceBufferFailed atomic.Bool
+
 	return &netlinkFuncs{
 		RouteSubscribe: func(ch chan<- netlink.RouteUpdate, done <-chan struct{}, errorCallback func(error)) error {
 			h := vns.NsHandle(cur.FD())
@@ -784,12 +840,11 @@ func makeNetlinkFuncs() (*netlinkFuncs, error) {
 		},
 		NeighSubscribe: func(ch chan<- netlink.NeighUpdate, done <-chan struct{}, errorCallback func(error)) error {
 			h := vns.NsHandle(cur.FD())
-			return safenetlink.NeighSubscribeWithOptions(ch, done,
-				netlink.NeighSubscribeOptions{
-					ListExisting:  false,
-					ErrorCallback: errorCallback,
-					Namespace:     &h,
-				})
+			return subscribeNeighWithBufferFallback(
+				func(opts netlink.NeighSubscribeOptions) error {
+					return safenetlink.NeighSubscribeWithOptions(ch, done, opts)
+				},
+				neighborReceiveBufferSize, &h, errorCallback, &neighForceBufferFailed, log)
 		},
 		LinkList: func() ([]netlink.Link, error) {
 			return safenetlink.WithRetryResult(func() ([]netlink.Link, error) {

Documentation/cmdref/cilium-agent_hive.md

@@ -31,7 +31,9 @@ import (
 	"golang.org/x/sys/unix"
 
 	"github.com/cilium/cilium/pkg/datapath/tables"
+	"github.com/cilium/cilium/pkg/defaults"
 	"github.com/cilium/cilium/pkg/hive"
+	"github.com/cilium/cilium/pkg/option"
 	"github.com/cilium/cilium/pkg/testutils"
 )
 
@@ -62,9 +64,9 @@ func TestPrivilegedDevicesControllerScript(t *testing.T) {
 
 		h := hive.New(
 			DevicesControllerCell,
-			cell.Provide(func() (*netlinkFuncs, error) {
+			cell.Provide(func(log *slog.Logger) (*netlinkFuncs, error) {
 				// Provide the normal netlink interface, restricted to the test network namespace.
-				return makeNetlinkFuncs()
+				return makeNetlinkFuncs(log, defaults.NeighborNetlinkBufferSize)
 			}),
 		)
 
@@ -97,6 +99,153 @@ func TestPrivilegedDevicesControllerScript(t *testing.T) {
 		"testdata/device-*.txtar")
 }
 
+func TestNeighSubscribeOptions(t *testing.T) {
+	errorCallback := func(error) {}
+	ns := netns.None()
+
+	t.Run("configured size is applied and forced", func(t *testing.T) {
+		opts := neighSubscribeOptions(defaults.NeighborNetlinkBufferSize, &ns, errorCallback)
+		assert.Equal(t, defaults.NeighborNetlinkBufferSize, opts.ReceiveBufferSize)
+		assert.True(t, opts.ReceiveBufferForceSize, "buffer size should be forced past net.core.rmem_max")
+		assert.Same(t, &ns, opts.Namespace)
+		assert.False(t, opts.ListExisting)
+	})
+
+	t.Run("zero size leaves kernel default in place", func(t *testing.T) {
+		opts := neighSubscribeOptions(0, &ns, errorCallback)
+		assert.Equal(t, 0, opts.ReceiveBufferSize)
+		assert.False(t, opts.ReceiveBufferForceSize)
+	})
+
+	t.Run("negative size is normalized to kernel default", func(t *testing.T) {
+		opts := neighSubscribeOptions(-1, &ns, errorCallback)
+		assert.Equal(t, 0, opts.ReceiveBufferSize)
+		assert.False(t, opts.ReceiveBufferForceSize)
+	})
+}
+
+func TestSubscribeNeighWithBufferFallback(t *testing.T) {
+	log := hivetest.Logger(t)
+	ns := netns.None()
+	cb := func(error) {}
+
+	type call struct {
+		size  int
+		force bool
+	}
+
+	newFake := func(errs ...error) (*[]call, func(netlink.NeighSubscribeOptions) error) {
+		calls := &[]call{}
+		i := 0
+		return calls, func(opts netlink.NeighSubscribeOptions) error {
+			*calls = append(*calls, call{opts.ReceiveBufferSize, opts.ReceiveBufferForceSize})
+			var err error
+			if i < len(errs) {
+				err = errs[i]
+			}
+			i++
+			return err
+		}
+	}
+
+	t.Run("success on forced attempt does not latch", func(t *testing.T) {
+		var flag atomic.Bool
+		calls, sub := newFake(nil)
+		require.NoError(t, subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log))
+		require.Len(t, *calls, 1)
+		assert.Equal(t, call{4096, true}, (*calls)[0])
+		assert.False(t, flag.Load())
+	})
+
+	t.Run("EPERM falls back to kernel default and latches", func(t *testing.T) {
+		var flag atomic.Bool
+		calls, sub := newFake(unix.EPERM, nil)
+		require.NoError(t, subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log))
+		require.Len(t, *calls, 2)
+		assert.Equal(t, call{4096, true}, (*calls)[0])
+		assert.Equal(t, call{0, false}, (*calls)[1], "fallback must not force the buffer")
+		assert.True(t, flag.Load(), "EPERM should latch so future restarts skip forcing")
+	})
+
+	t.Run("latched flag skips forcing without falling back", func(t *testing.T) {
+		var flag atomic.Bool
+		flag.Store(true)
+		calls, sub := newFake(nil)
+		require.NoError(t, subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log))
+		require.Len(t, *calls, 1)
+		assert.Equal(t, call{0, false}, (*calls)[0])
+	})
+
+	t.Run("non-EPERM error propagates without fallback or latch", func(t *testing.T) {
+		var flag atomic.Bool
+		wantErr := errors.New("socket setup failed")
+		calls, sub := newFake(wantErr)
+		err := subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log)
+		require.ErrorIs(t, err, wantErr)
+		require.Len(t, *calls, 1, "must not fall back on a non-EPERM error")
+		assert.False(t, flag.Load(), "must not latch on a transient/unrelated error")
+	})
+
+	t.Run("fallback error propagates and still latches", func(t *testing.T) {
+		var flag atomic.Bool
+		fallbackErr := errors.New("still broken")
+		calls, sub := newFake(unix.EPERM, fallbackErr)
+		err := subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log)
+		require.ErrorIs(t, err, fallbackErr)
+		require.Len(t, *calls, 2)
+		assert.True(t, flag.Load())
+	})
+
+	t.Run("latch persists across restarts: force once, then never again", func(t *testing.T) {
+		var flag atomic.Bool
+		calls, sub := newFake(unix.EPERM, nil, nil)
+
+		require.NoError(t, subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log))
+		require.NoError(t, subscribeNeighWithBufferFallback(sub, 4096, &ns, cb, &flag, log))
+
+		require.Len(t, *calls, 3, "first call forces+falls back (2), second skips forcing (1)")
+		assert.Equal(t, call{4096, true}, (*calls)[0])
+		assert.Equal(t, call{0, false}, (*calls)[1])
+		assert.Equal(t, call{0, false}, (*calls)[2], "second restart must not re-attempt forcing")
+	})
+
+	t.Run("disabled buffer returns the error without forcing or latching", func(t *testing.T) {
+		var flag atomic.Bool
+		wantErr := unix.EPERM // even EPERM must not latch when forcing wasn't requested
+		calls, sub := newFake(wantErr)
+		err := subscribeNeighWithBufferFallback(sub, 0, &ns, cb, &flag, log)
+		require.ErrorIs(t, err, wantErr)
+		require.Len(t, *calls, 1)
+		assert.Equal(t, call{0, false}, (*calls)[0])
+		assert.False(t, flag.Load())
+	})
+}
+
+func TestNeighborNetlinkBufferSize_Config(t *testing.T) {
+	newConfig := func(t *testing.T, args ...string) DevicesConfig {
+		var got DevicesConfig
+		h := hive.New(
+			cell.Config(DevicesConfig{}),
+			cell.Invoke(func(cfg DevicesConfig) { got = cfg }),
+		)
+		flags := pflag.NewFlagSet("", pflag.ContinueOnError)
+		h.RegisterFlags(flags)
+		require.NoError(t, flags.Parse(args))
+		require.NoError(t, h.Populate(hivetest.Logger(t)))
+		return got
+	}
+
+	t.Run("default is applied", func(t *testing.T) {
+		cfg := newConfig(t)
+		assert.Equal(t, defaults.NeighborNetlinkBufferSize, cfg.NeighborNetlinkBufferSize)
+	})
+
+	t.Run("flag overrides default", func(t *testing.T) {
+		cfg := newConfig(t, "--"+option.NeighborNetlinkBufferSize+"=1024")
+		assert.Equal(t, 1024, cfg.NeighborNetlinkBufferSize)
+	})
+}
+
 func TestDevicesController_Restarts(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()

Documentation/cmdref/cilium-agent_hive_dot-graph.md

@@ -20,6 +20,10 @@ const (
 )
 
 const (
+	// Kernel default overflows under high neighbor churn, causing
+	// ENOBUFS-triggered subscription restarts; 4MiB absorbs bursts.
+	NeighborNetlinkBufferSize = 4 << 20 // 4MiB
+
 	// ClusterHealthPort is the default value for option.ClusterHealthPort
 	ClusterHealthPort = 4240
 

pkg/datapath/linux/devices_controller.go

@@ -112,6 +112,10 @@ const (
 	// Forces the auto-detection of devices, even if specific devices are explicitly listed
 	ForceDeviceDetection = "force-device-detection"
 
+	// NeighborNetlinkBufferSize sets the size in bytes of the netlink socket
+	// receive buffer used by the devices controller's neighbor subscription
+	NeighborNetlinkBufferSize = "neighbor-netlink-buffer-size"
+
 	// DirectRoutingDevice is the name of a device used to connect nodes in
 	// direct routing mode (only required by BPF NodePort)
 	DirectRoutingDevice = "direct-routing-device"