[client] Reuse one wgctrl client in the kernel configurer (#7912)

* [client] Reuse one wgctrl client in the kernel configurer The kernel configurer opened a new wgctrl client for every request, paying a socket open, a generic netlink family lookup and a close on each peer update. With 10000 peers a single update dropped from about 37us to 10us per call once the client is kept open. The configurer now opens the client lazily on the first request and keeps it. A failed request drops the client so the next one opens a fresh socket: a reply cut short mid-dump leaves stale messages in the netlink buffer, and reusing that socket would fail every later request on a sequence mismatch. The failed request is not retried. Close releases the client and rejects later requests instead of reopening a socket nobody would close. The kernel device never closed its configurer, so it now keeps it and closes it on teardown.
2026-10-13 00:49:07 +02:00 · 2026-10-12 11:28:37 +02:00
parent 01185ba353
commit 5c9c64e3e0
3 changed files with 271 additions and 58 deletions
@@ -3,10 +3,12 @@
 package configurer

 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
 	"slices"
+	"sync"
 	"time"

 	log "github.com/sirupsen/logrus"
@@ -16,18 +18,38 @@ import (
 	"github.com/netbirdio/netbird/monotime"
 )

+var errConfigurerClosed = errors.New("configurer closed")
+
+type wgClient interface {
+	Device(name string) (*wgtypes.Device, error)
+	ConfigureDevice(name string, cfg wgtypes.Config) error
+	Close() error
+}
+
 type KernelConfigurer struct {
-	deviceName string
-	statsCache *statsCache
-	allowedIPs *allowedIPStore
+	deviceName  string
+	statsCache  *statsCache
+	allowedIPs  *allowedIPStore
+	newWgctrlFn func() (wgClient, error)
+
+	clientMu sync.Mutex
+	client   wgClient
+	closed   bool
 }

 // NewKernelConfigurer creates a configurer with an empty allowed IP mirror
 // and a statistics cache for the named kernel device.
 func NewKernelConfigurer(deviceName string) *KernelConfigurer {
+	return newKernelConfigurer(deviceName, func() (wgClient, error) {
+		return wgctrl.New()
+	})
+}
+
+func newKernelConfigurer(deviceName string, newWgctrlFn func() (wgClient, error)) *KernelConfigurer {
 	c := &KernelConfigurer{
-		deviceName: deviceName,
-		allowedIPs: newAllowedIPStore(),
+		deviceName:  deviceName,
+		allowedIPs:  newAllowedIPStore(),
+		newWgctrlFn: newWgctrlFn,
 	}
 	c.statsCache = newStatsCache(statsCacheTTL, c.fetchStats)
 	return c
@@ -242,7 +264,7 @@ func (c *KernelConfigurer) peerAllowedIPs(peerKey wgtypes.Key) ([]netip.Prefix,
 		return prefixes, nil
 	}

-	existingPeer, err := c.getPeer(c.deviceName, peerKey)
+	existingPeer, err := c.getPeer(peerKey)
 	if err != nil {
 		return nil, fmt.Errorf("get peer: %w", err)
 	}
@@ -254,21 +276,10 @@ func (c *KernelConfigurer) peerAllowedIPs(peerKey wgtypes.Key) ([]netip.Prefix,

 // getPeer scans the device for one peer. wgtypes.Key is an array, so the comparison is a
 // plain equality: Key.String would base64 encode into a fresh allocation for every peer.
-func (c *KernelConfigurer) getPeer(ifaceName string, peerPubKey wgtypes.Key) (wgtypes.Peer, error) {
-	wg, err := wgctrl.New()
+func (c *KernelConfigurer) getPeer(peerPubKey wgtypes.Key) (wgtypes.Peer, error) {
+	wgDevice, err := c.device()
 	if err != nil {
-		return wgtypes.Peer{}, fmt.Errorf("wgctl: %w", err)
-	}
-	defer func() {
-		err = wg.Close()
-		if err != nil {
-			log.Errorf("Got error while closing wgctl: %v", err)
-		}
-	}()
-
-	wgDevice, err := wg.Device(ifaceName)
-	if err != nil {
-		return wgtypes.Peer{}, fmt.Errorf("get device %s: %w", ifaceName, err)
+		return wgtypes.Peer{}, err
 	}
 	for _, peer := range wgDevice.Peers {
 		if peer.PublicKey == peerPubKey {
@@ -279,37 +290,73 @@ func (c *KernelConfigurer) getPeer(ifaceName string, peerPubKey wgtypes.Key) (wg
 }

 func (c *KernelConfigurer) configure(config wgtypes.Config) error {
-	wg, err := wgctrl.New()
+	c.clientMu.Lock()
+	defer c.clientMu.Unlock()
+
+	client, err := c.clientLocked()
 	if err != nil {
 		return err
 	}
-	defer func() {
-		if err := wg.Close(); err != nil {
-			log.Errorf("Failed to close wgctrl client: %v", err)
-		}
-	}()
-
-	return wg.ConfigureDevice(c.deviceName, config)
+	if err := client.ConfigureDevice(c.deviceName, config); err != nil {
+		c.discardLocked()
+		return err
+	}
+	return nil
 }

-func (c *KernelConfigurer) Close() {
+func (c *KernelConfigurer) device() (*wgtypes.Device, error) {
+	c.clientMu.Lock()
+	defer c.clientMu.Unlock()
+
+	client, err := c.clientLocked()
+	if err != nil {
+		return nil, err
+	}
+	wgDevice, err := client.Device(c.deviceName)
+	if err != nil {
+		c.discardLocked()
+		return nil, fmt.Errorf("get device %s: %w", c.deviceName, err)
+	}
+	return wgDevice, nil
 }

-func (c *KernelConfigurer) FullStats() (*Stats, error) {
-	wg, err := wgctrl.New()
+func (c *KernelConfigurer) clientLocked() (wgClient, error) {
+	if c.closed {
+		return nil, errConfigurerClosed
+	}
+	if c.client != nil {
+		return c.client, nil
+	}
+	client, err := c.newWgctrlFn()
 	if err != nil {
 		return nil, fmt.Errorf("wgctl: %w", err)
 	}
-	defer func() {
-		err = wg.Close()
-		if err != nil {
-			log.Errorf("Got error while closing wgctl: %v", err)
-		}
-	}()
+	c.client = client
+	return client, nil
+}

-	wgDevice, err := wg.Device(c.deviceName)
+func (c *KernelConfigurer) discardLocked() {
+	if c.client == nil {
+		return
+	}
+	if err := c.client.Close(); err != nil {
+		log.Warnf("failed to close wgctrl client: %v", err)
+	}
+	c.client = nil
+}
+
+func (c *KernelConfigurer) Close() {
+	c.clientMu.Lock()
+	defer c.clientMu.Unlock()
+
+	c.closed = true
+	c.discardLocked()
+}
+
+func (c *KernelConfigurer) FullStats() (*Stats, error) {
+	wgDevice, err := c.device()
 	if err != nil {
-		return nil, fmt.Errorf("get device %s: %w", c.deviceName, err)
+		return nil, err
 	}
 	fullStats := &Stats{
 		DeviceName: wgDevice.Name,
@@ -345,23 +392,12 @@ func (c *KernelConfigurer) LastActivities() map[string]monotime.Time {
 }

 func (c *KernelConfigurer) fetchStats() (map[string]WGStats, error) {
+	wgDevice, err := c.device()
+	if err != nil {
+		return nil, err
+	}
+
 	stats := make(map[string]WGStats)
-	wg, err := wgctrl.New()
-	if err != nil {
-		return nil, fmt.Errorf("wgctl: %w", err)
-	}
-	defer func() {
-		err = wg.Close()
-		if err != nil {
-			log.Errorf("Got error while closing wgctl: %v", err)
-		}
-	}()
-
-	wgDevice, err := wg.Device(c.deviceName)
-	if err != nil {
-		return nil, fmt.Errorf("get device %s: %w", c.deviceName, err)
-	}
-
 	for _, peer := range wgDevice.Peers {
 		stats[peer.PublicKey.String()] = WGStats{
 			LastHandshake: peer.LastHandshakeTime,
@@ -0,0 +1,167 @@
+//go:build (linux && !android) || freebsd
+
+package configurer
+
+import (
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+)
+
+type fakeWGClient struct {
+	configureErr error
+	deviceErr    error
+	calls        int
+	closed       bool
+}
+
+type fakeWGClientFactory struct {
+	openErr error
+	clients []*fakeWGClient
+}
+
+func (f *fakeWGClient) Device(string) (*wgtypes.Device, error) {
+	f.calls++
+	if f.deviceErr != nil {
+		return nil, f.deviceErr
+	}
+	return &wgtypes.Device{}, nil
+}
+
+func (f *fakeWGClient) ConfigureDevice(string, wgtypes.Config) error {
+	f.calls++
+	return f.configureErr
+}
+
+func (f *fakeWGClient) Close() error {
+	f.closed = true
+	return nil
+}
+
+func (f *fakeWGClientFactory) open() (wgClient, error) {
+	if f.openErr != nil {
+		return nil, f.openErr
+	}
+	client := &fakeWGClient{}
+	f.clients = append(f.clients, client)
+	return client, nil
+}
+
+func TestKernelConfigurer_ReusesClientAcrossRequests(t *testing.T) {
+	c, factory := newFakeKernelConfigurer()
+	peerKey := testPeerKey(t)
+
+	require.NoError(t, c.RemovePeer(peerKey))
+	require.NoError(t, c.RemovePeer(peerKey))
+	_, err := c.FullStats()
+	require.NoError(t, err)
+
+	require.Len(t, factory.clients, 1, "all requests should share one client")
+	assert.Equal(t, 3, factory.clients[0].calls, "every request should go through the shared client")
+	assert.False(t, factory.clients[0].closed, "a healthy client should stay open")
+}
+
+func TestKernelConfigurer_ReopensClientAfterFailedConfigure(t *testing.T) {
+	c, factory := newFakeKernelConfigurer()
+	peerKey := testPeerKey(t)
+
+	require.NoError(t, c.RemovePeer(peerKey))
+	require.Len(t, factory.clients, 1)
+
+	factory.clients[0].configureErr = errors.New("netlink down")
+	require.Error(t, c.RemovePeer(peerKey))
+	assert.True(t, factory.clients[0].closed, "the failed client should be closed")
+	assert.Len(t, factory.clients, 1, "a failure must not open a client on its own")
+
+	require.NoError(t, c.RemovePeer(peerKey))
+	require.Len(t, factory.clients, 2, "the next request should open a fresh client")
+	assert.Equal(t, 1, factory.clients[1].calls, "the request should run on the fresh client")
+	assert.False(t, factory.clients[1].closed, "the fresh client should stay open")
+}
+
+func TestKernelConfigurer_ReopensClientAfterFailedDeviceQuery(t *testing.T) {
+	c, factory := newFakeKernelConfigurer()
+
+	_, err := c.FullStats()
+	require.NoError(t, err)
+	require.Len(t, factory.clients, 1)
+
+	factory.clients[0].deviceErr = errors.New("netlink down")
+	_, err = c.FullStats()
+	require.Error(t, err)
+	assert.True(t, factory.clients[0].closed, "the failed client should be closed")
+
+	_, err = c.FullStats()
+	require.NoError(t, err)
+	assert.Len(t, factory.clients, 2, "the next request should open a fresh client")
+}
+
+func TestKernelConfigurer_RetriesOpenOnNextRequest(t *testing.T) {
+	c, factory := newFakeKernelConfigurer()
+	peerKey := testPeerKey(t)
+
+	factory.openErr = errors.New("no netlink")
+	require.Error(t, c.RemovePeer(peerKey))
+	assert.Empty(t, factory.clients, "a failed open leaves no client behind")
+
+	factory.openErr = nil
+	require.NoError(t, c.RemovePeer(peerKey))
+	assert.Len(t, factory.clients, 1, "the next request should open the client")
+}
+
+func TestKernelConfigurer_CloseRejectsFurtherRequests(t *testing.T) {
+	c, factory := newFakeKernelConfigurer()
+	peerKey := testPeerKey(t)
+
+	require.NoError(t, c.RemovePeer(peerKey))
+	require.Len(t, factory.clients, 1)
+
+	c.Close()
+	c.Close()
+	assert.True(t, factory.clients[0].closed, "Close should release the client")
+
+	require.ErrorIs(t, c.RemovePeer(peerKey), errConfigurerClosed)
+	_, err := c.FullStats()
+	require.ErrorIs(t, err, errConfigurerClosed)
+	assert.Len(t, factory.clients, 1, "a closed configurer must not open a client")
+}
+
+func TestKernelConfigurer_ConcurrentRequestsShareOneClient(t *testing.T) {
+	c, factory := newFakeKernelConfigurer()
+	peerKey := testPeerKey(t)
+
+	const workers = 16
+	errs := make(chan error, workers)
+	var wg sync.WaitGroup
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			errs <- c.RemovePeer(peerKey)
+		}()
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+
+	require.Len(t, factory.clients, 1, "concurrent requests should share one client")
+	assert.Equal(t, workers, factory.clients[0].calls, "every request should reach the shared client")
+}
+
+func newFakeKernelConfigurer() (*KernelConfigurer, *fakeWGClientFactory) {
+	factory := &fakeWGClientFactory{}
+	return newKernelConfigurer("wt0", factory.open), factory
+}
+
+func testPeerKey(t *testing.T) string {
+	t.Helper()
+	key, err := wgtypes.GeneratePrivateKey()
+	require.NoError(t, err)
+	return key.PublicKey().String()
+}
@@ -32,6 +32,7 @@ type TunKernelDevice struct {
 	link       *wgLink
 	udpMuxConn net.PacketConn
 	udpMux     *udpmux.UniversalUDPMuxDefault
+	configurer WGConfigurer
 }

 func NewKernelDevice(name string, address wgaddr.Address, wgPort int, key string, mtu uint16, transportNet transport.Net) *TunKernelDevice {
@@ -68,13 +69,18 @@ func (t *TunKernelDevice) Create() (WGConfigurer, error) {
 		return nil, fmt.Errorf("set mtu: %w", err)
 	}

-	configurer := configurer.NewKernelConfigurer(t.name)
+	cfg := configurer.NewKernelConfigurer(t.name)

-	if err := configurer.ConfigureInterface(t.key, t.wgPort); err != nil {
+	if err := cfg.ConfigureInterface(t.key, t.wgPort); err != nil {
+		cfg.Close()
 		return nil, fmt.Errorf("error configuring interface: %s", err)
 	}

-	return configurer, nil
+	if t.configurer != nil {
+		t.configurer.Close()
+	}
+	t.configurer = cfg
+	return cfg, nil
 }

 func (t *TunKernelDevice) Up() (*udpmux.UniversalUDPMuxDefault, error) {
@@ -132,6 +138,10 @@ func (t *TunKernelDevice) Close() error {
 		closErr = err
 	}

+	if t.configurer != nil {
+		t.configurer.Close()
+	}
+
 	if t.udpMux != nil {
 		if err := t.udpMux.Close(); err != nil {
 			log.Debugf("failed to close udp mux: %s", err)