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

Refactor replica_list (#2790)

Browse Source 
As suggested in
https://github.com/zalando/patroni/pull/2668/files#r1276115738, introduce a couple of classes that represent a single replica and collection of replicas.
This commit is contained in:
Alexander Kukushkin
2023-07-31 15:52:43 +02:00
committed by GitHub
parent a26e46cf76
commit e4703d4f74
1 changed files with 88 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/postgresql/sync.py
+88 -36
View File
@@ -153,6 +153,72 @@ def parse_sync_standby_names(value: str) -> _SSN:
return _SSN(sync_type, has_star, num, members)
class _Replica(NamedTuple):
"""Class representing a single replica that is eligible to be synchronous.
Attributes are taken from ``pg_stat_replication`` view and respective ``Cluster.members``.
:ivar pid: PID of walsender process.
:ivar application_name: matches with the ``Member.name``.
:ivar sync_state: possible values are: ``async``, ``potential``, ``quorum``, and ``sync``.
:ivar lsn: ``write_lsn``, ``flush_lsn``, or ``replay_lsn``, depending on the value of ``synchronous_commit`` GUC.
:ivar nofailover: whether the corresponding member has ``nofailover`` tag set to ``True``.
"""
pid: int
application_name: str
sync_state: str
lsn: int
nofailover: bool
class _ReplicaList(List[_Replica]):
"""A collection of :class:``_Replica`` objects.
Values are reverse ordered by ``_Replica.sync_state`` and ``_Replica.lsn``.
That is, first there will be replicas that have ``sync_state`` == ``sync``, even if they are not
the most up-to-date in term of write/flush/replay LSN. It helps to keep the result of chosing new
synchronous nodes consistent in case if a synchronous standby member is slowed down OR async node
is receiving changes faster than the sync member. Such cases would trigger sync standby member
swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``).
:ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one
element in the list we take value of ``pg_current_wal_lsn()``.
"""
def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:
"""Create :class:``_ReplicaList`` object.
:param postgresql: reference to :class:``Postgresql`` object.
:param cluster: currently known cluster state from DCS.
"""
super().__init__()
# We want to prioritize candidates based on `write_lsn``, ``flush_lsn``, or ``replay_lsn``.
# Which column exactly to pick depends on the values of ``synchronous_commit`` GUC.
sort_col = {
'remote_apply': 'replay',
'remote_write': 'write'
}.get(postgresql.synchronous_commit(), 'flush') + '_lsn'
members = CaseInsensitiveDict({m.name: m for m in cluster.members})
for row in postgresql.pg_stat_replication():
member = members.get(row['application_name'])
# We want to consider only rows from ``pg_stat_replication` that:
# 1. are known to be streaming (write/flush/replay LSN are not NULL).
# 2. can be mapped to a ``Member`` of the ``Cluster``:
# a. ``Member`` doesn't have ``nosync`` tag set;
# b. PostgreSQL on the member is known to be running and accepting client connections.
if member and row[sort_col] is not None and member.is_running and not member.tags.get('nosync', False):
self.append(_Replica(row['pid'], row['application_name'],
row['sync_state'], row[sort_col], bool(member.nofailover)))
# Prefer replicas that are in state ``sync`` and with higher values of ``write``/``flush``/``replay`` LSN.
self.sort(key=lambda r: (r.sync_state, r.lsn), reverse=True)
self.max_lsn = max(self, key=lambda x: x.lsn).lsn if len(self) > 1 else postgresql.last_operation()
class SyncHandler(object):
"""Class responsible for working with the `synchronous_standby_names`.
@@ -201,6 +267,21 @@ BEGIN
END;$$""")
self._postgresql.reset_cluster_info_state(None) # Reset internal cache to query fresh values
def _process_replica_readiness(self, cluster: Cluster, replica_list: _ReplicaList) -> None:
"""Flags replicas as truly "synchronous" when they have caught up with ``_primary_flush_lsn``.
:param cluster: current cluster topology from DCS
:param replica_list: collection of replicas that we want to evaluate.
"""
for replica in replica_list:
# if standby name is listed in the /sync key we can count it as synchronous, otherwise
# it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
if replica.application_name not in self._ready_replicas\
and replica.application_name in self._ssn_data.members\
and (cluster.sync.matches(replica.application_name)
or replica.sync_state == 'sync' and replica.lsn >= self._primary_flush_lsn):
self._ready_replicas[replica.application_name] = replica.pid
def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
"""Finds best candidates to be the synchronous standbys.
@@ -218,31 +299,8 @@ END;$$""")
"""
self._handle_synchronous_standby_names_change()
# Pick candidates based on who has higher replay/remote_write/flush lsn.
sort_col = {
'remote_apply': 'replay',
'remote_write': 'write'
}.get(self._postgresql.synchronous_commit(), 'flush') + '_lsn'
pg_stat_replication = [(r['pid'], r['application_name'], r['sync_state'], r[sort_col])
for r in self._postgresql.pg_stat_replication()
if r[sort_col] is not None]
members = CaseInsensitiveDict({m.name: m for m in cluster.members})
replica_list: List[Tuple[int, str, str, int, bool]] = []
# pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
# That is, alphabetically they are in the reversed order of priority.
# Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result
# consistent in case if a synchronous standby member is slowed down OR async node receiving
# changes faster than the sync member (very rare but possible).
# Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold.
for pid, app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: r[2:4], reverse=True):
member = members.get(app_name)
if member and member.is_running and not member.tags.get('nosync', False):
replica_list.append((pid, member.name, sync_state, replica_lsn, bool(member.nofailover)))
max_lsn = max(replica_list, key=lambda x: x[3])[3]\
if len(replica_list) > 1 else self._postgresql.last_operation()
replica_list = _ReplicaList(self._postgresql, cluster)
self._process_replica_readiness(cluster, replica_list)
if TYPE_CHECKING: # pragma: no cover
assert self._postgresql.global_config is not None
@@ -253,17 +311,11 @@ END;$$""")
candidates = CaseInsensitiveSet()
sync_nodes = CaseInsensitiveSet()
# Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
for pid, app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[4]):
# if standby name is listed in the /sync key we can count it as synchronous, otherwice
# it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
if app_name not in self._ready_replicas and app_name in self._ssn_data.members and\
(cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn):
self._ready_replicas[app_name] = pid
if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
candidates.add(app_name)
if sync_state == 'sync' and app_name in self._ready_replicas:
sync_nodes.add(app_name)
for replica in sorted(replica_list, key=lambda x: x.nofailover):
if sync_node_maxlag <= 0 or replica_list.max_lsn - replica.lsn <= sync_node_maxlag:
candidates.add(replica.application_name)
if replica.sync_state == 'sync' and replica.application_name in self._ready_replicas:
sync_nodes.add(replica.application_name)
if len(candidates) >= sync_node_count:
break