Merge pull request #5070 from trozet/fix_cm_node

Limit cluster manager node allocator updates/logs

Author: trozet

go-controller/pkg/clustermanager/clustermanager.go

@@ -25,11 +25,6 @@ import (
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
 )
 
-const (
-	// ID of the default network.
-	defaultNetworkID = 0
-)
-
 // ClusterManager structure is the object which manages the cluster nodes.
 // It creates a default network controller for the default network and a
 // secondary network cluster controller manager to manage the multi networks.

go-controller/pkg/clustermanager/clustermanager_test.go

@@ -123,23 +123,6 @@ var _ = ginkgo.Describe("Cluster Manager", func() {
 					}, 2).Should(gomega.HaveLen(1))
 				}
 
-				// Check that the network id 0 is allocated for the default network
-				for _, n := range nodes {
-					gomega.Eventually(func() error {
-						updatedNode, err := fakeClient.KubeClient.CoreV1().Nodes().Get(context.TODO(), n.Name, metav1.GetOptions{})
-						if err != nil {
-							return err
-						}
-
-						networkId, err := util.ParseNetworkIDAnnotation(updatedNode, "default")
-						if err != nil {
-							return fmt.Errorf("expected node network id annotation for node %s to have been allocated", n.Name)
-						}
-
-						gomega.Expect(networkId).To(gomega.Equal(0))
-						return nil
-					}).ShouldNot(gomega.HaveOccurred())
-				}
 				return nil
 			}
 

go-controller/pkg/clustermanager/network_cluster_controller.go

@@ -120,7 +120,7 @@ func newDefaultNetworkClusterController(netInfo util.NetInfo, ovnClient *util.OV
 	// defaiult network
 	networkIDAllocator := id.NewIDAllocator(types.DefaultNetworkName, 1)
 	// Reserve the id 0 for the default network.
-	err := networkIDAllocator.ReserveID(types.DefaultNetworkName, defaultNetworkID)
+	err := networkIDAllocator.ReserveID(types.DefaultNetworkName, types.DefaultNetworkID)
 	if err != nil {
 		panic(fmt.Errorf("could not reserve default network ID: %w", err))
 	}
@@ -175,7 +175,7 @@ func (ncc *networkClusterController) init() error {
 	if util.DoesNetworkRequireTunnelIDs(ncc.GetNetInfo()) {
 		ncc.tunnelIDAllocator = id.NewIDAllocator(ncc.GetNetworkName(), types.MaxLogicalPortTunnelKey)
 		// Reserve the id 0. We don't want to assign this id to any of the pods or nodes.
-		if err = ncc.tunnelIDAllocator.ReserveID("zero", util.NoID); err != nil {
+		if err = ncc.tunnelIDAllocator.ReserveID("zero", types.NoTunnelID); err != nil {
 			return err
 		}
 		if util.IsNetworkSegmentationSupportEnabled() && ncc.IsPrimaryNetwork() {
@@ -196,7 +196,7 @@ func (ncc *networkClusterController) init() error {
 							node.Name, ncc.GetNetworkName(), err)
 					}
 				}
-				if tunnelID != util.InvalidID {
+				if tunnelID != types.InvalidID {
 					if err := ncc.tunnelIDAllocator.ReserveID(ncc.GetNetworkName()+"_"+node.Name, tunnelID); err != nil {
 						return fmt.Errorf("unable to reserve id for network %s, node %s: %w", ncc.GetNetworkName(), node.Name, err)
 					}
@@ -489,6 +489,8 @@ type networkClusterControllerEventHandler struct {
 	objType  reflect.Type
 	ncc      *networkClusterController
 	syncFunc func([]interface{}) error
+
+	nodeSyncFailed sync.Map
 }
 
 func (h *networkClusterControllerEventHandler) FilterOutResource(_ interface{}) bool {
@@ -520,6 +522,9 @@ func (h *networkClusterControllerEventHandler) AddResource(obj interface{}, _ bo
 		err = h.ncc.nodeAllocator.HandleAddUpdateNodeEvent(node)
 		if err == nil {
 			h.clearInitialNodeNetworkUnavailableCondition(node)
+			h.nodeSyncFailed.Delete(node.Name)
+		} else {
+			h.nodeSyncFailed.Store(node.Name, true)
 		}
 		statusErr := h.ncc.updateNetworkStatus(node.Name, err)
 		joinedErr := errors.Join(err, statusErr)
@@ -557,19 +562,35 @@ func (h *networkClusterControllerEventHandler) UpdateResource(oldObj, newObj int
 			return err
 		}
 	case factory.NodeType:
-		node, ok := newObj.(*corev1.Node)
+		oldNode, ok := oldObj.(*corev1.Node)
+		if !ok {
+			return fmt.Errorf("could not cast %T object to *corev1.Node", oldObj)
+		}
+		newNode, ok := newObj.(*corev1.Node)
 		if !ok {
 			return fmt.Errorf("could not cast %T object to *corev1.Node", newObj)
 		}
-		err = h.ncc.nodeAllocator.HandleAddUpdateNodeEvent(node)
+		_, nodeFailed := h.nodeSyncFailed.Load(newNode.GetName())
+		// Note: (trozet) It might be pedantic to check if the NeedsNodeAllocation. This assumes one of the following:
+		// 1. we missed an add event (bug in kapi informer code)
+		// 2. a user removed the annotation on the node
+		// Either way to play it safe for now do a partial json unmarshal check
+		if !nodeFailed && util.NoHostSubnet(oldNode) != util.NoHostSubnet(newNode) && !h.ncc.nodeAllocator.NeedsNodeAllocation(newNode) {
+			// no other node updates would require us to reconcile again
+			return nil
+		}
+		err = h.ncc.nodeAllocator.HandleAddUpdateNodeEvent(newNode)
 		if err == nil {
-			h.clearInitialNodeNetworkUnavailableCondition(node)
+			h.clearInitialNodeNetworkUnavailableCondition(newNode)
+			h.nodeSyncFailed.Delete(newNode.GetName())
+		} else {
+			h.nodeSyncFailed.Store(newNode.Name, true)
 		}
-		statusErr := h.ncc.updateNetworkStatus(node.Name, err)
+		statusErr := h.ncc.updateNetworkStatus(newNode.Name, err)
 		joinedErr := errors.Join(err, statusErr)
 		if joinedErr != nil {
 			klog.Infof("Cluster Manager Network Controller %q: Node update failed for %s, will try again later: %v",
-				h.ncc.GetNetworkName(), node.Name, err)
+				h.ncc.GetNetworkName(), newNode.Name, err)
 			return err
 		}
 	case factory.IPAMClaimsType:
@@ -600,7 +621,12 @@ func (h *networkClusterControllerEventHandler) DeleteResource(obj, _ interface{}
 		}
 		err := h.ncc.nodeAllocator.HandleDeleteNode(node)
 		statusErr := h.ncc.updateNetworkStatus(node.Name, err)
-		return errors.Join(err, statusErr)
+		jErr := errors.Join(err, statusErr)
+		if jErr != nil {
+			return jErr
+		}
+		h.nodeSyncFailed.Delete(node.Name)
+		return nil
 	case factory.IPAMClaimsType:
 		ipamClaim, ok := obj.(*ipamclaimsapi.IPAMClaim)
 		if !ok {

go-controller/pkg/clustermanager/node/node_allocator.go

@@ -175,6 +175,43 @@ func (na *NodeAllocator) releaseHybridOverlayNodeSubnet(nodeName string) {
 	klog.Infof("Deleted hybrid overlay HostSubnets for node %s", nodeName)
 }
 
+// NeedsNodeAllocation determines if the annotations that are assigned by NodeAllocator are missing on a node
+func (na *NodeAllocator) NeedsNodeAllocation(node *corev1.Node) bool {
+	// hybrid overlay check
+	if util.NoHostSubnet(node) {
+		if na.hasHybridOverlayAllocation() {
+			if _, ok := node.Annotations[hotypes.HybridOverlayNodeSubnet]; !ok {
+				return true
+			}
+		}
+		return false
+	}
+
+	// ovn node check
+	// allocation is all or nothing, so if one field was allocated from:
+	// nodeSubnets, joinSubnet, layer 2 tunnel id, then all of them were
+	if na.hasNodeSubnetAllocation() {
+		if util.HasNodeHostSubnetAnnotation(node, na.netInfo.GetNetworkName()) {
+			return false
+		}
+	}
+
+	if na.hasJoinSubnetAllocation() {
+		if util.HasNodeGatewayRouterJoinNetwork(node, na.netInfo.GetNetworkName()) {
+			return false
+		}
+	}
+
+	if util.IsNetworkSegmentationSupportEnabled() && na.netInfo.IsPrimaryNetwork() && util.DoesNetworkRequireTunnelIDs(na.netInfo) {
+		if util.HasUDNLayer2NodeGRLRPTunnelID(node, na.netInfo.GetNetworkName()) {
+			return false
+		}
+	}
+
+	return true
+
+}
+
 // HandleAddUpdateNodeEvent handles the add or update node event
 func (na *NodeAllocator) HandleAddUpdateNodeEvent(node *corev1.Node) error {
 	defer na.recordSubnetUsage()
@@ -202,14 +239,21 @@ func (na *NodeAllocator) HandleAddUpdateNodeEvent(node *corev1.Node) error {
 
 // syncNodeNetworkAnnotations does 2 things
 //   - syncs the node's allocated subnets in the node subnet annotation
-//   - syncs the network id in the node network id annotation
+//   - syncs the join subnet annotation
+//   - syncs the layer 2 tunnel id annotation
+//   - syncs the network id in the node network id annotation (legacy)
 func (na *NodeAllocator) syncNodeNetworkAnnotations(node *corev1.Node) error {
 	networkName := na.netInfo.GetNetworkName()
 
 	networkID, err := util.ParseNetworkIDAnnotation(node, networkName)
-	if err != nil && !util.IsAnnotationNotSetError(err) {
-		// Log the error and try to allocate new subnets
-		klog.Warningf("Failed to get node %s network id annotations for network %s : %v", node.Name, networkName, err)
+	if err != nil {
+		if !util.IsAnnotationNotSetError(err) {
+			// Log the error and try to allocate new subnets
+			klog.Warningf("Failed to get node %s network id annotations for network %s : %v", node.Name, networkName, err)
+		}
+		// if there is an error we are going to set networkID here to NoNetworkID to prevent
+		// the annotation being updated
+		networkID = types.NoNetworkID
 	}
 
 	updatedSubnetsMap := map[string][]*net.IPNet{}
@@ -280,14 +324,14 @@ func (na *NodeAllocator) syncNodeNetworkAnnotations(node *corev1.Node) error {
 			updatedSubnetsMap[networkName] = validExistingSubnets
 		}
 	}
-	newTunnelID := util.NoID
+	newTunnelID := types.NoTunnelID
 	if util.IsNetworkSegmentationSupportEnabled() && na.netInfo.IsPrimaryNetwork() && util.DoesNetworkRequireTunnelIDs(na.netInfo) {
 		existingTunnelID, err := util.ParseUDNLayer2NodeGRLRPTunnelIDs(node, networkName)
 		if err != nil && !util.IsAnnotationNotSetError(err) {
 			return fmt.Errorf("failed to fetch tunnelID annotation from the node %s for network %s, err: %v",
 				node.Name, networkName, err)
 		}
-		if existingTunnelID == util.InvalidID {
+		if existingTunnelID == types.InvalidID {
 			if newTunnelID, err = na.idAllocator.AllocateID(networkName + "_" + node.Name); err != nil {
 				return fmt.Errorf("failed to assign node %s tunnel id for network %s: %w", node.Name, networkName, err)
 			}
@@ -301,14 +345,24 @@ func (na *NodeAllocator) syncNodeNetworkAnnotations(node *corev1.Node) error {
 		}
 	}
 
+	// only update node annotation with ID if it had it before (networkID is not NoNetworkID)
+	// and does not match.
+	// NoNetworkID means do not update the annotation
+	if networkID != types.NoNetworkID && networkID != na.networkID {
+		networkID = na.networkID
+	} else if networkID == na.networkID {
+		// don't need to update if there was no change to the ID
+		networkID = types.NoNetworkID
+	}
+
 	// Also update the node annotation if the networkID doesn't match
-	if len(updatedSubnetsMap) > 0 || na.networkID != networkID || len(allocatedJoinSubnets) > 0 || newTunnelID != util.NoID {
-		err = na.updateNodeNetworkAnnotationsWithRetry(node.Name, updatedSubnetsMap, na.networkID, newTunnelID, allocatedJoinSubnets)
+	if len(updatedSubnetsMap) > 0 || networkID != types.NoNetworkID || len(allocatedJoinSubnets) > 0 || newTunnelID != types.NoTunnelID {
+		err = na.updateNodeNetworkAnnotationsWithRetry(node.Name, updatedSubnetsMap, networkID, newTunnelID, allocatedJoinSubnets)
 		if err != nil {
 			if errR := na.clusterSubnetAllocator.ReleaseNetworks(node.Name, allocatedSubnets...); errR != nil {
 				klog.Warningf("Error releasing node %s subnets: %v", node.Name, errR)
 			}
-			if util.IsNetworkSegmentationSupportEnabled() && na.netInfo.IsPrimaryNetwork() && util.DoesNetworkRequireTunnelIDs(na.netInfo) {
+			if newTunnelID != types.NoTunnelID {
 				na.idAllocator.ReleaseID(networkName + "_" + node.Name)
 				klog.Infof("Releasing node %s tunnelID for network %s since annotation update failed", node.Name, networkName)
 			}
@@ -406,12 +460,14 @@ func (na *NodeAllocator) updateNodeNetworkAnnotationsWithRetry(nodeName string,
 			return fmt.Errorf("failed to update node %q annotation LRPAddrAnnotation %s: %w",
 				node.Name, util.JoinIPNets(joinAddr, ","), err)
 		}
-		cnode.Annotations, err = util.UpdateNetworkIDAnnotation(cnode.Annotations, networkName, networkId)
-		if err != nil {
-			return fmt.Errorf("failed to update node %q network id annotation %d for network %s: %w",
-				node.Name, networkId, networkName, err)
+		if networkId != types.NoNetworkID {
+			cnode.Annotations, err = util.UpdateNetworkIDAnnotation(cnode.Annotations, networkName, networkId)
+			if err != nil {
+				return fmt.Errorf("failed to update node %q network id annotation %d for network %s: %w",
+					node.Name, networkId, networkName, err)
+			}
 		}
-		if tunnelID != util.NoID {
+		if tunnelID != types.NoTunnelID {
 			cnode.Annotations, err = util.UpdateUDNLayer2NodeGRLRPTunnelIDs(cnode.Annotations, networkName, tunnelID)
 			if err != nil {
 				return fmt.Errorf("failed to update node %q tunnel id annotation %d for network %s: %w",
@@ -448,7 +504,7 @@ func (na *NodeAllocator) Cleanup() error {
 
 		hostSubnetsMap := map[string][]*net.IPNet{networkName: nil}
 		// passing util.InvalidID deletes the network/tunnel id annotation for the network.
-		err = na.updateNodeNetworkAnnotationsWithRetry(node.Name, hostSubnetsMap, util.InvalidID, util.InvalidID, nil)
+		err = na.updateNodeNetworkAnnotationsWithRetry(node.Name, hostSubnetsMap, types.InvalidID, types.InvalidID, nil)
 		if err != nil {
 			return fmt.Errorf("failed to clear node %q subnet annotation for network %s",
 				node.Name, networkName)
@@ -472,8 +528,6 @@ func (na *NodeAllocator) allocateNodeSubnets(allocator SubnetAllocator, nodeName
 		expectedHostSubnets = 2
 	}
 
-	klog.Infof("Expected %d subnets on node %s, found %d: %v", expectedHostSubnets, nodeName, len(existingSubnets), existingSubnets)
-
 	// If any existing subnets the node has are valid, mark them as reserved.
 	// The node might have invalid or already-reserved subnets, or it might
 	// have more subnets than configured in OVN (like for dual-stack to/from
@@ -486,7 +540,6 @@ func (na *NodeAllocator) allocateNodeSubnets(allocator SubnetAllocator, nodeName
 	for _, subnet := range existingSubnets {
 		if (ipv4Mode && utilnet.IsIPv4CIDR(subnet) && !foundIPv4) || (ipv6Mode && utilnet.IsIPv6CIDR(subnet) && !foundIPv6) {
 			if err := allocator.MarkAllocatedNetworks(nodeName, subnet); err == nil {
-				klog.Infof("Valid subnet %v allocated on node %s", subnet, nodeName)
 				existingSubnets[n] = subnet
 				n++
 				if utilnet.IsIPv4CIDR(subnet) {
@@ -498,17 +551,18 @@ func (na *NodeAllocator) allocateNodeSubnets(allocator SubnetAllocator, nodeName
 			}
 		}
 		// this subnet is no longer needed; release it
-		klog.Infof("Releasing unused or invalid subnet %v on node %s", subnet, nodeName)
+		klog.Infof("Releasing unused or invalid subnet %v on node %s, network %s",
+			subnet, na.netInfo.GetNetworkName(), nodeName)
 		if err := allocator.ReleaseNetworks(nodeName, subnet); err != nil {
-			klog.Warningf("Failed to release subnet %v on node %s: %v", subnet, nodeName, err)
+			klog.Warningf("Failed to release subnet %v on node %s, network %s: %v",
+				subnet, nodeName, na.netInfo.GetNetworkName(), err)
 		}
 	}
 	// recreate existingSubnets with the valid subnets
 	existingSubnets = existingSubnets[:n]
 
 	// Node has enough valid subnets already allocated
 	if len(existingSubnets) == expectedHostSubnets {
-		klog.Infof("Allowed existing subnets %v on node %s", existingSubnets, nodeName)
 		return existingSubnets, allocatedSubnets, nil
 	}
 
@@ -517,9 +571,10 @@ func (na *NodeAllocator) allocateNodeSubnets(allocator SubnetAllocator, nodeName
 	defer func() {
 		if releaseAllocatedSubnets {
 			for _, subnet := range allocatedSubnets {
-				klog.Warningf("Releasing subnet %v on node %s", subnet, nodeName)
+				klog.Warningf("Releasing subnet %v on node %s, network: %s", subnet, nodeName, na.netInfo.GetNetworkName())
 				if errR := allocator.ReleaseNetworks(nodeName, subnet); errR != nil {
-					klog.Warningf("Error releasing subnet %v on node %s: %v", subnet, nodeName, errR)
+					klog.Warningf("Error releasing subnet %v on node %s, network %s: %v",
+						subnet, na.netInfo.GetNetworkName(), nodeName, errR)
 				}
 			}
 		}
@@ -528,12 +583,14 @@ func (na *NodeAllocator) allocateNodeSubnets(allocator SubnetAllocator, nodeName
 	// allocateOneSubnet is a helper to process the result of a subnet allocation
 	allocateOneSubnet := func(allocatedHostSubnet *net.IPNet, allocErr error) error {
 		if allocErr != nil {
-			return fmt.Errorf("error allocating network for node %s: %v", nodeName, allocErr)
+			return fmt.Errorf("error allocating network for node %s, network name %s: %v",
+				nodeName, na.netInfo.GetNetworkName(), allocErr)
 		}
 		// the allocator returns nil if it can't provide a subnet
 		// we should filter them out or they will be appended to the slice
 		if allocatedHostSubnet != nil {
-			klog.V(5).Infof("Allocating subnet %v on node %s", allocatedHostSubnet, nodeName)
+			klog.V(5).Infof("Allocating subnet %v on node %s for network: %s",
+				allocatedHostSubnet, nodeName, na.netInfo.GetNetworkName())
 			allocatedSubnets = append(allocatedSubnets, allocatedHostSubnet)
 		}
 		return nil
@@ -556,12 +613,12 @@ func (na *NodeAllocator) allocateNodeSubnets(allocator SubnetAllocator, nodeName
 	// so it will require a reconfiguration and restart.
 	wantedSubnets := expectedHostSubnets - len(existingSubnets)
 	if wantedSubnets > 0 && len(allocatedSubnets) != wantedSubnets {
-		return nil, nil, fmt.Errorf("error allocating networks for node %s: %d subnets expected only new %d subnets allocated",
-			nodeName, expectedHostSubnets, len(allocatedSubnets))
+		return nil, nil, fmt.Errorf("error allocating networks for network: %s, node %s: %d subnets expected only new %d subnets allocated",
+			na.netInfo.GetNetworkName(), nodeName, expectedHostSubnets, len(allocatedSubnets))
 	}
 
 	hostSubnets := append(existingSubnets, allocatedSubnets...)
-	klog.Infof("Allocated Subnets %v on Node %s", hostSubnets, nodeName)
+	klog.Infof("Allocated Subnets %v on Node %s for Network: %s", hostSubnets, nodeName, na.netInfo.GetNetworkName())
 
 	// Success; prevent the release-on-error from triggering and return all node subnets
 	releaseAllocatedSubnets = false