Simplify methods related to replication slots in the Cluster class (#2958)

Instead of passing around names, specific tags, and Postgres version just pass Postgresql object and objects implementing Tags interface.

It should simplify implementation of #2842

patroni/dcs/__init__.py

@@ -24,6 +24,7 @@ from ..utils import parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
+    from ..postgresql import Postgresql
 
 SLOT_ADVANCE_AVAILABLE_VERSION = 110000
 CITUS_COORDINATOR_GROUP_ID = 0
@@ -956,28 +957,29 @@ class Cluster(NamedTuple('Cluster',
         """Dictionary of permanent ``logical`` replication slots."""
         return {name: value for name, value in self.__permanent_slots.items() if self.is_logical_slot(value)}
 
-    def get_replication_slots(self, my_name: str, role: str, nofailover: bool, major_version: int, *,
+    def get_replication_slots(self, postgresql: 'Postgresql', member: Tags, *,
-                              show_error: bool = False) -> Dict[str, Dict[str, Any]]:
+                              role: Optional[str] = None, 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:
 
             * Any logical slots are disabled, due to version compatibility, and *show_error* is ``True``.
 
-        :param my_name: name of this node.
+        :param postgresql: reference to :class:`Postgresql` object.
-        :param role: role of this node.
+        :param member: reference to an object implementing :class:`Tags` interface.
-        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param role: role of the node, if not set will be taken from *postgresql*.
-        :param major_version: postgresql major version.
         :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.
         """
-        slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role)
+        name = member.name if isinstance(member, Member) else postgresql.name
-        permanent_slots: Dict[str, Any] = self._get_permanent_slots(role=role, nofailover=nofailover,
+        role = role or postgresql.role
-                                                                    major_version=major_version)
+
+        slots: Dict[str, Dict[str, str]] = self._get_members_slots(name, role)
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role)
 
         disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
-            slots, permanent_slots, my_name, major_version)
+            slots, permanent_slots, name, postgresql.major_version)
 
         if disabled_permanent_logical_slots and show_error:
             logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
@@ -985,7 +987,7 @@ class Cluster(NamedTuple('Cluster',
 
         return slots
 
-    def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str,
+    def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], name: str,
                                major_version: int) -> List[str]:
         """Merge replication *slots* for members with *permanent_slots*.
 
@@ -995,7 +997,7 @@ class Cluster(NamedTuple('Cluster',
         Type is assumed to be ``physical`` if there are no attributes stored as the slot value.
 
         :param slots: Slot names with existing attributes if known.
-        :param my_name: name of this node.
+        :param name: name of this node.
         :param permanent_slots: dictionary containing slot name key and slot information values.
         :param major_version: postgresql major version.
 
@@ -1003,9 +1005,9 @@ class Cluster(NamedTuple('Cluster',
         """
         disabled_permanent_logical_slots: List[str] = []
 
-        for name, value in permanent_slots.items():
+        for slot_name, value in permanent_slots.items():
-            if not slot_name_re.match(name):
+            if not slot_name_re.match(slot_name):
-                logger.error("Invalid permanent replication slot name '%s'", name)
+                logger.error("Invalid permanent replication slot name '%s'", slot_name)
                 logger.error("Slot name may only contain lower case letters, numbers, and the underscore chars")
                 continue
 
@@ -1016,24 +1018,24 @@ class Cluster(NamedTuple('Cluster',
 
                 if value['type'] == 'physical':
                     # Don't try to create permanent physical replication slot for yourself
-                    if name != slot_name_from_member_name(my_name):
+                    if slot_name != slot_name_from_member_name(name):
-                        slots[name] = value
+                        slots[slot_name] = value
                     continue
 
                 if self.is_logical_slot(value):
                     if major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
-                        disabled_permanent_logical_slots.append(name)
+                        disabled_permanent_logical_slots.append(slot_name)
-                    elif name in slots:
+                    elif slot_name in slots:
                         logger.error("Permanent logical replication slot {'%s': %s} is conflicting with"
-                                     " physical replication slot for cluster member", name, value)
+                                     " physical replication slot for cluster member", slot_name, value)
                     else:
-                        slots[name] = value
+                        slots[slot_name] = value
                     continue
 
-            logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
+            logger.error("Bad value for slot '%s' in permanent_slots: %s", slot_name, permanent_slots[slot_name])
         return disabled_permanent_logical_slots
 
-    def _get_permanent_slots(self, *, role: str, nofailover: bool, major_version: int) -> Dict[str, Any]:
+    def _get_permanent_slots(self, postgresql: 'Postgresql', tags: Tags, role: str) -> Dict[str, Any]:
         """Get configured permanent replication slots.
 
         .. note::
@@ -1045,23 +1047,23 @@ class Cluster(NamedTuple('Cluster',
             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 role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
+        :param postgresql: reference to :class:`Postgresql` object.
-        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param tags: reference to an object implementing :class:`Tags` interface.
-        :param major_version: postgresql major version.
+        :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``.
 
         :returns: dictionary of permanent slot names mapped to attributes.
         """
-        if not global_config.use_slots or nofailover:
+        if not global_config.use_slots or tags.nofailover:
             return {}
 
         if global_config.is_standby_cluster:
             return self.__permanent_physical_slots \
-                if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION or role == 'standby_leader' else {}
+                if postgresql.major_version >= SLOT_ADVANCE_AVAILABLE_VERSION or role == 'standby_leader' else {}
 
-        return self.__permanent_slots if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION\
+        return self.__permanent_slots if postgresql.major_version >= SLOT_ADVANCE_AVAILABLE_VERSION\
             or role in ('master', 'primary') else self.__permanent_logical_slots
 
-    def _get_members_slots(self, my_name: str, role: str) -> Dict[str, Dict[str, str]]:
+    def _get_members_slots(self, name: str, role: str) -> Dict[str, Dict[str, str]]:
         """Get physical replication slots configuration for members that sourcing from this node.
 
         If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on
@@ -1073,7 +1075,7 @@ class Cluster(NamedTuple('Cluster',
 
             * Conflicting slot names between members are found
 
-        :param my_name: name of this node.
+        :param name: name of this node.
         :param role: role of this node, if this is a ``primary`` or ``standby_leader`` return list of members
                      replicating from this node. If not then return a list of members replicating as cascaded
                      replicas from this node.
@@ -1084,14 +1086,14 @@ class Cluster(NamedTuple('Cluster',
             return {}
 
         # we always want to exclude the member with our name from the list
-        members = filter(lambda m: m.name != my_name, self.members)
+        members = filter(lambda m: m.name != name, self.members)
 
         if role in ('master', 'primary', 'standby_leader'):
             members = [m for m in members if m.replicatefrom is None
-                       or m.replicatefrom == my_name or not self.has_member(m.replicatefrom)]
+                       or m.replicatefrom == name or not self.has_member(m.replicatefrom)]
         else:
             # only manage slots for replicas that replicate from this one, except for the leader among them
-            members = [m for m in members if m.replicatefrom == my_name and m.name != self.leader_name]
+            members = [m for m in members if m.replicatefrom == name and m.name != self.leader_name]
 
         slots = {slot_name_from_member_name(m.name): {'type': 'physical'} for m in members}
         if len(slots) < len(members):
@@ -1104,76 +1106,76 @@ class Cluster(NamedTuple('Cluster',
                                    for k, v in slot_conflicts.items() if len(v) > 1))
         return slots
 
-    def has_permanent_slots(self, my_name: str, *, nofailover: bool = False,
+    def has_permanent_slots(self, postgresql: 'Postgresql', member: Tags) -> bool:
-                            major_version: int = SLOT_ADVANCE_AVAILABLE_VERSION) -> bool:
+        """Check if our node has permanent replication slots configured.
-        """Check if the given member node has permanent replication slots configured.
 
-        :param my_name: name of the member node to check.
+        :param postgresql: reference to :class:`Postgresql` object.
-        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param member: reference to an object implementing :class:`Tags` interface for
-        :param major_version: postgresql major version.
+                       the node that we are checking permanent logical replication slots for.
 
         :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``.
         """
         role = 'replica'
-        members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role)
+        members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(postgresql.name, role)
-        permanent_slots: Dict[str, Any] = self._get_permanent_slots(role=role, nofailover=nofailover,
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role)
-                                                                    major_version=major_version)
         slots = deepcopy(members_slots)
-        self._merge_permanent_slots(slots, permanent_slots, my_name, major_version)
+        self._merge_permanent_slots(slots, permanent_slots, postgresql.name, postgresql.major_version)
         return len(slots) > len(members_slots) or any(self.is_physical_slot(v) for v in permanent_slots.values())
 
-    def filter_permanent_slots(self, slots: Dict[str, int], major_version: int) -> Dict[str, int]:
+    def filter_permanent_slots(self, postgresql: 'Postgresql', slots: Dict[str, int]) -> Dict[str, int]:
         """Filter out all non-permanent slots from provided *slots* dict.
 
-        :param slots: slot names with LSN values
+        :param postgresql: reference to :class:`Postgresql` object.
-        :param major_version: postgresql major version.
+        :param slots: slot names with LSN values.
 
         :returns: a :class:`dict` object that contains only slots that are known to be permanent.
         """
-        if major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
+        if postgresql.major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
             return {}  # for legacy PostgreSQL we don't support permanent slots on standby nodes
 
-        permanent_slots: Dict[str, Any] = self._get_permanent_slots(role='replica', nofailover=False,
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, RemoteMember('', {}), 'replica')
-                                                                    major_version=major_version)
         members_slots = {slot_name_from_member_name(m.name) for m in self.members}
 
         return {name: value for name, value in slots.items() if name in permanent_slots
                 and (self.is_physical_slot(permanent_slots[name])
                      or self.is_logical_slot(permanent_slots[name]) and name not in members_slots)}
 
-    def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool:
+    def _has_permanent_logical_slots(self, postgresql: 'Postgresql', member: Tags) -> bool:
         """Check if the given member node has permanent ``logical`` replication slots configured.
 
-        :param my_name: name of the member node to check.
+        :param postgresql: reference to a :class:`Postgresql` object.
-        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param member: reference to an object implementing :class:`Tags` interface for
+                       the node that we are checking permanent logical replication slots for.
 
         :returns: ``True`` if any detected replications slots are ``logical``, otherwise ``False``.
         """
-        slots = self.get_replication_slots(my_name, 'replica', nofailover, SLOT_ADVANCE_AVAILABLE_VERSION).values()
+        slots = self.get_replication_slots(postgresql, member, role='replica').values()
         return any(v for v in slots if v.get("type") == "logical")
 
-    def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool) -> bool:
+    def should_enforce_hot_standby_feedback(self, postgresql: 'Postgresql', member: Tags) -> bool:
         """Determine whether ``hot_standby_feedback`` should be enabled for the given member.
 
         The ``hot_standby_feedback`` must be enabled if the current replica has ``logical`` slots,
         or it is working as a cascading replica for the other node that has ``logical`` slots.
 
-        :param my_name: name of the member node to check.
+        :param postgresql: reference to a :class:`Postgresql` object.
-        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param member: reference to an object implementing :class:`Tags` interface for
+                       the node that we are checking permanent logical replication slots for.
 
         :returns: ``True`` if this node or any member replicating from this node has
                   permanent logical slots, otherwise ``False``.
         """
-        if self._has_permanent_logical_slots(my_name, nofailover):
+        if self._has_permanent_logical_slots(postgresql, member):
             return True
 
         if global_config.use_slots:
-            members = [m for m in self.members if m.replicatefrom == my_name and m.name != self.leader_name]
+            name = member.name if isinstance(member, Member) else postgresql.name
-            return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover) for m in members)
+            members = [m for m in self.members if m.replicatefrom == name and m.name != self.leader_name]
+            return any(self.should_enforce_hot_standby_feedback(postgresql, m) for m in members)
         return False
 
-    def get_my_slot_name_on_primary(self, my_name: str, replicatefrom: Optional[str]) -> str:
+    def get_slot_name_on_primary(self, name: str, tags: Tags) -> str:
-        """Canonical slot name for physical replication.
+        """Get the name of physical replication slot for this node on the primary.
 
         .. note::
             P <-- I <-- L
@@ -1181,14 +1183,14 @@ class Cluster(NamedTuple('Cluster',
             In case of cascading replication we have to check not our physical slot, but slot of the replica that
             connects us to the primary.
 
-        :param my_name: the member node name that is replicating.
+        :param name: name of the member node to check.
-        :param replicatefrom: the Intermediate member name that is configured to replicate for cascading replication.
+        :param tags: reference to an object implementing :class:`Tags` interface.
 
-        :returns: The slot name that is in use for physical replication on this no`de.
+        :returns: the slot name on the primary that is in use for physical replication on this node.
         """
-        m = self.get_member(replicatefrom, False) if replicatefrom else None
+        replicatefrom = self.get_member(tags.replicatefrom, False) if tags.replicatefrom else None
-        return self.get_my_slot_name_on_primary(m.name, m.replicatefrom) \
+        return self.get_slot_name_on_primary(replicatefrom.name, replicatefrom) \
-            if isinstance(m, Member) else slot_name_from_member_name(my_name)
+            if isinstance(replicatefrom, Member) else slot_name_from_member_name(name)
 
     @property
     def timeline(self) -> int:

patroni/ha.py

@@ -294,8 +294,8 @@ class Ha(object):
             try:
                 last_lsn = self.state_handler.last_operation()
                 slots = self.cluster.filter_permanent_slots(
-                    {**self.state_handler.slots(), slot_name_from_member_name(self.state_handler.name): last_lsn},
+                    self.state_handler,
-                    self.state_handler.major_version)
+                    {**self.state_handler.slots(), slot_name_from_member_name(self.state_handler.name): last_lsn})
             except Exception:
                 logger.exception('Exception when called state_handler.last_operation()')
         if TYPE_CHECKING:  # pragma: no cover
@@ -1745,7 +1745,7 @@ class Ha(object):
             try:
                 self.load_cluster_from_dcs()
                 global_config.update(self.cluster)
-                self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover)
+                self.state_handler.reset_cluster_info_state(self.cluster, self.patroni)
             except Exception:
                 self.state_handler.reset_cluster_info_state(None)
                 raise
@@ -1902,7 +1902,7 @@ class Ha(object):
                 if not is_promoting and create_slots and self.cluster.leader:
                     err = self._async_executor.try_run_async('copy_logical_slots',
                                                              self.state_handler.slots_handler.copy_logical_slots,
-                                                             args=(self.cluster, create_slots))
+                                                             args=(self.cluster, self.patroni, create_slots))
                     if not err:
                         ret = 'Copying logical slots {0} from the primary'.format(create_slots)
             return ret
@@ -1958,9 +1958,7 @@ class Ha(object):
         cluster = self._failsafe.update_cluster(self.cluster)\
             if self.is_failsafe_mode() and not self.is_leader() else self.cluster
         if cluster:
-            slots = self.state_handler.slots_handler.sync_replication_slots(cluster,
+            slots = self.state_handler.slots_handler.sync_replication_slots(cluster, self.patroni)
-                                                                            self.patroni.nofailover,
-                                                                            self.patroni.replicatefrom)
         # Don't copy replication slots if failsafe_mode is active
         return [] if self.failsafe_is_active() else slots
 

patroni/postgresql/__init__.py

@@ -30,6 +30,7 @@ from ..collections import CaseInsensitiveSet
 from ..dcs import Cluster, Leader, Member, SLOT_ADVANCE_AVAILABLE_VERSION
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
+from ..tags import Tags
 
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Connection as Connection3, Cursor
@@ -424,18 +425,20 @@ class Postgresql(object):
                 self.config.write_postgresql_conf()
                 self.reload()
 
-    def reset_cluster_info_state(self, cluster: Union[Cluster, None], nofailover: bool = False) -> None:
+    def reset_cluster_info_state(self, cluster: Optional[Cluster], tags: Optional[Tags] = None) -> None:
         """Reset monitoring query cache.
 
-        It happens in the beginning of heart-beat loop and on change of `synchronous_standby_names`.
+        .. note::
+            It happens in the beginning of heart-beat loop and on change of `synchronous_standby_names`.
 
         :param cluster: currently known cluster state from DCS
-        :param nofailover: whether this node could become a new primary.
+        :param tags: reference to an object implementing :class:`Tags` interface.
-                           Important when there are logical permanent replication slots because "nofailover"
-                           node could do cascading replication and should enable `hot_standby_feedback`
         """
         self._cluster_info_state = {}
 
+        if not tags:
+            return
+
         if global_config.is_standby_cluster:
             # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback
             self.set_enforce_hot_standby_feedback(False)
@@ -444,13 +447,8 @@ class Postgresql(object):
             # We want to enable hot_standby_feedback if the replica is supposed
             # to have a logical slot or in case if it is the cascading replica.
             self.set_enforce_hot_standby_feedback(not global_config.is_standby_cluster and self.can_advance_slots
-                                                  and cluster.should_enforce_hot_standby_feedback(self.name,
+                                                  and cluster.should_enforce_hot_standby_feedback(self, tags))
-                                                                                                  nofailover))
+            self._has_permanent_slots = cluster.has_permanent_slots(self, tags)
-
-            self._has_permanent_slots = cluster.has_permanent_slots(
-                my_name=self.name,
-                nofailover=nofailover,
-                major_version=self.major_version)
 
     def _cluster_info_state_get(self, name: str) -> Optional[Any]:
         if not self._cluster_info_state:

patroni/postgresql/slots.py

@@ -17,6 +17,7 @@ from .. import global_config
 from ..dcs import Cluster, Leader
 from ..file_perm import pg_perm
 from ..psycopg import OperationalError
+from ..tags import Tags
 
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
@@ -492,8 +493,7 @@ class SlotsHandler:
             self._schedule_load_slots = True
         return create_slots + copy_slots
 
-    def sync_replication_slots(self, cluster: Cluster, nofailover: bool,
+    def sync_replication_slots(self, cluster: Cluster, tags: Tags) -> List[str]:
-                               replicatefrom: Optional[str] = None) -> List[str]:
         """During the HA loop read, check and alter replication slots found in the cluster.
 
         Read physical and logical slots from ``pg_replication_slots``, then compare to those configured in the DCS.
@@ -503,8 +503,7 @@ class SlotsHandler:
         them on replica nodes by copying slot files from the primary.
 
         :param cluster: object containing stateful information for the cluster.
-        :param nofailover: ``True`` if this node has been tagged to not be a failover candidate.
+        :param tags: reference to an object implementing :class:`Tags` interface.
-        :param replicatefrom: the tag containing the node to replicate from.
 
         :returns: list of logical replication slots names that should be copied from the primary.
         """
@@ -513,8 +512,7 @@ class SlotsHandler:
             try:
                 self.load_replication_slots()
 
-                slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role,
+                slots = cluster.get_replication_slots(self._postgresql, tags, show_error=True)
-                                                      nofailover, self._postgresql.major_version, show_error=True)
 
                 self._drop_incorrect_slots(cluster, slots)
 
@@ -524,7 +522,7 @@ class SlotsHandler:
                     self._logical_slots_processing_queue.clear()
                     self._ensure_logical_slots_primary(slots)
                 else:
-                    self.check_logical_slots_readiness(cluster, replicatefrom)
+                    self.check_logical_slots_readiness(cluster, tags)
                     ret = self._ensure_logical_slots_replica(slots)
 
                 self._replication_slots = slots
@@ -550,7 +548,7 @@ class SlotsHandler:
         with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
             yield cur
 
-    def check_logical_slots_readiness(self, cluster: Cluster, replicatefrom: Optional[str]) -> bool:
+    def check_logical_slots_readiness(self, cluster: Cluster, tags: Tags) -> bool:
         """Determine whether all known logical slots are synchronised from the leader.
 
         1) Retrieve the current ``catalog_xmin`` value for the physical slot from the cluster leader, and
@@ -559,13 +557,13 @@ class SlotsHandler:
         3) store logical slot ``catalog_xmin`` when the physical slot ``catalog_xmin`` becomes valid.
 
         :param cluster: object containing stateful information for the cluster.
-        :param replicatefrom: name of the member that should be used to replicate from.
+        :param tags: reference to an object implementing :class:`Tags` interface.
 
         :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise.
         """
         catalog_xmin = None
         if self._logical_slots_processing_queue and cluster.leader:
-            slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
+            slot_name = cluster.get_slot_name_on_primary(self._postgresql.name, tags)
             try:
                 with self._get_leader_connection_cursor(cluster.leader) as cur:
                     cur.execute("SELECT slot_name, catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
@@ -643,16 +641,17 @@ class SlotsHandler:
                 if standby_logical_slot:
                     logger.info('Logical slot %s is safe to be used after a failover', name)
 
-    def copy_logical_slots(self, cluster: Cluster, create_slots: List[str]) -> None:
+    def copy_logical_slots(self, cluster: Cluster, tags: Tags, create_slots: List[str]) -> None:
         """Create logical replication slots on standby nodes.
 
         :param cluster: object containing stateful information for the cluster.
+        :param tags: reference to an object implementing :class:`Tags` interface.
         :param create_slots: list of slot names to copy from the primary.
         """
         leader = cluster.leader
         if not leader:
             return
-        slots = cluster.get_replication_slots(self._postgresql.name, 'replica', False, self._postgresql.major_version)
+        slots = cluster.get_replication_slots(self._postgresql, tags, role='replica')
         copy_slots: Dict[str, Dict[str, Any]] = {}
         with self._get_leader_connection_cursor(leader) as cur:
             try:

tests/test_slots.py

@@ -11,10 +11,18 @@ from patroni.dcs import Cluster, ClusterConfig, Member, Status, SyncState
 from patroni.postgresql import Postgresql
 from patroni.postgresql.misc import fsync_dir
 from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler
+from patroni.tags import Tags
 
 from . import BaseTestPostgresql, psycopg_connect, MockCursor
 
 
+class TestTags(Tags):
+
+    @property
+    def tags(self):
+        return {}
+
+
 @patch('subprocess.call', Mock(return_value=0))
 @patch('patroni.psycopg.connect', psycopg_connect)
 @patch.object(Thread, 'start', Mock())
@@ -34,6 +42,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.cluster = Cluster(True, config, self.leader, Status(0, {'ls': 12345, 'ls2': 12345}),
                                [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
         global_config.update(self.cluster)
+        self.tags = TestTags()
 
     def test_sync_replication_slots(self):
         config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
@@ -43,36 +52,36 @@ class TestSlotsHandler(BaseTestPostgresql):
                           [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
         global_config.update(cluster)
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
-            self.s.sync_replication_slots(cluster, False)
+            self.s.sync_replication_slots(cluster, self.tags)
         self.p.set_role('standby_leader')
         with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
                 patch.object(global_config.__class__, 'is_standby_cluster', PropertyMock(return_value=True)), \
                 patch('patroni.postgresql.slots.logger.debug') as mock_debug:
-            self.s.sync_replication_slots(cluster, False)
+            self.s.sync_replication_slots(cluster, self.tags)
             mock_debug.assert_called_once()
         self.p.set_role('replica')
         with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \
                 patch.object(global_config.__class__, 'is_paused', PropertyMock(return_value=True)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
             config.data['slots'].pop('ls')
-            self.s.sync_replication_slots(cluster, False)
+            self.s.sync_replication_slots(cluster, self.tags)
             mock_drop.assert_not_called()
         self.p.set_role('primary')
         with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
-            self.s.sync_replication_slots(cluster, False)
+            self.s.sync_replication_slots(cluster, self.tags)
         with patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
             alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
             alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
             cluster.members.extend([alias1, alias2])
-            self.s.sync_replication_slots(cluster, False)
+            self.s.sync_replication_slots(cluster, self.tags)
             self.assertEqual(errorlog_mock.call_count, 5)
             ca = errorlog_mock.call_args_list[0][0][1]
             self.assertTrue("test-3" in ca, "non matching {0}".format(ca))
             self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
             with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
-                self.s.sync_replication_slots(cluster, False)
+                self.s.sync_replication_slots(cluster, self.tags)
                 self.p.set_role('replica')
-                self.s.sync_replication_slots(cluster, False)
+                self.s.sync_replication_slots(cluster, self.tags)
 
     def test_cascading_replica_sync_replication_slots(self):
         """Test sync with a cascading replica so physical slots are present on a replica."""
@@ -87,7 +96,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(Postgresql, '_query') as mock_query, \
                 patch.object(Postgresql, 'is_primary', Mock(return_value=False)):
             mock_query.return_value = [('ls', 'logical', 104, 'b', 'a', 5, 12345, 105)]
-            ret = self.s.sync_replication_slots(cluster, False)
+            ret = self.s.sync_replication_slots(cluster, self.tags)
         self.assertEqual(ret, [])
 
     def test_process_permanent_slots(self):
@@ -97,7 +106,7 @@ class TestSlotsHandler(BaseTestPostgresql):
                           None, SyncState.empty(), None, None)
         global_config.update(cluster)
 
-        self.s.sync_replication_slots(cluster, False)
+        self.s.sync_replication_slots(cluster, self.tags)
         with patch.object(Postgresql, '_query') as mock_query:
             self.p.reset_cluster_info_state(None)
             mock_query.return_value = [(
@@ -120,48 +129,48 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('replica')
         self.cluster.slots['ls'] = 12346
         with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
-            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
         with patch.object(SlotsHandler, '_query', Mock(return_value=[('ls', 'logical', 499, 'b', 'a', 5, 100, 500)])), \
                 patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
                 patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
-            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), ['ls'])
         self.cluster.slots['ls'] = 'a'
-        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
+        self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
         self.cluster.config.data['slots']['ls']['database'] = 'b'
         self.cluster.slots['ls'] = '500'
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), ['ls'])
 
     def test_copy_logical_slots(self):
         self.cluster.config.data['slots']['ls']['database'] = 'b'
-        self.s.copy_logical_slots(self.cluster, ['ls'])
+        self.s.copy_logical_slots(self.cluster, self.tags, ['ls'])
         with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)):
-            self.s.copy_logical_slots(self.cluster, ['foo'])
+            self.s.copy_logical_slots(self.cluster, self.tags, ['foo'])
         with patch.object(Cluster, 'leader', PropertyMock(return_value=None)):
-            self.s.copy_logical_slots(self.cluster, ['foo'])
+            self.s.copy_logical_slots(self.cluster, self.tags, ['foo'])
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
-        self.s.copy_logical_slots(self.cluster, ['ls'])
+        self.s.copy_logical_slots(self.cluster, self.tags, ['ls'])
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
                 patch.object(MockCursor, 'fetchall', Mock(side_effect=Exception)):
-            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
+            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, self.tags))
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
                 patch.object(MockCursor, 'fetchall', Mock(return_value=[(False,)])):
-            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
+            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, self.tags))
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
-            self.s.check_logical_slots_readiness(self.cluster, None)
+            self.s.check_logical_slots_readiness(self.cluster, self.tags)
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test_on_promote(self):
         self.s.schedule_advance_slots({'foo': {'bar': 100}})
-        self.s.copy_logical_slots(self.cluster, ['ls'])
+        self.s.copy_logical_slots(self.cluster, self.tags, ['ls'])
         self.s.on_promote()
 
     @unittest.skipIf(os.name == 'nt', "Windows not supported")
@@ -192,11 +201,11 @@ class TestSlotsHandler(BaseTestPostgresql):
         cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}),
                           [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
         global_config.update(cluster)
-        self.s.sync_replication_slots(cluster, False)
+        self.s.sync_replication_slots(cluster, self.tags)
         with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('blabla', 'physical', 12345, None, None, None,
                                                                       None, None)], Exception])) as mock_query, \
                 patch('patroni.postgresql.slots.logger.error') as mock_error:
-            self.s.sync_replication_slots(cluster, False)
+            self.s.sync_replication_slots(cluster, self.tags)
             self.assertEqual(mock_query.call_args[0],
                              ("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", "blabla", '0/303A'))
             self.assertEqual(mock_error.call_args[0][0],