reflect feedback

Author: idanovinda

e2e/tests/test_e2e.py

@@ -395,13 +395,22 @@ def test_config_update(self):
             "spec": {
                 "postgresql": {
                     "parameters": {
-                        "max_connections": new_max_connections_value
+                        "max_connections": new_max_connections_value,
+                        "wal_level": "logical"
                      }
                  },
-                 "patroni": {
+                "users": {
+                    "test_user": []
+                },
+                "databases": {
+                    "foo": "test_user"
+                },
+                "patroni": {
                     "slots": {
                         "test_slot": {
-                            "type": "physical"
+                            "type": "logical",
+                            "database": "foo",
+                            "plugin": "pgoutput"
                         },
                         "test_slot_2": {
                             "type": "physical"
@@ -500,14 +509,22 @@ def compare_config():
             self.eventuallyEqual(lambda: self.query_database(replica.metadata.name, "postgres", setting_query)[0], lower_max_connections_value,
                 "Previous max_connections setting not applied on replica", 10, 5)
 
-            # delete test_slot_2 from config
+            # patch new slot via Patroni REST
+            patroni_slot = "test_patroni_slot"
+            patch_slot_command = """curl -s -XPATCH -d '{"slots": {"test_patroni_slot": {"type": "physical"}}}' http://localhost:8008/config"""
+            k8s.exec_with_kubectl(replica.metadata.name, patch_slot_command)
+
+            # delete test_slot_2 from config and change the plugin type for test_slot
             slot_to_remove = "test_slot_2"
+            slot_to_change = "test_slot"
             pg_patch_slots = {
                 "spec": {
                     "patroni": {
                         "slots": {
                             "test_slot": {
-                                "type": "physical"
+                                "type": "logical",
+                                "database": "foo",
+                                "plugin": "wal2json"
                             }
                         }
                     }
@@ -526,7 +543,26 @@ def compare_config():
             """ % (slot_to_remove)
 
             self.eventuallyEqual(lambda: len(self.query_database(replica.metadata.name, "postgres", deleted_slot_query)), 0,
-                "The replication slot cannot be deleted", 10, 5)       
+                "The replication slot cannot be deleted", 10, 5)
+
+            changed_slot_query = """
+                SELECT plugin
+                  FROM pg_replication_slots
+                 WHERE slot_name = '%s';
+            """ % (slot_to_change)
+
+            self.eventuallyEqual(lambda: self.query_database(replica.metadata.name, "postgres", changed_slot_query)[0], "wal2json",
+                "The replication slot cannot be updated", 10, 5)
+            
+            # make sure slot from Patroni didn't get deleted
+            patroni_slot_query = """
+                SELECT slot_name
+                  FROM pg_replication_slots
+                 WHERE slot_name = '%s';
+            """ % (patroni_slot)
+
+            self.eventuallyEqual(lambda: len(self.query_database(replica.metadata.name, "postgres", patroni_slot_query)), 1,
+                "The replication slot from Patroni gets deleted", 10, 5)
 
         except timeout_decorator.TimeoutError:
             print('Operator log: {}'.format(k8s.get_operator_log()))

pkg/cluster/cluster.go

@@ -377,10 +377,8 @@ func (c *Cluster) Create() error {
 		}
 	}
 
-	if len(c.Spec.Patroni.Slots) > 0 {
-		for slotName, desiredSlot := range c.Spec.Patroni.Slots {
-			c.replicationSlots[slotName] = desiredSlot
-		}
+	for slotName, desiredSlot := range c.Spec.Patroni.Slots {
+		c.replicationSlots[slotName] = desiredSlot
 	}
 
 	return nil

pkg/cluster/sync_test.go

@@ -144,6 +144,13 @@ func TestCheckAndSetGlobalPostgreSQLConfiguration(t *testing.T) {
 	client, _ := newFakeK8sSyncClient()
 	clusterName := "acid-test-cluster"
 	namespace := "default"
+	testSlots := map[string]map[string]string{
+		"slot1": {
+			"type":     "logical",
+			"plugin":   "wal2json",
+			"database": "foo",
+		},
+	}
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -210,6 +217,7 @@ func TestCheckAndSetGlobalPostgreSQLConfiguration(t *testing.T) {
 	tests := []struct {
 		subtest        string
 		patroni        acidv1.Patroni
+		slots          map[string]map[string]string
 		pgParams       map[string]string
 		restartPrimary bool
 	}{
@@ -251,9 +259,62 @@ func TestCheckAndSetGlobalPostgreSQLConfiguration(t *testing.T) {
 			},
 			restartPrimary: true,
 		},
+		{
+			subtest: "slot does not exist but is desired",
+			patroni: acidv1.Patroni{
+				TTL: 20,
+			},
+			slots: testSlots,
+			pgParams: map[string]string{
+				"log_min_duration_statement": "200",
+				"max_connections":            "50",
+			},
+			restartPrimary: false,
+		},
+		{
+			subtest: "slot exist, nothing specified in manifest",
+			patroni: acidv1.Patroni{
+				TTL: 20,
+				Slots: map[string]map[string]string{
+					"slot1": {
+						"type":     "logical",
+						"plugin":   "pgoutput",
+						"database": "foo",
+					},
+				},
+			},
+			pgParams: map[string]string{
+				"log_min_duration_statement": "200",
+				"max_connections":            "50",
+			},
+			restartPrimary: false,
+		},
+		{
+			subtest: "slot plugin differs",
+			patroni: acidv1.Patroni{
+				TTL: 20,
+				Slots: map[string]map[string]string{
+					"slot1": {
+						"type":     "logical",
+						"plugin":   "pgoutput",
+						"database": "foo",
+					},
+				},
+			},
+			slots: testSlots,
+			pgParams: map[string]string{
+				"log_min_duration_statement": "200",
+				"max_connections":            "50",
+			},
+			restartPrimary: false,
+		},
 	}
 
 	for _, tt := range tests {
+		if tt.slots != nil {
+			cluster.Spec.Patroni.Slots = tt.slots
+		}
+
 		configPatched, requirePrimaryRestart, err := cluster.checkAndSetGlobalPostgreSQLConfiguration(mockPod, tt.patroni, cluster.Spec.Patroni, tt.pgParams, cluster.Spec.Parameters)
 		assert.NoError(t, err)
 		if configPatched != true {