fix gabage u2o openflow (#5757)

* fix delete pod not remove u2o flow

Signed-off-by: clyi <[email protected]>

Author: changluyi

pkg/daemon/controller.go

@@ -54,7 +54,7 @@ type Controller struct {
 	podsLister     listerv1.PodLister
 	podsSynced     cache.InformerSynced
 	updatePodQueue workqueue.TypedRateLimitingInterface[string]
-	deletePodQueue workqueue.TypedRateLimitingInterface[string]
+	deletePodQueue workqueue.TypedRateLimitingInterface[*podEvent]
 
 	nodesLister listerv1.NodeLister
 	nodesSynced cache.InformerSynced
@@ -116,7 +116,7 @@ func NewController(config *Configuration, stopCh <-chan struct{}, podInformerFac
 		podsLister:     podInformer.Lister(),
 		podsSynced:     podInformer.Informer().HasSynced,
 		updatePodQueue: newTypedRateLimitingQueue[string]("UpdatePod", nil),
-		deletePodQueue: newTypedRateLimitingQueue[string]("DeletePod", nil),
+		deletePodQueue: newTypedRateLimitingQueue[*podEvent]("DeletePod", nil),
 
 		nodesLister: nodeInformer.Lister(),
 		nodesSynced: nodeInformer.Informer().HasSynced,
@@ -458,6 +458,10 @@ type serviceEvent struct {
 	oldObj, newObj any
 }
 
+type podEvent struct {
+	oldObj any
+}
+
 func (c *Controller) enqueueAddSubnet(obj any) {
 	c.subnetQueue.Add(&subnetEvent{newObj: obj})
 }
@@ -590,8 +594,8 @@ func (c *Controller) enqueueDeletePod(obj any) {
 		return
 	}
 
-	key := cache.MetaObjectToName(pod).String()
-	c.deletePodQueue.Add(key)
+	klog.V(3).Infof("enqueue delete pod %s", pod.Name)
+	c.deletePodQueue.Add(&podEvent{oldObj: pod})
 }
 
 func (c *Controller) runUpdatePodWorker() {
@@ -627,20 +631,20 @@ func (c *Controller) processNextUpdatePodWorkItem() bool {
 }
 
 func (c *Controller) processNextDeletePodWorkItem() bool {
-	key, shutdown := c.deletePodQueue.Get()
+	event, shutdown := c.deletePodQueue.Get()
 	if shutdown {
 		return false
 	}
 
-	err := func(key string) error {
-		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)
+	err := func(event *podEvent) error {
+		defer c.deletePodQueue.Done(event)
+		if err := c.handleDeletePod(event); err != nil {
+			c.deletePodQueue.AddRateLimited(event)
+			return fmt.Errorf("error syncing pod event: %w, requeuing", err)
 		}
-		c.deletePodQueue.Forget(key)
+		c.deletePodQueue.Forget(event)
 		return nil
-	}(key)
+	}(event)
 	if err != nil {
 		utilruntime.HandleError(err)
 		return true

pkg/daemon/controller_linux.go

@@ -971,27 +971,20 @@ func (c *Controller) handleUpdatePod(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))
+func (c *Controller) handleDeletePod(event *podEvent) error {
+	var pod *v1.Pod
+	if event.oldObj != nil {
+		pod = event.oldObj.(*v1.Pod)
+	} else {
 		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 {
+	logicalSwitch, ok := pod.Annotations[util.LogicalSwitchAnnotation]
+	if !ok {
 		return nil
 	}
 
-	subnet, err := c.subnetsLister.Get(pod.Annotations[util.LogicalSwitchAnnotation])
+	subnet, err := c.subnetsLister.Get(logicalSwitch)
 	if err != nil {
 		klog.Error(err)
 		return err

pkg/daemon/controller_windows.go

@@ -141,7 +141,7 @@ func routeDiff(existingRoutes, v4Cidrs, v6Cidrs []string) (toAddV4, toAddV6, toD
 	return
 }
 
-func (c *Controller) handleDeletePod(key string) error {
+func (c *Controller) handleDeletePod(event *podEvent) error {
 	return nil
 }
 

pkg/ovs/ovs-ofctl.go

@@ -214,11 +214,18 @@ func AddOrUpdateU2OKeepSrcMac(client *ovs.Client, bridgeName, podIP, podMac, cha
 		return err
 	}
 
-	for _, existingFlow := range flows {
-		if existingFlow.Priority == util.U2OKeepSrcMacPriority && existingFlow.InPort == localnetPatchPortID {
-			klog.V(3).Infof("flow already exists in bridge %s for in_port=%d, ip_src=%s, dl_src=%s",
-				bridgeName, localnetPatchPortID, podIP, chassisMac)
-			return nil
+	expectedAction := fmt.Sprintf("mod_dl_src:%s", podMac)
+	for _, flow := range flows {
+		if flow.Priority != util.U2OKeepSrcMacPriority || flow.InPort != localnetPatchPortID {
+			continue
+		}
+
+		for _, action := range flow.Actions {
+			if actionText, _ := action.MarshalText(); string(actionText) == expectedAction {
+				klog.V(3).Infof("flow already exists in bridge %s for in_port=%d, ip_src=%s, dl_src=%s, actions=mod_dl_src:%s",
+					bridgeName, localnetPatchPortID, podIP, chassisMac, podMac)
+				return nil
+			}
 		}
 	}