resolve conflicts

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

@@ -421,6 +421,43 @@ var PostgresCRDResourceValidation = apiextv1beta1.CustomResourceValidation{
 							},
 						},
 					},
+					"preparedDatabases": {
+						Type: "object",
+						AdditionalProperties: &apiextv1beta1.JSONSchemaPropsOrBool{
+							Schema: &apiextv1beta1.JSONSchemaProps{
+								Type: "object",
+								Properties: map[string]apiextv1beta1.JSONSchemaProps{
+									"defaultUsers": {
+										Type: "boolean",
+									},
+									"extensions": {
+										Type: "object",
+										AdditionalProperties: &apiextv1beta1.JSONSchemaPropsOrBool{
+											Schema: &apiextv1beta1.JSONSchemaProps{
+												Type: "string",
+											},
+										},
+									},
+									"schemas": {
+										Type: "object",
+										AdditionalProperties: &apiextv1beta1.JSONSchemaPropsOrBool{
+											Schema: &apiextv1beta1.JSONSchemaProps{
+												Type: "object",
+												Properties: map[string]apiextv1beta1.JSONSchemaProps{
+													"defaultUsers": {
+														Type: "boolean",
+													},
+													"defaultRoles": {
+														Type: "boolean",
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
 					"replicaLoadBalancer": {
 						Type:        "boolean",
 						Description: "Deprecated",
@@ -513,6 +550,9 @@ var PostgresCRDResourceValidation = apiextv1beta1.CustomResourceValidation{
 							"caFile": {
 								Type: "string",
 							},
+							"caSecretName": {
+								Type: "string",
+							},
 						},
 					},
 					"tolerations": {
@@ -758,6 +798,9 @@ var OperatorConfigCRDResourceValidation = apiextv1beta1.CustomResourceValidation
 					"enable_crd_validation": {
 						Type: "boolean",
 					},
+					"enable_lazy_spilo_upgrade": {
+						Type: "boolean",
+					},
 					"enable_shm_volume": {
 						Type: "boolean",
 					},
@@ -794,6 +837,17 @@ var OperatorConfigCRDResourceValidation = apiextv1beta1.CustomResourceValidation
 							},
 						},
 					},
+					"sidecars": {
+						Type: "array",
+						Items: &apiextv1beta1.JSONSchemaPropsOrArray{
+							Schema: &apiextv1beta1.JSONSchemaProps{
+								Type: "object",
+								AdditionalProperties: &apiextv1beta1.JSONSchemaPropsOrBool{
+									Allows: true,
+								},
+							},
+						},
+					},
 					"workers": {
 						Type:    "integer",
 						Minimum: &min1,

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

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/zalando/postgres-operator/pkg/spec"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -181,18 +182,21 @@ type OperatorLogicalBackupConfiguration struct {
 
 // OperatorConfigurationData defines the operation config
 type OperatorConfigurationData struct {
-	EnableCRDValidation        *bool                              `json:"enable_crd_validation,omitempty"`
-	EtcdHost                   string                             `json:"etcd_host,omitempty"`
-	KubernetesUseConfigMaps    bool                               `json:"kubernetes_use_configmaps,omitempty"`
-	DockerImage                string                             `json:"docker_image,omitempty"`
-	Workers                    uint32                             `json:"workers,omitempty"`
-	MinInstances               int32                              `json:"min_instances,omitempty"`
-	MaxInstances               int32                              `json:"max_instances,omitempty"`
-	ResyncPeriod               Duration                           `json:"resync_period,omitempty"`
-	RepairPeriod               Duration                           `json:"repair_period,omitempty"`
-	SetMemoryRequestToLimit    bool                               `json:"set_memory_request_to_limit,omitempty"`
-	ShmVolume                  *bool                              `json:"enable_shm_volume,omitempty"`
-	Sidecars                   map[string]string                  `json:"sidecar_docker_images,omitempty"`
+	EnableCRDValidation     *bool    `json:"enable_crd_validation,omitempty"`
+	EnableLazySpiloUpgrade  bool     `json:"enable_lazy_spilo_upgrade,omitempty"`
+	EtcdHost                string   `json:"etcd_host,omitempty"`
+	KubernetesUseConfigMaps bool     `json:"kubernetes_use_configmaps,omitempty"`
+	DockerImage             string   `json:"docker_image,omitempty"`
+	Workers                 uint32   `json:"workers,omitempty"`
+	MinInstances            int32    `json:"min_instances,omitempty"`
+	MaxInstances            int32    `json:"max_instances,omitempty"`
+	ResyncPeriod            Duration `json:"resync_period,omitempty"`
+	RepairPeriod            Duration `json:"repair_period,omitempty"`
+	SetMemoryRequestToLimit bool     `json:"set_memory_request_to_limit,omitempty"`
+	ShmVolume               *bool    `json:"enable_shm_volume,omitempty"`
+	// deprecated in favour of SidecarContainers
+	SidecarImages              map[string]string                  `json:"sidecar_docker_images,omitempty"`
+	SidecarContainers          []v1.Container                     `json:"sidecars,omitempty"`
 	PostgresUsersConfiguration PostgresUsersConfiguration         `json:"users"`
 	Kubernetes                 KubernetesMetaConfiguration        `json:"kubernetes"`
 	PostgresPodResources       PostgresPodResourcesDefaults       `json:"postgres_pod_resources"`

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

@@ -50,24 +50,25 @@ type PostgresSpec struct {
 	// load balancers' source ranges are the same for master and replica services
 	AllowedSourceRanges []string `json:"allowedSourceRanges"`
 
-	NumberOfInstances     int32                `json:"numberOfInstances"`
-	Users                 map[string]UserFlags `json:"users"`
-	MaintenanceWindows    []MaintenanceWindow  `json:"maintenanceWindows,omitempty"`
-	Clone                 CloneDescription     `json:"clone"`
-	ClusterName           string               `json:"-"`
-	Databases             map[string]string    `json:"databases,omitempty"`
-	Tolerations           []v1.Toleration      `json:"tolerations,omitempty"`
-	Sidecars              []Sidecar            `json:"sidecars,omitempty"`
-	InitContainers        []v1.Container       `json:"initContainers,omitempty"`
-	PodPriorityClassName  string               `json:"podPriorityClassName,omitempty"`
-	ShmVolume             *bool                `json:"enableShmVolume,omitempty"`
-	EnableLogicalBackup   bool                 `json:"enableLogicalBackup,omitempty"`
-	LogicalBackupSchedule string               `json:"logicalBackupSchedule,omitempty"`
-	StandbyCluster        *StandbyDescription  `json:"standby"`
-	PodAnnotations        map[string]string    `json:"podAnnotations"`
-	ServiceAnnotations    map[string]string    `json:"serviceAnnotations"`
-	TLS                   *TLSDescription      `json:"tls"`
-	AdditionalVolumes     []AdditionalVolume   `json:"additionalVolumes,omitempty"`
+	NumberOfInstances     int32                       `json:"numberOfInstances"`
+	Users                 map[string]UserFlags        `json:"users"`
+	MaintenanceWindows    []MaintenanceWindow         `json:"maintenanceWindows,omitempty"`
+	Clone                 CloneDescription            `json:"clone"`
+	ClusterName           string                      `json:"-"`
+	Databases             map[string]string           `json:"databases,omitempty"`
+	PreparedDatabases     map[string]PreparedDatabase `json:"preparedDatabases,omitempty"`
+	Tolerations           []v1.Toleration             `json:"tolerations,omitempty"`
+	Sidecars              []Sidecar                   `json:"sidecars,omitempty"`
+	InitContainers        []v1.Container              `json:"initContainers,omitempty"`
+	PodPriorityClassName  string                      `json:"podPriorityClassName,omitempty"`
+	ShmVolume             *bool                       `json:"enableShmVolume,omitempty"`
+	EnableLogicalBackup   bool                        `json:"enableLogicalBackup,omitempty"`
+	LogicalBackupSchedule string                      `json:"logicalBackupSchedule,omitempty"`
+	StandbyCluster        *StandbyDescription         `json:"standby"`
+	PodAnnotations        map[string]string           `json:"podAnnotations"`
+	ServiceAnnotations    map[string]string           `json:"serviceAnnotations"`
+	TLS                   *TLSDescription             `json:"tls"`
+	AdditionalVolumes     []AdditionalVolume          `json:"additionalVolumes,omitempty"`
 
 	// deprecated json tags
 	InitContainersOld       []v1.Container `json:"init_containers,omitempty"`
@@ -84,6 +85,19 @@ type PostgresqlList struct {
 	Items []Postgresql `json:"items"`
 }
 
+// PreparedDatabase describes elements to be bootstrapped
+type PreparedDatabase struct {
+	PreparedSchemas map[string]PreparedSchema `json:"schemas,omitempty"`
+	DefaultUsers    bool                      `json:"defaultUsers,omitempty" defaults:"false"`
+	Extensions      map[string]string         `json:"extensions,omitempty"`
+}
+
+// PreparedSchema describes elements to be bootstrapped per schema
+type PreparedSchema struct {
+	DefaultRoles *bool `json:"defaultRoles,omitempty" defaults:"true"`
+	DefaultUsers bool  `json:"defaultUsers,omitempty" defaults:"false"`
+}
+
 // MaintenanceWindow describes the time window when the operator is allowed to do maintenance on a cluster.
 type MaintenanceWindow struct {
 	Everyday  bool
@@ -104,7 +118,7 @@ type AdditionalVolume struct {
 	MountPath        string          `json:"mountPath"`
 	SubPath          string          `json:"subPath"`
 	TargetContainers []string        `json:"targetContainers"`
-	VolumeSource     v1.VolumeSource `json:"volume"`
+	VolumeSource     v1.VolumeSource `json:"volumeSource"`
 }
 
 // PostgresqlParam describes PostgreSQL version and pairs of configuration parameter name - values.
@@ -148,6 +162,7 @@ type TLSDescription struct {
 	CertificateFile string `json:"certificateFile,omitempty"`
 	PrivateKeyFile  string `json:"privateKeyFile,omitempty"`
 	CAFile          string `json:"caFile,omitempty"`
+	CASecretName    string `json:"caSecretName,omitempty"`
 }
 
 // CloneDescription describes which cluster the new should clone and up to which point in time

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

@@ -47,6 +47,28 @@ func (in *AWSGCPConfiguration) DeepCopy() *AWSGCPConfiguration {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *AdditionalVolume) DeepCopyInto(out *AdditionalVolume) {
+	*out = *in
+	if in.TargetContainers != nil {
+		in, out := &in.TargetContainers, &out.TargetContainers
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	in.VolumeSource.DeepCopyInto(&out.VolumeSource)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AdditionalVolume.
+func (in *AdditionalVolume) DeepCopy() *AdditionalVolume {
+	if in == nil {
+		return nil
+	}
+	out := new(AdditionalVolume)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CloneDescription) DeepCopyInto(out *CloneDescription) {
 	*out = *in
@@ -290,13 +312,20 @@ func (in *OperatorConfigurationData) DeepCopyInto(out *OperatorConfigurationData
 		*out = new(bool)
 		**out = **in
 	}
-	if in.Sidecars != nil {
-		in, out := &in.Sidecars, &out.Sidecars
+	if in.SidecarImages != nil {
+		in, out := &in.SidecarImages, &out.SidecarImages
 		*out = make(map[string]string, len(*in))
 		for key, val := range *in {
 			(*out)[key] = val
 		}
 	}
+	if in.SidecarContainers != nil {
+		in, out := &in.SidecarContainers, &out.SidecarContainers
+		*out = make([]corev1.Container, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	out.PostgresUsersConfiguration = in.PostgresUsersConfiguration
 	in.Kubernetes.DeepCopyInto(&out.Kubernetes)
 	out.PostgresPodResources = in.PostgresPodResources
@@ -541,6 +570,13 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
 			(*out)[key] = val
 		}
 	}
+	if in.PreparedDatabases != nil {
+		in, out := &in.PreparedDatabases, &out.PreparedDatabases
+		*out = make(map[string]PreparedDatabase, len(*in))
+		for key, val := range *in {
+			(*out)[key] = *val.DeepCopy()
+		}
+	}
 	if in.Tolerations != nil {
 		in, out := &in.Tolerations, &out.Tolerations
 		*out = make([]corev1.Toleration, len(*in))
@@ -591,6 +627,13 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
 		*out = new(TLSDescription)
 		**out = **in
 	}
+	if in.AdditionalVolumes != nil {
+		in, out := &in.AdditionalVolumes, &out.AdditionalVolumes
+		*out = make([]AdditionalVolume, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	if in.InitContainersOld != nil {
 		in, out := &in.InitContainersOld, &out.InitContainersOld
 		*out = make([]corev1.Container, len(*in))
@@ -727,6 +770,57 @@ func (in *PostgresqlParam) DeepCopy() *PostgresqlParam {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PreparedDatabase) DeepCopyInto(out *PreparedDatabase) {
+	*out = *in
+	if in.PreparedSchemas != nil {
+		in, out := &in.PreparedSchemas, &out.PreparedSchemas
+		*out = make(map[string]PreparedSchema, len(*in))
+		for key, val := range *in {
+			(*out)[key] = *val.DeepCopy()
+		}
+	}
+	if in.Extensions != nil {
+		in, out := &in.Extensions, &out.Extensions
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PreparedDatabase.
+func (in *PreparedDatabase) DeepCopy() *PreparedDatabase {
+	if in == nil {
+		return nil
+	}
+	out := new(PreparedDatabase)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PreparedSchema) DeepCopyInto(out *PreparedSchema) {
+	*out = *in
+	if in.DefaultRoles != nil {
+		in, out := &in.DefaultRoles, &out.DefaultRoles
+		*out = new(bool)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PreparedSchema.
+func (in *PreparedSchema) DeepCopy() *PreparedSchema {
+	if in == nil {
+		return nil
+	}
+	out := new(PreparedSchema)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ResourceDescription) DeepCopyInto(out *ResourceDescription) {
 	*out = *in

pkg/cluster/cluster.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"reflect"
 	"regexp"
+	"strings"
 	"sync"
 	"time"
 
@@ -21,8 +22,11 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/reference"
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
+	"github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/scheme"
 	"github.com/zalando/postgres-operator/pkg/spec"
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/config"
@@ -81,6 +85,7 @@ type Cluster struct {
 	acidv1.Postgresql
 	Config
 	logger           *logrus.Entry
+	eventRecorder    record.EventRecorder
 	patroni          patroni.Interface
 	pgUsers          map[string]spec.PgUser
 	systemUsers      map[string]spec.PgUser
@@ -109,7 +114,7 @@ type compareStatefulsetResult struct {
 }
 
 // New creates a new cluster. This function should be called from a controller.
-func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgresql, logger *logrus.Entry) *Cluster {
+func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgresql, logger *logrus.Entry, eventRecorder record.EventRecorder) *Cluster {
 	deletePropagationPolicy := metav1.DeletePropagationOrphan
 
 	podEventsQueue := cache.NewFIFO(func(obj interface{}) (string, error) {
@@ -140,7 +145,7 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgres
 	cluster.teamsAPIClient = teams.NewTeamsAPI(cfg.OpConfig.TeamsAPIUrl, logger)
 	cluster.oauthTokenGetter = newSecretOauthTokenGetter(&kubeClient, cfg.OpConfig.OAuthTokenSecretName)
 	cluster.patroni = patroni.New(cluster.logger)
-
+	cluster.eventRecorder = eventRecorder
 	return cluster
 }
 
@@ -166,6 +171,16 @@ func (c *Cluster) setProcessName(procName string, args ...interface{}) {
 	}
 }
 
+// GetReference of Postgres CR object
+// i.e. required to emit events to this resource
+func (c *Cluster) GetReference() *v1.ObjectReference {
+	ref, err := reference.GetReference(scheme.Scheme, &c.Postgresql)
+	if err != nil {
+		c.logger.Errorf("could not get reference for Postgresql CR %v/%v: %v", c.Postgresql.Namespace, c.Postgresql.Name, err)
+	}
+	return ref
+}
+
 // SetStatus of Postgres cluster
 // TODO: eventually switch to updateStatus() for kubernetes 1.11 and above
 func (c *Cluster) setStatus(status string) {
@@ -213,6 +228,10 @@ func (c *Cluster) initUsers() error {
 		return fmt.Errorf("could not init infrastructure roles: %v", err)
 	}
 
+	if err := c.initPreparedDatabaseRoles(); err != nil {
+		return fmt.Errorf("could not init default users: %v", err)
+	}
+
 	if err := c.initRobotUsers(); err != nil {
 		return fmt.Errorf("could not init robot users: %v", err)
 	}
@@ -245,6 +264,7 @@ func (c *Cluster) Create() error {
 	}()
 
 	c.setStatus(acidv1.ClusterStatusCreating)
+	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Create", "Started creation of new cluster resources")
 
 	if err = c.enforceMinResourceLimits(&c.Spec); err != nil {
 		return fmt.Errorf("could not enforce minimum resource limits: %v", err)
@@ -263,6 +283,7 @@ func (c *Cluster) Create() error {
 				return fmt.Errorf("could not create %s endpoint: %v", role, err)
 			}
 			c.logger.Infof("endpoint %q has been successfully created", util.NameFromMeta(ep.ObjectMeta))
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Endpoints", "Endpoint %q has been successfully created", util.NameFromMeta(ep.ObjectMeta))
 		}
 
 		if c.Services[role] != nil {
@@ -273,6 +294,7 @@ func (c *Cluster) Create() error {
 			return fmt.Errorf("could not create %s service: %v", role, err)
 		}
 		c.logger.Infof("%s service %q has been successfully created", role, util.NameFromMeta(service.ObjectMeta))
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Services", "The service %q for role %s has been successfully created", util.NameFromMeta(service.ObjectMeta), role)
 	}
 
 	if err = c.initUsers(); err != nil {
@@ -284,6 +306,7 @@ func (c *Cluster) Create() error {
 		return fmt.Errorf("could not create secrets: %v", err)
 	}
 	c.logger.Infof("secrets have been successfully created")
+	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Secrets", "The secrets have been successfully created")
 
 	if c.PodDisruptionBudget != nil {
 		return fmt.Errorf("pod disruption budget already exists in the cluster")
@@ -302,6 +325,7 @@ func (c *Cluster) Create() error {
 		return fmt.Errorf("could not create statefulset: %v", err)
 	}
 	c.logger.Infof("statefulset %q has been successfully created", util.NameFromMeta(ss.ObjectMeta))
+	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "StatefulSet", "Statefulset %q has been successfully created", util.NameFromMeta(ss.ObjectMeta))
 
 	c.logger.Info("waiting for the cluster being ready")
 
@@ -310,6 +334,7 @@ func (c *Cluster) Create() error {
 		return err
 	}
 	c.logger.Infof("pods are ready")
+	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "StatefulSet", "Pods are ready")
 
 	// create database objects unless we are running without pods or disabled
 	// that feature explicitly
@@ -323,6 +348,9 @@ func (c *Cluster) Create() error {
 		if err = c.syncDatabases(); err != nil {
 			return fmt.Errorf("could not sync databases: %v", err)
 		}
+		if err = c.syncPreparedDatabases(); err != nil {
+			return fmt.Errorf("could not sync prepared databases: %v", err)
+		}
 		c.logger.Infof("databases have been successfully created")
 	}
 
@@ -450,6 +478,14 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		}
 	}
 
+	// lazy Spilo update: modify the image in the statefulset itself but let its pods run with the old image
+	// until they are re-created for other reasons, for example node rotation
+	if c.OpConfig.EnableLazySpiloUpgrade && !reflect.DeepEqual(c.Statefulset.Spec.Template.Spec.Containers[0].Image, statefulSet.Spec.Template.Spec.Containers[0].Image) {
+		needsReplace = true
+		needsRollUpdate = false
+		reasons = append(reasons, "lazy Spilo update: new statefulset's pod image doesn't match the current one")
+	}
+
 	if needsRollUpdate || needsReplace {
 		match = false
 	}
@@ -481,8 +517,6 @@ func (c *Cluster) compareContainers(description string, setA, setB []v1.Containe
 	checks := []containerCheck{
 		newCheck("new statefulset %s's %s (index %d) name doesn't match the current one",
 			func(a, b v1.Container) bool { return a.Name != b.Name }),
-		newCheck("new statefulset %s's %s (index %d) image doesn't match the current one",
-			func(a, b v1.Container) bool { return a.Image != b.Image }),
 		newCheck("new statefulset %s's %s (index %d) ports don't match the current one",
 			func(a, b v1.Container) bool { return !reflect.DeepEqual(a.Ports, b.Ports) }),
 		newCheck("new statefulset %s's %s (index %d) resources don't match the current ones",
@@ -493,6 +527,11 @@ func (c *Cluster) compareContainers(description string, setA, setB []v1.Containe
 			func(a, b v1.Container) bool { return !reflect.DeepEqual(a.EnvFrom, b.EnvFrom) }),
 	}
 
+	if !c.OpConfig.EnableLazySpiloUpgrade {
+		checks = append(checks, newCheck("new statefulset %s's %s (index %d) image doesn't match the current one",
+			func(a, b v1.Container) bool { return a.Image != b.Image }))
+	}
+
 	for index, containerA := range setA {
 		containerB := setB[index]
 		for _, check := range checks {
@@ -555,6 +594,7 @@ func (c *Cluster) enforceMinResourceLimits(spec *acidv1.PostgresSpec) error {
 		}
 		if isSmaller {
 			c.logger.Warningf("defined CPU limit %s is below required minimum %s and will be set to it", cpuLimit, minCPULimit)
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "ResourceLimits", "defined CPU limit %s is below required minimum %s and will be set to it", cpuLimit, minCPULimit)
 			spec.Resources.ResourceLimits.CPU = minCPULimit
 		}
 	}
@@ -567,6 +607,7 @@ func (c *Cluster) enforceMinResourceLimits(spec *acidv1.PostgresSpec) error {
 		}
 		if isSmaller {
 			c.logger.Warningf("defined memory limit %s is below required minimum %s and will be set to it", memoryLimit, minMemoryLimit)
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "ResourceLimits", "defined memory limit %s is below required minimum %s and will be set to it", memoryLimit, minMemoryLimit)
 			spec.Resources.ResourceLimits.Memory = minMemoryLimit
 		}
 	}
@@ -598,6 +639,8 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	if oldSpec.Spec.PostgresqlParam.PgVersion != newSpec.Spec.PostgresqlParam.PgVersion { // PG versions comparison
 		c.logger.Warningf("postgresql version change(%q -> %q) has no effect",
 			oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "PostgreSQL", "postgresql version change(%q -> %q) has no effect",
+			oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
 		//we need that hack to generate statefulset with the old version
 		newSpec.Spec.PostgresqlParam.PgVersion = oldSpec.Spec.PostgresqlParam.PgVersion
 	}
@@ -614,7 +657,8 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 
 	// connection pooler needs one system user created, which is done in
 	// initUsers. Check if it needs to be called.
-	sameUsers := reflect.DeepEqual(oldSpec.Spec.Users, newSpec.Spec.Users)
+	sameUsers := reflect.DeepEqual(oldSpec.Spec.Users, newSpec.Spec.Users) &&
+		reflect.DeepEqual(oldSpec.Spec.PreparedDatabases, newSpec.Spec.PreparedDatabases)
 	needConnectionPooler := c.needConnectionPoolerWorker(&newSpec.Spec)
 	if !sameUsers || needConnectionPooler {
 		c.logger.Debugf("syncing secrets")
@@ -731,18 +775,28 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 			c.logger.Errorf("could not sync roles: %v", err)
 			updateFailed = true
 		}
-		if !reflect.DeepEqual(oldSpec.Spec.Databases, newSpec.Spec.Databases) {
+		if !reflect.DeepEqual(oldSpec.Spec.Databases, newSpec.Spec.Databases) ||
+			!reflect.DeepEqual(oldSpec.Spec.PreparedDatabases, newSpec.Spec.PreparedDatabases) {
 			c.logger.Infof("syncing databases")
 			if err := c.syncDatabases(); err != nil {
 				c.logger.Errorf("could not sync databases: %v", err)
 				updateFailed = true
 			}
 		}
+		if !reflect.DeepEqual(oldSpec.Spec.PreparedDatabases, newSpec.Spec.PreparedDatabases) {
+			c.logger.Infof("syncing prepared databases")
+			if err := c.syncPreparedDatabases(); err != nil {
+				c.logger.Errorf("could not sync prepared databases: %v", err)
+				updateFailed = true
+			}
+		}
 	}
 
 	// sync connection pooler
-	if err := c.syncConnectionPooler(oldSpec, newSpec, c.installLookupFunction); err != nil {
-		return fmt.Errorf("could not sync connection pooler: %v", err)
+	if _, err := c.syncConnectionPooler(oldSpec, newSpec,
+		c.installLookupFunction); err != nil {
+		c.logger.Errorf("could not sync connection pooler: %v", err)
+		updateFailed = true
 	}
 
 	return nil
@@ -756,6 +810,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 func (c *Cluster) Delete() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Delete", "Started deletion of new cluster resources")
 
 	// delete the backup job before the stateful set of the cluster to prevent connections to non-existing pods
 	// deleting the cron job also removes pods and batch jobs it created
@@ -783,8 +838,10 @@ func (c *Cluster) Delete() {
 
 	for _, role := range []PostgresRole{Master, Replica} {
 
-		if err := c.deleteEndpoint(role); err != nil {
-			c.logger.Warningf("could not delete %s endpoint: %v", role, err)
+		if !c.patroniKubernetesUseConfigMaps() {
+			if err := c.deleteEndpoint(role); err != nil {
+				c.logger.Warningf("could not delete %s endpoint: %v", role, err)
+			}
 		}
 
 		if err := c.deleteService(role); err != nil {
@@ -910,6 +967,100 @@ func (c *Cluster) initSystemUsers() {
 	}
 }
 
+func (c *Cluster) initPreparedDatabaseRoles() error {
+
+	if c.Spec.PreparedDatabases != nil && len(c.Spec.PreparedDatabases) == 0 { // TODO: add option to disable creating such a default DB
+		c.Spec.PreparedDatabases = map[string]acidv1.PreparedDatabase{strings.Replace(c.Name, "-", "_", -1): {}}
+	}
+
+	// create maps with default roles/users as keys and their membership as values
+	defaultRoles := map[string]string{
+		constants.OwnerRoleNameSuffix:  "",
+		constants.ReaderRoleNameSuffix: "",
+		constants.WriterRoleNameSuffix: constants.ReaderRoleNameSuffix,
+	}
+	defaultUsers := map[string]string{
+		constants.OwnerRoleNameSuffix + constants.UserRoleNameSuffix:  constants.OwnerRoleNameSuffix,
+		constants.ReaderRoleNameSuffix + constants.UserRoleNameSuffix: constants.ReaderRoleNameSuffix,
+		constants.WriterRoleNameSuffix + constants.UserRoleNameSuffix: constants.WriterRoleNameSuffix,
+	}
+
+	for preparedDbName, preparedDB := range c.Spec.PreparedDatabases {
+		// default roles per database
+		if err := c.initDefaultRoles(defaultRoles, "admin", preparedDbName); err != nil {
+			return fmt.Errorf("could not initialize default roles for database %s: %v", preparedDbName, err)
+		}
+		if preparedDB.DefaultUsers {
+			if err := c.initDefaultRoles(defaultUsers, "admin", preparedDbName); err != nil {
+				return fmt.Errorf("could not initialize default roles for database %s: %v", preparedDbName, err)
+			}
+		}
+
+		// default roles per database schema
+		preparedSchemas := preparedDB.PreparedSchemas
+		if len(preparedDB.PreparedSchemas) == 0 {
+			preparedSchemas = map[string]acidv1.PreparedSchema{"data": {DefaultRoles: util.True()}}
+		}
+		for preparedSchemaName, preparedSchema := range preparedSchemas {
+			if preparedSchema.DefaultRoles == nil || *preparedSchema.DefaultRoles {
+				if err := c.initDefaultRoles(defaultRoles,
+					preparedDbName+constants.OwnerRoleNameSuffix,
+					preparedDbName+"_"+preparedSchemaName); err != nil {
+					return fmt.Errorf("could not initialize default roles for database schema %s: %v", preparedSchemaName, err)
+				}
+				if preparedSchema.DefaultUsers {
+					if err := c.initDefaultRoles(defaultUsers,
+						preparedDbName+constants.OwnerRoleNameSuffix,
+						preparedDbName+"_"+preparedSchemaName); err != nil {
+						return fmt.Errorf("could not initialize default users for database schema %s: %v", preparedSchemaName, err)
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (c *Cluster) initDefaultRoles(defaultRoles map[string]string, admin, prefix string) error {
+
+	for defaultRole, inherits := range defaultRoles {
+
+		roleName := prefix + defaultRole
+
+		flags := []string{constants.RoleFlagNoLogin}
+		if defaultRole[len(defaultRole)-5:] == constants.UserRoleNameSuffix {
+			flags = []string{constants.RoleFlagLogin}
+		}
+
+		memberOf := make([]string, 0)
+		if inherits != "" {
+			memberOf = append(memberOf, prefix+inherits)
+		}
+
+		adminRole := ""
+		if strings.Contains(defaultRole, constants.OwnerRoleNameSuffix) {
+			adminRole = admin
+		} else {
+			adminRole = prefix + constants.OwnerRoleNameSuffix
+		}
+
+		newRole := spec.PgUser{
+			Origin:    spec.RoleOriginBootstrap,
+			Name:      roleName,
+			Password:  util.RandomPassword(constants.PasswordLength),
+			Flags:     flags,
+			MemberOf:  memberOf,
+			AdminRole: adminRole,
+		}
+		if currentRole, present := c.pgUsers[roleName]; present {
+			c.pgUsers[roleName] = c.resolveNameConflict(&currentRole, &newRole)
+		} else {
+			c.pgUsers[roleName] = newRole
+		}
+	}
+	return nil
+}
+
 func (c *Cluster) initRobotUsers() error {
 	for username, userFlags := range c.Spec.Users {
 		if !isValidUsername(username) {
@@ -1092,6 +1243,7 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
 
 	var err error
 	c.logger.Debugf("switching over from %q to %q", curMaster.Name, candidate)
+	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Switchover", "Switching over from %q to %q", curMaster.Name, candidate)
 
 	var wg sync.WaitGroup
 
@@ -1118,6 +1270,7 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
 
 	if err = c.patroni.Switchover(curMaster, candidate.Name); err == nil {
 		c.logger.Debugf("successfully switched over from %q to %q", curMaster.Name, candidate)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Switchover", "Successfully switched over from %q to %q", curMaster.Name, candidate)
 		if err = <-podLabelErr; err != nil {
 			err = fmt.Errorf("could not get master pod label: %v", err)
 		}
@@ -1133,6 +1286,7 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
 	// close the label waiting channel no sooner than the waiting goroutine terminates.
 	close(podLabelErr)
 
+	c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Switchover", "Switchover from %q to %q FAILED: %v", curMaster.Name, candidate, err)
 	return err
 
 }
@@ -1160,11 +1314,19 @@ type clusterObjectDelete func(name string) error
 
 func (c *Cluster) deletePatroniClusterObjects() error {
 	// TODO: figure out how to remove leftover patroni objects in other cases
+	var actionsList []simpleActionWithResult
+
 	if !c.patroniUsesKubernetes() {
 		c.logger.Infof("not cleaning up Etcd Patroni objects on cluster delete")
 	}
-	c.logger.Debugf("removing leftover Patroni objects (endpoints, services and configmaps)")
-	for _, deleter := range []simpleActionWithResult{c.deletePatroniClusterEndpoints, c.deletePatroniClusterServices, c.deletePatroniClusterConfigMaps} {
+
+	if !c.patroniKubernetesUseConfigMaps() {
+		actionsList = append(actionsList, c.deletePatroniClusterEndpoints)
+	}
+	actionsList = append(actionsList, c.deletePatroniClusterServices, c.deletePatroniClusterConfigMaps)
+
+	c.logger.Debugf("removing leftover Patroni objects (endpoints / services and configmaps)")
+	for _, deleter := range actionsList {
 		if err := deleter(); err != nil {
 			return err
 		}

pkg/cluster/cluster_test.go

@@ -13,6 +13,8 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
 	"github.com/zalando/postgres-operator/pkg/util/teams"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
 )
 
 const (
@@ -21,6 +23,8 @@ const (
 )
 
 var logger = logrus.New().WithField("test", "cluster")
+var eventRecorder = record.NewFakeRecorder(1)
+
 var cl = New(
 	Config{
 		OpConfig: config.Config{
@@ -32,8 +36,9 @@ var cl = New(
 		},
 	},
 	k8sutil.NewMockKubernetesClient(),
-	acidv1.Postgresql{},
+	acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test", Namespace: "test"}},
 	logger,
+	eventRecorder,
 )
 
 func TestInitRobotUsers(t *testing.T) {
@@ -756,3 +761,89 @@ func TestInitSystemUsers(t *testing.T) {
 		t.Errorf("%s, System users are not allowed to be a connection pool user", testName)
 	}
 }
+
+func TestPreparedDatabases(t *testing.T) {
+	testName := "TestDefaultPreparedDatabase"
+
+	cl.Spec.PreparedDatabases = map[string]acidv1.PreparedDatabase{}
+	cl.initPreparedDatabaseRoles()
+
+	for _, role := range []string{"acid_test_owner", "acid_test_reader", "acid_test_writer",
+		"acid_test_data_owner", "acid_test_data_reader", "acid_test_data_writer"} {
+		if _, exist := cl.pgUsers[role]; !exist {
+			t.Errorf("%s, default role %q for prepared database not present", testName, role)
+		}
+	}
+
+	testName = "TestPreparedDatabaseWithSchema"
+
+	cl.Spec.PreparedDatabases = map[string]acidv1.PreparedDatabase{
+		"foo": {
+			DefaultUsers: true,
+			PreparedSchemas: map[string]acidv1.PreparedSchema{
+				"bar": {
+					DefaultUsers: true,
+				},
+			},
+		},
+	}
+	cl.initPreparedDatabaseRoles()
+
+	for _, role := range []string{
+		"foo_owner", "foo_reader", "foo_writer",
+		"foo_owner_user", "foo_reader_user", "foo_writer_user",
+		"foo_bar_owner", "foo_bar_reader", "foo_bar_writer",
+		"foo_bar_owner_user", "foo_bar_reader_user", "foo_bar_writer_user"} {
+		if _, exist := cl.pgUsers[role]; !exist {
+			t.Errorf("%s, default role %q for prepared database not present", testName, role)
+		}
+	}
+
+	roleTests := []struct {
+		subTest  string
+		role     string
+		memberOf string
+		admin    string
+	}{
+		{
+			subTest:  "Test admin role of owner",
+			role:     "foo_owner",
+			memberOf: "",
+			admin:    "admin",
+		},
+		{
+			subTest:  "Test writer is a member of reader",
+			role:     "foo_writer",
+			memberOf: "foo_reader",
+			admin:    "foo_owner",
+		},
+		{
+			subTest:  "Test reader LOGIN role",
+			role:     "foo_reader_user",
+			memberOf: "foo_reader",
+			admin:    "foo_owner",
+		},
+		{
+			subTest:  "Test schema owner",
+			role:     "foo_bar_owner",
+			memberOf: "",
+			admin:    "foo_owner",
+		},
+		{
+			subTest:  "Test schema writer LOGIN role",
+			role:     "foo_bar_writer_user",
+			memberOf: "foo_bar_writer",
+			admin:    "foo_bar_owner",
+		},
+	}
+
+	for _, tt := range roleTests {
+		user := cl.pgUsers[tt.role]
+		if (tt.memberOf == "" && len(user.MemberOf) > 0) || (tt.memberOf != "" && user.MemberOf[0] != tt.memberOf) {
+			t.Errorf("%s, incorrect membership for default role %q. Expected %q, got %q", tt.subTest, tt.role, tt.memberOf, user.MemberOf[0])
+		}
+		if user.AdminRole != tt.admin {
+			t.Errorf("%s, incorrect admin role for default role %q. Expected %q, got %q", tt.subTest, tt.role, tt.admin, user.AdminRole)
+		}
+	}
+}

pkg/cluster/database.go

@@ -27,9 +27,35 @@ const (
 	 WHERE a.rolname = ANY($1)
 	 ORDER BY 1;`
 
-	getDatabasesSQL        = `SELECT datname, pg_get_userbyid(datdba) AS owner FROM pg_database;`
-	createDatabaseSQL      = `CREATE DATABASE "%s" OWNER "%s";`
-	alterDatabaseOwnerSQL  = `ALTER DATABASE "%s" OWNER TO "%s";`
+	getDatabasesSQL = `SELECT datname, pg_get_userbyid(datdba) AS owner FROM pg_database;`
+	getSchemasSQL   = `SELECT n.nspname AS dbschema FROM pg_catalog.pg_namespace n
+			WHERE n.nspname !~ '^pg_' AND n.nspname <> 'information_schema' ORDER BY 1`
+	getExtensionsSQL = `SELECT e.extname, n.nspname FROM pg_catalog.pg_extension e
+	        LEFT JOIN pg_catalog.pg_namespace n ON n.oid = e.extnamespace ORDER BY 1;`
+
+	createDatabaseSQL       = `CREATE DATABASE "%s" OWNER "%s";`
+	createDatabaseSchemaSQL = `SET ROLE TO "%s"; CREATE SCHEMA IF NOT EXISTS "%s" AUTHORIZATION "%s"`
+	alterDatabaseOwnerSQL   = `ALTER DATABASE "%s" OWNER TO "%s";`
+	createExtensionSQL      = `CREATE EXTENSION IF NOT EXISTS "%s" SCHEMA "%s"`
+	alterExtensionSQL       = `ALTER EXTENSION "%s" SET SCHEMA "%s"`
+
+	globalDefaultPrivilegesSQL = `SET ROLE TO "%s";
+			ALTER DEFAULT PRIVILEGES GRANT USAGE ON SCHEMAS TO "%s","%s";
+			ALTER DEFAULT PRIVILEGES GRANT SELECT ON TABLES TO "%s";
+			ALTER DEFAULT PRIVILEGES GRANT SELECT ON SEQUENCES TO "%s";
+			ALTER DEFAULT PRIVILEGES GRANT INSERT, UPDATE, DELETE ON TABLES TO "%s";
+			ALTER DEFAULT PRIVILEGES GRANT USAGE, UPDATE ON SEQUENCES TO "%s";
+			ALTER DEFAULT PRIVILEGES GRANT EXECUTE ON FUNCTIONS TO "%s","%s";
+			ALTER DEFAULT PRIVILEGES GRANT USAGE ON TYPES TO "%s","%s";`
+	schemaDefaultPrivilegesSQL = `SET ROLE TO "%s";
+			GRANT USAGE ON SCHEMA "%s" TO "%s","%s";
+			ALTER DEFAULT PRIVILEGES IN SCHEMA "%s" GRANT SELECT ON TABLES TO "%s";
+			ALTER DEFAULT PRIVILEGES IN SCHEMA "%s" GRANT SELECT ON SEQUENCES TO "%s";
+			ALTER DEFAULT PRIVILEGES IN SCHEMA "%s" GRANT INSERT, UPDATE, DELETE ON TABLES TO "%s";
+			ALTER DEFAULT PRIVILEGES IN SCHEMA "%s" GRANT USAGE, UPDATE ON SEQUENCES TO "%s";
+			ALTER DEFAULT PRIVILEGES IN SCHEMA "%s" GRANT EXECUTE ON FUNCTIONS TO "%s","%s";
+			ALTER DEFAULT PRIVILEGES IN SCHEMA "%s" GRANT USAGE ON TYPES TO "%s","%s";`
+
 	connectionPoolerLookup = `
 		CREATE SCHEMA IF NOT EXISTS {{.pooler_schema}};
 
@@ -221,43 +247,141 @@ func (c *Cluster) getDatabases() (dbs map[string]string, err error) {
 }
 
 // executeCreateDatabase creates new database with the given owner.
-// The caller is responsible for openinging and closing the database connection.
-func (c *Cluster) executeCreateDatabase(datname, owner string) error {
-	return c.execCreateOrAlterDatabase(datname, owner, createDatabaseSQL,
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeCreateDatabase(databaseName, owner string) error {
+	return c.execCreateOrAlterDatabase(databaseName, owner, createDatabaseSQL,
 		"creating database", "create database")
 }
 
-// executeCreateDatabase changes the owner of the given database.
-// The caller is responsible for openinging and closing the database connection.
-func (c *Cluster) executeAlterDatabaseOwner(datname string, owner string) error {
-	return c.execCreateOrAlterDatabase(datname, owner, alterDatabaseOwnerSQL,
+// executeAlterDatabaseOwner changes the owner of the given database.
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeAlterDatabaseOwner(databaseName string, owner string) error {
+	return c.execCreateOrAlterDatabase(databaseName, owner, alterDatabaseOwnerSQL,
 		"changing owner for database", "alter database owner")
 }
 
-func (c *Cluster) execCreateOrAlterDatabase(datname, owner, statement, doing, operation string) error {
-	if !c.databaseNameOwnerValid(datname, owner) {
+func (c *Cluster) execCreateOrAlterDatabase(databaseName, owner, statement, doing, operation string) error {
+	if !c.databaseNameOwnerValid(databaseName, owner) {
 		return nil
 	}
-	c.logger.Infof("%s %q owner %q", doing, datname, owner)
-	if _, err := c.pgDb.Exec(fmt.Sprintf(statement, datname, owner)); err != nil {
+	c.logger.Infof("%s %q owner %q", doing, databaseName, owner)
+	if _, err := c.pgDb.Exec(fmt.Sprintf(statement, databaseName, owner)); err != nil {
 		return fmt.Errorf("could not execute %s: %v", operation, err)
 	}
 	return nil
 }
 
-func (c *Cluster) databaseNameOwnerValid(datname, owner string) bool {
+func (c *Cluster) databaseNameOwnerValid(databaseName, owner string) bool {
 	if _, ok := c.pgUsers[owner]; !ok {
-		c.logger.Infof("skipping creation of the %q database, user %q does not exist", datname, owner)
+		c.logger.Infof("skipping creation of the %q database, user %q does not exist", databaseName, owner)
 		return false
 	}
 
-	if !databaseNameRegexp.MatchString(datname) {
-		c.logger.Infof("database %q has invalid name", datname)
+	if !databaseNameRegexp.MatchString(databaseName) {
+		c.logger.Infof("database %q has invalid name", databaseName)
 		return false
 	}
 	return true
 }
 
+// getSchemas returns the list of current database schemas
+// The caller is responsible for opening and closing the database connection
+func (c *Cluster) getSchemas() (schemas []string, err error) {
+	var (
+		rows      *sql.Rows
+		dbschemas []string
+	)
+
+	if rows, err = c.pgDb.Query(getSchemasSQL); err != nil {
+		return nil, fmt.Errorf("could not query database schemas: %v", err)
+	}
+
+	defer func() {
+		if err2 := rows.Close(); err2 != nil {
+			if err != nil {
+				err = fmt.Errorf("error when closing query cursor: %v, previous error: %v", err2, err)
+			} else {
+				err = fmt.Errorf("error when closing query cursor: %v", err2)
+			}
+		}
+	}()
+
+	for rows.Next() {
+		var dbschema string
+
+		if err = rows.Scan(&dbschema); err != nil {
+			return nil, fmt.Errorf("error when processing row: %v", err)
+		}
+		dbschemas = append(dbschemas, dbschema)
+	}
+
+	return dbschemas, err
+}
+
+// executeCreateDatabaseSchema creates new database schema with the given owner.
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeCreateDatabaseSchema(databaseName, schemaName, dbOwner string, schemaOwner string) error {
+	return c.execCreateDatabaseSchema(databaseName, schemaName, dbOwner, schemaOwner, createDatabaseSchemaSQL,
+		"creating database schema", "create database schema")
+}
+
+func (c *Cluster) execCreateDatabaseSchema(databaseName, schemaName, dbOwner, schemaOwner, statement, doing, operation string) error {
+	if !c.databaseSchemaNameValid(schemaName) {
+		return nil
+	}
+	c.logger.Infof("%s %q owner %q", doing, schemaName, schemaOwner)
+	if _, err := c.pgDb.Exec(fmt.Sprintf(statement, dbOwner, schemaName, schemaOwner)); err != nil {
+		return fmt.Errorf("could not execute %s: %v", operation, err)
+	}
+
+	// set default privileges for schema
+	c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner, databaseName)
+	if schemaOwner != dbOwner {
+		c.execAlterSchemaDefaultPrivileges(schemaName, dbOwner, databaseName+"_"+schemaName)
+		c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner, databaseName+"_"+schemaName)
+	}
+
+	return nil
+}
+
+func (c *Cluster) databaseSchemaNameValid(schemaName string) bool {
+	if !databaseNameRegexp.MatchString(schemaName) {
+		c.logger.Infof("database schema %q has invalid name", schemaName)
+		return false
+	}
+	return true
+}
+
+func (c *Cluster) execAlterSchemaDefaultPrivileges(schemaName, owner, rolePrefix string) error {
+	if _, err := c.pgDb.Exec(fmt.Sprintf(schemaDefaultPrivilegesSQL, owner,
+		schemaName, rolePrefix+constants.ReaderRoleNameSuffix, rolePrefix+constants.WriterRoleNameSuffix, // schema
+		schemaName, rolePrefix+constants.ReaderRoleNameSuffix, // tables
+		schemaName, rolePrefix+constants.ReaderRoleNameSuffix, // sequences
+		schemaName, rolePrefix+constants.WriterRoleNameSuffix, // tables
+		schemaName, rolePrefix+constants.WriterRoleNameSuffix, // sequences
+		schemaName, rolePrefix+constants.ReaderRoleNameSuffix, rolePrefix+constants.WriterRoleNameSuffix, // types
+		schemaName, rolePrefix+constants.ReaderRoleNameSuffix, rolePrefix+constants.WriterRoleNameSuffix)); err != nil { // functions
+		return fmt.Errorf("could not alter default privileges for database schema %s: %v", schemaName, err)
+	}
+
+	return nil
+}
+
+func (c *Cluster) execAlterGlobalDefaultPrivileges(owner, rolePrefix string) error {
+	if _, err := c.pgDb.Exec(fmt.Sprintf(globalDefaultPrivilegesSQL, owner,
+		rolePrefix+constants.WriterRoleNameSuffix, rolePrefix+constants.ReaderRoleNameSuffix, // schemas
+		rolePrefix+constants.ReaderRoleNameSuffix,                                            // tables
+		rolePrefix+constants.ReaderRoleNameSuffix,                                            // sequences
+		rolePrefix+constants.WriterRoleNameSuffix,                                            // tables
+		rolePrefix+constants.WriterRoleNameSuffix,                                            // sequences
+		rolePrefix+constants.ReaderRoleNameSuffix, rolePrefix+constants.WriterRoleNameSuffix, // types
+		rolePrefix+constants.ReaderRoleNameSuffix, rolePrefix+constants.WriterRoleNameSuffix)); err != nil { // functions
+		return fmt.Errorf("could not alter default privileges for database %s: %v", rolePrefix, err)
+	}
+
+	return nil
+}
+
 func makeUserFlags(rolsuper, rolinherit, rolcreaterole, rolcreatedb, rolcanlogin bool) (result []string) {
 	if rolsuper {
 		result = append(result, constants.RoleFlagSuperuser)
@@ -278,8 +402,67 @@ func makeUserFlags(rolsuper, rolinherit, rolcreaterole, rolcreatedb, rolcanlogin
 	return result
 }
 
-// Creates a connection pooler credentials lookup function in every database to
-// perform remote authentication.
+// getExtension returns the list of current database extensions
+// The caller is responsible for opening and closing the database connection
+func (c *Cluster) getExtensions() (dbExtensions map[string]string, err error) {
+	var (
+		rows *sql.Rows
+	)
+
+	if rows, err = c.pgDb.Query(getExtensionsSQL); err != nil {
+		return nil, fmt.Errorf("could not query database extensions: %v", err)
+	}
+
+	defer func() {
+		if err2 := rows.Close(); err2 != nil {
+			if err != nil {
+				err = fmt.Errorf("error when closing query cursor: %v, previous error: %v", err2, err)
+			} else {
+				err = fmt.Errorf("error when closing query cursor: %v", err2)
+			}
+		}
+	}()
+
+	dbExtensions = make(map[string]string)
+
+	for rows.Next() {
+		var extension, schema string
+
+		if err = rows.Scan(&extension, &schema); err != nil {
+			return nil, fmt.Errorf("error when processing row: %v", err)
+		}
+		dbExtensions[extension] = schema
+	}
+
+	return dbExtensions, err
+}
+
+// executeCreateExtension creates new extension in the given schema.
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeCreateExtension(extName, schemaName string) error {
+	return c.execCreateOrAlterExtension(extName, schemaName, createExtensionSQL,
+		"creating extension", "create extension")
+}
+
+// executeAlterExtension changes the schema of the given extension.
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeAlterExtension(extName, schemaName string) error {
+	return c.execCreateOrAlterExtension(extName, schemaName, alterExtensionSQL,
+		"changing schema for extension", "alter extension schema")
+}
+
+func (c *Cluster) execCreateOrAlterExtension(extName, schemaName, statement, doing, operation string) error {
+
+	c.logger.Infof("%s %q schema %q", doing, extName, schemaName)
+	if _, err := c.pgDb.Exec(fmt.Sprintf(statement, extName, schemaName)); err != nil {
+		return fmt.Errorf("could not execute %s: %v", operation, err)
+	}
+
+	return nil
+}
+
+// Creates a connection pool credentials lookup function in every database to
+// perform remote authentification.
 func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
 	var stmtBytes bytes.Buffer
 	c.logger.Info("Installing lookup function")
@@ -305,7 +488,7 @@ func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
 
 	templater := template.Must(template.New("sql").Parse(connectionPoolerLookup))
 
-	for dbname, _ := range currentDatabases {
+	for dbname := range currentDatabases {
 		if dbname == "template0" || dbname == "template1" {
 			continue
 		}

pkg/cluster/k8sres.go

@@ -462,8 +462,7 @@ func generateContainer(
 }
 
 func generateSidecarContainers(sidecars []acidv1.Sidecar,
-	volumeMounts []v1.VolumeMount, defaultResources acidv1.Resources,
-	superUserName string, credentialsSecretName string, logger *logrus.Entry) ([]v1.Container, error) {
+	defaultResources acidv1.Resources, startIndex int, logger *logrus.Entry) ([]v1.Container, error) {
 
 	if len(sidecars) > 0 {
 		result := make([]v1.Container, 0)
@@ -482,7 +481,7 @@ func generateSidecarContainers(sidecars []acidv1.Sidecar,
 				return nil, err
 			}
 
-			sc := getSidecarContainer(sidecar, index, volumeMounts, resources, superUserName, credentialsSecretName, logger)
+			sc := getSidecarContainer(sidecar, startIndex+index, resources)
 			result = append(result, *sc)
 		}
 		return result, nil
@@ -490,6 +489,55 @@ func generateSidecarContainers(sidecars []acidv1.Sidecar,
 	return nil, nil
 }
 
+// adds common fields to sidecars
+func patchSidecarContainers(in []v1.Container, volumeMounts []v1.VolumeMount, superUserName string, credentialsSecretName string, logger *logrus.Entry) []v1.Container {
+	result := []v1.Container{}
+
+	for _, container := range in {
+		container.VolumeMounts = append(container.VolumeMounts, volumeMounts...)
+		env := []v1.EnvVar{
+			{
+				Name: "POD_NAME",
+				ValueFrom: &v1.EnvVarSource{
+					FieldRef: &v1.ObjectFieldSelector{
+						APIVersion: "v1",
+						FieldPath:  "metadata.name",
+					},
+				},
+			},
+			{
+				Name: "POD_NAMESPACE",
+				ValueFrom: &v1.EnvVarSource{
+					FieldRef: &v1.ObjectFieldSelector{
+						APIVersion: "v1",
+						FieldPath:  "metadata.namespace",
+					},
+				},
+			},
+			{
+				Name:  "POSTGRES_USER",
+				Value: superUserName,
+			},
+			{
+				Name: "POSTGRES_PASSWORD",
+				ValueFrom: &v1.EnvVarSource{
+					SecretKeyRef: &v1.SecretKeySelector{
+						LocalObjectReference: v1.LocalObjectReference{
+							Name: credentialsSecretName,
+						},
+						Key: "password",
+					},
+				},
+			},
+		}
+		mergedEnv := append(container.Env, env...)
+		container.Env = deduplicateEnvVars(mergedEnv, container.Name, logger)
+		result = append(result, container)
+	}
+
+	return result
+}
+
 // Check whether or not we're requested to mount an shm volume,
 // taking into account that PostgreSQL manifest has precedence.
 func mountShmVolumeNeeded(opConfig config.Config, spec *acidv1.PostgresSpec) *bool {
@@ -519,7 +567,6 @@ func (c *Cluster) generatePodTemplate(
 	podAntiAffinityTopologyKey string,
 	additionalSecretMount string,
 	additionalSecretMountPath string,
-	volumes []v1.Volume,
 	additionalVolumes []acidv1.AdditionalVolume,
 ) (*v1.PodTemplateSpec, error) {
 
@@ -539,7 +586,6 @@ func (c *Cluster) generatePodTemplate(
 		InitContainers:                initContainers,
 		Tolerations:                   *tolerationsSpec,
 		SecurityContext:               &securityContext,
-		Volumes:                       volumes,
 	}
 
 	if shmVolume != nil && *shmVolume {
@@ -726,58 +772,18 @@ func deduplicateEnvVars(input []v1.EnvVar, containerName string, logger *logrus.
 	return result
 }
 
-func getSidecarContainer(sidecar acidv1.Sidecar, index int, volumeMounts []v1.VolumeMount,
-	resources *v1.ResourceRequirements, superUserName string, credentialsSecretName string, logger *logrus.Entry) *v1.Container {
+func getSidecarContainer(sidecar acidv1.Sidecar, index int, resources *v1.ResourceRequirements) *v1.Container {
 	name := sidecar.Name
 	if name == "" {
 		name = fmt.Sprintf("sidecar-%d", index)
 	}
 
-	env := []v1.EnvVar{
-		{
-			Name: "POD_NAME",
-			ValueFrom: &v1.EnvVarSource{
-				FieldRef: &v1.ObjectFieldSelector{
-					APIVersion: "v1",
-					FieldPath:  "metadata.name",
-				},
-			},
-		},
-		{
-			Name: "POD_NAMESPACE",
-			ValueFrom: &v1.EnvVarSource{
-				FieldRef: &v1.ObjectFieldSelector{
-					APIVersion: "v1",
-					FieldPath:  "metadata.namespace",
-				},
-			},
-		},
-		{
-			Name:  "POSTGRES_USER",
-			Value: superUserName,
-		},
-		{
-			Name: "POSTGRES_PASSWORD",
-			ValueFrom: &v1.EnvVarSource{
-				SecretKeyRef: &v1.SecretKeySelector{
-					LocalObjectReference: v1.LocalObjectReference{
-						Name: credentialsSecretName,
-					},
-					Key: "password",
-				},
-			},
-		},
-	}
-	if len(sidecar.Env) > 0 {
-		env = append(env, sidecar.Env...)
-	}
 	return &v1.Container{
 		Name:            name,
 		Image:           sidecar.DockerImage,
 		ImagePullPolicy: v1.PullIfNotPresent,
 		Resources:       *resources,
-		VolumeMounts:    volumeMounts,
-		Env:             deduplicateEnvVars(env, name, logger),
+		Env:             sidecar.Env,
 		Ports:           sidecar.Ports,
 	}
 }
@@ -854,7 +860,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		sidecarContainers   []v1.Container
 		podTemplate         *v1.PodTemplateSpec
 		volumeClaimTemplate *v1.PersistentVolumeClaim
-		volumes             []v1.Volume
+		additionalVolumes   = spec.AdditionalVolumes
 	)
 
 	// Improve me. Please.
@@ -1007,8 +1013,10 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		// this is combined with the FSGroup in the section above
 		// to give read access to the postgres user
 		defaultMode := int32(0640)
-		volumes = append(volumes, v1.Volume{
-			Name: "tls-secret",
+		mountPath := "/tls"
+		additionalVolumes = append(additionalVolumes, acidv1.AdditionalVolume{
+			Name:      spec.TLS.SecretName,
+			MountPath: mountPath,
 			VolumeSource: v1.VolumeSource{
 				Secret: &v1.SecretVolumeSource{
 					SecretName:  spec.TLS.SecretName,
@@ -1017,13 +1025,6 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 			},
 		})
 
-		mountPath := "/tls"
-		volumeMounts = append(volumeMounts, v1.VolumeMount{
-			MountPath: mountPath,
-			Name:      "tls-secret",
-			ReadOnly:  true,
-		})
-
 		// use the same filenames as Secret resources by default
 		certFile := ensurePath(spec.TLS.CertificateFile, mountPath, "tls.crt")
 		privateKeyFile := ensurePath(spec.TLS.PrivateKeyFile, mountPath, "tls.key")
@@ -1034,11 +1035,31 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		)
 
 		if spec.TLS.CAFile != "" {
-			caFile := ensurePath(spec.TLS.CAFile, mountPath, "")
+			// support scenario when the ca.crt resides in a different secret, diff path
+			mountPathCA := mountPath
+			if spec.TLS.CASecretName != "" {
+				mountPathCA = mountPath + "ca"
+			}
+
+			caFile := ensurePath(spec.TLS.CAFile, mountPathCA, "")
 			spiloEnvVars = append(
 				spiloEnvVars,
 				v1.EnvVar{Name: "SSL_CA_FILE", Value: caFile},
 			)
+
+			// the ca file from CASecretName secret takes priority
+			if spec.TLS.CASecretName != "" {
+				additionalVolumes = append(additionalVolumes, acidv1.AdditionalVolume{
+					Name:      spec.TLS.CASecretName,
+					MountPath: mountPathCA,
+					VolumeSource: v1.VolumeSource{
+						Secret: &v1.SecretVolumeSource{
+							SecretName:  spec.TLS.CASecretName,
+							DefaultMode: &defaultMode,
+						},
+					},
+				})
+			}
 		}
 	}
 
@@ -1052,37 +1073,63 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		c.OpConfig.Resources.SpiloPrivileged,
 	)
 
-	// resolve conflicts between operator-global and per-cluster sidecars
-	sideCars := c.mergeSidecars(spec.Sidecars)
+	// generate container specs for sidecars specified in the cluster manifest
+	clusterSpecificSidecars := []v1.Container{}
+	if spec.Sidecars != nil && len(spec.Sidecars) > 0 {
+		// warn if sidecars are defined, but globally disabled (does not apply to globally defined sidecars)
+		if c.OpConfig.EnableSidecars != nil && !(*c.OpConfig.EnableSidecars) {
+			c.logger.Warningf("sidecars specified but disabled in configuration - next statefulset creation would fail")
+		}
 
-	resourceRequirementsScalyrSidecar := makeResources(
-		c.OpConfig.ScalyrCPURequest,
-		c.OpConfig.ScalyrMemoryRequest,
-		c.OpConfig.ScalyrCPULimit,
-		c.OpConfig.ScalyrMemoryLimit,
-	)
+		if clusterSpecificSidecars, err = generateSidecarContainers(spec.Sidecars, defaultResources, 0, c.logger); err != nil {
+			return nil, fmt.Errorf("could not generate sidecar containers: %v", err)
+		}
+	}
+
+	// decrapted way of providing global sidecars
+	var globalSidecarContainersByDockerImage []v1.Container
+	var globalSidecarsByDockerImage []acidv1.Sidecar
+	for name, dockerImage := range c.OpConfig.SidecarImages {
+		globalSidecarsByDockerImage = append(globalSidecarsByDockerImage, acidv1.Sidecar{Name: name, DockerImage: dockerImage})
+	}
+	if globalSidecarContainersByDockerImage, err = generateSidecarContainers(globalSidecarsByDockerImage, defaultResources, len(clusterSpecificSidecars), c.logger); err != nil {
+		return nil, fmt.Errorf("could not generate sidecar containers: %v", err)
+	}
+	// make the resulting list reproducible
+	// c.OpConfig.SidecarImages is unsorted by Golang definition
+	// .Name is unique
+	sort.Slice(globalSidecarContainersByDockerImage, func(i, j int) bool {
+		return globalSidecarContainersByDockerImage[i].Name < globalSidecarContainersByDockerImage[j].Name
+	})
 
 	// generate scalyr sidecar container
-	if scalyrSidecar :=
+	var scalyrSidecars []v1.Container
+	if scalyrSidecar, err :=
 		generateScalyrSidecarSpec(c.Name,
 			c.OpConfig.ScalyrAPIKey,
 			c.OpConfig.ScalyrServerURL,
 			c.OpConfig.ScalyrImage,
-			&resourceRequirementsScalyrSidecar, c.logger); scalyrSidecar != nil {
-		sideCars = append(sideCars, *scalyrSidecar)
+			c.OpConfig.ScalyrCPURequest,
+			c.OpConfig.ScalyrMemoryRequest,
+			c.OpConfig.ScalyrCPULimit,
+			c.OpConfig.ScalyrMemoryLimit,
+			defaultResources,
+			c.logger); err != nil {
+		return nil, fmt.Errorf("could not generate Scalyr sidecar: %v", err)
+	} else {
+		if scalyrSidecar != nil {
+			scalyrSidecars = append(scalyrSidecars, *scalyrSidecar)
+		}
 	}
 
-	// generate sidecar containers
-	if sideCars != nil && len(sideCars) > 0 {
-		if c.OpConfig.EnableSidecars != nil && !(*c.OpConfig.EnableSidecars) {
-			c.logger.Warningf("sidecars specified but disabled in configuration - next statefulset creation would fail")
-		}
-		if sidecarContainers, err = generateSidecarContainers(sideCars, volumeMounts, defaultResources,
-			c.OpConfig.SuperUsername, c.credentialSecretName(c.OpConfig.SuperUsername), c.logger); err != nil {
-			return nil, fmt.Errorf("could not generate sidecar containers: %v", err)
-		}
+	sidecarContainers, conflicts := mergeContainers(clusterSpecificSidecars, c.Config.OpConfig.SidecarContainers, globalSidecarContainersByDockerImage, scalyrSidecars)
+	for containerName := range conflicts {
+		c.logger.Warningf("a sidecar is specified twice. Ignoring sidecar %q in favor of %q with high a precendence",
+			containerName, containerName)
 	}
 
+	sidecarContainers = patchSidecarContainers(sidecarContainers, volumeMounts, c.OpConfig.SuperUsername, c.credentialSecretName(c.OpConfig.SuperUsername), c.logger)
+
 	tolerationSpec := tolerations(&spec.Tolerations, c.OpConfig.PodToleration)
 	effectivePodPriorityClassName := util.Coalesce(spec.PodPriorityClassName, c.OpConfig.PodPriorityClassName)
 
@@ -1108,8 +1155,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		c.OpConfig.PodAntiAffinityTopologyKey,
 		c.OpConfig.AdditionalSecretMount,
 		c.OpConfig.AdditionalSecretMountPath,
-		volumes,
-		spec.AdditionalVolumes)
+		additionalVolumes)
 
 	if err != nil {
 		return nil, fmt.Errorf("could not generate pod template: %v", err)
@@ -1176,57 +1222,44 @@ func (c *Cluster) generatePodAnnotations(spec *acidv1.PostgresSpec) map[string]s
 }
 
 func generateScalyrSidecarSpec(clusterName, APIKey, serverURL, dockerImage string,
-	containerResources *acidv1.Resources, logger *logrus.Entry) *acidv1.Sidecar {
+	scalyrCPURequest string, scalyrMemoryRequest string, scalyrCPULimit string, scalyrMemoryLimit string,
+	defaultResources acidv1.Resources, logger *logrus.Entry) (*v1.Container, error) {
 	if APIKey == "" || dockerImage == "" {
 		if APIKey == "" && dockerImage != "" {
 			logger.Warning("Not running Scalyr sidecar: SCALYR_API_KEY must be defined")
 		}
-		return nil
+		return nil, nil
 	}
-	scalarSpec := &acidv1.Sidecar{
-		Name:        "scalyr-sidecar",
-		DockerImage: dockerImage,
-		Env: []v1.EnvVar{
-			{
-				Name:  "SCALYR_API_KEY",
-				Value: APIKey,
-			},
-			{
-				Name:  "SCALYR_SERVER_HOST",
-				Value: clusterName,
-			},
+	resourcesScalyrSidecar := makeResources(
+		scalyrCPURequest,
+		scalyrMemoryRequest,
+		scalyrCPULimit,
+		scalyrMemoryLimit,
+	)
+	resourceRequirementsScalyrSidecar, err := generateResourceRequirements(resourcesScalyrSidecar, defaultResources)
+	if err != nil {
+		return nil, fmt.Errorf("invalid resources for Scalyr sidecar: %v", err)
+	}
+	env := []v1.EnvVar{
+		{
+			Name:  "SCALYR_API_KEY",
+			Value: APIKey,
+		},
+		{
+			Name:  "SCALYR_SERVER_HOST",
+			Value: clusterName,
 		},
-		Resources: *containerResources,
 	}
 	if serverURL != "" {
-		scalarSpec.Env = append(scalarSpec.Env, v1.EnvVar{Name: "SCALYR_SERVER_URL", Value: serverURL})
+		env = append(env, v1.EnvVar{Name: "SCALYR_SERVER_URL", Value: serverURL})
 	}
-	return scalarSpec
-}
-
-// mergeSidecar merges globally-defined sidecars with those defined in the cluster manifest
-func (c *Cluster) mergeSidecars(sidecars []acidv1.Sidecar) []acidv1.Sidecar {
-	globalSidecarsToSkip := map[string]bool{}
-	result := make([]acidv1.Sidecar, 0)
-
-	for i, sidecar := range sidecars {
-		dockerImage, ok := c.OpConfig.Sidecars[sidecar.Name]
-		if ok {
-			if dockerImage != sidecar.DockerImage {
-				c.logger.Warningf("merging definitions for sidecar %q: "+
-					"ignoring %q in the global scope in favor of %q defined in the cluster",
-					sidecar.Name, dockerImage, sidecar.DockerImage)
-			}
-			globalSidecarsToSkip[sidecar.Name] = true
-		}
-		result = append(result, sidecars[i])
-	}
-	for name, dockerImage := range c.OpConfig.Sidecars {
-		if !globalSidecarsToSkip[name] {
-			result = append(result, acidv1.Sidecar{Name: name, DockerImage: dockerImage})
-		}
-	}
-	return result
+	return &v1.Container{
+		Name:            "scalyr-sidecar",
+		Image:           dockerImage,
+		Env:             env,
+		ImagePullPolicy: v1.PullIfNotPresent,
+		Resources:       *resourceRequirementsScalyrSidecar,
+	}, nil
 }
 
 func (c *Cluster) getNumberOfInstances(spec *acidv1.PostgresSpec) int32 {
@@ -1437,6 +1470,13 @@ func (c *Cluster) generateSingleUserSecret(namespace string, pgUser spec.PgUser)
 		return nil
 	}
 
+	//skip NOLOGIN users
+	for _, flag := range pgUser.Flags {
+		if flag == constants.RoleFlagNoLogin {
+			return nil
+		}
+	}
+
 	username := pgUser.Name
 	secret := v1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1614,11 +1654,11 @@ func (c *Cluster) generateCloneEnvironment(description *acidv1.CloneDescription)
 			c.logger.Info(msg, description.S3WalPath)
 
 			envs := []v1.EnvVar{
-				v1.EnvVar{
+				{
 					Name:  "CLONE_WAL_S3_BUCKET",
 					Value: c.OpConfig.WALES3Bucket,
 				},
-				v1.EnvVar{
+				{
 					Name:  "CLONE_WAL_BUCKET_SCOPE_SUFFIX",
 					Value: getBucketScopeSuffix(description.UID),
 				},
@@ -1790,9 +1830,8 @@ func (c *Cluster) generateLogicalBackupJob() (*batchv1beta1.CronJob, error) {
 		"",
 		c.OpConfig.AdditionalSecretMount,
 		c.OpConfig.AdditionalSecretMountPath,
-		nil,
 		[]acidv1.AdditionalVolume{}); err != nil {
-			return nil, fmt.Errorf("could not generate pod template for logical backup pod: %v", err)
+		return nil, fmt.Errorf("could not generate pod template for logical backup pod: %v", err)
 	}
 
 	// overwrite specific params of logical backups pods

pkg/cluster/k8sres_test.go

@@ -18,6 +18,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
@@ -37,7 +38,7 @@ func TestGenerateSpiloJSONConfiguration(t *testing.T) {
 					ReplicationUsername: replicationUserName,
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	testName := "TestGenerateSpiloConfig"
 	tests := []struct {
@@ -102,7 +103,7 @@ func TestCreateLoadBalancerLogic(t *testing.T) {
 					ReplicationUsername: replicationUserName,
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	testName := "TestCreateLoadBalancerLogic"
 	tests := []struct {
@@ -164,7 +165,8 @@ func TestGeneratePodDisruptionBudget(t *testing.T) {
 				acidv1.Postgresql{
 					ObjectMeta: metav1.ObjectMeta{Name: "myapp-database", Namespace: "myapp"},
 					Spec:       acidv1.PostgresSpec{TeamID: "myapp", NumberOfInstances: 3}},
-				logger),
+				logger,
+				eventRecorder),
 			policyv1beta1.PodDisruptionBudget{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "postgres-myapp-database-pdb",
@@ -187,7 +189,8 @@ func TestGeneratePodDisruptionBudget(t *testing.T) {
 				acidv1.Postgresql{
 					ObjectMeta: metav1.ObjectMeta{Name: "myapp-database", Namespace: "myapp"},
 					Spec:       acidv1.PostgresSpec{TeamID: "myapp", NumberOfInstances: 0}},
-				logger),
+				logger,
+				eventRecorder),
 			policyv1beta1.PodDisruptionBudget{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "postgres-myapp-database-pdb",
@@ -210,7 +213,8 @@ func TestGeneratePodDisruptionBudget(t *testing.T) {
 				acidv1.Postgresql{
 					ObjectMeta: metav1.ObjectMeta{Name: "myapp-database", Namespace: "myapp"},
 					Spec:       acidv1.PostgresSpec{TeamID: "myapp", NumberOfInstances: 3}},
-				logger),
+				logger,
+				eventRecorder),
 			policyv1beta1.PodDisruptionBudget{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "postgres-myapp-database-pdb",
@@ -233,7 +237,8 @@ func TestGeneratePodDisruptionBudget(t *testing.T) {
 				acidv1.Postgresql{
 					ObjectMeta: metav1.ObjectMeta{Name: "myapp-database", Namespace: "myapp"},
 					Spec:       acidv1.PostgresSpec{TeamID: "myapp", NumberOfInstances: 3}},
-				logger),
+				logger,
+				eventRecorder),
 			policyv1beta1.PodDisruptionBudget{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "postgres-myapp-database-databass-budget",
@@ -368,7 +373,7 @@ func TestCloneEnv(t *testing.T) {
 					ReplicationUsername: replicationUserName,
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	for _, tt := range tests {
 		envs := cluster.generateCloneEnvironment(tt.cloneOpts)
@@ -502,7 +507,7 @@ func TestGetPgVersion(t *testing.T) {
 					ReplicationUsername: replicationUserName,
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	for _, tt := range tests {
 		pgVersion, err := cluster.getNewPgVersion(tt.pgContainer, tt.newPgVersion)
@@ -678,7 +683,7 @@ func TestConnectionPoolerPodSpec(t *testing.T) {
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	var clusterNoDefaultRes = New(
 		Config{
@@ -690,7 +695,7 @@ func TestConnectionPoolerPodSpec(t *testing.T) {
 				},
 				ConnectionPooler: config.ConnectionPooler{},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	noCheck := func(cluster *Cluster, podSpec *v1.PodTemplateSpec) error { return nil }
 
@@ -803,7 +808,7 @@ func TestConnectionPoolerDeploymentSpec(t *testing.T) {
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 	cluster.Statefulset = &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "test-sts",
@@ -904,7 +909,7 @@ func TestConnectionPoolerServiceSpec(t *testing.T) {
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 	cluster.Statefulset = &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "test-sts",
@@ -961,6 +966,7 @@ func TestTLS(t *testing.T) {
 	var spec acidv1.PostgresSpec
 	var cluster *Cluster
 	var spiloFSGroup = int64(103)
+	var additionalVolumes = spec.AdditionalVolumes
 
 	makeSpec := func(tls acidv1.TLSDescription) acidv1.PostgresSpec {
 		return acidv1.PostgresSpec{
@@ -989,7 +995,7 @@ func TestTLS(t *testing.T) {
 					SpiloFSGroup: &spiloFSGroup,
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 	spec = makeSpec(acidv1.TLSDescription{SecretName: "my-secret", CAFile: "ca.crt"})
 	s, err := cluster.generateStatefulSet(&spec)
 	if err != nil {
@@ -1000,8 +1006,20 @@ func TestTLS(t *testing.T) {
 	assert.Equal(t, &fsGroup, s.Spec.Template.Spec.SecurityContext.FSGroup, "has a default FSGroup assigned")
 
 	defaultMode := int32(0640)
+	mountPath := "/tls"
+	additionalVolumes = append(additionalVolumes, acidv1.AdditionalVolume{
+		Name:      spec.TLS.SecretName,
+		MountPath: mountPath,
+		VolumeSource: v1.VolumeSource{
+			Secret: &v1.SecretVolumeSource{
+				SecretName:  spec.TLS.SecretName,
+				DefaultMode: &defaultMode,
+			},
+		},
+	})
+
 	volume := v1.Volume{
-		Name: "tls-secret",
+		Name: "my-secret",
 		VolumeSource: v1.VolumeSource{
 			Secret: &v1.SecretVolumeSource{
 				SecretName:  "my-secret",
@@ -1013,8 +1031,7 @@ func TestTLS(t *testing.T) {
 
 	assert.Contains(t, s.Spec.Template.Spec.Containers[0].VolumeMounts, v1.VolumeMount{
 		MountPath: "/tls",
-		Name:      "tls-secret",
-		ReadOnly:  true,
+		Name:      "my-secret",
 	}, "the volume gets mounted in /tls")
 
 	assert.Contains(t, s.Spec.Template.Spec.Containers[0].Env, v1.EnvVar{Name: "SSL_CERTIFICATE_FILE", Value: "/tls/tls.crt"})
@@ -1100,7 +1117,7 @@ func TestAdditionalVolume(t *testing.T) {
 					ReplicationUsername: replicationUserName,
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	for _, tt := range tests {
 		// Test with additional volume mounted in all containers
@@ -1190,3 +1207,201 @@ func TestAdditionalVolume(t *testing.T) {
 		}
 	}
 }
+
+// inject sidecars through all available mechanisms and check the resulting container specs
+func TestSidecars(t *testing.T) {
+	var err error
+	var spec acidv1.PostgresSpec
+	var cluster *Cluster
+
+	generateKubernetesResources := func(cpuRequest string, cpuLimit string, memoryRequest string, memoryLimit string) v1.ResourceRequirements {
+		parsedCPURequest, err := resource.ParseQuantity(cpuRequest)
+		assert.NoError(t, err)
+		parsedCPULimit, err := resource.ParseQuantity(cpuLimit)
+		assert.NoError(t, err)
+		parsedMemoryRequest, err := resource.ParseQuantity(memoryRequest)
+		assert.NoError(t, err)
+		parsedMemoryLimit, err := resource.ParseQuantity(memoryLimit)
+		assert.NoError(t, err)
+		return v1.ResourceRequirements{
+			Requests: v1.ResourceList{
+				v1.ResourceCPU:    parsedCPURequest,
+				v1.ResourceMemory: parsedMemoryRequest,
+			},
+			Limits: v1.ResourceList{
+				v1.ResourceCPU:    parsedCPULimit,
+				v1.ResourceMemory: parsedMemoryLimit,
+			},
+		}
+	}
+
+	spec = acidv1.PostgresSpec{
+		TeamID: "myapp", NumberOfInstances: 1,
+		Resources: acidv1.Resources{
+			ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+			ResourceLimits:   acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+		},
+		Volume: acidv1.Volume{
+			Size: "1G",
+		},
+		Sidecars: []acidv1.Sidecar{
+			acidv1.Sidecar{
+				Name: "cluster-specific-sidecar",
+			},
+			acidv1.Sidecar{
+				Name: "cluster-specific-sidecar-with-resources",
+				Resources: acidv1.Resources{
+					ResourceRequests: acidv1.ResourceDescription{CPU: "210m", Memory: "0.8Gi"},
+					ResourceLimits:   acidv1.ResourceDescription{CPU: "510m", Memory: "1.4Gi"},
+				},
+			},
+			acidv1.Sidecar{
+				Name:        "replace-sidecar",
+				DockerImage: "overwrite-image",
+			},
+		},
+	}
+
+	cluster = New(
+		Config{
+			OpConfig: config.Config{
+				PodManagementPolicy: "ordered_ready",
+				ProtectedRoles:      []string{"admin"},
+				Auth: config.Auth{
+					SuperUsername:       superUserName,
+					ReplicationUsername: replicationUserName,
+				},
+				Resources: config.Resources{
+					DefaultCPURequest:    "200m",
+					DefaultCPULimit:      "500m",
+					DefaultMemoryRequest: "0.7Gi",
+					DefaultMemoryLimit:   "1.3Gi",
+				},
+				SidecarImages: map[string]string{
+					"deprecated-global-sidecar": "image:123",
+				},
+				SidecarContainers: []v1.Container{
+					v1.Container{
+						Name: "global-sidecar",
+					},
+					// will be replaced by a cluster specific sidecar with the same name
+					v1.Container{
+						Name:  "replace-sidecar",
+						Image: "replaced-image",
+					},
+				},
+				Scalyr: config.Scalyr{
+					ScalyrAPIKey:        "abc",
+					ScalyrImage:         "scalyr-image",
+					ScalyrCPURequest:    "220m",
+					ScalyrCPULimit:      "520m",
+					ScalyrMemoryRequest: "0.9Gi",
+					// ise default memory limit
+				},
+			},
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
+
+	s, err := cluster.generateStatefulSet(&spec)
+	assert.NoError(t, err)
+
+	env := []v1.EnvVar{
+		{
+			Name: "POD_NAME",
+			ValueFrom: &v1.EnvVarSource{
+				FieldRef: &v1.ObjectFieldSelector{
+					APIVersion: "v1",
+					FieldPath:  "metadata.name",
+				},
+			},
+		},
+		{
+			Name: "POD_NAMESPACE",
+			ValueFrom: &v1.EnvVarSource{
+				FieldRef: &v1.ObjectFieldSelector{
+					APIVersion: "v1",
+					FieldPath:  "metadata.namespace",
+				},
+			},
+		},
+		{
+			Name:  "POSTGRES_USER",
+			Value: superUserName,
+		},
+		{
+			Name: "POSTGRES_PASSWORD",
+			ValueFrom: &v1.EnvVarSource{
+				SecretKeyRef: &v1.SecretKeySelector{
+					LocalObjectReference: v1.LocalObjectReference{
+						Name: "",
+					},
+					Key: "password",
+				},
+			},
+		},
+	}
+	mounts := []v1.VolumeMount{
+		v1.VolumeMount{
+			Name:      "pgdata",
+			MountPath: "/home/postgres/pgdata",
+		},
+	}
+
+	// deduplicated sidecars and Patroni
+	assert.Equal(t, 7, len(s.Spec.Template.Spec.Containers), "wrong number of containers")
+
+	// cluster specific sidecar
+	assert.Contains(t, s.Spec.Template.Spec.Containers, v1.Container{
+		Name:            "cluster-specific-sidecar",
+		Env:             env,
+		Resources:       generateKubernetesResources("200m", "500m", "0.7Gi", "1.3Gi"),
+		ImagePullPolicy: v1.PullIfNotPresent,
+		VolumeMounts:    mounts,
+	})
+
+	// container specific resources
+	expectedResources := generateKubernetesResources("210m", "510m", "0.8Gi", "1.4Gi")
+	assert.Equal(t, expectedResources.Requests[v1.ResourceCPU], s.Spec.Template.Spec.Containers[2].Resources.Requests[v1.ResourceCPU])
+	assert.Equal(t, expectedResources.Limits[v1.ResourceCPU], s.Spec.Template.Spec.Containers[2].Resources.Limits[v1.ResourceCPU])
+	assert.Equal(t, expectedResources.Requests[v1.ResourceMemory], s.Spec.Template.Spec.Containers[2].Resources.Requests[v1.ResourceMemory])
+	assert.Equal(t, expectedResources.Limits[v1.ResourceMemory], s.Spec.Template.Spec.Containers[2].Resources.Limits[v1.ResourceMemory])
+
+	// deprecated global sidecar
+	assert.Contains(t, s.Spec.Template.Spec.Containers, v1.Container{
+		Name:            "deprecated-global-sidecar",
+		Image:           "image:123",
+		Env:             env,
+		Resources:       generateKubernetesResources("200m", "500m", "0.7Gi", "1.3Gi"),
+		ImagePullPolicy: v1.PullIfNotPresent,
+		VolumeMounts:    mounts,
+	})
+
+	// global sidecar
+	assert.Contains(t, s.Spec.Template.Spec.Containers, v1.Container{
+		Name:         "global-sidecar",
+		Env:          env,
+		VolumeMounts: mounts,
+	})
+
+	// replaced sidecar
+	assert.Contains(t, s.Spec.Template.Spec.Containers, v1.Container{
+		Name:            "replace-sidecar",
+		Image:           "overwrite-image",
+		Resources:       generateKubernetesResources("200m", "500m", "0.7Gi", "1.3Gi"),
+		ImagePullPolicy: v1.PullIfNotPresent,
+		Env:             env,
+		VolumeMounts:    mounts,
+	})
+
+	// replaced sidecar
+	// the order in env is important
+	scalyrEnv := append([]v1.EnvVar{v1.EnvVar{Name: "SCALYR_API_KEY", Value: "abc"}, v1.EnvVar{Name: "SCALYR_SERVER_HOST", Value: ""}}, env...)
+	assert.Contains(t, s.Spec.Template.Spec.Containers, v1.Container{
+		Name:            "scalyr-sidecar",
+		Image:           "scalyr-image",
+		Resources:       generateKubernetesResources("220m", "520m", "0.9Gi", "1.3Gi"),
+		ImagePullPolicy: v1.PullIfNotPresent,
+		Env:             scalyrEnv,
+		VolumeMounts:    mounts,
+	})
+
+}

pkg/cluster/pod.go

@@ -294,6 +294,27 @@ func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
 	return pod, nil
 }
 
+func (c *Cluster) isSafeToRecreatePods(pods *v1.PodList) bool {
+
+	/*
+	 Operator should not re-create pods if there is at least one replica being bootstrapped
+	 because Patroni might use other replicas to take basebackup from (see Patroni's "clonefrom" tag).
+
+	 XXX operator cannot forbid replica re-init, so we might still fail if re-init is started
+	 after this check succeeds but before a pod is re-created
+	*/
+
+	for _, pod := range pods.Items {
+		state, err := c.patroni.GetPatroniMemberState(&pod)
+		if err != nil || state == "creating replica" {
+			c.logger.Warningf("cannot re-create replica %s: it is currently being initialized", pod.Name)
+			return false
+		}
+
+	}
+	return true
+}
+
 func (c *Cluster) recreatePods() error {
 	c.setProcessName("starting to recreate pods")
 	ls := c.labelsSet(false)
@@ -309,6 +330,10 @@ func (c *Cluster) recreatePods() error {
 	}
 	c.logger.Infof("there are %d pods in the cluster to recreate", len(pods.Items))
 
+	if !c.isSafeToRecreatePods(pods) {
+		return fmt.Errorf("postpone pod recreation until next Sync: recreation is unsafe because pods are being initilalized")
+	}
+
 	var (
 		masterPod, newMasterPod, newPod *v1.Pod
 	)

pkg/cluster/resources_test.go

@@ -36,7 +36,7 @@ func TestConnectionPoolerCreationAndDeletion(t *testing.T) {
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
 				},
 			},
-		}, k8sutil.NewMockKubernetesClient(), acidv1.Postgresql{}, logger)
+		}, k8sutil.NewMockKubernetesClient(), acidv1.Postgresql{}, logger, eventRecorder)
 
 	cluster.Statefulset = &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -85,7 +85,7 @@ func TestNeedConnectionPooler(t *testing.T) {
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
 				},
 			},
-		}, k8sutil.NewMockKubernetesClient(), acidv1.Postgresql{}, logger)
+		}, k8sutil.NewMockKubernetesClient(), acidv1.Postgresql{}, logger, eventRecorder)
 
 	cluster.Spec = acidv1.PostgresSpec{
 		ConnectionPooler: &acidv1.ConnectionPooler{},

pkg/cluster/sync.go

@@ -3,11 +3,7 @@ package cluster
 import (
 	"context"
 	"fmt"
-
-	batchv1beta1 "k8s.io/api/batch/v1beta1"
-	v1 "k8s.io/api/core/v1"
-	policybeta1 "k8s.io/api/policy/v1beta1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"strings"
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/spec"
@@ -15,6 +11,11 @@ import (
 	"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"
+	policybeta1 "k8s.io/api/policy/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // Sync syncs the cluster, making sure the actual Kubernetes objects correspond to what is defined in the manifest.
@@ -108,6 +109,11 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 			err = fmt.Errorf("could not sync databases: %v", err)
 			return err
 		}
+		c.logger.Debugf("syncing prepared databases with schemas")
+		if err = c.syncPreparedDatabases(); err != nil {
+			err = fmt.Errorf("could not sync prepared database: %v", err)
+			return err
+		}
 	}
 
 	// remove PVCs of shut down pods
@@ -117,7 +123,7 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 	}
 
 	// sync connection pooler
-	if err = c.syncConnectionPooler(&oldSpec, newSpec, c.installLookupFunction); err != nil {
+	if _, err = c.syncConnectionPooler(&oldSpec, newSpec, c.installLookupFunction); err != nil {
 		return fmt.Errorf("could not sync connection pooler: %v", err)
 	}
 
@@ -128,10 +134,11 @@ func (c *Cluster) syncServices() error {
 	for _, role := range []PostgresRole{Master, Replica} {
 		c.logger.Debugf("syncing %s service", role)
 
-		if err := c.syncEndpoint(role); err != nil {
-			return fmt.Errorf("could not sync %s endpoint: %v", role, err)
+		if !c.patroniKubernetesUseConfigMaps() {
+			if err := c.syncEndpoint(role); err != nil {
+				return fmt.Errorf("could not sync %s endpoint: %v", role, err)
+			}
 		}
-
 		if err := c.syncService(role); err != nil {
 			return fmt.Errorf("could not sync %s service: %v", role, err)
 		}
@@ -257,6 +264,28 @@ func (c *Cluster) syncPodDisruptionBudget(isUpdate bool) error {
 	return nil
 }
 
+func (c *Cluster) mustUpdatePodsAfterLazyUpdate(desiredSset *appsv1.StatefulSet) (bool, error) {
+
+	pods, err := c.listPods()
+	if err != nil {
+		return false, fmt.Errorf("could not list pods of the statefulset: %v", err)
+	}
+
+	for _, pod := range pods {
+
+		effectivePodImage := pod.Spec.Containers[0].Image
+		ssImage := desiredSset.Spec.Template.Spec.Containers[0].Image
+
+		if ssImage != effectivePodImage {
+			c.logger.Infof("not all pods were re-started when the lazy upgrade was enabled; forcing the rolling upgrade now")
+			return true, nil
+		}
+
+	}
+
+	return false, nil
+}
+
 func (c *Cluster) syncStatefulSet() error {
 	var (
 		podsRollingUpdateRequired bool
@@ -335,6 +364,19 @@ func (c *Cluster) syncStatefulSet() error {
 				}
 			}
 		}
+
+		if !podsRollingUpdateRequired && !c.OpConfig.EnableLazySpiloUpgrade {
+			// even if desired and actual statefulsets match
+			// there still may be not up-to-date pods on condition
+			//  (a) the lazy update was just disabled
+			// and
+			//  (b) some of the pods were not restarted when the lazy update was still in place
+			podsRollingUpdateRequired, err = c.mustUpdatePodsAfterLazyUpdate(desiredSS)
+			if err != nil {
+				return fmt.Errorf("could not list pods of the statefulset: %v", err)
+			}
+		}
+
 	}
 
 	// Apply special PostgreSQL parameters that can only be set via the Patroni API.
@@ -348,10 +390,12 @@ func (c *Cluster) syncStatefulSet() error {
 	// statefulset or those that got their configuration from the outdated statefulset)
 	if podsRollingUpdateRequired {
 		c.logger.Debugln("performing rolling update")
+		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Performing rolling update")
 		if err := c.recreatePods(); err != nil {
 			return fmt.Errorf("could not recreate pods: %v", err)
 		}
 		c.logger.Infof("pods have been recreated")
+		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Rolling update done - pods have been recreated")
 		if err := c.applyRollingUpdateFlagforStatefulSet(false); err != nil {
 			c.logger.Warningf("could not clear rolling update for the statefulset: %v", err)
 		}
@@ -531,6 +575,7 @@ func (c *Cluster) syncDatabases() error {
 
 	createDatabases := make(map[string]string)
 	alterOwnerDatabases := make(map[string]string)
+	preparedDatabases := make([]string, 0)
 
 	if err := c.initDbConn(); err != nil {
 		return fmt.Errorf("could not init database connection")
@@ -546,12 +591,24 @@ func (c *Cluster) syncDatabases() error {
 		return fmt.Errorf("could not get current databases: %v", err)
 	}
 
-	for datname, newOwner := range c.Spec.Databases {
-		currentOwner, exists := currentDatabases[datname]
+	// if no prepared databases are specified create a database named like the cluster
+	if c.Spec.PreparedDatabases != nil && len(c.Spec.PreparedDatabases) == 0 { // TODO: add option to disable creating such a default DB
+		c.Spec.PreparedDatabases = map[string]acidv1.PreparedDatabase{strings.Replace(c.Name, "-", "_", -1): {}}
+	}
+	for preparedDatabaseName := range c.Spec.PreparedDatabases {
+		_, exists := currentDatabases[preparedDatabaseName]
 		if !exists {
-			createDatabases[datname] = newOwner
+			createDatabases[preparedDatabaseName] = preparedDatabaseName + constants.OwnerRoleNameSuffix
+			preparedDatabases = append(preparedDatabases, preparedDatabaseName)
+		}
+	}
+
+	for databaseName, newOwner := range c.Spec.Databases {
+		currentOwner, exists := currentDatabases[databaseName]
+		if !exists {
+			createDatabases[databaseName] = newOwner
 		} else if currentOwner != newOwner {
-			alterOwnerDatabases[datname] = newOwner
+			alterOwnerDatabases[databaseName] = newOwner
 		}
 	}
 
@@ -559,13 +616,116 @@ func (c *Cluster) syncDatabases() error {
 		return nil
 	}
 
-	for datname, owner := range createDatabases {
-		if err = c.executeCreateDatabase(datname, owner); err != nil {
+	for databaseName, owner := range createDatabases {
+		if err = c.executeCreateDatabase(databaseName, owner); err != nil {
 			return err
 		}
 	}
-	for datname, owner := range alterOwnerDatabases {
-		if err = c.executeAlterDatabaseOwner(datname, owner); err != nil {
+	for databaseName, owner := range alterOwnerDatabases {
+		if err = c.executeAlterDatabaseOwner(databaseName, owner); err != nil {
+			return err
+		}
+	}
+
+	// set default privileges for prepared database
+	for _, preparedDatabase := range preparedDatabases {
+		if err = c.execAlterGlobalDefaultPrivileges(preparedDatabase+constants.OwnerRoleNameSuffix, preparedDatabase); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Cluster) syncPreparedDatabases() error {
+	c.setProcessName("syncing prepared databases")
+	for preparedDbName, preparedDB := range c.Spec.PreparedDatabases {
+		if err := c.initDbConnWithName(preparedDbName); err != nil {
+			return fmt.Errorf("could not init connection to database %s: %v", preparedDbName, err)
+		}
+		defer func() {
+			if err := c.closeDbConn(); err != nil {
+				c.logger.Errorf("could not close database connection: %v", err)
+			}
+		}()
+
+		// now, prepare defined schemas
+		preparedSchemas := preparedDB.PreparedSchemas
+		if len(preparedDB.PreparedSchemas) == 0 {
+			preparedSchemas = map[string]acidv1.PreparedSchema{"data": {DefaultRoles: util.True()}}
+		}
+		if err := c.syncPreparedSchemas(preparedDbName, preparedSchemas); err != nil {
+			return err
+		}
+
+		// install extensions
+		if err := c.syncExtensions(preparedDB.Extensions); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Cluster) syncPreparedSchemas(databaseName string, preparedSchemas map[string]acidv1.PreparedSchema) error {
+	c.setProcessName("syncing prepared schemas")
+
+	currentSchemas, err := c.getSchemas()
+	if err != nil {
+		return fmt.Errorf("could not get current schemas: %v", err)
+	}
+
+	var schemas []string
+
+	for schema := range preparedSchemas {
+		schemas = append(schemas, schema)
+	}
+
+	if createPreparedSchemas, equal := util.SubstractStringSlices(schemas, currentSchemas); !equal {
+		for _, schemaName := range createPreparedSchemas {
+			owner := constants.OwnerRoleNameSuffix
+			dbOwner := databaseName + owner
+			if preparedSchemas[schemaName].DefaultRoles == nil || *preparedSchemas[schemaName].DefaultRoles {
+				owner = databaseName + "_" + schemaName + owner
+			} else {
+				owner = dbOwner
+			}
+			if err = c.executeCreateDatabaseSchema(databaseName, schemaName, dbOwner, owner); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (c *Cluster) syncExtensions(extensions map[string]string) error {
+	c.setProcessName("syncing database extensions")
+
+	createExtensions := make(map[string]string)
+	alterExtensions := make(map[string]string)
+
+	currentExtensions, err := c.getExtensions()
+	if err != nil {
+		return fmt.Errorf("could not get current database extensions: %v", err)
+	}
+
+	for extName, newSchema := range extensions {
+		currentSchema, exists := currentExtensions[extName]
+		if !exists {
+			createExtensions[extName] = newSchema
+		} else if currentSchema != newSchema {
+			alterExtensions[extName] = newSchema
+		}
+	}
+
+	for extName, schema := range createExtensions {
+		if err = c.executeCreateExtension(extName, schema); err != nil {
+			return err
+		}
+	}
+	for extName, schema := range alterExtensions {
+		if err = c.executeAlterExtension(extName, schema); err != nil {
 			return err
 		}
 	}
@@ -626,7 +786,13 @@ func (c *Cluster) syncLogicalBackupJob() error {
 	return nil
 }
 
-func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, lookup InstallFunction) error {
+func (c *Cluster) syncConnectionPooler(oldSpec,
+	newSpec *acidv1.Postgresql,
+	lookup InstallFunction) (SyncReason, error) {
+
+	var reason SyncReason
+	var err error
+
 	if c.ConnectionPooler == nil {
 		c.ConnectionPooler = &ConnectionPoolerObjects{}
 	}
@@ -663,20 +829,20 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, look
 				specUser,
 				c.OpConfig.ConnectionPooler.User)
 
-			if err := lookup(schema, user); err != nil {
-				return err
+			if err = lookup(schema, user); err != nil {
+				return NoSync, err
 			}
 		}
 
-		if err := c.syncConnectionPoolerWorker(oldSpec, newSpec); err != nil {
+		if reason, err = c.syncConnectionPoolerWorker(oldSpec, newSpec); err != nil {
 			c.logger.Errorf("could not sync connection pooler: %v", err)
-			return err
+			return reason, err
 		}
 	}
 
 	if oldNeedConnectionPooler && !newNeedConnectionPooler {
 		// delete and cleanup resources
-		if err := c.deleteConnectionPooler(); err != nil {
+		if err = c.deleteConnectionPooler(); err != nil {
 			c.logger.Warningf("could not remove connection pooler: %v", err)
 		}
 	}
@@ -687,20 +853,22 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, look
 			(c.ConnectionPooler.Deployment != nil ||
 				c.ConnectionPooler.Service != nil) {
 
-			if err := c.deleteConnectionPooler(); err != nil {
+			if err = c.deleteConnectionPooler(); err != nil {
 				c.logger.Warningf("could not remove connection pooler: %v", err)
 			}
 		}
 	}
 
-	return nil
+	return reason, nil
 }
 
 // Synchronize connection pooler resources. Effectively we're interested only in
 // synchronizing the corresponding deployment, but in case of deployment or
 // service is missing, create it. After checking, also remember an object for
 // the future references.
-func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql) error {
+func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql) (
+	SyncReason, error) {
+
 	deployment, err := c.KubeClient.
 		Deployments(c.Namespace).
 		Get(context.TODO(), c.connectionPoolerName(), metav1.GetOptions{})
@@ -712,7 +880,7 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		deploymentSpec, err := c.generateConnectionPoolerDeployment(&newSpec.Spec)
 		if err != nil {
 			msg = "could not generate deployment for connection pooler: %v"
-			return fmt.Errorf(msg, err)
+			return NoSync, fmt.Errorf(msg, err)
 		}
 
 		deployment, err := c.KubeClient.
@@ -720,18 +888,35 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 			Create(context.TODO(), deploymentSpec, metav1.CreateOptions{})
 
 		if err != nil {
-			return err
+			return NoSync, err
 		}
 
 		c.ConnectionPooler.Deployment = deployment
 	} else if err != nil {
-		return fmt.Errorf("could not get connection pooler deployment to sync: %v", err)
+		msg := "could not get connection pooler deployment to sync: %v"
+		return NoSync, fmt.Errorf(msg, err)
 	} else {
 		c.ConnectionPooler.Deployment = deployment
 
 		// actual synchronization
 		oldConnectionPooler := oldSpec.Spec.ConnectionPooler
 		newConnectionPooler := newSpec.Spec.ConnectionPooler
+
+		// sync implementation below assumes that both old and new specs are
+		// not nil, but it can happen. To avoid any confusion like updating a
+		// deployment because the specification changed from nil to an empty
+		// struct (that was initialized somewhere before) replace any nil with
+		// an empty spec.
+		if oldConnectionPooler == nil {
+			oldConnectionPooler = &acidv1.ConnectionPooler{}
+		}
+
+		if newConnectionPooler == nil {
+			newConnectionPooler = &acidv1.ConnectionPooler{}
+		}
+
+		c.logger.Infof("Old: %+v, New %+v", oldConnectionPooler, newConnectionPooler)
+
 		specSync, specReason := c.needSyncConnectionPoolerSpecs(oldConnectionPooler, newConnectionPooler)
 		defaultsSync, defaultsReason := c.needSyncConnectionPoolerDefaults(newConnectionPooler, deployment)
 		reason := append(specReason, defaultsReason...)
@@ -742,7 +927,7 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 			newDeploymentSpec, err := c.generateConnectionPoolerDeployment(&newSpec.Spec)
 			if err != nil {
 				msg := "could not generate deployment for connection pooler: %v"
-				return fmt.Errorf(msg, err)
+				return reason, fmt.Errorf(msg, err)
 			}
 
 			oldDeploymentSpec := c.ConnectionPooler.Deployment
@@ -752,11 +937,11 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 				newDeploymentSpec)
 
 			if err != nil {
-				return err
+				return reason, err
 			}
 
 			c.ConnectionPooler.Deployment = deployment
-			return nil
+			return reason, nil
 		}
 	}
 
@@ -774,16 +959,17 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 			Create(context.TODO(), serviceSpec, metav1.CreateOptions{})
 
 		if err != nil {
-			return err
+			return NoSync, err
 		}
 
 		c.ConnectionPooler.Service = service
 	} else if err != nil {
-		return fmt.Errorf("could not get connection pooler service to sync: %v", err)
+		msg := "could not get connection pooler service to sync: %v"
+		return NoSync, fmt.Errorf(msg, err)
 	} else {
 		// Service updates are not supported and probably not that useful anyway
 		c.ConnectionPooler.Service = service
 	}
 
-	return nil
+	return NoSync, nil
 }

pkg/cluster/sync_test.go

@@ -2,6 +2,7 @@ package cluster
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
@@ -17,7 +18,7 @@ func int32ToPointer(value int32) *int32 {
 	return &value
 }
 
-func deploymentUpdated(cluster *Cluster, err error) error {
+func deploymentUpdated(cluster *Cluster, err error, reason SyncReason) error {
 	if cluster.ConnectionPooler.Deployment.Spec.Replicas == nil ||
 		*cluster.ConnectionPooler.Deployment.Spec.Replicas != 2 {
 		return fmt.Errorf("Wrong nubmer of instances")
@@ -26,7 +27,7 @@ func deploymentUpdated(cluster *Cluster, err error) error {
 	return nil
 }
 
-func objectsAreSaved(cluster *Cluster, err error) error {
+func objectsAreSaved(cluster *Cluster, err error, reason SyncReason) error {
 	if cluster.ConnectionPooler == nil {
 		return fmt.Errorf("Connection pooler resources are empty")
 	}
@@ -42,7 +43,7 @@ func objectsAreSaved(cluster *Cluster, err error) error {
 	return nil
 }
 
-func objectsAreDeleted(cluster *Cluster, err error) error {
+func objectsAreDeleted(cluster *Cluster, err error, reason SyncReason) error {
 	if cluster.ConnectionPooler != nil {
 		return fmt.Errorf("Connection pooler was not deleted")
 	}
@@ -50,6 +51,16 @@ func objectsAreDeleted(cluster *Cluster, err error) error {
 	return nil
 }
 
+func noEmptySync(cluster *Cluster, err error, reason SyncReason) error {
+	for _, msg := range reason {
+		if strings.HasPrefix(msg, "update [] from '<nil>' to '") {
+			return fmt.Errorf("There is an empty reason, %s", msg)
+		}
+	}
+
+	return nil
+}
+
 func TestConnectionPoolerSynchronization(t *testing.T) {
 	testName := "Test connection pooler synchronization"
 	var cluster = New(
@@ -68,7 +79,7 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 					NumberOfInstances:                    int32ToPointer(1),
 				},
 			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
 
 	cluster.Statefulset = &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -91,15 +102,15 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 
 	clusterNewDefaultsMock := *cluster
 	clusterNewDefaultsMock.KubeClient = k8sutil.NewMockKubernetesClient()
-	cluster.OpConfig.ConnectionPooler.Image = "pooler:2.0"
-	cluster.OpConfig.ConnectionPooler.NumberOfInstances = int32ToPointer(2)
 
 	tests := []struct {
-		subTest string
-		oldSpec *acidv1.Postgresql
-		newSpec *acidv1.Postgresql
-		cluster *Cluster
-		check   func(cluster *Cluster, err error) error
+		subTest          string
+		oldSpec          *acidv1.Postgresql
+		newSpec          *acidv1.Postgresql
+		cluster          *Cluster
+		defaultImage     string
+		defaultInstances int32
+		check            func(cluster *Cluster, err error, reason SyncReason) error
 	}{
 		{
 			subTest: "create if doesn't exist",
@@ -113,8 +124,10 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 					ConnectionPooler: &acidv1.ConnectionPooler{},
 				},
 			},
-			cluster: &clusterMissingObjects,
-			check:   objectsAreSaved,
+			cluster:          &clusterMissingObjects,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            objectsAreSaved,
 		},
 		{
 			subTest: "create if doesn't exist with a flag",
@@ -126,8 +139,10 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 					EnableConnectionPooler: boolToPointer(true),
 				},
 			},
-			cluster: &clusterMissingObjects,
-			check:   objectsAreSaved,
+			cluster:          &clusterMissingObjects,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            objectsAreSaved,
 		},
 		{
 			subTest: "create from scratch",
@@ -139,8 +154,10 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 					ConnectionPooler: &acidv1.ConnectionPooler{},
 				},
 			},
-			cluster: &clusterMissingObjects,
-			check:   objectsAreSaved,
+			cluster:          &clusterMissingObjects,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            objectsAreSaved,
 		},
 		{
 			subTest: "delete if not needed",
@@ -152,8 +169,10 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 			newSpec: &acidv1.Postgresql{
 				Spec: acidv1.PostgresSpec{},
 			},
-			cluster: &clusterMock,
-			check:   objectsAreDeleted,
+			cluster:          &clusterMock,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            objectsAreDeleted,
 		},
 		{
 			subTest: "cleanup if still there",
@@ -163,8 +182,10 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 			newSpec: &acidv1.Postgresql{
 				Spec: acidv1.PostgresSpec{},
 			},
-			cluster: &clusterDirtyMock,
-			check:   objectsAreDeleted,
+			cluster:          &clusterDirtyMock,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            objectsAreDeleted,
 		},
 		{
 			subTest: "update deployment",
@@ -182,8 +203,10 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 					},
 				},
 			},
-			cluster: &clusterMock,
-			check:   deploymentUpdated,
+			cluster:          &clusterMock,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            deploymentUpdated,
 		},
 		{
 			subTest: "update image from changed defaults",
@@ -197,14 +220,40 @@ func TestConnectionPoolerSynchronization(t *testing.T) {
 					ConnectionPooler: &acidv1.ConnectionPooler{},
 				},
 			},
-			cluster: &clusterNewDefaultsMock,
-			check:   deploymentUpdated,
+			cluster:          &clusterNewDefaultsMock,
+			defaultImage:     "pooler:2.0",
+			defaultInstances: 2,
+			check:            deploymentUpdated,
+		},
+		{
+			subTest: "there is no sync from nil to an empty spec",
+			oldSpec: &acidv1.Postgresql{
+				Spec: acidv1.PostgresSpec{
+					EnableConnectionPooler: boolToPointer(true),
+					ConnectionPooler:       nil,
+				},
+			},
+			newSpec: &acidv1.Postgresql{
+				Spec: acidv1.PostgresSpec{
+					EnableConnectionPooler: boolToPointer(true),
+					ConnectionPooler:       &acidv1.ConnectionPooler{},
+				},
+			},
+			cluster:          &clusterMock,
+			defaultImage:     "pooler:1.0",
+			defaultInstances: 1,
+			check:            noEmptySync,
 		},
 	}
 	for _, tt := range tests {
-		err := tt.cluster.syncConnectionPooler(tt.oldSpec, tt.newSpec, mockInstallLookupFunction)
+		tt.cluster.OpConfig.ConnectionPooler.Image = tt.defaultImage
+		tt.cluster.OpConfig.ConnectionPooler.NumberOfInstances =
+			int32ToPointer(tt.defaultInstances)
 
-		if err := tt.check(tt.cluster, err); err != nil {
+		reason, err := tt.cluster.syncConnectionPooler(tt.oldSpec,
+			tt.newSpec, mockInstallLookupFunction)
+
+		if err := tt.check(tt.cluster, err, reason); err != nil {
 			t.Errorf("%s [%s]: Could not synchronize, %+v",
 				testName, tt.subTest, err)
 		}

pkg/cluster/types.go

@@ -73,3 +73,8 @@ type ClusterStatus struct {
 type TemplateParams map[string]interface{}
 
 type InstallFunction func(schema string, user string) error
+
+type SyncReason []string
+
+// no sync happened, empty value
+var NoSync SyncReason = []string{}

pkg/cluster/util.go

@@ -530,3 +530,22 @@ func (c *Cluster) needConnectionPoolerWorker(spec *acidv1.PostgresSpec) bool {
 func (c *Cluster) needConnectionPooler() bool {
 	return c.needConnectionPoolerWorker(&c.Spec)
 }
+
+// Earlier arguments take priority
+func mergeContainers(containers ...[]v1.Container) ([]v1.Container, []string) {
+	containerNameTaken := map[string]bool{}
+	result := make([]v1.Container, 0)
+	conflicts := make([]string, 0)
+
+	for _, containerArray := range containers {
+		for _, container := range containerArray {
+			if _, taken := containerNameTaken[container.Name]; taken {
+				conflicts = append(conflicts, container.Name)
+			} else {
+				containerNameTaken[container.Name] = true
+				result = append(result, container)
+			}
+		}
+	}
+	return result, conflicts
+}

pkg/controller/controller.go

@@ -7,24 +7,24 @@ import (
 	"sync"
 
 	"github.com/sirupsen/logrus"
-	v1 "k8s.io/api/core/v1"
-	rbacv1 "k8s.io/api/rbac/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/tools/cache"
-
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/apiserver"
 	"github.com/zalando/postgres-operator/pkg/cluster"
+	acidv1informer "github.com/zalando/postgres-operator/pkg/generated/informers/externalversions/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/spec"
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/config"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
 	"github.com/zalando/postgres-operator/pkg/util/ringlog"
-
-	acidv1informer "github.com/zalando/postgres-operator/pkg/generated/informers/externalversions/acid.zalan.do/v1"
+	v1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 )
 
 // Controller represents operator controller
@@ -36,6 +36,9 @@ type Controller struct {
 	KubeClient k8sutil.KubernetesClient
 	apiserver  *apiserver.Server
 
+	eventRecorder    record.EventRecorder
+	eventBroadcaster record.EventBroadcaster
+
 	stopCh chan struct{}
 
 	controllerID     string
@@ -67,10 +70,21 @@ type Controller struct {
 func NewController(controllerConfig *spec.ControllerConfig, controllerId string) *Controller {
 	logger := logrus.New()
 
+	var myComponentName = "postgres-operator"
+	if controllerId != "" {
+		myComponentName += "/" + controllerId
+	}
+
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(logger.Infof)
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: myComponentName})
+
 	c := &Controller{
 		config:           *controllerConfig,
 		opConfig:         &config.Config{},
 		logger:           logger.WithField("pkg", "controller"),
+		eventRecorder:    recorder,
+		eventBroadcaster: eventBroadcaster,
 		controllerID:     controllerId,
 		curWorkerCluster: sync.Map{},
 		clusterWorkers:   make(map[spec.NamespacedName]uint32),
@@ -93,6 +107,11 @@ func (c *Controller) initClients() {
 	if err != nil {
 		c.logger.Fatalf("could not create kubernetes clients: %v", err)
 	}
+	c.eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: c.KubeClient.EventsGetter.Events("")})
+	if err != nil {
+		c.logger.Fatalf("could not setup kubernetes event sink: %v", err)
+	}
+
 }
 
 func (c *Controller) initOperatorConfig() {
@@ -159,6 +178,11 @@ func (c *Controller) warnOnDeprecatedOperatorParameters() {
 		c.logger.Warningf("Operator configuration parameter 'enable_load_balancer' is deprecated and takes no effect. " +
 			"Consider using the 'enable_master_load_balancer' or 'enable_replica_load_balancer' instead.")
 	}
+
+	if len(c.opConfig.SidecarImages) > 0 {
+		c.logger.Warningf("Operator configuration parameter 'sidecar_docker_images' is deprecated. " +
+			"Consider using 'sidecars' instead.")
+	}
 }
 
 func (c *Controller) initPodServiceAccount() {

pkg/controller/operator_config.go

@@ -34,6 +34,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
 
 	// general config
 	result.EnableCRDValidation = fromCRD.EnableCRDValidation
+	result.EnableLazySpiloUpgrade = fromCRD.EnableLazySpiloUpgrade
 	result.EtcdHost = fromCRD.EtcdHost
 	result.KubernetesUseConfigMaps = fromCRD.KubernetesUseConfigMaps
 	result.DockerImage = fromCRD.DockerImage
@@ -44,7 +45,8 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
 	result.RepairPeriod = time.Duration(fromCRD.RepairPeriod)
 	result.SetMemoryRequestToLimit = fromCRD.SetMemoryRequestToLimit
 	result.ShmVolume = fromCRD.ShmVolume
-	result.Sidecars = fromCRD.Sidecars
+	result.SidecarImages = fromCRD.SidecarImages
+	result.SidecarContainers = fromCRD.SidecarContainers
 
 	// user config
 	result.SuperUsername = fromCRD.PostgresUsersConfiguration.SuperUsername

pkg/controller/postgresql.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/sirupsen/logrus"
 
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
@@ -157,7 +158,7 @@ func (c *Controller) acquireInitialListOfClusters() error {
 }
 
 func (c *Controller) addCluster(lg *logrus.Entry, clusterName spec.NamespacedName, pgSpec *acidv1.Postgresql) *cluster.Cluster {
-	cl := cluster.New(c.makeClusterConfig(), c.KubeClient, *pgSpec, lg)
+	cl := cluster.New(c.makeClusterConfig(), c.KubeClient, *pgSpec, lg, c.eventRecorder)
 	cl.Run(c.stopCh)
 	teamName := strings.ToLower(cl.Spec.TeamID)
 
@@ -236,6 +237,7 @@ func (c *Controller) processEvent(event ClusterEvent) {
 		if err := cl.Create(); err != nil {
 			cl.Error = fmt.Sprintf("could not create cluster: %v", err)
 			lg.Error(cl.Error)
+			c.eventRecorder.Eventf(cl.GetReference(), v1.EventTypeWarning, "Create", "%v", cl.Error)
 
 			return
 		}
@@ -274,6 +276,8 @@ func (c *Controller) processEvent(event ClusterEvent) {
 
 		c.curWorkerCluster.Store(event.WorkerID, cl)
 		cl.Delete()
+		// Fixme - no error handling for delete ?
+		// c.eventRecorder.Eventf(cl.GetReference, v1.EventTypeWarning, "Delete", "%v", cl.Error)
 
 		func() {
 			defer c.clustersMu.Unlock()
@@ -304,6 +308,7 @@ func (c *Controller) processEvent(event ClusterEvent) {
 		c.curWorkerCluster.Store(event.WorkerID, cl)
 		if err := cl.Sync(event.NewSpec); err != nil {
 			cl.Error = fmt.Sprintf("could not sync cluster: %v", err)
+			c.eventRecorder.Eventf(cl.GetReference(), v1.EventTypeWarning, "Sync", "%v", cl.Error)
 			lg.Error(cl.Error)
 			return
 		}

pkg/controller/types.go

@@ -1,9 +1,10 @@
 package controller
 
 import (
-	"k8s.io/apimachinery/pkg/types"
 	"time"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 )
 

pkg/spec/types.go

@@ -31,6 +31,7 @@ const (
 	RoleOriginInfrastructure
 	RoleOriginTeamsAPI
 	RoleOriginSystem
+	RoleOriginBootstrap
 	RoleConnectionPooler
 )
 
@@ -180,6 +181,8 @@ func (r RoleOrigin) String() string {
 		return "teams API role"
 	case RoleOriginSystem:
 		return "system role"
+	case RoleOriginBootstrap:
+		return "bootstrapped role"
 	case RoleConnectionPooler:
 		return "connection pooler role"
 	default:

pkg/util/config/config.go

@@ -9,6 +9,7 @@ import (
 
 	"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
@@ -107,12 +108,14 @@ type Config struct {
 	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'
-	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-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-cdp-12:1.6-p115"`
+	// deprecated in favour of SidecarContainers
+	SidecarImages         map[string]string `name:"sidecar_docker_images"`
+	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:""`
 	PodServiceAccountRoleBindingDefinition string            `name:"pod_service_account_role_binding_definition" default:""`
@@ -155,6 +158,7 @@ type Config struct {
 	PostgresSuperuserTeams    []string          `name:"postgres_superuser_teams" default:""`
 	SetMemoryRequestToLimit   bool              `name:"set_memory_request_to_limit" default:"false"`
 	EnableUnusedPVCDeletion   bool              `name:"enable_unused_pvc_deletion" default:"false"`
+	EnableLazySpiloUpgrade    bool              `name:"enable_lazy_spilo_upgrade" default:"false"`
 }
 
 // MustMarshal marshals the config or panics

pkg/util/constants/roles.go

@@ -14,4 +14,8 @@ const (
 	RoleFlagCreateDB          = "CREATEDB"
 	RoleFlagReplication       = "REPLICATION"
 	RoleFlagByPassRLS         = "BYPASSRLS"
+	OwnerRoleNameSuffix       = "_owner"
+	ReaderRoleNameSuffix      = "_reader"
+	WriterRoleNameSuffix      = "_writer"
+	UserRoleNameSuffix        = "_user"
 )

pkg/util/k8sutil/k8sutil.go

@@ -45,6 +45,7 @@ type KubernetesClient struct {
 	corev1.NodesGetter
 	corev1.NamespacesGetter
 	corev1.ServiceAccountsGetter
+	corev1.EventsGetter
 	appsv1.StatefulSetsGetter
 	appsv1.DeploymentsGetter
 	rbacv1.RoleBindingsGetter
@@ -142,6 +143,7 @@ func NewFromConfig(cfg *rest.Config) (KubernetesClient, error) {
 	kubeClient.RESTClient = client.CoreV1().RESTClient()
 	kubeClient.RoleBindingsGetter = client.RbacV1()
 	kubeClient.CronJobsGetter = client.BatchV1beta1()
+	kubeClient.EventsGetter = client.CoreV1()
 
 	apiextClient, err := apiextclient.NewForConfig(cfg)
 	if err != nil {

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/users/users.go

@@ -73,26 +73,44 @@ 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) {