Full AWS gp3 support for iops and througput config. (#1261)

Support new AWS EBS volume type `gp3` with `iops` and `throughput` in the manifest. Co-authored-by: Felix Kunde <felix-kunde@gmx.de>
2026-10-08 10:51:37 +02:00 · 2021-01-25 10:07:18 +01:00
parent 4ea0b5f432
commit 4a88f00a3f
15 changed files with 535 additions and 165 deletions
@@ -10,13 +10,215 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

+	"github.com/aws/aws-sdk-go/aws"
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/spec"
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/filesystems"
+	"github.com/zalando/postgres-operator/pkg/util/volumes"
 )

+func (c *Cluster) syncVolumes() error {
+	c.logger.Debugf("syncing volumes using %q storage resize mode", c.OpConfig.StorageResizeMode)
+	var err error
+
+	// check quantity string once, and do not bother with it anymore anywhere else
+	_, err = resource.ParseQuantity(c.Spec.Volume.Size)
+	if err != nil {
+		return fmt.Errorf("could not parse volume size from the manifest: %v", err)
+	}
+
+	if c.OpConfig.StorageResizeMode == "mixed" {
+		// mixed op uses AWS API to adjust size, throughput, iops, and calls pvc change for file system resize
+		// in case of errors we proceed to let K8s do its work, favoring disk space increase of other adjustments
+
+		err = c.populateVolumeMetaData()
+		if err != nil {
+			c.logger.Errorf("populating EBS meta data failed, skipping potential adjustements: %v", err)
+		} else {
+			err = c.syncUnderlyingEBSVolume()
+			if err != nil {
+				c.logger.Errorf("errors occured during EBS volume adjustments: %v", err)
+			}
+		}
+
+		// resize pvc to adjust filesystem size until better K8s support
+		if err = c.syncVolumeClaims(); err != nil {
+			err = fmt.Errorf("could not sync persistent volume claims: %v", err)
+			return err
+		}
+	} else if c.OpConfig.StorageResizeMode == "pvc" {
+		if err = c.syncVolumeClaims(); err != nil {
+			err = fmt.Errorf("could not sync persistent volume claims: %v", err)
+			return err
+		}
+	} else if c.OpConfig.StorageResizeMode == "ebs" {
+		// potentially enlarge volumes before changing the statefulset. By doing that
+		// in this order we make sure the operator is not stuck waiting for a pod that
+		// cannot start because it ran out of disk space.
+		// TODO: handle the case of the cluster that is downsized and enlarged again
+		// (there will be a volume from the old pod for which we can't act before the
+		//  the statefulset modification is concluded)
+		if err = c.syncEbsVolumes(); err != nil {
+			err = fmt.Errorf("could not sync persistent volumes: %v", err)
+			return err
+		}
+	} else {
+		c.logger.Infof("Storage resize is disabled (storage_resize_mode is off). Skipping volume sync.")
+	}
+
+	return nil
+}
+
+func (c *Cluster) syncUnderlyingEBSVolume() error {
+	c.logger.Infof("starting to sync EBS volumes: type, iops, throughput, and size")
+
+	var err error
+
+	targetValue := c.Spec.Volume
+	newSize, err := resource.ParseQuantity(targetValue.Size)
+	targetSize := quantityToGigabyte(newSize)
+
+	awsGp3 := aws.String("gp3")
+	awsIo2 := aws.String("io2")
+
+	errors := []string{}
+
+	for _, volume := range c.EBSVolumes {
+		var modifyIops *int64
+		var modifyThroughput *int64
+		var modifySize *int64
+		var modifyType *string
+
+		if targetValue.Iops != nil {
+			if volume.Iops != *targetValue.Iops {
+				modifyIops = targetValue.Iops
+			}
+		}
+
+		if targetValue.Throughput != nil {
+			if volume.Throughput != *targetValue.Throughput {
+				modifyThroughput = targetValue.Throughput
+			}
+		}
+
+		if targetSize > volume.Size {
+			modifySize = &targetSize
+		}
+
+		if modifyIops != nil || modifyThroughput != nil || modifySize != nil {
+			if modifyIops != nil || modifyThroughput != nil {
+				// we default to gp3 if iops and throughput are configured
+				modifyType = awsGp3
+				if targetValue.VolumeType == "io2" {
+					modifyType = awsIo2
+				}
+			} else if targetValue.VolumeType == "gp3" && volume.VolumeType != "gp3" {
+				modifyType = awsGp3
+			} else {
+				// do not touch type
+				modifyType = nil
+			}
+
+			err = c.VolumeResizer.ModifyVolume(volume.VolumeID, modifyType, modifySize, modifyIops, modifyThroughput)
+			if err != nil {
+				errors = append(errors, fmt.Sprintf("modify volume failed: volume=%s size=%d iops=%d throughput=%d", volume.VolumeID, volume.Size, volume.Iops, volume.Throughput))
+			}
+		}
+	}
+
+	if len(errors) > 0 {
+		for _, s := range errors {
+			c.logger.Warningf(s)
+		}
+		// c.logger.Errorf("failed to modify %d of %d volumes", len(c.EBSVolumes), len(errors))
+	}
+	return nil
+}
+
+func (c *Cluster) populateVolumeMetaData() error {
+	c.logger.Infof("starting reading ebs meta data")
+
+	pvs, err := c.listPersistentVolumes()
+	if err != nil {
+		return fmt.Errorf("could not list persistent volumes: %v", err)
+	}
+	c.logger.Debugf("found %d volumes, size of known volumes %d", len(pvs), len(c.EBSVolumes))
+
+	volumeIds := []string{}
+	var volumeID string
+	for _, pv := range pvs {
+		volumeID, err = c.VolumeResizer.ExtractVolumeID(pv.Spec.AWSElasticBlockStore.VolumeID)
+		if err != nil {
+			continue
+		}
+
+		volumeIds = append(volumeIds, volumeID)
+	}
+
+	currentVolumes, err := c.VolumeResizer.DescribeVolumes(volumeIds)
+	if nil != err {
+		return err
+	}
+
+	if len(currentVolumes) != len(c.EBSVolumes) {
+		c.logger.Debugf("number of ebs volumes (%d) discovered differs from already known volumes (%d)", len(currentVolumes), len(c.EBSVolumes))
+	}
+
+	// reset map, operator is not responsible for dangling ebs volumes
+	c.EBSVolumes = make(map[string]volumes.VolumeProperties)
+	for _, volume := range currentVolumes {
+		c.EBSVolumes[volume.VolumeID] = volume
+	}
+
+	return nil
+}
+
+// syncVolumeClaims reads all persistent volume claims and checks that their size matches the one declared in the statefulset.
+func (c *Cluster) syncVolumeClaims() error {
+	c.setProcessName("syncing volume claims")
+
+	needsResizing, err := c.volumeClaimsNeedResizing(c.Spec.Volume)
+	if err != nil {
+		return fmt.Errorf("could not compare size of the volume claims: %v", err)
+	}
+
+	if !needsResizing {
+		c.logger.Infof("volume claims do not require changes")
+		return nil
+	}
+
+	if err := c.resizeVolumeClaims(c.Spec.Volume); err != nil {
+		return fmt.Errorf("could not sync volume claims: %v", err)
+	}
+
+	c.logger.Infof("volume claims have been synced successfully")
+
+	return nil
+}
+
+// syncVolumes reads all persistent volumes and checks that their size matches the one declared in the statefulset.
+func (c *Cluster) syncEbsVolumes() error {
+	c.setProcessName("syncing EBS and Claims volumes")
+
+	act, err := c.volumesNeedResizing()
+	if err != nil {
+		return fmt.Errorf("could not compare size of the volumes: %v", err)
+	}
+	if !act {
+		return nil
+	}
+
+	if err := c.resizeVolumes(); err != nil {
+		return fmt.Errorf("could not sync volumes: %v", err)
+	}
+
+	c.logger.Infof("volumes have been synced successfully")
+
+	return nil
+}
+
 func (c *Cluster) listPersistentVolumeClaims() ([]v1.PersistentVolumeClaim, error) {
 	ns := c.Namespace
 	listOptions := metav1.ListOptions{
@@ -125,15 +327,16 @@ func (c *Cluster) resizeVolumes() error {

 	c.setProcessName("resizing EBS volumes")

-	resizer := c.VolumeResizer
-	var totalIncompatible int
-
 	newQuantity, err := resource.ParseQuantity(c.Spec.Volume.Size)
 	if err != nil {
 		return fmt.Errorf("could not parse volume size: %v", err)
 	}

-	pvs, newSize, err := c.listVolumesWithManifestSize(c.Spec.Volume)
+	newSize := quantityToGigabyte(newQuantity)
+	resizer := c.VolumeResizer
+	var totalIncompatible int
+
+	pvs, err := c.listPersistentVolumes()
 	if err != nil {
 		return fmt.Errorf("could not list persistent volumes: %v", err)
 	}
@@ -214,33 +417,23 @@ func (c *Cluster) volumeClaimsNeedResizing(newVolume acidv1.Volume) (bool, error
 	return false, nil
 }

-func (c *Cluster) volumesNeedResizing(newVolume acidv1.Volume) (bool, error) {
-	vols, manifestSize, err := c.listVolumesWithManifestSize(newVolume)
+func (c *Cluster) volumesNeedResizing() (bool, error) {
+	newQuantity, _ := resource.ParseQuantity(c.Spec.Volume.Size)
+	newSize := quantityToGigabyte(newQuantity)
+
+	vols, err := c.listPersistentVolumes()
 	if err != nil {
 		return false, err
 	}
 	for _, pv := range vols {
 		currentSize := quantityToGigabyte(pv.Spec.Capacity[v1.ResourceStorage])
-		if currentSize != manifestSize {
+		if currentSize != newSize {
 			return true, nil
 		}
 	}
 	return false, nil
 }

-func (c *Cluster) listVolumesWithManifestSize(newVolume acidv1.Volume) ([]*v1.PersistentVolume, int64, error) {
-	newSize, err := resource.ParseQuantity(newVolume.Size)
-	if err != nil {
-		return nil, 0, fmt.Errorf("could not parse volume size from the manifest: %v", err)
-	}
-	manifestSize := quantityToGigabyte(newSize)
-	vols, err := c.listPersistentVolumes()
-	if err != nil {
-		return nil, 0, fmt.Errorf("could not list persistent volumes: %v", err)
-	}
-	return vols, manifestSize, nil
-}
-
 // getPodNameFromPersistentVolume returns a pod name that it extracts from the volume claim ref.
 func getPodNameFromPersistentVolume(pv *v1.PersistentVolume) *spec.NamespacedName {
 	namespace := pv.Spec.ClaimRef.Namespace
@@ -258,7 +451,7 @@ func (c *Cluster) executeEBSMigration() error {
 	}
 	c.logger.Infof("starting EBS gp2 to gp3 migration")

-	pvs, _, err := c.listVolumesWithManifestSize(c.Spec.Volume)
+	pvs, err := c.listPersistentVolumes()
 	if err != nil {
 		return fmt.Errorf("could not list persistent volumes: %v", err)
 	}
@@ -294,10 +487,13 @@ func (c *Cluster) executeEBSMigration() error {
 		return err
 	}

+	var i3000 int64 = 3000
+	var i125 int64 = 125
+
 	for _, volume := range awsVolumes {
 		if volume.VolumeType == "gp2" && volume.Size < c.OpConfig.EnableEBSGp3MigrationMaxSize {
 			c.logger.Infof("modifying EBS volume %s to type gp3 migration (%d)", volume.VolumeID, volume.Size)
-			err = c.VolumeResizer.ModifyVolume(volume.VolumeID, "gp3", volume.Size, 3000, 125)
+			err = c.VolumeResizer.ModifyVolume(volume.VolumeID, aws.String("gp3"), &volume.Size, &i3000, &i125)
 			if nil != err {
 				c.logger.Warningf("modifying volume %s failed: %v", volume.VolumeID, err)
 			}