Merge pull request #243 from andyzhangx/read-from-secret

fix: account key list throttling issue

Author: andyzhangx

charts/latest/blob-csi-driver/templates/rbac-csi-blob-controller.yaml

@@ -133,7 +133,7 @@ metadata:
 rules:
   - apiGroups: [""]
     resources: ["secrets"]
-    verbs: ["get", "list"]
+    verbs: ["get", "list", "create"]
 
 ---
 kind: ClusterRoleBinding

deploy/rbac-csi-blob-controller.yaml

@@ -132,7 +132,7 @@ metadata:
 rules:
   - apiGroups: [""]
     resources: ["secrets"]
-    verbs: ["get", "list"]
+    verbs: ["get", "list", "create"]
 
 ---
 kind: ClusterRoleBinding

pkg/blob/azure.go

@@ -72,10 +72,14 @@ func GetCloudProvider(kubeconfig string) (*azureprovider.Cloud, error) {
 		defer f.Close()
 
 		klog.V(2).Infof("read cloud config from file: %s successfully", credFile)
-		return azureprovider.NewCloudWithoutFeatureGates(f)
+		if az, err = azureprovider.NewCloudWithoutFeatureGates(f); err != nil {
+			return az, err
+		}
 	}
 
-	klog.V(2).Infof("read cloud config from secret successfully")
+	if kubeClient != nil {
+		az.KubeClient = kubeClient
+	}
 	return az, nil
 }
 

pkg/blob/blob.go

@@ -27,6 +27,10 @@ import (
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/pborman/uuid"
 	"golang.org/x/net/context"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	k8sutil "k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/legacy-cloud-providers/azure"
@@ -36,20 +40,27 @@ import (
 
 const (
 	// DriverName holds the name of the csi-driver
-	DriverName       = "blob.csi.azure.com"
-	separator        = "#"
-	volumeIDTemplate = "%s#%s#%s"
-	fileMode         = "file_mode"
-	dirMode          = "dir_mode"
-	vers             = "vers"
-	defaultFileMode  = "0777"
-	defaultDirMode   = "0777"
-	defaultVers      = "3.0"
-	serverNameField  = "server"
-	tagsField        = "tags"
-	protocolField    = "protocol"
-	fuse             = "fuse"
-	nfs              = "nfs"
+	DriverName               = "blob.csi.azure.com"
+	separator                = "#"
+	volumeIDTemplate         = "%s#%s#%s"
+	secretNameTemplate       = "azure-storage-account-%s-secret"
+	fileMode                 = "file_mode"
+	dirMode                  = "dir_mode"
+	vers                     = "vers"
+	defaultFileMode          = "0777"
+	defaultDirMode           = "0777"
+	defaultVers              = "3.0"
+	serverNameField          = "server"
+	tagsField                = "tags"
+	protocolField            = "protocol"
+	secretNamespaceField     = "secretnamespace"
+	storeAccountKeyField     = "storeaccountkey"
+	storeAccountKeyFalse     = "false"
+	defaultSecretAccountName = "azurestorageaccountname"
+	defaultSecretAccountKey  = "azurestorageaccountkey"
+	defaultSecretNamespace   = "default"
+	fuse                     = "fuse"
+	nfs                      = "nfs"
 
 	// See https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
 	containerNameMinLength = 3
@@ -296,9 +307,14 @@ func (d *Driver) GetAuthEnv(ctx context.Context, volumeID string, attrib, secret
 				resourceGroupName = d.cloud.ResourceGroup
 			}
 
-			accountKey, err = d.cloud.GetStorageAccesskey(accountName, resourceGroupName)
+			// read from k8s secret first
+			accountKey, err = d.GetStorageAccesskeyFromSecret(accountName, attrib[secretNamespaceField])
 			if err != nil {
-				return accountName, containerName, authEnv, fmt.Errorf("no key for storage account(%s) under resource group(%s), err %v", accountName, resourceGroupName, err)
+				klog.V(2).Infof("could not get account(%s) key from secret, error: %v, use cluster identity to get account key instead", accountName, err)
+				accountKey, err = d.cloud.GetStorageAccesskey(accountName, resourceGroupName)
+				if err != nil {
+					return accountName, containerName, authEnv, fmt.Errorf("no key for storage account(%s) under resource group(%s), err %v", accountName, resourceGroupName, err)
+				}
 			}
 		} else {
 			for k, v := range secrets {
@@ -468,3 +484,70 @@ func getStorageAccount(secrets map[string]string) (string, string, error) {
 	klog.V(4).Infof("got storage account(%s) from secret", accountName)
 	return accountName, accountKey, nil
 }
+
+func setAzureCredentials(kubeClient kubernetes.Interface, accountName, accountKey, secretNamespace string) (string, error) {
+	if kubeClient == nil {
+		klog.Warningf("could not create secret: kubeClient is nil")
+		return "", nil
+	}
+	if accountName == "" || accountKey == "" {
+		return "", fmt.Errorf("the account info is not enough, accountName(%v), accountKey(%v)", accountName, accountKey)
+	}
+	if secretNamespace == "" {
+		secretNamespace = defaultSecretNamespace
+	}
+	secretName := fmt.Sprintf(secretNameTemplate, accountName)
+	secret := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: defaultSecretNamespace,
+			Name:      secretName,
+		},
+		Data: map[string][]byte{
+			defaultSecretAccountName: []byte(accountName),
+			defaultSecretAccountKey:  []byte(accountKey),
+		},
+		Type: "Opaque",
+	}
+	_, err := kubeClient.CoreV1().Secrets(secretNamespace).Create(context.TODO(), secret, metav1.CreateOptions{})
+	if errors.IsAlreadyExists(err) {
+		err = nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("couldn't create secret %v", err)
+	}
+	return secretName, err
+}
+
+// GetStorageAccesskey get Azure storage account key
+func (d *Driver) GetStorageAccesskey(accountOptions *azure.AccountOptions, secrets map[string]string, secretNamespace string) (string, error) {
+	if len(secrets) > 0 {
+		_, accountKey, err := getStorageAccount(secrets)
+		return accountKey, err
+	}
+
+	// read from k8s secret first
+	accountKey, err := d.GetStorageAccesskeyFromSecret(accountOptions.Name, secretNamespace)
+	if err != nil {
+		klog.V(2).Infof("could not get account(%s) key from secret, error: %v, use cluster identity to get account key instead", accountOptions.Name, err)
+		return d.cloud.GetStorageAccesskey(accountOptions.Name, accountOptions.ResourceGroup)
+	}
+	return accountKey, err
+}
+
+// GetStorageAccesskeyFromSecret get storage account key from k8s secret
+func (d *Driver) GetStorageAccesskeyFromSecret(accountName, secretNamespace string) (string, error) {
+	if d.cloud.KubeClient == nil {
+		return "", fmt.Errorf("could not get account(%s) key from secret: KubeClient is nil", accountName)
+	}
+
+	secretName := fmt.Sprintf(secretNameTemplate, accountName)
+	if secretNamespace == "" {
+		secretNamespace = defaultSecretNamespace
+	}
+	secret, err := d.cloud.KubeClient.CoreV1().Secrets(secretNamespace).Get(context.TODO(), secretName, metav1.GetOptions{})
+	if err != nil {
+		return "", fmt.Errorf("could not get secret(%v): %v", secretName, err)
+	}
+
+	return string(secret.Data[defaultSecretAccountKey][:]), nil
+}

pkg/blob/blob_test.go

@@ -31,6 +31,8 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/legacy-cloud-providers/azure"
 	"k8s.io/legacy-cloud-providers/azure/clients/storageaccountclient/mockstorageaccountclient"
 )
@@ -700,3 +702,59 @@ func TestGetStorageAccount(t *testing.T) {
 		}
 	}
 }
+
+func TestSetAzureCredentials(t *testing.T) {
+	fakeClient := fake.NewSimpleClientset()
+
+	tests := []struct {
+		desc            string
+		kubeClient      kubernetes.Interface
+		accountName     string
+		accountKey      string
+		secretNamespace string
+		expectedName    string
+		expectedErr     error
+	}{
+		{
+			desc:        "[failure] accountName is nil",
+			kubeClient:  fakeClient,
+			expectedErr: fmt.Errorf("the account info is not enough, accountName(), accountKey()"),
+		},
+		{
+			desc:        "[failure] accountKey is nil",
+			kubeClient:  fakeClient,
+			accountName: "testName",
+			accountKey:  "",
+			expectedErr: fmt.Errorf("the account info is not enough, accountName(testName), accountKey()"),
+		},
+		{
+			desc:        "[success] kubeClient is nil",
+			kubeClient:  nil,
+			expectedErr: nil,
+		},
+		{
+			desc:         "[success] normal scenario",
+			kubeClient:   fakeClient,
+			accountName:  "testName",
+			accountKey:   "testKey",
+			expectedName: "azure-storage-account-testName-secret",
+			expectedErr:  nil,
+		},
+		{
+			desc:         "[success] already exist",
+			kubeClient:   fakeClient,
+			accountName:  "testName",
+			accountKey:   "testKey",
+			expectedName: "azure-storage-account-testName-secret",
+			expectedErr:  nil,
+		},
+	}
+
+	for _, test := range tests {
+		result, err := setAzureCredentials(test.kubeClient, test.accountName, test.accountKey, test.secretNamespace)
+		if result != test.expectedName || !reflect.DeepEqual(err, test.expectedErr) {
+			t.Errorf("desc: %s,\n input: kubeClient(%v), accountName(%v), accountKey(%v),\n setAzureCredentials result: %v, expectedName: %v err: %v, expectedErr: %v",
+				test.desc, test.kubeClient, test.accountName, test.accountKey, result, test.expectedName, err, test.expectedErr)
+		}
+	}
+}

pkg/blob/controllerserver.go

@@ -53,7 +53,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	requestGiB := int(util.RoundUpGiB(volSizeBytes))
 
 	parameters := req.GetParameters()
-	var storageAccountType, resourceGroup, location, account, containerName, protocol, customTags string
+	var storageAccountType, resourceGroup, location, account, containerName, protocol, customTags, storeAccountKey, secretNamespace string
 
 	// Apply ProvisionerParameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
@@ -75,6 +75,10 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 			protocol = v
 		case tagsField:
 			customTags = v
+		case secretNamespaceField:
+			secretNamespace = v
+		case storeAccountKeyField:
+			storeAccountKey = v
 		}
 	}
 
@@ -95,6 +99,8 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 			return nil, status.Errorf(codes.InvalidArgument, "storage account must be specified when provisioning nfs file share")
 		}
 		enableHTTPSTrafficOnly = false
+		// NFS protocol does not need account key
+		storeAccountKey = storeAccountKeyFalse
 	}
 
 	accountKind := string(storage.StorageV2)
@@ -117,9 +123,10 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		Tags:                   tags,
 	}
 
-	var accountName, accountKey string
-	if len(req.GetSecrets()) == 0 { // check whether account is provided by secret
-		lockKey := account + storageAccountType + accountKind + resourceGroup + location
+	var accountKey string
+	accountName := account
+	if len(req.GetSecrets()) == 0 && accountName == "" {
+		lockKey := storageAccountType + accountKind + resourceGroup + location
 		d.volLockMap.LockEntry(lockKey)
 		defer d.volLockMap.UnlockEntry(lockKey)
 
@@ -135,10 +142,12 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to ensure storage account: %v", err)
 		}
-	} else {
-		accountName, accountKey, err = getStorageAccount(req.GetSecrets())
-		if err != nil {
-			return nil, status.Errorf(codes.Internal, "failed to get storage account from secrets: %v", err)
+	}
+	accountOptions.Name = accountName
+
+	if accountKey == "" {
+		if accountKey, err = d.GetStorageAccesskey(accountOptions, req.GetSecrets(), secretNamespace); err != nil {
+			return nil, fmt.Errorf("failed to GetStorageAccesskey on account(%s) rg(%s), error: %v", accountOptions.Name, accountOptions.ResourceGroup, err)
 		}
 	}
 
@@ -159,15 +168,17 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	}
 
 	volumeID := fmt.Sprintf(volumeIDTemplate, resourceGroup, accountName, containerName)
+	klog.V(2).Infof("create container %s on storage account %s successfully", containerName, accountName)
 
-	/* todo: snapshot support
-	if req.GetVolumeContentSource() != nil {
-		contentSource := req.GetVolumeContentSource()
-		if contentSource.GetSnapshot() != nil {
+	if storeAccountKey != storeAccountKeyFalse && len(req.GetSecrets()) == 0 {
+		secretName, err := setAzureCredentials(d.cloud.KubeClient, accountName, accountKey, secretNamespace)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to store storage account key: %v", err)
+		}
+		if secretName != "" {
+			klog.V(2).Infof("store account key to k8s secret(%v) in %s namespace", secretName, secretNamespace)
 		}
 	}
-	*/
-	klog.V(2).Infof("create container %s on storage account %s successfully", containerName, accountName)
 
 	return &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{