gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

Browse Source
This commit is contained in:
Alexander Kukushkin
2023-11-29 14:34:56 +01:00
parent 193f5f1d7d 92f4aa2ef9
commit 91a6059055
12 changed files with 148 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -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
Dockerfile
+3 -3
View File
@@ -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 {} \; \
Dockerfile.citus
+3 -3
View File
@@ -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 {} \; \
README.rst
+1 -1
View File
@@ -151,7 +151,7 @@ run:
YAML Configuration
==================
Go `here <https://github.com/zalando/patroni/blob/master/docs/SETTINGS.rst>`__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml <https://github.com/zalando/patroni/blob/master/postgres0.yml>`__.
Go `here <https://github.com/zalando/patroni/blob/master/docs/dynamic_configuration.rst>`__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml <https://github.com/zalando/patroni/blob/master/postgres0.yml>`__.
=========================
Environment Configuration
docker/entrypoint.sh
+3 -1
View File
@@ -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
features/steps/citus.py
+1 -1
View File
@@ -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)
patroni/ctl.py
+2 -2
View File
@@ -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.
patroni/dcs/__init__.py
+71 -69
View File
@@ -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:
patroni/ha.py
+5 -8
View File
@@ -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
patroni/postgresql/__init__.py
+10 -12
View File
@@ -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:
patroni/postgresql/slots.py
+14 -17
View File
@@ -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:
tests/test_slots.py
+33 -23
View File
@@ -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:[email protected]:5436/postgres'})
alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
cluster.members.extend([alias1, alias2])
self.s.sync_replication_slots(cluster, False)
self.s.sync_replication_slots(cluster, self.tags)
self.assertEqual(errorlog_mock.call_count, 5)
ca = errorlog_mock.call_args_list[0][0][1]
self.assertTrue("test-3" in ca, "non matching {0}".format(ca))
self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
self.s.sync_replication_slots(cluster, False)
self.s.sync_replication_slots(cluster, self.tags)
self.p.set_role('replica')
self.s.sync_replication_slots(cluster, False)
self.s.sync_replication_slots(cluster, self.tags)
def test_cascading_replica_sync_replication_slots(self):
"""Test sync with a cascading replica so physical slots are present on a replica."""
@@ -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],