Don't retains member slots on nodes with nofailover tag (patroni#3169)

Followup on patroni#3142

Author: CyberDem0n

features/permanent_slots.feature

@@ -7,12 +7,13 @@ Feature: permanent slots
     Then I receive a response code 200
     And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds
     When I start postgres-1
-    And I start postgres-2
+    And I configure and start postgres-2 with a tag nofailover true
     And I configure and start postgres-3 with a tag replicatefrom postgres-2
     Then postgres-0 has a physical replication slot named test_physical after 10 seconds
     And postgres-0 has a physical replication slot named postgres_1 after 10 seconds
     And postgres-0 has a physical replication slot named postgres_2 after 10 seconds
     And postgres-2 has a physical replication slot named postgres_3 after 10 seconds
+    And postgres-2 does not have a replication slot named test_physical
 
   @slot-advance
   Scenario: check that logical permanent slots are created
@@ -24,10 +25,9 @@ Feature: permanent slots
   Scenario: check that permanent slots are created on replicas
     Given postgres-1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
     Then Logical slot test_logical is in sync between postgres-0 and postgres-1 after 10 seconds
-    And Logical slot test_logical is in sync between postgres-0 and postgres-2 after 10 seconds
     And Logical slot test_logical is in sync between postgres-0 and postgres-3 after 10 seconds
     And postgres-1 has a physical replication slot named test_physical after 2 seconds
-    And postgres-2 has a physical replication slot named test_physical after 2 seconds
+    And postgres-2 does not have a replication slot named test_logical
     And postgres-3 has a physical replication slot named test_physical after 2 seconds
 
   @slot-advance
@@ -36,9 +36,9 @@ Feature: permanent slots
     And postgres-1 has a physical replication slot named postgres_0 after 2 seconds
     And postgres-1 has a physical replication slot named postgres_2 after 2 seconds
     And postgres-1 has a physical replication slot named postgres_3 after 2 seconds
-    And postgres-2 has a physical replication slot named postgres_0 after 2 seconds
+    And postgres-2 does not have a replication slot named postgres_0
+    And postgres-2 does not have a replication slot named postgres_1
     And postgres-2 has a physical replication slot named postgres_3 after 2 seconds
-    And postgres-2 has a physical replication slot named postgres_1 after 2 seconds
     And postgres-3 has a physical replication slot named postgres_0 after 2 seconds
     And postgres-3 has a physical replication slot named postgres_1 after 2 seconds
     And postgres-3 has a physical replication slot named postgres_2 after 2 seconds
@@ -50,11 +50,8 @@ Feature: permanent slots
     And I get all changes from physical slot test_physical on postgres-0
     Then Logical slot test_logical is in sync between postgres-0 and postgres-1 after 10 seconds
     And Physical slot test_physical is in sync between postgres-0 and postgres-1 after 10 seconds
-    And Logical slot test_logical is in sync between postgres-0 and postgres-2 after 10 seconds
-    And Physical slot test_physical is in sync between postgres-0 and postgres-2 after 10 seconds
     And Logical slot test_logical is in sync between postgres-0 and postgres-3 after 10 seconds
     And Physical slot test_physical is in sync between postgres-0 and postgres-3 after 10 seconds
-    And Physical slot postgres_1 is in sync between postgres-0 and postgres-2 after 10 seconds
     And Physical slot postgres_1 is in sync between postgres-0 and postgres-3 after 10 seconds
     And Physical slot postgres_3 is in sync between postgres-2 and postgres-0 after 20 seconds
     And Physical slot postgres_3 is in sync between postgres-2 and postgres-1 after 10 seconds
@@ -69,7 +66,7 @@ Feature: permanent slots
 
   @slot-advance
   Scenario: check that only non-permanent member slots are written to the retain_slots in /status key
-    And "status" key in DCS has postgres_0 in retain_slots
+    Given "status" key in DCS has postgres_0 in retain_slots
     And "status" key in DCS has postgres_1 in retain_slots
     And "status" key in DCS has postgres_2 in retain_slots
     And "status" key in DCS does not have postgres_3 in retain_slots

patroni/dcs/__init__.py

@@ -1116,16 +1116,23 @@ def _get_permanent_slots(self, postgresql: 'Postgresql', tags: Tags, role: str)
 
     def _get_members_slots(self, name: str, role: str, nofailover: bool,
                            can_advance_slots: bool) -> Dict[str, Dict[str, Any]]:
-        """Get physical replication slots configuration for members that sourcing from this node.
+        """Get physical replication slots configuration for a given member.
 
-        If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on
-        the current primary, because that member would replicate from elsewhere. We still create the slot if
-        the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the
-        primary), or if ``replicatefrom`` destination member happens to be the current primary.
+        There are following situations possible:
 
-        If the ``nostream`` tag is set on the member - we should not create the replication slot for it on
-        the current primary or any other member even if ``replicatefrom`` is set, because ``nostream`` disables
-        WAL streaming.
+            * If the ``nostream`` tag is set on the member - we should not have the replication slot for it
+              on the current primary or any other member even if ``replicatefrom`` is set, because
+              ``nostream`` disables WAL streaming.
+
+            * PostgreSQL is 11 and newer and configuration allows retention of member replication slots. In this case
+              we want to have replication slots for every member except the case when we have ``nofailover`` tag set.
+
+            * PostgreSQL is older than 11 or configuration doesn't allow member slots retention. In this case we want:
+
+                * On primary have replication slots for all members that don't have ``replicatefrom`` tag pointing
+                  to the existing member.
+
+                * On replica node have replication slots only for members which ``replicatefrom`` tag pointing to us.
 
         Will log an error if:
 
@@ -1195,7 +1202,7 @@ def replica_filter(member: Member) -> bool:
             ret[slot_name] = {'type': 'physical', 'lsn': lsn, 'expected_active': expected_active(member)}
         slot_name = slot_name_from_member_name(name)
         ret.update({slot: {'type': 'physical'} for slot in self.status.retain_slots
-                    if slot not in ret and slot != slot_name})
+                    if not nofailover and slot not in ret and slot != slot_name})
 
         if len(ret) < len(members):
             # Find which names are conflicting for a nicer error message

tests/test_slots.py

@@ -328,3 +328,17 @@ def test_advance_physical_slots(self):
                                                                       None, None, None)], Exception])), \
                 patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(False, False))):
             self.s.sync_replication_slots(cluster, self.tags)
+
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
+    @patch.object(Postgresql, 'role', PropertyMock(return_value='replica'))
+    @patch.object(TestTags, 'tags', PropertyMock(return_value={'nofailover': True}))
+    def test_slots_nofailover_tag(self):
+        self.p.name = self.leadermem.name
+        cluster = Cluster(True, ClusterConfig(1, {}, 1), self.leader,
+                          Status(0, {}, [self.leadermem.name, self.other.name, self.me.name]),
+                          [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
+        global_config.update(cluster)
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None,
+                                                                      None, None, None)], Exception])) as mock_query:
+            self.s.sync_replication_slots(cluster, self.tags)
+            self.assertTrue(mock_query.call_args[0][0].startswith('SELECT slot_name, slot_type, xmin, '))