diff --git a/Dockerfile b/Dockerfile index 4aa0ce5c..c5b927ee 100644 --- a/Dockerfile +++ b/Dockerfile @@ -25,8 +25,7 @@ RUN set -ex \ | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \ | xargs apt-get install -y vim curl less jq locales haproxy sudo \ python3-etcd python3-kazoo python3-pip busybox \ - net-tools iputils-ping --fix-missing \ - && pip3 install dumb-init \ + net-tools iputils-ping dumb-init --fix-missing \ \ # Cleanup all locales but en_US.UTF-8 && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \ @@ -71,7 +70,7 @@ RUN set -ex \ # Clean up all useless packages and some files && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \ libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \ - exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \ + exim4-config gnupg-agent dirmngr \ git make \ && apt-get autoremove -y \ && apt-get clean -y \ diff --git a/Dockerfile.citus b/Dockerfile.citus index 8693b504..7e6ec18c 100644 --- a/Dockerfile.citus +++ b/Dockerfile.citus @@ -25,7 +25,7 @@ RUN set -ex \ | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \ | xargs apt-get install -y vim curl less jq locales haproxy sudo \ python3-etcd python3-kazoo python3-pip busybox \ - net-tools iputils-ping lsb-release --fix-missing \ + net-tools iputils-ping lsb-release dumb-init --fix-missing \ && if [ $(dpkg --print-architecture) = 'arm64' ]; then \ apt-get install -y postgresql-server-dev-$PG_MAJOR \ git gcc make autoconf \ @@ -42,7 +42,6 @@ RUN set -ex \ && apt-get update -y \ && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \ fi \ - && pip3 install dumb-init \ \ # Cleanup all locales but en_US.UTF-8 && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \ @@ -88,7 +87,7 @@ RUN set -ex \ # Clean up all useless packages and some files && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \ libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \ - exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \ + exim4-config gnupg-agent dirmngr \ postgresql-server-dev-$PG_MAJOR git gcc make autoconf \ libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \ libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \ diff --git a/features/basic_replication.feature b/features/basic_replication.feature index 0e2e8c4b..b58520bd 100644 --- a/features/basic_replication.feature +++ b/features/basic_replication.feature @@ -79,6 +79,7 @@ Feature: basic replication When I add the table buz to postgres2 Then table buz is present on postgres0 after 20 seconds + @reject-duplicate-name Scenario: check graceful rejection when two nodes have the same name Given I start duplicate postgres0 on port 8011 Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log diff --git a/features/environment.py b/features/environment.py index f7690ad2..5906becd 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1144,3 +1144,5 @@ def before_scenario(context, scenario): break if 'dcs-failsafe' in scenario.effective_tags and not context.dcs_ctl._handle: scenario.skip('it is not possible to control state of {0} from tests'.format(context.dcs_ctl.name())) + if 'reject-duplicate-name' in scenario.effective_tags and context.dcs_ctl.name() == 'raft': + scenario.skip('Flaky test with Raft') diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index fb8310ec..dfd71bbd 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -15,7 +15,9 @@ from collections import defaultdict from copy import deepcopy from random import randint from threading import Event, Lock -from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING +from types import ModuleType +from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING, \ + Type, Iterator from urllib.parse import urlparse, urlunparse, parse_qsl from ..exceptions import PatroniFatalException @@ -85,38 +87,83 @@ def dcs_modules() -> List[str]: return [module_prefix + name for _, name, is_pkg in pkgutil.iter_modules([dcs_dirname]) if not is_pkg] -def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': - modules = dcs_modules() +def iter_dcs_classes( + config: Optional[Union['Config', Dict[str, Any]]] = None +) -> Iterator[Tuple[str, Type['AbstractDCS']]]: + """Attempt to import DCS modules that are present in the given configuration. + + .. note:: + If a module successfully imports we can assume that all its requirements are installed. + + :param config: configuration information with possible DCS names as keys. If given, only attempt to import DCS + modules defined in the configuration. Else, if ``None``, attempt to import any supported DCS module. + + :yields: a tuple containing the module ``name`` and the imported DCS class object. + """ + for mod_name in dcs_modules(): + name = mod_name.rpartition('.')[2] + if config is None or name in config: - for module_name in modules: - name = module_name.split('.')[-1] - if name in config: # we will try to import only modules which have configuration section in the config file try: - module = importlib.import_module(module_name) - for key, item in module.__dict__.items(): # iterate through the module content - # try to find implementation of AbstractDCS interface, class name must match with module_name - if key.lower() == name and inspect.isclass(item) and issubclass(item, AbstractDCS): - # propagate some parameters - config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', - 'patronictl', 'ttl', 'retry_timeout') if p in config}) - # From citus section we only need "group" parameter, but will propagate everything just in case. - if isinstance(config.get('citus'), dict): - config[name].update(config['citus']) - return item(config[name]) - except ImportError: - logger.debug('Failed to import %s', module_name) + module = importlib.import_module(mod_name) + dcs_module = find_dcs_class_in_module(module) + if dcs_module: + yield name, dcs_module - available_implementations: List[str] = [] - for module_name in modules: - name = module_name.split('.')[-1] - try: - module = importlib.import_module(module_name) - available_implementations.extend(name for key, item in module.__dict__.items() if key.lower() == name - and inspect.isclass(item) and issubclass(item, AbstractDCS)) - except ImportError: - logger.info('Failed to import %s', module_name) - raise PatroniFatalException("""Can not find suitable configuration of distributed configuration store -Available implementations: """ + ', '.join(sorted(set(available_implementations)))) + except ImportError: + logger.log(logging.DEBUG if config is not None else logging.INFO, + 'Failed to import %s', mod_name) + + +def find_dcs_class_in_module(module: ModuleType) -> Optional[Type['AbstractDCS']]: + """Try to find the implementation of :class:`AbstractDCS` interface in *module* matching the *module* name. + + :param module: Imported DCS module. + + :returns: class with a name matching the name of *module* that implements :class:`AbstractDCS` or ``None`` if not + found. + """ + module_name = module.__name__.rpartition('.')[2] + return next( + (obj for obj_name, obj in module.__dict__.items() + if (obj_name.lower() == module_name + and inspect.isclass(obj) and issubclass(obj, AbstractDCS))), + None) + + +def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': + """Attempt to load a Distributed Configuration Store from known available implementations. + + .. note:: + Using the list of available DCS modules returned by :func:`iter_dcs_modules` attempt to dynamically import and + instantiate the class that implements a DCS using the abstract class :class:`AbstractDCS`. + + Basic top-level configuration parameters retrieved from *config* are propagated to the DCS specific config + before being passed to the module DCS class. + + If no module is found to satisfy configuration then report and log an error. This will cause Patroni to exit. + + :raises :exc:`PatroniFatalException`: if a load of all available DCS modules have been tried and none succeeded. + + :param config: object or dictionary with Patroni configuration. This is normally a representation of the main + Patroni + + :returns: The first successfully loaded DCS module which is an implementation of :class:`AbstractDCS`. + """ + for name, dcs_class in iter_dcs_classes(config): + # Propagate some parameters from top level of config if defined to the DCS specific config section. + config[name].update({ + p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', + 'patronictl', 'ttl', 'retry_timeout') + if p in config}) + # From citus section we only need "group" parameter, but will propagate everything just in case. + if isinstance(config.get('citus'), dict): + config[name].update(config['citus']) + return dcs_class(config[name]) + + raise PatroniFatalException( + f"Can not find suitable configuration of distributed configuration store\n" + f"Available implementations: {', '.join(sorted([n for n, _ in iter_dcs_classes()]))}") _Version = Union[int, str] @@ -596,24 +643,25 @@ class Cluster(NamedTuple): def get_replication_slots(self, my_name: str, role: str, nofailover: bool, major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]: - # if the replicatefrom tag is set on the member - we should not create the replication slot for it on - # the current primary, because that member would replicate from elsewhere. We still create the slot if - # the replicatefrom destination member is currently not a member of the cluster (fallback to the - # primary), or if replicatefrom destination member happens to be the current primary - use_slots = self.use_slots - if role in ('master', 'primary', 'standby_leader'): - slot_members = [m.name for m in self.members if use_slots and m.name != my_name - and (m.replicatefrom is None or m.replicatefrom == my_name - or not self.has_member(m.replicatefrom))] - permanent_slots = self.__permanent_slots if use_slots and \ - role in ('master', 'primary') else self.__permanent_physical_slots - else: - # only manage slots for replicas that replicate from this one, except for the leader among them - slot_members = [m.name for m in self.members if use_slots - and m.replicatefrom == my_name and m.name != self.leader_name] - permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {} + """Lookup configured slot names in the DCS, report issues found and merge with permanent slots. - slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members} + Will log an error if: + + * Conflicting slot names between members are found + * 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 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 [] + + slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'} + for name in slot_members} if len(slots) < len(slot_members): # Find which names are conflicting for a nicer error message @@ -621,11 +669,38 @@ class Cluster(NamedTuple): for name in slot_members: slot_conflicts[slot_name_from_member_name(name)].append(name) logger.error("Following cluster members share a replication slot name: %s", - "; ".join("{} map to {}".format(", ".join(v), k) + "; ".join(f"{', '.join(v)} map to {k}" for k, v in slot_conflicts.items() if len(v) > 1)) - # "merge" replication slots for members with permanent_replication_slots + permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {} + disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( + slots, permanent_slots, my_name, major_version) + + if disabled_permanent_logical_slots and show_error: + logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. " + "Following slots will not be created: %s.", disabled_permanent_logical_slots) + + return slots + + @staticmethod + def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str, + major_version: int) -> List[str]: + """Merge replication *slots* for members with *permanent_slots*. + + Perform validation of configured permanent slot name, skipping invalid names. + + Will update *slots* in-line based on ``type`` of slot, ``physical`` or ``logical``, and name of node. + 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 permanent_slots: dictionary containing slot name key and slot information values. + :param major_version: postgresql major version. + + :returns: List of disabled permanent, logical slot names, if postgresql version < 11. + """ 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) @@ -642,7 +717,8 @@ class Cluster(NamedTuple): if name != slot_name_from_member_name(my_name): slots[name] = value continue - elif value['type'] == 'logical' and value.get('database') and value.get('plugin'): + + if value['type'] == 'logical' and value.get('database') and value.get('plugin'): if major_version < 110000: disabled_permanent_logical_slots.append(name) elif name in slots: @@ -653,12 +729,62 @@ class Cluster(NamedTuple): continue logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name]) + return disabled_permanent_logical_slots - if disabled_permanent_logical_slots and show_error: - logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. " - "Following slots will not be created: %s.", disabled_permanent_logical_slots) + def _get_permanent_slots(self, role: str, nofailover: bool) -> Dict[str, Any]: + """Get configured permanent slot names. - return slots + .. 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 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. + + :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 + + 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. + + If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on + the current primary, because that member would replicate from elsewhere. We still create the slot if + the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the + primary), or if ``replicatefrom`` destination member happens to be the current primary. + + :param my_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. + + :returns: list of member names. + """ + if role in ('master', 'primary', 'standby_leader'): + slot_members = [m.name for m in self.members + if m.name != my_name + and (m.replicatefrom is None + or m.replicatefrom == my_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 + slot_members = [m.name for m in self.members + if m.replicatefrom == my_name and m.name != self.leader_name] + return slot_members def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool: if major_version < 110000: diff --git a/patroni/ha.py b/patroni/ha.py index 3482121d..ced4aabd 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1267,7 +1267,9 @@ class Ha(object): # It could happen if Postgres is still archiving the backlog of WAL files. # If we know that there are replicas that received the shutdown checkpoint # location, we can remove the leader key and allow them to start leader race. - if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location): + + # for a manual failover/switchover with a candidate, we should check the requested candidate only + if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location): self.state_handler.set_role('demoted') with self._async_executor: self.release_leader_key_voluntarily(checkpoint_location) @@ -1371,15 +1373,12 @@ class Ha(object): logger.warning('Failover is possible only to a specific candidate in a paused state') else: if self.is_synchronous_mode(): - if failover.candidate and not self.cluster.sync.matches(failover.candidate): + members = self.get_failover_candidates(check_sync=True) + if failover.candidate and not members: logger.warning('Failover candidate=%s does not match with sync_standbys=%s', failover.candidate, self.cluster.sync.sync_standby) - members = [] - else: - members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)] else: - members = [m for m in self.cluster.members - if not failover.candidate or m.name == failover.candidate] + members = self.get_failover_candidates() if self.is_failover_possible(members, False): # check that there are healthy members ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',)) return ret or 'manual failover: demoting myself' @@ -2031,7 +2030,9 @@ class Ha(object): # It could happen if Postgres is still archiving the backlog of WAL files. # If we know that there are replicas that received the shutdown checkpoint # location, we can remove the leader key and allow them to start leader race. - if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location): + + # for a manual failover/switchover with a candidate, we should check the requested candidate only + if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location): self.dcs.delete_leader(checkpoint_location) status['deleted'] = True else: @@ -2091,3 +2092,24 @@ class Ha(object): name = member.name if member else 'remote_member:{}'.format(uuid.uuid1()) return RemoteMember.from_name_and_data(name, data) + + def get_failover_candidates(self, check_sync: bool = False) -> List[Member]: + """Return list of candidates for either manual or automatic failover. + + Mainly used to later be passed to ``Ha.is_failover_possible()``. + + :param check_sync: if ``True``, also check against the sync key members + + :returns: a list of ``Member`` ojects or an empty list if there is no candidate available + """ + failover = self.cluster.failover + if check_sync: + # TODO: allow manual failover (=no leader specified) to async node + # every sync_standby or the candidate specified if is in sync_standbys + return [m for m in self.cluster.members + if self.cluster.sync.matches(m.name) + and (not failover or not failover.candidate or m.name == failover.candidate)] + else: + # every member or the candidate specified + return [m for m in self.cluster.members + if not failover or not failover.candidate or m.name == failover.candidate] diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index f70ad065..ff6fc751 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -370,7 +370,7 @@ class Rewind(object): # it is the author of archive_command, who is responsible # for not overriding the WALs already present in archive logger.info('Trying to archive %s: %s', wal, cmd) - if self._postgresql.cancellable.call(shlex.split(cmd)) == 0: + if self._postgresql.cancellable.call([cmd], shell=True) == 0: new_name = os.path.join(status_dir, wal + '.done') try: shutil.move(old_name, new_name) diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index d08a09c3..fde81a7a 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -130,7 +130,7 @@ class SlotsHandler(object): self._postgresql = postgresql self._advance = None self._replication_slots: Dict[str, Dict[str, Any]] = {} # already existing replication slots - self._unready_logical_slots: Dict[str, Optional[int]] = {} + self._logical_slots_processing_queue: Dict[str, Optional[int]] = {} self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot') self.schedule() @@ -190,7 +190,8 @@ class SlotsHandler(object): self._replication_slots = replication_slots self._schedule_load_slots = False if self._force_readiness_check: - self._unready_logical_slots = {n: None for n, v in replication_slots.items() if v['type'] == 'logical'} + self._logical_slots_processing_queue = {n: None for n, v in replication_slots.items() + if v['type'] == 'logical'} self._force_readiness_check = False def ignore_replication_slot(self, cluster: Cluster, name: str) -> bool: @@ -327,10 +328,10 @@ class SlotsHandler(object): self._ensure_physical_slots(slots) if self._postgresql.is_leader(): - self._unready_logical_slots.clear() + self._logical_slots_processing_queue.clear() self._ensure_logical_slots_primary(slots) elif cluster.slots and slots: - self.check_logical_slots_readiness(cluster, nofailover, replicatefrom) + self.check_logical_slots_readiness(cluster, replicatefrom) ret = self._ensure_logical_slots_replica(cluster, slots) @@ -347,48 +348,100 @@ class SlotsHandler(object): 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, nofailover: bool, replicatefrom: Optional[str]) -> None: + def check_logical_slots_readiness(self, cluster: Cluster, replicatefrom: Optional[str]) -> 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 + 2) using previously stored list of "unready" logical slots, those which have yet to be checked hence have no + stored slot attributes, + 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. + + :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise. + """ catalog_xmin = None - if self._unready_logical_slots and cluster.leader: + if self._logical_slots_processing_queue and cluster.leader: slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom) 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()" " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)", - ([n for n, v in self._unready_logical_slots.items() if v is None] + [slot_name],)) + ([n for n, v in self._logical_slots_processing_queue.items() + if v is None] + [slot_name],)) slots = {row[0]: row[1] for row in cur} if slot_name not in slots: - return logger.warning('Physical slot %s does not exist on the primary', slot_name) + logger.warning('Physical slot %s does not exist on the primary', slot_name) + return False catalog_xmin = slots.pop(slot_name) except Exception as e: - return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e) - # Remember catalog_xmin of logical slots on the primary when catalog_xmin of - # the physical slot became valid. Logical slots on replica will be safe to use after - # promote when catalog_xmin of the physical slot overtakes these values. - if catalog_xmin is not None: - for name, value in slots.items(): - self._unready_logical_slots[name] = value - else: # Replica isn't streaming or the hot_standby_feedback isn't enabled - try: - cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean") - row = cur.fetchone() - if row and not row[0]: - logger.error('Logical slot failover requires "hot_standby_feedback".' - ' Please check postgresql.auto.conf') - except Exception as e: - logger.error('Failed to check the hot_standby_feedback setting: %r', e) - return # since `catalog_xmin` isn't valid further checks don't make any sense + logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e) + return False - for name in list(self._unready_logical_slots): - value = self._replication_slots.get(name) - # The logical slot on a replica is safe to use when the physical replica slot on the primary: - # 1. has a nonzero/non-null catalog_xmin - # 2. has a catalog_xmin that is not newer (greater) than the catalog_xmin of any slot on the standby - # 3. overtook the catalog_xmin of remembered values of logical slots on the primary. - if not value or catalog_xmin is not None and\ - self._unready_logical_slots[name] <= catalog_xmin <= value['catalog_xmin']: - del self._unready_logical_slots[name] - if value: + if not self._update_pending_logical_slot_primary(slots, catalog_xmin): + return False # since `catalog_xmin` isn't valid further checks don't make any sense + + self._ready_logical_slots(catalog_xmin) + return True + + def _update_pending_logical_slot_primary(self, slots: Dict[str, Any], catalog_xmin: Optional[int] = None) -> bool: + """Store pending logical slot information for ``catalog_xmin`` on the primary. + + Remember ``catalog_xmin`` of logical slots on the primary when ``catalog_xmin`` of the physical slot became + valid. Logical slots on replica will be safe to use after promote when ``catalog_xmin`` of the physical slot + overtakes these values. + + :param slots: dictionary of slot information from the primary + :param catalog_xmin: ``catalog_xmin`` of the physical slot used by this replica to stream changes from primary. + + :returns: ``False`` if any issue was faced while processing, ``True`` otherwise. + """ + if catalog_xmin is not None: + for name, value in slots.items(): + self._logical_slots_processing_queue[name] = value + return True + + # Replica isn't streaming or the hot_standby_feedback isn't enabled + try: + cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean") + row = cur.fetchone() + if row and not row[0]: + logger.error('Logical slot failover requires "hot_standby_feedback".' + ' Please check postgresql.auto.conf') + except Exception as e: + logger.error('Failed to check the hot_standby_feedback setting: %r', e) + return False + + def _ready_logical_slots(self, primary_physical_catalog_xmin: Optional[int] = None) -> None: + """Ready logical slots by comparing primary physical slot ``catalog_xmin`` to logical ``catalog_xmin``. + + The logical slot on a replica is safe to use when the physical replica slot on the primary: + + 1. has a nonzero/non-null ``catalog_xmin`` represented by ``primary_physical_xmin``. + 2. has a ``catalog_xmin`` that is not newer (greater) than the ``catalog_xmin`` of any slot on the standby + 3. overtook the ``catalog_xmin`` of remembered values of logical slots on the primary. + + :param primary_physical_catalog_xmin: is the value retrieved from ``pg_catalog.pg_get_replication_slots()`` for + the physical replication slot on the primary. + """ + # Make a copy of processing queue keys as a list as the queue dictionary is modified inside the loop. + for name in list(self._logical_slots_processing_queue): + primary_logical_catalog_xmin = self._logical_slots_processing_queue[name] + standby_logical_slot = self._replication_slots.get(name, {}) + standby_logical_catalog_xmin = standby_logical_slot.get('catalog_xmin', 0) + if TYPE_CHECKING: # pragma: no cover + assert primary_logical_catalog_xmin is not None + + if ( + not standby_logical_slot + or primary_physical_catalog_xmin is not None + and primary_logical_catalog_xmin <= primary_physical_catalog_xmin <= standby_logical_catalog_xmin + ): + + del self._logical_slots_processing_queue[name] + + 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: @@ -433,7 +486,7 @@ class SlotsHandler(object): shutil.rmtree(slot_dir) os.rename(slot_tmp_dir, slot_dir) fsync_dir(slot_dir) - self._unready_logical_slots[name] = None + self._logical_slots_processing_queue[name] = None fsync_dir(self._postgresql.slots_handler.pg_replslot_dir) self._postgresql.start() @@ -446,6 +499,6 @@ class SlotsHandler(object): if self._advance: self._advance.on_promote() - if self._unready_logical_slots: + if self._logical_slots_processing_queue: logger.warning('Logical replication slots that might be unsafe to use after promote: %s', - set(self._unready_logical_slots)) + set(self._logical_slots_processing_queue)) diff --git a/tests/test_ha.py b/tests/test_ha.py index 116cc5af..b6b2d334 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -280,8 +280,10 @@ class TestHa(PostgresInit): self.p.follow = true self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') self.p.is_running = true + ha_dcs_orig_name = self.ha.dcs.__class__.__name__ self.ha.dcs.__class__.__name__ = 'Raft' self.assertEqual(self.ha.run_cycle(), 'started as a secondary') + self.ha.dcs.__class__.__name__ = ha_dcs_orig_name def test_recover_former_primary(self): self.p.follow = false diff --git a/tests/test_rewind.py b/tests/test_rewind.py index b5858d8e..a4fafcff 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -239,7 +239,7 @@ class TestRewind(BaseTestPostgresql): with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])): # successful archive_command call - with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)): + with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)) as mock_subprocess_call: get_guc_value_res = [ 'on', 'command %f', 'always', 'command %f', @@ -252,6 +252,10 @@ class TestRewind(BaseTestPostgresql): '000000000000000000000000', 'command 000000000000000000000000'), mock_logger_info.call_args[0]) mock_logger_info.reset_mock() + mock_subprocess_call.assert_called_once() + self.assertEqual(mock_subprocess_call.call_args.args[0], ['command 000000000000000000000000']) + self.assertEqual(mock_subprocess_call.call_args.kwargs['shell'], True) + mock_subprocess_call.reset_mock() # failed archive_command call with patch.object(CancellableSubprocess, 'call', Mock(return_value=1)): diff --git a/tests/test_slots.py b/tests/test_slots.py index a962dbe3..3f21998f 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -93,7 +93,7 @@ class TestSlotsHandler(BaseTestPostgresql): def test__ensure_logical_slots_replica(self): self.p.set_role('replica') self.cluster.slots['ls'] = 12346 - with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()): + with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) self.s._schedule_load_slots = False with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\ @@ -121,12 +121,12 @@ class TestSlotsHandler(BaseTestPostgresql): self.s.copy_logical_slots(self.cluster, ['ls']) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\ patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)): - self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None)) + self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\ patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))): - self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None)) + self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))): - self.s.check_logical_slots_readiness(self.cluster, False, None) + self.s.check_logical_slots_readiness(self.cluster, None) @patch.object(Postgresql, 'stop', Mock(return_value=True)) @patch.object(Postgresql, 'start', Mock(return_value=True))