fix: use keymutex to serialize pod add/delete operation

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/kelseyhightower/envconfig v1.4.0 // indirect
 	github.com/kubesphere/porter v0.1.1
 	github.com/moul/http2curl v1.0.0 // indirect
+	github.com/neverlee/keymutex v0.0.0-20171121013845-f593aa834bf9
 	github.com/oilbeater/go-ping v0.0.0-20200413021620-332b7197c5b5
 	github.com/onsi/ginkgo v1.12.0
 	github.com/onsi/gomega v1.9.0

go.sum

@@ -281,6 +281,8 @@ github.com/moul/http2curl v1.0.0/go.mod h1:8UbvGypXm98wA/IqH45anm5Y2Z6ep6O31QGOA
 github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw=
+github.com/neverlee/keymutex v0.0.0-20171121013845-f593aa834bf9 h1:UfW5pM66x0MWE72ySrpd2Ymrn+b62kNHirozKkY3ojE=
+github.com/neverlee/keymutex v0.0.0-20171121013845-f593aa834bf9/go.mod h1:3hf2IoUXDKjCg/EuqSLUB5TY8StGS3haWYJiqzP907c=
 github.com/oilbeater/go-ping v0.0.0-20200413021620-332b7197c5b5 h1:r2QP8IbxbQJeRsxlNEMjS48F5LajUp/g/LDlLhEtxQ0=
 github.com/oilbeater/go-ping v0.0.0-20200413021620-332b7197c5b5/go.mod h1:ZeD7oOS+O/ywjMQWgHqo1tUGfNzB6mNAmv/MYAWxmWg=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=

pkg/controller/controller.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/alauda/kube-ovn/pkg/ovs"
 	"github.com/alauda/kube-ovn/pkg/util"
+	"github.com/neverlee/keymutex"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -36,12 +37,12 @@ type Controller struct {
 	ovnClient *ovs.Client
 	ipam      *ovnipam.IPAM
 
-	podsLister v1.PodLister
-	podsSynced cache.InformerSynced
-
+	podsLister     v1.PodLister
+	podsSynced     cache.InformerSynced
 	addPodQueue    workqueue.RateLimitingInterface
 	deletePodQueue workqueue.RateLimitingInterface
 	updatePodQueue workqueue.RateLimitingInterface
+	podKeyMutex    *keymutex.KeyMutex
 
 	subnetsLister           kubeovnlister.SubnetLister
 	subnetSynced            cache.InformerSynced
@@ -145,6 +146,7 @@ func NewController(config *Configuration) *Controller {
 		addPodQueue:    workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "AddPod"),
 		deletePodQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "DeletePod"),
 		updatePodQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "UpdatePod"),
+		podKeyMutex:    keymutex.New(97),
 
 		namespacesLister:  namespaceInformer.Lister(),
 		namespacesSynced:  namespaceInformer.Informer().HasSynced,

pkg/controller/pod.go

@@ -222,10 +222,13 @@ func (c *Controller) processNextAddPodWorkItem() bool {
 			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
 			return nil
 		}
+		klog.Infof("handle add pod %s", key)
 		if err := c.handleAddPod(key); err != nil {
 			c.addPodQueue.AddRateLimited(key)
 			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
 		}
+		last := time.Since(now)
+		klog.Infof("take %d ms to handle add pod %s", last.Milliseconds(), key)
 		c.addPodQueue.Forget(obj)
 		return nil
 	}(obj)
@@ -234,8 +237,6 @@ func (c *Controller) processNextAddPodWorkItem() bool {
 		utilruntime.HandleError(err)
 		return true
 	}
-	last := time.Since(now)
-	klog.Infof("take %d ms to deal with add pod", last.Milliseconds())
 	return true
 }
 
@@ -256,20 +257,21 @@ func (c *Controller) processNextDeletePodWorkItem() bool {
 			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
 			return nil
 		}
+		klog.Infof("handle delete pod %s", key)
 		if err := c.handleDeletePod(key); err != nil {
 			c.deletePodQueue.AddRateLimited(key)
 			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
 		}
 		c.deletePodQueue.Forget(obj)
+		last := time.Since(now)
+		klog.Infof("take %d ms to handle delete pod %s", last.Milliseconds(), key)
 		return nil
 	}(obj)
 
 	if err != nil {
 		utilruntime.HandleError(err)
 		return true
 	}
-	last := time.Since(now)
-	klog.Infof("take %d ms to deal with delete pod", last.Milliseconds())
 	return true
 }
 
@@ -306,6 +308,8 @@ func (c *Controller) processNextUpdatePodWorkItem() bool {
 }
 
 func (c *Controller) handleAddPod(key string) error {
+	c.podKeyMutex.Lock(key)
+	defer c.podKeyMutex.Unlock(key)
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
@@ -399,6 +403,8 @@ func (c *Controller) handleAddPod(key string) error {
 }
 
 func (c *Controller) handleDeletePod(key string) error {
+	c.podKeyMutex.Lock(key)
+	defer c.podKeyMutex.Unlock(key)
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
@@ -434,6 +440,8 @@ func (c *Controller) handleDeletePod(key string) error {
 }
 
 func (c *Controller) handleUpdatePod(key string) error {
+	c.podKeyMutex.Lock(key)
+	defer c.podKeyMutex.Unlock(key)
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))

vendor/modules.txt

@@ -110,6 +110,8 @@ github.com/modern-go/concurrent
 github.com/modern-go/reflect2
 # github.com/moul/http2curl v1.0.0
 github.com/moul/http2curl
+# github.com/neverlee/keymutex v0.0.0-20171121013845-f593aa834bf9
+github.com/neverlee/keymutex
 # github.com/oilbeater/go-ping v0.0.0-20200413021620-332b7197c5b5
 github.com/oilbeater/go-ping
 # github.com/onsi/ginkgo v1.12.0