Merge pull request ovn-org#3820 from jcaamano/localnet-node-watch

Watch nodes for localnet secondary networks

go-controller/pkg/ovn/base_network_controller_secondary.go

@@ -266,6 +266,24 @@ func (bsnc *BaseSecondaryNetworkController) addLogicalPortToNetworkForNAD(pod *k
 		if err != nil {
 			return err
 		}
+	} else if bsnc.TopologyType() == types.LocalnetTopology {
+		// On localnet networks, we might be processing the pod as a result of a
+		// node changing zone local -> remote so cleanup the logical port in
+		// case it exists and is no longer needed.
+		// This should be an idempotent operation.
+		// Not needed for layer3 networks as in that case the whole node switch
+		// is removed
+		// No need to release IPs as those are allocated from cluster manager
+		logicalPort := bsnc.GetLogicalPortName(pod, nadName)
+		expectedSwitchName, err := bsnc.getExpectedSwitchName(pod)
+		if err != nil {
+			return err
+		}
+		ops, err = bsnc.delLSPOps(logicalPort, expectedSwitchName, "")
+		if err != nil {
+			return err
+		}
+		bsnc.logicalPortCache.remove(pod, nadName)
 	}
 
 	if podAnnotation == nil {
@@ -343,7 +361,12 @@ func (bsnc *BaseSecondaryNetworkController) removePodForSecondaryNetwork(pod *ka
 			return bsnc.removeRemoteZonePodFromNamespaceAddressSet(pod)
 		}
 
-		return nil
+		// except for localnet networks, continue the delete flow in case a node just
+		// became remote where we might still need to cleanup. On L3 networks
+		// the node switch is removed so there is no need to do this.
+		if bsnc.TopologyType() != types.LocalnetTopology {
+			return nil
+		}
 	}
 
 	// for a specific NAD belongs to this network, Pod's logical port might already be created half-way

go-controller/pkg/ovn/base_secondary_layer2_network_controller.go

@@ -16,6 +16,8 @@ import (
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
 
+	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/klog/v2"
 	utilnet "k8s.io/utils/net"
 )
@@ -89,22 +91,49 @@ func (h *baseSecondaryLayer2NetworkControllerEventHandler) IsResourceScheduled(o
 // if any, yielded during object creation.
 // Given an object to add and a boolean specifying if the function was executed from iterateRetryResources
 func (h *baseSecondaryLayer2NetworkControllerEventHandler) AddResource(obj interface{}, fromRetryLoop bool) error {
-	return h.oc.AddSecondaryNetworkResourceCommon(h.objType, obj)
+	switch h.objType {
+	case factory.NodeType:
+		node, ok := obj.(*corev1.Node)
+		if !ok {
+			return fmt.Errorf("could not cast %T object to Node", obj)
+		}
+		return h.oc.addUpdateNodeEvent(node)
+	default:
+		return h.oc.AddSecondaryNetworkResourceCommon(h.objType, obj)
+	}
 }
 
 // UpdateResource updates the specified object in the cluster to its version in newObj according to its
 // type and returns the error, if any, yielded during the object update.
 // Given an old and a new object; The inRetryCache boolean argument is to indicate if the given resource
 // is in the retryCache or not.
 func (h *baseSecondaryLayer2NetworkControllerEventHandler) UpdateResource(oldObj, newObj interface{}, inRetryCache bool) error {
-	return h.oc.UpdateSecondaryNetworkResourceCommon(h.objType, oldObj, newObj, inRetryCache)
+	switch h.objType {
+	case factory.NodeType:
+		node, ok := newObj.(*corev1.Node)
+		if !ok {
+			return fmt.Errorf("could not cast %T object to Node", newObj)
+		}
+		return h.oc.addUpdateNodeEvent(node)
+	default:
+		return h.oc.UpdateSecondaryNetworkResourceCommon(h.objType, oldObj, newObj, inRetryCache)
+	}
 }
 
 // DeleteResource deletes the object from the cluster according to the delete logic of its resource type.
 // Given an object and optionally a cachedObj; cachedObj is the internal cache entry for this object,
 // used for now for pods and network policies.
 func (h *baseSecondaryLayer2NetworkControllerEventHandler) DeleteResource(obj, cachedObj interface{}) error {
-	return h.oc.DeleteSecondaryNetworkResourceCommon(h.objType, obj, cachedObj)
+	switch h.objType {
+	case factory.NodeType:
+		node, ok := obj.(*corev1.Node)
+		if !ok {
+			return fmt.Errorf("could not cast %T object to Node", obj)
+		}
+		return h.oc.deleteNodeEvent(node)
+	default:
+		return h.oc.DeleteSecondaryNetworkResourceCommon(h.objType, obj, cachedObj)
+	}
 }
 
 func (h *baseSecondaryLayer2NetworkControllerEventHandler) SyncFunc(objs []interface{}) error {
@@ -115,6 +144,9 @@ func (h *baseSecondaryLayer2NetworkControllerEventHandler) SyncFunc(objs []inter
 		syncFunc = h.syncFunc
 	} else {
 		switch h.objType {
+		case factory.NodeType:
+			syncFunc = h.oc.syncNodes
+
 		case factory.PodType:
 			syncFunc = h.oc.syncPodsForSecondaryNetwork
 
@@ -147,6 +179,7 @@ type BaseSecondaryLayer2NetworkController struct {
 }
 
 func (oc *BaseSecondaryLayer2NetworkController) initRetryFramework() {
+	oc.retryNodes = oc.newRetryFramework(factory.NodeType)
 	oc.retryPods = oc.newRetryFramework(factory.PodType)
 
 	// For secondary networks, we don't have to watch namespace events if
@@ -195,6 +228,9 @@ func (oc *BaseSecondaryLayer2NetworkController) stop() {
 	if oc.podHandler != nil {
 		oc.watchFactory.RemovePodHandler(oc.podHandler)
 	}
+	if oc.nodeHandler != nil {
+		oc.watchFactory.RemoveNodeHandler(oc.nodeHandler)
+	}
 	if oc.namespaceHandler != nil {
 		oc.watchFactory.RemoveNamespaceHandler(oc.namespaceHandler)
 	}
@@ -232,6 +268,10 @@ func (oc *BaseSecondaryLayer2NetworkController) run() error {
 		return err
 	}
 
+	if err := oc.WatchNodes(); err != nil {
+		return err
+	}
+
 	if err := oc.WatchPods(); err != nil {
 		return err
 	}
@@ -288,3 +328,66 @@ func (oc *BaseSecondaryLayer2NetworkController) initializeLogicalSwitch(switchNa
 
 	return &logicalSwitch, nil
 }
+
+func (oc *BaseSecondaryLayer2NetworkController) addUpdateNodeEvent(node *corev1.Node) error {
+	if oc.isLocalZoneNode(node) {
+		return oc.addUpdateLocalNodeEvent(node)
+	}
+	return oc.addUpdateRemoteNodeEvent(node)
+}
+
+func (oc *BaseSecondaryLayer2NetworkController) addUpdateLocalNodeEvent(node *corev1.Node) error {
+	_, present := oc.localZoneNodes.LoadOrStore(node.Name, true)
+
+	if !present {
+		// process all pods so they are reconfigured as local
+		errs := oc.addAllPodsOnNode(node.Name)
+		if errs != nil {
+			err := kerrors.NewAggregate(errs)
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (oc *BaseSecondaryLayer2NetworkController) addUpdateRemoteNodeEvent(node *corev1.Node) error {
+	_, present := oc.localZoneNodes.Load(node.Name)
+
+	if present {
+		err := oc.deleteNodeEvent(node)
+		if err != nil {
+			return err
+		}
+
+		// process all pods so they are reconfigured as remote
+		errs := oc.addAllPodsOnNode(node.Name)
+		if errs != nil {
+			err = kerrors.NewAggregate(errs)
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (oc *BaseSecondaryLayer2NetworkController) deleteNodeEvent(node *corev1.Node) error {
+	oc.localZoneNodes.Delete(node.Name)
+	return nil
+}
+
+func (oc *BaseSecondaryLayer2NetworkController) syncNodes(nodes []interface{}) error {
+	for _, tmp := range nodes {
+		node, ok := tmp.(*corev1.Node)
+		if !ok {
+			return fmt.Errorf("spurious object in syncNodes: %v", tmp)
+		}
+
+		// Add the node to the foundNodes only if it belongs to the local zone.
+		if oc.isLocalZoneNode(node) {
+			oc.localZoneNodes.Store(node.Name, true)
+		}
+	}
+
+	return nil
+}