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

Please codacy

Browse Source
This commit is contained in:
Alexander Kukushkin
2023-07-18 15:35:32 +02:00
parent e6d251bda0
commit 300740c919
2 changed files with 100 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/postgresql/sync.py
+56 -43
View File
@@ -3,7 +3,7 @@ import re
import time
from copy import deepcopy
from typing import Collection, List, NamedTuple, Optional, Tuple, TYPE_CHECKING
from typing import Collection, Iterator, List, NamedTuple, Optional, Tuple, TYPE_CHECKING
from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
from ..dcs import Cluster
@@ -137,7 +137,7 @@ def parse_sync_standby_names(value: str) -> _SSN:
if len(synclist) == i + 1: # except the last token
raise ValueError("Unparseable synchronous_standby_names value %r: Unexpected token %s %r at %d" %
(value, a_type, a_value, a_pos))
elif a_type != 'comma':
if a_type != 'comma':
raise ValueError("Unparseable synchronous_standby_names value %r: ""Got token %s %r while"
" expecting comma at %d" % (value, a_type, a_value, a_pos))
elif a_type in {'ident', 'first', 'any'}:
@@ -209,6 +209,49 @@ BEGIN
END;$$""")
self._postgresql.reset_cluster_info_state(None) # Reset internal cache to query fresh values
def _get_replica_list(self, cluster: Cluster) -> Iterator[Tuple[int, str, str, int, bool]]:
"""Yields candidates based on higher replay/remote_write/flush lsn."""
# What column from pg_stat_replication we want to sort on? Choose based on ``synchronous_commit`` value.
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})
# 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):
yield (pid, member.name, sync_state, replica_lsn, bool(member.nofailover))
def _process_replica_readiness(self, cluster: Cluster, replica_list: List[Tuple[int, str, str, int, bool]]) -> None:
"""Flags replicas as truely "synchronous" when they caught up with "_primary_flush_lsn"."""
if TYPE_CHECKING: # pragma: no cover
assert self._postgresql.global_config is not None
for pid, app_name, sync_state, replica_lsn, _ in replica_list:
if app_name not in self._ready_replicas and app_name in self._ssn_data.members:
if self._postgresql.global_config.is_quorum_commit_mode:
# When quorum commit is enabled we can't check against cluster.sync because nodes
# are written there when at least one of them caught up with _primary_flush_lsn.
if replica_lsn >= self._primary_flush_lsn\
and (sync_state == 'quorum' or (not self._postgresql.supports_quorum_commit
and sync_state in ('sync', 'potential'))):
self._ready_replicas[app_name] = pid
elif cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn:
# 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 we known that it managed to catch up
self._ready_replicas[app_name] = pid
def current_state(self, cluster: Cluster) -> _SyncState:
"""Finds best candidates to be the synchronous standbys.
@@ -227,31 +270,12 @@ 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'
replica_list = list(self._get_replica_list(cluster))
self._process_replica_readiness(cluster, replica_list)
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()
active = CaseInsensitiveSet()
sync_nodes = CaseInsensitiveSet()
numsync_confirmed = 0
if TYPE_CHECKING: # pragma: no cover
assert self._postgresql.global_config is not None
@@ -259,24 +283,13 @@ END;$$""")
if self._postgresql.supports_multiple_sync else 1
sync_node_maxlag = self._postgresql.global_config.maximum_lag_on_syncnode
active = CaseInsensitiveSet()
sync_nodes = CaseInsensitiveSet()
numsync_confirmed = 0
# 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, nofailover in sorted(replica_list, key=lambda x: x[4]):
if app_name not in self._ready_replicas and app_name in self._ssn_data.members:
if self._postgresql.global_config.is_quorum_commit_mode:
# When quorum commit is enabled we can't check against cluster.sync because nodes
# are written there when at least one of them caught up with _primary_flush_lsn.
if replica_lsn >= self._primary_flush_lsn\
and (sync_state == 'quorum' or (not self._postgresql.supports_quorum_commit
and sync_state in ('sync', 'potential'))):
self._ready_replicas[app_name] = pid
elif cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn:
# 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 we known that it managed to catch up
self._ready_replicas[app_name] = pid
# When checking *maximum_lag_on_syncnode* we want to compare with the most
# up-to-date replica or with cluster LSN if there is only one replica.
max_lsn = max(replica_list, key=lambda x: x[3])[3]\
if len(replica_list) > 1 else self._postgresql.last_operation()
# Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
for _, app_name, sync_state, replica_lsn, nofailover in sorted(replica_list, key=lambda x: x[4]):
if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
if self._postgresql.global_config.is_quorum_commit_mode:
# add nodes with nofailover tag only to get enough "active" nodes
patroni/quorum.py
+44 -33
View File
@@ -157,37 +157,7 @@ class QuorumStateResolver(object):
if cur_transition[0] == 'restart':
break
def _generate_transitions(self) -> Iterator[Tuple[str, str, int, CaseInsensitiveSet]]:
logger.debug("Quorum state: leader %s quorum %s, voters %s, numsync %s, sync %s, "
"numsync_confirmed %s, active %s, sync_wanted %s leader_wanted %s",
self.leader, self.quorum, self.voters, self.numsync, self.sync,
self.numsync_confirmed, self.active, self.sync_wanted, self.leader_wanted)
try:
if self.leader_wanted != self.leader:
voters = (self.voters - CaseInsensitiveSet([self.leader_wanted])) | CaseInsensitiveSet([self.leader])
if not self.sync:
# If sync is empty we need to update synchronous_standby_names first
numsync = len(voters) - self.quorum
yield from self.sync_update(numsync, CaseInsensitiveSet(voters))
# If leader changed we need to add the old leader to quorum (voters)
yield from self.quorum_update(self.quorum, CaseInsensitiveSet(voters), self.leader_wanted)
# right after promote there could be no replication connections yet
if not self.sync & self.active:
return # give another loop_wait seconds for replicas to reconnect before removing them from quorum
else:
self.check_invariants()
except QuorumError as e:
logger.warning('%s', e)
yield from self.quorum_update(len(self.sync) - self.numsync, self.sync)
assert self.leader == self.leader_wanted
# numsync_confirmed could be 0 after restart/failover, we will calculate it from quorum
if self.numsync_confirmed == 0 and self.sync & self.active:
self.numsync_confirmed = min(len(self.sync & self.active), len(self.voters) - self.quorum)
logger.debug('numsync_confirmed=0, adjusting it to %d', self.numsync_confirmed)
# Handle non steady state cases
def __handle_non_steady_cases(self) -> Iterator[Tuple[str, str, int, CaseInsensitiveSet]]:
if self.sync < self.voters:
logger.debug("Case 1: synchronous_standby_names subset of DCS state")
# Case 1: quorum is superset of sync nodes. In the middle of changing quorum.
@@ -233,8 +203,7 @@ class QuorumStateResolver(object):
if self.numsync == self.sync_wanted and safety_margin > 0 and self.numsync > self.numsync_confirmed:
yield from self.quorum_update(len(self.sync) - self.numsync, self.voters)
# We are in a steady state point. Find if desired state is different and act accordingly.
def __remove_gone_nodes(self) -> Iterator[Tuple[str, str, int, CaseInsensitiveSet]]:
# If any nodes have gone away, evict them
to_remove = self.sync - self.active
if to_remove and self.sync == to_remove:
@@ -266,6 +235,7 @@ class QuorumStateResolver(object):
yield from self.quorum_update(quorum, voters, adjust_quorum=False)
yield from self.sync_update(numsync, sync)
def __add_new_nodes(self) -> Iterator[Tuple[str, str, int, CaseInsensitiveSet]]:
# If any new nodes, join them to quorum
to_add = self.active - self.sync
if to_add:
@@ -289,6 +259,7 @@ class QuorumStateResolver(object):
adjust_quorum=sync_wanted > self.numsync_confirmed)
yield from self.sync_update(sync_wanted, CaseInsensitiveSet(self.sync | to_add))
def __handle_replication_factor_change(self) -> Iterator[Tuple[str, str, int, CaseInsensitiveSet]]:
# Apply requested replication factor change
sync_increase = min(self.sync_wanted, len(self.sync)) - self.numsync
if sync_increase > 0:
@@ -303,3 +274,43 @@ class QuorumStateResolver(object):
yield from self.quorum_update(len(self.voters) - self.numsync - sync_increase, self.voters,
adjust_quorum=self.sync_wanted > self.numsync_confirmed)
yield from self.sync_update(self.numsync + sync_increase, self.sync)
def _generate_transitions(self) -> Iterator[Tuple[str, str, int, CaseInsensitiveSet]]:
logger.debug("Quorum state: leader %s quorum %s, voters %s, numsync %s, sync %s, "
"numsync_confirmed %s, active %s, sync_wanted %s leader_wanted %s",
self.leader, self.quorum, self.voters, self.numsync, self.sync,
self.numsync_confirmed, self.active, self.sync_wanted, self.leader_wanted)
try:
if self.leader_wanted != self.leader:
voters = (self.voters - CaseInsensitiveSet([self.leader_wanted])) | CaseInsensitiveSet([self.leader])
if not self.sync:
# If sync is empty we need to update synchronous_standby_names first
numsync = len(voters) - self.quorum
yield from self.sync_update(numsync, CaseInsensitiveSet(voters))
# If leader changed we need to add the old leader to quorum (voters)
yield from self.quorum_update(self.quorum, CaseInsensitiveSet(voters), self.leader_wanted)
# right after promote there could be no replication connections yet
if not self.sync & self.active:
return # give another loop_wait seconds for replicas to reconnect before removing them from quorum
else:
self.check_invariants()
except QuorumError as e:
logger.warning('%s', e)
yield from self.quorum_update(len(self.sync) - self.numsync, self.sync)
assert self.leader == self.leader_wanted
# numsync_confirmed could be 0 after restart/failover, we will calculate it from quorum
if self.numsync_confirmed == 0 and self.sync & self.active:
self.numsync_confirmed = min(len(self.sync & self.active), len(self.voters) - self.quorum)
logger.debug('numsync_confirmed=0, adjusting it to %d', self.numsync_confirmed)
yield from self.__handle_non_steady_cases()
# We are in a steady state point. Find if desired state is different and act accordingly.
yield from self.__remove_gone_nodes()
yield from self.__add_new_nodes()
yield from self.__handle_replication_factor_change()