Don't return logical slots for standby cluster (#2816)

Cluster.get_replication_slots() didn't take into account that there can not be logical replication slots in a standby cluster replicas. It was only skipping logical slots for the standby_leader, but replicas were expecting that they will have to copy them over.

Also on replicas in a standby cluster these logical slots were falsely added to the `_replication_slots` dict.

patroni/dcs/__init__.py

@@ -932,8 +932,8 @@ class Cluster(NamedTuple('Cluster',
         """``True`` if cluster is configured to use replication slots."""
         return bool(self.config and (self.config.data.get('postgresql') or {}).get('use_slots', True))
 
-    def get_replication_slots(self, my_name: str, role: str, nofailover: bool,
-                              major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
+    def get_replication_slots(self, my_name: str, role: str, nofailover: bool, major_version: int, *,
+                              is_standby_cluster: bool = False, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
         """Lookup configured slot names in the DCS, report issues found and merge with permanent slots.
 
         Will log an error if:
@@ -945,11 +945,13 @@ class Cluster(NamedTuple('Cluster',
         :param role: role of this node.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
         :param major_version: postgresql major version.
+        :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from
+                                   the outside because we want to protect from the ``/config`` key removal.
         :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found.
 
         :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
         """
-        slot_members: List[str] = self._get_slot_members(my_name, role) if self.use_slots else []
+        slot_members: List[str] = self._get_slot_members(my_name, role)
 
         slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'}
                                             for name in slot_members}
@@ -963,7 +965,7 @@ class Cluster(NamedTuple('Cluster',
                          "; ".join(f"{', '.join(v)} map to {k}"
                                    for k, v in slot_conflicts.items() if len(v) > 1))
 
-        permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {}
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover)
         disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
             slots, permanent_slots, my_name, major_version)
 
@@ -1022,33 +1024,32 @@ class Cluster(NamedTuple('Cluster',
             logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
         return disabled_permanent_logical_slots
 
-    def _get_permanent_slots(self, role: str, nofailover: bool) -> Dict[str, Any]:
+    def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]:
         """Get configured permanent slot names.
 
         .. note::
-            Permanent logical replication slots are only considered if ``use_slots`` configuration is enabled. Also,
-            only considered if *role* is ``primary`` or if it is a promotable ``replica`` -- what excludes a
-            ``standby_leader`` or ``replica`` with ``nofailover`` tag enabled. That combination is used for failing
-            over logical replication slots, and the latter nodes are not eligible for such task.
+            Permanent replication slots are only considered if ``use_slots`` configuration is enabled.
+            A node that is not supposed to become a leader (*nofailover*) will not have permanent replication slots.
 
-            Permanent physical slots are only considered if *role* is ``primary`` or ``standby_leader``, independently
-            if ``use_slots`` is enabled or not. That is done that way because even if Patroni itself is not using slots
-            to replicate among its members when ``use_slots`` is disabled, the user may still have configured Patroni to
-            keep permanent physical slots used out of Patroni.
+            In a standby cluster we only support physical replication slots.
 
+            The returned dictionary for a non-standby cluster always contains permanent logical replication slots in
+            order to show a warning if they are not supported by PostgreSQL before v11.
+
+        :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from
+                                   the outside because we want to protect from the ``/config`` key removal.
         :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
-                     or logical slots being consumed.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
 
         :returns: dictionary of permanent slot names mapped to attributes.
         """
-        if role in ('master', 'primary', 'standby_leader'):
-            permanent_slots = (self.__permanent_slots
-                               if role in ('master', 'primary')
-                               else self.__permanent_physical_slots)
-        else:
-            permanent_slots = self.__permanent_logical_slots if not nofailover else {}
-        return permanent_slots
+        if not self.use_slots or nofailover:
+            return {}
+
+        if is_standby_cluster:
+            return self.__permanent_physical_slots if role == 'standby_leader' else {}
+
+        return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots
 
     def _get_slot_members(self, my_name: str, role: str) -> List[str]:
         """Get a list of member names that have replication slots sourcing from this node.
@@ -1065,6 +1066,9 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: list of member names.
         """
+        if not self.use_slots:
+            return []
+
         if role in ('master', 'primary', 'standby_leader'):
             slot_members = [m.name for m in self.members
                             if m.name != my_name

patroni/postgresql/__init__.py

@@ -429,7 +429,18 @@ class Postgresql(object):
         :param global_config: last known :class:`GlobalConfig` object
         """
         self._cluster_info_state = {}
-        if cluster and cluster.config and cluster.config.modify_version:
+
+        if global_config:
+            self._global_config = global_config
+
+        if not self._global_config:
+            return
+
+        if self._global_config.is_standby_cluster:
+            # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback
+            self._has_permanent_logical_slots = False
+            self.set_enforce_hot_standby_feedback(False)
+        elif cluster and cluster.config and cluster.config.modify_version:
             self._has_permanent_logical_slots =\
                 cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
 
@@ -439,9 +450,6 @@ class Postgresql(object):
                 self._has_permanent_logical_slots
                 or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
 
-        if global_config:
-            self._global_config = global_config
-
     def _cluster_info_state_get(self, name: str) -> Optional[Any]:
         if not self._cluster_info_state:
             try:

patroni/postgresql/slots.py

@@ -471,6 +471,11 @@ class SlotsHandler:
             elif cluster.slots and name in cluster.slots:  # We want to copy only slots with feedback in a DCS
                 create_slots.append(name)
 
+        # Slots to be copied from the primary should be removed from the *slots* structure,
+        # otherwise Patroni falsely assumes that they already exist.
+        for name in create_slots:
+            slots.pop(name)
+
         error, copy_slots = self.schedule_advance_slots(advance_slots)
         if error:
             self._schedule_load_slots = True
@@ -493,12 +498,13 @@ class SlotsHandler:
         :returns: list of logical replication slots names that should be copied from the primary.
         """
         ret = []
-        if self._postgresql.major_version >= 90400 and cluster.config:
+        if self._postgresql.major_version >= 90400 and self._postgresql.global_config and cluster.config:
             try:
                 self.load_replication_slots()
 
-                slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role,
-                                                      nofailover, self._postgresql.major_version, True)
+                slots = cluster.get_replication_slots(
+                    self._postgresql.name, self._postgresql.role, nofailover, self._postgresql.major_version,
+                    is_standby_cluster=self._postgresql.global_config.is_standby_cluster, show_error=True)
 
                 self._drop_incorrect_slots(cluster, slots, paused)
 

tests/test_slots.py

@@ -7,6 +7,7 @@ from mock import Mock, PropertyMock, patch
 from threading import Thread
 
 from patroni import psycopg
+from patroni.config import GlobalConfig
 from patroni.dcs import Cluster, ClusterConfig, Member, SyncState
 from patroni.postgresql import Postgresql
 from patroni.postgresql.misc import fsync_dir
@@ -28,6 +29,7 @@ class TestSlotsHandler(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def setUp(self):
         super(TestSlotsHandler, self).setUp()
+        self.p._global_config = GlobalConfig({})
         self.s = self.p.slots_handler
         self.p.start()
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
@@ -44,6 +46,7 @@ class TestSlotsHandler(BaseTestPostgresql):
             self.s.sync_replication_slots(cluster, False)
         self.p.set_role('standby_leader')
         with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
+                patch.object(GlobalConfig, 'is_standby_cluster', PropertyMock(return_value=True)), \
                 patch('patroni.postgresql.slots.logger.debug') as mock_debug:
             self.s.sync_replication_slots(cluster, False)
             mock_debug.assert_called_once()