diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index f60d740a..7f792826 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -398,6 +398,7 @@ class Postgresql(object): self._has_permanent_logical_slots or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) + if global_config: self._global_config = global_config def _cluster_info_state_get(self, name: str) -> Optional[Any]: diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 7ddd7722..78e7b427 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 Any, Collection, Dict, List, Tuple, TYPE_CHECKING, Union +from typing import Collection, List, NamedTuple, Tuple, TYPE_CHECKING from ..collections import CaseInsensitiveDict, CaseInsensitiveSet from ..dcs import Cluster @@ -27,7 +27,6 @@ SYNC_REP_PARSER_RE = re.compile(r""" | (?P \) ) | (?P . ) """, re.X) -_EMPTY_SSN = {'type': 'off', 'num': 0, 'members': CaseInsensitiveSet()} def quote_ident(value: str) -> str: @@ -35,37 +34,52 @@ def quote_ident(value: str) -> str: return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value) -def parse_sync_standby_names(value: str) -> Dict[str, Any]: - """Parse postgresql synchronous_standby_names to constituent parts. - Returns dict with the following keys: - * type: 'quorum'|'priority' - * num: int - * members: CaseInsensitiveSet, with name - * has_star: bool - Present if true - If the configuration value can not be parsed, raises a ValueError. +class _SSN(NamedTuple): + """class representing "synchronous_standby_names" value after parsing. - >>> parse_sync_standby_names('')['type'] + :ivar sync_type: possible values: 'off', 'priority', 'quorum' + :ivar has_star: is set to `True` if "synchronous_standby_names" contains '*' + :ivar num: how many nodes are required to be synchronous + :ivar members: collection of standby names listed in "synchronous_standby_names" + """ + sync_type: str + has_star: bool + num: int + members: CaseInsensitiveSet + + +_EMPTY_SSN = _SSN('off', False, 0, CaseInsensitiveSet()) + + +def parse_sync_standby_names(value: str) -> _SSN: + """Parse postgresql synchronous_standby_names to constituent parts. + + :param value: the value of `synchronous_standby_names` + :returns: :class:`_SSN` object + :raises `ValueError`: if the configuration value can not be parsed + + >>> parse_sync_standby_names('').sync_type 'off' - >>> parse_sync_standby_names('FiRsT')['type'] + >>> parse_sync_standby_names('FiRsT').sync_type 'priority' - >>> 'first' in parse_sync_standby_names('FiRsT')['members'] + >>> 'first' in parse_sync_standby_names('FiRsT').members True - >>> set(parse_sync_standby_names('"1"')['members']) + >>> set(parse_sync_standby_names('"1"').members) {'1'} - >>> parse_sync_standby_names(' a , b ')['members'] == {'a', 'b'} + >>> parse_sync_standby_names(' a , b ').members == {'a', 'b'} True - >>> parse_sync_standby_names(' a , b ')['num'] + >>> parse_sync_standby_names(' a , b ').num 1 - >>> parse_sync_standby_names('ANY 4("a",*,b)')['has_star'] + >>> parse_sync_standby_names('ANY 4("a",*,b)').has_star True - >>> parse_sync_standby_names('ANY 4("a",*,b)')['num'] + >>> parse_sync_standby_names('ANY 4("a",*,b)').num 4 >>> parse_sync_standby_names('1') # doctest: +IGNORE_EXCEPTION_DETAIL @@ -100,18 +114,24 @@ def parse_sync_standby_names(value: str) -> Dict[str, Any]: return deepcopy(_EMPTY_SSN) if [t[0] for t in tokens[0:3]] == ['any', 'num', 'parenstart'] and tokens[-1][0] == 'parenend': - result = {'type': 'quorum', 'num': int(tokens[1][1])} + sync_type = 'quorum' + num = int(tokens[1][1]) synclist = tokens[3:-1] elif [t[0] for t in tokens[0:3]] == ['first', 'num', 'parenstart'] and tokens[-1][0] == 'parenend': - result = {'type': 'priority', 'num': int(tokens[1][1])} + sync_type = 'priority' + num = int(tokens[1][1]) synclist = tokens[3:-1] elif [t[0] for t in tokens[0:2]] == ['num', 'parenstart'] and tokens[-1][0] == 'parenend': - result = {'type': 'priority', 'num': int(tokens[0][1])} + sync_type = 'priority' + num = int(tokens[0][1]) synclist = tokens[2:-1] else: - result: Dict[str, Union[int, str, CaseInsensitiveSet]] = {'type': 'priority', 'num': 1} + sync_type = 'priority' + num = 1 synclist = tokens - result['members'] = CaseInsensitiveSet() + + has_star = False + members = CaseInsensitiveSet() for i, (a_type, a_value, a_pos) in enumerate(synclist): if i % 2 == 1: # odd elements are supposed to be commas if len(synclist) == i + 1: # except the last token @@ -121,16 +141,16 @@ def parse_sync_standby_names(value: str) -> Dict[str, Any]: 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'}: - result['members'].add(a_value) + members.add(a_value) elif a_type == 'star': - result['members'].add(a_value) - result['has_star'] = True + members.add(a_value) + has_star = True elif a_type == 'dquot': - result['members'].add(a_value[1:-1].replace('""', '"')) + members.add(a_value[1:-1].replace('""', '"')) else: raise ValueError("Unparseable synchronous_standby_names value %r: Unexpected token %s %r at %d" % (value, a_type, a_value, a_pos)) - return result + return _SSN(sync_type, has_star, num, members) class SyncHandler(object): @@ -150,8 +170,11 @@ class SyncHandler(object): self._ready_replicas = CaseInsensitiveDict({}) # keys: member names, values: connection pids def _handle_synchronous_standby_names_change(self) -> None: - """If synchronous_standby_names has changed we need to check that newly added replicas - have reached self._primary_flush_lsn. Only after that they could be counted as sync.""" + """Handles changes of "synchronous_standby_names" GUC. + + If "synchronous_standby_names" was changed, we need to check that newly added replicas have + reached `self._primary_flush_lsn`. Only after that they could be counted as synchronous. + """ synchronous_standby_names = self._postgresql.synchronous_standby_names() if synchronous_standby_names == self._synchronous_standby_names: return @@ -165,7 +188,7 @@ class SyncHandler(object): # Invalidate cache of "sync" connections for app_name in list(self._ready_replicas.keys()): - if isinstance(self._ssn_data['members'], CaseInsensitiveSet) and app_name not in self._ssn_data['members']: + if app_name not in self._ssn_data.members: del self._ready_replicas[app_name] # Newly connected replicas will be counted as sync only when reached self._primary_flush_lsn @@ -216,7 +239,8 @@ class SyncHandler(object): max_lsn = max(replica_list, key=lambda x: x[3])[3]\ if len(replica_list) > 1 else self._postgresql.last_operation() - assert self._postgresql._global_config is not None + if TYPE_CHECKING: # pragma: no cover + assert self._postgresql._global_config is not None sync_node_count = self._postgresql._global_config.synchronous_node_count\ if self._postgresql.supports_multiple_sync else 1 sync_node_maxlag = self._postgresql._global_config.maximum_lag_on_syncnode @@ -227,8 +251,7 @@ class SyncHandler(object): 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 isinstance(self._ssn_data['members'], CaseInsensitiveSet)\ - and app_name in self._ssn_data['members'] and\ + 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