gateway: Refactor gateway initialization and DPU host handling

- Improve DPU host gateway configuration with better IP handling
- Fix node interface address handling for DPU mode
- Clean up gateway interface selection logic for DPU and DPU-HOST
- Update node annotations for DPU host mode
- Remove redundant code and improve error handling
- Fix typos and improve code readability

This change simplifies the gateway initialization flow and improves
the handling of DPU (Data Processing Unit) host configurations.

The main idea is that the DPU in host mode is the one that dictates
how the network looks and the DPU side just follows.

Signed-off-by: Alin Serdean <[email protected]>

Author: aserdean

go-controller/pkg/node/bridgeconfig/bridgeconfig.go

@@ -244,12 +244,15 @@ func (b *BridgeConfiguration) UpdateInterfaceIPAddresses(node *corev1.Node) ([]*
 	// For DPU, here we need to use the DPU host's IP address which is the tenant cluster's
 	// host internal IP address instead of the DPU's external bridge IP address.
 	if config.OvnKubeNode.Mode == types.NodeModeDPU {
-		nodeAddrStr, err := util.GetNodePrimaryIP(node)
+		nodeIfAddr, err := util.GetNodePrimaryDPUHostAddrAnnotation(node)
 		if err != nil {
 			return nil, err
 		}
-		nodeAddr := net.ParseIP(nodeAddrStr)
-		if nodeAddr == nil {
+		// For DPU mode, we only support IPv4 for now.
+		nodeAddrStr := nodeIfAddr.IPv4
+
+		nodeAddr, _, err := net.ParseCIDR(nodeAddrStr)
+		if err != nil {
 			return nil, fmt.Errorf("failed to parse node IP address. %v", nodeAddrStr)
 		}
 		ifAddrs, err = nodeutil.GetDPUHostPrimaryIPAddresses(nodeAddr, ifAddrs)

go-controller/pkg/node/default_node_network_controller.go

@@ -965,8 +965,12 @@ func (nc *DefaultNodeNetworkController) Init(ctx context.Context) error {
 
 	// First part of gateway initialization. It will be completed by (nc *DefaultNodeNetworkController) Start()
 	if config.OvnKubeNode.Mode != types.NodeModeDPUHost {
+		// IPv6 is not supported in DPU enabled nodes, error out if ovnkube is not set in IPv4 mode
+		if config.IPv6Mode && config.OvnKubeNode.Mode == types.NodeModeDPU {
+			return fmt.Errorf("IPv6 mode is not supported on a DPU enabled node")
+		}
 		// Initialize gateway for OVS internal port or representor management port
-		gw, err := nc.initGatewayPreStart(subnets, nodeAnnotator, nc.mgmtPortController, nodeAddr)
+		gw, err := nc.initGatewayPreStart(subnets, nodeAnnotator, nc.mgmtPortController)
 		if err != nil {
 			return err
 		}
@@ -1059,7 +1063,7 @@ func (nc *DefaultNodeNetworkController) Start(ctx context.Context) error {
 			netdevName = netdevs[0]
 			config.Gateway.Interface = netdevName
 		}
-		err = nc.initGatewayDPUHost(nc.nodeAddress)
+		err = nc.initGatewayDPUHost(nc.nodeAddress, nodeAnnotator)
 		if err != nil {
 			return err
 		}

go-controller/pkg/node/gateway_init.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/vishvananda/netlink"
 
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	utilnet "k8s.io/utils/net"
 
@@ -195,7 +197,6 @@ func (nc *DefaultNodeNetworkController) initGatewayPreStart(
 	subnets []*net.IPNet,
 	nodeAnnotator kube.Annotator,
 	mgmtPort managementport.Interface,
-	kubeNodeIP net.IP,
 ) (*gateway, error) {
 
 	klog.Info("Initializing Gateway Functionality for Gateway PreStart")
@@ -219,13 +220,39 @@ func (nc *DefaultNodeNetworkController) initGatewayPreStart(
 		return nil, err
 	}
 
-	// For DPU need to use the host IP addr which currently is assumed to be K8s Node cluster
-	// internal IP address.
+	// For DPU mode, we need to use the host IP address which is stored as a Kubernetes
+	// node annotation rather than using the gateway interface IP addresses.
 	if config.OvnKubeNode.Mode == types.NodeModeDPU {
-		ifAddrs, err = nodeutil.GetDPUHostPrimaryIPAddresses(kubeNodeIP, ifAddrs)
+		// Retrieve the current node object from the Kubernetes API
+		var node *corev1.Node
+		if node, err = nc.watchFactory.GetNode(nc.name); err != nil {
+			return nil, fmt.Errorf("error retrieving node %s: %v", nc.name, err)
+		}
+
+		// Extract the primary DPU address annotation from the node
+		nodeIfAddr, err := util.GetNodePrimaryDPUHostAddrAnnotation(node)
 		if err != nil {
 			return nil, err
 		}
+		// For DPU mode, we only support IPv4 for now.
+		nodeAddrStr := nodeIfAddr.IPv4
+		if nodeAddrStr == "" {
+			return nil, fmt.Errorf("node primary DPU address annotation is empty for node %s", nc.name)
+		}
+
+		// Parse the IPv4 address string into IP and network components
+		nodeIP, nodeAddrs, err := net.ParseCIDR(nodeAddrStr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse node IP address %s: %v", nodeAddrStr, err)
+		}
+
+		// Set the parsed IP as the network address
+		nodeAddrs.IP = nodeIP
+
+		// Create a new slice and replace ifAddrs with the DPU host address
+		// This overrides the gateway interface addresses for DPU mode
+		var gwIps []*net.IPNet
+		ifAddrs = append(gwIps, nodeAddrs)
 	}
 
 	if err := util.SetNodePrimaryIfAddrs(nodeAnnotator, ifAddrs); err != nil {
@@ -359,43 +386,79 @@ func interfaceForEXGW(intfName string) string {
 	return intfName
 }
 
-func (nc *DefaultNodeNetworkController) initGatewayDPUHost(kubeNodeIP net.IP) error {
+func (nc *DefaultNodeNetworkController) initGatewayDPUHost(kubeNodeIP net.IP, nodeAnnotator kube.Annotator) error {
 	// A DPU host gateway is complementary to the shared gateway running
 	// on the DPU embedded CPU. it performs some initializations and
 	// watch on services for iptable rule updates and run a loadBalancerHealth checker
 	// Note: all K8s Node related annotations are handled from DPU.
 	klog.Info("Initializing Shared Gateway Functionality on DPU host")
 	var err error
 
-	// Force gateway interface to be the interface associated with kubeNodeIP
-	gwIntf, err := getInterfaceByIP(kubeNodeIP)
+	// Find the network interface that has the Kubernetes node IP assigned to it
+	// This interface will be used for DPU host gateway operations
+	kubeIntf, err := getInterfaceByIP(kubeNodeIP)
 	if err != nil {
 		return err
 	}
-	config.Gateway.Interface = gwIntf
 
-	_, gatewayIntf, err := getGatewayNextHops()
+	// Get all IP addresses (IPv4 and IPv6) configured on the detected interface
+	ifAddrs, err := nodeutil.GetNetworkInterfaceIPAddresses(kubeIntf)
 	if err != nil {
 		return err
 	}
 
-	ifAddrs, err := nodeutil.GetNetworkInterfaceIPAddresses(gatewayIntf)
-	if err != nil {
+	// Extract the IPv4 address from the interface addresses for node annotation
+	nodeIPNet, _ := util.MatchFirstIPNetFamily(false, ifAddrs)
+	nodeAddrSet := sets.New[string](nodeIPNet.String())
+
+	// If no gateway interface is explicitly configured, use the detected interface
+	if config.Gateway.Interface == "" {
+		config.Gateway.Interface = kubeIntf
+	}
+
+	// If a different gateway interface is configured than the one with used for the kubernetes node IP,
+	// get its addresses and add them to the node address set for routing purposes
+	if config.Gateway.Interface != kubeIntf {
+		ifAddrs, err = nodeutil.GetNetworkInterfaceIPAddresses(config.Gateway.Interface)
+		if err != nil {
+			return err
+		}
+		detectedIPNetv4, _ := util.MatchFirstIPNetFamily(false, ifAddrs)
+		nodeAddrSet.Insert(detectedIPNetv4.String())
+		// Use the configured interface for the masquerade route instead of the auto-detected one
+		kubeIntf = config.Gateway.Interface
+	}
+
+	// Set the primary DPU address annotation on the node with the interface addresses
+	if err := util.SetNodePrimaryDPUHostAddr(nodeAnnotator, ifAddrs); err != nil {
+		klog.Errorf("Unable to set primary IP net label on node, err: %v", err)
+		return err
+	}
+
+	// Set the host CIDRs annotation to include all detected network addresses
+	// This helps with routing decisions for traffic coming from the host
+	if err := util.SetNodeHostCIDRs(nodeAnnotator, nodeAddrSet); err != nil {
+		klog.Errorf("Unable to set host-cidrs on node, err: %v", err)
 		return err
 	}
 
+	// Apply all node annotations to the Kubernetes node object
+	if err := nodeAnnotator.Run(); err != nil {
+		return fmt.Errorf("failed to set node %s annotations: %w", nc.name, err)
+	}
+
 	// Delete stale masquerade resources if there are any. This is to make sure that there
 	// are no Linux resources with IP from old masquerade subnet when masquerade subnet
 	// gets changed as part of day2 operation.
-	if err := deleteStaleMasqueradeResources(gwIntf, nc.name, nc.watchFactory); err != nil {
+	if err := deleteStaleMasqueradeResources(kubeIntf, nc.name, nc.watchFactory); err != nil {
 		return fmt.Errorf("failed to remove stale masquerade resources: %w", err)
 	}
 
-	if err := setNodeMasqueradeIPOnExtBridge(gwIntf); err != nil {
-		return fmt.Errorf("failed to set the node masquerade IP on the ext bridge %s: %v", gwIntf, err)
+	if err := setNodeMasqueradeIPOnExtBridge(kubeIntf); err != nil {
+		return fmt.Errorf("failed to set the node masquerade IP on the ext bridge %s: %v", kubeIntf, err)
 	}
 
-	if err := addMasqueradeRoute(nc.routeManager, gwIntf, nc.name, ifAddrs, nc.watchFactory); err != nil {
+	if err := addMasqueradeRoute(nc.routeManager, kubeIntf, nc.name, ifAddrs, nc.watchFactory); err != nil {
 		return fmt.Errorf("failed to set the node masquerade route to OVN: %v", err)
 	}
 
@@ -404,7 +467,7 @@ func (nc *DefaultNodeNetworkController) initGatewayDPUHost(kubeNodeIP net.IP) er
 		return fmt.Errorf("failed to update masquerade subnet annotation on node: %s, error: %v", nc.name, err)
 	}
 
-	err = configureSvcRouteViaInterface(nc.routeManager, gatewayIntf, DummyNextHopIPs())
+	err = configureSvcRouteViaInterface(nc.routeManager, config.Gateway.Interface, DummyNextHopIPs())
 	if err != nil {
 		return err
 	}
@@ -430,7 +493,7 @@ func (nc *DefaultNodeNetworkController) initGatewayDPUHost(kubeNodeIP net.IP) er
 		gw.portClaimWatcher = portClaimWatcher
 	}
 
-	if err := addHostMACBindings(gwIntf); err != nil {
+	if err := addHostMACBindings(kubeIntf); err != nil {
 		return fmt.Errorf("failed to add MAC bindings for service routing")
 	}
 

go-controller/pkg/node/gateway_init_linux_test.go

@@ -725,6 +725,9 @@ func shareGatewayInterfaceDPUTest(app *cli.App, testNS ns.NetNS,
 		k := &kube.Kube{KClient: kubeFakeClient}
 
 		nodeAnnotator := kube.NewNodeAnnotator(k, existingNode.Name)
+		err = util.SetNodePrimaryDPUHostAddr(nodeAnnotator, ovntest.MustParseIPNets(nodeSubnet))
+		config.Gateway.RouterSubnet = nodeSubnet
+		Expect(err).NotTo(HaveOccurred())
 
 		err = util.SetNodeHostSubnetAnnotation(nodeAnnotator, ovntest.MustParseIPNets(nodeSubnet))
 		Expect(err).NotTo(HaveOccurred())
@@ -893,8 +896,11 @@ func shareGatewayInterfaceDPUHostTest(app *cli.App, testNS ns.NetNS, uplinkName,
 
 		err = testNS.Do(func(ns.NetNS) error {
 			defer GinkgoRecover()
+			k := &kube.Kube{KClient: kubeFakeClient}
+
+			nodeAnnotator := kube.NewNodeAnnotator(k, existingNode.Name)
 
-			err := nc.initGatewayDPUHost(net.ParseIP(hostIP))
+			err := nc.initGatewayDPUHost(net.ParseIP(hostIP), nodeAnnotator)
 			Expect(err).NotTo(HaveOccurred())
 
 			link, err := netlink.LinkByName(uplinkName)

go-controller/pkg/node/node_ip_handler_linux.go

@@ -65,27 +65,11 @@ func newAddressManagerInternal(nodeName string, k kube.Interface, mgmtPort manag
 	}
 	mgr.nodeAnnotator = kube.NewNodeAnnotator(k, nodeName)
 	if config.OvnKubeNode.Mode == types.NodeModeDPU {
-		var ifAddrs []*net.IPNet
-
-		// update k8s.ovn.org/host-cidrs
-		node, err := watchFactory.GetNode(nodeName)
-		if err != nil {
-			klog.Errorf("Failed to get node %s: %v", nodeName, err)
-			return nil
-		}
-		if useNetlink {
-			// get updated interface IP addresses for the gateway bridge
-			ifAddrs, err = gwBridge.UpdateInterfaceIPAddresses(node)
-			if err != nil {
-				klog.Errorf("Failed to obtain interface IP addresses for node %s: %v", nodeName, err)
-				return nil
-			}
-		}
-		if err = mgr.updateHostCIDRs(ifAddrs); err != nil {
+		if err := mgr.updateHostCIDRs(); err != nil {
 			klog.Errorf("Failed to update host-cidrs annotations on node %s: %v", nodeName, err)
 			return nil
 		}
-		if err = mgr.nodeAnnotator.Run(); err != nil {
+		if err := mgr.nodeAnnotator.Run(); err != nil {
 			klog.Errorf("Failed to set host-cidrs annotations on node %s: %v", nodeName, err)
 			return nil
 		}
@@ -286,7 +270,7 @@ func (c *addressManager) updateNodeAddressAnnotations() error {
 	}
 
 	// update k8s.ovn.org/host-cidrs
-	if err = c.updateHostCIDRs(ifAddrs); err != nil {
+	if err = c.updateHostCIDRs(); err != nil {
 		return err
 	}
 
@@ -316,14 +300,10 @@ func (c *addressManager) updateNodeAddressAnnotations() error {
 	return nil
 }
 
-func (c *addressManager) updateHostCIDRs(ifAddrs []*net.IPNet) error {
+func (c *addressManager) updateHostCIDRs() error {
 	if config.OvnKubeNode.Mode == types.NodeModeDPU {
-		// For DPU mode, here we need to use the DPU host's IP address which is the tenant cluster's
-		// host internal IP address instead.
-		// Currently we are only intentionally supporting IPv4 for DPU here.
-		nodeIPNetv4, _ := util.MatchFirstIPNetFamily(false, ifAddrs)
-		nodeAddrSet := sets.New[string](nodeIPNetv4.String())
-		return util.SetNodeHostCIDRs(c.nodeAnnotator, nodeAddrSet)
+		// For DPU mode, we don't need to update the host-cidrs annotation.
+		return nil
 	}
 
 	return util.SetNodeHostCIDRs(c.nodeAnnotator, c.cidrs)

go-controller/pkg/ovn/controller/services/lb_config.go

@@ -91,10 +91,16 @@ func makeNodeRouterTargetIPs(node *nodeInfo, c *lbConfig, hostMasqueradeIPV4, ho
 		targetIPsV6 = localIPsV6
 	}
 
+	// TODO: For all scenarios the lbAddress should be set to hostAddressesStr but this is breaking CI needs more investigation
+	lbAddresses := node.hostAddressesStr()
+	if config.OvnKubeNode.Mode == types.NodeModeFull {
+		lbAddresses = node.l3gatewayAddressesStr()
+	}
+
 	// Any targets local to the node need to have a special
 	// harpin IP added, but only for the router LB
-	targetIPsV4, v4Updated := util.UpdateIPsSlice(targetIPsV4, node.l3gatewayAddressesStr(), []string{hostMasqueradeIPV4})
-	targetIPsV6, v6Updated := util.UpdateIPsSlice(targetIPsV6, node.l3gatewayAddressesStr(), []string{hostMasqueradeIPV6})
+	targetIPsV4, v4Updated := util.UpdateIPsSlice(targetIPsV4, lbAddresses, []string{hostMasqueradeIPV4})
+	targetIPsV6, v6Updated := util.UpdateIPsSlice(targetIPsV6, lbAddresses, []string{hostMasqueradeIPV6})
 
 	// Local endpoints are a subset of cluster endpoints, so it is enough to compare their length
 	v4Changed = len(targetIPsV4) != len(c.clusterEndpoints.V4IPs) || v4Updated

go-controller/pkg/ovn/default_network_controller.go

@@ -949,6 +949,7 @@ func (h *defaultNetworkControllerEventHandler) UpdateResource(oldObj, newObj int
 		zoneClusterChanged := h.oc.nodeZoneClusterChanged(oldNode, newNode, newNodeIsLocalZoneNode, types.DefaultNetworkName)
 		nodeSubnetChange := nodeSubnetChanged(oldNode, newNode, types.DefaultNetworkName)
 		nodeEncapIPsChanged := util.NodeEncapIPsChanged(oldNode, newNode)
+		nodePrimaryDPUHostAddrChanged := util.NodePrimaryDPUHostAddrAnnotationChanged(oldNode, newNode)
 
 		var aggregatedErrors []error
 		if newNodeIsLocalZoneNode {
@@ -1006,11 +1007,18 @@ func (h *defaultNetworkControllerEventHandler) UpdateResource(oldObj, newObj int
 			// Also check if node subnet changed, so static routes are properly set
 			// Also check if the node is used to be a hybrid overlay node
 			syncZoneIC = syncZoneIC || h.oc.isLocalZoneNode(oldNode) || nodeSubnetChange || zoneClusterChanged ||
-				switchToOvnNode || nodeEncapIPsChanged
+				switchToOvnNode || nodeEncapIPsChanged || nodePrimaryDPUHostAddrChanged
 			if syncZoneIC {
 				klog.Infof("Node %q in remote zone %q, network %q, needs interconnect zone sync up. Zone cluster changed: %v",
 					newNode.Name, util.GetNodeZone(newNode), h.oc.GetNetworkName(), zoneClusterChanged)
 			}
+			// Reprovisioning the DPU (including OVS), which is pinned to a host, will change the system ID but not the node.
+			if config.OvnKubeNode.Mode == types.NodeModeDPU && nodeChassisChanged(oldNode, newNode) {
+				if err := h.oc.zoneChassisHandler.DeleteRemoteZoneNode(oldNode); err != nil {
+					aggregatedErrors = append(aggregatedErrors, err)
+				}
+				syncZoneIC = true
+			}
 			if err := h.oc.addUpdateRemoteNodeEvent(newNode, syncZoneIC); err != nil {
 				aggregatedErrors = append(aggregatedErrors, err)
 			}

go-controller/pkg/util/node_annotations.go

@@ -97,6 +97,9 @@ const (
 	// OVNNodeHostCIDRs is used to track the different host IP addresses and subnet masks on the node
 	OVNNodeHostCIDRs = "k8s.ovn.org/host-cidrs"
 
+	// OVNNodePrimaryDPUHostAddr is used to track the primary DPU host address on the node
+	OVNNodePrimaryDPUHostAddr = "k8s.ovn.org/primary-dpu-host-addr"
+
 	// OVNNodeSecondaryHostEgressIPs contains EgressIP addresses that aren't managed by OVN. The EIP addresses are assigned to
 	// standard linux interfaces and not interfaces of type OVS.
 	OVNNodeSecondaryHostEgressIPs = "k8s.ovn.org/secondary-host-egress-ips"
@@ -1534,3 +1537,39 @@ func ParseNodeEncapIPsAnnotation(node *corev1.Node) ([]string, error) {
 func NodeEncapIPsChanged(oldNode, newNode *corev1.Node) bool {
 	return oldNode.Annotations[OVNNodeEncapIPs] != newNode.Annotations[OVNNodeEncapIPs]
 }
+
+// SetNodePrimaryDPUHostAddr sets the primary DPU host address annotation on a node
+func SetNodePrimaryDPUHostAddr(nodeAnnotator kube.Annotator, ifAddrs []*net.IPNet) error {
+	nodeIPNetv4, _ := MatchFirstIPNetFamily(false, ifAddrs)
+	nodeIPNetv6, _ := MatchFirstIPNetFamily(true, ifAddrs)
+
+	ifAddrAnnotation := ifAddr{}
+	if nodeIPNetv4 != nil {
+		ifAddrAnnotation.IPv4 = nodeIPNetv4.String()
+	}
+	if nodeIPNetv6 != nil {
+		ifAddrAnnotation.IPv6 = nodeIPNetv6.String()
+	}
+	return nodeAnnotator.Set(OVNNodePrimaryDPUHostAddr, ifAddrAnnotation)
+}
+
+// NodePrimaryDPUHostAddrAnnotationChanged returns true if the primary DPU host address annotation changed
+func NodePrimaryDPUHostAddrAnnotationChanged(oldNode, newNode *corev1.Node) bool {
+	return oldNode.Annotations[OVNNodePrimaryDPUHostAddr] != newNode.Annotations[OVNNodePrimaryDPUHostAddr]
+}
+
+// GetNodePrimaryDPUHostAddrAnnotation returns the raw primary DPU host address annotation from a node
+func GetNodePrimaryDPUHostAddrAnnotation(node *corev1.Node) (*ifAddr, error) {
+	addrAnnotation, ok := node.Annotations[OVNNodePrimaryDPUHostAddr]
+	if !ok {
+		return nil, newAnnotationNotSetError("%s annotation not found for node %q", OVNNodePrimaryDPUHostAddr, node.Name)
+	}
+	nodeIfAddr := &ifAddr{}
+	if err := json.Unmarshal([]byte(addrAnnotation), nodeIfAddr); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal annotation: %s for node %q, err: %v", OVNNodePrimaryDPUHostAddr, node.Name, err)
+	}
+	if nodeIfAddr.IPv4 == "" && nodeIfAddr.IPv6 == "" {
+		return nil, fmt.Errorf("node: %q does not have any IP information set", node.Name)
+	}
+	return nodeIfAddr, nil
+}