Add Disk Auto Grow for pgBackrest repo host volumes (#4247)

* Add Disk Auto Grow for pgBackrest repo host volumes

This update adds the ability to automatically grow repo host PVCs. The
AutoGrowVolumes feature gate must be enabled and the relevant volumes
must include a Limit value.

Once enabled, this feature tracks the current disk utilization and, when
utilization reaches 75%, the disk request is updated to 150% of the
observed value. At this point and beyond, the requested value will be
tracked by CPK.

The volume request can grow up to the configured limit value.
Note: This change now treats limit values as authoritative
regardless of the feature gate setting. However, the implementation
also now allows limits to be updated after being set.

Issue: PGO-1427

Author: tjmoore4

config/crd/bases/postgres-operator.crunchydata.com_postgresclusters.yaml

@@ -18759,6 +18759,9 @@ spec:
                           description: Whether or not the pgBackRest repository PersistentVolumeClaim
                             is bound to a volume
                           type: boolean
+                        desiredRepoVolume:
+                          description: Desired Size of the repo volume
+                          type: string
                         name:
                           description: The name of the pgBackRest repository
                           type: string
@@ -37681,6 +37684,9 @@ spec:
                           description: Whether or not the pgBackRest repository PersistentVolumeClaim
                             is bound to a volume
                           type: boolean
+                        desiredRepoVolume:
+                          description: Desired Size of the repo volume
+                          type: string
                         name:
                           description: The name of the pgBackRest repository
                           type: string

internal/controller/postgrescluster/autogrow.go

@@ -6,7 +6,9 @@ package postgrescluster
 
 import (
 	"context"
+	"strings"
 
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
@@ -19,7 +21,7 @@ import (
 // status. If the value has grown, create an Event.
 func (r *Reconciler) storeDesiredRequest(
 	ctx context.Context, cluster *v1beta1.PostgresCluster,
-	volumeType, instanceSetName, desiredRequest, desiredRequestBackup string,
+	volumeType, host, desiredRequest, desiredRequestBackup string,
 ) string {
 	var current resource.Quantity
 	var previous resource.Quantity
@@ -31,7 +33,7 @@ func (r *Reconciler) storeDesiredRequest(
 		current, err = resource.ParseQuantity(desiredRequest)
 		if err != nil {
 			log.Error(err, "Unable to parse "+volumeType+" volume request from status ("+
-				desiredRequest+") for "+cluster.Name+"/"+instanceSetName)
+				desiredRequest+") for "+cluster.Name+"/"+host)
 			// If there was an error parsing the value, treat as unset (equivalent to zero).
 			desiredRequest = ""
 			current, _ = resource.ParseQuantity("")
@@ -44,7 +46,7 @@ func (r *Reconciler) storeDesiredRequest(
 		previous, err = resource.ParseQuantity(desiredRequestBackup)
 		if err != nil {
 			log.Error(err, "Unable to parse "+volumeType+" volume request from status backup ("+
-				desiredRequestBackup+") for "+cluster.Name+"/"+instanceSetName)
+				desiredRequestBackup+") for "+cluster.Name+"/"+host)
 			// If there was an error parsing the value, treat as unset (equivalent to zero).
 			desiredRequestBackup = ""
 			previous, _ = resource.ParseQuantity("")
@@ -53,12 +55,12 @@ func (r *Reconciler) storeDesiredRequest(
 	}
 
 	// determine if the appropriate volume limit is set
-	limitSet := limitIsSet(cluster, volumeType, instanceSetName)
+	limitSet := limitIsSet(cluster, volumeType, host)
 
 	if limitSet && current.Value() > previous.Value() {
 		r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "VolumeAutoGrow",
 			"%s volume expansion to %v requested for %s/%s.",
-			volumeType, current.String(), cluster.Name, instanceSetName)
+			volumeType, current.String(), cluster.Name, host)
 	}
 
 	// If the desired size was not observed, update with previously stored value.
@@ -77,34 +79,44 @@ func limitIsSet(cluster *v1beta1.PostgresCluster, volumeType, instanceSetName st
 
 	var limitSet bool
 
-	switch volumeType {
+	switch {
 
 	// Cycle through the instance sets to ensure the correct limit is identified.
-	case "pgData":
+	case volumeType == "pgData":
 		for _, specInstance := range cluster.Spec.InstanceSets {
 			if specInstance.Name == instanceSetName {
 				limitSet = !specInstance.DataVolumeClaimSpec.Resources.Limits.Storage().IsZero()
 			}
 		}
+
+	// VolumeType for the repository host volumes should be in the form 'repoN'
+	// where N is 1-4. As above, cycle through any defined repositories and ensure
+	// the correct limit is identified.
+	case strings.HasPrefix(volumeType, "repo"):
+		for _, specRepo := range cluster.Spec.Backups.PGBackRest.Repos {
+			if specRepo.Name == volumeType && specRepo.Volume != nil {
+				limitSet = !specRepo.Volume.VolumeClaimSpec.Resources.Limits.Storage().IsZero()
+			}
+		}
 	}
-	// TODO: Add cases for pgWAL and repo volumes
+	// TODO: Add case for pgWAL
 
 	return limitSet
 
 }
 
-// setVolumeSize compares the potential sizes from the instance spec, status
-// and limit and sets the appropriate current value.
+// setVolumeSize compares the potential sizes from the cluster status, volume request
+// and volume limit and sets the appropriate current value.
 func (r *Reconciler) setVolumeSize(ctx context.Context, cluster *v1beta1.PostgresCluster,
-	pvc *corev1.PersistentVolumeClaim, volumeType, instanceSpecName string) {
+	spec *corev1.PersistentVolumeClaimSpec, volumeType, host string) {
 
 	log := logging.FromContext(ctx)
 
 	// Store the limit for this instance set. This value will not change below.
-	volumeLimitFromSpec := pvc.Spec.Resources.Limits.Storage()
+	volumeLimitFromSpec := spec.Resources.Limits.Storage()
 
 	// This value will capture our desired update.
-	volumeRequestSize := pvc.Spec.Resources.Requests.Storage()
+	volumeRequestSize := spec.Resources.Requests.Storage()
 
 	// A limit of 0 is ignorned, so the volume request is used.
 	if volumeLimitFromSpec.IsZero() {
@@ -116,19 +128,19 @@ func (r *Reconciler) setVolumeSize(ctx context.Context, cluster *v1beta1.Postgre
 	if volumeRequestSize.Value() > volumeLimitFromSpec.Value() {
 		r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "VolumeRequestOverLimit",
 			"%s volume request (%v) for %s/%s is greater than set limit (%v). Limit value will be used.",
-			volumeType, volumeRequestSize, cluster.Name, instanceSpecName, volumeLimitFromSpec)
+			volumeType, volumeRequestSize, cluster.Name, host, volumeLimitFromSpec)
 
-		pvc.Spec.Resources.Requests = corev1.ResourceList{
+		spec.Resources.Requests = corev1.ResourceList{
 			corev1.ResourceStorage: *resource.NewQuantity(volumeLimitFromSpec.Value(), resource.BinarySI),
 		}
 		// Otherwise, if the feature gate is not enabled, do not autogrow.
 	} else if feature.Enabled(ctx, feature.AutoGrowVolumes) {
 
 		// determine the appropriate volume request based on what's set in the status
 		if dpv, err := getDesiredVolumeSize(
-			cluster, volumeType, instanceSpecName, volumeRequestSize,
+			cluster, volumeType, host, volumeRequestSize,
 		); err != nil {
-			log.Error(err, "For "+cluster.Name+"/"+instanceSpecName+
+			log.Error(err, "For "+cluster.Name+"/"+host+
 				": Unable to parse "+volumeType+" volume request: "+dpv)
 		}
 
@@ -140,19 +152,19 @@ func (r *Reconciler) setVolumeSize(ctx context.Context, cluster *v1beta1.Postgre
 
 			r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "VolumeLimitReached",
 				"%s volume(s) for %s/%s are at size limit (%v).", volumeType,
-				cluster.Name, instanceSpecName, volumeLimitFromSpec)
+				cluster.Name, host, volumeLimitFromSpec)
 
 			// If the volume size request is greater than the limit, issue an
 			// additional event warning.
 			if volumeRequestSize.Value() > volumeLimitFromSpec.Value() {
 				r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "DesiredVolumeAboveLimit",
 					"The desired size (%v) for the %s/%s %s volume(s) is greater than the size limit (%v).",
-					volumeRequestSize, cluster.Name, instanceSpecName, volumeType, volumeLimitFromSpec)
+					volumeRequestSize, cluster.Name, host, volumeType, volumeLimitFromSpec)
 			}
 
 			volumeRequestSize = volumeLimitFromSpec
 		}
-		pvc.Spec.Resources.Requests = corev1.ResourceList{
+		spec.Resources.Requests = corev1.ResourceList{
 			corev1.ResourceStorage: *resource.NewQuantity(volumeRequestSize.Value(), resource.BinarySI),
 		}
 	}
@@ -164,8 +176,8 @@ func getDesiredVolumeSize(cluster *v1beta1.PostgresCluster,
 	volumeType, instanceSpecName string,
 	volumeRequestSize *resource.Quantity) (string, error) {
 
-	switch volumeType {
-	case "pgData":
+	switch {
+	case volumeType == "pgData":
 		for i := range cluster.Status.InstanceSets {
 			if instanceSpecName == cluster.Status.InstanceSets[i].Name {
 				for _, dpv := range cluster.Status.InstanceSets[i].DesiredPGDataVolume {
@@ -182,7 +194,30 @@ func getDesiredVolumeSize(cluster *v1beta1.PostgresCluster,
 				}
 			}
 		}
-		// TODO: Add cases for pgWAL and repo volumes (requires relevant status sections)
+
+	// VolumeType for the repository host volumes should be in the form 'repoN'
+	// where N is 1-4. As above, cycle through any defined repositories and ensure
+	// the correct limit is identified.
+	case strings.HasPrefix(volumeType, "repo"):
+		if cluster.Status.PGBackRest == nil {
+			return "", errors.New("PostgresCluster.Status.PGBackRest is nil")
+		}
+		for i := range cluster.Status.PGBackRest.Repos {
+			if volumeType == cluster.Status.PGBackRest.Repos[i].Name {
+				dpv := cluster.Status.PGBackRest.Repos[i].DesiredRepoVolume
+				if dpv != "" {
+					desiredRequest, err := resource.ParseQuantity(dpv)
+					if err == nil {
+						if desiredRequest.Value() > volumeRequestSize.Value() {
+							*volumeRequestSize = desiredRequest
+						}
+					} else {
+						return dpv, err
+					}
+				}
+			}
+		}
 	}
+	// TODO: Add case for pgWAL
 	return "", nil
 }