[WIP] Add the ability to configure replications slots in Patroni (#398)

* Add the ability to configure replication slots in Patroni

* Add debugging  to Makefile for CDP builds

Makefile

@@ -30,6 +30,11 @@ else
 	DOCKERFILE = Dockerfile
 endif
 
+ifdef CDP_PULL_REQUEST_NUMBER
+	CDP_TAG := -${CDP_BUILD_VERSION}
+endif
+
+
 PATH := $(GOPATH)/bin:$(PATH)
 SHELL := env PATH=$(PATH) $(SHELL)
 
@@ -52,13 +57,18 @@ docker-context: scm-source.json linux
 	cp build/linux/${BINARY} scm-source.json docker/build/
 
 docker: ${DOCKERDIR}/${DOCKERFILE} docker-context
-	cd "${DOCKERDIR}" && docker build --rm -t "$(IMAGE):$(TAG)$(DEBUG_POSTFIX)" -f "${DOCKERFILE}" .
+	echo `(env)`
+	echo "Tag ${TAG}"
+	echo "Version ${VERSION}"
+	echo "CDP tag ${CDP_TAG}"
+	echo "git describe $(shell git describe --tags --always --dirty)"
+	cd "${DOCKERDIR}" && docker build --rm -t "$(IMAGE):$(TAG)$(CDP_TAG)$(DEBUG_POSTFIX)" -f "${DOCKERFILE}" .
 
 indocker-race:
 	docker run --rm -v "${GOPATH}":"${GOPATH}" -e GOPATH="${GOPATH}" -e RACE=1 -w ${PWD} golang:1.8.1 bash -c "make linux"
 
 push:
-	docker push "$(IMAGE):$(TAG)"
+	docker push "$(IMAGE):$(TAG)$(CDP_TAG)"
 
 scm-source.json: .git
 	echo '{\n "url": "git:$(GITURL)",\n "revision": "$(GITHEAD)",\n "author": "$(USER)",\n "status": "$(GITSTATUS)"\n}' > scm-source.json

docs/reference/cluster_manifest.md

@@ -151,6 +151,9 @@ explanation of `ttl` and `loop_wait` parameters.
   patroni `maximum_lag_on_failover` parameter value, optional. The default is
   set by the Spilo docker image. Optional.
 
+* **replication_slots**
+  permanent replication slots that Patroni preserves after failover by re-creating them on the new primary immediately after doing a promote. Slots could be reconfigured with the help of `patronictl edit-config`. It is the responsibility of a user to avoid clashes in names between replication slots automatically created by Patroni for cluster members and permanent replication slots. Optional.
+
 ## Postgres container resources
 
 Those parameters define [CPU and memory requests and

manifests/complete-postgres-manifest.yaml

@@ -40,6 +40,13 @@ spec:
     pg_hba:
     - hostssl all all 0.0.0.0/0 md5
     - host    all all 0.0.0.0/0 md5
+    replication_slots:
+      permanent_physical_1:
+        type: physical
+      permanent_logical_1:
+        type: logical
+        database: foo
+        plugin: pgoutput
     ttl: 30
     loop_wait: &loop_wait 10
     retry_timeout: 10

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

@@ -102,6 +102,7 @@ type Patroni struct {
 	LoopWait             uint32                       `json:"loop_wait"`
 	RetryTimeout         uint32                       `json:"retry_timeout"`
 	MaximumLagOnFailover float32                      `json:"maximum_lag_on_failover"` // float32 because https://github.com/kubernetes/kubernetes/issues/30213
+	ReplicationSlots     map[string]map[string]string `json:"replication_slots"`
 }
 
 // CloneDescription describes which cluster the new should clone and up to which point in time

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

@@ -132,7 +132,7 @@ var unmarshalCluster = []struct {
 		// This error message can vary between Go versions, so compute it for the current version.
 		Error: json.Unmarshal([]byte(`{"teamId": 0}`), &PostgresSpec{}).Error(),
 	},
-	[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
+	[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"replication_slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
 	{[]byte(`{
   "kind": "Postgresql",
   "apiVersion": "acid.zalan.do/v1",
@@ -189,7 +189,14 @@ var unmarshalCluster = []struct {
       "ttl": 30,
       "loop_wait": 10,
       "retry_timeout": 10,
-      "maximum_lag_on_failover": 33554432
+	  "maximum_lag_on_failover": 33554432,
+	  "replication_slots" : {
+		  "permanent_logical_1" : {
+			  "type"     : "logical",
+			  "database" : "foo",
+			  "plugin"   : "pgoutput"
+	       }
+	  }
     },
     "maintenanceWindows": [
       "Mon:01:00-06:00",
@@ -230,6 +237,7 @@ var unmarshalCluster = []struct {
 					LoopWait:             10,
 					RetryTimeout:         10,
 					MaximumLagOnFailover: 33554432,
+					ReplicationSlots:     map[string]map[string]string{"permanent_logical_1": {"type": "logical", "database": "foo", "plugin": "pgoutput"}},
 				},
 				Resources: Resources{
 					ResourceRequest: ResourceDescription{CPU: "10m", Memory: "50Mi"},
@@ -265,7 +273,7 @@ var unmarshalCluster = []struct {
 			},
 			Error: "",
 		},
-		[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"9.6","parameters":{"log_statement":"all","max_connections":"10","shared_buffers":"32MB"}},"volume":{"size":"5Gi","storageClass":"SSD"},"patroni":{"initdb":{"data-checksums":"true","encoding":"UTF8","locale":"en_US.UTF-8"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host    all all 0.0.0.0/0 md5"],"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432},"resources":{"requests":{"cpu":"10m","memory":"50Mi"},"limits":{"cpu":"300m","memory":"3000Mi"}},"teamId":"ACID","allowedSourceRanges":["127.0.0.1/32"],"numberOfInstances":2,"users":{"zalando":["superuser","createdb"]},"maintenanceWindows":["Mon:01:00-06:00","Sat:00:00-04:00","05:00-05:15"],"clone":{"cluster":"acid-batman"}}}`), nil},
+		[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"9.6","parameters":{"log_statement":"all","max_connections":"10","shared_buffers":"32MB"}},"volume":{"size":"5Gi","storageClass":"SSD"},"patroni":{"initdb":{"data-checksums":"true","encoding":"UTF8","locale":"en_US.UTF-8"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host    all all 0.0.0.0/0 md5"],"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"replication_slots":{"permanent_logical_1":{"database":"foo","plugin":"pgoutput","type":"logical"}}},"resources":{"requests":{"cpu":"10m","memory":"50Mi"},"limits":{"cpu":"300m","memory":"3000Mi"}},"teamId":"ACID","allowedSourceRanges":["127.0.0.1/32"],"numberOfInstances":2,"users":{"zalando":["superuser","createdb"]},"maintenanceWindows":["Mon:01:00-06:00","Sat:00:00-04:00","05:00-05:15"],"clone":{"cluster":"acid-batman"}}}`), nil},
 	{
 		[]byte(`{"kind": "Postgresql","apiVersion": "acid.zalan.do/v1","metadata": {"name": "teapot-testcluster1"}, "spec": {"teamId": "acid"}}`),
 		Postgresql{
@@ -280,7 +288,7 @@ var unmarshalCluster = []struct {
 			Status: ClusterStatusInvalid,
 			Error:  errors.New("name must match {TEAM}-{NAME} format").Error(),
 		},
-		[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"teapot-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
+		[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"teapot-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"replication_slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
 	{
 		in: []byte(`{"kind": "Postgresql","apiVersion": "acid.zalan.do/v1","metadata": {"name": "acid-testcluster1"}, "spec": {"teamId": "acid", "clone": {"cluster": "team-batman"}}}`),
 		out: Postgresql{
@@ -300,12 +308,12 @@ var unmarshalCluster = []struct {
 			},
 			Error: "",
 		},
-		marshal: []byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{"cluster":"team-batman"}}}`), err: nil},
+		marshal: []byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"replication_slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{"cluster":"team-batman"}}}`), err: nil},
 	{[]byte(`{"kind": "Postgresql","apiVersion": "acid.zalan.do/v1"`),
 		Postgresql{},
 		[]byte{},
 		errors.New("unexpected end of JSON input")},
-	{[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster","creationTimestamp":qaz},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`),
+	{[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster","creationTimestamp":qaz},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"replication_slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`),
 		Postgresql{},
 		[]byte{},
 		errors.New("invalid character 'q' looking for beginning of value")}}

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

@@ -320,6 +320,23 @@ func (in *Patroni) DeepCopyInto(out *Patroni) {
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.ReplicationSlots != nil {
+		in, out := &in.ReplicationSlots, &out.ReplicationSlots
+		*out = make(map[string]map[string]string, len(*in))
+		for key, val := range *in {
+			var outVal map[string]string
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(map[string]string, len(*in))
+				for key, val := range *in {
+					(*out)[key] = val
+				}
+			}
+			(*out)[key] = outVal
+		}
+	}
 	return
 }
 

pkg/cluster/k8sres.go

@@ -41,6 +41,7 @@ type patroniDCS struct {
 	RetryTimeout             uint32                       `json:"retry_timeout,omitempty"`
 	MaximumLagOnFailover     float32                      `json:"maximum_lag_on_failover,omitempty"`
 	PGBootstrapConfiguration map[string]interface{}       `json:"postgresql,omitempty"`
+	ReplicationSlots         map[string]map[string]string `json:"replication_slots,omitempty"`
 }
 
 type pgBootstrap struct {
@@ -215,6 +216,9 @@ PatroniInitDBParams:
 	if patroni.TTL != 0 {
 		config.Bootstrap.DCS.TTL = patroni.TTL
 	}
+	if patroni.ReplicationSlots != nil {
+		config.Bootstrap.DCS.ReplicationSlots = patroni.ReplicationSlots
+	}
 
 	config.PgLocalConfiguration = make(map[string]interface{})
 	config.PgLocalConfiguration[patroniPGBinariesParameterName] = fmt.Sprintf(pgBinariesLocationTemplate, pg.PgVersion)