resolve conflicts

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

@@ -534,6 +534,9 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
 							"synchronous_mode_strict": {
 								Type: "boolean",
 							},
+							"synchronous_node_count": {
+								Type: "integer",
+							},
 							"ttl": {
 								Type: "integer",
 							},

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

@@ -165,6 +165,7 @@ type Patroni struct {
 	Slots                 map[string]map[string]string `json:"slots,omitempty"`
 	SynchronousMode       bool                         `json:"synchronous_mode,omitempty"`
 	SynchronousModeStrict bool                         `json:"synchronous_mode_strict,omitempty"`
+	SynchronousNodeCount  uint32                       `json:"synchronous_node_count,omitempty" defaults:1`
 }
 
 // StandbyDescription contains s3 wal path

pkg/cluster/k8sres.go

@@ -50,6 +50,7 @@ type patroniDCS struct {
 	MaximumLagOnFailover     float32                      `json:"maximum_lag_on_failover,omitempty"`
 	SynchronousMode          bool                         `json:"synchronous_mode,omitempty"`
 	SynchronousModeStrict    bool                         `json:"synchronous_mode_strict,omitempty"`
+	SynchronousNodeCount     uint32                       `json:"synchronous_node_count,omitempty"`
 	PGBootstrapConfiguration map[string]interface{}       `json:"postgresql,omitempty"`
 	Slots                    map[string]map[string]string `json:"slots,omitempty"`
 }
@@ -262,6 +263,9 @@ PatroniInitDBParams:
 	if patroni.SynchronousModeStrict {
 		config.Bootstrap.DCS.SynchronousModeStrict = patroni.SynchronousModeStrict
 	}
+	if patroni.SynchronousNodeCount >= 1 {
+		config.Bootstrap.DCS.SynchronousNodeCount = patroni.SynchronousNodeCount
+	}
 
 	config.PgLocalConfiguration = make(map[string]interface{})
 

pkg/cluster/k8sres_test.go

@@ -91,11 +91,12 @@ func TestGenerateSpiloJSONConfiguration(t *testing.T) {
 				MaximumLagOnFailover:  33554432,
 				SynchronousMode:       true,
 				SynchronousModeStrict: true,
+				SynchronousNodeCount:  1,
 				Slots:                 map[string]map[string]string{"permanent_logical_1": {"type": "logical", "database": "foo", "plugin": "pgoutput"}},
 			},
 			role:     "zalandos",
 			opConfig: config.Config{},
-			result:   `{"postgresql":{"bin_dir":"/usr/lib/postgresql/11/bin","pg_hba":["hostssl all all 0.0.0.0/0 md5","host    all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"zalandos":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"synchronous_mode":true,"synchronous_mode_strict":true,"slots":{"permanent_logical_1":{"database":"foo","plugin":"pgoutput","type":"logical"}}}}}`,
+			result:   `{"postgresql":{"bin_dir":"/usr/lib/postgresql/11/bin","pg_hba":["hostssl all all 0.0.0.0/0 md5","host    all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"zalandos":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"synchronous_mode":true,"synchronous_mode_strict":true,"synchronous_node_count":1,"slots":{"permanent_logical_1":{"database":"foo","plugin":"pgoutput","type":"logical"}}}}}`,
 		},
 	}
 	for _, tt := range tests {

pkg/cluster/pod.go

@@ -13,6 +13,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 
+	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/patroni"
@@ -349,6 +350,54 @@ func (c *Cluster) MigrateReplicaPod(podName spec.NamespacedName, fromNodeName st
 	return nil
 }
 
+func (c *Cluster) getPatroniConfig(pod *v1.Pod) (acidv1.Patroni, map[string]string, error) {
+	var (
+		patroniConfig acidv1.Patroni
+		pgParameters  map[string]string
+	)
+	podName := util.NameFromMeta(pod.ObjectMeta)
+	err := retryutil.Retry(1*time.Second, 5*time.Second,
+		func() (bool, error) {
+			var err error
+			patroniConfig, pgParameters, err = c.patroni.GetConfig(pod)
+
+			if err != nil {
+				return false, err
+			}
+			return true, nil
+		},
+	)
+
+	if err != nil {
+		return acidv1.Patroni{}, nil, fmt.Errorf("could not get Postgres config from pod %s: %v", podName, err)
+	}
+
+	return patroniConfig, pgParameters, nil
+}
+
+func (c *Cluster) getPatroniMemberData(pod *v1.Pod) (patroni.MemberData, error) {
+	var memberData patroni.MemberData
+	err := retryutil.Retry(1*time.Second, 5*time.Second,
+		func() (bool, error) {
+			var err error
+			memberData, err = c.patroni.GetMemberData(pod)
+
+			if err != nil {
+				return false, err
+			}
+			return true, nil
+		},
+	)
+	if err != nil {
+		return patroni.MemberData{}, fmt.Errorf("could not get member data: %v", err)
+	}
+	if memberData.State == "creating replica" {
+		return patroni.MemberData{}, fmt.Errorf("replica currently being initialized")
+	}
+
+	return memberData, nil
+}
+
 func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
 	ch := c.registerPodSubscriber(podName)
 	defer c.unregisterPodSubscriber(podName)
@@ -380,54 +429,10 @@ func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
 	return pod, nil
 }
 
-func (c *Cluster) isSafeToRecreatePods(pods []v1.Pod) 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 {
-		c.logger.Debugf("name=%s phase=%s ip=%s", pod.Name, pod.Status.Phase, pod.Status.PodIP)
-	}
-
-	for _, pod := range pods {
-
-		var data patroni.MemberData
-
-		err := retryutil.Retry(1*time.Second, 5*time.Second,
-			func() (bool, error) {
-				var err error
-				data, err = c.patroni.GetMemberData(&pod)
-
-				if err != nil {
-					return false, err
-				}
-				return true, nil
-			},
-		)
-
-		if err != nil {
-			c.logger.Errorf("failed to get Patroni state for pod: %s", err)
-			return false
-		} else if data.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(pods []v1.Pod, switchoverCandidates []spec.NamespacedName) error {
 	c.setProcessName("starting to recreate pods")
 	c.logger.Infof("there are %d pods in the cluster to recreate", len(pods))
 
-	if !c.isSafeToRecreatePods(pods) {
-		return fmt.Errorf("postpone pod recreation until next Sync: recreation is unsafe because pods are being initialized")
-	}
-
 	var (
 		masterPod, newMasterPod *v1.Pod
 	)

pkg/cluster/sync.go

@@ -285,6 +285,7 @@ func (c *Cluster) syncStatefulSet() error {
 		restartMasterFirst bool
 	)
 	podsToRecreate := make([]v1.Pod, 0)
+	isSafeToRecreatePods := true
 	switchoverCandidates := make([]spec.NamespacedName, 0)
 
 	pods, err := c.listPods()
@@ -410,23 +411,21 @@ func (c *Cluster) syncStatefulSet() error {
 	// get Postgres config, compare with manifest and update via Patroni PATCH endpoint if it differs
 	// Patroni's config endpoint is just a "proxy" to DCS. It is enough to patch it only once and it doesn't matter which pod is used
 	for i, pod := range pods {
-		emptyPatroniConfig := acidv1.Patroni{}
-		podName := util.NameFromMeta(pods[i].ObjectMeta)
-		patroniConfig, pgParameters, err := c.patroni.GetConfig(&pod)
+		patroniConfig, pgParameters, err := c.getPatroniConfig(&pod)
 		if err != nil {
-			c.logger.Warningf("could not get Postgres config from pod %s: %v", podName, err)
+			c.logger.Warningf("%v", err)
+			isSafeToRecreatePods = false
 			continue
 		}
 		restartWait = patroniConfig.LoopWait
 
 		// empty config probably means cluster is not fully initialized yet, e.g. restoring from backup
 		// do not attempt a restart
-		if !reflect.DeepEqual(patroniConfig, emptyPatroniConfig) || len(pgParameters) > 0 {
+		if !reflect.DeepEqual(patroniConfig, acidv1.Patroni{}) || len(pgParameters) > 0 {
 			// compare config returned from Patroni with what is specified in the manifest
 			restartMasterFirst, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, patroniConfig, c.Spec.Patroni, pgParameters, c.Spec.Parameters)
-
 			if err != nil {
-				c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", podName, err)
+				c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", pods[i].Name, err)
 				continue
 			}
 
@@ -448,50 +447,59 @@ func (c *Cluster) syncStatefulSet() error {
 			remainingPods = append(remainingPods, &pods[i])
 			continue
 		}
-		c.restartInstance(&pod, restartWait)
+		if err = c.restartInstance(&pod, restartWait); err != nil {
+			c.logger.Errorf("%v", err)
+			isSafeToRecreatePods = false
+		}
 	}
 
 	// in most cases only the master should be left to restart
 	if len(remainingPods) > 0 {
 		for _, remainingPod := range remainingPods {
-			c.restartInstance(remainingPod, restartWait)
+			if err = c.restartInstance(remainingPod, restartWait); err != nil {
+				c.logger.Errorf("%v", err)
+				isSafeToRecreatePods = false
+			}
 		}
 	}
 
 	// if we get here we also need to re-create the pods (either leftovers from the old
 	// statefulset or those that got their configuration from the outdated statefulset)
 	if len(podsToRecreate) > 0 {
-		c.logger.Debugln("performing rolling update")
-		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Performing rolling update")
-		if err := c.recreatePods(podsToRecreate, switchoverCandidates); err != nil {
-			return fmt.Errorf("could not recreate pods: %v", err)
+		if isSafeToRecreatePods {
+			c.logger.Debugln("performing rolling update")
+			c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Performing rolling update")
+			if err := c.recreatePods(podsToRecreate, switchoverCandidates); err != nil {
+				return fmt.Errorf("could not recreate pods: %v", err)
+			}
+			c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Rolling update done - pods have been recreated")
+		} else {
+			c.logger.Warningf("postpone pod recreation until next sync")
 		}
-		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Rolling update done - pods have been recreated")
 	}
 	return nil
 }
 
-func (c *Cluster) restartInstance(pod *v1.Pod, restartWait uint32) {
+func (c *Cluster) restartInstance(pod *v1.Pod, restartWait uint32) error {
+	// if the config update requires a restart, call Patroni restart
 	podName := util.NameFromMeta(pod.ObjectMeta)
 	role := PostgresRole(pod.Labels[c.OpConfig.PodRoleLabel])
-
-	// if the config update requires a restart, call Patroni restart
-	memberData, err := c.patroni.GetMemberData(pod)
+	memberData, err := c.getPatroniMemberData(pod)
 	if err != nil {
-		c.logger.Debugf("could not get member data of %s pod %s - skipping possible restart attempt: %v", role, podName, err)
-		return
+		return fmt.Errorf("could not restart Postgres in %s pod %s: %v", role, podName, err)
 	}
 
 	// do restart only when it is pending
 	if memberData.PendingRestart {
-		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("restarting Postgres server within %s pod %s", role, pod.Name))
+		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("restarting Postgres server within %s pod %s", role, podName))
 		if err := c.patroni.Restart(pod); err != nil {
-			c.logger.Warningf("could not restart Postgres server within %s pod %s: %v", role, podName, err)
-			return
+			return err
 		}
 		time.Sleep(time.Duration(restartWait) * time.Second)
-		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("Postgres server restart done for %s pod %s", role, pod.Name))
+		c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", fmt.Sprintf("Postgres server restart done for %s pod %s", role, podName))
 	}
+
+	return nil
 }
 
 // AnnotationsToPropagate get the annotations to update if required
@@ -620,11 +628,6 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectiv
 	return requiresMasterRestart, nil
 }
 
-func (c *Cluster) getNextRotationDate(currentDate time.Time) (time.Time, string) {
-	nextRotationDate := currentDate.AddDate(0, 0, int(c.OpConfig.PasswordRotationInterval))
-	return nextRotationDate, nextRotationDate.Format("2006-01-02 15:04:05")
-}
-
 func (c *Cluster) syncSecrets() error {
 
 	c.logger.Info("syncing secrets")
@@ -682,6 +685,11 @@ func (c *Cluster) syncSecrets() error {
 	return nil
 }
 
+func (c *Cluster) getNextRotationDate(currentDate time.Time) (time.Time, string) {
+	nextRotationDate := currentDate.AddDate(0, 0, int(c.OpConfig.PasswordRotationInterval))
+	return nextRotationDate, nextRotationDate.Format(time.RFC3339)
+}
+
 func (c *Cluster) updateSecret(
 	secretUsername string,
 	generatedSecret *v1.Secret,
@@ -727,7 +735,7 @@ func (c *Cluster) updateSecret(
 
 		// initialize password rotation setting first rotation date
 		nextRotationDateStr = string(secret.Data["nextRotation"])
-		if nextRotationDate, err = time.ParseInLocation("2006-01-02 15:04:05", nextRotationDateStr, time.Local); err != nil {
+		if nextRotationDate, err = time.ParseInLocation(time.RFC3339, nextRotationDateStr, currentTime.UTC().Location()); err != nil {
 			nextRotationDate, nextRotationDateStr = c.getNextRotationDate(currentTime)
 			secret.Data["nextRotation"] = []byte(nextRotationDateStr)
 			updateSecret = true
@@ -736,7 +744,7 @@ func (c *Cluster) updateSecret(
 
 		// check if next rotation can happen sooner
 		// if rotation interval has been decreased
-		currentRotationDate, _ := c.getNextRotationDate(currentTime)
+		currentRotationDate, nextRotationDateStr := c.getNextRotationDate(currentTime)
 		if nextRotationDate.After(currentRotationDate) {
 			nextRotationDate = currentRotationDate
 		}
@@ -756,8 +764,6 @@ func (c *Cluster) updateSecret(
 				*retentionUsers = append(*retentionUsers, secretUsername)
 			}
 			secret.Data["password"] = []byte(util.RandomPassword(constants.PasswordLength))
-
-			_, nextRotationDateStr = c.getNextRotationDate(nextRotationDate)
 			secret.Data["nextRotation"] = []byte(nextRotationDateStr)
 
 			updateSecret = true

pkg/cluster/sync_test.go

@@ -270,13 +270,29 @@ func TestUpdateSecret(t *testing.T) {
 
 	clusterName := "acid-test-cluster"
 	namespace := "default"
-	username := "foo"
+	dbname := "app"
+	dbowner := "appowner"
 	secretTemplate := config.StringTemplate("{username}.{cluster}.credentials")
 	rotationUsers := make(spec.PgUserMap)
 	retentionUsers := make([]string, 0)
-	yesterday := time.Now().AddDate(0, 0, -1)
 
-	// new cluster with pvc storage resize mode and configured labels
+	// define manifest users and enable rotation for dbowner
+	pg := acidv1.Postgresql{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      clusterName,
+			Namespace: namespace,
+		},
+		Spec: acidv1.PostgresSpec{
+			Databases:                      map[string]string{dbname: dbowner},
+			Users:                          map[string]acidv1.UserFlags{"foo": {}, dbowner: {}},
+			UsersWithInPlaceSecretRotation: []string{dbowner},
+			Volume: acidv1.Volume{
+				Size: "1Gi",
+			},
+		},
+	}
+
+	// new cluster with enabled password rotation
 	var cluster = New(
 		Config{
 			OpConfig: config.Config{
@@ -291,44 +307,61 @@ func TestUpdateSecret(t *testing.T) {
 					ClusterNameLabel: "cluster-name",
 				},
 			},
-		}, client, acidv1.Postgresql{}, logger, eventRecorder)
+		}, client, pg, logger, eventRecorder)
 
 	cluster.Name = clusterName
 	cluster.Namespace = namespace
 	cluster.pgUsers = map[string]spec.PgUser{}
-	cluster.Spec.Users = map[string]acidv1.UserFlags{username: {}}
 	cluster.initRobotUsers()
 
-	// create a secret for user foo
+	// create secrets
+	cluster.syncSecrets()
+	// initialize rotation with current time
 	cluster.syncSecrets()
 
-	secret, err := cluster.KubeClient.Secrets(namespace).Get(context.TODO(), secretTemplate.Format("username", username, "cluster", clusterName), metav1.GetOptions{})
-	assert.NoError(t, err)
-	generatedSecret := cluster.Secrets[secret.UID]
+	dayAfterTomorrow := time.Now().AddDate(0, 0, 2)
 
-	// now update the secret setting next rotation date (yesterday + interval)
-	cluster.updateSecret(username, generatedSecret, &rotationUsers, &retentionUsers, yesterday)
-	updatedSecret, err := cluster.KubeClient.Secrets(namespace).Get(context.TODO(), secretTemplate.Format("username", username, "cluster", clusterName), metav1.GetOptions{})
-	assert.NoError(t, err)
+	for username := range cluster.Spec.Users {
+		pgUser := cluster.pgUsers[username]
 
-	nextRotation := string(updatedSecret.Data["nextRotation"])
-	_, nextRotationDate := cluster.getNextRotationDate(yesterday)
-	if nextRotation != nextRotationDate {
-		t.Errorf("%s: updated secret does not contain correct rotation date: expected %s, got %s", testName, nextRotationDate, nextRotation)
-	}
+		// first, get the secret
+		secret, err := cluster.KubeClient.Secrets(namespace).Get(context.TODO(), secretTemplate.Format("username", username, "cluster", clusterName), metav1.GetOptions{})
+		assert.NoError(t, err)
+		secretPassword := string(secret.Data["password"])
 
-	// update secret again but use current time to trigger rotation
-	cluster.updateSecret(username, generatedSecret, &rotationUsers, &retentionUsers, time.Now())
-	updatedSecret, err = cluster.KubeClient.Secrets(namespace).Get(context.TODO(), secretTemplate.Format("username", username, "cluster", clusterName), metav1.GetOptions{})
-	assert.NoError(t, err)
+		// now update the secret setting a next rotation date (tomorrow + interval)
+		cluster.updateSecret(username, secret, &rotationUsers, &retentionUsers, dayAfterTomorrow)
+		updatedSecret, err := cluster.KubeClient.Secrets(namespace).Get(context.TODO(), secretTemplate.Format("username", username, "cluster", clusterName), metav1.GetOptions{})
+		assert.NoError(t, err)
 
-	if len(rotationUsers) != 1 && len(retentionUsers) != 1 {
-		t.Errorf("%s: unexpected number of users to rotate - expected only foo, found %d", testName, len(rotationUsers))
-	}
+		// check that passwords are different
+		rotatedPassword := string(updatedSecret.Data["password"])
+		if secretPassword == rotatedPassword {
+			t.Errorf("%s: password unchanged in updated secret for %s", testName, username)
+		}
 
-	secretUsername := string(updatedSecret.Data["username"])
-	rotatedUsername := username + time.Now().Format("060102")
-	if secretUsername != rotatedUsername {
-		t.Errorf("%s: updated secret does not contain correct username: expected %s, got %s", testName, rotatedUsername, secretUsername)
+		// check that next rotation date is tomorrow + interval, not date in secret + interval
+		nextRotation := string(updatedSecret.Data["nextRotation"])
+		_, nextRotationDate := cluster.getNextRotationDate(dayAfterTomorrow)
+		if nextRotation != nextRotationDate {
+			t.Errorf("%s: updated secret of %s does not contain correct rotation date: expected %s, got %s", testName, username, nextRotationDate, nextRotation)
+		}
+
+		// compare username, when it's dbowner they should be equal because of UsersWithInPlaceSecretRotation
+		secretUsername := string(updatedSecret.Data["username"])
+		if pgUser.IsDbOwner {
+			if secretUsername != username {
+				t.Errorf("%s: username differs in updated secret: expected %s, got %s", testName, username, secretUsername)
+			}
+		} else {
+			rotatedUsername := username + dayAfterTomorrow.Format("060102")
+			if secretUsername != rotatedUsername {
+				t.Errorf("%s: updated secret does not contain correct username: expected %s, got %s", testName, rotatedUsername, secretUsername)
+			}
+
+			if len(rotationUsers) != 1 && len(retentionUsers) != 1 {
+				t.Errorf("%s: unexpected number of users to rotate - expected only %s, found %d", testName, username, len(rotationUsers))
+			}
+		}
 	}
 }