diff --git a/.gitignore b/.gitignore index c902c6eb..07e227ee 100644 --- a/.gitignore +++ b/.gitignore @@ -27,7 +27,7 @@ lib64 pip-log.txt # Unit test / coverage reports -.coverage +.coverage* .tox nosetests.xml coverage.xml @@ -35,6 +35,7 @@ htmlcov junit.xml features/output* dummy +result.json # Translations *.mo diff --git a/Dockerfile b/Dockerfile index 3e638518..b74cdf24 100644 --- a/Dockerfile +++ b/Dockerfile @@ -94,9 +94,9 @@ RUN set -ex \ /usr/share/locale/??_?? \ /usr/share/postgresql/*/man \ /usr/share/postgresql-common/pg_wrapper \ - /usr/share/vim/vim80/doc \ - /usr/share/vim/vim80/lang \ - /usr/share/vim/vim80/tutor \ + /usr/share/vim/vim*/doc \ + /usr/share/vim/vim*/lang \ + /usr/share/vim/vim*/tutor \ # /var/lib/dpkg/info/* \ && find /usr/bin -xtype l -delete \ && find /var/log -type f -exec truncate --size 0 {} \; \ diff --git a/Dockerfile.citus b/Dockerfile.citus index f52a36e7..5f0164b4 100644 --- a/Dockerfile.citus +++ b/Dockerfile.citus @@ -113,9 +113,9 @@ RUN set -ex \ /usr/share/locale/??_?? \ /usr/share/postgresql/*/man \ /usr/share/postgresql-common/pg_wrapper \ - /usr/share/vim/vim80/doc \ - /usr/share/vim/vim80/lang \ - /usr/share/vim/vim80/tutor \ + /usr/share/vim/vim*/doc \ + /usr/share/vim/vim*/lang \ + /usr/share/vim/vim*/tutor \ # /var/lib/dpkg/info/* \ && find /usr/bin -xtype l -delete \ && find /var/log -type f -exec truncate --size 0 {} \; \ diff --git a/README.rst b/README.rst index c2fc165e..adade6b4 100644 --- a/README.rst +++ b/README.rst @@ -151,7 +151,7 @@ run: YAML Configuration ================== -Go `here `__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml `__. +Go `here `__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml `__. ========================= Environment Configuration diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index fb30bee7..1e6e91b5 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -13,6 +13,8 @@ readonly PATRONI_NAMESPACE="${PATRONI_NAMESPACE%/}" DOCKER_IP=$(hostname --ip-address) readonly DOCKER_IP +export DUMB_INIT_SETSID=0 + case "$1" in haproxy) haproxy -f /etc/haproxy/haproxy.cfg -p /var/run/haproxy.pid -D @@ -72,4 +74,4 @@ export PATRONI_SUPERUSER_SSLKEY="${PATRONI_SUPERUSER_SSLKEY:-$PGSSLKEY}" export PATRONI_SUPERUSER_SSLCERT="${PATRONI_SUPERUSER_SSLCERT:-$PGSSLCERT}" export PATRONI_SUPERUSER_SSLROOTCERT="${PATRONI_SUPERUSER_SSLROOTCERT:-$PGSSLROOTCERT}" -exec python3 /patroni.py postgres0.yml +exec dumb-init python3 /patroni.py postgres0.yml diff --git a/features/steps/citus.py b/features/steps/citus.py index 4dd2ffa6..7277cccc 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -131,5 +131,5 @@ def check_transaction(context, name, time_limit): @step("a transaction finishes in {timeout:d} seconds") def check_transaction_timeout(context, timeout): - assert (datetime.now(tzutc) - context.xact_start).seconds > timeout, \ + assert (datetime.now(tzutc) - context.xact_start).seconds >= timeout, \ "a transaction finished earlier than in {0} seconds".format(timeout) diff --git a/patroni/ctl.py b/patroni/ctl.py index 13959896..9345d18e 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -432,9 +432,9 @@ def print_output(columns: Optional[List[str]], rows: List[List[Any]], alignment: def watching(w: bool, watch: Optional[int], max_count: Optional[int] = None, clear: bool = True) -> Iterator[int]: - """Yield a value every ``x`` seconds. + """Yield a value every ``watch`` seconds. - Used to run a command with a watch-based aproach. + Used to run a command with a watch-based approach. :param w: if ``True`` and *watch* is ``None``, then *watch* assumes the value ``2``. :param watch: amount of seconds to wait before yielding another value. diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index c725e94a..8fbce78f 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 @@ -561,7 +562,7 @@ class SyncState(NamedTuple): """Factory method to parse *value* as synchronisation state information. :param version: optional *version* number for the object. - :param value: (optionally JSON serialised) sychronisation state information + :param value: (optionally JSON serialised) synchronisation state information :returns: constructed :class:`SyncState` object. @@ -969,28 +970,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. " @@ -998,7 +1000,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*. @@ -1008,7 +1010,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. @@ -1016,9 +1018,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 @@ -1029,24 +1031,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:: @@ -1058,23 +1060,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 @@ -1086,7 +1088,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. @@ -1097,14 +1099,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): @@ -1117,76 +1119,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 @@ -1194,14 +1196,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 5699be69..0519ed06 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -305,8 +305,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 @@ -1937,7 +1937,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, self.patroni.nofailover) raise @@ -2094,7 +2094,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 @@ -2150,10 +2150,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, - self.is_paused()) + 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 3f2589e9..8bed8d36 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 @@ -429,18 +430,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) @@ -449,13 +452,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 66bb1f00..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 @@ -320,7 +321,7 @@ class SlotsHandler: ' FULL OUTER JOIN dropped ON true'), name) return (rows[0][0], rows[0][1]) if rows else (False, False) - def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any], paused: bool) -> None: + def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any]) -> None: """Compare required slots and configured as permanent slots with those found, dropping extraneous ones. .. note:: @@ -331,11 +332,10 @@ class SlotsHandler: :param cluster: cluster state information object. :param slots: dictionary of desired slot names as keys with slot attributes as a dictionary value, if known. - :param paused: ``True`` if the patroni cluster is currently in a paused state. """ # drop old replication slots which are not presented in desired slots. for name in set(self._replication_slots) - set(slots): - if not paused and not self.ignore_replication_slot(cluster, name): + if not global_config.is_paused and not self.ignore_replication_slot(cluster, name): active, dropped = self.drop_replication_slot(name) if dropped: logger.info("Dropped unknown replication slot '%s'", name) @@ -493,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, paused: bool = False) -> 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. @@ -504,9 +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 paused: ``True`` if the cluster is in maintenance mode. + :param tags: reference to an object implementing :class:`Tags` interface. :returns: list of logical replication slots names that should be copied from the primary. """ @@ -515,10 +512,9 @@ 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, paused) + self._drop_incorrect_slots(cluster, slots) self._ensure_physical_slots(slots) @@ -526,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 @@ -552,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 @@ -561,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()" @@ -645,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 215c9b84..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,35 +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, paused=True) + 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.""" @@ -86,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): @@ -96,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 = [( @@ -119,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") @@ -191,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],