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

More typing in sync.py (#2666)

Browse Source 
- make parse_sync_standby_names() return NamedTuple instead of dict
- fix little issue in Postgresql.reset_cluster_info_state(), it should check global_config independently from cluster and cluster.config
This commit is contained in:
Alexander Kukushkin
2023-05-09 12:47:34 +02:00
committed by GitHub
parent 76b3b99de2
commit 13164daf28
2 changed files with 58 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/postgresql/__init__.py
+1
View File
@@ -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]:
patroni/postgresql/sync.py
+57 -34
View File
@@ -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<parenend> \) )
| (?P<JUNK> . )
""", 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