Support EBS gp2 to gp3 migration on sync for below 1tb volumes (#1242)

* initial commit for gp3 migration.

* Default volume migration done.

* Added Gomock and one test case with mock.

* Dep update.

* more changes for code gen.

* push fake package.

* Rename var.

* Changes to Makefile and return value.

* Macke mocks phony due to overlap in foldername.

* Learning as one goes. Initialize map.

* Wrong toggle.

* Expect modify call.

* Fix mapping of ids in test.

* Fix volume id.

* volume ids.

* Fixing test setup. Late night...

* create all pvs.

* Fix test case config.

* store volumes and compare.

* More logs.

* Logging of migration action.

* Ensure to log errors.

* Log warning if modify failed, e.g. due to ebs volume state.

* Add more output.

* Skip local e2e tests.

* Reflect k8s volume id in test data. Extract aws volume id from k8s value.

* Finalizing ebs migration.

* More logs. describe fails.

* Fix non existing fields in gp2 discovery.

* Remove nothing to do flag for migration.

* Final commit for migration.

* add new options to all places

Co-authored-by: Felix Kunde <felix-kunde@gmx.de>

pkg/apis/acid.zalan.do/v1/crds.go

@@ -1169,6 +1169,15 @@ var OperatorConfigCRDResourceValidation = apiextv1.CustomResourceValidation{
 							"aws_region": {
 								Type: "string",
 							},
+							"enable_ebs_gp3_migration": {
+								Type: "boolean",
+							},
+							"enable_ebs_gp3_migration_max_size": {
+								Type: "integer",
+							},
+							"gcp_credentials": {
+								Type: "string",
+							},
 							"kube_iam_role": {
 								Type: "string",
 							},

pkg/apis/acid.zalan.do/v1/operator_configuration_type.go

@@ -117,14 +117,16 @@ type LoadBalancerConfiguration struct {
 // AWSGCPConfiguration defines the configuration for AWS
 // TODO complete Google Cloud Platform (GCP) configuration
 type AWSGCPConfiguration struct {
-	WALES3Bucket              string `json:"wal_s3_bucket,omitempty"`
-	AWSRegion                 string `json:"aws_region,omitempty"`
-	WALGSBucket               string `json:"wal_gs_bucket,omitempty"`
-	GCPCredentials            string `json:"gcp_credentials,omitempty"`
-	LogS3Bucket               string `json:"log_s3_bucket,omitempty"`
-	KubeIAMRole               string `json:"kube_iam_role,omitempty"`
-	AdditionalSecretMount     string `json:"additional_secret_mount,omitempty"`
-	AdditionalSecretMountPath string `json:"additional_secret_mount_path" default:"/meta/credentials"`
+	WALES3Bucket                 string `json:"wal_s3_bucket,omitempty"`
+	AWSRegion                    string `json:"aws_region,omitempty"`
+	WALGSBucket                  string `json:"wal_gs_bucket,omitempty"`
+	GCPCredentials               string `json:"gcp_credentials,omitempty"`
+	LogS3Bucket                  string `json:"log_s3_bucket,omitempty"`
+	KubeIAMRole                  string `json:"kube_iam_role,omitempty"`
+	AdditionalSecretMount        string `json:"additional_secret_mount,omitempty"`
+	AdditionalSecretMountPath    string `json:"additional_secret_mount_path" default:"/meta/credentials"`
+	EnableEBSGp3Migration        bool   `json:"enable_ebs_gp3_migration" default:"false"`
+	EnableEBSGp3MigrationMaxSize int64  `json:"enable_ebs_gp3_migration_max_size" default:"1000"`
 }
 
 // OperatorDebugConfiguration defines options for the debug mode

pkg/apis/acid.zalan.do/v1/postgresql_type.go

@@ -115,8 +115,11 @@ type Volume struct {
 	Size         string `json:"size"`
 	StorageClass string `json:"storageClass,omitempty"`
 	SubPath      string `json:"subPath,omitempty"`
+	Iops         *int64 `json:"iops,omitempty"`
+	Throughput   *int64 `json:"throughput,omitempty"`
 }
 
+// AdditionalVolume specs additional optional volumes for statefulset
 type AdditionalVolume struct {
 	Name             string          `json:"name"`
 	MountPath        string          `json:"mountPath"`
@@ -156,11 +159,12 @@ type Patroni struct {
 	SynchronousModeStrict bool                         `json:"synchronous_mode_strict,omitempty"`
 }
 
-//StandbyCluster
+// StandbyDescription contains s3 wal path
 type StandbyDescription struct {
 	S3WalPath string `json:"s3_wal_path,omitempty"`
 }
 
+// TLSDescription specs TLS properties
 type TLSDescription struct {
 	SecretName      string `json:"secretName,omitempty"`
 	CertificateFile string `json:"certificateFile,omitempty"`
@@ -198,7 +202,7 @@ type PostgresStatus struct {
 	PostgresClusterStatus string `json:"PostgresClusterStatus"`
 }
 
-// Options for connection pooler
+// ConnectionPooler Options for connection pooler
 //
 // TODO: prepared snippets of configuration, one can choose via type, e.g.
 // pgbouncer-large (with higher resources) or odyssey-small (with smaller

pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go

@@ -524,7 +524,7 @@ func (in *PostgresPodResourcesDefaults) DeepCopy() *PostgresPodResourcesDefaults
 func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
 	*out = *in
 	in.PostgresqlParam.DeepCopyInto(&out.PostgresqlParam)
-	out.Volume = in.Volume
+	in.Volume.DeepCopyInto(&out.Volume)
 	in.Patroni.DeepCopyInto(&out.Patroni)
 	out.Resources = in.Resources
 	if in.EnableConnectionPooler != nil {
@@ -623,6 +623,11 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
 			(*out)[key] = *val.DeepCopy()
 		}
 	}
+	if in.SchedulerName != nil {
+		in, out := &in.SchedulerName, &out.SchedulerName
+		*out = new(string)
+		**out = **in
+	}
 	if in.Tolerations != nil {
 		in, out := &in.Tolerations, &out.Tolerations
 		*out = make([]corev1.Toleration, len(*in))
@@ -687,11 +692,6 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
-	if in.SchedulerName != nil {
-		in, out := &in.SchedulerName, &out.SchedulerName
-		*out = new(string)
-		**out = **in
-	}
 	return
 }
 
@@ -1160,6 +1160,16 @@ func (in UserFlags) DeepCopy() UserFlags {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Volume) DeepCopyInto(out *Volume) {
 	*out = *in
+	if in.Iops != nil {
+		in, out := &in.Iops, &out.Iops
+		*out = new(int64)
+		**out = **in
+	}
+	if in.Throughput != nil {
+		in, out := &in.Throughput, &out.Throughput
+		*out = new(int64)
+		**out = **in
+	}
 	return
 }
 

pkg/cluster/cluster.go

@@ -25,6 +25,7 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util/patroni"
 	"github.com/zalando/postgres-operator/pkg/util/teams"
 	"github.com/zalando/postgres-operator/pkg/util/users"
+	"github.com/zalando/postgres-operator/pkg/util/volumes"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	policybeta1 "k8s.io/api/policy/v1beta1"
@@ -89,7 +90,10 @@ type Cluster struct {
 	processMu        sync.RWMutex // protects the current operation for reporting, no need to hold the master mutex
 	specMu           sync.RWMutex // protects the spec for reporting, no need to hold the master mutex
 	ConnectionPooler map[PostgresRole]*ConnectionPoolerObjects
+	EBSVolumes       map[string]volumes.VolumeProperties
+	VolumeResizer    volumes.VolumeResizer
 }
+
 type compareStatefulsetResult struct {
 	match         bool
 	replace       bool
@@ -134,6 +138,13 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgres
 	cluster.oauthTokenGetter = newSecretOauthTokenGetter(&kubeClient, cfg.OpConfig.OAuthTokenSecretName)
 	cluster.patroni = patroni.New(cluster.logger)
 	cluster.eventRecorder = eventRecorder
+
+	cluster.EBSVolumes = make(map[string]volumes.VolumeProperties)
+	if cfg.OpConfig.StorageResizeMode != "pvc" || cfg.OpConfig.EnableEBSGp3Migration {
+		cluster.VolumeResizer = &volumes.EBSVolumeResizer{AWSRegion: cfg.OpConfig.AWSRegion}
+
+	}
+
 	return cluster
 }
 

pkg/cluster/sync.go

@@ -11,7 +11,6 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
-	"github.com/zalando/postgres-operator/pkg/util/volumes"
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	v1 "k8s.io/api/core/v1"
@@ -55,7 +54,23 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 	}
 
 	c.logger.Debugf("syncing volumes using %q storage resize mode", c.OpConfig.StorageResizeMode)
-	if c.OpConfig.StorageResizeMode == "pvc" {
+
+	if c.OpConfig.EnableEBSGp3Migration {
+		err = c.executeEBSMigration()
+		if nil != err {
+			return err
+		}
+	}
+
+	if c.OpConfig.StorageResizeMode == "mixed" {
+		// mixed op uses AWS API to adjust size,throughput,iops and calls pvc chance for file system resize
+
+		// 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
@@ -599,7 +614,8 @@ func (c *Cluster) syncVolumes() error {
 	if !act {
 		return nil
 	}
-	if err := c.resizeVolumes(c.Spec.Volume, []volumes.VolumeResizer{&volumes.EBSVolumeResizer{AWSRegion: c.OpConfig.AWSRegion}}); err != nil {
+
+	if err := c.resizeVolumes(); err != nil {
 		return fmt.Errorf("could not sync volumes: %v", err)
 	}
 

pkg/cluster/volumes.go

@@ -15,7 +15,6 @@ import (
 	"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) listPersistentVolumeClaims() ([]v1.PersistentVolumeClaim, error) {
@@ -119,19 +118,26 @@ func (c *Cluster) listPersistentVolumes() ([]*v1.PersistentVolume, error) {
 }
 
 // resizeVolumes resize persistent volumes compatible with the given resizer interface
-func (c *Cluster) resizeVolumes(newVolume acidv1.Volume, resizers []volumes.VolumeResizer) error {
-	c.setProcessName("resizing volumes")
+func (c *Cluster) resizeVolumes() error {
+	if c.VolumeResizer == nil {
+		return fmt.Errorf("no volume resizer set for EBS volume handling")
+	}
 
+	c.setProcessName("resizing EBS volumes")
+
+	resizer := c.VolumeResizer
 	var totalIncompatible int
 
-	newQuantity, err := resource.ParseQuantity(newVolume.Size)
+	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(newVolume)
+
+	pvs, newSize, err := c.listVolumesWithManifestSize(c.Spec.Volume)
 	if err != nil {
 		return fmt.Errorf("could not list persistent volumes: %v", err)
 	}
+
 	for _, pv := range pvs {
 		volumeSize := quantityToGigabyte(pv.Spec.Capacity[v1.ResourceStorage])
 		if volumeSize >= newSize {
@@ -141,43 +147,43 @@ func (c *Cluster) resizeVolumes(newVolume acidv1.Volume, resizers []volumes.Volu
 			continue
 		}
 		compatible := false
-		for _, resizer := range resizers {
-			if !resizer.VolumeBelongsToProvider(pv) {
-				continue
-			}
-			compatible = true
-			if !resizer.IsConnectedToProvider() {
-				err := resizer.ConnectToProvider()
-				if err != nil {
-					return fmt.Errorf("could not connect to the volume provider: %v", err)
-				}
-				defer func() {
-					if err := resizer.DisconnectFromProvider(); err != nil {
-						c.logger.Errorf("%v", err)
-					}
-				}()
-			}
-			awsVolumeID, err := resizer.GetProviderVolumeID(pv)
-			if err != nil {
-				return err
-			}
-			c.logger.Debugf("updating persistent volume %q to %d", pv.Name, newSize)
-			if err := resizer.ResizeVolume(awsVolumeID, newSize); err != nil {
-				return fmt.Errorf("could not resize EBS volume %q: %v", awsVolumeID, err)
-			}
-			c.logger.Debugf("resizing the filesystem on the volume %q", pv.Name)
-			podName := getPodNameFromPersistentVolume(pv)
-			if err := c.resizePostgresFilesystem(podName, []filesystems.FilesystemResizer{&filesystems.Ext234Resize{}}); err != nil {
-				return fmt.Errorf("could not resize the filesystem on pod %q: %v", podName, err)
-			}
-			c.logger.Debugf("filesystem resize successful on volume %q", pv.Name)
-			pv.Spec.Capacity[v1.ResourceStorage] = newQuantity
-			c.logger.Debugf("updating persistent volume definition for volume %q", pv.Name)
-			if _, err := c.KubeClient.PersistentVolumes().Update(context.TODO(), pv, metav1.UpdateOptions{}); err != nil {
-				return fmt.Errorf("could not update persistent volume: %q", err)
-			}
-			c.logger.Debugf("successfully updated persistent volume %q", pv.Name)
+
+		if !resizer.VolumeBelongsToProvider(pv) {
+			continue
 		}
+		compatible = true
+		if !resizer.IsConnectedToProvider() {
+			err := resizer.ConnectToProvider()
+			if err != nil {
+				return fmt.Errorf("could not connect to the volume provider: %v", err)
+			}
+			defer func() {
+				if err := resizer.DisconnectFromProvider(); err != nil {
+					c.logger.Errorf("%v", err)
+				}
+			}()
+		}
+		awsVolumeID, err := resizer.GetProviderVolumeID(pv)
+		if err != nil {
+			return err
+		}
+		c.logger.Debugf("updating persistent volume %q to %d", pv.Name, newSize)
+		if err := resizer.ResizeVolume(awsVolumeID, newSize); err != nil {
+			return fmt.Errorf("could not resize EBS volume %q: %v", awsVolumeID, err)
+		}
+		c.logger.Debugf("resizing the filesystem on the volume %q", pv.Name)
+		podName := getPodNameFromPersistentVolume(pv)
+		if err := c.resizePostgresFilesystem(podName, []filesystems.FilesystemResizer{&filesystems.Ext234Resize{}}); err != nil {
+			return fmt.Errorf("could not resize the filesystem on pod %q: %v", podName, err)
+		}
+		c.logger.Debugf("filesystem resize successful on volume %q", pv.Name)
+		pv.Spec.Capacity[v1.ResourceStorage] = newQuantity
+		c.logger.Debugf("updating persistent volume definition for volume %q", pv.Name)
+		if _, err := c.KubeClient.PersistentVolumes().Update(context.TODO(), pv, metav1.UpdateOptions{}); err != nil {
+			return fmt.Errorf("could not update persistent volume: %q", err)
+		}
+		c.logger.Debugf("successfully updated persistent volume %q", pv.Name)
+
 		if !compatible {
 			c.logger.Warningf("volume %q is incompatible with all available resizing providers, consider switching storage_resize_mode to pvc or off", pv.Name)
 			totalIncompatible++
@@ -245,3 +251,61 @@ func getPodNameFromPersistentVolume(pv *v1.PersistentVolume) *spec.NamespacedNam
 func quantityToGigabyte(q resource.Quantity) int64 {
 	return q.ScaledValue(0) / (1 * constants.Gigabyte)
 }
+
+func (c *Cluster) executeEBSMigration() error {
+	if !c.OpConfig.EnableEBSGp3Migration {
+		return nil
+	}
+	c.logger.Infof("starting EBS gp2 to gp3 migration")
+
+	pvs, _, err := c.listVolumesWithManifestSize(c.Spec.Volume)
+	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)
+	}
+
+	if len(volumeIds) == len(c.EBSVolumes) {
+		hasGp2 := false
+		for _, v := range c.EBSVolumes {
+			if v.VolumeType == "gp2" {
+				hasGp2 = true
+			}
+		}
+
+		if !hasGp2 {
+			c.logger.Infof("no EBS gp2 volumes left to migrate")
+			return nil
+		}
+	}
+
+	awsVolumes, err := c.VolumeResizer.DescribeVolumes(volumeIds)
+	if nil != err {
+		return err
+	}
+
+	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)
+			if nil != err {
+				c.logger.Warningf("modifying volume %s failed: %v", volume.VolumeID, err)
+			}
+		} else {
+			c.logger.Debugf("skipping EBS volume %s to type gp3 migration (%d)", volume.VolumeID, volume.Size)
+		}
+		c.EBSVolumes[volume.VolumeID] = volume
+	}
+
+	return nil
+}

pkg/cluster/volumes_test.go

@@ -1,6 +1,7 @@
 package cluster
 
 import (
+	"fmt"
 	"testing"
 
 	"context"
@@ -10,11 +11,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 
+	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"github.com/zalando/postgres-operator/mocks"
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/util/config"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
+	"github.com/zalando/postgres-operator/pkg/util/volumes"
 	"k8s.io/client-go/kubernetes/fake"
 )
 
@@ -23,6 +27,8 @@ func NewFakeKubernetesClient() (k8sutil.KubernetesClient, *fake.Clientset) {
 
 	return k8sutil.KubernetesClient{
 		PersistentVolumeClaimsGetter: clientSet.CoreV1(),
+		PersistentVolumesGetter:      clientSet.CoreV1(),
+		PodsGetter:                   clientSet.CoreV1(),
 	}, clientSet
 }
 
@@ -33,6 +39,9 @@ func TestResizeVolumeClaim(t *testing.T) {
 	namespace := "default"
 	newVolumeSize := "2Gi"
 
+	storage1Gi, err := resource.ParseQuantity("1Gi")
+	assert.NoError(t, err)
+
 	// new cluster with pvc storage resize mode and configured labels
 	var cluster = New(
 		Config{
@@ -51,55 +60,9 @@ func TestResizeVolumeClaim(t *testing.T) {
 	filterLabels := cluster.labelsSet(false)
 
 	// define and create PVCs for 1Gi volumes
-	storage1Gi, err := resource.ParseQuantity("1Gi")
-	assert.NoError(t, err)
-
-	pvcList := &v1.PersistentVolumeClaimList{
-		Items: []v1.PersistentVolumeClaim{
-			{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      constants.DataVolumeName + "-" + clusterName + "-0",
-					Namespace: namespace,
-					Labels:    filterLabels,
-				},
-				Spec: v1.PersistentVolumeClaimSpec{
-					Resources: v1.ResourceRequirements{
-						Requests: v1.ResourceList{
-							v1.ResourceStorage: storage1Gi,
-						},
-					},
-				},
-			},
-			{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      constants.DataVolumeName + "-" + clusterName + "-1",
-					Namespace: namespace,
-					Labels:    filterLabels,
-				},
-				Spec: v1.PersistentVolumeClaimSpec{
-					Resources: v1.ResourceRequirements{
-						Requests: v1.ResourceList{
-							v1.ResourceStorage: storage1Gi,
-						},
-					},
-				},
-			},
-			{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      constants.DataVolumeName + "-" + clusterName + "-2-0",
-					Namespace: namespace,
-					Labels:    labels.Set{},
-				},
-				Spec: v1.PersistentVolumeClaimSpec{
-					Resources: v1.ResourceRequirements{
-						Requests: v1.ResourceList{
-							v1.ResourceStorage: storage1Gi,
-						},
-					},
-				},
-			},
-		},
-	}
+	pvcList := CreatePVCs(namespace, clusterName, filterLabels, 2, "1Gi")
+	// add another PVC with different cluster name
+	pvcList.Items = append(pvcList.Items, CreatePVCs(namespace, clusterName+"-2", labels.Set{}, 1, "1Gi").Items[0])
 
 	for _, pvc := range pvcList.Items {
 		cluster.KubeClient.PersistentVolumeClaims(namespace).Create(context.TODO(), &pvc, metav1.CreateOptions{})
@@ -169,3 +132,132 @@ func TestQuantityToGigabyte(t *testing.T) {
 		}
 	}
 }
+
+func CreatePVCs(namespace string, clusterName string, labels labels.Set, n int, size string) v1.PersistentVolumeClaimList {
+	// define and create PVCs for 1Gi volumes
+	storage1Gi, _ := resource.ParseQuantity(size)
+	pvcList := v1.PersistentVolumeClaimList{
+		Items: []v1.PersistentVolumeClaim{},
+	}
+
+	for i := 0; i < n; i++ {
+		pvc := v1.PersistentVolumeClaim{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("%s-%s-%d", constants.DataVolumeName, clusterName, i),
+				Namespace: namespace,
+				Labels:    labels,
+			},
+			Spec: v1.PersistentVolumeClaimSpec{
+				Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceStorage: storage1Gi,
+					},
+				},
+				VolumeName: fmt.Sprintf("persistent-volume-%d", i),
+			},
+		}
+		pvcList.Items = append(pvcList.Items, pvc)
+	}
+
+	return pvcList
+}
+
+func TestMigrateEBS(t *testing.T) {
+	client, _ := NewFakeKubernetesClient()
+	clusterName := "acid-test-cluster"
+	namespace := "default"
+
+	// new cluster with pvc storage resize mode and configured labels
+	var cluster = New(
+		Config{
+			OpConfig: config.Config{
+				Resources: config.Resources{
+					ClusterLabels:    map[string]string{"application": "spilo"},
+					ClusterNameLabel: "cluster-name",
+				},
+				StorageResizeMode:            "pvc",
+				EnableEBSGp3Migration:        true,
+				EnableEBSGp3MigrationMaxSize: 1000,
+			},
+		}, client, acidv1.Postgresql{}, logger, eventRecorder)
+	cluster.Spec.Volume.Size = "1Gi"
+
+	// set metadata, so that labels will get correct values
+	cluster.Name = clusterName
+	cluster.Namespace = namespace
+	filterLabels := cluster.labelsSet(false)
+
+	pvcList := CreatePVCs(namespace, clusterName, filterLabels, 2, "1Gi")
+
+	ps := v1.PersistentVolumeSpec{}
+	ps.AWSElasticBlockStore = &v1.AWSElasticBlockStoreVolumeSource{}
+	ps.AWSElasticBlockStore.VolumeID = "aws://eu-central-1b/ebs-volume-1"
+
+	ps2 := v1.PersistentVolumeSpec{}
+	ps2.AWSElasticBlockStore = &v1.AWSElasticBlockStoreVolumeSource{}
+	ps2.AWSElasticBlockStore.VolumeID = "aws://eu-central-1b/ebs-volume-2"
+
+	pvList := &v1.PersistentVolumeList{
+		Items: []v1.PersistentVolume{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "persistent-volume-0",
+				},
+				Spec: ps,
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "persistent-volume-1",
+				},
+				Spec: ps2,
+			},
+		},
+	}
+
+	for _, pvc := range pvcList.Items {
+		cluster.KubeClient.PersistentVolumeClaims(namespace).Create(context.TODO(), &pvc, metav1.CreateOptions{})
+	}
+
+	for _, pv := range pvList.Items {
+		cluster.KubeClient.PersistentVolumes().Create(context.TODO(), &pv, metav1.CreateOptions{})
+	}
+
+	pod := v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   clusterName + "-0",
+			Labels: filterLabels,
+		},
+		Spec: v1.PodSpec{},
+	}
+
+	cluster.KubeClient.Pods(namespace).Create(context.TODO(), &pod, metav1.CreateOptions{})
+
+	pod = v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   clusterName + "-1",
+			Labels: filterLabels,
+		},
+		Spec: v1.PodSpec{},
+	}
+
+	cluster.KubeClient.Pods(namespace).Create(context.TODO(), &pod, metav1.CreateOptions{})
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	resizer := mocks.NewMockVolumeResizer(ctrl)
+
+	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-1")).Return("ebs-volume-1", nil)
+	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-2")).Return("ebs-volume-2", nil)
+
+	resizer.EXPECT().DescribeVolumes(gomock.Eq([]string{"ebs-volume-1", "ebs-volume-2"})).Return(
+		[]volumes.VolumeProperties{
+			{VolumeID: "ebs-volume-1", VolumeType: "gp2", Size: 100},
+			{VolumeID: "ebs-volume-2", VolumeType: "gp3", Size: 100}}, nil)
+
+	// expect only gp2 volume to be modified
+	resizer.EXPECT().ModifyVolume(gomock.Eq("ebs-volume-1"), gomock.Eq("gp3"), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
+
+	cluster.VolumeResizer = resizer
+	cluster.executeEBSMigration()
+}

pkg/controller/operator_config.go

@@ -138,6 +138,8 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
 	result.GCPCredentials = fromCRD.AWSGCP.GCPCredentials
 	result.AdditionalSecretMount = fromCRD.AWSGCP.AdditionalSecretMount
 	result.AdditionalSecretMountPath = util.Coalesce(fromCRD.AWSGCP.AdditionalSecretMountPath, "/meta/credentials")
+	result.EnableEBSGp3Migration = fromCRD.AWSGCP.EnableEBSGp3Migration
+	result.EnableEBSGp3MigrationMaxSize = fromCRD.AWSGCP.EnableEBSGp3MigrationMaxSize
 
 	// logical backup config
 	result.LogicalBackupSchedule = util.Coalesce(fromCRD.LogicalBackup.Schedule, "30 00 * * *")

pkg/util/config/config.go

@@ -163,6 +163,8 @@ type Config struct {
 	GCPCredentials                         string            `name:"gcp_credentials"`
 	AdditionalSecretMount                  string            `name:"additional_secret_mount"`
 	AdditionalSecretMountPath              string            `name:"additional_secret_mount_path" default:"/meta/credentials"`
+	EnableEBSGp3Migration                  bool              `name:"enable_ebs_gp3_migration" default:"false"`
+	EnableEBSGp3MigrationMaxSize           int64             `name:"enable_ebs_gp3_migration_max_size" default:"1000"`
 	DebugLogging                           bool              `name:"debug_logging" default:"true"`
 	EnableDBAccess                         bool              `name:"enable_database_access" default:"true"`
 	EnableTeamsAPI                         bool              `name:"enable_teams_api" default:"true"`

pkg/util/volumes/ebs.go

@@ -7,7 +7,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/retryutil"
@@ -20,42 +20,80 @@ type EBSVolumeResizer struct {
 }
 
 // ConnectToProvider connects to AWS.
-func (c *EBSVolumeResizer) ConnectToProvider() error {
-	sess, err := session.NewSession(&aws.Config{Region: aws.String(c.AWSRegion)})
+func (r *EBSVolumeResizer) ConnectToProvider() error {
+	sess, err := session.NewSession(&aws.Config{Region: aws.String(r.AWSRegion)})
 	if err != nil {
 		return fmt.Errorf("could not establish AWS session: %v", err)
 	}
-	c.connection = ec2.New(sess)
+	r.connection = ec2.New(sess)
 	return nil
 }
 
 // IsConnectedToProvider checks if AWS connection is established.
-func (c *EBSVolumeResizer) IsConnectedToProvider() bool {
-	return c.connection != nil
+func (r *EBSVolumeResizer) IsConnectedToProvider() bool {
+	return r.connection != nil
 }
 
 // VolumeBelongsToProvider checks if the given persistent volume is backed by EBS.
-func (c *EBSVolumeResizer) VolumeBelongsToProvider(pv *v1.PersistentVolume) bool {
+func (r *EBSVolumeResizer) VolumeBelongsToProvider(pv *v1.PersistentVolume) bool {
 	return pv.Spec.AWSElasticBlockStore != nil && pv.Annotations[constants.VolumeStorateProvisionerAnnotation] == constants.EBSProvisioner
 }
 
-// GetProviderVolumeID converts aws://eu-central-1b/vol-00f93d4827217c629 to vol-00f93d4827217c629 for EBS volumes
-func (c *EBSVolumeResizer) GetProviderVolumeID(pv *v1.PersistentVolume) (string, error) {
-	volumeID := pv.Spec.AWSElasticBlockStore.VolumeID
-	if volumeID == "" {
-		return "", fmt.Errorf("volume id is empty for volume %q", pv.Name)
-	}
+// ExtractVolumeID extracts volumeID
+func (r *EBSVolumeResizer) ExtractVolumeID(volumeID string) (string, error) {
 	idx := strings.LastIndex(volumeID, constants.EBSVolumeIDStart) + 1
 	if idx == 0 {
-		return "", fmt.Errorf("malfored EBS volume id %q", volumeID)
+		return "", fmt.Errorf("malformed EBS volume id %q", volumeID)
 	}
 	return volumeID[idx:], nil
 }
 
+// GetProviderVolumeID converts aws://eu-central-1b/vol-00f93d4827217c629 to vol-00f93d4827217c629 for EBS volumes
+func (r *EBSVolumeResizer) GetProviderVolumeID(pv *v1.PersistentVolume) (string, error) {
+	volumeID := pv.Spec.AWSElasticBlockStore.VolumeID
+	if volumeID == "" {
+		return "", fmt.Errorf("got empty volume id for volume %v", pv)
+	}
+
+	return r.ExtractVolumeID(volumeID)
+}
+
+// DescribeVolumes ...
+func (r *EBSVolumeResizer) DescribeVolumes(volumeIds []string) ([]VolumeProperties, error) {
+	if !r.IsConnectedToProvider() {
+		err := r.ConnectToProvider()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	volumeOutput, err := r.connection.DescribeVolumes(&ec2.DescribeVolumesInput{VolumeIds: aws.StringSlice((volumeIds))})
+	if err != nil {
+		return nil, err
+	}
+
+	p := []VolumeProperties{}
+	if nil == volumeOutput.Volumes {
+		return p, nil
+	}
+
+	for _, v := range volumeOutput.Volumes {
+		if *v.VolumeType == "gp3" {
+			p = append(p, VolumeProperties{VolumeID: *v.VolumeId, Size: *v.Size, VolumeType: *v.VolumeType, Iops: *v.Iops, Throughput: *v.Throughput})
+		} else if *v.VolumeType == "gp2" {
+			p = append(p, VolumeProperties{VolumeID: *v.VolumeId, Size: *v.Size, VolumeType: *v.VolumeType})
+		} else {
+			return nil, fmt.Errorf("Discovered unexpected volume type %s %s", *v.VolumeId, *v.VolumeType)
+		}
+	}
+
+	return p, nil
+}
+
 // ResizeVolume actually calls AWS API to resize the EBS volume if necessary.
-func (c *EBSVolumeResizer) ResizeVolume(volumeID string, newSize int64) error {
+func (r *EBSVolumeResizer) ResizeVolume(volumeID string, newSize int64) error {
 	/* first check if the volume is already of a requested size */
-	volumeOutput, err := c.connection.DescribeVolumes(&ec2.DescribeVolumesInput{VolumeIds: []*string{&volumeID}})
+	volumeOutput, err := r.connection.DescribeVolumes(&ec2.DescribeVolumesInput{VolumeIds: []*string{&volumeID}})
 	if err != nil {
 		return fmt.Errorf("could not get information about the volume: %v", err)
 	}
@@ -68,7 +106,7 @@ func (c *EBSVolumeResizer) ResizeVolume(volumeID string, newSize int64) error {
 		return nil
 	}
 	input := ec2.ModifyVolumeInput{Size: &newSize, VolumeId: &volumeID}
-	output, err := c.connection.ModifyVolume(&input)
+	output, err := r.connection.ModifyVolume(&input)
 	if err != nil {
 		return fmt.Errorf("could not modify persistent volume: %v", err)
 	}
@@ -87,7 +125,54 @@ func (c *EBSVolumeResizer) ResizeVolume(volumeID string, newSize int64) error {
 	in := ec2.DescribeVolumesModificationsInput{VolumeIds: []*string{&volumeID}}
 	return retryutil.Retry(constants.EBSVolumeResizeWaitInterval, constants.EBSVolumeResizeWaitTimeout,
 		func() (bool, error) {
-			out, err := c.connection.DescribeVolumesModifications(&in)
+			out, err := r.connection.DescribeVolumesModifications(&in)
+			if err != nil {
+				return false, fmt.Errorf("could not describe volume modification: %v", err)
+			}
+			if len(out.VolumesModifications) != 1 {
+				return false, fmt.Errorf("describe volume modification didn't return one record for volume %q", volumeID)
+			}
+			if *out.VolumesModifications[0].VolumeId != volumeID {
+				return false, fmt.Errorf("non-matching volume id when describing modifications: %q is different from %q",
+					*out.VolumesModifications[0].VolumeId, volumeID)
+			}
+			return *out.VolumesModifications[0].ModificationState != constants.EBSVolumeStateModifying, nil
+		})
+}
+
+// ModifyVolume Modify EBS volume
+func (r *EBSVolumeResizer) ModifyVolume(volumeID string, newType string, newSize int64, iops int64, throughput int64) error {
+	/* first check if the volume is already of a requested size */
+	volumeOutput, err := r.connection.DescribeVolumes(&ec2.DescribeVolumesInput{VolumeIds: []*string{&volumeID}})
+	if err != nil {
+		return fmt.Errorf("could not get information about the volume: %v", err)
+	}
+	vol := volumeOutput.Volumes[0]
+	if *vol.VolumeId != volumeID {
+		return fmt.Errorf("describe volume %q returned information about a non-matching volume %q", volumeID, *vol.VolumeId)
+	}
+
+	input := ec2.ModifyVolumeInput{Size: &newSize, VolumeId: &volumeID, VolumeType: &newType, Iops: &iops, Throughput: &throughput}
+	output, err := r.connection.ModifyVolume(&input)
+	if err != nil {
+		return fmt.Errorf("could not modify persistent volume: %v", err)
+	}
+
+	state := *output.VolumeModification.ModificationState
+	if state == constants.EBSVolumeStateFailed {
+		return fmt.Errorf("could not modify persistent volume %q: modification state failed", volumeID)
+	}
+	if state == "" {
+		return fmt.Errorf("received empty modification status")
+	}
+	if state == constants.EBSVolumeStateOptimizing || state == constants.EBSVolumeStateCompleted {
+		return nil
+	}
+	// wait until the volume reaches the "optimizing" or "completed" state
+	in := ec2.DescribeVolumesModificationsInput{VolumeIds: []*string{&volumeID}}
+	return retryutil.Retry(constants.EBSVolumeResizeWaitInterval, constants.EBSVolumeResizeWaitTimeout,
+		func() (bool, error) {
+			out, err := r.connection.DescribeVolumesModifications(&in)
 			if err != nil {
 				return false, fmt.Errorf("could not describe volume modification: %v", err)
 			}
@@ -103,7 +188,7 @@ func (c *EBSVolumeResizer) ResizeVolume(volumeID string, newSize int64) error {
 }
 
 // DisconnectFromProvider closes connection to the EC2 instance
-func (c *EBSVolumeResizer) DisconnectFromProvider() error {
-	c.connection = nil
+func (r *EBSVolumeResizer) DisconnectFromProvider() error {
+	r.connection = nil
 	return nil
 }

pkg/util/volumes/volumes.go

@@ -1,8 +1,17 @@
 package volumes
 
-import (
-	"k8s.io/api/core/v1"
-)
+//go:generate mockgen -package mocks -destination=$PWD/mocks/$GOFILE -source=$GOFILE -build_flags=-mod=vendor
+
+import v1 "k8s.io/api/core/v1"
+
+// VolumeProperties ...
+type VolumeProperties struct {
+	VolumeID   string
+	VolumeType string
+	Size       int64
+	Iops       int64
+	Throughput int64
+}
 
 // VolumeResizer defines the set of methods used to implememnt provider-specific resizing of persistent volumes.
 type VolumeResizer interface {
@@ -10,6 +19,9 @@ type VolumeResizer interface {
 	IsConnectedToProvider() bool
 	VolumeBelongsToProvider(pv *v1.PersistentVolume) bool
 	GetProviderVolumeID(pv *v1.PersistentVolume) (string, error)
+	ExtractVolumeID(volumeID string) (string, error)
 	ResizeVolume(providerVolumeID string, newSize int64) error
+	ModifyVolume(providerVolumeID string, newType string, newSize int64, iops int64, throughput int64) error
 	DisconnectFromProvider() error
+	DescribeVolumes(providerVolumesID []string) ([]VolumeProperties, error)
 }