diff --git a/pkg/blob/blob.go b/pkg/blob/blob.go
index 8b9c3d630..b22ec8472 100644
--- a/pkg/blob/blob.go
+++ b/pkg/blob/blob.go
@@ -27,7 +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"
@@ -37,24 +40,27 @@ import (
 
 const (
 	// DriverName holds the name of the csi-driver
-	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"
-	defaultSecretAccountKey = "azurestorageaccountkey"
-	defaultSecretNamespace  = "default"
-	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
@@ -301,23 +307,10 @@ func (d *Driver) GetAuthEnv(ctx context.Context, volumeID string, attrib, secret
 				resourceGroupName = d.cloud.ResourceGroup
 			}
 
-			if d.cloud.KubeClient != nil {
-				secretName := fmt.Sprintf(secretNameTemplate, accountName)
-				secretNamespace := attrib[secretNamespaceField]
-				if secretNamespace == "" {
-					secretNamespace = defaultSecretNamespace
-				}
-				secret, err := d.cloud.KubeClient.CoreV1().Secrets(secretNamespace).Get(context.TODO(), secretName, metav1.GetOptions{})
-				if err != nil {
-					klog.V(4).Infof("could not get secret(%v): %v", secretName, err)
-				} else {
-					accountKey = string(secret.Data[defaultSecretAccountKey][:])
-				}
-			} else {
-				klog.V(5).Infof("could not get account(%s) key from secret: KubeClient is nil", accountName)
-			}
-
-			if accountKey == "" {
+			// read from k8s secret first
+			accountKey, err = d.GetStorageAccesskeyFromSecret(accountName, attrib[secretNamespaceField])
+			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", 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)
@@ -491,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
+}
diff --git a/pkg/blob/blob_test.go b/pkg/blob/blob_test.go
index e2289eb1c..ece261285 100644
--- a/pkg/blob/blob_test.go
+++ b/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)
+		}
+	}
+}
diff --git a/pkg/blob/controllerserver.go b/pkg/blob/controllerserver.go
index 56c35b9a8..307384f7f 100644
--- a/pkg/blob/controllerserver.go
+++ b/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{