Postgres Operator automated cluster upgrade

Previously, the existing cluster upgrade focused on the updating
of a cluster's underlying container images. However, due to the
various changes in the Postgres Operator's operation, including numerous
updates to the relevant CRDs, integration of Patroni for HA and other
significant changes between versions, updates between Postgres Operator
releases required the manual deletion of the existing clusters while
being sure to preserve the underlying PVC storage. After installing the
new version, the clusters could be recreated manually so long as the
name of the new cluster matched the existing PVC's name.

This process has been completely overhauled so that now, when the Postgres
Operator has been updated to a new version, the existing pgclusters will
be left in place. Normal Operator functionality will be restricted until
the clusters are upgraded to match the currently installed version, although
the pods, services, etc will still be in place and accessible via other
methods (e.g. kubectl, service IP, etc).

To upgrade a particular cluster, use

pgo upgrade mycluster

This will follow a similar process to the existing documented manual process,
where the pods, deployments, replicasets, pgtasks and jobs are deleted, the
cluster's replicas are scaled down and replica PVCs deleted, but the primary
PVC and backrest-repo PVC are left in place. Existing services for the primary,
replica and backrest-shared-repo are also kept. Configmaps and secrets are
kept except where deletion is required. For a cluster 'mycluster', the
following configmaps will be deleted (if they exist) and recreated:

mycluster-leader
mycluster-pgha-config

One exception to this is during the upgrade of a standby cluster, where
the 'mycluster-pgha-config' configmap is modified, but not deleted. This is
necessary to ensure the standby cluster is not reinitialized during upgrade.

The secret 'mycluster-backrest-repo-config' will also be deleted and
recreated in this example to ensure the encryption key used matches the current
version requirements.

The pgcluster CRD will then be read, updated automatically and replaced, at which
point the normal cluster creation process will take over. The end result of the
upgrade should be an identical numer of pods, deployments, etc with a new pgbackrest
backup taken, but existing backups still available.

Please note, this upgrade currently supports cluster upgrades from Postgres
Operator version 4.1.0 and later, and can be run multiple times against the
same cluster if needed.

Author: tjmoore4

apis/crunchydata.com/v1/task.go

@@ -27,7 +27,10 @@ const PgtaskDeleteData = "delete-data"
 const PgtaskFailover = "failover"
 const PgtaskAutoFailover = "autofailover"
 const PgtaskAddPolicies = "addpolicies"
-const PgtaskMinorUpgrade = "minorupgradecluster"
+
+const PgtaskUpgrade = "clusterupgrade"
+const PgtaskUpgradeCreated = "cluster upgrade - task created"
+const PgtaskUpgradeInProgress = "cluster upgrade - in progress"
 
 const PgtaskWorkflow = "workflow"
 const PgtaskWorkflowCloneType = "cloneworkflow"

apiserver/upgradeservice/upgradeimpl.go

@@ -16,26 +16,42 @@ limitations under the License.
 */
 
 import (
+	"fmt"
+	"io/ioutil"
+	"regexp"
+	"strconv"
+	"time"
+
 	crv1 "github.com/crunchydata/postgres-operator/apis/crunchydata.com/v1"
 	"github.com/crunchydata/postgres-operator/apiserver"
 	msgs "github.com/crunchydata/postgres-operator/apiservermsgs"
 	"github.com/crunchydata/postgres-operator/config"
 	"github.com/crunchydata/postgres-operator/kubeapi"
+
 	log "github.com/sirupsen/logrus"
 	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 )
 
-// CreateUpgrade ...
-func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns string) msgs.CreateUpgradeResponse {
+// Currently supported version information for upgrades
+const (
+	REQUIRED_MAJOR_PGO_VERSION = 4
+	MAXIMUM_MINOR_PGO_VERSION  = 3
+	MINIMUM_MINOR_PGO_VERSION  = 1
+)
+
+// CreateUpgrade accepts the CreateUpgradeRequest performs the necessary validation checks and
+// organizes the needed upgrade information before creating the required pgtask
+// Command format: pgo upgrade mycluster
+func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns, pgouser string) msgs.CreateUpgradeResponse {
 	response := msgs.CreateUpgradeResponse{}
 	response.Status = msgs.Status{Code: msgs.Ok, Msg: ""}
-	response.Results = make([]string, 1)
+	response.Results = make([]string, 0)
 
 	log.Debugf("createUpgrade called %v", request)
 
 	if request.Selector != "" {
-		//use the selector instead of an argument list to filter on
+		// use the selector instead of an argument list to filter on
 
 		myselector, err := labels.Parse(request.Selector)
 		if err != nil {
@@ -46,7 +62,7 @@ func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns string) msgs.CreateUpg
 		}
 		log.Debugf("myselector is %s", myselector.String())
 
-		//get the clusters list
+		// get the clusters list
 		clusterList := crv1.PgclusterList{}
 		err = kubeapi.GetpgclustersBySelector(apiserver.RESTClient,
 			&clusterList, request.Selector, ns)
@@ -56,6 +72,7 @@ func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns string) msgs.CreateUpg
 			return response
 		}
 
+		// check that the cluster can be found
 		if len(clusterList.Items) == 0 {
 			log.Debug("no clusters found")
 			response.Status.Msg = "no clusters found"
@@ -72,16 +89,37 @@ func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns string) msgs.CreateUpg
 	for _, clusterName := range request.Args {
 		log.Debugf("create upgrade called for %s", clusterName)
 
-		//build the pgtask for the minor upgrade
+		// build the pgtask for the upgrade
 		spec := crv1.PgtaskSpec{}
-		spec.TaskType = crv1.PgtaskMinorUpgrade
-		spec.Status = "requested"
+		spec.TaskType = crv1.PgtaskUpgrade
+		// set the status as created
+		spec.Status = crv1.PgtaskUpgradeCreated
 		spec.Parameters = make(map[string]string)
 		spec.Parameters[config.LABEL_PG_CLUSTER] = clusterName
-		spec.Name = clusterName + "-" + config.LABEL_MINOR_UPGRADE
+		spec.Parameters[crv1.PgtaskWorkflowSubmittedStatus] = time.Now().Format(time.RFC3339)
+
+		u, err := ioutil.ReadFile("/proc/sys/kernel/random/uuid")
+		if err != nil {
+			log.Error(err)
+			response.Status.Code = msgs.Error
+			response.Status.Msg = fmt.Sprintf("Could not generate UUID for upgrade task. Error: %s", err.Error())
+			return response
+		}
+		spec.Parameters[crv1.PgtaskWorkflowID] = string(u[:len(u)-1])
+
+		// pass the CCP Image Tag from the apiserver
+		spec.Parameters[config.LABEL_CCP_IMAGE_KEY] = apiserver.Pgo.Cluster.CCPImageTag
+		// pass the PGO version for the upgrade
+		spec.Parameters[config.LABEL_PGO_VERSION] = msgs.PGO_VERSION
+		// pass the PGO username for use in the updated CR if missing
+		spec.Parameters[config.LABEL_PGOUSER] = pgouser
+
+		spec.Name = clusterName + "-" + config.LABEL_UPGRADE
 		spec.Namespace = ns
 		labels := make(map[string]string)
 		labels[config.LABEL_PG_CLUSTER] = clusterName
+		labels[config.LABEL_PGOUSER] = pgouser
+		labels[crv1.PgtaskWorkflowID] = spec.Parameters[crv1.PgtaskWorkflowID]
 
 		newInstance := &crv1.Pgtask{
 			ObjectMeta: meta_v1.ObjectMeta{
@@ -91,20 +129,17 @@ func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns string) msgs.CreateUpg
 			Spec: spec,
 		}
 
-		// remove any existing pgtask for this minor upgrade
-		result := crv1.Pgtask{}
-		found, err := kubeapi.Getpgtask(apiserver.RESTClient,
-			&result, spec.Name, ns)
-		if found {
-			err := kubeapi.Deletepgtask(apiserver.RESTClient, spec.Name, ns)
-			if err != nil {
-				response.Status.Code = msgs.Error
-				response.Status.Msg = err.Error()
-				return response
-			}
+		// remove any existing pgtask for this upgrade
+		task := crv1.Pgtask{}
+		found, err := kubeapi.Getpgtask(apiserver.RESTClient, &task, spec.Name, ns)
+
+		if found && task.Spec.Status != crv1.CompletedStatus {
+			response.Status.Code = msgs.Error
+			response.Status.Msg = fmt.Sprintf("Could not upgrade cluster: there exists an ongoing upgrade task: [%s]. If you believe this is an error, try deleting this pgtask CR.", task.Spec.Name)
+			return response
 		}
 
-		//validate the cluster name and ensure autofail is turned off for each cluster.
+		// validate the cluster name and ensure autofail is turned off for each cluster.
 		cl := crv1.Pgcluster{}
 		found, err = kubeapi.Getpgcluster(apiserver.RESTClient,
 			&cl, clusterName, ns)
@@ -114,31 +149,131 @@ func CreateUpgrade(request *msgs.CreateUpgradeRequest, ns string) msgs.CreateUpg
 			return response
 		}
 
-		//figure out what version we are upgrading to
-		imageToUpgradeTo := apiserver.Pgo.Cluster.CCPImageTag
-		if request.CCPImageTag != "" {
-			imageToUpgradeTo = request.CCPImageTag
+		if !supportedOperatorVersion(cl.ObjectMeta.Labels[config.LABEL_PGO_VERSION]) {
+			response.Status.Code = msgs.Error
+			response.Status.Msg = "Cannot upgrade " + clusterName + " from Postgres Operator version " + cl.ObjectMeta.Labels[config.LABEL_PGO_VERSION]
+			return response
 		}
-		if imageToUpgradeTo == cl.Spec.CCPImageTag {
+
+		// for the upgrade procedure, we only upgrade to the current image used by the
+		// Postgres Operator. As such, we will validate that the Postgres Operator's configured
+		// image tag (first value) is compatible (i.e. is the same Major PostgreSQL version) as the
+		// existing cluster's PG value, unless the --ignore-validation flag is set.
+		if !upgradeTagValid(cl.Spec.CCPImageTag, apiserver.Pgo.Cluster.CCPImageTag) && !request.IgnoreValidation {
+			log.Debugf("Cannot upgrade from %s to %s. Image must be the same base OS and the upgrade must be within the same major PG version.", cl.Spec.CCPImageTag, apiserver.Pgo.Cluster.CCPImageTag)
 			response.Status.Code = msgs.Error
-			response.Status.Msg = "can not upgrade to the same image tag " + imageToUpgradeTo + " " + cl.Spec.CCPImageTag
+			response.Status.Msg = fmt.Sprintf("cannot upgrade from %s to %s, upgrade task failed.", cl.Spec.CCPImageTag, apiserver.Pgo.Cluster.CCPImageTag)
 			return response
 		}
-		log.Debugf("upgrading to image tag %s", imageToUpgradeTo)
-		spec.Parameters["CCPImageTag"] = imageToUpgradeTo
+
+		// given the above check passed, save the CCP image tag value
+		spec.Parameters["CCPImageTag"] = cl.Spec.CCPImageTag
 
 		// Create an instance of our CRD
 		err = kubeapi.Createpgtask(apiserver.RESTClient, newInstance, ns)
 		if err != nil {
 			response.Status.Code = msgs.Error
 			response.Status.Msg = err.Error()
+			response.WorkflowID = spec.Parameters[crv1.PgtaskWorkflowID]
 			return response
 		}
 
-		msg := "created minor upgrade task for " + clusterName
+		msg := "created upgrade task for " + clusterName
 		response.Results = append(response.Results, msg)
-
+		response.WorkflowID = spec.Parameters[crv1.PgtaskWorkflowID]
 	}
 
 	return response
 }
+
+// supportedOperatorVersion validates the Postgres Operator version
+// information for the candidate pgcluster. If this value is in the
+// required range, return true so that the upgrade may continue. Otherwise,
+// return false.
+func supportedOperatorVersion(version string) bool {
+	// get the Operator version
+	operatorVersionRegex := regexp.MustCompile(`^(\d)\.(\d)\.(\d)`)
+	operatorVersion := operatorVersionRegex.FindStringSubmatch(version)
+
+	// if this regex passes, the returned array should always contain
+	// 4 values. At 0, the full match, then 1-3 are the three defined groups
+	// If this is not true, the upgrade cannot continue (and we won't want to
+	// reference potentially missing array items).
+	if len(operatorVersion) != 4 {
+		return false
+	}
+
+	// if the first group does not equal the current major version
+	// then the upgrade cannot continue
+	if major, err := strconv.Atoi(operatorVersion[1]); err != nil {
+		log.Error(err)
+		return false
+	} else if major != REQUIRED_MAJOR_PGO_VERSION {
+		return false
+	}
+
+	// if the second group does is not in the supported range,
+	// then the upgrade cannot continue
+	minor, err := strconv.Atoi(operatorVersion[2])
+	if err != nil {
+		log.Errorf("Cannot convert Postgres Operator's minor version to an integer. Error: ", err)
+		return false
+	}
+	if minor < MINIMUM_MINOR_PGO_VERSION || minor > MAXIMUM_MINOR_PGO_VERSION {
+		return false
+	}
+
+	// If none of the above is true, the upgrade can continue
+	return true
+
+}
+
+// upgradeTagValid compares and validates the PostgreSQL version values stored
+// in the image tag of the existing pgcluster CR against the values set in the
+// Postgres Operator's configuration
+func upgradeTagValid(upgradeFrom, upgradeTo string) bool {
+
+	log.Debugf("Validating upgrade from %s to %s", upgradeFrom, upgradeTo)
+
+	versionRegex := regexp.MustCompile(`-(\d+)\.(\d+)(\.\d+)?-`)
+
+	// get the PostgreSQL version values
+	upgradeFromValue := versionRegex.FindStringSubmatch(upgradeFrom)
+	upgradeToValue := versionRegex.FindStringSubmatch(upgradeTo)
+
+	// if this regex passes, the returned array should always contain
+	// 4 values. At 0, the full match, then 1-3 are the three defined groups
+	// If this is not true, the upgrade cannot continue (and we won't want to
+	// reference potentially missing array items).
+	if len(upgradeFromValue) != 4 || len(upgradeToValue) != 4 {
+		return false
+	}
+
+	// if the first group does not match (PG version 9, 10, 11, 12 etc), or if a value is
+	// missing, then the upgrade cannot continue
+	if upgradeFromValue[1] != upgradeToValue[1] && upgradeToValue[1] != "" {
+		return false
+	}
+
+	// if the above check passed, and there is no fourth value, then the PG
+	// version has only two digits (e.g. PG 10, 11 or 12), meaning this is a minor upgrade.
+	// After validating the second value is at least equal (this is to allow for multiple executions of the
+	// upgrade in case an error occurs), the upgrade can continue
+	if upgradeFromValue[3] == "" && upgradeToValue[3] == "" && upgradeFromValue[2] <= upgradeToValue[2] {
+		return true
+	}
+
+	// finally, if the second group matches and is not empty, then, based on the
+	// possibilities remaining for Operator container image tags, this is either PG 9.5 or 9.6.
+	// if the second group value matches, and the third group was already validated as not
+	// empty, check that the third value is at least equal (this is to allow for multiple executions of the
+	// upgrade in case an error occurs). If so, the upgrade can continue.
+	if upgradeFromValue[2] == upgradeToValue[2] && upgradeToValue[2] != "" && upgradeFromValue[3] <= upgradeToValue[3] {
+		return true
+	}
+
+	// if none of the above conditions are met, a two digit Major version upgrade is likely being
+	// attempted, or a tag value or general error occurred, so we cannot continue
+	return false
+
+}

apiserver/upgradeservice/upgradeservice.go

@@ -17,25 +17,26 @@ limitations under the License.
 
 import (
 	"encoding/json"
+	"net/http"
+
 	"github.com/crunchydata/postgres-operator/apiserver"
 	msgs "github.com/crunchydata/postgres-operator/apiservermsgs"
 	log "github.com/sirupsen/logrus"
-	"net/http"
 )
 
 // CreateUpgradeHandler ...
 // pgo upgrade mycluster
-// parameters --upgrade-type
-// parameters --ccp-image-tag
 func CreateUpgradeHandler(w http.ResponseWriter, r *http.Request) {
 	// swagger:operation POST /upgrades upgradeservice upgrades
 	/*```
-	UPGRADE performs an upgrade on a PostgreSQL cluster. For example:
-
+	UPGRADE performs an upgrade on a PostgreSQL cluster from an earlier version
+	of the Postgres Operator to the current version.
 
+	OTHER UPGRADE DESCRIPTION:
+	This upgrade will update the scale down any existing replicas while saving the primary
+	and pgbackrest repo PVCs, then update the existing pgcluster CR and resubmit it for
+	re-creation.
 
-	This upgrade will update the CCPImageTag of the deployment for the following: primary, replicas, and backrest-repo.
-	The running containers are upgraded one at a time, sequentially, in the following order: replicas, backrest-repo, then primary.
 	*/
 	// ---
 	//  produces:
@@ -81,6 +82,6 @@ func CreateUpgradeHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	resp = CreateUpgrade(&request, ns)
+	resp = CreateUpgrade(&request, ns, username)
 	json.NewEncoder(w).Encode(resp)
 }

apiservermsgs/common.go

@@ -19,6 +19,8 @@ const PGO_VERSION = "4.3.0"
 
 // Ok status
 const Ok = "ok"
+
+// Error code string
 const Error = "error"
 
 // UpgradeError is the error used for when a command is tried against a cluster that has not

apiservermsgs/upgrademsgs.go

@@ -18,16 +18,17 @@ limitations under the License.
 // CreateUpgradeRequest ...
 // swagger:model
 type CreateUpgradeRequest struct {
-	Args          []string
-	Selector      string
-	Namespace     string
-	CCPImageTag   string
-	ClientVersion string
+	Args             []string
+	Selector         string
+	Namespace        string
+	ClientVersion    string
+	IgnoreValidation bool
 }
 
 // CreateUpgradeResponse ...
 // swagger:model
 type CreateUpgradeResponse struct {
 	Results []string
 	Status
+	WorkflowID string
 }

config/labels.go

@@ -55,7 +55,6 @@ const LABEL_STORAGE_CONFIG = "storage-config"
 const LABEL_NODE_LABEL = "node-label"
 const LABEL_VERSION = "version"
 const LABEL_PGO_VERSION = "pgo-version"
-const LABEL_UPGRADE_DATE = "operator-upgrade-date"
 const LABEL_DELETE_DATA = "delete-data"
 const LABEL_DELETE_DATA_STARTED = "delete-data-started"
 const LABEL_DELETE_BACKUPS = "delete-backups"
@@ -64,12 +63,8 @@ const LABEL_IS_BACKUP = "is-backup"
 const LABEL_STARTUP = "startup"
 const LABEL_SHUTDOWN = "shutdown"
 
-const LABEL_MINOR_UPGRADE = "minor-upgrade"
-const LABEL_UPGRADE_IN_PROGRESS = "upgrade-in-progress"
-const LABEL_UPGRADE_COMPLETED = "upgrade-complete"
-const LABEL_UPGRADE_REPLICA = "upgrade-replicas"
-const LABEL_UPGRADE_PRIMARY = "upgrade-primary"
-const LABEL_UPGRADE_BACKREST = "upgrade-backrest"
+// label for the pgcluster upgrade
+const LABEL_UPGRADE = "upgrade"
 
 const LABEL_BACKREST = "pgo-backrest"
 const LABEL_BACKREST_JOB = "pgo-backrest-job"

controller/pgtask/pgtaskcontroller.go

@@ -101,8 +101,8 @@ func (c *Controller) processNextItem() bool {
 
 	//process the incoming task
 	switch tmpTask.Spec.TaskType {
-	case crv1.PgtaskMinorUpgrade:
-		log.Debug("delete minor upgrade task added")
+	case crv1.PgtaskUpgrade:
+		log.Debug("upgrade task added")
 		clusteroperator.AddUpgrade(c.PgtaskClientset, c.PgtaskClient, &tmpTask, keyNamespace)
 	case crv1.PgtaskFailover:
 		log.Debug("failover task added")