Enable promote/demote standby via operator

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

@@ -117,7 +117,8 @@ type Patroni struct {
 
 //StandbyCluster
 type StandbyDescription struct {
-	S3WalPath string `json:"s3_wal_path,omitempty"`
+	S3WalPath      string `json:"s3_wal_path,omitempty"`
+	StandbyOptions map[string]string
 }
 
 // CloneDescription describes which cluster the new should clone and up to which point in time

pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go

@@ -506,7 +506,7 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
 	if in.StandbyCluster != nil {
 		in, out := &in.StandbyCluster, &out.StandbyCluster
 		*out = new(StandbyDescription)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	return
 }
@@ -719,6 +719,13 @@ func (in *Sidecar) DeepCopy() *Sidecar {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *StandbyDescription) DeepCopyInto(out *StandbyDescription) {
 	*out = *in
+	if in.StandbyOptions != nil {
+		in, out := &in.StandbyOptions, &out.StandbyOptions
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 	return
 }
 

pkg/cluster/sync.go

@@ -23,8 +23,16 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	c.setSpec(newSpec)
+	//Find out how NewSpec requires to edit the standby options of this cluster
+	if newSpec.Spec.StandbyCluster != nil && c.Spec.StandbyCluster == nil {
+		newSpec.Spec.StandbyCluster.StandbyOptions["restore_command"] = ""
+		c.logger.Info("Setting this cluster as standby cluster")
+	} else if newSpec.Spec.StandbyCluster == nil && c.Spec.StandbyCluster != nil {
+		newSpec.Spec.StandbyCluster.StandbyOptions["restore_command"] = "envdir \"/home/postgres/etc/wal-e.d/env-standby\" /scripts/restore_command.sh \"%f\" \"%p\""
+		c.logger.Info("Promoting this cluster into master from standby")
+	}
 
+	c.setSpec(newSpec)
 	defer func() {
 		if err != nil {
 			c.logger.Warningf("error while syncing cluster state: %v", err)
@@ -341,6 +349,8 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
 	optionsToSet := make(map[string]string)
 	pgOptions := c.Spec.Parameters
 
+	standbyOptions := c.Spec.StandbyCluster.StandbyOptions
+
 	for k, v := range pgOptions {
 		if isBootstrapOnlyParameter(k) {
 			optionsToSet[k] = v
@@ -361,6 +371,11 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
 	// carries the request to change configuration through
 	for _, pod := range pods {
 		podName := util.NameFromMeta(pod.ObjectMeta)
+		c.logger.Debugf("calling Patroni API on a pod %s to promote the Standby",
+			podName)
+		if err = c.patroni.EditStandby(&pod, standbyOptions); err != nil {
+			c.logger.Warningf("could not patch postgres for standby with a pod %s: %v", podName, err)
+		}
 		c.logger.Debugf("calling Patroni API on a pod %s to set the following Postgres options: %v",
 			podName, optionsToSet)
 		if err = c.patroni.SetPostgresParameters(&pod, optionsToSet); err == nil {

pkg/util/patroni/patroni.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 )
 
 const (
@@ -23,6 +23,7 @@ const (
 type Interface interface {
 	Switchover(master *v1.Pod, candidate string) error
 	SetPostgresParameters(server *v1.Pod, options map[string]string) error
+	EditStandby(server *v1.Pod, options map[string]string) error
 }
 
 // Patroni API client
@@ -102,3 +103,14 @@ func (p *Patroni) SetPostgresParameters(server *v1.Pod, parameters map[string]st
 	}
 	return p.httpPostOrPatch(http.MethodPatch, apiURL(server)+configPath, buf)
 }
+
+//SetPostgresParameters sets Postgres options via Patroni patch API call.
+func (p *Patroni) EditStandby(server *v1.Pod, options map[string]string) error {
+	buf := &bytes.Buffer{}
+	err := json.NewEncoder(buf).Encode(map[string]map[string]interface{}{"standby_cluster": {"restore_command": options}})
+
+	if err != nil {
+		return fmt.Errorf("could not encode json: %v", err)
+	}
+	return p.httpPostOrPatch(http.MethodPatch, apiURL(server)+configPath, buf)
+}