Add setting to enable wireguard NAPI threading (#9260)

* Add setting to enable wireguard NAPI threading

NAPI threading can be used with wireguard to help increase maximum
packets-per-second by distributing the tunnel's load from one core to
many cores via a NAPI kernel thread.

This option is off by default and can be configured via the
FelixConfiguration the same way wireguard can be toggled on and off.

* Fix lowercase typo in wireguard test

* Move bool conversion to fv's utils package

* Only warn when toggling NAPI threading fails

* Revert "Move bool conversion to fv's utils package"

This reverts commit 1614c10.

* Use Infof and Warnf for wireguard NAPI threading logs

* Enable wireguard NAPI threading in topology options

* Remove unused threading variable in topology

* Fix threading enabled setting flag

* Remove wireguard threading topology option

* Ability to toggle wireguard threading setting

* Give wireguard threading more time to be configured properly

* fix wireguard threaded tests

* Only toggle threading on and off once

* Do the same with an earlier test

* Move NAPI threading test to own It block and clean up unused test parameter

* Move napi threading flip logic into test

* fix whitespace

* Add config-params after rebase

* Increment number of felix configs

api/pkg/apis/projectcalico/v3/felixconfig.go

@@ -775,7 +775,8 @@ type FelixConfigurationSpec struct {
 
 	// WireguardEnabledV6 controls whether Wireguard is enabled for IPv6 (encapsulating IPv6 traffic over an IPv6 underlay network). [Default: false]
 	WireguardEnabledV6 *bool `json:"wireguardEnabledV6,omitempty"`
-
+	// WireguardThreadingEnabled controls whether Wireguard has NAPI threading enabled. [Default: false]
+	WireguardThreadingEnabled *bool `json:"wireguardThreadingEnabled,omitempty"`
 	// WireguardListeningPort controls the listening port used by IPv4 Wireguard. [Default: 51820]
 	WireguardListeningPort *int `json:"wireguardListeningPort,omitempty" validate:"omitempty,gt=0,lte=65535"`
 

felix/config/config_params.go

@@ -169,6 +169,7 @@ type Config struct {
 	WireguardMTUV6                 int           `config:"int;0"`
 	WireguardHostEncryptionEnabled bool          `config:"bool;false"`
 	WireguardPersistentKeepAlive   time.Duration `config:"seconds;0"`
+	WireguardThreadingEnabled      bool          `config:"bool;false"`
 
 	// nftables configuration.
 	NFTablesMode string `config:"oneof(Enabled,Disabled);Disabled"`

felix/dataplane/driver.go

@@ -297,6 +297,7 @@ func StartDataplaneDriver(configParams *config.Config,
 				RouteSource:         configParams.RouteSource,
 				EncryptHostTraffic:  configParams.WireguardHostEncryptionEnabled,
 				PersistentKeepAlive: configParams.WireguardPersistentKeepAlive,
+				ThreadedNAPI:        configParams.WireguardThreadingEnabled,
 				RouteSyncDisabled:   configParams.RouteSyncDisabled,
 			},
 			IPIPMTU:                        configParams.IpInIpMtu,

felix/fv/wireguard_test.go

@@ -100,7 +100,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported", []api
 			}
 		})
 
-		Describe(fmt.Sprintf("wireguardEnabledV4: %v, wireguardEnabledV6: %v, ", wireguardEnabledV4, wireguardEnabledV6), func() {
+		Describe(fmt.Sprintf("wireguardEnabledV4: %v, wireguardEnabledV6: %v", wireguardEnabledV4, wireguardEnabledV6), func() {
 			BeforeEach(func() {
 				// Run these tests only when the Host has Wireguard kernel module installed.
 				if os.Getenv("FELIX_FV_WIREGUARD_AVAILABLE") != "true" {
@@ -123,7 +123,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported", []api
 				infra = getInfra()
 				ipipEnabled := !BPFMode() || !wireguardEnabledV6
 				topologyOptions := wireguardTopologyOptions(
-					"CalicoIPAM", ipipEnabled, wireguardEnabledV4, wireguardEnabledV6,
+					"CalicoIPAM", ipipEnabled, wireguardEnabledV4, wireguardEnabledV6, false,
 					map[string]string{
 						"FELIX_DebugDisableLogDropping": "true",
 						"FELIX_DBG_WGBOOTSTRAP":         "true",
@@ -351,7 +351,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported", []api
 					_, err = client.FelixConfigurations().Update(ctx, fc, options.SetOptions{})
 					Expect(err).NotTo(HaveOccurred())
 
-					updateWireguardEnabledConfig(client, wireguardEnabledV4, wireguardEnabledV6)
+					updateWireguardEnabledConfig(client, wireguardEnabledV4, wireguardEnabledV6, false)
 
 					// New Wireguard device should appear with default MTU, etc.
 					for _, felix := range topologyContainers.Felixes {
@@ -398,6 +398,30 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported", []api
 					}
 				})
 
+				It("the Wireguard device should have napi threading set correctly", func() {
+					// transitions wireguard napi threading on then off
+					wireguardThreadingStates := []string{"0", "1", "0"}
+					for _, state := range wireguardThreadingStates {
+						stateBool, err := strconv.ParseBool(state)
+						Expect(err).NotTo(HaveOccurred())
+						updateWireguardEnabledConfig(client, wireguardEnabledV4, wireguardEnabledV6, stateBool)
+						for _, felix := range topologyContainers.Felixes {
+							if wireguardEnabledV4 {
+								Eventually(func() string {
+									s, _ := felix.ExecCombinedOutput("cat", fmt.Sprintf("/sys/class/net/%s/threaded", wireguardInterfaceNameDefault))
+									return s
+								}, "60s", "5s").Should(ContainSubstring(state))
+							}
+							if wireguardEnabledV6 {
+								Eventually(func() string {
+									s, _ := felix.ExecCombinedOutput("cat", fmt.Sprintf("/sys/class/net/%s/threaded", wireguardInterfaceNameV6Default))
+									return s
+								}, "60s", "5s").Should(ContainSubstring(state))
+							}
+						}
+					}
+				})
+
 				It("v3 node resource annotations should contain public-keys", func() {
 					for _, felix := range topologyContainers.Felixes {
 						if wireguardEnabledV4 {
@@ -1088,7 +1112,7 @@ var _ = infrastructure.DatastoreDescribe("WireGuard-Unsupported", []apiconfig.Da
 
 				infra = getInfra()
 				ipipEnabled := !BPFMode() || !wireguardEnabledV6
-				tc, _ = infrastructure.StartNNodeTopology(nodeCount, wireguardTopologyOptions("CalicoIPAM", ipipEnabled, wireguardEnabledV4, wireguardEnabledV6), infra)
+				tc, _ = infrastructure.StartNNodeTopology(nodeCount, wireguardTopologyOptions("CalicoIPAM", ipipEnabled, wireguardEnabledV4, wireguardEnabledV6, false), infra)
 
 				// Install a default profile that allows all ingress and egress, in the absence of any Policy.
 				infra.AddDefaultAllow()
@@ -1170,7 +1194,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported 3 node
 		}
 
 		infra = getInfra()
-		topologyOptions := wireguardTopologyOptions("CalicoIPAM", true, true, false)
+		topologyOptions := wireguardTopologyOptions("CalicoIPAM", true, true, false, false)
 		tc, client = infrastructure.StartNNodeTopology(nodeCount, topologyOptions, infra)
 
 		// To allow all ingress and egress, in absence of any Policy.
@@ -1442,7 +1466,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported 3-node
 		}
 
 		infra = getInfra()
-		topologyOptions := wireguardTopologyOptions("WorkloadIPs", false, true, false)
+		topologyOptions := wireguardTopologyOptions("WorkloadIPs", false, true, false, false)
 		tc, client = infrastructure.StartNNodeTopology(nodeCount, topologyOptions, infra)
 
 		// To allow all ingress and egress, in absence of any Policy.
@@ -1733,7 +1757,7 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ WireGuard-Supported 3-node
 
 // Setup cluster topology options.
 // mainly, enable Wireguard with delayed start option.
-func wireguardTopologyOptions(routeSource string, ipipEnabled, wireguardIPv4Enabled, wireguardIPv6Enabled bool, extraEnvs ...map[string]string) infrastructure.TopologyOptions {
+func wireguardTopologyOptions(routeSource string, ipipEnabled, wireguardIPv4Enabled, wireguardIPv6Enabled, wireguardThreadingEnabled bool, extraEnvs ...map[string]string) infrastructure.TopologyOptions {
 	topologyOptions := infrastructure.DefaultTopologyOptions()
 
 	// Waiting for calico-node to be ready.
@@ -1772,22 +1796,26 @@ func wireguardTopologyOptions(routeSource string, ipipEnabled, wireguardIPv4Enab
 	if wireguardIPv6Enabled {
 		felixConfig.Spec.WireguardEnabledV6 = &enabled
 	}
+	if wireguardThreadingEnabled {
+		felixConfig.Spec.WireguardThreadingEnabled = &enabled
+	}
 	topologyOptions.InitialFelixConfiguration = felixConfig
 
 	return topologyOptions
 }
 
 func disableWireguard(client clientv3.Interface) {
-	updateWireguardEnabledConfig(client, false, false)
+	updateWireguardEnabledConfig(client, false, false, false)
 }
 
-func updateWireguardEnabledConfig(client clientv3.Interface, valueV4, valueV6 bool) {
+func updateWireguardEnabledConfig(client clientv3.Interface, valueV4, valueV6, valueThreadingEnabled bool) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	felixConfig, err := client.FelixConfigurations().Get(ctx, "default", options.GetOptions{})
 	Expect(err).NotTo(HaveOccurred())
 	felixConfig.Spec.WireguardEnabled = &valueV4
 	felixConfig.Spec.WireguardEnabledV6 = &valueV6
+	felixConfig.Spec.WireguardThreadingEnabled = &valueThreadingEnabled
 	felixConfig, err = client.FelixConfigurations().Update(ctx, felixConfig, options.SetOptions{})
 	Expect(err).NotTo(HaveOccurred())
 }

felix/wireguard/config.go

@@ -33,4 +33,5 @@ type Config struct {
 	EncryptHostTraffic  bool
 	PersistentKeepAlive time.Duration
 	RouteSyncDisabled   bool
+	ThreadedNAPI        bool
 }

felix/wireguard/wireguard.go

@@ -18,6 +18,7 @@ package wireguard
 
 import (
 	"errors"
+	"fmt"
 	"net"
 	"os"
 	"sync"
@@ -1496,6 +1497,16 @@ func (w *Wireguard) ensureLink(netlinkClient netlinkshim.Interface) (bool, error
 		}
 	}
 
+	// Can only enable NAPI threading once the link is up
+	if attrs.Flags&net.FlagUp != 0 {
+		threadedNAPIBit := boolToBinaryString(w.config.ThreadedNAPI)
+		w.logCtx.WithField("flags", attrs.Flags).Infof("Set NAPI threading to %s for wireguard interface %s", threadedNAPIBit, w.interfaceName)
+		napiThreadedPath := fmt.Sprintf("/sys/class/net/%s/threaded", w.interfaceName)
+		if err := w.writeProcSys(napiThreadedPath, threadedNAPIBit); err != nil {
+			w.logCtx.WithError(err).Warnf("failed to set NAPI threading to %s for wireguard for interface %s", threadedNAPIBit, w.interfaceName)
+		}
+	}
+
 	// Track whether the interface is oper up or not. We halt programming when it is down.
 	return link.Attrs().Flags&net.FlagUp != 0, nil
 }
@@ -1836,3 +1847,10 @@ func writeProcSys(path, value string) error {
 	}
 	return nil
 }
+
+func boolToBinaryString(input bool) string {
+	if input {
+		return "1"
+	}
+	return "0"
+}

libcalico-go/lib/backend/syncersv1/updateprocessors/configurationprocessor_test.go

@@ -43,7 +43,7 @@ const (
 )
 
 const (
-	numBaseFelixConfigs = 147
+	numBaseFelixConfigs = 148
 )
 
 var _ = Describe("Test the generic configuration update processor and the concrete implementations", func() {