chore: fix build error using cloud provider lib

Author: andyzhangx

pkg/blobfuse/blobfuse.go

@@ -44,10 +44,21 @@ const (
 	defaultDirMode   = "0777"
 	defaultVers      = "3.0"
 	serverNameField  = "server"
+	tagsField        = "tags"
 
 	// See https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
 	containerNameMinLength = 3
 	containerNameMaxLength = 63
+
+	accountNotProvisioned = "StorageAccountIsNotProvisioned"
+	tooManyRequests       = "TooManyRequests"
+	shareNotFound         = "The specified share does not exist"
+	shareBeingDeleted     = "The specified share is being deleted"
+	clientThrottled       = "client throttled"
+)
+
+var (
+	retriableErrors = []string{accountNotProvisioned, tooManyRequests, shareNotFound, shareBeingDeleted, clientThrottled}
 )
 
 // Driver implements all interfaces of CSI drivers
@@ -395,3 +406,14 @@ func IsCorruptedDir(dir string) bool {
 	fmt.Printf("IsCorruptedDir(%s) returned with error: (%v, %v)\\n", dir, pathExists, pathErr)
 	return pathErr != nil && mount.IsCorruptedMnt(pathErr)
 }
+
+func isRetriableError(err error) bool {
+	if err != nil {
+		for _, v := range retriableErrors {
+			if strings.Contains(strings.ToLower(err.Error()), strings.ToLower(v)) {
+				return true
+			}
+		}
+	}
+	return false
+}

pkg/blobfuse/blobfuse_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package blobfuse
 
 import (
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -136,6 +137,53 @@ func TestGetContainerInfo(t *testing.T) {
 	}
 }
 
+func TestIsRetriableError(t *testing.T) {
+	tests := []struct {
+		desc         string
+		rpcErr       error
+		expectedBool bool
+	}{
+		{
+			desc:         "non-retriable error",
+			rpcErr:       nil,
+			expectedBool: false,
+		},
+		{
+			desc:         "accountNotProvisioned",
+			rpcErr:       errors.New("could not get storage key for storage account : could not get storage key for storage account f233333: Retriable: true, RetryAfter: 0001-01-01 00:00:00 +0000 UTC, HTTPStatusCode: 409, RawError: storage.AccountsClient#ListKeys: Failure sending request: StatusCode=409 -- Original Error: autorest/azure: Service returned an error. Status=<nil> Code=\"StorageAccountIsNotProvisioned\" Message=\"The storage account provisioning state must be 'Succeeded' before executing the operation.\""),
+			expectedBool: true,
+		},
+		{
+			desc:         "tooManyRequests",
+			rpcErr:       errors.New("could not get storage key for storage account : could not list storage accounts for account type Premium_LRS: Retriable: true, RetryAfter: 0001-01-01 00:00:00 +0000 UTC m=+231.866923225, HTTPStatusCode: 429, RawError: storage.AccountsClient#ListByResourceGroup: Failure responding to request: StatusCode=429 -- Original Error: autorest/azure: Service returned an error. Status=429 Code=\"TooManyRequests\" Message=\"The request is being throttled as the limit has been reached for operation type - List. For more information, see - https://aka.ms/srpthrottlinglimits\""),
+			expectedBool: true,
+		},
+		{
+			desc:         "shareNotFound",
+			rpcErr:       errors.New("storage.FileSharesClient#Get: Failure responding to request: StatusCode=404 -- Original Error: autorest/azure: Service returned an error. Status=404 Code=\"ShareNotFound\" Message=\"The specified share does not exist\""),
+			expectedBool: true,
+		},
+		{
+			desc:         "shareBeingDeleted",
+			rpcErr:       errors.New("storage.FileSharesClient#Create: Failure sending request: StatusCode=409 -- Original Error: autorest/azure: Service returned an error. Status=<nil> Code=\"ShareBeingDeleted\" Message=\"The specified share is being deleted. Try operation later.\""),
+			expectedBool: true,
+		},
+		{
+			desc:         "clientThrottled",
+			rpcErr:       errors.New("could not list storage accounts for account type : Retriable: true, RetryAfter: 16s, HTTPStatusCode: 0, RawError: azure cloud provider throttled for operation StorageAccountListByResourceGroup with reason \"client throttled\""),
+			expectedBool: true,
+		},
+	}
+
+	for _, test := range tests {
+		result := isRetriableError(test.rpcErr)
+		if result != test.expectedBool {
+			t.Errorf("desc: (%s), input: rpcErr(%v), isRetriableError returned with bool(%v), not equal to expectedBool(%v)",
+				test.desc, test.rpcErr, result, test.expectedBool)
+		}
+	}
+}
+
 func TestGetValidContainerName(t *testing.T) {
 	tests := []struct {
 		volumeName string

pkg/blobfuse/controllerserver.go

@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
+	"k8s.io/legacy-cloud-providers/azure"
 )
 
 const blobfuseAccountNamePrefix = "fuse"
@@ -54,7 +55,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	requestGiB := int(util.RoundUpGiB(volSizeBytes))
 
 	parameters := req.GetParameters()
-	var storageAccountType, resourceGroup, location, accountName, containerName string
+	var storageAccountType, resourceGroup, location, account, containerName, customTags string
 
 	// Apply ProvisionerParameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
@@ -67,11 +68,13 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		case "location":
 			location = v
 		case "storageaccount":
-			accountName = v
+			account = v
 		case "resourcegroup":
 			resourceGroup = v
 		case "containername":
 			containerName = v
+		case tagsField:
+			customTags = v
 		}
 	}
 
@@ -83,11 +86,38 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	if strings.HasPrefix(strings.ToLower(storageAccountType), "premium") {
 		accountKind = string(storage.BlockBlobStorage)
 	}
-	account, accountKey, err := d.cloud.EnsureStorageAccount(accountName, storageAccountType, accountKind, resourceGroup, location, blobfuseAccountNamePrefix, false)
+
+	enableHTTPSTrafficOnly := true
+
+	tags, err := azure.ConvertTagsToMap(customTags)
+	if err != nil {
+		return nil, err
+	}
+
+	accountOptions := &azure.AccountOptions{
+		Name:                   account,
+		Type:                   storageAccountType,
+		Kind:                   accountKind,
+		ResourceGroup:          resourceGroup,
+		Location:               location,
+		EnableHTTPSTrafficOnly: enableHTTPSTrafficOnly,
+		Tags:                   tags,
+	}
+
+	var accountName, accountKey string
+	err = wait.ExponentialBackoff(d.cloud.RequestBackoff(), func() (bool, error) {
+		var retErr error
+		accountName, accountKey, retErr = d.cloud.EnsureStorageAccount(accountOptions, blobfuseAccountNamePrefix)
+		if isRetriableError(retErr) {
+			klog.Warningf("EnsureStorageAccount(%s) failed with error(%v), waiting for retrying", account, retErr)
+			return false, nil
+		}
+		return true, retErr
+	})
+
 	if err != nil {
-		return nil, fmt.Errorf("could not get storage key for storage account %s: %v", accountName, err)
+		return nil, status.Errorf(codes.Internal, "failed to ensure storage account: %v", err)
 	}
-	accountName = account
 
 	if containerName == "" {
 		containerName = getValidContainerName(name)