[PERFSCALE-4024] [RoCE] ib_write_bw support

README.md

@@ -4,23 +4,24 @@ A tool to run network performance tests in Kubernetes clusters.
 
 ## Overview
 
-`k8s-netperf` is a comprehensive network performance testing tool designed specifically for Kubernetes environments. It supports multiple benchmark tools (netperf, iperf3, uperf) and can test various network scenarios including pod-to-pod, host networking, and cross-availability zone communications.
+`k8s-netperf` is a comprehensive network performance testing tool designed specifically for Kubernetes environments. It supports multiple benchmark tools (netperf, iperf3, uperf and ib_write_bw) and can test various network scenarios including pod-to-pod, host networking, and cross-availability zone communications.
 
 ### Supported Benchmarks
 
-| Tool    | Test       | Status  |
-| ------- | ---------- | ------- |
-| netperf | TCP_STREAM | Working |
-| netperf | UDP_STREAM | Working |
-| netperf | TCP_RR     | Working |
-| netperf | UDP_RR     | Working |
-| netperf | TCP_CRR    | Working |
-| uperf   | TCP_STREAM | Working |
-| uperf   | UDP_STREAM | Working |
-| uperf   | TCP_RR     | Working |
-| uperf   | UDP_RR     | Working |
-| iperf3  | TCP_STREAM | Working |
-| iperf3  | UDP_STREAM | Working |
+| Tool        | Test       | Status  |
+| ----------- | ---------- | ------- |
+| netperf     | TCP_STREAM | Working |
+| netperf     | UDP_STREAM | Working |
+| netperf     | TCP_RR     | Working |
+| netperf     | UDP_RR     | Working |
+| netperf     | TCP_CRR    | Working |
+| uperf       | TCP_STREAM | Working |
+| uperf       | UDP_STREAM | Working |
+| uperf       | TCP_RR     | Working |
+| uperf       | UDP_RR     | Working |
+| iperf3      | TCP_STREAM | Working |
+| iperf3      | UDP_STREAM | Working |
+| ib_write_bw | UDP_STREAM | Working |
 
 ## Quick Start
 
@@ -44,7 +45,7 @@ For detailed documentation, please refer to the following guides:
 
 ## Features
 
-- **Multiple benchmark tools**: Support for netperf, iperf3, and uperf
+- **Multiple benchmark tools**: Support for netperf, iperf3, uperf and ib_write_bw
 - **Flexible test scenarios**: Pod-to-pod, host networking, cross-AZ testing
 - **Virtual Machine support**: Test with KubeVirt VMs
 - **Advanced networking**: User Defined Networks (UDN) and bridge interfaces

cmd/k8s-netperf/k8s-netperf.go

@@ -41,6 +41,7 @@ var (
 	netperf          bool
 	iperf3           bool
 	uperf            bool
+	ibWriteBw        bool
 	udnl2            bool
 	udnl3            bool
 	udnPluginBinding string
@@ -86,9 +87,17 @@ var rootCmd = &cobra.Command{
 			fmt.Println("OS/Arch:", cmdVersion.OsArch)
 			os.Exit(0)
 		}
-		if !(uperf || netperf || iperf3) {
+		if !(uperf || netperf || iperf3 || ibWriteBw) {
 			log.Fatalf("😭 At least one driver needs to be enabled")
 		}
+		if ibWriteBw && (!privileged || !hostNetOnly) {
+			log.Fatalf("😭 ib_write_bw driver requires both --privileged and --hostNet flags")
+		}
+
+		// If a specific driver is explicitly requested, disable the default netperf driver
+		if (iperf3 || uperf || ibWriteBw) && !cmd.Flags().Changed("netperf") {
+			netperf = false
+		}
 		uid := ""
 		if len(id) > 0 {
 			uid = id
@@ -245,6 +254,28 @@ var rootCmd = &cobra.Command{
 			}
 		}
 
+		// Determine requested drivers BEFORE building SUT
+		var requestedDrivers []string
+		if netperf {
+			requestedDrivers = append(requestedDrivers, "netperf")
+		}
+		if uperf {
+			requestedDrivers = append(requestedDrivers, "uperf")
+		}
+		if iperf3 {
+			requestedDrivers = append(requestedDrivers, "iperf3")
+		}
+		if ibWriteBw {
+			requestedDrivers = append(requestedDrivers, "ib_write_bw")
+		}
+
+		// Set requested drivers BEFORE BuildSUT
+		s.RequestedDrivers = requestedDrivers
+
+		// Debug: Print requested drivers
+		log.Debugf("🔥 Requested drivers: %v", requestedDrivers)
+		log.Debugf("🔥 netperf=%v, iperf3=%v, uperf=%v, ibWriteBw=%v", netperf, iperf3, uperf, ibWriteBw)
+
 		// Build the SUT (Deployments)
 		err = k8s.BuildSUT(client, &s)
 		if err != nil {
@@ -262,16 +293,6 @@ var rootCmd = &cobra.Command{
 			acrossAZ = true
 		}
 		time.Sleep(5 * time.Second) // Wait some seconds to ensure service is ready
-		var requestedDrivers []string
-		if netperf {
-			requestedDrivers = append(requestedDrivers, "netperf")
-		}
-		if uperf {
-			requestedDrivers = append(requestedDrivers, "uperf")
-		}
-		if iperf3 {
-			requestedDrivers = append(requestedDrivers, "iperf3")
-		}
 
 		// Run through each test
 		if !s.VM {
@@ -512,6 +533,8 @@ func executeWorkload(nc config.Config,
 			serverIP = s.IperfService.Spec.ClusterIP
 		} else if driverName == "uperf" {
 			serverIP = s.UperfService.Spec.ClusterIP
+		} else if driverName == "ib_write_bw" {
+			serverIP = s.ServerHost.Items[0].Status.PodIP
 		} else {
 			serverIP = s.NetperfService.Spec.ClusterIP
 		}
@@ -578,6 +601,9 @@ func executeWorkload(nc config.Config,
 		} else if driverName == "uperf" {
 			driver = drivers.NewDriver("uperf")
 			npr.Driver = "uperf"
+		} else if driverName == "ib_write_bw" {
+			driver = drivers.NewDriver("ib_write_bw")
+			npr.Driver = "ib_write_bw"
 		} else {
 			driver = drivers.NewDriver("netperf")
 			npr.Driver = "netperf"
@@ -634,6 +660,7 @@ func main() {
 	rootCmd.Flags().BoolVar(&netperf, "netperf", true, "Use netperf as load driver")
 	rootCmd.Flags().BoolVar(&iperf3, "iperf", false, "Use iperf3 as load driver")
 	rootCmd.Flags().BoolVar(&uperf, "uperf", false, "Use uperf as load driver")
+	rootCmd.Flags().BoolVar(&ibWriteBw, "ib-write-bw", false, "Use ib_write_bw as load driver (requires --privileged and --hostNet)")
 	rootCmd.Flags().BoolVar(&clean, "clean", true, "Clean-up resources created by k8s-netperf")
 	rootCmd.Flags().BoolVar(&json, "json", false, "Instead of human-readable output, return JSON to stdout")
 	rootCmd.Flags().BoolVar(&nl, "local", false, "Run network performance tests with Server-Pods/Client-Pods on the same Node")

docs/advanced-usage.md

@@ -41,7 +41,7 @@ $ k8s-netperf --udnl2 --vm --udnPluginBinding=l2bridge
 > Warning! Support of k8s Services with UDN is not fully supported yet, you may faced inconsistent results when using a service in your tests. 
 
 ## Using a Linux Bridge Interface
-When using `--bridge`, a NetworkAttachmentDefinition defining a bridge interface is attached to the VMs and is used for the test. It requires the name of the bridge as it is defined in the NetworkNodeConfigurationPolicy, NMstate operator is required. 
+When using `--bridge`, a NetworkAttachmentDefinition defining a bridge interface is attached to the VMs and is used for the test. It requires the name of the bridge as it is defined in the NetworkNodeConfigurationPolicy, NMstate operator is required.
 
 For example:
 ```yaml
@@ -82,4 +82,11 @@ k8s-netperf --vm --bridge br0 --bridgeNetwork /path/to/my/bridgeConfig.json
 If your use case requires running pods with privileged security context, use the `--privileged` flag:
 ```
 $ k8s-netperf --privileged
+```
+
+## RoCEv2 testing
+
+Using the `ib_write_bw` driver, your hardware should include RDMA devices:
+```
+$ k8s-netperf --ib-write-bw --privileged --hostNet
 ```

pkg/config/config.go

@@ -51,6 +51,7 @@ type PerfScenarios struct {
 	Sockets             uint32
 	Cores               uint32
 	Threads             uint32
+	RequestedDrivers    []string
 	ServerNodeInfo      metrics.NodeInfo
 	ClientNodeInfo      metrics.NodeInfo
 	Client              apiv1.PodList

pkg/drivers/driver.go

@@ -41,6 +41,10 @@ func NewDriver(driverName string) Driver {
 		return &uperf{
 			driverName: driverName,
 		}
+	case "ib_write_bw":
+		return &ibWriteBw{
+			driverName: driverName,
+		}
 	default:
 		return &netperf{
 			driverName: driverName,

pkg/drivers/ibwritebw.go

@@ -0,0 +1,181 @@
+package drivers
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	apiv1 "k8s.io/api/core/v1"
+
+	"github.com/cloud-bulldozer/k8s-netperf/pkg/config"
+	"github.com/cloud-bulldozer/k8s-netperf/pkg/k8s"
+	log "github.com/cloud-bulldozer/k8s-netperf/pkg/logging"
+	"github.com/cloud-bulldozer/k8s-netperf/pkg/sample"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+var IbWriteBw ibWriteBw
+
+func init() {
+	IbWriteBw = ibWriteBw{
+		driverName: "ib_write_bw",
+	}
+}
+
+type ibWriteBw struct {
+	driverName string
+}
+
+// IsTestSupported determines if the test is supported for ib_write_bw driver
+func (i *ibWriteBw) IsTestSupported(test string) bool {
+	// ib_write_bw only supports UDP_STREAM profile
+	return strings.ToUpper(test) == "UDP_STREAM"
+}
+
+// Run will invoke ib_write_bw in a client container
+func (i *ibWriteBw) Run(c *kubernetes.Clientset,
+	rc rest.Config,
+	nc config.Config,
+	client apiv1.PodList,
+	serverIP string, perf *config.PerfScenarios) (bytes.Buffer, error) {
+	var stdout, stderr bytes.Buffer
+	pod := client.Items[0]
+	clientIp := pod.Status.PodIP
+
+	if perf.Udn {
+		if udnIp, _ := k8s.ExtractUdnIp(pod); udnIp != "" {
+			clientIp = udnIp
+		}
+	} else if perf.BridgeNetwork != "" {
+		if bridgeClientIp, err := k8s.ExtractBridgeIp(pod, perf.BridgeNetwork, perf.BridgeNamespace); err == nil {
+			clientIp = bridgeClientIp
+		}
+	}
+	log.Debugf("🔥 Client (%s,%s) starting ib_write_bw against server: %s", pod.Name, clientIp, serverIP)
+	config.Show(nc, i.driverName)
+
+	// ib_write_bw client command: "ib_write_bw -d mlx5_0 -x 3 -F $server_ip"
+	cmd := []string{"stdbuf", "-oL", "-eL", "ib_write_bw", "-d", "mlx5_0", "-x", "3", "-F", serverIP}
+
+	// Add duration if specified (ib_write_bw uses -D for duration in seconds)
+	if nc.Duration > 0 {
+		cmd = append(cmd, "-D", fmt.Sprint(nc.Duration))
+	}
+
+	log.Debug(cmd)
+	if !perf.VM {
+		req := c.CoreV1().RESTClient().
+			Post().
+			Namespace(pod.Namespace).
+			Resource("pods").
+			Name(pod.Name).
+			SubResource("exec").
+			VersionedParams(&apiv1.PodExecOptions{
+				Container: pod.Spec.Containers[0].Name,
+				Command:   cmd,
+				Stdin:     false,
+				Stdout:    true,
+				Stderr:    true,
+				TTY:       true,
+			}, scheme.ParameterCodec)
+		exec, err := remotecommand.NewSPDYExecutor(&rc, "POST", req.URL())
+		if err != nil {
+			return stdout, err
+		}
+		// Connect this process' std{in,out,err} to the remote shell process.
+		err = exec.StreamWithContext(context.Background(), remotecommand.StreamOptions{
+			Stdin:  nil,
+			Stdout: &stdout,
+			Stderr: &stderr,
+		})
+		if err != nil {
+			return stdout, err
+		}
+	} else {
+		retry := 3
+		present := false
+		sshclient, err := k8s.SSHConnect(perf)
+		if err != nil {
+			return stdout, err
+		}
+		for i := 0; i <= retry; i++ {
+			log.Debug("⏰ Waiting for ib_write_bw to be present on VM")
+			_, err = sshclient.Run("until ib_write_bw -h; do sleep 30; done")
+			if err == nil {
+				present = true
+				break
+			}
+			time.Sleep(10 * time.Second)
+		}
+		if !present {
+			sshclient.Close()
+			return stdout, fmt.Errorf("ib_write_bw binary is not present on the VM")
+		}
+		var stdoutBytes []byte
+		ran := false
+		for i := 0; i <= retry; i++ {
+			stdoutBytes, err = sshclient.Run(strings.Join(cmd[:], " "))
+			if err == nil {
+				ran = true
+				break
+			}
+			log.Debugf("Failed running command %s", err)
+			log.Debugf("⏰ Retrying ib_write_bw command -- cloud-init still finishing up")
+			time.Sleep(60 * time.Second)
+		}
+		sshclient.Close()
+		if !ran {
+			return *bytes.NewBuffer(stdoutBytes), fmt.Errorf("Unable to run ib_write_bw")
+		}
+		stdout = *bytes.NewBuffer(stdoutBytes)
+	}
+
+	log.Debug(strings.TrimSpace(stdout.String()))
+	return stdout, nil
+}
+
+// ParseResults accepts the stdout from the execution of ib_write_bw benchmark.
+// It will return a Sample struct or error
+func (i *ibWriteBw) ParseResults(stdout *bytes.Buffer, _ config.Config) (sample.Sample, error) {
+	sample := sample.Sample{}
+	sample.Driver = i.driverName
+	sample.Metric = "MiB/s"
+
+	output := stdout.String()
+	lines := strings.Split(output, "\n")
+
+	// Look for the results line that contains the bandwidth data
+	// Format: " 65536      130030           0.00               2708.88              0.043342"
+	// We want the 4th column which is "BW average[MiB/sec]"
+	for _, line := range lines {
+		line = strings.TrimSpace(line)
+
+		// Skip header lines and empty lines
+		if strings.Contains(line, "#bytes") || strings.Contains(line, "---") || line == "" {
+			continue
+		}
+
+		// Look for data lines with numeric values
+		fields := strings.Fields(line)
+		if len(fields) >= 4 {
+			// Check if first field is numeric (bytes)
+			if _, err := fmt.Sscanf(fields[0], "%d", new(int)); err == nil {
+				// Parse the BW average field (4th column)
+				var bwAverage float64
+				if _, err := fmt.Sscanf(fields[3], "%f", &bwAverage); err == nil {
+					sample.Throughput = bwAverage
+					log.Debugf("Parsed ib_write_bw BW average: %.2f MiB/s", bwAverage)
+					return sample, nil
+				}
+			}
+		}
+	}
+
+	log.Debugf("Failed to parse ib_write_bw output: %s", output)
+	return sample, fmt.Errorf("failed to parse BW average from ib_write_bw output")
+}