diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index df6801b7..1ceab579 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -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 diff --git a/patroni/quorum.py b/patroni/quorum.py index a41597b5..2cb4539c 100644 --- a/patroni/quorum.py +++ b/patroni/quorum.py @@ -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()