feat: support data plane api using Oauth on smb file share operations

Author: andyzhangx

pkg/azurefile/azure.go

@@ -181,7 +181,7 @@ func getCloudProvider(ctx context.Context, kubeconfig, nodeID, secretName, secre
 			klog.V(2).Infof("starting node server on node(%s)", nodeID)
 		}
 
-		repo, err = storage.NewRepository(*config, env, az.ComputeClientFactory, az.NetworkClientFactory)
+		repo, err = storage.NewRepository(*config, env, az.AuthProvider, az.ComputeClientFactory, az.NetworkClientFactory)
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to create storage repository: %v", err)
 		}

pkg/azurefile/azure_test.go

@@ -318,7 +318,7 @@ var _ = ginkgo.Describe("AzureFile", func() {
 		var err error
 		d.cloud, err = storage.NewRepository(config, &azclient.Environment{
 			StorageEndpointSuffix: "fake-endpoint",
-		}, mockClientFactory, mockClientFactory)
+		}, nil, mockClientFactory, mockClientFactory)
 		gomega.Expect(err).NotTo(gomega.HaveOccurred(), "Unexpected error:", err)
 	})
 	ginkgo.AfterEach(func() {

pkg/azurefile/azurefile.go

@@ -166,7 +166,7 @@ const (
 	accountLimitExceedManagementAPI = "TotalSharesProvisionedCapacityExceedsAccountLimit"
 	accountLimitExceedDataPlaneAPI  = "specified share does not exist"
 
-	fileShareNotFound  = "ErrorCode=ShareNotFound"
+	fileShareNotFound  = "ShareNotFound"
 	statusCodeNotFound = "StatusCode=404"
 	httpCodeNotFound   = "HTTPStatusCode: 404"
 
@@ -196,6 +196,8 @@ const (
 	FSGroupChangeNone = "None"
 	// define tag value delimiter and default is comma
 	tagValueDelimiterField = "tagvaluedelimiter"
+	// for data plane API
+	oauth = "oauth"
 )
 
 var (
@@ -260,7 +262,7 @@ type Driver struct {
 	accountCacheMap azcache.Resource
 	// a map storing all secret names created by this driver <secretCacheKey, "">
 	secretCacheMap azcache.Resource
-	// a map storing all volumes using data plane API <volumeID, "">
+	// a map storing all volumes using data plane API <volumeID, value>
 	dataPlaneAPIVolMap sync.Map
 	// a timed cache storing all storage accounts that are using data plane API temporarily
 	dataPlaneAPIAccountCache azcache.Resource
@@ -473,25 +475,27 @@ func (d *Driver) Run(ctx context.Context) error {
 }
 
 // getFileShareQuota return (-1, nil) means file share does not exist
-func (d *Driver) getFileShareQuota(ctx context.Context, accountOptions *storage.AccountOptions, fileShareName string, secrets map[string]string) (int, error) {
+func (d *Driver) getFileShareQuota(ctx context.Context, accountOptions *storage.AccountOptions, fileShareName string, secrets map[string]string, useDataPlaneAPI string) (int, error) {
 	var fileClient azureFileClient
 	var err error
 	if len(secrets) > 0 {
-		accountName, accountKey, err := getStorageAccount(secrets)
-		if err != nil {
-			return -1, err
+		accountName, accountKey, rerr := getStorageAccount(secrets)
+		if rerr != nil {
+			return -1, rerr
 		}
 		storageEndPointSuffix := d.getStorageEndPointSuffix()
 		if accountOptions != nil && accountOptions.StorageEndpointSuffix != "" {
 			storageEndPointSuffix = accountOptions.StorageEndpointSuffix
 		}
-		if fileClient, err = newAzureFileClient(accountName, accountKey, storageEndPointSuffix); err != nil {
-			return -1, err
-		}
+		fileClient, err = newAzureFileClient(accountName, accountKey, storageEndPointSuffix)
+	} else if d.cloud != nil && d.cloud.AuthProvider != nil && strings.EqualFold(useDataPlaneAPI, oauth) {
+		fileClient, err = newAzureFileClientWithOAuth(d.cloud.AuthProvider.GetAzIdentity(), accountOptions.Name, d.getStorageEndPointSuffix())
 	} else {
-		if fileClient, err = newAzureFileMgmtClient(d.cloud, accountOptions); err != nil {
-			return -1, err
-		}
+		fileClient, err = newAzureFileMgmtClient(d.cloud, accountOptions)
+	}
+
+	if err != nil {
+		return -1, err
 	}
 	quota, err := fileClient.GetFileShareQuota(ctx, fileShareName)
 	if err != nil {
@@ -709,7 +713,7 @@ func getDirectoryClient(accountName, accountKey, storageEndpointSuffix, fileShar
 	if u == nil {
 		return nil, fmt.Errorf("parse fileURLTemplate error: url is nil")
 	}
-	serviceURL := fmt.Sprintf("https://%s.file.%s/", accountName, storageEndpointSuffix)
+	serviceURL := getFileServiceURL(accountName, storageEndpointSuffix)
 
 	serviceClient, err := service.NewClientWithSharedKeyCredential(serviceURL, credential, nil)
 	if err != nil {
@@ -942,7 +946,7 @@ func isSupportedFSGroupChangePolicy(policy string) bool {
 }
 
 // CreateFileShare creates a file share
-func (d *Driver) CreateFileShare(ctx context.Context, accountOptions *storage.AccountOptions, shareOptions *ShareOptions, secrets map[string]string) error {
+func (d *Driver) CreateFileShare(ctx context.Context, accountOptions *storage.AccountOptions, shareOptions *ShareOptions, secrets map[string]string, useDataPlaneAPI string) error {
 	return wait.ExponentialBackoff(getBackOff(d.cloud.Config), func() (bool, error) {
 		var err error
 		var fileClient azureFileClient
@@ -959,12 +963,20 @@ func (d *Driver) CreateFileShare(ctx context.Context, accountOptions *storage.Ac
 			if fileClient, err = newAzureFileClient(accountName, accountKey, storageEndPointSuffix); err != nil {
 				return true, err
 			}
+		} else if d.cloud != nil && d.cloud.AuthProvider != nil && strings.EqualFold(useDataPlaneAPI, oauth) {
+			fileClient, err = newAzureFileClientWithOAuth(d.cloud.AuthProvider.GetAzIdentity(), accountOptions.Name, d.getStorageEndPointSuffix())
 		} else {
-			if fileClient, err = newAzureFileMgmtClient(d.cloud, accountOptions); err != nil {
-				return true, err
-			}
+			fileClient, err = newAzureFileMgmtClient(d.cloud, accountOptions)
 		}
+		if err != nil {
+			return true, err
+		}
+
 		if err = fileClient.CreateFileShare(ctx, shareOptions); err != nil {
+			if strings.Contains(err.Error(), "ShareAlreadyExists") {
+				klog.Warningf("CreateFileShare(%s) on account(%s) failed with error(%v), return as success", shareOptions.Name, accountOptions.Name, err)
+				return true, nil
+			}
 			if isRetriableError(err) {
 				klog.Warningf("CreateFileShare(%s) on account(%s) failed with error(%v), waiting for retrying", shareOptions.Name, accountOptions.Name, err)
 				sleepIfThrottled(err, fileOpThrottlingSleepSec)
@@ -977,7 +989,7 @@ func (d *Driver) CreateFileShare(ctx context.Context, accountOptions *storage.Ac
 }
 
 // DeleteFileShare deletes a file share using storage account name and key
-func (d *Driver) DeleteFileShare(ctx context.Context, subsID, resourceGroup, accountName, shareName string, secrets map[string]string) error {
+func (d *Driver) DeleteFileShare(ctx context.Context, subsID, resourceGroup, accountName, shareName string, secrets map[string]string, useDataPlaneAPI string) error {
 	return wait.ExponentialBackoff(getBackOff(d.cloud.Config), func() (bool, error) {
 		var err error
 		if len(secrets) > 0 {
@@ -990,6 +1002,12 @@ func (d *Driver) DeleteFileShare(ctx context.Context, subsID, resourceGroup, acc
 				return true, rerr
 			}
 			err = fileClient.DeleteFileShare(ctx, shareName)
+		} else if d.cloud != nil && d.cloud.AuthProvider != nil && strings.EqualFold(useDataPlaneAPI, oauth) {
+			fileClient, rerr := newAzureFileClientWithOAuth(d.cloud.AuthProvider.GetAzIdentity(), accountName, d.getStorageEndPointSuffix())
+			if rerr != nil {
+				return true, rerr
+			}
+			err = fileClient.DeleteFileShare(ctx, shareName)
 		} else {
 			fileClient, errGetClient := d.getFileShareClientForSub(subsID)
 			if errGetClient != nil {
@@ -1022,7 +1040,7 @@ func (d *Driver) DeleteFileShare(ctx context.Context, subsID, resourceGroup, acc
 }
 
 // ResizeFileShare resizes a file share
-func (d *Driver) ResizeFileShare(ctx context.Context, subsID, resourceGroup, accountName, shareName string, sizeGiB int, secrets map[string]string) error {
+func (d *Driver) ResizeFileShare(ctx context.Context, subsID, resourceGroup, accountName, shareName string, sizeGiB int, secrets map[string]string, useDataPlaneAPI string) error {
 	return wait.ExponentialBackoff(getBackOff(d.cloud.Config), func() (bool, error) {
 		var err error
 		if len(secrets) > 0 {
@@ -1035,14 +1053,20 @@ func (d *Driver) ResizeFileShare(ctx context.Context, subsID, resourceGroup, acc
 				return true, rerr
 			}
 			err = fileClient.ResizeFileShare(ctx, shareName, sizeGiB)
+		} else if d.cloud != nil && d.cloud.AuthProvider != nil && strings.EqualFold(useDataPlaneAPI, oauth) {
+			fileClient, rerr := newAzureFileClientWithOAuth(d.cloud.AuthProvider.GetAzIdentity(), accountName, d.getStorageEndPointSuffix())
+			if rerr != nil {
+				return true, rerr
+			}
+			err = fileClient.ResizeFileShare(ctx, shareName, sizeGiB)
 		} else {
-			fileClient, err := d.getFileShareClientForSub(subsID)
-			if err != nil {
-				return true, err
+			fileClient, rerr := d.getFileShareClientForSub(subsID)
+			if rerr != nil {
+				return true, rerr
 			}
-			fileShare, err := fileClient.Get(ctx, resourceGroup, accountName, shareName, nil)
-			if err != nil {
-				return true, err
+			fileShare, rerr := fileClient.Get(ctx, resourceGroup, accountName, shareName, nil)
+			if rerr != nil {
+				return true, rerr
 			}
 			if ptr.Deref(fileShare.FileShareProperties.ShareQuota, 0) >= int32(sizeGiB) {
 				klog.Warningf("file share size(%dGi) is already greater or equal than requested size(%dGi), accountName: %s, shareName: %s",
@@ -1051,7 +1075,6 @@ func (d *Driver) ResizeFileShare(ctx context.Context, subsID, resourceGroup, acc
 			}
 			fileShare.FileShareProperties.ShareQuota = to.Ptr(int32(sizeGiB))
 			_, err = fileClient.Update(ctx, resourceGroup, accountName, shareName, *fileShare)
-			return true, err
 		}
 		if isRetriableError(err) {
 			klog.Warningf("ResizeFileShare(%s) on account(%s) with new size(%d) failed with error(%v), waiting for retrying", shareName, accountName, sizeGiB, err)
@@ -1232,20 +1255,21 @@ func (d *Driver) getSubnetResourceID(vnetResourceGroup, vnetName, subnetName str
 	return fmt.Sprintf(subnetTemplate, subsID, vnetResourceGroup, vnetName, subnetName)
 }
 
-func (d *Driver) useDataPlaneAPI(ctx context.Context, volumeID, accountName string) bool {
-	_, useDataPlaneAPI := d.dataPlaneAPIVolMap.Load(volumeID)
+func (d *Driver) useDataPlaneAPI(ctx context.Context, volumeID, accountName string) string {
+	v, useDataPlaneAPI := d.dataPlaneAPIVolMap.Load(volumeID)
 	if useDataPlaneAPI {
-		return true
+		return v.(string)
 	}
 
 	cache, err := d.dataPlaneAPIAccountCache.Get(ctx, accountName, azcache.CacheReadTypeDefault)
 	if err != nil {
 		klog.Errorf("get(%s) from dataPlaneAPIAccountCache failed with error: %v", accountName, err)
+		return ""
 	}
 	if cache != nil {
-		return true
+		return cache.(string)
 	}
-	return false
+	return ""
 }
 
 func (d *Driver) SetAzureCredentials(ctx context.Context, accountName, accountKey, secretName, secretNamespace string) (string, error) {

pkg/azurefile/azurefile_dataplane_client.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net/http"
 
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/service"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/share"
@@ -47,18 +48,15 @@ type azureFileDataplaneClient struct {
 }
 
 func newAzureFileClient(accountName, accountKey, storageEndpointSuffix string) (azureFileClient, error) {
-	if storageEndpointSuffix == "" {
-		storageEndpointSuffix = defaultStorageEndPointSuffix
-	}
 	keyCred, err := service.NewSharedKeyCredential(accountName, accountKey)
 	if err != nil {
 		return nil, fmt.Errorf("error creating azure client: %v", err)
 	}
-	storageEndpoint := fmt.Sprintf("https://%s.file."+storageEndpointSuffix, accountName)
+	serviceURL := getFileServiceURL(accountName, storageEndpointSuffix)
 	clientOps := utils.GetDefaultAzCoreClientOption()
 	clientOps.Retry.StatusCodes = defaultValidStatusCodes
 
-	fileClient, err := service.NewClientWithSharedKeyCredential(storageEndpoint, keyCred, &service.ClientOptions{
+	fileClient, err := service.NewClientWithSharedKeyCredential(serviceURL, keyCred, &service.ClientOptions{
 		ClientOptions: clientOps,
 	})
 	if err != nil {
@@ -71,6 +69,21 @@ func newAzureFileClient(accountName, accountKey, storageEndpointSuffix string) (
 	}, nil
 }
 
+func newAzureFileClientWithOAuth(cred azcore.TokenCredential, accountName, storageEndpointSuffix string) (azureFileClient, error) {
+	serviceURL := getFileServiceURL(accountName, storageEndpointSuffix)
+	fileClient, err := service.NewClient(serviceURL, cred, &service.ClientOptions{
+		ClientOptions: utils.GetDefaultAzCoreClientOption(),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error creating azure client with oauth: %v", err)
+	}
+	klog.V(2).Infof("created azure file client with oauth, accountName: %s", accountName)
+	return &azureFileDataplaneClient{
+		accountName: accountName,
+		Client:      fileClient,
+	}, nil
+}
+
 func (f *azureFileDataplaneClient) CreateFileShare(ctx context.Context, shareOptions *ShareOptions) error {
 	if shareOptions == nil {
 		return fmt.Errorf("shareOptions of account(%s) is nil", f.accountName)

pkg/azurefile/azurefile_test.go

@@ -1015,7 +1015,7 @@ func TestGetFileShareQuota(t *testing.T) {
 			ResourceGroup:  resourceGroupName,
 			Name:           accountName,
 			SubscriptionID: "subsID",
-		}, fileShareName, test.secrets)
+		}, fileShareName, test.secrets, "")
 		if !reflect.DeepEqual(err, test.expectedError) {
 			t.Errorf("test name: %s, Unexpected error: %v, expected error: %v", test.desc, err, test.expectedError)
 		}