Added option to create streaming standby cluster.

docs/reference/cluster_manifest.md

@@ -385,16 +385,33 @@ under the `clone` top-level key and do not affect the already running cluster.
 ## Standby cluster
 
 On startup, an existing `standby` top-level key creates a standby Postgres
-cluster streaming from a remote location. So far streaming from S3 and GCS WAL
-archives is supported.
+cluster streaming from a remote location. There are two options supported
+for a streaming cluster. Continious streaming from WAL storage, and
+streaming directly from host.
 
 * **s3_wal_path**
   the url to S3 bucket containing the WAL archive of the remote primary.
-  Optional, but `s3_wal_path`  or `gs_wal_path` is required.
+  Optional.
 
 * **gs_wal_path**
   the url to GS bucket containing the WAL archive of the remote primary.
-  Optional, but `s3_wal_path`  or `gs_wal_path` is required.
+  Optional.
+
+* **standby_host**
+  Standby host is IP or hostname from which standby clustere is streaming
+  WAL records. This should be specified when using `streaming_host` standby method.
+
+* **standby_method**
+  Specify desired method: `s3_wal` or `streaming_host`. Required.
+
+* **standby_port**
+  Specify port that should be used when connecting to primary.
+  This should be specified when using `streaming_host` standby method.
+
+* **standby_secret_name**
+  Specify secret name that is used to look-up for a password that
+  is need to connect to primary host. Password should be under `password` key
+  of a secret. This is *required* be specified when using `streaming_host` standby method.
 
 ## Volume properties
 

manifests/postgresql.crd.yaml

@@ -468,9 +468,17 @@ spec:
               standby:
                 type: object
                 properties:
+                  gs_wal_path:
+                    type: string
                   s3_wal_path:
                     type: string
-                  gs_wal_path:
+                  standby_host:
+                    type: string
+                  standby_method:
+                    type: string
+                  standby_port:
+                    type: string
+                  standby_secret_name:
                     type: string
               streams:
                 type: array

manifests/standby-manifest.yaml

@@ -12,4 +12,8 @@ spec:
     version: "14"
 # Make this a standby cluster and provide the s3 bucket path of source cluster for continuous streaming.
   standby:
-    s3_wal_path: "s3://path/to/bucket/containing/wal/of/source/cluster/"
+    s3_wal_path: "s3://path/to/bucket/containing/wal/of/source/cluster/" # Make this a standby cluster and provide the s3 bucket path of source cluster for continuous streaming. 
+    standby_method: "s3_wal" # s3_wal or streaming_host; 
+    # standby_host: ""
+    # standby_port: ""
+    # standby_secret_name: ""

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

@@ -708,12 +708,24 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
 					"standby": {
 						Type: "object",
 						Properties: map[string]apiextv1.JSONSchemaProps{
+              "gs_wal_path": {
+								Type: "string",
+							},
 							"s3_wal_path": {
 								Type: "string",
 							},
-							"gs_wal_path": {
+							"standby_host": {
+								Type: "string",
+							},
+							"standby_method": {
 								Type: "string",
 							},
+							"standby_port": {
+								Type: "string",
+							},
+							"standby_secret_name": {
+								Type: "string",
+              },
 						},
 					},
 					"streams": {

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

@@ -170,8 +170,12 @@ type Patroni struct {
 
 // StandbyDescription contains s3 wal path
 type StandbyDescription struct {
-	S3WalPath string `json:"s3_wal_path,omitempty"`
-	GSWalPath string `json:"gs_wal_path,omitempty"`
+	S3WalPath         string `json:"s3_wal_path,omitempty"`
+	GSWalPath         string `json:"gs_wal_path,omitempty"`
+	StandbyHost       string `json:"standby_host,omitempty"`
+	StandbyMethod     string `json:"standby_method,omitempty"`
+	StandbyPort       string `json:"standby_port,omitempty"`
+	StandbySecretName string `json:"standby_secret_name,omitempty"`
 }
 
 // TLSDescription specs TLS properties

pkg/cluster/k8sres.go

@@ -745,17 +745,6 @@ func (c *Cluster) generateSpiloPodEnvVars(uid types.UID, spiloConfiguration stri
 			Name:  "PGUSER_STANDBY",
 			Value: c.OpConfig.ReplicationUsername,
 		},
-		{
-			Name: "PGPASSWORD_STANDBY",
-			ValueFrom: &v1.EnvVarSource{
-				SecretKeyRef: &v1.SecretKeySelector{
-					LocalObjectReference: v1.LocalObjectReference{
-						Name: c.credentialSecretName(c.OpConfig.ReplicationUsername),
-					},
-					Key: "password",
-				},
-			},
-		},
 		{
 			Name:  "PAM_OAUTH2",
 			Value: c.OpConfig.PamConfiguration,
@@ -765,6 +754,24 @@ func (c *Cluster) generateSpiloPodEnvVars(uid types.UID, spiloConfiguration stri
 			Value: c.OpConfig.PamRoleName,
 		},
 	}
+
+	var standbySecretName string
+	if standbyDescription != nil && standbyDescription.StandbyMethod == "streaming_host" {
+		standbySecretName = standbyDescription.StandbySecretName
+	} else {
+		standbySecretName = c.credentialSecretName(c.OpConfig.ReplicationUsername)
+	}
+
+	envVars = append(envVars, v1.EnvVar{Name: "PGPASSWORD_STANDBY",
+		ValueFrom: &v1.EnvVarSource{
+			SecretKeyRef: &v1.SecretKeySelector{
+				LocalObjectReference: v1.LocalObjectReference{
+					Name: standbySecretName,
+				},
+				Key: "password",
+			},
+		}})
+
 	if c.OpConfig.EnablePgVersionEnvVar {
 		envVars = append(envVars, v1.EnvVar{Name: "PGVERSION", Value: c.GetDesiredMajorVersion()})
 	}
@@ -1080,6 +1087,18 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 	if err != nil {
 		return nil, err
 	}
+	if spec.StandbyCluster != nil && spec.StandbyCluster.StandbyMethod == "" {
+		// In order to be backward compatible, we fallback to s3_wal when standby_method is not specified
+		if spec.StandbyCluster.S3WalPath != "" {
+			c.logger.Warningf("Fallback to a s3_wal as standby_method is not specified.")
+			spec.StandbyCluster.StandbyMethod = "s3_wal"
+    } else if spec.StandbyCluster.GSWalPath != "" {
+			c.logger.Warningf("Fallback to a gs_wal as standby_method is not specified.")
+			spec.StandbyCluster.StandbyMethod = "gs_wal"
+		} else {
+			return nil, fmt.Errorf("standby_method is and s3_wal_path are empty, what standby method to use!?")
+		}
+	}
 
 	// fetch env vars from custom ConfigMap
 	secretEnvVarsList, err := c.getPodEnvironmentSecretVariables()
@@ -1093,11 +1112,6 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 	sort.Slice(customPodEnvVarsList,
 		func(i, j int) bool { return customPodEnvVarsList[i].Name < customPodEnvVarsList[j].Name })
 
-	if spec.StandbyCluster != nil && spec.StandbyCluster.S3WalPath == "" &&
-		spec.StandbyCluster.GSWalPath == "" {
-		return nil, fmt.Errorf("one of s3_wal_path or gs_wal_path must be set for standby cluster")
-	}
-
 	// backward compatible check for InitContainers
 	if spec.InitContainersOld != nil {
 		msg := "Manifest parameter init_containers is deprecated."
@@ -1905,40 +1919,49 @@ func (c *Cluster) generateCloneEnvironment(description *acidv1.CloneDescription)
 func (c *Cluster) generateStandbyEnvironment(description *acidv1.StandbyDescription) []v1.EnvVar {
 	result := make([]v1.EnvVar, 0)
 
-	if description.S3WalPath == "" && description.GSWalPath == "" {
+	if description.StandbyMethod == "" {
 		return nil
 	}
 
-	if description.S3WalPath != "" {
-		// standby with S3, find out the bucket to setup standby
-		msg := "Standby from S3 bucket using custom parsed S3WalPath from the manifest %s "
-		c.logger.Infof(msg, description.S3WalPath)
-
-		result = append(result, v1.EnvVar{
-			Name:  "STANDBY_WALE_S3_PREFIX",
-			Value: description.S3WalPath,
-		})
-	} else if description.GSWalPath != "" {
-		msg := "Standby from GS bucket using custom parsed GSWalPath from the manifest %s "
-		c.logger.Infof(msg, description.GSWalPath)
-
-		envs := []v1.EnvVar{
-			{
-				Name:  "STANDBY_WALE_GS_PREFIX",
-				Value: description.GSWalPath,
-			},
-			{
-				Name:  "STANDBY_GOOGLE_APPLICATION_CREDENTIALS",
-				Value: c.OpConfig.GCPCredentials,
-			},
-		}
-		result = append(result, envs...)
+	if description.StandbyMethod == "s3_wal" || description.StandbyMethod == "gs_wal" {
+  	if description.S3WalPath != "" {
+  		// standby with S3, find out the bucket to setup standby
+	  	msg := "Standby from S3 bucket using custom parsed S3WalPath from the manifest %s "
+		  c.logger.Infof(msg, description.S3WalPath)
+
+	  	result = append(result, v1.EnvVar{
+	  		Name:  "STANDBY_WALE_S3_PREFIX",
+		  	Value: description.S3WalPath,
+		  })
+  	} else if description.GSWalPath != "" {
+	  	msg := "Standby from GS bucket using custom parsed GSWalPath from the manifest %s "
+	  	c.logger.Infof(msg, description.GSWalPath)
+
+	  	envs := []v1.EnvVar{
+	  		{
+	  			Name:  "STANDBY_WALE_GS_PREFIX",
+	  			Value: description.GSWalPath,
+	  		},
+		  	{
+			  	Name:  "STANDBY_GOOGLE_APPLICATION_CREDENTIALS",
+				  Value: c.OpConfig.GCPCredentials,
+			  },
+		  }
+		  result = append(result, envs...)
+  	}
+		result = append(result, v1.EnvVar{Name: "STANDBY_METHOD", Value: "STANDBY_WITH_WALE"})
+		result = append(result, v1.EnvVar{Name: "STANDBY_WAL_BUCKET_SCOPE_PREFIX", Value: ""})
+	} else if description.StandbyMethod == "streaming_host" {
+		// standby with streaming replication from standby host
+		msg := "Standby using streaming replication from standby host: %s "
+		c.logger.Infof(msg, description.StandbyHost)
+
+		result = append(result, v1.EnvVar{Name: "STANDBY_HOST", Value: description.StandbyHost})
+		result = append(result, v1.EnvVar{Name: "STANDBY_PORT", Value: description.StandbyPort})
+		result = append(result, v1.EnvVar{Name: "STANDBY_CLUSTER", Value: "True"})
 
 	}
 
-	result = append(result, v1.EnvVar{Name: "STANDBY_METHOD", Value: "STANDBY_WITH_WALE"})
-	result = append(result, v1.EnvVar{Name: "STANDBY_WAL_BUCKET_SCOPE_PREFIX", Value: ""})
-
 	return result
 }