Add support for NF-based accelerated mode

Plumb bridgeID across CNI and network function APIs. Add per-VF
accelerated mode and orphaned VF representor recovery on shutdown.
Signed-off-by: Alkama Hasan <alkamah@marvell.com>

Author: alkama-hasan

dpu-cni/pkgs/cnitypes/cnitypes.go

@@ -130,6 +130,8 @@ type NetConf struct {
 	RuntimeConfig struct {
 		Mac string `json:"mac,omitempty"`
 	} `json:"runtimeConfig,omitempty"`
-	LogLevel string `json:"logLevel,omitempty"`
-	LogFile  string `json:"logFile,omitempty"`
+	LogLevel      string `json:"logLevel,omitempty"`
+	LogFile       string `json:"logFile,omitempty"`
+	BridgeID      string `json:"bridgeID,omitempty"`
+	IsAccelerated bool   `json:"isAccelerated,omitempty"`
 }

dpu-cni/pkgs/networkfn/networkfn.go

@@ -322,11 +322,18 @@ func CmdDel(req *cnitypes.PodRequest) error {
 	conf := req.CNIConf
 
 	if req.Netns == "" {
+		klog.Warning("CmdDel: netns is empty, device may be orphaned")
 		return nil
 	}
 
 	containerNs, err := ns.GetNS(req.Netns)
 	if err != nil {
+		if _, ok := err.(ns.NSPathNotExistErr); ok {
+			klog.Warningf("CmdDel: netns %q no longer exists for device %s — attempting recovery",
+				req.Netns, conf.DeviceID)
+			recoverOrphanedDevice(conf.DeviceID)
+			return nil
+		}
 		return fmt.Errorf("failed to open netns %q: %v", req.Netns, err)
 	}
 	defer containerNs.Close()
@@ -347,3 +354,41 @@ func CmdDel(req *cnitypes.PodRequest) error {
 
 	return nil
 }
+
+// recoverOrphanedDevice attempts to find a device that was left behind in a
+// now-destroyed namespace and restore it to the init namespace.
+// Hardware-backed VF representors survive namespace destruction (unlike veths
+// which are auto-cleaned by the kernel), so they may be found under the
+// temporary name assigned by moveLinkInNetNamespace with the original name
+// stored in the device's alias.
+// For veth pairs this is a safe no-op: the kernel already destroyed both ends.
+func recoverOrphanedDevice(deviceName string) {
+	if _, err := netlink.LinkByName(deviceName); err == nil {
+		klog.Infof("recoverOrphanedDevice: %s already in init namespace", deviceName)
+		return
+	}
+
+	links, err := netlink.LinkList()
+	if err != nil {
+		klog.Errorf("recoverOrphanedDevice: failed to list links: %v", err)
+		return
+	}
+	for _, link := range links {
+		if link.Attrs().Alias == deviceName {
+			klog.Infof("recoverOrphanedDevice: found %s under temp name %s, renaming back",
+				deviceName, link.Attrs().Name)
+			if err := netlink.LinkSetName(link, deviceName); err != nil {
+				klog.Errorf("recoverOrphanedDevice: failed to rename %s to %s: %v",
+					link.Attrs().Name, deviceName, err)
+				return
+			}
+			if err := netlink.LinkSetUp(link); err != nil {
+				klog.Warningf("recoverOrphanedDevice: failed to bring up %s: %v", deviceName, err)
+			}
+			return
+		}
+	}
+
+	klog.Warningf("recoverOrphanedDevice: %s not found in init namespace — "+
+		"device was likely a veth pair already cleaned up by the kernel", deviceName)
+}

examples/host-pod.yaml

@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: host-pod
+  namespace: default
+  annotations:
+    dpu.config.openshift.io/dpu-network: net1
+spec:
+  nodeSelector:
+    node-role.kubernetes.io/worker: ""
+  containers:
+    - name: app-container
+      image: ghcr.io/ovn-kubernetes/kubernetes-traffic-flow-tests:latest
+      resources:
+        requests:
+          openshift.io/dpunetwork-net1: "1"
+        limits:
+          openshift.io/dpunetwork-net1: "1"

examples/nf-pod.yaml

@@ -0,0 +1,23 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: nf-pod
+  namespace: default
+  annotations:
+    dpu.config.openshift.io/dpu-network: net1
+spec:
+  nodeSelector:
+    dpu.config.openshift.io/dpuside: "dpu"
+  containers:
+    - name: nf-container
+      image: ghcr.io/ovn-kubernetes/kubernetes-traffic-flow-tests:latest
+      securityContext:
+        capabilities:
+          add: ["NET_ADMIN", "NET_RAW"]
+      resources:
+        requests:
+          openshift.io/dpunetwork-net1: "4"
+          openshift.io/dpu-accelerated: "1"
+        limits:
+          openshift.io/dpunetwork-net1: "4"
+          openshift.io/dpu-accelerated: "1"

internal/daemon/device-plugin/deviceplugin.go

@@ -31,7 +31,7 @@ import (
 )
 
 const (
-	DefaultDpuResourceName = "openshift.io/dpu"
+	DefaultDpuResourceName  = "openshift.io/dpu"
 	AcceleratedResourceName = "openshift.io/dpu-accelerated"
 
 	acceleratedDevicePrefix = "accelerated:"
@@ -924,4 +924,3 @@ func vfRangesEqual(a, b []string) bool {
 	}
 	return true
 }
-

internal/daemon/dpusidemanager.go

@@ -21,11 +21,13 @@ import (
 	"github.com/openshift/dpu-operator/pkgs/vars"
 	pb "github.com/opiproject/opi-api/network/evpn-gw/v1alpha1/gen/go"
 	lifecycleapi "github.com/opiproject/opi-api/v1/gen/go/lifecycle/v1alpha1"
+	"github.com/vishvananda/netlink"
 	"google.golang.org/grpc"
 	emptypb "google.golang.org/protobuf/types/known/emptypb"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
 )
@@ -113,7 +115,12 @@ func NewDpuSideManager(vsp plugin.VendorPlugin, config *rest.Config, opts ...fun
 		opt(d)
 	}
 
-	d.dp = deviceplugin.NewDevicePlugin(vsp, true, d.pathManager)
+	if k8sClient, err := client.New(d.config, client.Options{Scheme: scheme.Scheme}); err != nil {
+		d.log.Error(err, "Failed to create Kubernetes client for device plugin; falling back to default-only registration")
+		d.dp = deviceplugin.NewDevicePluginManager(vsp, true, d.pathManager, nil)
+	} else {
+		d.dp = deviceplugin.NewDevicePluginManager(vsp, true, d.pathManager, k8sClient)
+	}
 
 	return d, nil
 }
@@ -149,12 +156,21 @@ func (d *DpuSideManager) cniCmdNfAddHandler(req *cnitypes.PodRequest) (*cni100.R
 		return nil, fmt.Errorf("SRIOV manager failed in add handler: %v", err)
 	}
 
-	d.macStore[req.Netns] = append(d.macStore[req.Netns], req.CNIConf.MAC)
-	if len(d.macStore[req.Netns]) == 2 {
-		d.log.Info("cniCmdNfAddHandler", "req.Netns", req.Netns)
-		macs := d.macStore[req.Netns]
-		d.vsp.CreateNetworkFunction(macs[0], macs[1])
+	bridgeID := req.CNIConf.BridgeID
+
+	if req.CNIConf.IsAccelerated {
+		d.log.Info("cniCmdNfAddHandler accelerated mode: calling CNF per VF", "mac", req.CNIConf.MAC, "bridgeID", bridgeID)
+		d.macStore[req.Netns] = append(d.macStore[req.Netns], req.CNIConf.MAC)
+		d.vsp.CreateNetworkFunction(req.CNIConf.MAC, "", bridgeID)
+	} else {
+		d.macStore[req.Netns] = append(d.macStore[req.Netns], req.CNIConf.MAC)
+		if len(d.macStore[req.Netns]) == 2 {
+			d.log.Info("cniCmdNfAddHandler", "req.Netns", req.Netns, "bridgeID", bridgeID)
+			macs := d.macStore[req.Netns]
+			d.vsp.CreateNetworkFunction(macs[0], macs[1], bridgeID)
+		}
 	}
+
 	d.log.Info("cniCmdNfAddHandler CmdAdd succeeded")
 	return res, nil
 }
@@ -166,19 +182,71 @@ func (d *DpuSideManager) cniCmdNfDelHandler(req *cnitypes.PodRequest) (*cni100.R
 		return nil, errors.New("SRIOV manager failed in del handler")
 	}
 
-	macs := d.macStore[req.Netns]
+	bridgeID := req.CNIConf.BridgeID
 
-	if len(macs) == 2 {
-		d.log.Info("cniCmdNfDelHandler", "req.Netns", req.Netns)
-		d.vsp.DeleteNetworkFunction(macs[0], macs[1])
+	if req.CNIConf.IsAccelerated {
+		macs := d.macStore[req.Netns]
+		mac := ""
+		if len(macs) > 0 {
+			mac = macs[len(macs)-1]
+			d.macStore[req.Netns] = macs[:len(macs)-1]
+		}
+		d.log.Info("cniCmdNfDelHandler accelerated mode: calling DNF per VF", "mac", mac, "bridgeID", bridgeID)
+		d.vsp.DeleteNetworkFunction(mac, "", bridgeID)
+	} else {
+		macs := d.macStore[req.Netns]
+		if len(macs) == 2 {
+			d.log.Info("cniCmdNfDelHandler", "req.Netns", req.Netns, "bridgeID", bridgeID)
+			d.vsp.DeleteNetworkFunction(macs[0], macs[1], bridgeID)
+		}
+		if len(macs) > 0 {
+			d.macStore[req.Netns] = macs[:len(macs)-1]
+		}
 	}
 
-	d.macStore[req.Netns] = macs[:len(macs)-1]
-
 	d.log.Info("cniCmdNfDelHandler CmdDel succeeded")
 	return nil, nil
 }
 
+// releaseNfDevices recovers any devices (VF representors or veths) that are
+// still inside NF pod namespaces during graceful shutdown. This must run
+// before the CNI server and VSP are stopped so the host can safely reset
+// sriov_numvfs without hitting a kernel D-state hang.
+func (d *DpuSideManager) releaseNfDevices() {
+	d.log.Info("releaseNfDevices: checking for devices still in pod namespaces")
+
+	devices, err := d.vsp.GetDevices()
+	if err != nil {
+		d.log.Error(err, "releaseNfDevices: failed to get device list from VSP")
+		return
+	}
+
+	for _, dev := range devices.Devices {
+		devName := dev.ID
+		if _, err := netlink.LinkByName(devName); err == nil {
+			continue
+		}
+
+		links, listErr := netlink.LinkList()
+		if listErr != nil {
+			d.log.Error(listErr, "releaseNfDevices: failed to list links")
+			return
+		}
+		for _, link := range links {
+			if link.Attrs().Alias == devName {
+				d.log.Info("releaseNfDevices: found device under temp name, restoring",
+					"tempName", link.Attrs().Name, "originalName", devName)
+				if err := netlink.LinkSetName(link, devName); err != nil {
+					d.log.Error(err, "releaseNfDevices: failed to rename", "device", devName)
+				} else if err := netlink.LinkSetUp(link); err != nil {
+					d.log.Error(err, "releaseNfDevices: failed to bring up", "device", devName)
+				}
+				break
+			}
+		}
+	}
+}
+
 func (d *DpuSideManager) Listen() (net.Listener, error) {
 	d.startedWg.Add(1)
 	d.log.Info("Starting DpuDaemon")
@@ -225,6 +293,7 @@ func (d *DpuSideManager) Serve(ctx context.Context, listener net.Listener) error
 	go func() {
 		<-ctx.Done()
 		d.log.Info("Context cancelled, shutting down servers")
+		d.releaseNfDevices()
 		d.server.Stop()
 		d.dp.Stop()
 		d.vsp.Close()

internal/daemon/dpusidemanager_test.go

@@ -36,7 +36,7 @@ func waitAllNodesDpuAllocatable(client client.Client) {
 		}
 		readyNodes := 0
 		for _, node := range latestNodes.Items {
-			allocatableQuantity, ok := node.Status.Allocatable[deviceplugin.DpuResourceName]
+			allocatableQuantity, ok := node.Status.Allocatable[deviceplugin.DefaultDpuResourceName]
 			if ok {
 				allocatable, _ := allocatableQuantity.AsInt64()
 				if allocatable > 0 {

internal/daemon/hostsidemanager.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type HostSideManager struct {
@@ -51,7 +52,7 @@ type HostSideManager struct {
 	dpListener         net.Listener
 }
 
-func (d *HostSideManager) CreateBridgePort(pf int, vf int, vlan int, mac string) (*pb.BridgePort, error) {
+func (d *HostSideManager) CreateBridgePort(pf int, vf int, vlan int, mac string, bridgeID string) (*pb.BridgePort, error) {
 	err := d.connectWithRetry()
 	if err != nil {
 		return nil, fmt.Errorf("Failed to connect with retry: %v", err)
@@ -69,8 +70,7 @@ func (d *HostSideManager) CreateBridgePort(pf int, vf int, vlan int, mac string)
 				Ptype:      1,
 				MacAddress: m,
 				LogicalBridges: []string{
-					// TODO: Remove +2
-					fmt.Sprintf("%d", vf+2),
+					bridgeID,
 				},
 			},
 		},
@@ -100,10 +100,16 @@ func NewHostSideManager(vsp plugin.VendorPlugin, opts ...func(*HostSideManager))
 		opt(h)
 	}
 
-	h.dp = deviceplugin.NewDevicePlugin(vsp, false, h.pathManager)
 	if h.config == nil {
 		h.config = ctrl.GetConfigOrDie()
 	}
+
+	if k8sClient, err := client.New(h.config, client.Options{Scheme: scheme.Scheme}); err != nil {
+		h.log.Error(err, "Failed to create Kubernetes client for device plugin; falling back to default-only registration")
+		h.dp = deviceplugin.NewDevicePluginManager(vsp, false, h.pathManager, nil)
+	} else {
+		h.dp = deviceplugin.NewDevicePluginManager(vsp, false, h.pathManager, k8sClient)
+	}
 	return h, nil
 }
 
@@ -193,11 +199,12 @@ func (d *HostSideManager) cniCmdAddHandler(req *cnitypes.PodRequest) (*cni100.Re
 	pf := 0
 	vf := req.CNIConf.VFID
 	mac := req.CNIConf.OrigVfState.EffectiveMAC
+	bridgeID := req.CNIConf.BridgeID
 	d.log.Info("addHandler", "CNIConf", req.CNIConf)
 	// TODO: fix setting Vlan based on network definition in CR
 	vlan := 2 // *req.CNIConf.Vlan
-	d.log.Info("addHandler", "pf", pf, "vf", vf, "mac", mac, "vlan", vlan)
-	_, err = d.CreateBridgePort(pf, vf, vlan, mac)
+	d.log.Info("addHandler", "pf", pf, "vf", vf, "mac", mac, "vlan", vlan, "bridgeID", bridgeID)
+	_, err = d.CreateBridgePort(pf, vf, vlan, mac, bridgeID)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to call CreateBridgePort: %v", err)
 	}

internal/daemon/hostsidemanager_test.go

@@ -63,11 +63,15 @@ func (v *DummyPlugin) DeleteBridgePort(deleteRequest *opi.DeleteBridgePortReques
 	return nil
 }
 
-func (g *DummyPlugin) CreateNetworkFunction(input string, output string) error {
+func (g *DummyPlugin) CreateNetworkFunction(input string, output string, bridgeID string) error {
 	return nil
 }
 
-func (g *DummyPlugin) DeleteNetworkFunction(input string, output string) error {
+func (g *DummyPlugin) DeleteNetworkFunction(input string, output string, bridgeID string) error {
+	return nil
+}
+
+func (g *DummyPlugin) SetDpuNetworkConfig(isAccelerated bool) error {
 	return nil
 }