merge

pkg/cluster/cluster.go

@@ -14,19 +14,10 @@ import (
 
 	"github.com/r3labs/diff"
 	"github.com/sirupsen/logrus"
-	appsv1 "k8s.io/api/apps/v1"
-	v1 "k8s.io/api/core/v1"
-	policybeta1 "k8s.io/api/policy/v1beta1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"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"
+	pgteams "github.com/zalando/postgres-operator/pkg/teams"
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/config"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
@@ -34,7 +25,16 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util/patroni"
 	"github.com/zalando/postgres-operator/pkg/util/teams"
 	"github.com/zalando/postgres-operator/pkg/util/users"
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	policybeta1 "k8s.io/api/policy/v1beta1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"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"
 )
 
 var (
@@ -48,6 +48,7 @@ var (
 type Config struct {
 	OpConfig                     config.Config
 	RestConfig                   *rest.Config
+	PgTeamMap                    pgteams.PostgresTeamMap
 	InfrastructureRoles          map[string]spec.PgUser // inherited from the controller
 	PodServiceAccount            *v1.ServiceAccount
 	PodServiceAccountRoleBinding *rbacv1.RoleBinding
@@ -371,11 +372,11 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 	//TODO: improve me
 	if *c.Statefulset.Spec.Replicas != *statefulSet.Spec.Replicas {
 		match = false
-		reasons = append(reasons, "new statefulset's number of replicas doesn't match the current one")
+		reasons = append(reasons, "new statefulset's number of replicas does not match the current one")
 	}
 	if !reflect.DeepEqual(c.Statefulset.Annotations, statefulSet.Annotations) {
 		match = false
-		reasons = append(reasons, "new statefulset's annotations doesn't match the current one")
+		reasons = append(reasons, "new statefulset's annotations does not match the current one")
 	}
 
 	needsRollUpdate, reasons = c.compareContainers("initContainers", c.Statefulset.Spec.Template.Spec.InitContainers, statefulSet.Spec.Template.Spec.InitContainers, needsRollUpdate, reasons)
@@ -392,24 +393,24 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 	if c.Statefulset.Spec.Template.Spec.ServiceAccountName != statefulSet.Spec.Template.Spec.ServiceAccountName {
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's serviceAccountName service account name doesn't match the current one")
+		reasons = append(reasons, "new statefulset's serviceAccountName service account name does not match the current one")
 	}
 	if *c.Statefulset.Spec.Template.Spec.TerminationGracePeriodSeconds != *statefulSet.Spec.Template.Spec.TerminationGracePeriodSeconds {
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's terminationGracePeriodSeconds doesn't match the current one")
+		reasons = append(reasons, "new statefulset's terminationGracePeriodSeconds does not match the current one")
 	}
 	if !reflect.DeepEqual(c.Statefulset.Spec.Template.Spec.Affinity, statefulSet.Spec.Template.Spec.Affinity) {
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's pod affinity doesn't match the current one")
+		reasons = append(reasons, "new statefulset's pod affinity does not match the current one")
 	}
 
 	// Some generated fields like creationTimestamp make it not possible to use DeepCompare on Spec.Template.ObjectMeta
 	if !reflect.DeepEqual(c.Statefulset.Spec.Template.Labels, statefulSet.Spec.Template.Labels) {
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's metadata labels doesn't match the current one")
+		reasons = append(reasons, "new statefulset's metadata labels does not match the current one")
 	}
 	if (c.Statefulset.Spec.Selector != nil) && (statefulSet.Spec.Selector != nil) {
 		if !reflect.DeepEqual(c.Statefulset.Spec.Selector.MatchLabels, statefulSet.Spec.Selector.MatchLabels) {
@@ -420,7 +421,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 				return &compareStatefulsetResult{}
 			}
 			needsReplace = true
-			reasons = append(reasons, "new statefulset's selector doesn't match the current one")
+			reasons = append(reasons, "new statefulset's selector does not match the current one")
 		}
 	}
 
@@ -434,7 +435,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		match = false
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's pod template security context in spec doesn't match the current one")
+		reasons = append(reasons, "new statefulset's pod template security context in spec does not match the current one")
 	}
 	if len(c.Statefulset.Spec.VolumeClaimTemplates) != len(statefulSet.Spec.VolumeClaimTemplates) {
 		needsReplace = true
@@ -445,17 +446,17 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		// Some generated fields like creationTimestamp make it not possible to use DeepCompare on ObjectMeta
 		if name != statefulSet.Spec.VolumeClaimTemplates[i].Name {
 			needsReplace = true
-			reasons = append(reasons, fmt.Sprintf("new statefulset's name for volume %d doesn't match the current one", i))
+			reasons = append(reasons, fmt.Sprintf("new statefulset's name for volume %d does not match the current one", i))
 			continue
 		}
 		if !reflect.DeepEqual(c.Statefulset.Spec.VolumeClaimTemplates[i].Annotations, statefulSet.Spec.VolumeClaimTemplates[i].Annotations) {
 			needsReplace = true
-			reasons = append(reasons, fmt.Sprintf("new statefulset's annotations for volume %q doesn't match the current one", name))
+			reasons = append(reasons, fmt.Sprintf("new statefulset's annotations for volume %q does not match the current one", name))
 		}
 		if !reflect.DeepEqual(c.Statefulset.Spec.VolumeClaimTemplates[i].Spec, statefulSet.Spec.VolumeClaimTemplates[i].Spec) {
 			name := c.Statefulset.Spec.VolumeClaimTemplates[i].Name
 			needsReplace = true
-			reasons = append(reasons, fmt.Sprintf("new statefulset's volumeClaimTemplates specification for volume %q doesn't match the current one", name))
+			reasons = append(reasons, fmt.Sprintf("new statefulset's volumeClaimTemplates specification for volume %q does not match the current one", name))
 		}
 	}
 
@@ -465,14 +466,14 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		match = false
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's pod priority class in spec doesn't match the current one")
+		reasons = append(reasons, "new statefulset's pod priority class in spec does not match the current one")
 	}
 
 	// 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
-		reasons = append(reasons, "lazy Spilo update: new statefulset's pod image doesn't match the current one")
+		reasons = append(reasons, "lazy Spilo update: new statefulset's pod image does not match the current one")
 	}
 
 	if needsRollUpdate || needsReplace {
@@ -582,7 +583,7 @@ func (c *Cluster) enforceMinResourceLimits(spec *acidv1.PostgresSpec) error {
 			return fmt.Errorf("could not compare defined CPU limit %s with configured minimum value %s: %v", cpuLimit, minCPULimit, err)
 		}
 		if isSmaller {
-			c.logger.Warningf("defined CPU limit %s is below required minimum %s and will be set to it", cpuLimit, minCPULimit)
+			c.logger.Warningf("defined CPU limit %s is below required minimum %s and will be increased", 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
 		}
@@ -595,7 +596,7 @@ func (c *Cluster) enforceMinResourceLimits(spec *acidv1.PostgresSpec) error {
 			return fmt.Errorf("could not compare defined memory limit %s with configured minimum value %s: %v", memoryLimit, minMemoryLimit, err)
 		}
 		if isSmaller {
-			c.logger.Warningf("defined memory limit %s is below required minimum %s and will be set to it", memoryLimit, minMemoryLimit)
+			c.logger.Warningf("defined memory limit %s is below required minimum %s and will be increased", 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
 		}
@@ -625,13 +626,16 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		}
 	}()
 
-	if oldSpec.Spec.PostgresqlParam.PgVersion != newSpec.Spec.PostgresqlParam.PgVersion { // PG versions comparison
+	if oldSpec.Spec.PostgresqlParam.PgVersion >= newSpec.Spec.PostgresqlParam.PgVersion {
 		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
+		// we need that hack to generate statefulset with the old version
 		newSpec.Spec.PostgresqlParam.PgVersion = oldSpec.Spec.PostgresqlParam.PgVersion
+	} else {
+		c.logger.Infof("postgresql version increased (%q -> %q), major version upgrade can be done manually after StatefulSet Sync",
+			oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
 	}
 
 	// Service
@@ -780,7 +784,13 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		}
 	}
 
-	// sync connection pooler
+	// Sync connection pooler. Before actually doing sync reset lookup
+	// installation flag, since manifest updates could add another db which we
+	// need to process. In the future we may want to do this more careful and
+	// check which databases we need to process, but even repeating the whole
+	// installation process should be good enough.
+	c.ConnectionPooler.LookupFunction = false
+
 	if _, err := c.syncConnectionPooler(oldSpec, newSpec,
 		c.installLookupFunction); err != nil {
 		c.logger.Errorf("could not sync connection pooler: %v", err)
@@ -1101,7 +1111,7 @@ func (c *Cluster) initTeamMembers(teamID string, isPostgresSuperuserTeam bool) e
 		if c.shouldAvoidProtectedOrSystemRole(username, "API role") {
 			continue
 		}
-		if c.OpConfig.EnableTeamSuperuser || isPostgresSuperuserTeam {
+		if (c.OpConfig.EnableTeamSuperuser && teamID == c.Spec.TeamID) || isPostgresSuperuserTeam {
 			flags = append(flags, constants.RoleFlagSuperuser)
 		} else {
 			if c.OpConfig.TeamAdminRole != "" {
@@ -1130,17 +1140,38 @@ func (c *Cluster) initTeamMembers(teamID string, isPostgresSuperuserTeam bool) e
 func (c *Cluster) initHumanUsers() error {
 
 	var clusterIsOwnedBySuperuserTeam bool
+	superuserTeams := []string{}
+
+	if c.OpConfig.EnablePostgresTeamCRDSuperusers {
+		superuserTeams = c.PgTeamMap.GetAdditionalSuperuserTeams(c.Spec.TeamID, true)
+	}
 
 	for _, postgresSuperuserTeam := range c.OpConfig.PostgresSuperuserTeams {
-		err := c.initTeamMembers(postgresSuperuserTeam, true)
-		if err != nil {
-			return fmt.Errorf("Cannot create a team %q of Postgres superusers: %v", postgresSuperuserTeam, err)
+		if !(util.SliceContains(superuserTeams, postgresSuperuserTeam)) {
+			superuserTeams = append(superuserTeams, postgresSuperuserTeam)
 		}
-		if postgresSuperuserTeam == c.Spec.TeamID {
+	}
+
+	for _, superuserTeam := range superuserTeams {
+		err := c.initTeamMembers(superuserTeam, true)
+		if err != nil {
+			return fmt.Errorf("Cannot initialize members for team %q of Postgres superusers: %v", superuserTeam, err)
+		}
+		if superuserTeam == c.Spec.TeamID {
 			clusterIsOwnedBySuperuserTeam = true
 		}
 	}
 
+	additionalTeams := c.PgTeamMap.GetAdditionalTeams(c.Spec.TeamID, true)
+	for _, additionalTeam := range additionalTeams {
+		if !(util.SliceContains(superuserTeams, additionalTeam)) {
+			err := c.initTeamMembers(additionalTeam, false)
+			if err != nil {
+				return fmt.Errorf("Cannot initialize members for additional team %q for cluster owned by %q: %v", additionalTeam, c.Spec.TeamID, err)
+			}
+		}
+	}
+
 	if clusterIsOwnedBySuperuserTeam {
 		c.logger.Infof("Team %q owning the cluster is also a team of superusers. Created superuser roles for its members instead of admin roles.", c.Spec.TeamID)
 		return nil
@@ -1148,7 +1179,7 @@ func (c *Cluster) initHumanUsers() error {
 
 	err := c.initTeamMembers(c.Spec.TeamID, false)
 	if err != nil {
-		return fmt.Errorf("Cannot create a team %q of admins owning the PG cluster: %v", c.Spec.TeamID, err)
+		return fmt.Errorf("Cannot initialize members for team %q who owns the Postgres cluster: %v", c.Spec.TeamID, err)
 	}
 
 	return nil

pkg/cluster/database.go

@@ -101,15 +101,20 @@ func (c *Cluster) databaseAccessDisabled() bool {
 }
 
 func (c *Cluster) initDbConn() error {
-	return c.initDbConnWithName("")
-}
-
-func (c *Cluster) initDbConnWithName(dbname string) error {
-	c.setProcessName("initializing db connection")
 	if c.pgDb != nil {
 		return nil
 	}
 
+	return c.initDbConnWithName("")
+}
+
+// Worker function for connection initialization. This function does not check
+// if the connection is already open, if it is then it will be overwritten.
+// Callers need to make sure no connection is open, otherwise we could leak
+// connections
+func (c *Cluster) initDbConnWithName(dbname string) error {
+	c.setProcessName("initializing db connection")
+
 	var conn *sql.DB
 	connstring := c.pgConnectionString(dbname)
 
@@ -126,12 +131,12 @@ func (c *Cluster) initDbConnWithName(dbname string) error {
 			}
 
 			if _, ok := err.(*net.OpError); ok {
-				c.logger.Errorf("could not connect to PostgreSQL database: %v", err)
+				c.logger.Warningf("could not connect to Postgres database: %v", err)
 				return false, nil
 			}
 
 			if err2 := conn.Close(); err2 != nil {
-				c.logger.Errorf("error when closing PostgreSQL connection after another error: %v", err)
+				c.logger.Errorf("error when closing Postgres connection after another error: %v", err)
 				return false, err2
 			}
 
@@ -145,6 +150,12 @@ func (c *Cluster) initDbConnWithName(dbname string) error {
 	conn.SetMaxOpenConns(1)
 	conn.SetMaxIdleConns(-1)
 
+	if c.pgDb != nil {
+		msg := "closing an existing connection before opening a new one to %s"
+		c.logger.Warningf(msg, dbname)
+		c.closeDbConn()
+	}
+
 	c.pgDb = conn
 
 	return nil
@@ -155,7 +166,7 @@ func (c *Cluster) connectionIsClosed() bool {
 }
 
 func (c *Cluster) closeDbConn() (err error) {
-	c.setProcessName("closing db connection")
+	c.setProcessName("closing database connection")
 	if c.pgDb != nil {
 		c.logger.Debug("closing database connection")
 		if err = c.pgDb.Close(); err != nil {
@@ -170,7 +181,7 @@ func (c *Cluster) closeDbConn() (err error) {
 }
 
 func (c *Cluster) readPgUsersFromDatabase(userNames []string) (users spec.PgUserMap, err error) {
-	c.setProcessName("reading users from the db")
+	c.setProcessName("reading users from the database")
 	var rows *sql.Rows
 	users = make(spec.PgUserMap)
 	if rows, err = c.pgDb.Query(getUserSQL, pq.Array(userNames)); err != nil {
@@ -465,8 +476,11 @@ func (c *Cluster) execCreateOrAlterExtension(extName, schemaName, statement, doi
 // perform remote authentification.
 func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
 	var stmtBytes bytes.Buffer
+
 	c.logger.Info("Installing lookup function")
 
+	// Open a new connection if not yet done. This connection will be used only
+	// to get the list of databases, not for the actuall installation.
 	if err := c.initDbConn(); err != nil {
 		return fmt.Errorf("could not init database connection")
 	}
@@ -480,36 +494,40 @@ func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
 		}
 	}()
 
+	// List of databases we failed to process. At the moment it function just
+	// like a flag to retry on the next sync, but in the future we may want to
+	// retry only necessary parts, so let's keep the list.
+	failedDatabases := []string{}
 	currentDatabases, err := c.getDatabases()
 	if err != nil {
 		msg := "could not get databases to install pooler lookup function: %v"
 		return fmt.Errorf(msg, err)
 	}
 
+	// We've got the list of target databases, now close this connection to
+	// open a new one to every each of them.
+	if err := c.closeDbConn(); err != nil {
+		c.logger.Errorf("could not close database connection: %v", err)
+	}
+
 	templater := template.Must(template.New("sql").Parse(connectionPoolerLookup))
+	params := TemplateParams{
+		"pooler_schema": poolerSchema,
+		"pooler_user":   poolerUser,
+	}
+
+	if err := templater.Execute(&stmtBytes, params); err != nil {
+		msg := "could not prepare sql statement %+v: %v"
+		return fmt.Errorf(msg, params, err)
+	}
 
 	for dbname := range currentDatabases {
+
 		if dbname == "template0" || dbname == "template1" {
 			continue
 		}
 
-		if err := c.initDbConnWithName(dbname); err != nil {
-			return fmt.Errorf("could not init database connection to %s", dbname)
-		}
-
-		c.logger.Infof("Install pooler lookup function into %s", dbname)
-
-		params := TemplateParams{
-			"pooler_schema": poolerSchema,
-			"pooler_user":   poolerUser,
-		}
-
-		if err := templater.Execute(&stmtBytes, params); err != nil {
-			c.logger.Errorf("could not prepare sql statement %+v: %v",
-				params, err)
-			// process other databases
-			continue
-		}
+		c.logger.Infof("install pooler lookup function into database '%s'", dbname)
 
 		// golang sql will do retries couple of times if pq driver reports
 		// connections issues (driver.ErrBadConn), but since our query is
@@ -520,7 +538,20 @@ func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
 			constants.PostgresConnectTimeout,
 			constants.PostgresConnectRetryTimeout,
 			func() (bool, error) {
-				if _, err := c.pgDb.Exec(stmtBytes.String()); err != nil {
+
+				// At this moment we are not connected to any database
+				if err := c.initDbConnWithName(dbname); err != nil {
+					msg := "could not init database connection to %s"
+					return false, fmt.Errorf(msg, dbname)
+				}
+				defer func() {
+					if err := c.closeDbConn(); err != nil {
+						msg := "could not close database connection: %v"
+						c.logger.Errorf(msg, err)
+					}
+				}()
+
+				if _, err = c.pgDb.Exec(stmtBytes.String()); err != nil {
 					msg := fmt.Errorf("could not execute sql statement %s: %v",
 						stmtBytes.String(), err)
 					return false, msg
@@ -533,15 +564,16 @@ func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
 			c.logger.Errorf("could not execute after retries %s: %v",
 				stmtBytes.String(), err)
 			// process other databases
+			failedDatabases = append(failedDatabases, dbname)
 			continue
 		}
 
 		c.logger.Infof("pooler lookup function installed into %s", dbname)
-		if err := c.closeDbConn(); err != nil {
-			c.logger.Errorf("could not close database connection: %v", err)
-		}
 	}
 
-	c.ConnectionPooler.LookupFunction = true
+	if len(failedDatabases) == 0 {
+		c.ConnectionPooler.LookupFunction = true
+	}
+
 	return nil
 }

pkg/cluster/k8sres.go

@@ -844,7 +844,7 @@ func (c *Cluster) getPodEnvironmentSecretVariables() ([]v1.EnvVar, error) {
 		return secretPodEnvVarsList, nil
 	}
 
-	secret, err := c.KubeClient.Secrets(c.OpConfig.PodEnvironmentSecret).Get(
+	secret, err := c.KubeClient.Secrets(c.Namespace).Get(
 		context.TODO(),
 		c.OpConfig.PodEnvironmentSecret,
 		metav1.GetOptions{})
@@ -912,51 +912,6 @@ func extractPgVersionFromBinPath(binPath string, template string) (string, error
 	return fmt.Sprintf("%v", pgVersion), nil
 }
 
-func (c *Cluster) getNewPgVersion(container v1.Container, newPgVersion string) (string, error) {
-	var (
-		spiloConfiguration spiloConfiguration
-		runningPgVersion   string
-		version            string
-		err                error
-	)
-
-	for _, env := range container.Env {
-
-		if env.Name == "PGVERSION" {
-			err = json.Unmarshal([]byte(env.Value), &version)
-			if err != nil {
-				return newPgVersion, err
-			}
-			return version, nil
-		}
-
-		if env.Name != "SPILO_CONFIGURATION" {
-			continue
-		}
-		err = json.Unmarshal([]byte(env.Value), &spiloConfiguration)
-		if err != nil {
-			return newPgVersion, err
-		}
-	}
-
-	if len(spiloConfiguration.PgLocalConfiguration) > 0 {
-		currentBinPath := fmt.Sprintf("%v", spiloConfiguration.PgLocalConfiguration[patroniPGBinariesParameterName])
-		runningPgVersion, err = extractPgVersionFromBinPath(currentBinPath, pgBinariesLocationTemplate)
-		if err != nil {
-			return "", fmt.Errorf("could not extract Postgres version from %v in SPILO_CONFIGURATION", currentBinPath)
-		}
-	} else {
-		return "", fmt.Errorf("could not find %q setting in SPILO_CONFIGURATION", patroniPGBinariesParameterName)
-	}
-
-	if runningPgVersion != newPgVersion {
-		c.logger.Warningf("postgresql version change(%q -> %q) has no effect", runningPgVersion, newPgVersion)
-		newPgVersion = runningPgVersion
-	}
-
-	return newPgVersion, nil
-}
-
 func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.StatefulSet, error) {
 
 	var (
@@ -1170,7 +1125,9 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 	}
 
 	// generate the spilo container
-	c.logger.Debugf("Generating Spilo container, environment variables: %v", spiloEnvVars)
+	c.logger.Debugf("Generating Spilo container, environment variables")
+	c.logger.Debugf("%v", spiloEnvVars)
+
 	spiloContainer := generateContainer(c.containerName(),
 		&effectiveDockerImage,
 		resourceRequirements,
@@ -1765,11 +1722,31 @@ func (c *Cluster) generateCloneEnvironment(description *acidv1.CloneDescription)
 			msg := "Figure out which S3 bucket to use from env"
 			c.logger.Info(msg, description.S3WalPath)
 
+			if c.OpConfig.WALES3Bucket != "" {
+				envs := []v1.EnvVar{
+					{
+						Name:  "CLONE_WAL_S3_BUCKET",
+						Value: c.OpConfig.WALES3Bucket,
+					},
+				}
+				result = append(result, envs...)
+			} else if c.OpConfig.WALGSBucket != "" {
+				envs := []v1.EnvVar{
+					{
+						Name:  "CLONE_WAL_GS_BUCKET",
+						Value: c.OpConfig.WALGSBucket,
+					},
+					{
+						Name:  "CLONE_GOOGLE_APPLICATION_CREDENTIALS",
+						Value: c.OpConfig.GCPCredentials,
+					},
+				}
+				result = append(result, envs...)
+			} else {
+				c.logger.Error("Cannot figure out S3 or GS bucket. Both are empty.")
+			}
+
 			envs := []v1.EnvVar{
-				{
-					Name:  "CLONE_WAL_S3_BUCKET",
-					Value: c.OpConfig.WALES3Bucket,
-				},
 				{
 					Name:  "CLONE_WAL_BUCKET_SCOPE_SUFFIX",
 					Value: getBucketScopeSuffix(description.UID),
@@ -2068,7 +2045,8 @@ func (c *Cluster) generateLogicalBackupPodEnvVars() []v1.EnvVar {
 		envVars = append(envVars, v1.EnvVar{Name: "AWS_SECRET_ACCESS_KEY", Value: c.OpConfig.LogicalBackup.LogicalBackupS3SecretAccessKey})
 	}
 
-	c.logger.Debugf("Generated logical backup env vars %v", envVars)
+	c.logger.Debugf("Generated logical backup env vars")
+	c.logger.Debugf("%v", envVars)
 	return envVars
 }
 
@@ -2229,6 +2207,13 @@ func (c *Cluster) generateConnectionPoolerPodTemplate(spec *acidv1.PostgresSpec)
 			},
 		},
 		Env: envVars,
+		ReadinessProbe: &v1.Probe{
+			Handler: v1.Handler{
+				TCPSocket: &v1.TCPSocketAction{
+					Port: intstr.IntOrString{IntVal: pgPort},
+				},
+			},
+		},
 	}
 
 	podTemplate := &v1.PodTemplateSpec{

pkg/cluster/k8sres_test.go

@@ -557,95 +557,6 @@ func TestExtractPgVersionFromBinPath(t *testing.T) {
 	}
 }
 
-func TestGetPgVersion(t *testing.T) {
-	testName := "TestGetPgVersion"
-	tests := []struct {
-		subTest          string
-		pgContainer      v1.Container
-		currentPgVersion string
-		newPgVersion     string
-	}{
-		{
-			subTest: "new version with decimal point differs from current SPILO_CONFIGURATION",
-			pgContainer: v1.Container{
-				Name: "postgres",
-				Env: []v1.EnvVar{
-					{
-						Name:  "SPILO_CONFIGURATION",
-						Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/9.6/bin\"}}",
-					},
-				},
-			},
-			currentPgVersion: "9.6",
-			newPgVersion:     "12",
-		},
-		{
-			subTest: "new version differs from current SPILO_CONFIGURATION",
-			pgContainer: v1.Container{
-				Name: "postgres",
-				Env: []v1.EnvVar{
-					{
-						Name:  "SPILO_CONFIGURATION",
-						Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/11/bin\"}}",
-					},
-				},
-			},
-			currentPgVersion: "11",
-			newPgVersion:     "12",
-		},
-		{
-			subTest: "new version is lower than the one found in current SPILO_CONFIGURATION",
-			pgContainer: v1.Container{
-				Name: "postgres",
-				Env: []v1.EnvVar{
-					{
-						Name:  "SPILO_CONFIGURATION",
-						Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/12/bin\"}}",
-					},
-				},
-			},
-			currentPgVersion: "12",
-			newPgVersion:     "11",
-		},
-		{
-			subTest: "new version is the same like in the current SPILO_CONFIGURATION",
-			pgContainer: v1.Container{
-				Name: "postgres",
-				Env: []v1.EnvVar{
-					{
-						Name:  "SPILO_CONFIGURATION",
-						Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/12/bin\"}}",
-					},
-				},
-			},
-			currentPgVersion: "12",
-			newPgVersion:     "12",
-		},
-	}
-
-	var cluster = New(
-		Config{
-			OpConfig: config.Config{
-				ProtectedRoles: []string{"admin"},
-				Auth: config.Auth{
-					SuperUsername:       superUserName,
-					ReplicationUsername: replicationUserName,
-				},
-			},
-		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
-
-	for _, tt := range tests {
-		pgVersion, err := cluster.getNewPgVersion(tt.pgContainer, tt.newPgVersion)
-		if err != nil {
-			t.Errorf("Unexpected error: %v", err)
-		}
-		if pgVersion != tt.currentPgVersion {
-			t.Errorf("%s %s: Expected version %s, have %s instead",
-				testName, tt.subTest, tt.currentPgVersion, pgVersion)
-		}
-	}
-}
-
 func TestSecretVolume(t *testing.T) {
 	testName := "TestSecretVolume"
 	tests := []struct {

pkg/cluster/pod.go

@@ -304,9 +304,16 @@ func (c *Cluster) isSafeToRecreatePods(pods *v1.PodList) bool {
 	 after this check succeeds but before a pod is re-created
 	*/
 
+	for _, pod := range pods.Items {
+		c.logger.Debugf("name=%s phase=%s ip=%s", pod.Name, pod.Status.Phase, pod.Status.PodIP)
+	}
+
 	for _, pod := range pods.Items {
 		state, err := c.patroni.GetPatroniMemberState(&pod)
-		if err != nil || state == "creating replica" {
+		if err != nil {
+			c.logger.Errorf("failed to get Patroni state for pod: %s", err)
+			return false
+		} else if state == "creating replica" {
 			c.logger.Warningf("cannot re-create replica %s: it is currently being initialized", pod.Name)
 			return false
 		}

pkg/cluster/resources.go

@@ -293,7 +293,7 @@ func (c *Cluster) preScaleDown(newStatefulSet *appsv1.StatefulSet) error {
 
 // setRollingUpdateFlagForStatefulSet sets the indicator or the rolling update requirement
 // in the StatefulSet annotation.
-func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *appsv1.StatefulSet, val bool) {
+func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *appsv1.StatefulSet, val bool, msg string) {
 	anno := sset.GetAnnotations()
 	if anno == nil {
 		anno = make(map[string]string)
@@ -301,13 +301,13 @@ func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *appsv1.StatefulSet, v
 
 	anno[rollingUpdateStatefulsetAnnotationKey] = strconv.FormatBool(val)
 	sset.SetAnnotations(anno)
-	c.logger.Debugf("statefulset's rolling update annotation has been set to %t", val)
+	c.logger.Debugf("set statefulset's rolling update annotation to %t: caller/reason %s", val, msg)
 }
 
 // applyRollingUpdateFlagforStatefulSet sets the rolling update flag for the cluster's StatefulSet
 // and applies that setting to the actual running cluster.
 func (c *Cluster) applyRollingUpdateFlagforStatefulSet(val bool) error {
-	c.setRollingUpdateFlagForStatefulSet(c.Statefulset, val)
+	c.setRollingUpdateFlagForStatefulSet(c.Statefulset, val, "applyRollingUpdateFlag")
 	sset, err := c.updateStatefulSetAnnotations(c.Statefulset.GetAnnotations())
 	if err != nil {
 		return err
@@ -359,14 +359,13 @@ func (c *Cluster) mergeRollingUpdateFlagUsingCache(runningStatefulSet *appsv1.St
 			podsRollingUpdateRequired = false
 		} else {
 			c.logger.Infof("found a statefulset with an unfinished rolling update of the pods")
-
 		}
 	}
 	return podsRollingUpdateRequired
 }
 
 func (c *Cluster) updateStatefulSetAnnotations(annotations map[string]string) (*appsv1.StatefulSet, error) {
-	c.logger.Debugf("updating statefulset annotations")
+	c.logger.Debugf("patching statefulset annotations")
 	patchData, err := metaAnnotationsPatch(annotations)
 	if err != nil {
 		return nil, fmt.Errorf("could not form patch for the statefulset metadata: %v", err)

pkg/cluster/sync.go

@@ -332,29 +332,17 @@ func (c *Cluster) syncStatefulSet() error {
 		// statefulset is already there, make sure we use its definition in order to compare with the spec.
 		c.Statefulset = sset
 
-		// check if there is no Postgres version mismatch
-		for _, container := range c.Statefulset.Spec.Template.Spec.Containers {
-			if container.Name != "postgres" {
-				continue
-			}
-			pgVersion, err := c.getNewPgVersion(container, c.Spec.PostgresqlParam.PgVersion)
-			if err != nil {
-				return fmt.Errorf("could not parse current Postgres version: %v", err)
-			}
-			c.Spec.PostgresqlParam.PgVersion = pgVersion
-		}
-
 		desiredSS, err := c.generateStatefulSet(&c.Spec)
 		if err != nil {
 			return fmt.Errorf("could not generate statefulset: %v", err)
 		}
-		c.setRollingUpdateFlagForStatefulSet(desiredSS, podsRollingUpdateRequired)
+		c.setRollingUpdateFlagForStatefulSet(desiredSS, podsRollingUpdateRequired, "from cache")
 
 		cmp := c.compareStatefulSetWith(desiredSS)
 		if !cmp.match {
 			if cmp.rollingUpdate && !podsRollingUpdateRequired {
 				podsRollingUpdateRequired = true
-				c.setRollingUpdateFlagForStatefulSet(desiredSS, podsRollingUpdateRequired)
+				c.setRollingUpdateFlagForStatefulSet(desiredSS, podsRollingUpdateRequired, "statefulset changes")
 			}
 
 			c.logStatefulSetChanges(c.Statefulset, desiredSS, false, cmp.reasons)
@@ -497,11 +485,11 @@ func (c *Cluster) syncSecrets() error {
 				return fmt.Errorf("could not get current secret: %v", err)
 			}
 			if secretUsername != string(secret.Data["username"]) {
-				c.logger.Warningf("secret %q does not contain the role %q", secretSpec.Name, secretUsername)
+				c.logger.Warningf("secret %s does not contain the role %q", secretSpec.Name, secretUsername)
 				continue
 			}
 			c.Secrets[secret.UID] = secret
-			c.logger.Debugf("secret %q already exists, fetching its password", util.NameFromMeta(secret.ObjectMeta))
+			c.logger.Debugf("secret %s already exists, fetching its password", util.NameFromMeta(secret.ObjectMeta))
 			if secretUsername == c.systemUsers[constants.SuperuserKeyName].Name {
 				secretUsername = constants.SuperuserKeyName
 				userMap = c.systemUsers
@@ -804,7 +792,7 @@ func (c *Cluster) syncLogicalBackupJob() error {
 			return fmt.Errorf("could not generate the desired logical backup job state: %v", err)
 		}
 		if match, reason := k8sutil.SameLogicalBackupJob(job, desiredJob); !match {
-			c.logger.Infof("logical job %q is not in the desired state and needs to be updated",
+			c.logger.Infof("logical job %s is not in the desired state and needs to be updated",
 				c.getLogicalBackupJobName(),
 			)
 			if reason != "" {
@@ -825,12 +813,12 @@ func (c *Cluster) syncLogicalBackupJob() error {
 	c.logger.Info("could not find the cluster's logical backup job")
 
 	if err = c.createLogicalBackupJob(); err == nil {
-		c.logger.Infof("created missing logical backup job %q", jobName)
+		c.logger.Infof("created missing logical backup job %s", jobName)
 	} else {
 		if !k8sutil.ResourceAlreadyExists(err) {
 			return fmt.Errorf("could not create missing logical backup job: %v", err)
 		}
-		c.logger.Infof("logical backup job %q already exists", jobName)
+		c.logger.Infof("logical backup job %s already exists", jobName)
 		if _, err = c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Get(context.TODO(), jobName, metav1.GetOptions{}); err != nil {
 			return fmt.Errorf("could not fetch existing logical backup job: %v", err)
 		}
@@ -847,7 +835,9 @@ func (c *Cluster) syncConnectionPooler(oldSpec,
 	var err error
 
 	if c.ConnectionPooler == nil {
-		c.ConnectionPooler = &ConnectionPoolerObjects{}
+		c.ConnectionPooler = &ConnectionPoolerObjects{
+			LookupFunction: false,
+		}
 	}
 
 	newNeedConnectionPooler := c.needConnectionPoolerWorker(&newSpec.Spec)
@@ -885,6 +875,11 @@ func (c *Cluster) syncConnectionPooler(oldSpec,
 			if err = lookup(schema, user); err != nil {
 				return NoSync, err
 			}
+		} else {
+			// Lookup function installation seems to be a fragile point, so
+			// let's log for debugging if we skip it
+			msg := "Skip lookup function installation, old: %d, already installed %d"
+			c.logger.Debug(msg, oldNeedConnectionPooler, c.ConnectionPooler.LookupFunction)
 		}
 
 		if reason, err = c.syncConnectionPoolerWorker(oldSpec, newSpec); err != nil {
@@ -968,7 +963,7 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 			newConnectionPooler = &acidv1.ConnectionPooler{}
 		}
 
-		c.logger.Infof("Old: %+v, New %+v", oldConnectionPooler, newConnectionPooler)
+		logNiceDiff(c.logger, oldConnectionPooler, newConnectionPooler)
 
 		specSync, specReason := c.needSyncConnectionPoolerSpecs(oldConnectionPooler, newConnectionPooler)
 		defaultsSync, defaultsReason := c.needSyncConnectionPoolerDefaults(newConnectionPooler, deployment)

pkg/cluster/util.go

@@ -18,12 +18,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 
+	"github.com/sirupsen/logrus"
 	acidzalando "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do"
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/spec"
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
+	"github.com/zalando/postgres-operator/pkg/util/nicediff"
 	"github.com/zalando/postgres-operator/pkg/util/retryutil"
 )
 
@@ -166,40 +168,59 @@ func (c *Cluster) logPDBChanges(old, new *policybeta1.PodDisruptionBudget, isUpd
 		)
 	}
 
-	c.logger.Debugf("diff\n%s\n", util.PrettyDiff(old.Spec, new.Spec))
+	logNiceDiff(c.logger, old.Spec, new.Spec)
+}
+
+func logNiceDiff(log *logrus.Entry, old, new interface{}) {
+	o, erro := json.MarshalIndent(old, "", "  ")
+	n, errn := json.MarshalIndent(new, "", "  ")
+
+	if erro != nil || errn != nil {
+		panic("could not marshal API objects, should not happen")
+	}
+
+	nice := nicediff.Diff(string(o), string(n), true)
+	for _, s := range strings.Split(nice, "\n") {
+		// " is not needed in the value to understand
+		log.Debugf(strings.ReplaceAll(s, "\"", ""))
+	}
 }
 
 func (c *Cluster) logStatefulSetChanges(old, new *appsv1.StatefulSet, isUpdate bool, reasons []string) {
 	if isUpdate {
-		c.logger.Infof("statefulset %q has been changed", util.NameFromMeta(old.ObjectMeta))
+		c.logger.Infof("statefulset %s has been changed", util.NameFromMeta(old.ObjectMeta))
 	} else {
-		c.logger.Infof("statefulset %q is not in the desired state and needs to be updated",
+		c.logger.Infof("statefulset %s is not in the desired state and needs to be updated",
 			util.NameFromMeta(old.ObjectMeta),
 		)
 	}
+
+	logNiceDiff(c.logger, old.Spec, new.Spec)
+
 	if !reflect.DeepEqual(old.Annotations, new.Annotations) {
-		c.logger.Debugf("metadata.annotation diff\n%s\n", util.PrettyDiff(old.Annotations, new.Annotations))
+		c.logger.Debugf("metadata.annotation are different")
+		logNiceDiff(c.logger, old.Annotations, new.Annotations)
 	}
-	c.logger.Debugf("spec diff between old and new statefulsets: \n%s\n", util.PrettyDiff(old.Spec, new.Spec))
 
 	if len(reasons) > 0 {
 		for _, reason := range reasons {
-			c.logger.Infof("reason: %q", reason)
+			c.logger.Infof("reason: %s", reason)
 		}
 	}
 }
 
 func (c *Cluster) logServiceChanges(role PostgresRole, old, new *v1.Service, isUpdate bool, reason string) {
 	if isUpdate {
-		c.logger.Infof("%s service %q has been changed",
+		c.logger.Infof("%s service %s has been changed",
 			role, util.NameFromMeta(old.ObjectMeta),
 		)
 	} else {
-		c.logger.Infof("%s service %q is not in the desired state and needs to be updated",
+		c.logger.Infof("%s service %s is not in the desired state and needs to be updated",
 			role, util.NameFromMeta(old.ObjectMeta),
 		)
 	}
-	c.logger.Debugf("diff\n%s\n", util.PrettyDiff(old.Spec, new.Spec))
+
+	logNiceDiff(c.logger, old.Spec, new.Spec)
 
 	if reason != "" {
 		c.logger.Infof("reason: %s", reason)
@@ -208,7 +229,7 @@ func (c *Cluster) logServiceChanges(role PostgresRole, old, new *v1.Service, isU
 
 func (c *Cluster) logVolumeChanges(old, new acidv1.Volume) {
 	c.logger.Infof("volume specification has been changed")
-	c.logger.Debugf("diff\n%s\n", util.PrettyDiff(old, new))
+	logNiceDiff(c.logger, old, new)
 }
 
 func (c *Cluster) getTeamMembers(teamID string) ([]string, error) {
@@ -217,24 +238,37 @@ func (c *Cluster) getTeamMembers(teamID string) ([]string, error) {
 		return nil, fmt.Errorf("no teamId specified")
 	}
 
+	c.logger.Debugf("fetching possible additional team members for team %q", teamID)
+	members := []string{}
+	additionalMembers := c.PgTeamMap[c.Spec.TeamID].AdditionalMembers
+	for _, member := range additionalMembers {
+		members = append(members, member)
+	}
+
 	if !c.OpConfig.EnableTeamsAPI {
-		c.logger.Debugf("team API is disabled, returning empty list of members for team %q", teamID)
-		return []string{}, nil
+		c.logger.Debugf("team API is disabled, only returning %d members for team %q", len(members), teamID)
+		return members, nil
 	}
 
 	token, err := c.oauthTokenGetter.getOAuthToken()
 	if err != nil {
-		c.logger.Warnf("could not get oauth token to authenticate to team service API, returning empty list of team members: %v", err)
-		return []string{}, nil
+		c.logger.Warnf("could not get oauth token to authenticate to team service API, only returning %d members for team %q: %v", len(members), teamID, err)
+		return members, nil
 	}
 
 	teamInfo, err := c.teamsAPIClient.TeamInfo(teamID, token)
 	if err != nil {
-		c.logger.Warnf("could not get team info for team %q, returning empty list of team members: %v", teamID, err)
-		return []string{}, nil
+		c.logger.Warnf("could not get team info for team %q, only returning %d members: %v", teamID, len(members), err)
+		return members, nil
 	}
 
-	return teamInfo.Members, nil
+	for _, member := range teamInfo.Members {
+		if !(util.SliceContains(members, member)) {
+			members = append(members, member)
+		}
+	}
+
+	return members, nil
 }
 
 func (c *Cluster) waitForPodLabel(podEvents chan PodEvent, stopChan chan struct{}, role *PostgresRole) (*v1.Pod, error) {