From 92f4aa2ef9a9284d1bcda8a293f1df6a09ab54b4 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 29 Nov 2023 14:22:49 +0100 Subject: [PATCH] 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 | 138 +++++++++++++++++---------------- patroni/ha.py | 12 ++- patroni/postgresql/__init__.py | 22 +++--- patroni/postgresql/slots.py | 23 +++--- tests/test_slots.py | 55 +++++++------ 5 files changed, 128 insertions(+), 122 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 4a9f998f..28c3734f 100644 --- a/patroni/dcs/__init__.py +++ b/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, *, - show_error: bool = False) -> Dict[str, Dict[str, Any]]: + def get_replication_slots(self, postgresql: 'Postgresql', member: Tags, *, + 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 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 postgresql: reference to :class:`Postgresql` object. + :param member: reference to an object implementing :class:`Tags` interface. + :param role: role of the node, if not set will be taken from *postgresql*. :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) - permanent_slots: Dict[str, Any] = self._get_permanent_slots(role=role, nofailover=nofailover, - major_version=major_version) + name = member.name if isinstance(member, Member) else postgresql.name + role = role or postgresql.role + + 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(): - if not slot_name_re.match(name): - logger.error("Invalid permanent replication slot name '%s'", name) + for slot_name, value in permanent_slots.items(): + if not slot_name_re.match(slot_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): - slots[name] = value + if slot_name != slot_name_from_member_name(name): + slots[slot_name] = value continue if self.is_logical_slot(value): if major_version < SLOT_ADVANCE_AVAILABLE_VERSION: - disabled_permanent_logical_slots.append(name) - elif name in slots: + disabled_permanent_logical_slots.append(slot_name) + 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 nofailover: ``True`` if this node is tagged to not be a failover candidate. - :param major_version: postgresql major version. + :param postgresql: reference to :class:`Postgresql` object. + :param tags: reference to an object implementing :class:`Tags` interface. + :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, - major_version: int = SLOT_ADVANCE_AVAILABLE_VERSION) -> bool: - """Check if the given member node has permanent replication slots configured. + def has_permanent_slots(self, postgresql: 'Postgresql', member: Tags) -> bool: + """Check if our node has permanent replication slots configured. - :param my_name: name of the member node to check. - :param nofailover: ``True`` if this node is tagged to not be a failover candidate. - :param major_version: postgresql major version. + :param postgresql: reference to :class:`Postgresql` object. + :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 there are permanent replication slots configured, otherwise ``False``. """ role = 'replica' - members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) - permanent_slots: Dict[str, Any] = self._get_permanent_slots(role=role, nofailover=nofailover, - major_version=major_version) + members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(postgresql.name, role) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role) 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 major_version: postgresql major version. + :param postgresql: reference to :class:`Postgresql` object. + :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, - major_version=major_version) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, RemoteMember('', {}), 'replica') 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 nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param postgresql: reference to a :class:`Postgresql` object. + :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 nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param postgresql: reference to a :class:`Postgresql` object. + :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] - return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover) for m in members) + name = member.name if isinstance(member, Member) else postgresql.name + 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: - """Canonical slot name for physical replication. + def get_slot_name_on_primary(self, name: str, tags: Tags) -> str: + """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 replicatefrom: the Intermediate member name that is configured to replicate for cascading replication. + :param name: name of the member node to check. + :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 - return self.get_my_slot_name_on_primary(m.name, m.replicatefrom) \ - if isinstance(m, Member) else slot_name_from_member_name(my_name) + replicatefrom = self.get_member(tags.replicatefrom, False) if tags.replicatefrom else None + return self.get_slot_name_on_primary(replicatefrom.name, replicatefrom) \ + if isinstance(replicatefrom, Member) else slot_name_from_member_name(name) @property def timeline(self) -> int: diff --git a/patroni/ha.py b/patroni/ha.py index 3988cb75..ab1bc433 100644 --- a/patroni/ha.py +++ b/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.major_version) + self.state_handler, + {**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, - self.patroni.nofailover, - self.patroni.replicatefrom) + slots = self.state_handler.slots_handler.sync_replication_slots(cluster, self.patroni) # Don't copy replication slots if failsafe_mode is active return [] if self.failsafe_is_active() else slots diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index c1027dad..e373bd3c 100644 --- a/patroni/postgresql/__init__.py +++ b/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. - Important when there are logical permanent replication slots because "nofailover" - node could do cascading replication and should enable `hot_standby_feedback` + :param tags: reference to an object implementing :class:`Tags` interface. """ 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, - nofailover)) - - self._has_permanent_slots = cluster.has_permanent_slots( - my_name=self.name, - nofailover=nofailover, - major_version=self.major_version) + and cluster.should_enforce_hot_standby_feedback(self, tags)) + self._has_permanent_slots = cluster.has_permanent_slots(self, tags) def _cluster_info_state_get(self, name: str) -> Optional[Any]: if not self._cluster_info_state: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 025a88a9..fb9448cd 100644 --- a/patroni/postgresql/slots.py +++ b/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, - replicatefrom: Optional[str] = None) -> List[str]: + def sync_replication_slots(self, cluster: Cluster, tags: Tags) -> 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 replicatefrom: the tag containing the node to replicate from. + :param tags: reference to an object implementing :class:`Tags` interface. :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, - nofailover, self._postgresql.major_version, show_error=True) + slots = cluster.get_replication_slots(self._postgresql, tags, 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: diff --git a/tests/test_slots.py b/tests/test_slots.py index ee1c4ee9..3be2bbd7 100644 --- a/tests/test_slots.py +++ b/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:rep-pass@127.0.0.1:5436/postgres'}) alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1: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],