Merge pull request #186 from andyzhangx/upgrade-provider

chore: upgrade azure cloud provider

Author: andyzhangx

go.mod

@@ -23,7 +23,7 @@ require (
 	k8s.io/klog/v2 v2.2.0
 	k8s.io/kubernetes v1.18.5
 	k8s.io/legacy-cloud-providers v0.0.0
-	k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19
+	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
 	sigs.k8s.io/yaml v1.2.0
 )
 
@@ -50,7 +50,7 @@ replace (
 	k8s.io/kube-scheduler => k8s.io/kube-scheduler v0.18.1
 	k8s.io/kubectl => k8s.io/kubectl v0.18.1
 	k8s.io/kubelet => k8s.io/kubelet v0.18.1
-	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.0.0-20200711020551-224d5470706b
+	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.0.0-20200731174458-cce61a865d46
 	k8s.io/metrics => k8s.io/metrics v0.18.1
 	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.18.1
 	k8s.io/sample-cli-plugin => k8s.io/sample-cli-plugin v0.18.1

go.sum

@@ -594,6 +594,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200220183623-bac4c82f6975 h1:/Tl7pH94bvbAAHBdZJT947M/+gp0+CqQXDtMRC0fseo=
 golang.org/x/crypto v0.0.0-20200220183623-bac4c82f6975/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -852,15 +854,17 @@ k8s.io/kubectl v0.18.1/go.mod h1:7bLKKp5gSrtuSCIbclgifVOCwdLBs+h3oETVqLn1928=
 k8s.io/kubelet v0.18.1/go.mod h1:wlhv8yVMnbTp0haWYfuOzD7fqq7F35bWcIFIpdy9w9A=
 k8s.io/kubernetes v1.18.5 h1:sLbTilDIc02vNzrTORdW2YakzrPCOtmffaUXCwQHsRc=
 k8s.io/kubernetes v1.18.5/go.mod h1:Efg82S+Ti02A/Mww53bxroc7IgzX2bgPsf6hT8gAs3M=
-k8s.io/legacy-cloud-providers v0.0.0-20200711020551-224d5470706b h1:U6s1jf8ZIWD3G/5/vtRh75iDB3RfMP6Y9UpikHgh5to=
-k8s.io/legacy-cloud-providers v0.0.0-20200711020551-224d5470706b/go.mod h1:1PY7/iWuD99kgksPmcb90Jn3Xtl/yW1ICQvKWIwJGEU=
+k8s.io/legacy-cloud-providers v0.0.0-20200731174458-cce61a865d46 h1:ojY3Mxvwnh8f/mP1bVyptw8zzFSS6EyvtCdOxB26NQc=
+k8s.io/legacy-cloud-providers v0.0.0-20200731174458-cce61a865d46/go.mod h1:IJ/IyiwEloImiZd033RM/WAFc32p5KY1hlNIRtGUgT8=
 k8s.io/metrics v0.18.1/go.mod h1:UnoYSbaiakbuBYgpDEsXjVjk843af/WRHNPR3Wx+2XI=
 k8s.io/repo-infra v0.0.1-alpha.1/go.mod h1:wO1t9WaB99V80ljbeENTnayuEEwNZt7gECYh/CEyOJ8=
 k8s.io/sample-apiserver v0.18.1/go.mod h1:ir6PkLhIblVb9nNHzd8LG77QW4mhQAwahznNAIHhOEM=
 k8s.io/system-validators v1.0.4/go.mod h1:HgSgTg4NAGNoYYjKsUyk52gdNi2PVDswQ9Iyn66R7NI=
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19 h1:7Nu2dTj82c6IaWvL7hImJzcXoTPz1MsSCH7r+0m6rfo=
 k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
+k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
+k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=

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)