add sync map to fix cocurrent write (#2918)

* add sync map to fix cocurrent write * change type
kubeovn · Jun 9, 2023 · 517d379 · 517d379
1 parent dff950b
commit 517d379
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 22 deletions.
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
@@ -63,7 +63,7 @@ type Controller struct {
 	podsSynced             cache.InformerSynced
 	addOrUpdatePodQueue    workqueue.RateLimitingInterface
 	deletePodQueue         workqueue.RateLimitingInterface
-	deletingPodObjMap      map[string]*corev1.Pod
+	deletingPodObjMap      *sync.Map
 	updatePodSecurityQueue workqueue.RateLimitingInterface
 	podKeyMutex            keymutex.KeyMutex
 
@@ -298,12 +298,13 @@ func Run(ctx context.Context, config *Configuration) {
 		numKeyLocks = config.WorkerNum * 2
 	}
 	controller := &Controller{
-		config:          config,
-		vpcs:            &sync.Map{},
-		podSubnetMap:    &sync.Map{},
-		ovnLegacyClient: ovs.NewLegacyClient(config.OvnNbAddr, config.OvnTimeout, config.OvnSbAddr, config.ClusterRouter, config.ClusterTcpLoadBalancer, config.ClusterUdpLoadBalancer, config.ClusterTcpSessionLoadBalancer, config.ClusterUdpSessionLoadBalancer, config.NodeSwitch, config.NodeSwitchCIDR),
-		ipam:            ovnipam.NewIPAM(),
-		namedPort:       NewNamedPort(),
+		config:            config,
+		vpcs:              &sync.Map{},
+		podSubnetMap:      &sync.Map{},
+		deletingPodObjMap: &sync.Map{},
+		ovnLegacyClient:   ovs.NewLegacyClient(config.OvnNbAddr, config.OvnTimeout, config.OvnSbAddr, config.ClusterRouter, config.ClusterTcpLoadBalancer, config.ClusterUdpLoadBalancer, config.ClusterTcpSessionLoadBalancer, config.ClusterUdpSessionLoadBalancer, config.NodeSwitch, config.NodeSwitchCIDR),
+		ipam:              ovnipam.NewIPAM(),
+		namedPort:         NewNamedPort(),
 
 		vpcsLister:           vpcInformer.Lister(),
 		vpcSynced:            vpcInformer.Informer().HasSynced,
@@ -395,7 +396,6 @@ func Run(ctx context.Context, config *Configuration) {
 			workqueue.NewNamedDelayingQueue("DeletePod"),
 			workqueue.DefaultControllerRateLimiter(),
 		),
-		deletingPodObjMap:      make(map[string]*corev1.Pod),
 		updatePodSecurityQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "UpdatePodSecurity"),
 		podKeyMutex:            keymutex.NewHashed(numKeyLocks),
 

diff --git a/pkg/controller/pod.go b/pkg/controller/pod.go
@@ -199,17 +199,17 @@ func (c *Controller) enqueueAddPod(obj interface{}) {
 		if isStateful || (isVmPod && c.config.EnableKeepVmIP) {
 			if isStateful && isStatefulSetPodToDel(c.config.KubeClient, p, statefulSetName) {
 				klog.V(3).Infof("enqueue delete pod %s", key)
-				c.deletingPodObjMap[key] = p
+				c.deletingPodObjMap.Store(key, p)
 				c.deletePodQueue.Add(key)
 			}
 			if isVmPod && c.isVmPodToDel(p, vmName) {
 				klog.V(3).Infof("enqueue delete pod %s", key)
-				c.deletingPodObjMap[key] = p
+				c.deletingPodObjMap.Store(key, p)
 				c.deletePodQueue.Add(key)
 			}
 		} else {
 			klog.V(3).Infof("enqueue delete pod %s", key)
-			c.deletingPodObjMap[key] = p
+			c.deletingPodObjMap.Store(key, p)
 			c.deletePodQueue.Add(key)
 		}
 		return
@@ -240,7 +240,7 @@ func (c *Controller) enqueueDeletePod(obj interface{}) {
 	}
 
 	klog.V(3).Infof("enqueue delete pod %s", key)
-	c.deletingPodObjMap[key] = p
+	c.deletingPodObjMap.Store(key, p)
 	c.deletePodQueue.Add(key)
 }
 
@@ -294,7 +294,7 @@ func (c *Controller) enqueueUpdatePod(oldObj, newObj interface{}) {
 	isVmPod, vmName := isVmPod(newPod)
 	if !isPodStatusPhaseAlive(newPod) && !isStateful && !isVmPod {
 		klog.V(3).Infof("enqueue delete pod %s", key)
-		c.deletingPodObjMap[key] = newPod
+		c.deletingPodObjMap.Store(key, newPod)
 		c.deletePodQueue.Add(key)
 		return
 	}
@@ -314,7 +314,7 @@ func (c *Controller) enqueueUpdatePod(oldObj, newObj interface{}) {
 			// In case node get lost and pod can not be deleted,
 			// the ip address will not be recycled
 			klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)
-			c.deletingPodObjMap[key] = newPod
+			c.deletingPodObjMap.Store(key, newPod)
 			c.deletePodQueue.AddAfter(key, delay)
 		}()
 		return
@@ -324,15 +324,15 @@ func (c *Controller) enqueueUpdatePod(oldObj, newObj interface{}) {
 	if isStateful && isStatefulSetPodToDel(c.config.KubeClient, newPod, statefulSetName) {
 		go func() {
 			klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)
-			c.deletingPodObjMap[key] = newPod
+			c.deletingPodObjMap.Store(key, newPod)
 			c.deletePodQueue.AddAfter(key, delay)
 		}()
 		return
 	}
 	if isVmPod && c.isVmPodToDel(newPod, vmName) {
 		go func() {
 			klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)
-			c.deletingPodObjMap[key] = newPod
+			c.deletingPodObjMap.Store(key, newPod)
 			c.deletePodQueue.AddAfter(key, delay)
 		}()
 		return
@@ -420,8 +420,7 @@ func (c *Controller) processNextDeletePodWorkItem() bool {
 			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
 			return nil
 		}
-		_, exist := c.deletingPodObjMap[key]
-		if !exist {
+		if _, ok := c.deletingPodObjMap.Load(key); !ok {
 			return nil
 		}
 
@@ -435,7 +434,7 @@ func (c *Controller) processNextDeletePodWorkItem() bool {
 		// gc pod obj in c.deletingPodObjMap
 		go func() {
 			time.Sleep(5 * time.Minute)
-			delete(c.deletingPodObjMap, key)
+			c.deletingPodObjMap.Delete(key)
 		}()
 		return nil
 	}(obj)
@@ -718,7 +717,7 @@ func (c *Controller) reconcileAllocateSubnets(cachedPod, pod *v1.Pod, needAlloca
 			// Sometimes pod is deleted between kube-ovn configure ovn-nb and patch pod.
 			// Then we need to recycle the resource again.
 			key := strings.Join([]string{namespace, name}, "/")
-			c.deletingPodObjMap[key] = pod
+			c.deletingPodObjMap.Store(key, pod)
 			c.deletePodQueue.AddRateLimited(key)
 			return nil, nil
 		}
@@ -865,7 +864,7 @@ func (c *Controller) reconcileRouteSubnets(cachedPod, pod *v1.Pod, needRoutePodN
 			// Sometimes pod is deleted between kube-ovn configure ovn-nb and patch pod.
 			// Then we need to recycle the resource again.
 			key := strings.Join([]string{namespace, name}, "/")
-			c.deletingPodObjMap[key] = pod
+			c.deletingPodObjMap.Store(key, pod)
 			c.deletePodQueue.AddRateLimited(key)
 			return nil
 		}
@@ -876,7 +875,11 @@ func (c *Controller) reconcileRouteSubnets(cachedPod, pod *v1.Pod, needRoutePodN
 }
 
 func (c *Controller) handleDeletePod(key string) error {
-	pod := c.deletingPodObjMap[key]
+	podObj, ok := c.deletingPodObjMap.Load(key)
+	if !ok {
+		return nil
+	}
+	pod := podObj.(*v1.Pod)
 	podName := c.getNameByPod(pod)
 	c.podKeyMutex.LockKey(key)
 	defer func() { _ = c.podKeyMutex.UnlockKey(key) }()