avoid concurrent subnet status update (#1976)

pkg/controller/controller.go

@@ -75,6 +75,7 @@ type Controller struct {
 	deleteSubnetQueue       workqueue.RateLimitingInterface
 	deleteRouteQueue        workqueue.RateLimitingInterface
 	updateSubnetStatusQueue workqueue.RateLimitingInterface
+	subnetStatusKeyMutex    *keymutex.KeyMutex
 
 	ipsLister kubeovnlister.IPLister
 	ipSynced  cache.InformerSynced
@@ -199,6 +200,7 @@ func NewController(config *Configuration) *Controller {
 		deleteSubnetQueue:       workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "DeleteSubnet"),
 		deleteRouteQueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "DeleteRoute"),
 		updateSubnetStatusQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "UpdateSubnetStatus"),
+		subnetStatusKeyMutex:    keymutex.New(97),
 
 		ipsLister: ipInformer.Lister(),
 		ipSynced:  ipInformer.Informer().HasSynced,

pkg/controller/subnet.go

@@ -632,6 +632,9 @@ func (c *Controller) handleAddOrUpdateSubnet(key string) error {
 }
 
 func (c *Controller) handleUpdateSubnetStatus(key string) error {
+	c.subnetStatusKeyMutex.Lock(key)
+	defer c.subnetStatusKeyMutex.Unlock(key)
+
 	orisubnet, err := c.subnetsLister.Get(key)
 	subnet := orisubnet.DeepCopy()
 	if err != nil {
@@ -1193,6 +1196,13 @@ func calcDualSubnetStatusIP(subnet *kubeovnv1.Subnet, c *Controller) error {
 		v6availableIPs = 0
 	}
 
+	if subnet.Status.V4AvailableIPs == v4availableIPs &&
+		subnet.Status.V6AvailableIPs == v6availableIPs &&
+		subnet.Status.V4UsingIPs == float64(len(v4UsingIPs)) &&
+		subnet.Status.V6UsingIPs == float64(len(v6UsingIPs)) {
+		return nil
+	}
+
 	subnet.Status.V4AvailableIPs = v4availableIPs
 	subnet.Status.V6AvailableIPs = v6availableIPs
 	subnet.Status.V4UsingIPs = float64(len(v4UsingIPs))
@@ -1227,6 +1237,13 @@ func calcSubnetStatusIP(subnet *kubeovnv1.Subnet, c *Controller) error {
 		availableIPs = 0
 	}
 	usingIPs := float64(len(podUsedIPs.Items))
+	cachedFields := [4]float64{
+		subnet.Status.V4AvailableIPs,
+		subnet.Status.V4UsingIPs,
+		subnet.Status.V6AvailableIPs,
+		subnet.Status.V6UsingIPs,
+	}
+
 	if util.CheckProtocol(subnet.Spec.CIDRBlock) == kubeovnv1.ProtocolIPv4 {
 		subnet.Status.V4AvailableIPs = availableIPs
 		subnet.Status.V4UsingIPs = usingIPs
@@ -1238,6 +1255,14 @@ func calcSubnetStatusIP(subnet *kubeovnv1.Subnet, c *Controller) error {
 		subnet.Status.V4AvailableIPs = 0
 		subnet.Status.V4UsingIPs = 0
 	}
+	if cachedFields == [4]float64{
+		subnet.Status.V4AvailableIPs,
+		subnet.Status.V4UsingIPs,
+		subnet.Status.V6AvailableIPs,
+		subnet.Status.V6UsingIPs,
+	} {
+		return nil
+	}
 
 	bytes, err := subnet.Status.Bytes()
 	if err != nil {