[datapath] harden neighbor netlink receive buffer handling

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

Author: jaredledvina

Documentation/cmdref/cilium-agent.md

@@ -147,7 +147,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.Config.NeighborNetlinkBufferSize)
+		dc.params.NetlinkFuncs, err = makeNetlinkFuncs(dc.log, dc.params.Config.NeighborNetlinkBufferSize)
 		if err != nil {
 			return err
 		}
@@ -762,7 +762,16 @@ type netlinkFuncs struct {
 // past net.core.rmem_max: cilium-agent runs with CAP_NET_ADMIN, and a dropped
 // neighbor update is more costly than the extra kernel memory. A
 // non-positive receiveBufferSize leaves the kernel default in place.
+//
+// The underlying library only skips the setsockopt call when the size is
+// exactly 0, so negative values are normalized to 0 here to honor the
+// "non-positive means kernel default" contract; otherwise a negative value
+// would be passed to setsockopt(SO_RCVBUF) and clamped by the kernel rather
+// than leaving the default untouched.
 func neighSubscribeOptions(receiveBufferSize int, ns *vns.NsHandle, errorCallback func(error)) netlink.NeighSubscribeOptions {
+	if receiveBufferSize < 0 {
+		receiveBufferSize = 0
+	}
 	return netlink.NeighSubscribeOptions{
 		ListExisting:           false,
 		ErrorCallback:          errorCallback,
@@ -779,7 +788,10 @@ func neighSubscribeOptions(receiveBufferSize int, ns *vns.NsHandle, errorCallbac
 // the netlink socket used for the neighbor subscription. See
 // defaults.NeighborNetlinkBufferSize for rationale. A value <= 0 leaves the
 // kernel default in place.
-func makeNetlinkFuncs(neighborReceiveBufferSize int) (*netlinkFuncs, error) {
+//
+// log is used to report the one-time fallback to the kernel-default receive
+// buffer if forcing the configured size fails (see the NeighSubscribe closure).
+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)
@@ -820,8 +832,27 @@ func makeNetlinkFuncs(neighborReceiveBufferSize int) (*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,
+			err := safenetlink.NeighSubscribeWithOptions(ch, done,
 				neighSubscribeOptions(neighborReceiveBufferSize, &h, errorCallback))
+			if err != nil && neighborReceiveBufferSize > 0 {
+				// Forcing the receive buffer past net.core.rmem_max requires
+				// CAP_NET_ADMIN and a size the kernel accepts. If that fails the
+				// error is permanent, and without a fallback the controller would
+				// wedge in the restartWaitDuration loop in run(), retrying it once
+				// per second forever. Fall back once to the kernel-default buffer
+				// so neighbor updates keep flowing, at the cost of more frequent
+				// ENOBUFS-driven restarts under heavy churn. On the setsockopt
+				// failure path the library returns before starting the goroutine
+				// that closes ch, so ch is safe to reuse here.
+				log.Warn("Failed to subscribe to neighbor updates with forced netlink receive buffer, falling back to kernel default buffer",
+					logfields.Error, err,
+					logfields.BufferSize, neighborReceiveBufferSize,
+				)
+				h = vns.NsHandle(cur.FD())
+				return safenetlink.NeighSubscribeWithOptions(ch, done,
+					neighSubscribeOptions(0, &h, errorCallback))
+			}
+			return err
 		},
 		LinkList: func() ([]netlink.Link, error) {
 			return safenetlink.WithRetryResult(func() ([]netlink.Link, error) {

Documentation/cmdref/cilium-agent_hive.md

@@ -33,6 +33,7 @@ import (
 	"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"
 )
 
@@ -63,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(defaults.NeighborNetlinkBufferSize)
+				return makeNetlinkFuncs(log, defaults.NeighborNetlinkBufferSize)
 			}),
 		)
 
@@ -110,11 +111,50 @@ func TestNeighSubscribeOptions(t *testing.T) {
 		assert.False(t, opts.ListExisting)
 	})
 
-	t.Run("non-positive size leaves kernel default in place", func(t *testing.T) {
+	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) {
+		// A negative value must be normalized to 0 so the library skips the
+		// setsockopt call entirely; otherwise setsockopt(SO_RCVBUF, <negative>)
+		// would be issued and clamped by the kernel rather than left at default.
+		opts := neighSubscribeOptions(-1, &ns, errorCallback)
+		assert.Equal(t, 0, opts.ReceiveBufferSize)
+		assert.False(t, opts.ReceiveBufferForceSize)
+	})
+}
+
+// TestNeighborNetlinkBufferSize_Config verifies that the
+// neighbor-netlink-buffer-size flag is registered and that its value is
+// populated into DevicesConfig.NeighborNetlinkBufferSize via hive's
+// field-name<->flag-name convention (the actual wiring the devices controller
+// relies on at Start).
+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) {