From 5c9c64e3e0125785bfcb3a3f4b9eb1f05714e118 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Mon, 12 Oct 2026 11:28:37 +0200 Subject: [PATCH] [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. --- client/iface/configurer/kernel_unix.go | 146 ++++++++++------- client/iface/configurer/kernel_unix_test.go | 167 ++++++++++++++++++++ client/iface/device/device_kernel_unix.go | 16 +- 3 files changed, 271 insertions(+), 58 deletions(-) create mode 100644 client/iface/configurer/kernel_unix_test.go diff --git a/client/iface/configurer/kernel_unix.go b/client/iface/configurer/kernel_unix.go index 7b1ad9683..afe1300d5 100644 --- a/client/iface/configurer/kernel_unix.go +++ b/client/iface/configurer/kernel_unix.go @@ -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, diff --git a/client/iface/configurer/kernel_unix_test.go b/client/iface/configurer/kernel_unix_test.go new file mode 100644 index 000000000..8daa15a25 --- /dev/null +++ b/client/iface/configurer/kernel_unix_test.go @@ -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() +} diff --git a/client/iface/device/device_kernel_unix.go b/client/iface/device/device_kernel_unix.go index 3c429fb96..ee965dd24 100644 --- a/client/iface/device/device_kernel_unix.go +++ b/client/iface/device/device_kernel_unix.go @@ -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)