[datapath] avoid repeated netlink socket leak on forced-buffer fallback

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

Author: jaredledvina

pkg/datapath/linux/devices_controller.go

@@ -13,6 +13,7 @@ import (
 	"net/netip"
 	"slices"
 	"strings"
+	"sync/atomic"
 
 	"github.com/cilium/hive/cell"
 	"github.com/cilium/statedb"
@@ -802,6 +803,15 @@ func makeNetlinkFuncs(log *slog.Logger, neighborReceiveBufferSize int) (*netlink
 		return nil, fmt.Errorf("getting current netns: %w", err)
 	}
 
+	// neighForceBufferFailed records that forcing the neighbor subscription's
+	// receive buffer has already failed for this process (e.g. missing
+	// CAP_NET_ADMIN or a size rejected against net.core.rmem_max). Such failures
+	// are effectively permanent, so once we have seen one we stop attempting the
+	// forced subscription on subsequent restarts. This bounds the netlink socket
+	// leaked by the vendored library on the setsockopt-failure path to a single
+	// fd for the lifetime of the process instead of leaking one per restart.
+	var neighForceBufferFailed atomic.Bool
+
 	return &netlinkFuncs{
 		RouteSubscribe: func(ch chan<- netlink.RouteUpdate, done <-chan struct{}, errorCallback func(error)) error {
 			h := vns.NsHandle(cur.FD())
@@ -831,28 +841,41 @@ func makeNetlinkFuncs(log *slog.Logger, neighborReceiveBufferSize int) (*netlink
 				})
 		},
 		NeighSubscribe: func(ch chan<- netlink.NeighUpdate, done <-chan struct{}, errorCallback func(error)) error {
+			// Only force the receive buffer if forcing hasn't already failed
+			// for this process (see neighForceBufferFailed).
+			forceBufferSize := neighborReceiveBufferSize
+			if neighForceBufferFailed.Load() {
+				forceBufferSize = 0
+			}
+
 			h := vns.NsHandle(cur.FD())
 			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))
+				neighSubscribeOptions(forceBufferSize, &h, errorCallback))
+			if err == nil || forceBufferSize <= 0 {
+				return err
 			}
-			return err
+
+			// Subscribing with a forced receive buffer failed. Forcing the
+			// buffer past net.core.rmem_max requires CAP_NET_ADMIN and a size
+			// the kernel accepts, so the most likely cause is a permanent
+			// misconfiguration; without a fallback the controller would wedge in
+			// the restartWaitDuration loop in run(), retrying the same failure
+			// 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) and remember
+			// the failure so we don't re-attempt (and re-leak a socket via the
+			// vendored library) on every restart. The error may instead be an
+			// unrelated socket-setup failure (e.g. fd exhaustion); in that case
+			// the fallback re-hits and propagates it, so no genuine error is
+			// masked. On the setsockopt-failure path the library returns before
+			// starting the goroutine that closes ch, so ch is safe to reuse.
+			neighForceBufferFailed.Store(true)
+			log.Warn("Failed to subscribe to neighbor updates with a forced netlink receive buffer; retrying without forcing it. Neighbor updates may be dropped more often under high churn. This usually means the agent lacks CAP_NET_ADMIN or the configured size was rejected against net.core.rmem_max.",
+				logfields.Error, err,
+				logfields.BufferSize, neighborReceiveBufferSize,
+			)
+			return safenetlink.NeighSubscribeWithOptions(ch, done,
+				neighSubscribeOptions(0, &h, errorCallback))
 		},
 		LinkList: func() ([]netlink.Link, error) {
 			return safenetlink.WithRetryResult(func() ([]netlink.Link, error) {