Merge pull request #5307 from aserdean/switch_dpu_annotation

gateway: Refactor gateway initialization and DPU host handling

Author: trozet

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
+}