Economy volume limits

Fixed limitVolumeSize and added limitVolumePoolSize for ONTAP economy drivers

storage_drivers/ontap/api/ontap_rest.go

@@ -279,10 +279,6 @@ func (c RestClient) SupportsFeature(ctx context.Context, feature Feature) bool {
 		return false
 	}
 
-	if feature == LunGeometrySkip {
-		return !ontapSemVer.AtLeast(versionutils.MustParseSemantic("9.11.0"))
-	}
-
 	if minVersion, ok := featuresByVersion[feature]; ok {
 		return ontapSemVer.AtLeast(minVersion)
 	} else {

storage_drivers/ontap/api/ontap_zapi.go

@@ -201,7 +201,6 @@ var features = map[Feature]*versionutils.Version{
 	NetAppFlexGroupsClone:     versionutils.MustParseSemantic("1.170.0"), // cDOT 9.7.0
 	NetAppFabricPoolFlexVol:   versionutils.MustParseSemantic("1.120.0"), // cDOT 9.2.0
 	NetAppFabricPoolFlexGroup: versionutils.MustParseSemantic("1.150.0"), // cDOT 9.5.0
-	LunGeometrySkip:           versionutils.MustParseSemantic("1.150.0"), // cDOT 9.5.0
 	FabricPoolForSVMDR:        versionutils.MustParseSemantic("1.150.0"), // cDOT 9.5.0
 	QosPolicies:               versionutils.MustParseSemantic("1.180.0"), // cDOT 9.8.0
 	LIFServices:               versionutils.MustParseSemantic("1.160.0"), // cDOT 9.6.0
@@ -216,7 +215,6 @@ var featuresByVersion = map[Feature]*versionutils.Version{
 	NetAppFlexGroupsClone:     versionutils.MustParseSemantic("9.7.0"),
 	NetAppFabricPoolFlexVol:   versionutils.MustParseSemantic("9.2.0"),
 	NetAppFabricPoolFlexGroup: versionutils.MustParseSemantic("9.5.0"),
-	LunGeometrySkip:           versionutils.MustParseSemantic("9.5.0"),
 	FabricPoolForSVMDR:        versionutils.MustParseSemantic("9.5.0"),
 	QosPolicies:               versionutils.MustParseSemantic("9.8.0"),
 	LIFServices:               versionutils.MustParseSemantic("9.6.0"),

storage_drivers/ontap/ontap_common.go

@@ -1228,6 +1228,7 @@ const (
 	DefaultMirroring                 = "false"
 	DefaultLimitAggregateUsage       = ""
 	DefaultLimitVolumeSize           = ""
+	DefaultLimitVolumePoolSize       = ""
 	DefaultTieringPolicy             = ""
 )
 
@@ -1332,14 +1333,6 @@ func PopulateConfigurationDefaults(ctx context.Context, config *drivers.OntapSto
 		config.Mirroring = DefaultMirroring
 	}
 
-	if config.LimitAggregateUsage == "" {
-		config.LimitAggregateUsage = DefaultLimitAggregateUsage
-	}
-
-	if config.LimitVolumeSize == "" {
-		config.LimitVolumeSize = DefaultLimitVolumeSize
-	}
-
 	if config.TieringPolicy == "" {
 		config.TieringPolicy = DefaultTieringPolicy
 	}
@@ -1403,6 +1396,7 @@ func PopulateConfigurationDefaults(ctx context.Context, config *drivers.OntapSto
 		"Mirroring":              config.Mirroring,
 		"LimitAggregateUsage":    config.LimitAggregateUsage,
 		"LimitVolumeSize":        config.LimitVolumeSize,
+		"LimitVolumePoolSize":    config.LimitVolumePoolSize,
 		"Size":                   config.Size,
 		"TieringPolicy":          config.TieringPolicy,
 		"AutoExportPolicy":       config.AutoExportPolicy,
@@ -1527,6 +1521,43 @@ func GetVolumeSize(sizeBytes uint64, poolDefaultSizeBytes string) (uint64, error
 	return sizeBytes, nil
 }
 
+// CheckVolumePoolSizeLimits checks if a volume pool size limit has been set.
+func CheckVolumePoolSizeLimits(
+	ctx context.Context, requestedSize uint64, config *drivers.OntapStorageDriverConfig,
+) (bool, uint64, error) {
+	// If the user specified a limit for volume pool size, parse and enforce it
+	limitVolumePoolSize := config.LimitVolumePoolSize
+	Logc(ctx).WithFields(LogFields{
+		"limitVolumePoolSize": limitVolumePoolSize,
+	}).Debugf("Limits")
+
+	if limitVolumePoolSize == "" {
+		Logc(ctx).Debugf("No limits specified, not limiting volume pool size")
+		return false, 0, nil
+	}
+
+	var volumePoolSizeLimit uint64
+	volumePoolSizeLimitStr, parseErr := utils.ConvertSizeToBytes(limitVolumePoolSize)
+	if parseErr != nil {
+		return false, 0, fmt.Errorf("error parsing limitVolumePoolSize: %v", parseErr)
+	}
+	volumePoolSizeLimit, _ = strconv.ParseUint(volumePoolSizeLimitStr, 10, 64)
+
+	Logc(ctx).WithFields(LogFields{
+		"limitVolumePoolSize": limitVolumePoolSize,
+		"volumePoolSizeLimit": volumePoolSizeLimit,
+		"requestedSizeBytes":  requestedSize,
+	}).Debugf("Comparing pool limits")
+
+	// Check whether pool size limit would prevent *any* Flexvol from working
+	if requestedSize > volumePoolSizeLimit {
+		return true, volumePoolSizeLimit, errors.UnsupportedCapacityRangeError(fmt.Errorf(
+			"requested size: %d > the pool size limit: %d", requestedSize, volumePoolSizeLimit))
+	}
+
+	return true, volumePoolSizeLimit, nil
+}
+
 func GetSnapshotReserve(snapshotPolicy, snapshotReserve string) (int, error) {
 	if snapshotReserve != "" {
 		// snapshotReserve defaults to "", so if it is explicitly set

storage_drivers/ontap/ontap_common_test.go

@@ -4241,6 +4241,48 @@ func TestValidateNASDriver(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestCheckVolumePoolSizeLimits(t *testing.T) {
+	ctx := context.Background()
+	requestedSize := uint64(1073741824) // 1Gi
+
+	// Returns because LimitVolumePoolSize not specified.
+	config := &drivers.OntapStorageDriverConfig{
+		LimitVolumePoolSize: "",
+	}
+	shouldLimit, sizeLimit, err := CheckVolumePoolSizeLimits(ctx, requestedSize, config)
+	assert.False(t, shouldLimit, "expected should limit to be false")
+	assert.Zero(t, sizeLimit, "expected zero size limit")
+	assert.Nil(t, err, "expected nil error")
+
+	// Errors when LimitVolumePoolSize is not empty but invalid and cannot be parsed.
+	config = &drivers.OntapStorageDriverConfig{
+		LimitVolumePoolSize: "Gi",
+	}
+	shouldLimit, sizeLimit, err = CheckVolumePoolSizeLimits(ctx, requestedSize, config)
+	assert.False(t, shouldLimit, "expected should limit to be false")
+	assert.Zero(t, sizeLimit, "expected zero size limit")
+	assert.NotNil(t, err, "expected non-nil error")
+
+	// Errors when the requested volume is larger than LimitVolumePoolSize.
+	requestedSize = uint64(2000000000)
+	config = &drivers.OntapStorageDriverConfig{
+		LimitVolumePoolSize: "1Gi",
+	}
+	shouldLimit, sizeLimit, err = CheckVolumePoolSizeLimits(ctx, requestedSize, config)
+	assert.True(t, shouldLimit, "expected should limit to be true")
+	assert.Equal(t, sizeLimit, uint64(1073741824), "expected size limit of 1Gi")
+	assert.NotNil(t, err, "expected non-nil error")
+
+	requestedSize = uint64(1000000000)
+	config = &drivers.OntapStorageDriverConfig{
+		LimitVolumePoolSize: "1Gi",
+	}
+	shouldLimit, sizeLimit, err = CheckVolumePoolSizeLimits(ctx, requestedSize, config)
+	assert.True(t, shouldLimit, "expected should limit to be true")
+	assert.Equal(t, sizeLimit, uint64(1073741824), "expected size limit of 1Gi")
+	assert.Nil(t, err, "expected nil error")
+}
+
 func TestGetSnapshotReserve(t *testing.T) {
 	// snapshot reserve is not passed
 	snapshotPolicy := "fakePolicy"

storage_drivers/ontap/ontap_nas_qtree.go

@@ -355,6 +355,12 @@ func (d *NASQtreeStorageDriver) Create(
 		return err
 	}
 
+	if _, _, checkVolumeSizeLimitsError := drivers.CheckVolumeSizeLimits(
+		ctx, sizeBytes, d.Config.CommonStorageDriverConfig,
+	); checkVolumeSizeLimitsError != nil {
+		return checkVolumeSizeLimitsError
+	}
+
 	// Ensure qtree name isn't too long
 	if len(name) > maxQtreeNameLength {
 		return fmt.Errorf("volume %s name exceeds the limit of %d characters", name, maxQtreeNameLength)
@@ -429,7 +435,7 @@ func (d *NASQtreeStorageDriver) Create(
 		// Make sure we have a Flexvol for the new qtree
 		flexvol, err := d.ensureFlexvolForQtree(
 			ctx, aggregate, spaceReserve, snapshotPolicy, tieringPolicy, enableSnapshotDir, enableEncryption, sizeBytes,
-			d.Config, snapshotReserve, exportPolicy)
+			&d.Config, snapshotReserve, exportPolicy)
 		if err != nil {
 			errMessage := fmt.Sprintf("ONTAP-NAS-QTREE pool %s/%s; Flexvol location/creation failed %s: %v",
 				storagePool.Name(), aggregate, name, err)
@@ -1034,18 +1040,18 @@ func (d *NASQtreeStorageDriver) Get(ctx context.Context, name string) error {
 // qtree or it creates a new Flexvol with the needed attributes.
 func (d *NASQtreeStorageDriver) ensureFlexvolForQtree(
 	ctx context.Context, aggregate, spaceReserve, snapshotPolicy, tieringPolicy string, enableSnapshotDir bool,
-	enableEncryption *bool, sizeBytes uint64, config drivers.OntapStorageDriverConfig, snapshotReserve,
+	enableEncryption *bool, sizeBytes uint64, config *drivers.OntapStorageDriverConfig, snapshotReserve,
 	exportPolicy string,
 ) (string, error) {
-	shouldLimitVolumeSize, flexvolQuotaSizeLimit, checkVolumeSizeLimitsError := drivers.CheckVolumeSizeLimits(
-		ctx, sizeBytes, config.CommonStorageDriverConfig)
-	if checkVolumeSizeLimitsError != nil {
-		return "", checkVolumeSizeLimitsError
+	shouldLimitFlexvolSize, flexvolSizeLimit, checkFlexvolSizeLimitsError := CheckVolumePoolSizeLimits(
+		ctx, sizeBytes, config)
+	if checkFlexvolSizeLimitsError != nil {
+		return "", checkFlexvolSizeLimitsError
 	}
 
 	// Check if a suitable Flexvol already exists
 	flexvol, err := d.findFlexvolForQtree(ctx, aggregate, spaceReserve, snapshotPolicy, tieringPolicy, snapshotReserve,
-		enableSnapshotDir, enableEncryption, shouldLimitVolumeSize, sizeBytes, flexvolQuotaSizeLimit)
+		enableSnapshotDir, enableEncryption, shouldLimitFlexvolSize, sizeBytes, flexvolSizeLimit)
 	if err != nil {
 		return "", fmt.Errorf("error finding Flexvol for qtree: %v", err)
 	}
@@ -1175,8 +1181,8 @@ func (d *NASQtreeStorageDriver) createFlexvolForQtree(
 // is returned at random.
 func (d *NASQtreeStorageDriver) findFlexvolForQtree(
 	ctx context.Context, aggregate, spaceReserve, snapshotPolicy, tieringPolicy, snapshotReserve string,
-	enableSnapshotDir bool, enableEncryption *bool, shouldLimitFlexvolQuotaSize bool,
-	sizeBytes, flexvolQuotaSizeLimit uint64,
+	enableSnapshotDir bool, enableEncryption *bool, shouldLimitFlexvolSize bool,
+	sizeBytes, flexvolSizeLimit uint64,
 ) (string, error) {
 	snapshotReserveInt, err := GetSnapshotReserve(snapshotPolicy, snapshotReserve)
 	if err != nil {
@@ -1207,15 +1213,14 @@ func (d *NASQtreeStorageDriver) findFlexvolForQtree(
 		volName := volume.Name
 
 		// skip flexvols over the size limit
-		if shouldLimitFlexvolQuotaSize {
+		if shouldLimitFlexvolSize {
 			sizeWithRequest, err := d.getOptimalSizeForFlexvol(ctx, volName, sizeBytes)
 			if err != nil {
 				Logc(ctx).Errorf("Error checking size for existing qtree. %v %v", volName, err)
 				continue
 			}
-			if sizeWithRequest > flexvolQuotaSizeLimit {
-				Logc(ctx).Debugf("Flexvol quota size for %v is over the limit of %v", volName,
-					flexvolQuotaSizeLimit)
+			if sizeWithRequest > flexvolSizeLimit {
+				Logc(ctx).Debugf("Flexvol quota size for %v is over the limit of %v", volName, flexvolSizeLimit)
 				continue
 			}
 		}
@@ -1257,8 +1262,7 @@ func (d *NASQtreeStorageDriver) getOptimalSizeForFlexvol(
 		return 0, err
 	}
 
-	return calculateFlexvolEconomySizeBytes(ctx, flexvol, volAttrs, newQtreeSizeBytes,
-		totalDiskLimitBytes), nil
+	return calculateFlexvolEconomySizeBytes(ctx, flexvol, volAttrs, newQtreeSizeBytes, totalDiskLimitBytes), nil
 }
 
 // addDefaultQuotaForFlexvol adds a default quota rule to a Flexvol so that quotas for
@@ -1298,15 +1302,14 @@ func (d *NASQtreeStorageDriver) setQuotaForQtree(ctx context.Context, qtree, fle
 }
 
 // getQuotaDiskLimitSize returns the disk limit size for the specified quota.
-func (d *NASQtreeStorageDriver) getQuotaDiskLimitSize(ctx context.Context, name, flexvol string) (int64, error) {
+func (d *NASQtreeStorageDriver) getQuotaDiskLimitSize(ctx context.Context, name, flexvol string) (uint64, error) {
 	quota, err := d.API.QuotaGetEntry(ctx, flexvol, name, "tree")
 	if err != nil {
 		return 0, err
 	}
 
-	quotaSize := quota.DiskLimitBytes
-
-	return quotaSize, nil
+	// Report no quota as 0
+	return uint64(max(0, quota.DiskLimitBytes)), nil
 }
 
 // enableQuotas disables quotas on a Flexvol, optionally waiting for the operation to finish.
@@ -2085,29 +2088,49 @@ func (d *NASQtreeStorageDriver) Resize(ctx context.Context, volConfig *storage.V
 		return resizeError
 	}
 
-	volConfig.Size = strconv.FormatInt(quotaSize, 10)
-	if int64(sizeBytes) == quotaSize {
+	volConfig.Size = strconv.FormatUint(quotaSize, 10)
+	if sizeBytes == quotaSize {
 		Logc(ctx).Infof("Requested size and existing volume size are the same for volume %s.", name)
 		return nil
 	}
 
-	if quotaSize >= 0 && int64(sizeBytes) < quotaSize {
+	if sizeBytes < quotaSize {
 		return fmt.Errorf("requested size %d is less than existing volume size %d", sizeBytes, quotaSize)
 	}
-	deltaQuotaSize := int64(sizeBytes) - quotaSize
+	deltaQuotaSize := sizeBytes - quotaSize
 
+	// Enforce aggregate usage limit
 	if aggrLimitsErr := checkAggregateLimitsForFlexvol(
 		ctx, flexvol, sizeBytes, d.Config, d.GetAPI(),
 	); aggrLimitsErr != nil {
 		return aggrLimitsErr
 	}
 
+	// Enforce volume size limit
 	if _, _, checkVolumeSizeLimitsError := drivers.CheckVolumeSizeLimits(
 		ctx, sizeBytes, d.Config.CommonStorageDriverConfig,
 	); checkVolumeSizeLimitsError != nil {
 		return checkVolumeSizeLimitsError
 	}
 
+	// Enforce Flexvol pool size limit
+	shouldLimitFlexvolSize, flexvolSizeLimit, checkFlexvolSizeLimitsError := CheckVolumePoolSizeLimits(
+		ctx, sizeBytes, &d.Config)
+	if checkFlexvolSizeLimitsError != nil {
+		return checkFlexvolSizeLimitsError
+	}
+	if shouldLimitFlexvolSize {
+		flexvolSize, flexvolSizeErr := d.getOptimalSizeForFlexvol(ctx, flexvol, uint64(deltaQuotaSize))
+		if flexvolSizeErr != nil {
+			return flexvolSizeErr
+		}
+		if flexvolSize > flexvolSizeLimit {
+			return errors.UnsupportedCapacityRangeError(fmt.Errorf(
+				"requested size %d would exceed the pool size limit %d", sizeBytes, flexvolSizeLimit))
+		}
+	}
+
+	// Resize the Flexvol
 	err = d.resizeFlexvol(ctx, flexvol, uint64(deltaQuotaSize))
 	if err != nil {
 		Logc(ctx).WithField("error", err).Error("Failed to resize flexvol.")
@@ -2131,20 +2154,20 @@ func (d *NASQtreeStorageDriver) resizeFlexvol(ctx context.Context, flexvol strin
 	flexvolSizeBytes, err := d.getOptimalSizeForFlexvol(ctx, flexvol, sizeBytes)
 	if err != nil {
 		Logc(ctx).Warnf("Could not calculate optimal Flexvol size. %v", err)
-		// Lacking the optimal size, just grow the Flexvol to contain the new qtree
+
+		// Lacking the optimal size, just grow the Flexvol to contain the new qtree if Flexvol size isn't capped
 		size := strconv.FormatUint(sizeBytes, 10)
-		err := d.API.VolumeSetSize(ctx, flexvol, "+"+size)
-		if err != nil {
-			return fmt.Errorf("flexvol resize failed: %v", err)
-		}
-	} else {
-		// Got optimal size, so just set the Flexvol to that value
-		flexvolSizeStr := strconv.FormatUint(flexvolSizeBytes, 10)
-		err := d.API.VolumeSetSize(ctx, flexvol, flexvolSizeStr)
-		if err != nil {
+		if err = d.API.VolumeSetSize(ctx, flexvol, "+"+size); err != nil {
 			return fmt.Errorf("flexvol resize failed: %v", err)
 		}
 	}
+
+	// Got optimal size, so just set the Flexvol to that value
+	flexvolSizeStr := strconv.FormatUint(flexvolSizeBytes, 10)
+	if err = d.API.VolumeSetSize(ctx, flexvol, flexvolSizeStr); err != nil {
+		return fmt.Errorf("flexvol resize failed: %v", err)
+	}
+
 	return nil
 }