u2o keep src mac (#5192)

Signed-off-by: clyi <clyi@alauda.io>

Author: changluyi

pkg/daemon/controller.go

@@ -51,9 +51,10 @@ type Controller struct {
 	ovnEipsLister kubeovnlister.OvnEipLister
 	ovnEipsSynced cache.InformerSynced
 
-	podsLister listerv1.PodLister
-	podsSynced cache.InformerSynced
-	podQueue   workqueue.TypedRateLimitingInterface[string]
+	podsLister     listerv1.PodLister
+	podsSynced     cache.InformerSynced
+	updatePodQueue workqueue.TypedRateLimitingInterface[string]
+	deletePodQueue workqueue.TypedRateLimitingInterface[string]
 
 	nodesLister listerv1.NodeLister
 	nodesSynced cache.InformerSynced
@@ -112,9 +113,10 @@ func NewController(config *Configuration, stopCh <-chan struct{}, podInformerFac
 		ovnEipsLister: ovnEipInformer.Lister(),
 		ovnEipsSynced: ovnEipInformer.Informer().HasSynced,
 
-		podsLister: podInformer.Lister(),
-		podsSynced: podInformer.Informer().HasSynced,
-		podQueue:   newTypedRateLimitingQueue[string]("Pod", nil),
+		podsLister:     podInformer.Lister(),
+		podsSynced:     podInformer.Informer().HasSynced,
+		updatePodQueue: newTypedRateLimitingQueue[string]("UpdatePod", nil),
+		deletePodQueue: newTypedRateLimitingQueue[string]("DeletePod", nil),
 
 		nodesLister: nodeInformer.Lister(),
 		nodesSynced: nodeInformer.Informer().HasSynced,
@@ -175,7 +177,8 @@ func NewController(config *Configuration, stopCh <-chan struct{}, podInformerFac
 	}
 
 	if _, err = podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		UpdateFunc: controller.enqueuePod,
+		UpdateFunc: controller.enqueueUpdatePod,
+		DeleteFunc: controller.enqueueDeletePod,
 	}); err != nil {
 		return nil, err
 	}
@@ -518,7 +521,7 @@ func (c *Controller) processNextServiceWorkItem() bool {
 	return true
 }
 
-func (c *Controller) enqueuePod(oldObj, newObj any) {
+func (c *Controller) enqueueUpdatePod(oldObj, newObj any) {
 	oldPod := oldObj.(*v1.Pod)
 	newPod := newObj.(*v1.Pod)
 	key := cache.MetaObjectToName(newPod).String()
@@ -529,8 +532,9 @@ func (c *Controller) enqueuePod(oldObj, newObj any) {
 		oldPod.Annotations[util.NetemQosJitterAnnotation] != newPod.Annotations[util.NetemQosJitterAnnotation] ||
 		oldPod.Annotations[util.NetemQosLimitAnnotation] != newPod.Annotations[util.NetemQosLimitAnnotation] ||
 		oldPod.Annotations[util.NetemQosLossAnnotation] != newPod.Annotations[util.NetemQosLossAnnotation] ||
-		oldPod.Annotations[util.MirrorControlAnnotation] != newPod.Annotations[util.MirrorControlAnnotation] {
-		c.podQueue.Add(key)
+		oldPod.Annotations[util.MirrorControlAnnotation] != newPod.Annotations[util.MirrorControlAnnotation] ||
+		oldPod.Annotations[util.IPAddressAnnotation] != newPod.Annotations[util.IPAddressAnnotation] {
+		c.updatePodQueue.Add(key)
 		return
 	}
 
@@ -548,30 +552,63 @@ func (c *Controller) enqueuePod(oldObj, newObj any) {
 				oldPod.Annotations[fmt.Sprintf(util.NetemQosLimitAnnotationTemplate, provider)] != newPod.Annotations[fmt.Sprintf(util.NetemQosLimitAnnotationTemplate, provider)] ||
 				oldPod.Annotations[fmt.Sprintf(util.NetemQosLossAnnotationTemplate, provider)] != newPod.Annotations[fmt.Sprintf(util.NetemQosLossAnnotationTemplate, provider)] ||
 				oldPod.Annotations[fmt.Sprintf(util.MirrorControlAnnotationTemplate, provider)] != newPod.Annotations[fmt.Sprintf(util.MirrorControlAnnotationTemplate, provider)] {
-				c.podQueue.Add(key)
+				c.updatePodQueue.Add(key)
 			}
 		}
 	}
 }
 
-func (c *Controller) runPodWorker() {
-	for c.processNextPodWorkItem() {
+func (c *Controller) enqueueDeletePod(obj any) {
+	pod := obj.(*v1.Pod)
+	key := cache.MetaObjectToName(pod).String()
+	c.deletePodQueue.Add(key)
+}
+
+func (c *Controller) runUpdatePodWorker() {
+	for c.processNextUpdatePodWorkItem() {
+	}
+}
+
+func (c *Controller) runDeletePodWorker() {
+	for c.processNextDeletePodWorkItem() {
 	}
 }
 
-func (c *Controller) processNextPodWorkItem() bool {
-	key, shutdown := c.podQueue.Get()
+func (c *Controller) processNextUpdatePodWorkItem() bool {
+	key, shutdown := c.updatePodQueue.Get()
+	if shutdown {
+		return false
+	}
+
+	err := func(key string) error {
+		defer c.updatePodQueue.Done(key)
+		if err := c.handleUpdatePod(key); err != nil {
+			c.updatePodQueue.AddRateLimited(key)
+			return fmt.Errorf("error syncing %q: %w, requeuing", key, err)
+		}
+		c.updatePodQueue.Forget(key)
+		return nil
+	}(key)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return true
+	}
+	return true
+}
+
+func (c *Controller) processNextDeletePodWorkItem() bool {
+	key, shutdown := c.deletePodQueue.Get()
 	if shutdown {
 		return false
 	}
 
 	err := func(key string) error {
-		defer c.podQueue.Done(key)
-		if err := c.handlePod(key); err != nil {
-			c.podQueue.AddRateLimited(key)
+		defer c.deletePodQueue.Done(key)
+		if err := c.handleDeletePod(key); err != nil {
+			c.deletePodQueue.AddRateLimited(key)
 			return fmt.Errorf("error syncing %q: %w, requeuing", key, err)
 		}
-		c.podQueue.Forget(key)
+		c.deletePodQueue.Forget(key)
 		return nil
 	}(key)
 	if err != nil {
@@ -615,7 +652,8 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 	defer c.deleteProviderNetworkQueue.ShutDown()
 	defer c.subnetQueue.ShutDown()
 	defer c.serviceQueue.ShutDown()
-	defer c.podQueue.ShutDown()
+	defer c.updatePodQueue.ShutDown()
+	defer c.deletePodQueue.ShutDown()
 
 	go wait.Until(ovs.CleanLostInterface, time.Minute, stopCh)
 	go wait.Until(recompute, 10*time.Minute, stopCh)
@@ -633,7 +671,8 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 	go wait.Until(c.runAddOrUpdateServicekWorker, time.Second, stopCh)
 	go wait.Until(c.runDeleteProviderNetworkWorker, time.Second, stopCh)
 	go wait.Until(c.runSubnetWorker, time.Second, stopCh)
-	go wait.Until(c.runPodWorker, time.Second, stopCh)
+	go wait.Until(c.runUpdatePodWorker, time.Second, stopCh)
+	go wait.Until(c.runDeletePodWorker, time.Second, stopCh)
 	go wait.Until(c.runGateway, 3*time.Second, stopCh)
 	go wait.Until(c.loopEncapIPCheck, 3*time.Second, stopCh)
 	go wait.Until(c.ovnMetricsUpdate, 3*time.Second, stopCh)

pkg/daemon/controller_linux.go

@@ -13,6 +13,7 @@ import (
 	"syscall"
 
 	ovsutil "github.com/digitalocean/go-openvswitch/ovs"
+	nadv1 "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/apis/k8s.cni.cncf.io/v1"
 	nadutils "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/utils"
 	"github.com/kubeovn/felix/ipsets"
 	"github.com/kubeovn/go-iptables/iptables"
@@ -229,6 +230,13 @@ func (c *Controller) reconcileRouters(event *subnetEvent) error {
 			}
 		}
 
+		isAdd, needAction := c.CheckSubnetU2OChangeAction(oldSubnet, newSubnet)
+		if needAction {
+			if err := c.HandleU2OForSubnet(newSubnet, isAdd); err != nil {
+				return err
+			}
+		}
+
 		if err = c.handleEnableExternalLBAddressChange(oldSubnet, newSubnet); err != nil {
 			klog.Errorf("failed to handle enable external lb address change: %v", err)
 			return err
@@ -762,7 +770,100 @@ func (c *Controller) getPolicyRouting(subnet *kubeovnv1.Subnet) ([]netlink.Rule,
 	return rules, routes, nil
 }
 
-func (c *Controller) handlePod(key string) error {
+func (c *Controller) GetProviderInfoFromSubnet(subnet *kubeovnv1.Subnet) (bridgeName, chassisMac string, err error) {
+	if subnet == nil {
+		return "", "", nil
+	}
+	if subnet.Spec.Vlan == "" {
+		return "", "", nil
+	}
+
+	vlan, err := c.vlansLister.Get(subnet.Spec.Vlan)
+	if err != nil {
+		return "", "", fmt.Errorf("failed to get vlan %s: %w", subnet.Spec.Vlan, err)
+	}
+	providerName := vlan.Spec.Provider
+	chassisMac, err = GetProviderChassisMac(providerName)
+	if err != nil {
+		return "", "", fmt.Errorf("failed to get chassis mac for provider %s: %w", providerName, err)
+	}
+
+	bridgeName = util.ExternalBridgeName(providerName)
+	return bridgeName, chassisMac, nil
+}
+
+func HandleU2OForPod(ovsClient *ovsutil.Client, pod *v1.Pod, bridgeName, chassisMac string, isAdd bool) error {
+	if pod == nil {
+		return errors.New("pod is nil")
+	}
+
+	podMac := pod.Annotations[util.MacAddressAnnotation]
+
+	podIPs := []string{}
+	if pod.Annotations != nil && pod.Annotations[util.IPAddressAnnotation] != "" {
+		podIPs = append(podIPs, strings.Split(pod.Annotations[util.IPAddressAnnotation], ",")...)
+
+		for _, podIP := range podIPs {
+			var err error
+			if isAdd {
+				err = ovs.AddOrUpdateU2OKeepSrcMac(ovsClient, bridgeName, podIP, podMac, chassisMac)
+			} else {
+				err = ovs.DeleteU2OKeepSrcMac(ovsClient, bridgeName, podIP, chassisMac)
+			}
+
+			if err != nil {
+				action := "add"
+				if !isAdd {
+					action = "delete"
+				}
+				return fmt.Errorf("failed to %s U2O rule for pod %s/%s: %w", action, pod.Namespace, pod.Name, err)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (c *Controller) HandleU2OForSubnet(subnet *kubeovnv1.Subnet, isAdd bool) error {
+	klog.Infof("U2O processing for subnet %s, action: %v", subnet.Name, isAdd)
+
+	bridgeName, chassisMac, err := c.GetProviderInfoFromSubnet(subnet)
+	if err != nil {
+		return fmt.Errorf("failed to get provider info: %w", err)
+	}
+
+	pods, err := c.podsLister.List(labels.Everything())
+	if err != nil {
+		return fmt.Errorf("failed to list pods: %w", err)
+	}
+
+	for _, pod := range pods {
+		if pod.Annotations[util.LogicalSwitchAnnotation] != subnet.Name {
+			continue
+		}
+		if err := HandleU2OForPod(c.ovsClient, pod, bridgeName, chassisMac, isAdd); err != nil {
+			klog.Error(err)
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Controller) CheckSubnetU2OChangeAction(oldSubnet, newSubnet *kubeovnv1.Subnet) (bool, bool) {
+	if newSubnet == nil ||
+		(oldSubnet != nil && newSubnet != nil && oldSubnet.Spec.U2OInterconnection == newSubnet.Spec.U2OInterconnection) {
+		return false, false
+	}
+
+	if newSubnet.Spec.Vlan == "" || newSubnet.Spec.LogicalGateway {
+		return false, false
+	}
+
+	return newSubnet.Spec.U2OInterconnection, true
+}
+
+func (c *Controller) handleUpdatePod(key string) error {
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
@@ -785,6 +886,26 @@ func (c *Controller) handlePod(key string) error {
 		return err
 	}
 
+	if _, ok := pod.Annotations[util.LogicalSwitchAnnotation]; ok {
+		subnet, err := c.subnetsLister.Get(pod.Annotations[util.LogicalSwitchAnnotation])
+		if err != nil {
+			klog.Error(err)
+			return err
+		}
+
+		if subnet.Spec.U2OInterconnection {
+			bridgeName, chassisMac, err := c.GetProviderInfoFromSubnet(subnet)
+			if err != nil {
+				klog.Error(err)
+				return err
+			}
+			if err := HandleU2OForPod(c.ovsClient, pod, bridgeName, chassisMac, true); err != nil {
+				klog.Error(err)
+				return err
+			}
+		}
+	}
+
 	podName := pod.Name
 	if pod.Annotations[fmt.Sprintf(util.VMAnnotationTemplate, util.OvnProvider)] != "" {
 		podName = pod.Annotations[fmt.Sprintf(util.VMAnnotationTemplate, util.OvnProvider)]
@@ -815,6 +936,9 @@ func (c *Controller) handlePod(key string) error {
 	// set multus-nic bandwidth
 	attachNets, err := nadutils.ParsePodNetworkAnnotation(pod)
 	if err != nil {
+		if _, ok := err.(*nadv1.NoK8sNetworkError); ok {
+			return nil
+		}
 		klog.Error(err)
 		return err
 	}
@@ -843,6 +967,51 @@ func (c *Controller) handlePod(key string) error {
 			}
 		}
 	}
+
+	return nil
+}
+
+func (c *Controller) handleDeletePod(key string) error {
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return nil
+	}
+
+	pod, err := c.podsLister.Pods(namespace).Get(name)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+		klog.Error(err)
+		return err
+	}
+
+	if _, ok := pod.Annotations[util.LogicalSwitchAnnotation]; !ok {
+		return nil
+	}
+
+	subnet, err := c.subnetsLister.Get(pod.Annotations[util.LogicalSwitchAnnotation])
+	if err != nil {
+		klog.Error(err)
+		return err
+	}
+
+	if !subnet.Spec.U2OInterconnection {
+		return nil
+	}
+
+	bridgeName, chassisMac, err := c.GetProviderInfoFromSubnet(subnet)
+	if err != nil {
+		klog.Error(err)
+		return err
+	}
+
+	if err := HandleU2OForPod(c.ovsClient, pod, bridgeName, chassisMac, false); err != nil {
+		klog.Error(err)
+		return err
+	}
+
 	return nil
 }
 

pkg/daemon/controller_windows.go

@@ -141,7 +141,11 @@ func routeDiff(existingRoutes, v4Cidrs, v6Cidrs []string) (toAddV4, toAddV6, toD
 	return
 }
 
-func (c *Controller) handlePod(key string) error {
+func (c *Controller) handleDeletePod(key string) error {
+	return nil
+}
+
+func (c *Controller) handleUpdatePod(key string) error {
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))

pkg/daemon/ovs.go

@@ -286,3 +286,17 @@ func initProviderChassisMac(provider string) error {
 	}
 	return nil
 }
+
+func GetProviderChassisMac(provider string) (string, error) {
+	mappings, err := getOvnMappings("ovn-chassis-mac-mappings")
+	if err != nil {
+		return "", fmt.Errorf("failed to get chassis mac for provider %s: %w", provider, err)
+	}
+
+	mac, ok := mappings[provider]
+	if !ok {
+		return "", fmt.Errorf("no chassis mac found for provider %s", provider)
+	}
+
+	return mac, nil
+}