From 2be64e5131bc77d6c82b370b029d32e1ad65b5af Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 18 Aug 2023 13:36:32 +0200 Subject: [PATCH] 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 | 46 ++++++++++++++++++---------------- patroni/postgresql/__init__.py | 16 +++++++++--- patroni/postgresql/slots.py | 12 ++++++--- tests/test_slots.py | 3 +++ 4 files changed, 49 insertions(+), 28 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 279ce265..b76e1bda 100644 --- a/patroni/dcs/__init__.py +++ b/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 diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 7f3bdccb..9b1991a0 100644 --- a/patroni/postgresql/__init__.py +++ b/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: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 319119ce..e4543c71 100644 --- a/patroni/postgresql/slots.py +++ b/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) diff --git a/tests/test_slots.py b/tests/test_slots.py index df8c6955..add0fdf7 100644 --- a/tests/test_slots.py +++ b/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()