Merge pull request #1362 from kubernetes-sigs/set-account-quota

feat: add accountQuota parameter

docs/driver-parameters.md

@@ -41,6 +41,7 @@ storageEndpointSuffix | specify Azure storage endpoint suffix | `core.windows.ne
 tags | [tags](https://docs.microsoft.com/en-us/azure/azure-resource-manager/management/tag-resources) would be created in newly created storage account | tag format: 'foo=aaa,bar=bbb' | No | ""
 matchTags | whether matching tags when driver tries to find a suitable storage account | `true`,`false` | No | `false`
 selectRandomMatchingAccount | whether randomly selecting a matching account, by default, the driver would always select the first matching account in alphabetical order | `true`,`false` | No | `false`
+accountQuota | to limit the quota for an account, you can specify a maximum quota in GB (`102400`GB by default). If the account exceeds the specified quota, the driver would skip selecting the account | `` | No | `102400`
 --- | **Following parameters are only for SMB protocol** | --- | --- |
 subscriptionID | specify Azure subscription ID where Azure file share will be created | Azure subscription ID | No | if not empty, `resourceGroup` must be provided
 storeAccountKey | whether store account key to k8s secret <br><br> Note:  <br> `false` means driver would leverage kubelet identity to get account key | `true`,`false` | No | `true`

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/Azure/azure-storage-file-go v0.8.0
 	github.com/Azure/go-autorest/autorest v0.11.29
 	github.com/Azure/go-autorest/autorest/adal v0.9.23 // indirect
-	github.com/Azure/go-autorest/autorest/to v0.4.0 // indirect
+	github.com/Azure/go-autorest/autorest/to v0.4.0
 	github.com/container-storage-interface/spec v1.8.0
 	github.com/gofrs/uuid v4.2.0+incompatible // indirect
 	github.com/golang/mock v1.6.0

pkg/azurefile/azurefile.go

@@ -73,6 +73,7 @@ const (
 	// Minimum size of Azure Premium Files is 100GiB
 	// See https://docs.microsoft.com/en-us/azure/storage/files/storage-files-planning#provisioned-shares
 	defaultAzureFileQuota = 100
+	minimumAccountQuota   = 100 // GB
 
 	// key of snapshot name in metadata
 	snapshotNameKey = "initiator"
@@ -135,6 +136,7 @@ const (
 	enableMultichannelField           = "enablemultichannel"
 	premium                           = "premium"
 	selectRandomMatchingAccountField  = "selectrandommatchingaccount"
+	accountQuotaField                 = "accountquota"
 
 	accountNotProvisioned = "StorageAccountIsNotProvisioned"
 	// this is a workaround fix for 429 throttling issue, will update cloud provider for better fix later
@@ -870,6 +872,21 @@ func (d *Driver) ResizeFileShare(ctx context.Context, subsID, resourceGroup, acc
 	})
 }
 
+// GetTotalAccountQuota returns the total quota in GB of all file shares in the storage account and the number of file shares
+func (d *Driver) GetTotalAccountQuota(ctx context.Context, subsID, resourceGroup, accountName string) (int32, int32, error) {
+	fileshares, err := d.cloud.FileClient.WithSubscriptionID(subsID).ListFileShare(ctx, resourceGroup, accountName, "", "")
+	if err != nil {
+		return -1, -1, err
+	}
+	var totalQuotaGB int32
+	for _, fs := range fileshares {
+		if fs.ShareQuota != nil {
+			totalQuotaGB += *fs.ShareQuota
+		}
+	}
+	return totalQuotaGB, int32(len(fileshares)), nil
+}
+
 // RemoveStorageAccountTag remove tag from storage account
 func (d *Driver) RemoveStorageAccountTag(ctx context.Context, subsID, resourceGroup, account, key string) error {
 	// search in cache first

pkg/azurefile/azurefile_test.go

@@ -29,6 +29,8 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2021-09-01/storage"
 	azure2 "github.com/Azure/go-autorest/autorest/azure"
+	"github.com/Azure/go-autorest/autorest/to"
+	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"k8s.io/client-go/kubernetes/fake"
@@ -63,6 +65,11 @@ func NewFakeDriver() *Driver {
 			},
 		},
 	}
+	driver.AddControllerServiceCapabilities(
+		[]csi.ControllerServiceCapability_RPC_Type{
+			csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+			csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
+		})
 	return driver
 }
 
@@ -78,6 +85,10 @@ func NewFakeDriverCustomOptions(opts DriverOptions) *Driver {
 			},
 		},
 	}
+	driver.AddControllerServiceCapabilities(
+		[]csi.ControllerServiceCapability_RPC_Type{
+			csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+		})
 	return driver
 }
 
@@ -1274,3 +1285,68 @@ func TestGetSubnetResourceID(t *testing.T) {
 		t.Run(tc.name, tc.testFunc)
 	}
 }
+
+func TestGetTotalAccountQuota(t *testing.T) {
+	d := NewFakeDriver()
+	d.cloud = &azure.Cloud{}
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	fileShareItemsWithQuota := []storage.FileShareItem{
+		{
+			FileShareProperties: &storage.FileShareProperties{
+				ShareQuota: to.Int32Ptr(100),
+			},
+		},
+		{
+			FileShareProperties: &storage.FileShareProperties{
+				ShareQuota: to.Int32Ptr(200),
+			},
+		},
+	}
+
+	tests := []struct {
+		name             string
+		subsID           string
+		resourceGroup    string
+		accountName      string
+		fileShareItems   []storage.FileShareItem
+		listFileShareErr error
+		expectedQuota    int32
+		expectedShareNum int32
+		expectedErr      error
+	}{
+		{
+			name:             "GetTotalAccountQuota success",
+			expectedQuota:    0,
+			expectedShareNum: 0,
+		},
+		{
+			name:             "GetTotalAccountQuota success (with 2 file shares)",
+			fileShareItems:   fileShareItemsWithQuota,
+			expectedQuota:    300,
+			expectedShareNum: 2,
+		},
+		{
+			name:             "list file share error",
+			listFileShareErr: fmt.Errorf("list file share error"),
+			expectedQuota:    -1,
+			expectedShareNum: -1,
+			expectedErr:      fmt.Errorf("list file share error"),
+		},
+	}
+
+	for _, test := range tests {
+		mockFileClient := mockfileclient.NewMockInterface(ctrl)
+		d.cloud.FileClient = mockFileClient
+
+		mockFileClient.EXPECT().WithSubscriptionID(gomock.Any()).Return(mockFileClient).AnyTimes()
+		mockFileClient.EXPECT().ListFileShare(context.TODO(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(test.fileShareItems, test.listFileShareErr).AnyTimes()
+
+		quota, fileShareNum, err := d.GetTotalAccountQuota(context.Background(), test.subsID, test.resourceGroup, test.accountName)
+		assert.Equal(t, test.expectedErr, err, test.name)
+		assert.Equal(t, test.expectedQuota, quota, test.name)
+		assert.Equal(t, test.expectedShareNum, fileShareNum, test.name)
+	}
+}

pkg/azurefile/controllerserver.go

@@ -77,6 +77,7 @@ var (
 			Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
 		},
 	}
+	skipMatchingTag = map[string]*string{azure.SkipMatchingTag: pointer.String("")}
 )
 
 // CreateVolume provisions an azure file
@@ -123,6 +124,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	// store account key to k8s secret by default
 	storeAccountKey := true
 
+	var accountQuota int32
 	// Apply ProvisionerParameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
 	for k, v := range parameters {
@@ -245,6 +247,12 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 				return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("invalid %s: %s in storage class", getLatestAccountKeyField, v))
 			}
 			getLatestAccountKey = value
+		case accountQuotaField:
+			value, err := strconv.ParseInt(v, 10, 32)
+			if err != nil || value < minimumAccountQuota {
+				return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("invalid accountQuota %s in storage class, minimum quota: %d", v, minimumAccountQuota))
+			}
+			accountQuota = int32(value)
 		default:
 			return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("invalid parameter %q in storage class", k))
 		}
@@ -452,6 +460,22 @@ 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)
 				}
+				if accountQuota > minimumAccountQuota {
+					totalQuotaGB, fileshareNum, err := d.GetTotalAccountQuota(ctx, subsID, resourceGroup, accountName)
+					if err != nil {
+						return nil, status.Errorf(codes.Internal, "failed to get total quota on account(%s), error: %v", accountName, err)
+					}
+					klog.V(2).Infof("total used quota on account(%s) is %d GB, file share number: %d", accountName, totalQuotaGB, fileshareNum)
+					if totalQuotaGB > accountQuota {
+						klog.Warningf("account(%s) used quota(%d GB) is over %d GB, skip matching current account", accountName, accountQuota, totalQuotaGB)
+						if rerr := d.cloud.AddStorageAccountTags(ctx, subsID, resourceGroup, accountName, skipMatchingTag); rerr != nil {
+							klog.Warningf("AddStorageAccountTags(%v) on account(%s) subsID(%s) rg(%s) failed with error: %v", tags, accountName, subsID, resourceGroup, rerr.Error())
+						}
+						// release volume lock first to prevent deadlock
+						d.volumeLocks.Release(volName)
+						return d.CreateVolume(ctx, req)
+					}
+				}
 				d.accountSearchCache.Set(lockKey, accountName)
 				d.volMap.Store(volName, accountName)
 				if accountKey != "" {
@@ -502,10 +526,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	if err := d.CreateFileShare(ctx, accountOptions, shareOptions, secret); err != nil {
 		if strings.Contains(err.Error(), accountLimitExceedManagementAPI) || strings.Contains(err.Error(), accountLimitExceedDataPlaneAPI) {
 			klog.Warningf("create file share(%s) on account(%s) type(%s) subID(%s) rg(%s) location(%s) size(%d), error: %v, skip matching current account", validFileShareName, accountName, sku, subsID, resourceGroup, location, fileShareSize, err)
-			tags := map[string]*string{
-				azure.SkipMatchingTag: pointer.String(""),
-			}
-			if rerr := d.cloud.AddStorageAccountTags(ctx, subsID, resourceGroup, accountName, tags); rerr != nil {
+			if rerr := d.cloud.AddStorageAccountTags(ctx, subsID, resourceGroup, accountName, skipMatchingTag); rerr != nil {
 				klog.Warningf("AddStorageAccountTags(%v) on account(%s) subsID(%s) rg(%s) failed with error: %v", tags, accountName, subsID, resourceGroup, rerr.Error())
 			}
 			// release volume lock first to prevent deadlock
@@ -514,7 +535,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 			if err := d.accountSearchCache.Delete(lockKey); err != nil {
 				return nil, status.Errorf(codes.Internal, err.Error())
 			}
-			// remove the volName from the volMap to stop it matching the same storage account
+			// remove the volName from the volMap to stop matching the same storage account
 			d.volMap.Delete(volName)
 			return d.CreateVolume(ctx, req)
 		}