initial commit for gp3 migration.

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

@@ -115,8 +115,11 @@ type Volume struct {
 	Size         string `json:"size"`
 	StorageClass string `json:"storageClass"`
 	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"`
 }
 
-//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/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"
@@ -64,6 +65,14 @@ type kubeResources struct {
 	//PVCs are treated separately
 }
 
+type EBSVolume struct {
+	volumeId   string
+	volumeType string
+	size       int64
+	iops       int32
+	throughput int32
+}
+
 // Cluster describes postgresql cluster
 type Cluster struct {
 	kubeResources
@@ -89,7 +98,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]EBSVolume
+	VolumeResizer    volumes.VolumeResizer
 }
+
 type compareStatefulsetResult struct {
 	match         bool
 	replace       bool
@@ -134,6 +146,11 @@ 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
+
+	if cfg.OpConfig.StorageResizeMode != "pvc" {
+		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,15 @@ 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.StorageResizeMode == "mixed" {
+
+		// 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 +606,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++

pkg/util/config/config.go

@@ -198,6 +198,8 @@ type Config struct {
 	SetMemoryRequestToLimit                bool              `name:"set_memory_request_to_limit" default:"false"`
 	EnableLazySpiloUpgrade                 bool              `name:"enable_lazy_spilo_upgrade" default:"false"`
 	EnablePgVersionEnvVar                  bool              `name:"enable_pgversion_env_var" default:"false"`
+	EnableEBSGp3Migration                  bool              `name:"enable_ebs_gp3_migration" default:"false"`
+	EnableEBSGp3MaxSize                    int32             `name:"enable_ebs_gp3_max_size" default:"1000"`
 }
 
 // MustMarshal marshals the config or panics

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"
@@ -102,6 +102,57 @@ func (c *EBSVolumeResizer) ResizeVolume(volumeID string, newSize int64) error {
 		})
 }
 
+// ModifyVolume Modify EBS volume
+func (c *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 := c.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)
+	}
+	if *vol.Size == newSize {
+		// nothing to do
+		return nil
+	}
+
+	input := ec2.ModifyVolumeInput{Size: &newSize, VolumeId: &volumeID, VolumeType: &newType, Iops: &iops, Throughput: &throughput}
+	output, err := c.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 := c.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
+		})
+}
+
 // DisconnectFromProvider closes connection to the EC2 instance
 func (c *EBSVolumeResizer) DisconnectFromProvider() error {
 	c.connection = nil

pkg/util/volumes/volumes.go

@@ -1,8 +1,6 @@
 package volumes
 
-import (
-	"k8s.io/api/core/v1"
-)
+import v1 "k8s.io/api/core/v1"
 
 // VolumeResizer defines the set of methods used to implememnt provider-specific resizing of persistent volumes.
 type VolumeResizer interface {
@@ -11,5 +9,6 @@ type VolumeResizer interface {
 	VolumeBelongsToProvider(pv *v1.PersistentVolume) bool
 	GetProviderVolumeID(pv *v1.PersistentVolume) (string, error)
 	ResizeVolume(providerVolumeID string, newSize int64) error
+	ModifyVolume(providerVolumeID string, newType string, newSize int64, iops int64, throughput int64) error
 	DisconnectFromProvider() error
 }