Updates the cluster initialization process as needed to

support pgBackRest replica creation.  This means that
in addition to creating the pgBackRest stanza following
initialization of the primary, a full backup is also taken
(the backup is executed following a successful stanza
creation).  This full backup can then be utilized to
initialize any replicas in the cluster using the pgBackRest
initialization process included in the containers.

In order to support this initialization process, any
pgreplica CR's created prior to the successful
initialization of the primary are placed in a pending
state until the cluster is fully initialized (i.e. the stanza
and full backup have been created).  Then, once the
cluster is initialized, and update to the pgreplica's is
triggered via an annotation, triggering the creation
of the replicas.  A 'pgcluster Initialized' state is now
included for the pgcluster CR, and is therefore utilized
to determine if the cluster is fully initialized (e.g. in
order to trigger replica creation).

Also included are updates to the pgBackRest restore
workflow as needed to support the new
crunchy-postgres-ha and crunchy-postgres-gis containers.
Now when a restore is initiated, all replicas are deleted
in addition to the primary.  Then, once the restored
primary DB comes back online, the replicas are recreated
using any existing pgreplica CR's.  This essentially establishes
a fully working cluster following a successful restore,
which includes all desired replicas and autofailover
enabled by default.

Additionally, it should be noted that any  PVC's previously
utilized by those replicas are reused for those same replicas
in the restored cluster.  Also, the pgcluster CR now shows
a 'pgcluster Restoring' state during a restore, and any
pgreplica will show a "pgreplica Pending restore" state
while the restore is underway.

Author: andrewlecuyer

apis/cr/v1/cluster.go

@@ -87,4 +87,8 @@ const (
 	PgclusterStateCreated PgclusterState = "pgcluster Created"
 	// PgclusterStateProcessed ...
 	PgclusterStateProcessed PgclusterState = "pgcluster Processed"
+	// PgclusterStateInitialized ...
+	PgclusterStateInitialized PgclusterState = "pgcluster Initialized"
+	// PgclusterStateRestore ...
+	PgclusterStateRestore PgclusterState = "pgcluster Restoring"
 )

apis/cr/v1/replica.go

@@ -63,6 +63,11 @@ type PgreplicaState string
 const (
 	// PgreplicaStateCreated ...
 	PgreplicaStateCreated PgreplicaState = "pgreplica Created"
+	// PgreplicaStatePending ...
+	PgreplicaStatePendingInit PgreplicaState = "pgreplica Pending init"
+	// PgreplicaStatePendingRestore ...
+	PgreplicaStatePendingRestore PgreplicaState = "pgreplica Pending restore"
 	// PgreplicaStateProcessed ...
 	PgreplicaStateProcessed PgreplicaState = "pgreplica Processed"
+
 )

conf/postgres-operator/cluster-deployment.json

@@ -5,6 +5,7 @@
         "name": "{{.Name}}",
         "labels": {
             "vendor": "crunchydata",
+            "pgo-pg-database": "true",
             {{.DeploymentLabels }}
         }
     },
@@ -21,6 +22,7 @@
                 "labels": {
                     "name": "{{.Name}}",
                     "vendor": "crunchydata",
+                    "pgo-pg-database": "true",
                     {{.PodLabels }}
                 }
             },
@@ -37,8 +39,11 @@
                     "readinessProbe": {
                         "exec": {
                             "command": [
-                                "/opt/cpm/bin/readiness/readiness.sh"
-                            ]
+                                "/bin/bash",
+                                "-c",
+                                "[[ -f '/crunchyadm/pgha_initialized' ]]",
+                                "&& pg_isready -h /crunchyadm -U crunchyready"
+                             ]
                         },
                         "initialDelaySeconds": 15,
                         "timeoutSeconds": 8
@@ -52,8 +57,19 @@
                     }, {
                         "name": "PGHA_USER",
                         "value": "postgres"
+                    },
+                    {{if .IsInit}}
+                    {
+                        "name": "PGHA_INIT",
+                        "valueFrom": {
+                            "configMapKeyRef": {
+                               "name": "{{.ClusterName}}-pgha-default-config",
+                               "key": "init"
+                            }
+                         }
                     }, 
-                    {{if .IsReplica}}
+                    {{ end }}
+                    {{if not .IsInit}}
                     {
                         "name": "PGHA_PRIMARY_HOST",
                         "value": "{{.ClusterName}}"
@@ -75,26 +91,19 @@
                                 "fieldPath": "metadata.namespace"
                             }
                         }
-                    }, {
-                        "name": "PATRONI_NAME",
-                        "valueFrom": {
-                            "fieldRef": {
-                                "fieldPath": "metadata.name"
-                            }
-                        }
-                    }, {
+                    },  {
+                        "name": "PATRONI_KUBERNETES_SCOPE_LABEL",
+                        "value": "{{.ScopeLabel}}"
+                    },  {
                         "name": "PATRONI_SCOPE",
                         "valueFrom": {
                             "fieldRef": {
-                                "fieldPath": "metadata.labels['pg-cluster']"
+                                "fieldPath": "metadata.labels['{{.ScopeLabel}}']"
                             }
                         }
                     }, {
                         "name": "PATRONI_KUBERNETES_LABELS",
                         "value": "{vendor: \"crunchydata\"}"
-                    }, {
-                        "name": "PATRONI_KUBERNETES_SCOPE_LABEL",
-                        "value": "{{.ScopeLabel}}"
                     }, {
                         "name": "PATRONI_LOG_LEVEL",
                         "value": "INFO"
@@ -205,7 +214,7 @@
                                 {{end}}
                                 {
                                     "configMap": { 
-                                        "name": "pgo-pgha-default-config",
+                                        "name": "{{.ClusterName}}-pgha-default-config",
                                         "optional": true
                                     }
                                 }

conf/postgres-operator/cluster-service.json

@@ -44,7 +44,16 @@
     }
     ],
         "selector": {
+            {{ if or (eq .Name .ClusterName) (eq .Name (printf "%s%s" .ClusterName "-replica")) }}
+            "pg-cluster": "{{.ClusterName}}",
+            {{ if eq .Name (printf "%s%s" .ClusterName "-replica") }}
+            "role": "replica"
+            {{else}}
+            "role": "master"
+            {{end}}
+            {{else}}
             "service-name": "{{.ServiceName}}"
+            {{end}}
         },
         "type": "{{.ServiceType}}",
         "sessionAffinity": "None"

config/annotations.go

@@ -0,0 +1,19 @@
+package config
+
+/*
+ Copyright 2019 Crunchy Data Solutions, Inc.
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+// annotations used by the operator
+const ANNOTATION_PGHA_BOOTSTRAP_REPLICA = "pgo-pgha-bootstrap-replica"

config/labels.go

@@ -21,6 +21,7 @@ const LABEL_SELECTOR = "selector"
 const LABEL_OPERATOR = "postgres-operator"
 const LABEL_PG_CLUSTER = "pg-cluster"
 const LABEL_PG_CLUSTER_IDENTIFIER = "pg-cluster-id"
+const LABEL_PG_DATABASE = "pgo-pg-database"
 
 const LABEL_PGBACKUP = "pgbackup"
 const LABEL_PGTASK = "pg-task"
@@ -178,3 +179,7 @@ const LABEL_PGO_DEFAULT_SC = "pgo-default-sc"
 const LABEL_FAILOVER_STARTED = "failover-started"
 
 const GLOBAL_CUSTOM_CONFIGMAP = "pgo-custom-pg-config"
+
+const LABEL_PGHA_SCOPE = "crunchy-pgha-scope"
+const LABEL_PGHA_DEFAULT_CONFIGMAP = "pgha-default-config"
+const LABEL_PGHA_BOOTSTRAP_BACKUP = "pgha-bootstrap-backup"

controller/jobcontroller.go

@@ -34,6 +34,7 @@ import (
 	"github.com/crunchydata/postgres-operator/util"
 	log "github.com/sirupsen/logrus"
 	apiv1 "k8s.io/api/batch/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -100,6 +101,15 @@ func (c *JobController) onUpdate(oldObj, newObj interface{}) {
 
 	log.Debugf("[JobController] onUpdate ns=%s %s active=%d succeeded=%d conditions=[%v]", job.ObjectMeta.Namespace, job.ObjectMeta.SelfLink, job.Status.Active, job.Status.Succeeded, job.Status.Conditions)
 
+	// determine if a foreground deletion of this resource is in progress
+	isForegroundDeletion := false
+	for _, finalizer := range job.Finalizers {
+        if finalizer == meta_v1.FinalizerDeleteDependents {
+			isForegroundDeletion = true
+			break
+		}
+    }
+
 	var err error
 
 	//handle the case of rmdata jobs succeeding
@@ -250,10 +260,74 @@ func (c *JobController) onUpdate(oldObj, newObj interface{}) {
 				log.Error("error in patching pgtask " + job.ObjectMeta.SelfLink + err.Error())
 			}
 			publishBackupComplete(labels[config.LABEL_PG_CLUSTER], job.ObjectMeta.Labels[config.LABEL_PG_CLUSTER_IDENTIFIER], job.ObjectMeta.Labels[config.LABEL_PGOUSER], "pgbackrest", job.ObjectMeta.Namespace, "")
+		
+			// if initial cluster backup, now annotate all existing pgreplica's to initiate replica creation
+			pgreplicaList := &crv1.PgreplicaList{}
+			selector := config.LABEL_PG_CLUSTER+"="+labels[config.LABEL_PG_CLUSTER] 
+			if labels[config.LABEL_PGHA_BOOTSTRAP_BACKUP] == "true" {
+				log.Debugf("jobController onUpdate initial backup complete")
+				
+				// get the pgcluster resource for the cluster the replica is a part of
+				cluster := crv1.Pgcluster{}
+				_, err = kubeapi.Getpgcluster(c.JobClient, &cluster, labels[config.LABEL_PG_CLUSTER],
+					job.ObjectMeta.Namespace)
+				if err != nil {
+					log.Error(err)
+					return
+				}
+				message := "Cluster has been initialized"
+				err = kubeapi.PatchpgclusterStatus(c.JobClient, crv1.PgclusterStateInitialized, message, 
+					&cluster, job.ObjectMeta.Namespace)
+				if err != nil {
+					log.Error(err)
+					return
+				}
+
+				
+				err := kubeapi.GetpgreplicasBySelector(c.JobClient, pgreplicaList, selector, job.ObjectMeta.Namespace)
+				if err != nil {
+					log.Error(err)
+					return
+				}
+				for _, pgreplica := range pgreplicaList.Items {
+					if pgreplica.Annotations == nil {
+						pgreplica.Annotations = make(map[string]string)
+					}
+					pgreplica.Annotations[config.ANNOTATION_PGHA_BOOTSTRAP_REPLICA] = "true"
+					err = kubeapi.Updatepgreplica(c.JobClient, &pgreplica, pgreplica.Name, job.ObjectMeta.Namespace)
+					if err != nil {
+						log.Error(err)
+						return
+					}
+				}
+			}
 		}
-
 		return
+	}
 
+	// create an initial full backup for replica creation once stanza creation is complete
+	if !isForegroundDeletion && labels[config.LABEL_BACKREST] == "true" && 
+		labels[config.LABEL_BACKREST_COMMAND] == crv1.PgtaskBackrestStanzaCreate {
+		log.Debugf("jobController onUpdate backrest stanza-create job case")
+		if job.Status.Succeeded == 1 {
+			log.Debugf("backrest stanza successfully created for cluster %s", labels[config.LABEL_PG_CLUSTER])
+			log.Debugf("proceeding with the initial full backup for cluster %s as needed for replica creation", 
+				labels[config.LABEL_PG_CLUSTER])
+			
+			var backrestRepoPodName string
+			for _, cont := range job.Spec.Template.Spec.Containers {
+				for _, envVar := range cont.Env {
+					if envVar.Name == "PODNAME" {
+						backrestRepoPodName = envVar.Value
+						log.Debugf("the backrest repo pod for the initial backup of cluster %s is %s",
+							labels[config.LABEL_PG_CLUSTER], backrestRepoPodName)
+					}
+				}
+			}
+			backrestoperator.CreateInitialBackup(c.JobClient, job.ObjectMeta.Namespace, 
+				labels[config.LABEL_PG_CLUSTER], backrestRepoPodName)
+		}
+		return
 	}
 
 	if labels[config.LABEL_PGBASEBACKUP_RESTORE] == "true" {

controller/podcontroller.go

@@ -194,6 +194,37 @@ func (c *PodController) checkReadyStatus(oldpod, newpod *apiv1.Pod, cluster *crv
 			if v.Name == "database" {
 				//see if there are pgtasks for adding a policy
 				if oldDatabaseStatus == false && v.Ready {
+
+					// update pgreplica status if performing a restore
+					if cluster.Status.State == crv1.PgclusterStateRestore {
+						pgreplicaList := &crv1.PgreplicaList{}
+						selector := config.LABEL_PG_CLUSTER+"="+clusterName
+						log.Debugf("Restored cluster %s went to ready, patching replicas", clusterName)
+						err := kubeapi.GetpgreplicasBySelector(c.PodClient, pgreplicaList, selector, newpod.ObjectMeta.Namespace)
+						if err != nil {
+							log.Error(err)
+							return
+						}
+						message := "Cluster has been initialized"
+						err = kubeapi.PatchpgclusterStatus(c.PodClient, crv1.PgclusterStateInitialized, message, cluster, 
+							newpod.ObjectMeta.Namespace)
+						if err != nil {
+							log.Error(err)
+							return
+						}
+						for _, pgreplica := range pgreplicaList.Items {
+							pgreplica.Spec.Status = "restore"
+							err = kubeapi.Updatepgreplica(c.PodClient, &pgreplica, pgreplica.Name, newpod.ObjectMeta.Namespace)
+							if err != nil {
+								log.Error(err)
+								return
+							}
+						}
+						return
+					}
+
+					operator.UpdatePghaDefaultConfigInitFlag(c.PodClientset, false, clusterName, newpod.ObjectMeta.Namespace)
+
 					log.Debugf("%s went to Ready from Not Ready, apply policies...", clusterName)
 					taskoperator.ApplyPolicies(clusterName, c.PodClientset, c.PodClient, newpod.ObjectMeta.Namespace)
 
@@ -408,8 +439,8 @@ func publishClusterComplete(clusterName, namespace string, cluster *crv1.Pgclust
 			Timestamp: time.Now(),
 			EventType: events.EventCreateClusterCompleted,
 		},
-		Clustername:       clusterName,
-		WorkflowID:        cluster.Spec.UserLabels[config.LABEL_WORKFLOW_ID],
+		Clustername: clusterName,
+		WorkflowID:  cluster.Spec.UserLabels[config.LABEL_WORKFLOW_ID],
 	}
 
 	err := events.Publish(f)

controller/replicacontroller.go

@@ -126,13 +126,33 @@ func (c *PgreplicaController) processNextItem() bool {
 			log.Error(err)
 			return false
 		}
-		clusteroperator.ScaleBase(c.PgreplicaClientset, c.PgreplicaClient, &replica, replica.ObjectMeta.Namespace)
 
-		state := crv1.PgreplicaStateProcessed
-		message := "Successfully processed Pgreplica by controller"
-		err = kubeapi.PatchpgreplicaStatus(c.PgreplicaClient, state, message, &replica, replica.ObjectMeta.Namespace)
+		// get the pgcluster resource for the cluster the replica is a part of
+		cluster := crv1.Pgcluster{}
+		_, err = kubeapi.Getpgcluster(c.PgreplicaClient, &cluster, replica.Spec.ClusterName, keyNamespace)
 		if err != nil {
-			log.Errorf("ERROR updating pgreplica status: %s", err.Error())
+			log.Error(err)
+			return false
+		}
+
+		// only process pgreplica if cluster has been initialized
+		if cluster.Status.State == crv1.PgclusterStateInitialized {
+			clusteroperator.ScaleBase(c.PgreplicaClientset, c.PgreplicaClient, &replica, replica.ObjectMeta.Namespace)
+
+			state := crv1.PgreplicaStateProcessed
+			message := "Successfully processed Pgreplica by controller"
+			err = kubeapi.PatchpgreplicaStatus(c.PgreplicaClient, state, message, &replica, replica.ObjectMeta.Namespace)
+			if err != nil {
+				log.Errorf("ERROR updating pgreplica status: %s", err.Error())
+			}
+		} else  {
+
+			state := crv1.PgreplicaStatePendingInit
+			message := "Pgreplica processing pending the creation of the initial backup"
+			err = kubeapi.PatchpgreplicaStatus(c.PgreplicaClient, state, message, &replica, replica.ObjectMeta.Namespace)
+			if err != nil {
+				log.Errorf("ERROR updating pgreplica status: %s", err.Error())
+			}
 		}
 
 		//no error, tell the queue to stop tracking history
@@ -163,9 +183,34 @@ func (c *PgreplicaController) onAdd(obj interface{}) {
 
 // onUpdate is called when a pgreplica is updated
 func (c *PgreplicaController) onUpdate(oldObj, newObj interface{}) {
-	//newExample := newObj.(*crv1.Pgreplica)
-	//log.Debugf("[PgreplicaController] ns=%s %s ", newExample.ObjectMeta.Namespace, newExample.ObjectMeta.Name)
+	
+	newPgreplica := newObj.(*crv1.Pgreplica)
+
+	log.Debugf("[PgreplicaController] onUpdate ns=%s %s", newPgreplica.ObjectMeta.Namespace, 
+		newPgreplica.ObjectMeta.SelfLink)
 
+	// get the pgcluster resource for the cluster the replica is a part of
+	cluster := crv1.Pgcluster{}
+	_, err := kubeapi.Getpgcluster(c.PgreplicaClient, &cluster, newPgreplica.Spec.ClusterName, 
+		newPgreplica.ObjectMeta.Namespace)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+
+	// only process pgreplica if cluster has been initialized
+	if cluster.Status.State == crv1.PgclusterStateInitialized && newPgreplica.Spec.Status != "complete" {
+		clusteroperator.ScaleBase(c.PgreplicaClientset, c.PgreplicaClient, newPgreplica,
+			newPgreplica.ObjectMeta.Namespace)
+
+		state := crv1.PgreplicaStateProcessed
+		message := "Successfully processed Pgreplica by controller"
+		err := kubeapi.PatchpgreplicaStatus(c.PgreplicaClient, state, message, newPgreplica,
+			newPgreplica.ObjectMeta.Namespace)
+		if err != nil {
+			log.Errorf("ERROR updating pgreplica status: %s", err.Error())
+		}
+	}
 }
 
 // onDelete is called when a pgreplica is deleted