feat: add account search cache to prevent account list throttling

Author: andyzhangx

pkg/blob/blob.go

@@ -19,6 +19,7 @@ package blob
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"golang.org/x/net/context"
 
@@ -36,6 +37,7 @@ import (
 
 	csicommon "sigs.k8s.io/blob-csi-driver/pkg/csi-common"
 	"sigs.k8s.io/blob-csi-driver/pkg/util"
+	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
 )
 
@@ -137,6 +139,8 @@ type Driver struct {
 	volumeLocks *volumeLocks
 	// only for nfs feature
 	subnetLockMap *util.LockMap
+	// a timed cache storing acount search history (solve account list throttling issue)
+	accountSearchCache *azcache.TimedCache
 }
 
 // NewDriver Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
@@ -158,6 +162,12 @@ func NewDriver(options *DriverOptions) *Driver {
 	d.Name = options.DriverName
 	d.Version = driverVersion
 	d.NodeID = options.NodeID
+
+	var err error
+	getter := func(key string) (interface{}, error) { return nil, nil }
+	if d.accountSearchCache, err = azcache.NewTimedcache(time.Minute, getter); err != nil {
+		klog.Fatalf("%v", err)
+	}
 	return &d
 }
 

pkg/blob/blob_test.go

@@ -87,6 +87,7 @@ func TestNewDriver(t *testing.T) {
 	fakedriver := NewFakeDriver()
 	fakedriver.Name = DefaultDriverName
 	fakedriver.Version = driverVersion
+	fakedriver.accountSearchCache = driver.accountSearchCache
 	assert.Equal(t, driver, fakedriver)
 }
 

pkg/blob/controllerserver.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"sigs.k8s.io/blob-csi-driver/pkg/util"
+	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	"sigs.k8s.io/cloud-provider-azure/pkg/metrics"
 	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
 )
@@ -189,20 +190,30 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	var accountKey string
 	accountName := account
 	if len(req.GetSecrets()) == 0 && accountName == "" {
-		lockKey := storageAccountType + accountKind + resourceGroup + location
-		d.volLockMap.LockEntry(lockKey)
-		err = wait.ExponentialBackoff(d.cloud.RequestBackoff(), func() (bool, error) {
-			var retErr error
-			accountName, accountKey, retErr = d.cloud.EnsureStorageAccount(ctx, accountOptions, protocol)
-			if isRetriableError(retErr) {
-				klog.Warningf("EnsureStorageAccount(%s) failed with error(%v), waiting for retrying", account, retErr)
-				return false, nil
-			}
-			return true, retErr
-		})
-		d.volLockMap.UnlockEntry(lockKey)
+		lockKey := fmt.Sprintf("%s%s%s%s%s", storageAccountType, accountKind, resourceGroup, location, protocol)
+		// search in cache first
+		cache, err := d.accountSearchCache.Get(lockKey, azcache.CacheReadTypeDefault)
 		if err != nil {
-			return nil, status.Errorf(codes.Internal, "failed to ensure storage account: %v", err)
+			return nil, err
+		}
+		if cache != nil {
+			accountName = cache.(string)
+		} else {
+			d.volLockMap.LockEntry(lockKey)
+			err = wait.ExponentialBackoff(d.cloud.RequestBackoff(), func() (bool, error) {
+				var retErr error
+				accountName, accountKey, retErr = d.cloud.EnsureStorageAccount(ctx, accountOptions, protocol)
+				if isRetriableError(retErr) {
+					klog.Warningf("EnsureStorageAccount(%s) failed with error(%v), waiting for retrying", account, retErr)
+					return false, nil
+				}
+				return true, retErr
+			})
+			d.volLockMap.UnlockEntry(lockKey)
+			if err != nil {
+				return nil, status.Errorf(codes.Internal, "failed to ensure storage account: %v", err)
+			}
+			d.accountSearchCache.Set(lockKey, accountName)
 		}
 	}
 	accountOptions.Name = accountName