datapath: Increase the devices controller's netlink receive buffer

The devices controller subscribes to address, route, link and neighbor
updates on netlink sockets with the kernel default receive buffer. Under
high churn that buffer overflows with ENOBUFS, and any one socket erroring
restarts all four subscriptions.

Size all four explicitly: 4MiB by default, configurable with the new
--netlink-buffer-size flag.

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.NetlinkBufferSize, defaults.NetlinkBufferSize, "Size (in bytes) of the netlink socket receive buffer used by the devices controller's address, route, link and neighbor subscriptions; a larger buffer reduces the chance of ENOBUFS errors and subscription restarts under high churn")
 }
 
 var (
@@ -93,6 +97,9 @@ type DevicesConfig struct {
 	// ForceDeviceDetection forces the auto-detection of devices,
 	// even if user-specific devices are explicitly listed.
 	ForceDeviceDetection bool
+	// NetlinkBufferSize is the receive buffer size in bytes requested for
+	// the controller's netlink subscription sockets.
+	NetlinkBufferSize int
 }
 
 type devicesControllerParams struct {
@@ -141,7 +148,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.NetlinkBufferSize)
 		if err != nil {
 			return err
 		}
@@ -741,9 +748,36 @@ type netlinkFuncs struct {
 	NeighList         func(linkIndex, family int) ([]netlink.Neigh, error)
 }
 
+// The EPERM latch is required: netlink orphans its socket when setsockopt
+// fails, so re-forcing the buffer on every restart would leak sockets. It is
+// shared by every subscription because CAP_NET_ADMIN is process-wide.
+func subscribeWithForcedBuffer[Options any](
+	subscribe func(Options) error,
+	options func(size int) Options,
+	bufferSize int,
+	forceFailed *atomic.Bool,
+	log *slog.Logger,
+) error {
+	if forceFailed.Load() {
+		bufferSize = 0
+	}
+
+	err := subscribe(options(bufferSize))
+	if err == nil || bufferSize <= 0 || !errors.Is(err, unix.EPERM) {
+		return err
+	}
+
+	forceFailed.Store(true)
+	log.Warn("Failed to force the netlink receive buffer (SO_RCVBUFFORCE requires CAP_NET_ADMIN); falling back to the kernel default. Netlink updates may be dropped under high churn.",
+		logfields.Error, err,
+		logfields.BufferSize, bufferSize,
+	)
+	return subscribe(options(0))
+}
+
 // 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, netlinkBufferSize 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,42 +788,78 @@ func makeNetlinkFuncs() (*netlinkFuncs, error) {
 		return nil, fmt.Errorf("getting current netns: %w", err)
 	}
 
+	netlinkBufferSize = max(netlinkBufferSize, 0)
+
+	var forceBufferFailed atomic.Bool
+
 	return &netlinkFuncs{
 		RouteSubscribe: func(ch chan<- netlink.RouteUpdate, done <-chan struct{}, errorCallback func(error)) error {
 			h := vns.NsHandle(cur.FD())
-			return safenetlink.RouteSubscribeWithOptions(ch, done,
-				netlink.RouteSubscribeOptions{
-					ListExisting:  false,
-					ErrorCallback: errorCallback,
-					Namespace:     &h,
-				})
+			return subscribeWithForcedBuffer(
+				func(opts netlink.RouteSubscribeOptions) error {
+					return safenetlink.RouteSubscribeWithOptions(ch, done, opts)
+				},
+				func(size int) netlink.RouteSubscribeOptions {
+					return netlink.RouteSubscribeOptions{
+						ListExisting:           false,
+						ErrorCallback:          errorCallback,
+						Namespace:              &h,
+						ReceiveBufferSize:      size,
+						ReceiveBufferForceSize: size > 0,
+					}
+				},
+				netlinkBufferSize, &forceBufferFailed, log)
 		},
 		AddrSubscribe: func(ch chan<- netlink.AddrUpdate, done <-chan struct{}, errorCallback func(error)) error {
 			h := vns.NsHandle(cur.FD())
-			return netlink.AddrSubscribeWithOptions(ch, done,
-				netlink.AddrSubscribeOptions{
-					ListExisting:  false,
-					ErrorCallback: errorCallback,
-					Namespace:     &h,
-				})
+			return subscribeWithForcedBuffer(
+				func(opts netlink.AddrSubscribeOptions) error {
+					return netlink.AddrSubscribeWithOptions(ch, done, opts)
+				},
+				func(size int) netlink.AddrSubscribeOptions {
+					return netlink.AddrSubscribeOptions{
+						ListExisting:           false,
+						ErrorCallback:          errorCallback,
+						Namespace:              &h,
+						ReceiveBufferSize:      size,
+						ReceiveBufferForceSize: size > 0,
+					}
+				},
+				netlinkBufferSize, &forceBufferFailed, log)
 		},
 		LinkSubscribe: func(ch chan<- netlink.LinkUpdate, done <-chan struct{}, errorCallback func(error)) error {
 			h := vns.NsHandle(cur.FD())
-			return safenetlink.LinkSubscribeWithOptions(ch, done,
-				netlink.LinkSubscribeOptions{
-					ListExisting:  false,
-					ErrorCallback: errorCallback,
-					Namespace:     &h,
-				})
+			return subscribeWithForcedBuffer(
+				func(opts netlink.LinkSubscribeOptions) error {
+					return safenetlink.LinkSubscribeWithOptions(ch, done, opts)
+				},
+				func(size int) netlink.LinkSubscribeOptions {
+					return netlink.LinkSubscribeOptions{
+						ListExisting:           false,
+						ErrorCallback:          errorCallback,
+						Namespace:              &h,
+						ReceiveBufferSize:      size,
+						ReceiveBufferForceSize: size > 0,
+					}
+				},
+				netlinkBufferSize, &forceBufferFailed, log)
 		},
 		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 subscribeWithForcedBuffer(
+				func(opts netlink.NeighSubscribeOptions) error {
+					return safenetlink.NeighSubscribeWithOptions(ch, done, opts)
+				},
+				func(size int) netlink.NeighSubscribeOptions {
+					return netlink.NeighSubscribeOptions{
+						ListExisting:           false,
+						ErrorCallback:          errorCallback,
+						Namespace:              &h,
+						ReceiveBufferSize:      size,
+						ReceiveBufferForceSize: size > 0,
+					}
+				},
+				netlinkBufferSize, &forceBufferFailed, 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.NetlinkBufferSize)
 			}),
 		)
 
@@ -97,6 +99,137 @@ func TestPrivilegedDevicesControllerScript(t *testing.T) {
 		"testdata/device-*.txtar")
 }
 
+func TestSubscribeWithForcedBuffer(t *testing.T) {
+	log := hivetest.Logger(t)
+	ns := netns.None()
+
+	options := func(size int) netlink.NeighSubscribeOptions {
+		return netlink.NeighSubscribeOptions{
+			ListExisting:           false,
+			ErrorCallback:          func(error) {},
+			Namespace:              &ns,
+			ReceiveBufferSize:      size,
+			ReceiveBufferForceSize: size > 0,
+		}
+	}
+
+	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, subscribeWithForcedBuffer(sub, options, 4096, &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, subscribeWithForcedBuffer(sub, options, 4096, &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, subscribeWithForcedBuffer(sub, options, 4096, &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 := subscribeWithForcedBuffer(sub, options, 4096, &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 := subscribeWithForcedBuffer(sub, options, 4096, &flag, log)
+		require.ErrorIs(t, err, fallbackErr)
+		require.Len(t, *calls, 2)
+		assert.True(t, flag.Load())
+	})
+
+	t.Run("one latch is shared by every subscription", func(t *testing.T) {
+		var flag atomic.Bool
+		calls, sub := newFake(unix.EPERM, nil, nil)
+
+		require.NoError(t, subscribeWithForcedBuffer(sub, options, 4096, &flag, log))
+		require.NoError(t, subscribeWithForcedBuffer(sub, options, 4096, &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], "later subscriptions 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
+		calls, sub := newFake(wantErr)
+		err := subscribeWithForcedBuffer(sub, options, 0, &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 TestNetlinkBufferSizeConfig(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.NetlinkBufferSize, cfg.NetlinkBufferSize)
+	})
+
+	t.Run("flag overrides default", func(t *testing.T) {
+		cfg := newConfig(t, "--"+option.NetlinkBufferSize+"=1024")
+		assert.Equal(t, 1024, cfg.NetlinkBufferSize)
+	})
+}
+
 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,9 @@ const (
 )
 
 const (
+	// NetlinkBufferSize is the default value for option.NetlinkBufferSize
+	NetlinkBufferSize = 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"
 
+	// NetlinkBufferSize sets the size in bytes of the netlink socket receive
+	// buffer used by the devices controller's subscriptions
+	NetlinkBufferSize = "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"