take over several changes

Author: Polina Bungina

go.mod

@@ -33,6 +33,7 @@ require (
 )
 
 require (
+	github.com/Masterminds/semver v1.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect

go.sum

@@ -1,6 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
+github.com/Masterminds/semver v1.5.0/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=

pkg/cluster/majorversionupgrade.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/Masterminds/semver"
 	"github.com/cybertec-postgresql/cybertec-pg-operator/pkg/spec"
 	"github.com/cybertec-postgresql/cybertec-pg-operator/pkg/util"
 	v1 "k8s.io/api/core/v1"
@@ -44,7 +45,7 @@ func (c *Cluster) GetDesiredMajorVersionAsInt() int {
 func (c *Cluster) GetDesiredMajorVersion() string {
 
 	if c.Config.OpConfig.MajorVersionUpgradeMode == "full" {
-		// e.g. current is 10, minimal is 11 allowing 11 to 15 clusters, everything below is upgraded
+		// e.g. current is 13, minimal is 13 allowing 13 to 17 clusters, everything below is upgraded
 		if IsBiggerPostgresVersion(c.Spec.PgVersion, c.Config.OpConfig.MinimalMajorVersion) {
 			c.logger.Infof("overwriting configured major version %s to %s", c.Spec.PgVersion, c.Config.OpConfig.TargetMajorVersion)
 			return c.Config.OpConfig.TargetMajorVersion
@@ -136,22 +137,33 @@ func (c *Cluster) majorVersionUpgrade() error {
 	desiredVersion := c.GetDesiredMajorVersionAsInt()
 
 	if c.currentMajorVersion >= desiredVersion {
+		if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists { // if failure annotation exists, remove it
+			c.removeFailuresAnnotation()
+			c.logger.Infof("removing failure annotation as the cluster is already up to date")
+		}
 		c.logger.Infof("cluster version up to date. current: %d, min desired: %d", c.currentMajorVersion, desiredVersion)
 		return nil
 	}
 
-	pods, err := c.listPodsOfType(TYPE_POSTGRESQL)
+	pods, err := c.listPods()
 	if err != nil {
 		return err
 	}
 
 	allRunning := true
+	isStandbyCluster := false
 
 	var masterPod *v1.Pod
 
 	for i, pod := range pods {
 		ps, _ := c.patroni.GetMemberData(&pod)
 
+		if ps.Role == "standby_leader" {
+			isStandbyCluster = true
+			c.currentMajorVersion = ps.ServerVersion
+			break
+		}
+
 		if ps.State != "running" {
 			allRunning = false
 			c.logger.Infof("identified non running pod, potentially skipping major version upgrade")
@@ -163,37 +175,114 @@ func (c *Cluster) majorVersionUpgrade() error {
 		}
 	}
 
+	if masterPod == nil {
+		c.logger.Infof("no master in the cluster, skipping major version upgrade")
+		return nil
+	}
+
+	// Recheck version with newest data from Patroni
+	if c.currentMajorVersion >= desiredVersion {
+		if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists { // if failure annotation exists, remove it
+			c.removeFailuresAnnotation()
+			c.logger.Infof("removing failure annotation as the cluster is already up to date")
+		}
+		c.logger.Infof("recheck cluster version is already up to date. current: %d, min desired: %d", c.currentMajorVersion, desiredVersion)
+		return nil
+	} else if isStandbyCluster {
+		c.logger.Warnf("skipping major version upgrade for %s/%s standby cluster. Re-deploy standby cluster with the required Postgres version specified", c.Namespace, c.Name)
+		return nil
+	}
+
+	if _, exists := c.ObjectMeta.Annotations[majorVersionUpgradeFailureAnnotation]; exists {
+		c.logger.Infof("last major upgrade failed, skipping upgrade")
+		return nil
+	}
+
+	if !isInMaintenanceWindow(c.Spec.MaintenanceWindows) {
+		c.logger.Infof("skipping major version upgrade, not in maintenance window")
+		return nil
+	}
+
+	members, err := c.patroni.GetClusterMembers(masterPod)
+	if err != nil {
+		c.logger.Error("could not get cluster members data from Patroni API, skipping major version upgrade")
+		return err
+	}
+	patroniData, err := c.patroni.GetMemberData(masterPod)
+	if err != nil {
+		c.logger.Error("could not get members data from Patroni API, skipping major version upgrade")
+		return err
+	}
+	patroniVer, err := semver.NewVersion(patroniData.Patroni.Version)
+	if err != nil {
+		c.logger.Error("error parsing Patroni version")
+		patroniVer, _ = semver.NewVersion("3.0.4")
+	}
+	verConstraint, _ := semver.NewConstraint(">= 3.0.4")
+	checkStreaming, _ := verConstraint.Validate(patroniVer)
+
+	for _, member := range members {
+		if PostgresRole(member.Role) == Leader {
+			continue
+		}
+		if checkStreaming && member.State != "streaming" {
+			c.logger.Infof("skipping major version upgrade, replica %s is not streaming from primary", member.Name)
+			return nil
+		}
+		if member.Lag > 16*1024*1024 {
+			c.logger.Infof("skipping major version upgrade, replication lag on member %s is too high", member.Name)
+			return nil
+		}
+	}
+
+	isUpgradeSuccess := true
 	numberOfPods := len(pods)
 	if allRunning && masterPod != nil {
 		c.logger.Infof("healthy cluster ready to upgrade, current: %d desired: %d", c.currentMajorVersion, desiredVersion)
 		if c.currentMajorVersion < desiredVersion {
+			defer func() error {
+				if err = c.criticalOperationLabel(pods, nil); err != nil {
+					return fmt.Errorf("failed to remove critical-operation label: %s", err)
+				}
+				return nil
+			}()
+			val := "true"
+			if err = c.criticalOperationLabel(pods, &val); err != nil {
+				return fmt.Errorf("failed to assign critical-operation label: %s", err)
+			}
+
 			podName := &spec.NamespacedName{Namespace: masterPod.Namespace, Name: masterPod.Name}
 			c.logger.Infof("triggering major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)
-			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "Starting major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "starting major version upgrade on pod %s of %d pods", masterPod.Name, numberOfPods)
 			upgradeCommand := fmt.Sprintf("set -o pipefail && /usr/bin/python3 /scripts/inplace_upgrade.py %d 2>&1 | tee last_upgrade.log", numberOfPods)
 
-			c.logger.Debugf("checking if the spilo image runs with root or non-root (check for user id=0)")
+			c.logger.Debug("checking if the spilo image runs with root or non-root (check for user id=0)")
 			resultIdCheck, errIdCheck := c.ExecCommand(podName, "/bin/bash", "-c", "/usr/bin/id -u")
 			if errIdCheck != nil {
-				c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "Checking user id to run upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, errIdCheck)
+				c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "checking user id to run upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, errIdCheck)
 			}
 
 			resultIdCheck = strings.TrimSuffix(resultIdCheck, "\n")
-			var result string
+			var result, scriptErrMsg string
 			if resultIdCheck != "0" {
-				c.logger.Infof("User id was identified as: %s, hence default user is non-root already", resultIdCheck)
+				c.logger.Infof("user id was identified as: %s, hence default user is non-root already", resultIdCheck)
 				result, err = c.ExecCommand(podName, "/bin/bash", "-c", upgradeCommand)
+				scriptErrMsg, _ = c.ExecCommand(podName, "/bin/bash", "-c", "tail -n 1 last_upgrade.log")
 			} else {
-				c.logger.Infof("User id was identified as: %s, using su to reach the postgres user", resultIdCheck)
+				c.logger.Infof("user id was identified as: %s, using su to reach the postgres user", resultIdCheck)
 				result, err = c.ExecCommand(podName, "/bin/su", "postgres", "-c", upgradeCommand)
+				scriptErrMsg, _ = c.ExecCommand(podName, "/bin/bash", "-c", "tail -n 1 last_upgrade.log")
 			}
 			if err != nil {
-				c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "Upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, err)
-				return err
+				isUpgradeSuccess = false
+				c.annotatePostgresResource(isUpgradeSuccess)
+				c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Major Version Upgrade", "upgrade from %d to %d FAILED: %v", c.currentMajorVersion, desiredVersion, scriptErrMsg)
+				return fmt.Errorf("%s", scriptErrMsg)
 			}
-			c.logger.Infof("upgrade action triggered and command completed: %s", result[:100])
 
-			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "Upgrade from %d to %d finished", c.currentMajorVersion, desiredVersion)
+			c.annotatePostgresResource(isUpgradeSuccess)
+			c.logger.Infof("upgrade action triggered and command completed: %s", result[:100])
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeNormal, "Major Version Upgrade", "upgrade from %d to %d finished", c.currentMajorVersion, desiredVersion)
 		}
 	}