Add lock to subnet to avoid race between gc and controller (vmware-tanzu#770)

Subnetport/pod controller will acquire a subnet, create nsx subnetport on it and
save the subnetport to store after it is created. But Subnetset GC may delete the
subnet during the subnetport creation

The PR adds locks for subnet by subnet path to avoid this race condition.

Testing done:
Decreased the subnetset garbage collector running interval to 10 seconds to increase
the potential of race. Created 8 pods, waited for all related subnetports created without
error and delete the pods. Repeated for 5 times and not found deadlock or race issue.

Signed-off-by: Yanjun Zhou <[email protected]>

Author: yanjunz97

pkg/controllers/pod/pod_controller.go

@@ -88,6 +88,11 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 			return common.ResultRequeue, err
 		}
 		contextID := *node.UniqueId
+		// There is a race condition that the subnetset controller may delete the
+		// subnet during CollectGarbage. So check the subnet under lock.
+		r.SubnetService.LockSubnet(&nsxSubnetPath)
+		defer r.SubnetService.UnlockSubnet(&nsxSubnetPath)
+
 		nsxSubnet, err := r.SubnetService.GetSubnetByPath(nsxSubnetPath)
 		if err != nil {
 			return common.ResultRequeue, err

pkg/controllers/subnetport/subnetport_controller.go

@@ -105,6 +105,11 @@ func (r *SubnetPortReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			updateFail(r, ctx, subnetPort, &err)
 			return common.ResultRequeue, err
 		}
+		// There is a race condition that the subnetset controller may delete the
+		// subnet during CollectGarbage. So check the subnet under lock.
+		r.SubnetService.LockSubnet(&nsxSubnetPath)
+		defer r.SubnetService.UnlockSubnet(&nsxSubnetPath)
+
 		nsxSubnet, err := r.SubnetService.GetSubnetByPath(nsxSubnetPath)
 		if err != nil {
 			updateFail(r, ctx, subnetPort, &err)

pkg/controllers/subnetset/subnetset_controller.go

@@ -293,16 +293,18 @@ func (r *SubnetSetReconciler) DeleteSubnetForSubnetSet(obj v1alpha1.SubnetSet, u
 	hitError := false
 	hasStaleSubnetPorts := false
 	for _, subnet := range nsxSubnets {
+		r.SubnetService.LockSubnet(subnet.Path)
 		portNums := len(r.SubnetPortService.GetPortsOfSubnet(*subnet.Id))
 		if portNums > 0 {
 			hasStaleSubnetPorts = true
+			r.SubnetService.UnlockSubnet(subnet.Path)
 			continue
 		}
 		if err := r.SubnetService.DeleteSubnet(*subnet); err != nil {
 			log.Error(err, "fail to delete subnet from subnetset cr", "ID", *subnet.Id)
 			hitError = true
 		}
-
+		r.SubnetService.UnlockSubnet(subnet.Path)
 	}
 	if updataStatus {
 		if err := r.SubnetService.UpdateSubnetSetStatus(&obj); err != nil {

pkg/nsx/services/common/services.go

@@ -30,6 +30,8 @@ type SubnetServiceProvider interface {
 	GetSubnetsByIndex(key, value string) []*model.VpcSubnet
 	CreateOrUpdateSubnet(obj client.Object, vpcInfo VPCResourceInfo, tags []model.Tag) (string, error)
 	GenerateSubnetNSTags(obj client.Object, nsUID string) []model.Tag
+	LockSubnet(path *string)
+	UnlockSubnet(path *string)
 }
 
 type SubnetPortServiceProvider interface {

pkg/nsx/services/subnet/store.go

@@ -2,6 +2,7 @@ package subnet
 
 import (
 	"errors"
+	"sync"
 
 	"github.com/vmware/vsphere-automation-sdk-go/services/nsxt/model"
 
@@ -51,6 +52,41 @@ func subnetSetIndexFunc(obj interface{}) ([]string, error) {
 // SubnetStore is a store for subnet.
 type SubnetStore struct {
 	common.ResourceStore
+	// save locks for subnet by path
+	pathLocks sync.Map
+}
+
+func (subnetStore *SubnetStore) Add(i interface{}) error {
+	subnet := i.(*model.VpcSubnet)
+	if subnet.Path == nil {
+		log.Info("Store a subnet without path", "subnet", subnet)
+		return subnetStore.ResourceStore.Add(i)
+	}
+	lock := sync.Mutex{}
+	subnetStore.pathLocks.LoadOrStore(*subnet.Path, &lock)
+	return subnetStore.ResourceStore.Add(i)
+}
+
+func (subnetStore *SubnetStore) Delete(i interface{}) error {
+	subnet := i.(*model.VpcSubnet)
+	if subnet.Path == nil {
+		log.Info("Delete a subnet without path", "subnet", subnet)
+		return subnetStore.ResourceStore.Delete(i)
+	}
+	subnetStore.pathLocks.Delete(*subnet.Path)
+	return subnetStore.ResourceStore.Delete(i)
+}
+
+func (subnetStore *SubnetStore) Lock(path string) {
+	lock := sync.Mutex{}
+	subnetLock, _ := subnetStore.pathLocks.LoadOrStore(path, &lock)
+	subnetLock.(*sync.Mutex).Lock()
+}
+
+func (subnetStore *SubnetStore) Unlock(path string) {
+	if subnetLock, existed := subnetStore.pathLocks.Load(path); existed {
+		subnetLock.(*sync.Mutex).Unlock()
+	}
 }
 
 func (subnetStore *SubnetStore) Apply(i interface{}) error {

pkg/nsx/services/subnet/subnet.go

@@ -421,3 +421,17 @@ func (service *SubnetService) UpdateSubnetSetTags(ns string, vpcSubnets []*model
 	}
 	return nil
 }
+
+func (service *SubnetService) LockSubnet(path *string) {
+	if path != nil && *path != "" {
+		log.V(1).Info("locked subnet", "path", *path)
+		service.SubnetStore.Lock(*path)
+	}
+}
+
+func (service *SubnetService) UnlockSubnet(path *string) {
+	if path != nil && *path != "" {
+		log.V(1).Info("unlocked subnet", "path", *path)
+		service.SubnetStore.Unlock(*path)
+	}
+}