merge with master

pkg/util/config/config.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 
 	"github.com/zalando/postgres-operator/pkg/spec"
+	"github.com/zalando/postgres-operator/pkg/util/constants"
+	v1 "k8s.io/api/core/v1"
 )
 
 // CRD describes CustomResourceDefinition specific configuration parameters
@@ -21,43 +23,79 @@ type CRD struct {
 
 // Resources describes kubernetes resource specific configuration parameters
 type Resources struct {
-	ResourceCheckInterval   time.Duration     `name:"resource_check_interval" default:"3s"`
-	ResourceCheckTimeout    time.Duration     `name:"resource_check_timeout" default:"10m"`
-	PodLabelWaitTimeout     time.Duration     `name:"pod_label_wait_timeout" default:"10m"`
-	PodDeletionWaitTimeout  time.Duration     `name:"pod_deletion_wait_timeout" default:"10m"`
-	PodTerminateGracePeriod time.Duration     `name:"pod_terminate_grace_period" default:"5m"`
-	SpiloFSGroup            *int64            `name:"spilo_fsgroup"`
-	PodPriorityClassName    string            `name:"pod_priority_class_name"`
-	ClusterDomain           string            `name:"cluster_domain" default:"cluster.local"`
-	SpiloPrivileged         bool              `name:"spilo_privileged" default:"false"`
-	ClusterLabels           map[string]string `name:"cluster_labels" default:"application:spilo"`
-	InheritedLabels         []string          `name:"inherited_labels" default:""`
-	ClusterNameLabel        string            `name:"cluster_name_label" default:"cluster-name"`
-	PodRoleLabel            string            `name:"pod_role_label" default:"spilo-role"`
-	PodToleration           map[string]string `name:"toleration" default:""`
-	DefaultCPURequest       string            `name:"default_cpu_request" default:"100m"`
-	DefaultMemoryRequest    string            `name:"default_memory_request" default:"100Mi"`
-	DefaultCPULimit         string            `name:"default_cpu_limit" default:"1"`
-	DefaultMemoryLimit      string            `name:"default_memory_limit" default:"500Mi"`
-	MinCPULimit             string            `name:"min_cpu_limit" default:"250m"`
-	MinMemoryLimit          string            `name:"min_memory_limit" default:"250Mi"`
-	PodEnvironmentConfigMap string            `name:"pod_environment_configmap" default:""`
-	NodeReadinessLabel      map[string]string `name:"node_readiness_label" default:""`
-	MaxInstances            int32             `name:"max_instances" default:"-1"`
-	MinInstances            int32             `name:"min_instances" default:"-1"`
-	ShmVolume               *bool             `name:"enable_shm_volume" default:"true"`
+	ResourceCheckInterval   time.Duration       `name:"resource_check_interval" default:"3s"`
+	ResourceCheckTimeout    time.Duration       `name:"resource_check_timeout" default:"10m"`
+	PodLabelWaitTimeout     time.Duration       `name:"pod_label_wait_timeout" default:"10m"`
+	PodDeletionWaitTimeout  time.Duration       `name:"pod_deletion_wait_timeout" default:"10m"`
+	PodTerminateGracePeriod time.Duration       `name:"pod_terminate_grace_period" default:"5m"`
+	SpiloRunAsUser          *int64              `json:"spilo_runasuser,omitempty"`
+	SpiloRunAsGroup         *int64              `json:"spilo_runasgroup,omitempty"`
+	SpiloFSGroup            *int64              `name:"spilo_fsgroup"`
+	PodPriorityClassName    string              `name:"pod_priority_class_name"`
+	ClusterDomain           string              `name:"cluster_domain" default:"cluster.local"`
+	SpiloPrivileged         bool                `name:"spilo_privileged" default:"false"`
+	ClusterLabels           map[string]string   `name:"cluster_labels" default:"application:spilo"`
+	InheritedLabels         []string            `name:"inherited_labels" default:""`
+	InheritedAnnotations    []string            `name:"inherited_annotations" default:""`
+	DownscalerAnnotations   []string            `name:"downscaler_annotations"`
+	ClusterNameLabel        string              `name:"cluster_name_label" default:"cluster-name"`
+	DeleteAnnotationDateKey string              `name:"delete_annotation_date_key"`
+	DeleteAnnotationNameKey string              `name:"delete_annotation_name_key"`
+	PodRoleLabel            string              `name:"pod_role_label" default:"spilo-role"`
+	PodToleration           map[string]string   `name:"toleration" default:""`
+	DefaultCPURequest       string              `name:"default_cpu_request" default:"100m"`
+	DefaultMemoryRequest    string              `name:"default_memory_request" default:"100Mi"`
+	DefaultCPULimit         string              `name:"default_cpu_limit" default:"1"`
+	DefaultMemoryLimit      string              `name:"default_memory_limit" default:"500Mi"`
+	MinCPULimit             string              `name:"min_cpu_limit" default:"250m"`
+	MinMemoryLimit          string              `name:"min_memory_limit" default:"250Mi"`
+	PodEnvironmentConfigMap spec.NamespacedName `name:"pod_environment_configmap"`
+	PodEnvironmentSecret    string              `name:"pod_environment_secret"`
+	NodeReadinessLabel      map[string]string   `name:"node_readiness_label" default:""`
+	MaxInstances            int32               `name:"max_instances" default:"-1"`
+	MinInstances            int32               `name:"min_instances" default:"-1"`
+	ShmVolume               *bool               `name:"enable_shm_volume" default:"true"`
+}
+
+type InfrastructureRole struct {
+	// Name of a secret which describes the role, and optionally name of a
+	// configmap with an extra information
+	SecretName spec.NamespacedName
+
+	UserKey     string
+	PasswordKey string
+	RoleKey     string
+
+	DefaultUserValue string
+	DefaultRoleValue string
+
+	// This field point out the detailed yaml definition of the role, if exists
+	Details string
+
+	// Specify if a secret contains multiple fields in the following format:
+	//
+	// 	%(userkey)idx: ...
+	// 	%(passwordkey)idx: ...
+	// 	%(rolekey)idx: ...
+	//
+	// If it does, Name/Password/Role are interpreted not as unique field
+	// names, but as a template.
+
+	Template bool
 }
 
 // Auth describes authentication specific configuration parameters
 type Auth struct {
-	SecretNameTemplate            StringTemplate      `name:"secret_name_template" default:"{username}.{cluster}.credentials.{tprkind}.{tprgroup}"`
-	PamRoleName                   string              `name:"pam_role_name" default:"zalandos"`
-	PamConfiguration              string              `name:"pam_configuration" default:"https://info.example.com/oauth2/tokeninfo?access_token= uid realm=/employees"`
-	TeamsAPIUrl                   string              `name:"teams_api_url" default:"https://teams.example.com/api/"`
-	OAuthTokenSecretName          spec.NamespacedName `name:"oauth_token_secret_name" default:"postgresql-operator"`
-	InfrastructureRolesSecretName spec.NamespacedName `name:"infrastructure_roles_secret_name"`
-	SuperUsername                 string              `name:"super_username" default:"postgres"`
-	ReplicationUsername           string              `name:"replication_username" default:"standby"`
+	SecretNameTemplate            StringTemplate        `name:"secret_name_template" default:"{username}.{cluster}.credentials.{tprkind}.{tprgroup}"`
+	PamRoleName                   string                `name:"pam_role_name" default:"zalandos"`
+	PamConfiguration              string                `name:"pam_configuration" default:"https://info.example.com/oauth2/tokeninfo?access_token= uid realm=/employees"`
+	TeamsAPIUrl                   string                `name:"teams_api_url" default:"https://teams.example.com/api/"`
+	OAuthTokenSecretName          spec.NamespacedName   `name:"oauth_token_secret_name" default:"postgresql-operator"`
+	InfrastructureRolesSecretName spec.NamespacedName   `name:"infrastructure_roles_secret_name"`
+	InfrastructureRoles           []*InfrastructureRole `name:"-"`
+	InfrastructureRolesDefs       string                `name:"infrastructure_roles_secrets"`
+	SuperUsername                 string                `name:"super_username" default:"postgres"`
+	ReplicationUsername           string                `name:"replication_username" default:"standby"`
 }
 
 // Scalyr holds the configuration for the Scalyr Agent sidecar for log shipping:
@@ -73,14 +111,31 @@ type Scalyr struct {
 
 // LogicalBackup defines configuration for logical backup
 type LogicalBackup struct {
-	LogicalBackupSchedule          string `name:"logical_backup_schedule" default:"30 00 * * *"`
-	LogicalBackupDockerImage       string `name:"logical_backup_docker_image" default:"registry.opensource.zalan.do/acid/logical-backup"`
-	LogicalBackupS3Bucket          string `name:"logical_backup_s3_bucket" default:""`
-	LogicalBackupS3Region          string `name:"logical_backup_s3_region" default:""`
-	LogicalBackupS3Endpoint        string `name:"logical_backup_s3_endpoint" default:""`
-	LogicalBackupS3AccessKeyID     string `name:"logical_backup_s3_access_key_id" default:""`
-	LogicalBackupS3SecretAccessKey string `name:"logical_backup_s3_secret_access_key" default:""`
-	LogicalBackupS3SSE             string `name:"logical_backup_s3_sse" default:"AES256"`
+	LogicalBackupSchedule                     string `name:"logical_backup_schedule" default:"30 00 * * *"`
+	LogicalBackupDockerImage                  string `name:"logical_backup_docker_image" default:"registry.opensource.zalan.do/acid/logical-backup:v1.6.0"`
+	LogicalBackupProvider                     string `name:"logical_backup_provider" default:"s3"`
+	LogicalBackupS3Bucket                     string `name:"logical_backup_s3_bucket" default:""`
+	LogicalBackupS3Region                     string `name:"logical_backup_s3_region" default:""`
+	LogicalBackupS3Endpoint                   string `name:"logical_backup_s3_endpoint" default:""`
+	LogicalBackupS3AccessKeyID                string `name:"logical_backup_s3_access_key_id" default:""`
+	LogicalBackupS3SecretAccessKey            string `name:"logical_backup_s3_secret_access_key" default:""`
+	LogicalBackupS3SSE                        string `name:"logical_backup_s3_sse" default:""`
+	LogicalBackupGoogleApplicationCredentials string `name:"logical_backup_google_application_credentials" default:""`
+	LogicalBackupJobPrefix                    string `name:"logical_backup_job_prefix" default:"logical-backup-"`
+}
+
+// Operator options for connection pooler
+type ConnectionPooler struct {
+	NumberOfInstances                    *int32 `name:"connection_pooler_number_of_instances" default:"2"`
+	Schema                               string `name:"connection_pooler_schema" default:"pooler"`
+	User                                 string `name:"connection_pooler_user" default:"pooler"`
+	Image                                string `name:"connection_pooler_image" default:"registry.opensource.zalan.do/acid/pgbouncer"`
+	Mode                                 string `name:"connection_pooler_mode" default:"transaction"`
+	MaxDBConnections                     *int32 `name:"connection_pooler_max_db_connections" default:"60"`
+	ConnectionPoolerDefaultCPURequest    string `name:"connection_pooler_default_cpu_request" default:"500m"`
+	ConnectionPoolerDefaultMemoryRequest string `name:"connection_pooler_default_memory_request" default:"100Mi"`
+	ConnectionPoolerDefaultCPULimit      string `name:"connection_pooler_default_cpu_limit" default:"1"`
+	ConnectionPoolerDefaultMemoryLimit   string `name:"connection_pooler_default_memory_limit" default:"100Mi"`
 }
 
 // Config describes operator config
@@ -90,12 +145,15 @@ type Config struct {
 	Auth
 	Scalyr
 	LogicalBackup
+	ConnectionPooler
 
-	WatchedNamespace      string            `name:"watched_namespace"`    // special values: "*" means 'watch all namespaces', the empty string "" means 'watch a namespace where operator is deployed to'
-	EtcdHost              string            `name:"etcd_host" default:""` // special values: the empty string "" means Patroni will use K8s as a DCS
-	DockerImage           string            `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-12:1.6-p2"`
-	Sidecars              map[string]string `name:"sidecar_docker_images"`
-	PodServiceAccountName string            `name:"pod_service_account_name" default:"postgres-pod"`
+	WatchedNamespace        string            `name:"watched_namespace"` // special values: "*" means 'watch all namespaces', the empty string "" means 'watch a namespace where operator is deployed to'
+	KubernetesUseConfigMaps bool              `name:"kubernetes_use_configmaps" default:"false"`
+	EtcdHost                string            `name:"etcd_host" default:""` // special values: the empty string "" means Patroni will use K8s as a DCS
+	DockerImage             string            `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-13:2.0-p2"`
+	SidecarImages           map[string]string `name:"sidecar_docker_images"` // deprecated in favour of SidecarContainers
+	SidecarContainers       []v1.Container    `name:"sidecars"`
+	PodServiceAccountName   string            `name:"pod_service_account_name" default:"postgres-pod"`
 	// value of this string must be valid JSON or YAML; see initPodServiceAccount
 	PodServiceAccountDefinition            string            `name:"pod_service_account_definition" default:""`
 	PodServiceAccountRoleDefinition        string            `name:"pod_service_account_role_definition" default:""`
@@ -106,43 +164,53 @@ type Config struct {
 	WALES3Bucket                           string            `name:"wal_s3_bucket"`
 	LogS3Bucket                            string            `name:"log_s3_bucket"`
 	KubeIAMRole                            string            `name:"kube_iam_role"`
+	WALGSBucket                            string            `name:"wal_gs_bucket"`
+	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"`
 	EnableTeamSuperuser                    bool              `name:"enable_team_superuser" default:"false"`
 	TeamAdminRole                          string            `name:"team_admin_role" default:"admin"`
 	EnableAdminRoleForUsers                bool              `name:"enable_admin_role_for_users" default:"true"`
+	EnablePostgresTeamCRD                  bool              `name:"enable_postgres_team_crd" default:"false"`
+	EnablePostgresTeamCRDSuperusers        bool              `name:"enable_postgres_team_crd_superusers" default:"false"`
 	EnableMasterLoadBalancer               bool              `name:"enable_master_load_balancer" default:"true"`
 	EnableReplicaLoadBalancer              bool              `name:"enable_replica_load_balancer" default:"false"`
 	CustomServiceAnnotations               map[string]string `name:"custom_service_annotations"`
 	CustomPodAnnotations                   map[string]string `name:"custom_pod_annotations"`
 	EnablePodAntiAffinity                  bool              `name:"enable_pod_antiaffinity" default:"false"`
 	PodAntiAffinityTopologyKey             string            `name:"pod_antiaffinity_topology_key" default:"kubernetes.io/hostname"`
-	// deprecated and kept for backward compatibility
-	EnableLoadBalancer        *bool             `name:"enable_load_balancer"`
-	MasterDNSNameFormat       StringTemplate    `name:"master_dns_name_format" default:"{cluster}.{team}.{hostedzone}"`
-	ReplicaDNSNameFormat      StringTemplate    `name:"replica_dns_name_format" default:"{cluster}-repl.{team}.{hostedzone}"`
-	PDBNameFormat             StringTemplate    `name:"pdb_name_format" default:"postgres-{cluster}-pdb"`
-	EnablePodDisruptionBudget *bool             `name:"enable_pod_disruption_budget" default:"true"`
-	EnableInitContainers      *bool             `name:"enable_init_containers" default:"true"`
-	EnableSidecars            *bool             `name:"enable_sidecars" default:"true"`
-	Workers                   uint32            `name:"workers" default:"4"`
-	APIPort                   int               `name:"api_port" default:"8080"`
-	RingLogLines              int               `name:"ring_log_lines" default:"100"`
-	ClusterHistoryEntries     int               `name:"cluster_history_entries" default:"1000"`
-	TeamAPIRoleConfiguration  map[string]string `name:"team_api_role_configuration" default:"log_statement:all"`
-	PodTerminateGracePeriod   time.Duration     `name:"pod_terminate_grace_period" default:"5m"`
-	PodManagementPolicy       string            `name:"pod_management_policy" default:"ordered_ready"`
-	ProtectedRoles            []string          `name:"protected_role_names" default:"admin"`
-	PostgresSuperuserTeams    []string          `name:"postgres_superuser_teams" default:""`
-	SetMemoryRequestToLimit   bool              `name:"set_memory_request_to_limit" default:"false"`
+	StorageResizeMode                      string            `name:"storage_resize_mode" default:"pvc"`
+	EnableLoadBalancer                     *bool             `name:"enable_load_balancer"` // deprecated and kept for backward compatibility
+	ExternalTrafficPolicy                  string            `name:"external_traffic_policy" default:"Cluster"`
+	MasterDNSNameFormat                    StringTemplate    `name:"master_dns_name_format" default:"{cluster}.{team}.{hostedzone}"`
+	ReplicaDNSNameFormat                   StringTemplate    `name:"replica_dns_name_format" default:"{cluster}-repl.{team}.{hostedzone}"`
+	PDBNameFormat                          StringTemplate    `name:"pdb_name_format" default:"postgres-{cluster}-pdb"`
+	EnablePodDisruptionBudget              *bool             `name:"enable_pod_disruption_budget" default:"true"`
+	EnableInitContainers                   *bool             `name:"enable_init_containers" default:"true"`
+	EnableSidecars                         *bool             `name:"enable_sidecars" default:"true"`
+	Workers                                uint32            `name:"workers" default:"8"`
+	APIPort                                int               `name:"api_port" default:"8080"`
+	RingLogLines                           int               `name:"ring_log_lines" default:"100"`
+	ClusterHistoryEntries                  int               `name:"cluster_history_entries" default:"1000"`
+	TeamAPIRoleConfiguration               map[string]string `name:"team_api_role_configuration" default:"log_statement:all"`
+	PodTerminateGracePeriod                time.Duration     `name:"pod_terminate_grace_period" default:"5m"`
+	PodManagementPolicy                    string            `name:"pod_management_policy" default:"ordered_ready"`
+	ProtectedRoles                         []string          `name:"protected_role_names" default:"admin"`
+	PostgresSuperuserTeams                 []string          `name:"postgres_superuser_teams" default:""`
+	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:"true"`
+	EnableSpiloWalPathCompat               bool              `name:"enable_spilo_wal_path_compat" default:"false"`
 }
 
 // MustMarshal marshals the config or panics
 func (c Config) MustMarshal() string {
-	b, err := json.MarshalIndent(c, "", "\t")
+	b, err := json.MarshalIndent(c, "", "   ")
 	if err != nil {
 		panic(err)
 	}
@@ -197,5 +265,16 @@ func validate(cfg *Config) (err error) {
 	if cfg.Workers == 0 {
 		err = fmt.Errorf("number of workers should be higher than 0")
 	}
+
+	if *cfg.ConnectionPooler.NumberOfInstances < constants.ConnectionPoolerMinInstances {
+		msg := "number of connection pooler instances should be higher than %d"
+		err = fmt.Errorf(msg, constants.ConnectionPoolerMinInstances)
+	}
+
+	if cfg.ConnectionPooler.User == cfg.SuperUsername {
+		msg := "Connection pool user is not allowed to be the same as super user, username: %s"
+		err = fmt.Errorf(msg, cfg.ConnectionPooler.User)
+	}
+
 	return
 }

pkg/util/constants/annotations.go

@@ -7,4 +7,5 @@ const (
 	ElbTimeoutAnnotationValue          = "3600"
 	KubeIAmAnnotation                  = "iam.amazonaws.com/role"
 	VolumeStorateProvisionerAnnotation = "pv.kubernetes.io/provisioned-by"
+	PostgresqlControllerAnnotationKey  = "acid.zalan.do/controller"
 )

pkg/util/constants/pooler.go

@@ -0,0 +1,18 @@
+package constants
+
+// Connection pooler specific constants
+const (
+	ConnectionPoolerUserName             = "pooler"
+	ConnectionPoolerSchemaName           = "pooler"
+	ConnectionPoolerDefaultType          = "pgbouncer"
+	ConnectionPoolerDefaultMode          = "transaction"
+	ConnectionPoolerDefaultCpuRequest    = "500m"
+	ConnectionPoolerDefaultCpuLimit      = "1"
+	ConnectionPoolerDefaultMemoryRequest = "100Mi"
+	ConnectionPoolerDefaultMemoryLimit   = "100Mi"
+
+	ConnectionPoolerContainer            = 0
+	ConnectionPoolerMaxDBConnections     = 60
+	ConnectionPoolerMaxClientConnections = 10000
+	ConnectionPoolerMinInstances         = 1
+)

pkg/util/constants/roles.go

@@ -2,15 +2,21 @@ package constants
 
 // Roles specific constants
 const (
-	PasswordLength         = 64
-	SuperuserKeyName       = "superuser"
-	ReplicationUserKeyName = "replication"
-	RoleFlagSuperuser      = "SUPERUSER"
-	RoleFlagInherit        = "INHERIT"
-	RoleFlagLogin          = "LOGIN"
-	RoleFlagNoLogin        = "NOLOGIN"
-	RoleFlagCreateRole     = "CREATEROLE"
-	RoleFlagCreateDB       = "CREATEDB"
-	RoleFlagReplication    = "REPLICATION"
-	RoleFlagByPassRLS      = "BYPASSRLS"
+	PasswordLength              = 64
+	SuperuserKeyName            = "superuser"
+	ConnectionPoolerUserKeyName = "pooler"
+	ReplicationUserKeyName      = "replication"
+	RoleFlagSuperuser           = "SUPERUSER"
+	RoleFlagInherit             = "INHERIT"
+	RoleFlagLogin               = "LOGIN"
+	RoleFlagNoLogin             = "NOLOGIN"
+	RoleFlagCreateRole          = "CREATEROLE"
+	RoleFlagCreateDB            = "CREATEDB"
+	RoleFlagReplication         = "REPLICATION"
+	RoleFlagByPassRLS           = "BYPASSRLS"
+	OwnerRoleNameSuffix         = "_owner"
+	ReaderRoleNameSuffix        = "_reader"
+	WriterRoleNameSuffix        = "_writer"
+	UserRoleNameSuffix          = "_user"
+	DefaultSearchPath           = "\"$user\""
 )

pkg/util/k8sutil/k8sutil.go

@@ -1,19 +1,28 @@
 package k8sutil
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 
 	b64 "encoding/base64"
+	"encoding/json"
 
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	clientbatchv1beta1 "k8s.io/client-go/kubernetes/typed/batch/v1beta1"
 
+	apiacidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
+	acidv1client "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned"
+	acidv1 "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/typed/acid.zalan.do/v1"
+	"github.com/zalando/postgres-operator/pkg/spec"
+	apiappsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	policybeta1 "k8s.io/api/policy/v1beta1"
 	apiextclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
-	apiextbeta1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1beta1"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	appsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -21,11 +30,12 @@ import (
 	rbacv1 "k8s.io/client-go/kubernetes/typed/rbac/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
-
-	acidv1client "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+func Int32ToPointer(value int32) *int32 {
+	return &value
+}
+
 // KubernetesClient describes getters for Kubernetes objects
 type KubernetesClient struct {
 	corev1.SecretsGetter
@@ -38,12 +48,17 @@ type KubernetesClient struct {
 	corev1.NodesGetter
 	corev1.NamespacesGetter
 	corev1.ServiceAccountsGetter
+	corev1.EventsGetter
 	appsv1.StatefulSetsGetter
+	appsv1.DeploymentsGetter
 	rbacv1.RolesGetter
 	rbacv1.RoleBindingsGetter
 	policyv1beta1.PodDisruptionBudgetsGetter
-	apiextbeta1.CustomResourceDefinitionsGetter
+	apiextv1.CustomResourceDefinitionsGetter
 	clientbatchv1beta1.CronJobsGetter
+	acidv1.OperatorConfigurationsGetter
+	acidv1.PostgresTeamsGetter
+	acidv1.PostgresqlsGetter
 
 	RESTClient      rest.Interface
 	AcidV1ClientSet *acidv1client.Clientset
@@ -56,6 +71,34 @@ type mockSecret struct {
 type MockSecretGetter struct {
 }
 
+type mockDeployment struct {
+	appsv1.DeploymentInterface
+}
+
+type mockDeploymentNotExist struct {
+	appsv1.DeploymentInterface
+}
+
+type MockDeploymentGetter struct {
+}
+
+type MockDeploymentNotExistGetter struct {
+}
+
+type mockService struct {
+	corev1.ServiceInterface
+}
+
+type mockServiceNotExist struct {
+	corev1.ServiceInterface
+}
+
+type MockServiceGetter struct {
+}
+
+type MockServiceNotExistGetter struct {
+}
+
 type mockConfigMap struct {
 	corev1.ConfigMapInterface
 }
@@ -102,28 +145,65 @@ func NewFromConfig(cfg *rest.Config) (KubernetesClient, error) {
 	kubeClient.NodesGetter = client.CoreV1()
 	kubeClient.NamespacesGetter = client.CoreV1()
 	kubeClient.StatefulSetsGetter = client.AppsV1()
+	kubeClient.DeploymentsGetter = client.AppsV1()
 	kubeClient.PodDisruptionBudgetsGetter = client.PolicyV1beta1()
 	kubeClient.RESTClient = client.CoreV1().RESTClient()
 	kubeClient.RolesGetter = client.RbacV1()
 	kubeClient.RoleBindingsGetter = client.RbacV1()
 	kubeClient.CronJobsGetter = client.BatchV1beta1()
+	kubeClient.EventsGetter = client.CoreV1()
 
 	apiextClient, err := apiextclient.NewForConfig(cfg)
 	if err != nil {
 		return kubeClient, fmt.Errorf("could not create api client:%v", err)
 	}
 
-	kubeClient.CustomResourceDefinitionsGetter = apiextClient.ApiextensionsV1beta1()
+	kubeClient.CustomResourceDefinitionsGetter = apiextClient.ApiextensionsV1()
+
 	kubeClient.AcidV1ClientSet = acidv1client.NewForConfigOrDie(cfg)
+	if err != nil {
+		return kubeClient, fmt.Errorf("could not create acid.zalan.do clientset: %v", err)
+	}
+
+	kubeClient.OperatorConfigurationsGetter = kubeClient.AcidV1ClientSet.AcidV1()
+	kubeClient.PostgresTeamsGetter = kubeClient.AcidV1ClientSet.AcidV1()
+	kubeClient.PostgresqlsGetter = kubeClient.AcidV1ClientSet.AcidV1()
 
 	return kubeClient, nil
 }
 
+// SetPostgresCRDStatus of Postgres cluster
+func (client *KubernetesClient) SetPostgresCRDStatus(clusterName spec.NamespacedName, status string) (*apiacidv1.Postgresql, error) {
+	var pg *apiacidv1.Postgresql
+	var pgStatus apiacidv1.PostgresStatus
+	pgStatus.PostgresClusterStatus = status
+
+	patch, err := json.Marshal(struct {
+		PgStatus interface{} `json:"status"`
+	}{&pgStatus})
+
+	if err != nil {
+		return pg, fmt.Errorf("could not marshal status: %v", err)
+	}
+
+	// we cannot do a full scale update here without fetching the previous manifest (as the resourceVersion may differ),
+	// however, we could do patch without it. In the future, once /status subresource is there (starting Kubernetes 1.11)
+	// we should take advantage of it.
+	pg, err = client.PostgresqlsGetter.Postgresqls(clusterName.Namespace).Patch(
+		context.TODO(), clusterName.Name, types.MergePatchType, patch, metav1.PatchOptions{}, "status")
+	if err != nil {
+		return pg, fmt.Errorf("could not update status: %v", err)
+	}
+
+	// update the spec, maintaining the new resourceVersion.
+	return pg, nil
+}
+
 // SameService compares the Services
 func SameService(cur, new *v1.Service) (match bool, reason string) {
 	//TODO: improve comparison
 	if cur.Spec.Type != new.Spec.Type {
-		return false, fmt.Sprintf("new service's type %q doesn't match the current one %q",
+		return false, fmt.Sprintf("new service's type %q does not match the current one %q",
 			new.Spec.Type, cur.Spec.Type)
 	}
 
@@ -133,13 +213,13 @@ func SameService(cur, new *v1.Service) (match bool, reason string) {
 	/* work around Kubernetes 1.6 serializing [] as nil. See https://github.com/kubernetes/kubernetes/issues/43203 */
 	if (len(oldSourceRanges) != 0) || (len(newSourceRanges) != 0) {
 		if !reflect.DeepEqual(oldSourceRanges, newSourceRanges) {
-			return false, "new service's LoadBalancerSourceRange doesn't match the current one"
+			return false, "new service's LoadBalancerSourceRange does not match the current one"
 		}
 	}
 
 	match = true
 
-	reasonPrefix := "new service's annotations doesn't match the current one:"
+	reasonPrefix := "new service's annotations does not match the current one:"
 	for ann := range cur.Annotations {
 		if _, ok := new.Annotations[ann]; !ok {
 			match = false
@@ -175,7 +255,7 @@ func SamePDB(cur, new *policybeta1.PodDisruptionBudget) (match bool, reason stri
 	//TODO: improve comparison
 	match = reflect.DeepEqual(new.Spec, cur.Spec)
 	if !match {
-		reason = "new PDB spec doesn't match the current one"
+		reason = "new PDB spec does not match the current one"
 	}
 
 	return
@@ -189,62 +269,230 @@ func getJobImage(cronJob *batchv1beta1.CronJob) string {
 func SameLogicalBackupJob(cur, new *batchv1beta1.CronJob) (match bool, reason string) {
 
 	if cur.Spec.Schedule != new.Spec.Schedule {
-		return false, fmt.Sprintf("new job's schedule %q doesn't match the current one %q",
+		return false, fmt.Sprintf("new job's schedule %q does not match the current one %q",
 			new.Spec.Schedule, cur.Spec.Schedule)
 	}
 
 	newImage := getJobImage(new)
 	curImage := getJobImage(cur)
 	if newImage != curImage {
-		return false, fmt.Sprintf("new job's image %q doesn't match the current one %q",
+		return false, fmt.Sprintf("new job's image %q does not match the current one %q",
 			newImage, curImage)
 	}
 
 	return true, ""
 }
 
-func (c *mockSecret) Get(name string, options metav1.GetOptions) (*v1.Secret, error) {
-	if name != "infrastructureroles-test" {
-		return nil, fmt.Errorf("NotFound")
-	}
-	secret := &v1.Secret{}
-	secret.Name = "testcluster"
-	secret.Data = map[string][]byte{
+func (c *mockSecret) Get(ctx context.Context, name string, options metav1.GetOptions) (*v1.Secret, error) {
+	oldFormatSecret := &v1.Secret{}
+	oldFormatSecret.Name = "testcluster"
+	oldFormatSecret.Data = map[string][]byte{
 		"user1":     []byte("testrole"),
 		"password1": []byte("testpassword"),
 		"inrole1":   []byte("testinrole"),
 		"foobar":    []byte(b64.StdEncoding.EncodeToString([]byte("password"))),
 	}
-	return secret, nil
+
+	newFormatSecret := &v1.Secret{}
+	newFormatSecret.Name = "test-secret-new-format"
+	newFormatSecret.Data = map[string][]byte{
+		"user":       []byte("new-test-role"),
+		"password":   []byte("new-test-password"),
+		"inrole":     []byte("new-test-inrole"),
+		"new-foobar": []byte(b64.StdEncoding.EncodeToString([]byte("password"))),
+	}
+
+	secrets := map[string]*v1.Secret{
+		"infrastructureroles-old-test": oldFormatSecret,
+		"infrastructureroles-new-test": newFormatSecret,
+	}
+
+	for idx := 1; idx <= 2; idx++ {
+		newFormatStandaloneSecret := &v1.Secret{}
+		newFormatStandaloneSecret.Name = fmt.Sprintf("test-secret-new-format%d", idx)
+		newFormatStandaloneSecret.Data = map[string][]byte{
+			"user":     []byte(fmt.Sprintf("new-test-role%d", idx)),
+			"password": []byte(fmt.Sprintf("new-test-password%d", idx)),
+			"inrole":   []byte(fmt.Sprintf("new-test-inrole%d", idx)),
+		}
+
+		secrets[fmt.Sprintf("infrastructureroles-new-test%d", idx)] =
+			newFormatStandaloneSecret
+	}
+
+	if secret, exists := secrets[name]; exists {
+		return secret, nil
+	}
+
+	return nil, fmt.Errorf("NotFound")
 
 }
 
-func (c *mockConfigMap) Get(name string, options metav1.GetOptions) (*v1.ConfigMap, error) {
-	if name != "infrastructureroles-test" {
-		return nil, fmt.Errorf("NotFound")
-	}
-	configmap := &v1.ConfigMap{}
-	configmap.Name = "testcluster"
-	configmap.Data = map[string]string{
+func (c *mockConfigMap) Get(ctx context.Context, name string, options metav1.GetOptions) (*v1.ConfigMap, error) {
+	oldFormatConfigmap := &v1.ConfigMap{}
+	oldFormatConfigmap.Name = "testcluster"
+	oldFormatConfigmap.Data = map[string]string{
 		"foobar": "{}",
 	}
-	return configmap, nil
+
+	newFormatConfigmap := &v1.ConfigMap{}
+	newFormatConfigmap.Name = "testcluster"
+	newFormatConfigmap.Data = map[string]string{
+		"new-foobar": "{\"user_flags\": [\"createdb\"]}",
+	}
+
+	configmaps := map[string]*v1.ConfigMap{
+		"infrastructureroles-old-test": oldFormatConfigmap,
+		"infrastructureroles-new-test": newFormatConfigmap,
+	}
+
+	if configmap, exists := configmaps[name]; exists {
+		return configmap, nil
+	}
+
+	return nil, fmt.Errorf("NotFound")
 }
 
 // Secrets to be mocked
-func (c *MockSecretGetter) Secrets(namespace string) corev1.SecretInterface {
+func (mock *MockSecretGetter) Secrets(namespace string) corev1.SecretInterface {
 	return &mockSecret{}
 }
 
 // ConfigMaps to be mocked
-func (c *MockConfigMapsGetter) ConfigMaps(namespace string) corev1.ConfigMapInterface {
+func (mock *MockConfigMapsGetter) ConfigMaps(namespace string) corev1.ConfigMapInterface {
 	return &mockConfigMap{}
 }
 
+func (mock *MockDeploymentGetter) Deployments(namespace string) appsv1.DeploymentInterface {
+	return &mockDeployment{}
+}
+
+func (mock *MockDeploymentNotExistGetter) Deployments(namespace string) appsv1.DeploymentInterface {
+	return &mockDeploymentNotExist{}
+}
+
+func (mock *mockDeployment) Create(context.Context, *apiappsv1.Deployment, metav1.CreateOptions) (*apiappsv1.Deployment, error) {
+	return &apiappsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-deployment",
+		},
+		Spec: apiappsv1.DeploymentSpec{
+			Replicas: Int32ToPointer(1),
+		},
+	}, nil
+}
+
+func (mock *mockDeployment) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return nil
+}
+
+func (mock *mockDeployment) Get(ctx context.Context, name string, opts metav1.GetOptions) (*apiappsv1.Deployment, error) {
+	return &apiappsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-deployment",
+		},
+		Spec: apiappsv1.DeploymentSpec{
+			Replicas: Int32ToPointer(1),
+			Template: v1.PodTemplateSpec{
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						v1.Container{
+							Image: "pooler:1.0",
+						},
+					},
+				},
+			},
+		},
+	}, nil
+}
+
+func (mock *mockDeployment) Patch(ctx context.Context, name string, t types.PatchType, data []byte, opts metav1.PatchOptions, subres ...string) (*apiappsv1.Deployment, error) {
+	return &apiappsv1.Deployment{
+		Spec: apiappsv1.DeploymentSpec{
+			Replicas: Int32ToPointer(2),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-deployment",
+		},
+	}, nil
+}
+
+func (mock *mockDeploymentNotExist) Get(ctx context.Context, name string, opts metav1.GetOptions) (*apiappsv1.Deployment, error) {
+	return nil, &apierrors.StatusError{
+		ErrStatus: metav1.Status{
+			Reason: metav1.StatusReasonNotFound,
+		},
+	}
+}
+
+func (mock *mockDeploymentNotExist) Create(context.Context, *apiappsv1.Deployment, metav1.CreateOptions) (*apiappsv1.Deployment, error) {
+	return &apiappsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-deployment",
+		},
+		Spec: apiappsv1.DeploymentSpec{
+			Replicas: Int32ToPointer(1),
+		},
+	}, nil
+}
+
+func (mock *MockServiceGetter) Services(namespace string) corev1.ServiceInterface {
+	return &mockService{}
+}
+
+func (mock *MockServiceNotExistGetter) Services(namespace string) corev1.ServiceInterface {
+	return &mockServiceNotExist{}
+}
+
+func (mock *mockService) Create(context.Context, *v1.Service, metav1.CreateOptions) (*v1.Service, error) {
+	return &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-service",
+		},
+	}, nil
+}
+
+func (mock *mockService) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return nil
+}
+
+func (mock *mockService) Get(ctx context.Context, name string, opts metav1.GetOptions) (*v1.Service, error) {
+	return &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-service",
+		},
+	}, nil
+}
+
+func (mock *mockServiceNotExist) Create(context.Context, *v1.Service, metav1.CreateOptions) (*v1.Service, error) {
+	return &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-service",
+		},
+	}, nil
+}
+
+func (mock *mockServiceNotExist) Get(ctx context.Context, name string, opts metav1.GetOptions) (*v1.Service, error) {
+	return nil, &apierrors.StatusError{
+		ErrStatus: metav1.Status{
+			Reason: metav1.StatusReasonNotFound,
+		},
+	}
+}
+
 // NewMockKubernetesClient for other tests
 func NewMockKubernetesClient() KubernetesClient {
 	return KubernetesClient{
-		SecretsGetter:    &MockSecretGetter{},
-		ConfigMapsGetter: &MockConfigMapsGetter{},
+		SecretsGetter:     &MockSecretGetter{},
+		ConfigMapsGetter:  &MockConfigMapsGetter{},
+		DeploymentsGetter: &MockDeploymentGetter{},
+		ServicesGetter:    &MockServiceGetter{},
+	}
+}
+
+func ClientMissingObjects() KubernetesClient {
+	return KubernetesClient{
+		DeploymentsGetter: &MockDeploymentNotExistGetter{},
+		ServicesGetter:    &MockServiceNotExistGetter{},
 	}
 }

pkg/util/k8sutil/k8sutil_test.go

@@ -63,7 +63,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's type "LoadBalancer" doesn't match the current one "ClusterIP"`,
+			reason: `new service's type "LoadBalancer" does not match the current one "ClusterIP"`,
 		},
 		{
 			about: "services differ on lb source ranges",
@@ -82,7 +82,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"185.249.56.0/22"}),
 			match:  false,
-			reason: `new service's LoadBalancerSourceRange doesn't match the current one`,
+			reason: `new service's LoadBalancerSourceRange does not match the current one`,
 		},
 		{
 			about: "new service doesn't have lb source ranges",
@@ -101,7 +101,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{}),
 			match:  false,
-			reason: `new service's LoadBalancerSourceRange doesn't match the current one`,
+			reason: `new service's LoadBalancerSourceRange does not match the current one`,
 		},
 		{
 			about: "services differ on DNS annotation",
@@ -120,7 +120,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's annotations doesn't match the current one: 'external-dns.alpha.kubernetes.io/hostname' changed from 'clstr.acid.zalan.do' to 'new_clstr.acid.zalan.do'.`,
+			reason: `new service's annotations does not match the current one: 'external-dns.alpha.kubernetes.io/hostname' changed from 'clstr.acid.zalan.do' to 'new_clstr.acid.zalan.do'.`,
 		},
 		{
 			about: "services differ on AWS ELB annotation",
@@ -139,7 +139,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's annotations doesn't match the current one: 'service.beta.kubernetes.io/aws-load-balancer-connection-idle-timeout' changed from '3600' to '1800'.`,
+			reason: `new service's annotations does not match the current one: 'service.beta.kubernetes.io/aws-load-balancer-connection-idle-timeout' changed from '3600' to '1800'.`,
 		},
 		{
 			about: "service changes existing annotation",
@@ -160,7 +160,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's annotations doesn't match the current one: 'foo' changed from 'bar' to 'baz'.`,
+			reason: `new service's annotations does not match the current one: 'foo' changed from 'bar' to 'baz'.`,
 		},
 		{
 			about: "service changes multiple existing annotations",
@@ -184,7 +184,7 @@ func TestSameService(t *testing.T) {
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match: false,
 			// Test just the prefix to avoid flakiness and map sorting
-			reason: `new service's annotations doesn't match the current one:`,
+			reason: `new service's annotations does not match the current one:`,
 		},
 		{
 			about: "service adds a new custom annotation",
@@ -204,7 +204,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's annotations doesn't match the current one: Added 'foo' with value 'bar'.`,
+			reason: `new service's annotations does not match the current one: Added 'foo' with value 'bar'.`,
 		},
 		{
 			about: "service removes a custom annotation",
@@ -224,7 +224,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's annotations doesn't match the current one: Removed 'foo'.`,
+			reason: `new service's annotations does not match the current one: Removed 'foo'.`,
 		},
 		{
 			about: "service removes a custom annotation and adds a new one",
@@ -245,7 +245,7 @@ func TestSameService(t *testing.T) {
 				v1.ServiceTypeLoadBalancer,
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match:  false,
-			reason: `new service's annotations doesn't match the current one: Removed 'foo'. Added 'bar' with value 'foo'.`,
+			reason: `new service's annotations does not match the current one: Removed 'foo'. Added 'bar' with value 'foo'.`,
 		},
 		{
 			about: "service removes a custom annotation, adds a new one and change another",
@@ -269,7 +269,7 @@ func TestSameService(t *testing.T) {
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match: false,
 			// Test just the prefix to avoid flakiness and map sorting
-			reason: `new service's annotations doesn't match the current one: Removed 'foo'.`,
+			reason: `new service's annotations does not match the current one: Removed 'foo'.`,
 		},
 		{
 			about: "service add annotations",
@@ -286,7 +286,7 @@ func TestSameService(t *testing.T) {
 				[]string{"128.141.0.0/16", "137.138.0.0/16"}),
 			match: false,
 			// Test just the prefix to avoid flakiness and map sorting
-			reason: `new service's annotations doesn't match the current one: Added `,
+			reason: `new service's annotations does not match the current one: Added `,
 		},
 	}
 	for _, tt := range tests {

pkg/util/nicediff/diff.go

@@ -0,0 +1,191 @@
+// Copyright 2013 Google Inc.  All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package diff implements a linewise diff algorithm.
+package nicediff
+
+import (
+	"fmt"
+	"strings"
+)
+
+// Chunk represents a piece of the diff.  A chunk will not have both added and
+// deleted lines.  Equal lines are always after any added or deleted lines.
+// A Chunk may or may not have any lines in it, especially for the first or last
+// chunk in a computation.
+type Chunk struct {
+	Added   []string
+	Deleted []string
+	Equal   []string
+}
+
+func (c *Chunk) empty() bool {
+	return len(c.Added) == 0 && len(c.Deleted) == 0 && len(c.Equal) == 0
+}
+
+// Diff returns a string containing a line-by-line unified diff of the linewise
+// changes required to make A into B.  Each line is prefixed with '+', '-', or
+// ' ' to indicate if it should be added, removed, or is correct respectively.
+func Diff(A, B string, skipEqual bool) string {
+	aLines := strings.Split(A, "\n")
+	bLines := strings.Split(B, "\n")
+	return Render(DiffChunks(aLines, bLines), skipEqual)
+}
+
+// Render renders the slice of chunks into a representation that prefixes
+// the lines with '+', '-', or ' ' depending on whether the line was added,
+// removed, or equal (respectively).
+func Render(chunks []Chunk, skipEqual bool) string {
+	buf := new(strings.Builder)
+	for _, c := range chunks {
+		for _, line := range c.Added {
+			fmt.Fprintf(buf, "+%s\n", line)
+		}
+		for _, line := range c.Deleted {
+			fmt.Fprintf(buf, "-%s\n", line)
+		}
+		if !skipEqual {
+			for _, line := range c.Equal {
+				fmt.Fprintf(buf, " %s\n", line)
+			}
+		}
+	}
+	return strings.TrimRight(buf.String(), "\n")
+}
+
+// DiffChunks uses an O(D(N+M)) shortest-edit-script algorithm
+// to compute the edits required from A to B and returns the
+// edit chunks.
+func DiffChunks(a, b []string) []Chunk {
+	// algorithm: http://www.xmailserver.org/diff2.pdf
+
+	// We'll need these quantities a lot.
+	alen, blen := len(a), len(b) // M, N
+
+	// At most, it will require len(a) deletions and len(b) additions
+	// to transform a into b.
+	maxPath := alen + blen // MAX
+	if maxPath == 0 {
+		// degenerate case: two empty lists are the same
+		return nil
+	}
+
+	// Store the endpoint of the path for diagonals.
+	// We store only the a index, because the b index on any diagonal
+	// (which we know during the loop below) is aidx-diag.
+	// endpoint[maxPath] represents the 0 diagonal.
+	//
+	// Stated differently:
+	// endpoint[d] contains the aidx of a furthest reaching path in diagonal d
+	endpoint := make([]int, 2*maxPath+1) // V
+
+	saved := make([][]int, 0, 8) // Vs
+	save := func() {
+		dup := make([]int, len(endpoint))
+		copy(dup, endpoint)
+		saved = append(saved, dup)
+	}
+
+	var editDistance int // D
+dLoop:
+	for editDistance = 0; editDistance <= maxPath; editDistance++ {
+		// The 0 diag(onal) represents equality of a and b.  Each diagonal to
+		// the left is numbered one lower, to the right is one higher, from
+		// -alen to +blen.  Negative diagonals favor differences from a,
+		// positive diagonals favor differences from b.  The edit distance to a
+		// diagonal d cannot be shorter than d itself.
+		//
+		// The iterations of this loop cover either odds or evens, but not both,
+		// If odd indices are inputs, even indices are outputs and vice versa.
+		for diag := -editDistance; diag <= editDistance; diag += 2 { // k
+			var aidx int // x
+			switch {
+			case diag == -editDistance:
+				// This is a new diagonal; copy from previous iter
+				aidx = endpoint[maxPath-editDistance+1] + 0
+			case diag == editDistance:
+				// This is a new diagonal; copy from previous iter
+				aidx = endpoint[maxPath+editDistance-1] + 1
+			case endpoint[maxPath+diag+1] > endpoint[maxPath+diag-1]:
+				// diagonal d+1 was farther along, so use that
+				aidx = endpoint[maxPath+diag+1] + 0
+			default:
+				// diagonal d-1 was farther (or the same), so use that
+				aidx = endpoint[maxPath+diag-1] + 1
+			}
+			// On diagonal d, we can compute bidx from aidx.
+			bidx := aidx - diag // y
+			// See how far we can go on this diagonal before we find a difference.
+			for aidx < alen && bidx < blen && a[aidx] == b[bidx] {
+				aidx++
+				bidx++
+			}
+			// Store the end of the current edit chain.
+			endpoint[maxPath+diag] = aidx
+			// If we've found the end of both inputs, we're done!
+			if aidx >= alen && bidx >= blen {
+				save() // save the final path
+				break dLoop
+			}
+		}
+		save() // save the current path
+	}
+	if editDistance == 0 {
+		return nil
+	}
+	chunks := make([]Chunk, editDistance+1)
+
+	x, y := alen, blen
+	for d := editDistance; d > 0; d-- {
+		endpoint := saved[d]
+		diag := x - y
+		insert := diag == -d || (diag != d && endpoint[maxPath+diag-1] < endpoint[maxPath+diag+1])
+
+		x1 := endpoint[maxPath+diag]
+		var x0, xM, kk int
+		if insert {
+			kk = diag + 1
+			x0 = endpoint[maxPath+kk]
+			xM = x0
+		} else {
+			kk = diag - 1
+			x0 = endpoint[maxPath+kk]
+			xM = x0 + 1
+		}
+		y0 := x0 - kk
+
+		var c Chunk
+		if insert {
+			c.Added = b[y0:][:1]
+		} else {
+			c.Deleted = a[x0:][:1]
+		}
+		if xM < x1 {
+			c.Equal = a[xM:][:x1-xM]
+		}
+
+		x, y = x0, y0
+		chunks[d] = c
+	}
+	if x > 0 {
+		chunks[0].Equal = a[:x]
+	}
+	if chunks[0].empty() {
+		chunks = chunks[1:]
+	}
+	if len(chunks) == 0 {
+		return nil
+	}
+	return chunks
+}

pkg/util/patroni/patroni.go

@@ -3,6 +3,7 @@ package patroni
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net"
@@ -11,7 +12,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 )
 
 const (
@@ -25,6 +26,7 @@ const (
 type Interface interface {
 	Switchover(master *v1.Pod, candidate string) error
 	SetPostgresParameters(server *v1.Pod, options map[string]string) error
+	GetPatroniMemberState(pod *v1.Pod) (string, error)
 }
 
 // Patroni API client
@@ -123,3 +125,36 @@ func (p *Patroni) SetPostgresParameters(server *v1.Pod, parameters map[string]st
 	}
 	return p.httpPostOrPatch(http.MethodPatch, apiURLString+configPath, buf)
 }
+
+//GetPatroniMemberState returns a state of member of a Patroni cluster
+func (p *Patroni) GetPatroniMemberState(server *v1.Pod) (string, error) {
+
+	apiURLString, err := apiURL(server)
+	if err != nil {
+		return "", err
+	}
+	response, err := p.httpClient.Get(apiURLString)
+	if err != nil {
+		return "", fmt.Errorf("could not perform Get request: %v", err)
+	}
+	defer response.Body.Close()
+
+	body, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return "", fmt.Errorf("could not read response: %v", err)
+	}
+
+	data := make(map[string]interface{})
+	err = json.Unmarshal(body, &data)
+	if err != nil {
+		return "", err
+	}
+
+	state, ok := data["state"].(string)
+	if !ok {
+		return "", errors.New("Patroni Get call response contains wrong type for 'state' field")
+	}
+
+	return state, nil
+
+}

pkg/util/retryutil/retry_util.go

@@ -27,7 +27,7 @@ func (t *Ticker) Tick() { <-t.ticker.C }
 func Retry(interval time.Duration, timeout time.Duration, f func() (bool, error)) error {
 	//TODO: make the retry exponential
 	if timeout < interval {
-		return fmt.Errorf("timout(%s) should be greater than interval(%v)", timeout, interval)
+		return fmt.Errorf("timeout(%s) should be greater than interval(%v)", timeout, interval)
 	}
 	tick := &Ticker{time.NewTicker(interval)}
 	return RetryWorker(interval, timeout, tick, f)

pkg/util/teams/teams_test.go

@@ -133,11 +133,11 @@ var requestsURLtc = []struct {
 }{
 	{
 		"coffee://localhost/",
-		fmt.Errorf(`Get coffee://localhost/teams/acid: unsupported protocol scheme "coffee"`),
+		fmt.Errorf(`Get "coffee://localhost/teams/acid": unsupported protocol scheme "coffee"`),
 	},
 	{
 		"http://192.168.0.%31/",
-		fmt.Errorf(`parse http://192.168.0.%%31/teams/acid: invalid URL escape "%%31"`),
+		fmt.Errorf(`parse "http://192.168.0.%%31/teams/acid": invalid URL escape "%%31"`),
 	},
 }
 

pkg/util/users/users.go

@@ -28,6 +28,7 @@ const (
 // an existing roles of another role membership, nor it removes the already assigned flag
 // (except for the NOLOGIN). TODO: process other NOflags, i.e. NOSUPERUSER correctly.
 type DefaultUserSyncStrategy struct {
+	PasswordEncryption string
 }
 
 // ProduceSyncRequests figures out the types of changes that need to happen with the given users.
@@ -45,7 +46,7 @@ func (strategy DefaultUserSyncStrategy) ProduceSyncRequests(dbUsers spec.PgUserM
 			}
 		} else {
 			r := spec.PgSyncUserRequest{}
-			newMD5Password := util.PGUserPassword(newUser)
+			newMD5Password := util.NewEncryptor(strategy.PasswordEncryption).PGUserPassword(newUser)
 
 			if dbUser.Password != newMD5Password {
 				r.User.Password = newMD5Password
@@ -73,36 +74,54 @@ func (strategy DefaultUserSyncStrategy) ProduceSyncRequests(dbUsers spec.PgUserM
 }
 
 // ExecuteSyncRequests makes actual database changes from the requests passed in its arguments.
-func (strategy DefaultUserSyncStrategy) ExecuteSyncRequests(reqs []spec.PgSyncUserRequest, db *sql.DB) error {
-	for _, r := range reqs {
-		switch r.Kind {
+func (strategy DefaultUserSyncStrategy) ExecuteSyncRequests(requests []spec.PgSyncUserRequest, db *sql.DB) error {
+	var reqretries []spec.PgSyncUserRequest
+	var errors []string
+	for _, request := range requests {
+		switch request.Kind {
 		case spec.PGSyncUserAdd:
-			if err := strategy.createPgUser(r.User, db); err != nil {
-				return fmt.Errorf("could not create user %q: %v", r.User.Name, err)
+			if err := strategy.createPgUser(request.User, db); err != nil {
+				reqretries = append(reqretries, request)
+				errors = append(errors, fmt.Sprintf("could not create user %q: %v", request.User.Name, err))
 			}
 		case spec.PGsyncUserAlter:
-			if err := strategy.alterPgUser(r.User, db); err != nil {
-				return fmt.Errorf("could not alter user %q: %v", r.User.Name, err)
+			if err := strategy.alterPgUser(request.User, db); err != nil {
+				reqretries = append(reqretries, request)
+				errors = append(errors, fmt.Sprintf("could not alter user %q: %v", request.User.Name, err))
 			}
 		case spec.PGSyncAlterSet:
-			if err := strategy.alterPgUserSet(r.User, db); err != nil {
-				return fmt.Errorf("could not set custom user %q parameters: %v", r.User.Name, err)
+			if err := strategy.alterPgUserSet(request.User, db); err != nil {
+				reqretries = append(reqretries, request)
+				errors = append(errors, fmt.Sprintf("could not set custom user %q parameters: %v", request.User.Name, err))
 			}
 		default:
-			return fmt.Errorf("unrecognized operation: %v", r.Kind)
+			return fmt.Errorf("unrecognized operation: %v", request.Kind)
 		}
 
 	}
+
+	// creating roles might fail if group role members are created before the parent role
+	// retry adding roles as long as the number of failed attempts is shrinking
+	if len(reqretries) > 0 {
+		if len(reqretries) < len(requests) {
+			if err := strategy.ExecuteSyncRequests(reqretries, db); err != nil {
+				return err
+			}
+		} else {
+			return fmt.Errorf("could not execute sync requests for users: %v", errors)
+		}
+	}
+
 	return nil
 }
-func (strategy DefaultUserSyncStrategy) alterPgUserSet(user spec.PgUser, db *sql.DB) (err error) {
+
+func (strategy DefaultUserSyncStrategy) alterPgUserSet(user spec.PgUser, db *sql.DB) error {
 	queries := produceAlterRoleSetStmts(user)
 	query := fmt.Sprintf(doBlockStmt, strings.Join(queries, ";"))
-	if _, err = db.Exec(query); err != nil {
-		err = fmt.Errorf("dB error: %v, query: %s", err, query)
-		return
+	if _, err := db.Exec(query); err != nil {
+		return fmt.Errorf("dB error: %v, query: %s", err, query)
 	}
-	return
+	return nil
 }
 
 func (strategy DefaultUserSyncStrategy) createPgUser(user spec.PgUser, db *sql.DB) error {
@@ -122,7 +141,7 @@ func (strategy DefaultUserSyncStrategy) createPgUser(user spec.PgUser, db *sql.D
 	if user.Password == "" {
 		userPassword = "PASSWORD NULL"
 	} else {
-		userPassword = fmt.Sprintf(passwordTemplate, util.PGUserPassword(user))
+		userPassword = fmt.Sprintf(passwordTemplate, util.NewEncryptor(strategy.PasswordEncryption).PGUserPassword(user))
 	}
 	query := fmt.Sprintf(createUserSQL, user.Name, strings.Join(userFlags, " "), userPassword)
 
@@ -130,6 +149,12 @@ func (strategy DefaultUserSyncStrategy) createPgUser(user spec.PgUser, db *sql.D
 		return fmt.Errorf("dB error: %v, query: %s", err, query)
 	}
 
+	if len(user.Parameters) > 0 {
+		if err := strategy.alterPgUserSet(user, db); err != nil {
+			return fmt.Errorf("incomplete setup for user %s: %v", user.Name, err)
+		}
+	}
+
 	return nil
 }
 
@@ -137,7 +162,7 @@ func (strategy DefaultUserSyncStrategy) alterPgUser(user spec.PgUser, db *sql.DB
 	var resultStmt []string
 
 	if user.Password != "" || len(user.Flags) > 0 {
-		alterStmt := produceAlterStmt(user)
+		alterStmt := produceAlterStmt(user, strategy.PasswordEncryption)
 		resultStmt = append(resultStmt, alterStmt)
 	}
 	if len(user.MemberOf) > 0 {
@@ -156,14 +181,14 @@ func (strategy DefaultUserSyncStrategy) alterPgUser(user spec.PgUser, db *sql.DB
 	return nil
 }
 
-func produceAlterStmt(user spec.PgUser) string {
+func produceAlterStmt(user spec.PgUser, encryption string) string {
 	// ALTER ROLE ... LOGIN ENCRYPTED PASSWORD ..
 	result := make([]string, 0)
 	password := user.Password
 	flags := user.Flags
 
 	if password != "" {
-		result = append(result, fmt.Sprintf(passwordTemplate, util.PGUserPassword(user)))
+		result = append(result, fmt.Sprintf(passwordTemplate, util.NewEncryptor(encryption).PGUserPassword(user)))
 	}
 	if len(flags) != 0 {
 		result = append(result, strings.Join(flags, " "))

pkg/util/util.go

@@ -1,11 +1,18 @@
 package util
 
 import (
+	"crypto/hmac"
 	"crypto/md5" // #nosec we need it to for PostgreSQL md5 passwords
+	cryptoRand "crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
 	"encoding/hex"
 	"fmt"
+	"math/big"
 	"math/rand"
+	"reflect"
 	"regexp"
+	"sort"
 	"strings"
 	"time"
 
@@ -14,10 +21,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/zalando/postgres-operator/pkg/spec"
+	"golang.org/x/crypto/pbkdf2"
 )
 
 const (
-	md5prefix = "md5"
+	md5prefix         = "md5"
+	scramsha256prefix = "SCRAM-SHA-256"
+	saltlength        = 16
+	iterations        = 4096
 )
 
 var passwordChars = []byte("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
@@ -37,13 +48,17 @@ func False() *bool {
 	return &b
 }
 
-// RandomPassword generates random alphanumeric password of a given length.
+// RandomPassword generates a secure, random alphanumeric password of a given length.
 func RandomPassword(n int) string {
 	b := make([]byte, n)
 	for i := range b {
-		b[i] = passwordChars[rand.Intn(len(passwordChars))]
+		maxN := big.NewInt(int64(len(passwordChars)))
+		if n, err := cryptoRand.Int(cryptoRand.Reader, maxN); err != nil {
+			panic(fmt.Errorf("Unable to generate secure, random password: %v", err))
+		} else {
+			b[i] = passwordChars[n.Int64()]
+		}
 	}
-
 	return string(b)
 }
 
@@ -55,16 +70,62 @@ func NameFromMeta(meta metav1.ObjectMeta) spec.NamespacedName {
 	}
 }
 
-// PGUserPassword is used to generate md5 password hash for a given user. It does nothing for already hashed passwords.
-func PGUserPassword(user spec.PgUser) string {
-	if (len(user.Password) == md5.Size*2+len(md5prefix) && user.Password[:3] == md5prefix) || user.Password == "" {
+type Hasher func(user spec.PgUser) string
+type Random func(n int) string
+
+type Encryptor struct {
+	encrypt Hasher
+	random  Random
+}
+
+func NewEncryptor(encryption string) *Encryptor {
+	e := Encryptor{random: RandomPassword}
+	m := map[string]Hasher{
+		"md5":           e.PGUserPasswordMD5,
+		"scram-sha-256": e.PGUserPasswordScramSHA256,
+	}
+	hasher, ok := m[encryption]
+	if !ok {
+		hasher = e.PGUserPasswordMD5
+	}
+	e.encrypt = hasher
+	return &e
+}
+
+func (e *Encryptor) PGUserPassword(user spec.PgUser) string {
+	if (len(user.Password) == md5.Size*2+len(md5prefix) && user.Password[:3] == md5prefix) ||
+		(len(user.Password) > len(scramsha256prefix) && user.Password[:len(scramsha256prefix)] == scramsha256prefix) || user.Password == "" {
 		// Avoid processing already encrypted or empty passwords
 		return user.Password
 	}
+	return e.encrypt(user)
+}
+
+func (e *Encryptor) PGUserPasswordMD5(user spec.PgUser) string {
 	s := md5.Sum([]byte(user.Password + user.Name)) // #nosec, using md5 since PostgreSQL uses it for hashing passwords.
 	return md5prefix + hex.EncodeToString(s[:])
 }
 
+func (e *Encryptor) PGUserPasswordScramSHA256(user spec.PgUser) string {
+	salt := []byte(e.random(saltlength))
+	key := pbkdf2.Key([]byte(user.Password), salt, iterations, 32, sha256.New)
+	mac := hmac.New(sha256.New, key)
+	mac.Write([]byte("Server Key"))
+	serverKey := mac.Sum(nil)
+	mac = hmac.New(sha256.New, key)
+	mac.Write([]byte("Client Key"))
+	clientKey := mac.Sum(nil)
+	storedKey := sha256.Sum256(clientKey)
+	pass := fmt.Sprintf("%s$%v:%s$%s:%s",
+		scramsha256prefix,
+		iterations,
+		base64.StdEncoding.EncodeToString(salt),
+		base64.StdEncoding.EncodeToString(storedKey[:]),
+		base64.StdEncoding.EncodeToString(serverKey),
+	)
+	return pass
+}
+
 // Diff returns diffs between 2 objects
 func Diff(a, b interface{}) []string {
 	return pretty.Diff(a, b)
@@ -75,6 +136,21 @@ func PrettyDiff(a, b interface{}) string {
 	return strings.Join(Diff(a, b), "\n")
 }
 
+// Compare two string slices while ignoring the order of elements
+func IsEqualIgnoreOrder(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	a_copy := make([]string, len(a))
+	b_copy := make([]string, len(b))
+	copy(a_copy, a)
+	copy(b_copy, b)
+	sort.Strings(a_copy)
+	sort.Strings(b_copy)
+
+	return reflect.DeepEqual(a_copy, b_copy)
+}
+
 // SubstractStringSlices finds elements in a that are not in b and return them as a result slice.
 func SubstractStringSlices(a []string, b []string) (result []string, equal bool) {
 	// Slices are assumed to contain unique elements only
@@ -117,6 +193,20 @@ func FindNamedStringSubmatch(r *regexp.Regexp, s string) map[string]string {
 	return res
 }
 
+// SliceContains
+func SliceContains(slice interface{}, item interface{}) bool {
+	s := reflect.ValueOf(slice)
+	if s.Kind() != reflect.Slice {
+		panic("Invalid data-type")
+	}
+	for i := 0; i < s.Len(); i++ {
+		if s.Index(i).Interface() == item {
+			return true
+		}
+	}
+	return false
+}
+
 // MapContains returns true if and only if haystack contains all the keys from the needle with matching corresponding values
 func MapContains(haystack, needle map[string]string) bool {
 	if len(haystack) < len(needle) {
@@ -141,6 +231,100 @@ func Coalesce(val, defaultVal string) string {
 	return val
 }
 
+// CoalesceStrArr returns the first argument if it is not null, otherwise the second one.
+func CoalesceStrArr(val, defaultVal []string) []string {
+	if len(val) == 0 {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceStrMap returns the first argument if it is not null, otherwise the second one.
+func CoalesceStrMap(val, defaultVal map[string]string) map[string]string {
+	if len(val) == 0 {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceInt works like coalesce but for int
+func CoalesceInt(val, defaultVal int) int {
+	if val == 0 {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceInt32 works like coalesce but for *int32
+func CoalesceInt32(val, defaultVal *int32) *int32 {
+	if val == nil {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceUInt32 works like coalesce but for uint32
+func CoalesceUInt32(val, defaultVal uint32) uint32 {
+	if val == 0 {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceInt64 works like coalesce but for int64
+func CoalesceInt64(val, defaultVal int64) int64 {
+	if val == 0 {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceBool works like coalesce but for *bool
+func CoalesceBool(val, defaultVal *bool) *bool {
+	if val == nil {
+		return defaultVal
+	}
+	return val
+}
+
+// CoalesceDuration works like coalesce but for time.Duration
+func CoalesceDuration(val time.Duration, defaultVal string) time.Duration {
+	if val == 0 {
+		duration, err := time.ParseDuration(defaultVal)
+		if err != nil {
+			panic(err)
+		}
+		return duration
+	}
+	return val
+}
+
+// Test if any of the values is nil
+func testNil(values ...*int32) bool {
+	for _, v := range values {
+		if v == nil {
+			return true
+		}
+	}
+
+	return false
+}
+
+// MaxInt32 : Return maximum of two integers provided via pointers. If one value
+// is not defined, return the other one. If both are not defined, result is also
+// undefined, caller needs to check for that.
+func MaxInt32(a, b *int32) *int32 {
+	if testNil(a, b) {
+		return nil
+	}
+
+	if *a > *b {
+		return a
+	}
+
+	return b
+}
+
 // IsSmallerQuantity : checks if first resource is of a smaller quantity than the second
 func IsSmallerQuantity(requestStr, limitStr string) (bool, error) {
 

pkg/util/util_test.go

@@ -12,20 +12,27 @@ import (
 )
 
 var pgUsers = []struct {
-	in  spec.PgUser
-	out string
+	in             spec.PgUser
+	outmd5         string
+	outscramsha256 string
 }{{spec.PgUser{
 	Name:     "test",
 	Password: "password",
 	Flags:    []string{},
 	MemberOf: []string{}},
-	"md587f77988ccb5aa917c93201ba314fcd4"},
+	"md587f77988ccb5aa917c93201ba314fcd4", "SCRAM-SHA-256$4096:c2FsdA==$lF4cRm/Jky763CN4HtxdHnjV4Q8AWTNlKvGmEFFU8IQ=:ub8OgRsftnk2ccDMOt7ffHXNcikRkQkq1lh4xaAqrSw="},
 	{spec.PgUser{
 		Name:     "test",
 		Password: "md592f413f3974bdf3799bb6fecb5f9f2c6",
 		Flags:    []string{},
 		MemberOf: []string{}},
-		"md592f413f3974bdf3799bb6fecb5f9f2c6"}}
+		"md592f413f3974bdf3799bb6fecb5f9f2c6", "md592f413f3974bdf3799bb6fecb5f9f2c6"},
+	{spec.PgUser{
+		Name:     "test",
+		Password: "SCRAM-SHA-256$4096:S1ByZWhvYVV5VDlJNGZoVw==$ozLevu5k0pAQYRrSY+vZhetO6+/oB+qZvuutOdXR94U=:yADwhy0LGloXzh5RaVwLMFyUokwI17VkHVfKVuHu0Zs=",
+		Flags:    []string{},
+		MemberOf: []string{}},
+		"SCRAM-SHA-256$4096:S1ByZWhvYVV5VDlJNGZoVw==$ozLevu5k0pAQYRrSY+vZhetO6+/oB+qZvuutOdXR94U=:yADwhy0LGloXzh5RaVwLMFyUokwI17VkHVfKVuHu0Zs=", "SCRAM-SHA-256$4096:S1ByZWhvYVV5VDlJNGZoVw==$ozLevu5k0pAQYRrSY+vZhetO6+/oB+qZvuutOdXR94U=:yADwhy0LGloXzh5RaVwLMFyUokwI17VkHVfKVuHu0Zs="}}
 
 var prettyDiffTest = []struct {
 	inA interface{}
@@ -36,6 +43,17 @@ var prettyDiffTest = []struct {
 	{[]int{1, 2, 3, 4}, []int{1, 2, 3, 4}, ""},
 }
 
+var isEqualIgnoreOrderTest = []struct {
+	inA      []string
+	inB      []string
+	outEqual bool
+}{
+	{[]string{"a", "b", "c"}, []string{"a", "b", "c"}, true},
+	{[]string{"a", "b", "c"}, []string{"a", "c", "b"}, true},
+	{[]string{"a", "b"}, []string{"a", "c", "b"}, false},
+	{[]string{"a", "b", "c"}, []string{"a", "d", "c"}, false},
+}
+
 var substractTest = []struct {
 	inA      []string
 	inB      []string
@@ -46,6 +64,16 @@ var substractTest = []struct {
 	{[]string{"a", "b", "c", "d"}, []string{"a", "bb", "c", "d"}, []string{"b"}, false},
 }
 
+var sliceContaintsTest = []struct {
+	slice []string
+	item  string
+	out   bool
+}{
+	{[]string{"a", "b", "c"}, "a", true},
+	{[]string{"a", "b", "c"}, "d", false},
+	{[]string{}, "d", false},
+}
+
 var mapContaintsTest = []struct {
 	inA map[string]string
 	inB map[string]string
@@ -107,9 +135,16 @@ func TestNameFromMeta(t *testing.T) {
 
 func TestPGUserPassword(t *testing.T) {
 	for _, tt := range pgUsers {
-		pwd := PGUserPassword(tt.in)
-		if pwd != tt.out {
-			t.Errorf("PgUserPassword expected: %q, got: %q", tt.out, pwd)
+		e := NewEncryptor("md5")
+		pwd := e.PGUserPassword(tt.in)
+		if pwd != tt.outmd5 {
+			t.Errorf("PgUserPassword expected: %q, got: %q", tt.outmd5, pwd)
+		}
+		e = NewEncryptor("scram-sha-256")
+		e.random = func(n int) string { return "salt" }
+		pwd = e.PGUserPassword(tt.in)
+		if pwd != tt.outscramsha256 {
+			t.Errorf("PgUserPassword expected: %q, got: %q", tt.outscramsha256, pwd)
 		}
 	}
 }
@@ -122,6 +157,15 @@ func TestPrettyDiff(t *testing.T) {
 	}
 }
 
+func TestIsEqualIgnoreOrder(t *testing.T) {
+	for _, tt := range isEqualIgnoreOrderTest {
+		actualEqual := IsEqualIgnoreOrder(tt.inA, tt.inB)
+		if actualEqual != tt.outEqual {
+			t.Errorf("IsEqualIgnoreOrder expected: %t, got: %t", tt.outEqual, actualEqual)
+		}
+	}
+}
+
 func TestSubstractSlices(t *testing.T) {
 	for _, tt := range substractTest {
 		actualRes, actualEqual := SubstractStringSlices(tt.inA, tt.inB)
@@ -146,6 +190,15 @@ func TestFindNamedStringSubmatch(t *testing.T) {
 	}
 }
 
+func TestSliceContains(t *testing.T) {
+	for _, tt := range sliceContaintsTest {
+		res := SliceContains(tt.slice, tt.item)
+		if res != tt.out {
+			t.Errorf("SliceContains expected: %#v, got: %#v", tt.out, res)
+		}
+	}
+}
+
 func TestMapContains(t *testing.T) {
 	for _, tt := range mapContaintsTest {
 		res := MapContains(tt.inA, tt.inB)
@@ -166,3 +219,13 @@ func TestIsSmallerQuantity(t *testing.T) {
 		}
 	}
 }
+
+/*
+func TestNiceDiff(t *testing.T) {
+	o := "a\nb\nc\n"
+	n := "b\nd\n"
+	d := nicediff.Diff(o, n, true)
+	t.Log(d)
+	// t.Errorf("Lets see output")
+}
+*/

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,45 @@ 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 */
+	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 +179,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)
 }