Merge pull request #751 from andyzhangx/dataplane-issue

fix: delete volume failure when management api is throttled

Author: andyzhangx

pkg/blob/blob.go

@@ -173,8 +173,8 @@ type Driver struct {
 	subnetLockMap *util.LockMap
 	// a map storing all volumes created by this driver <volumeName, accountName>
 	volMap sync.Map
-	// a map storing all volumes using data plane API <volumeID, "">, <accountName, "">
-	dataPlaneAPIVolMap sync.Map
+	// a timed cache storing all volumeIDs and storage accounts that are using data plane API
+	dataPlaneAPIVolCache *azcache.TimedCache
 	// a timed cache storing account search history (solve account list throttling issue)
 	accountSearchCache *azcache.TimedCache
 }
@@ -208,6 +208,9 @@ func NewDriver(options *DriverOptions) *Driver {
 	if d.accountSearchCache, err = azcache.NewTimedcache(time.Minute, getter); err != nil {
 		klog.Fatalf("%v", err)
 	}
+	if d.dataPlaneAPIVolCache, err = azcache.NewTimedcache(10*time.Minute, getter); err != nil {
+		klog.Fatalf("%v", err)
+	}
 	return &d
 }
 
@@ -746,11 +749,21 @@ func (d *Driver) getSubnetResourceID() string {
 }
 
 func (d *Driver) useDataPlaneAPI(volumeID, accountName string) bool {
-	_, useDataPlaneAPI := d.dataPlaneAPIVolMap.Load(volumeID)
-	if !useDataPlaneAPI {
-		_, useDataPlaneAPI = d.dataPlaneAPIVolMap.Load(accountName)
+	cache, err := d.dataPlaneAPIVolCache.Get(volumeID, azcache.CacheReadTypeDefault)
+	if err != nil {
+		klog.Errorf("get(%s) from dataPlaneAPIVolCache failed with error: %v", volumeID, err)
+	}
+	if cache != nil {
+		return true
 	}
-	return useDataPlaneAPI
+	cache, err = d.dataPlaneAPIVolCache.Get(accountName, azcache.CacheReadTypeDefault)
+	if err != nil {
+		klog.Errorf("get(%s) from dataPlaneAPIVolCache failed with error: %v", accountName, err)
+	}
+	if cache != nil {
+		return true
+	}
+	return false
 }
 
 // appendDefaultMountOptions return mount options combined with mountOptions and defaultMountOptions

pkg/blob/blob_test.go

@@ -91,6 +91,7 @@ func TestNewDriver(t *testing.T) {
 	fakedriver.Name = DefaultDriverName
 	fakedriver.Version = driverVersion
 	fakedriver.accountSearchCache = driver.accountSearchCache
+	fakedriver.dataPlaneAPIVolCache = driver.dataPlaneAPIVolCache
 	assert.Equal(t, driver, fakedriver)
 }
 
@@ -1184,7 +1185,7 @@ func TestUseDataPlaneAPI(t *testing.T) {
 			name: "volumeID loads correctly",
 			testFunc: func(t *testing.T) {
 				d := NewFakeDriver()
-				d.dataPlaneAPIVolMap.LoadOrStore(fakeVolumeID, "foo")
+				d.dataPlaneAPIVolCache.Set(fakeVolumeID, "foo")
 				output := d.useDataPlaneAPI(fakeVolumeID, "")
 				if !output {
 					t.Errorf("Actual Output: %t, Expected Output: %t", output, true)
@@ -1195,7 +1196,7 @@ func TestUseDataPlaneAPI(t *testing.T) {
 			name: "account loads correctly",
 			testFunc: func(t *testing.T) {
 				d := NewFakeDriver()
-				d.dataPlaneAPIVolMap.LoadOrStore(fakeAccountName, "foo")
+				d.dataPlaneAPIVolCache.Set(fakeAccountName, "foo")
 				output := d.useDataPlaneAPI(fakeAccountName, "")
 				if !output {
 					t.Errorf("Actual Output: %t, Expected Output: %t", output, true)

pkg/blob/controllerserver.go

@@ -347,8 +347,8 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	klog.V(2).Infof("create container %s on storage account %s successfully", validContainerName, accountName)
 
 	if useDataPlaneAPI {
-		d.dataPlaneAPIVolMap.Store(volumeID, "")
-		d.dataPlaneAPIVolMap.Store(accountName, "")
+		d.dataPlaneAPIVolCache.Set(volumeID, "")
+		d.dataPlaneAPIVolCache.Set(accountName, "")
 	}
 
 	isOperationSucceeded = true

pkg/blob/controllerserver_test.go

@@ -792,7 +792,7 @@ func TestDeleteVolume(t *testing.T) {
 				req := &csi.DeleteVolumeRequest{
 					VolumeId: "rg#accountName#containerName",
 				}
-				d.dataPlaneAPIVolMap.Store("rg#accountName#containerName", "accountName")
+				d.dataPlaneAPIVolCache.Set("rg#accountName#containerName", "accountName")
 				_, err := d.DeleteVolume(context.Background(), req)
 				expectedErr := status.Errorf(codes.Internal, "GetAuthEnv(%s) failed with %v", "rg#accountName#containerName", fmt.Errorf("no key for storage account(%s) under resource group(%s), err %w", "accountName", "rg", fmt.Errorf("StorageAccountClient is nil")))
 				if !reflect.DeepEqual(err, expectedErr) {