From e30d96a46810eabaace7d914e29479ef88f7f4b7 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 5 Apr 2023 15:30:55 +0200 Subject: [PATCH] More use of CaseInsensitiveSet (#2631) 1. make `SyncHandler.current_state()` return `CaseInsensitiveSet` instead of `list` objects. 2. take `sync_node_count` and `sync_node_maxlag` from the `Postgresql._global_config` instead of passing them as arguments. 3. Make `AbstractDCS.write_sync_state()` accept any `Collection`-like objects. --- patroni/collections.py | 2 +- patroni/config.py | 7 +++- patroni/dcs/__init__.py | 32 ++++++++++++----- patroni/dcs/kubernetes.py | 25 ++++++++++--- patroni/ha.py | 45 ++++++++++++----------- patroni/postgresql/__init__.py | 21 +++++++---- patroni/postgresql/sync.py | 65 ++++++++++++++++++++-------------- tests/test_ha.py | 33 +++++++++-------- tests/test_sync.py | 29 ++++++++++----- 9 files changed, 166 insertions(+), 93 deletions(-) diff --git a/patroni/collections.py b/patroni/collections.py index a42060bb..f16a6f86 100644 --- a/patroni/collections.py +++ b/patroni/collections.py @@ -35,7 +35,7 @@ class CaseInsensitiveSet(MutableSet): def discard(self, value: str) -> None: self._values.pop(value.lower(), None) - def issubset(self, other: 'CaseInsensitiveSet'): + def issubset(self, other: 'CaseInsensitiveSet') -> bool: return self <= other diff --git a/patroni/config.py b/patroni/config.py index 6b645725..6d4a1cb0 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -107,10 +107,15 @@ class GlobalConfig(object): ret = parse_int(self.get(name)) return default if ret is None else ret + @property + def min_synchronous_nodes(self) -> int: + """:returns: the minimal number of synchronous nodes based on whether strict mode is requested or not.""" + return 1 if self.is_synchronous_mode_strict else 0 + @property def synchronous_node_count(self) -> int: """:returns: currently configured value from the global configuration or 1 if it is not set or invalid.""" - return self.get_int('synchronous_node_count', 0) + return max(self.get_int('synchronous_node_count', 1), self.min_synchronous_nodes) @property def maximum_lag_on_failover(self) -> int: diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index f5a5b89c..ae52e8ad 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -14,7 +14,7 @@ from collections import defaultdict, namedtuple from copy import deepcopy from random import randint from threading import Event, Lock -from typing import Any, Dict, List, Optional, Union +from typing import Any, Collection, Dict, List, Optional, Union from urllib.parse import urlparse, urlunparse, parse_qsl from ..exceptions import PatroniFatalException @@ -1024,13 +1024,24 @@ class AbstractDCS(abc.ABC): """Delete cluster from DCS""" @staticmethod - def sync_state(leader, sync_standby): - """Build sync_state dict - sync_standby dictionary key being kept for backward compatibility + def sync_state(leader: Union[str, None], sync_standby: Union[Collection[str], None]) -> Dict[str, Any]: + """Build sync_state dict. + The sync_standby key being kept for backward compatibility. + :param leader: name of the leader node that manages /sync key + :param sync_standby: collection of currently known synchronous standby node names + :returns: dictionary that later could be serialized to JSON or saved directly to DCS """ - return {'leader': leader, 'sync_standby': sync_standby and ','.join(sorted(sync_standby)) or None} + return {'leader': leader, 'sync_standby': ','.join(sorted(sync_standby)) if sync_standby else None} - def write_sync_state(self, leader, sync_standby, index=None): + def write_sync_state(self, leader: Union[str, None], sync_standby: Union[Collection[str], None], + index: Optional[Union[int, str]] = None) -> bool: + """Write the new synchronous state to DCS. + Calls :func:`sync_state` method to build a dict and than calls DCS specific :func:`set_sync_state_value` method. + :param leader: name of the leader node that manages /sync key + :param sync_standby: collection of currently known synchronous standby node names + :param index: for conditional update of the key/object + :returns: `True` if /sync key was successfully updated + """ sync_value = self.sync_state(leader, sync_standby) return self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index) @@ -1039,8 +1050,13 @@ class AbstractDCS(abc.ABC): """""" @abc.abstractmethod - def set_sync_state_value(self, value, index=None): - """""" + def set_sync_state_value(self, value: str, index: Optional[Union[int, str]] = None) -> bool: + """Set synchronous state in DCS, should be implemented in the child class. + + :param value: the new value of /sync key + :param index: for conditional update of the key/object + :returns: `True` if key/object was successfully updated + """ @abc.abstractmethod def delete_sync_state(self, index=None): diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 5c1db1b5..981980f7 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -16,7 +16,7 @@ from collections import defaultdict from copy import deepcopy from http.client import HTTPException from threading import Condition, Lock, Thread -from typing import Any, Dict, List, Optional +from typing import Any, Collection, Dict, List, Optional, Union from urllib3.exceptions import HTTPError from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ @@ -1267,11 +1267,26 @@ class Kubernetes(AbstractDCS): def set_sync_state_value(self, value, index=None): """Unused""" - def write_sync_state(self, leader, sync_standby, index=None): - return self.patch_or_create(self.sync_path, self.sync_state(leader, sync_standby), index, False) + def write_sync_state(self, leader: Union[str, None], sync_standby: Union[Collection[str], None], + index: Optional[Union[int, str]] = None) -> bool: + """Prepare and write annotations to $SCOPE-sync Endpoint or ConfigMap. - def delete_sync_state(self, index=None): - return self.write_sync_state(None, None, index) + :param leader: name of the leader node that manages /sync key + :param sync_standby: collection of currently known synchronous standby node names + :param index: last known `resource_version` for conditional update of the object + :returns: `True` if update was successful + """ + sync_state = self.sync_state(leader, sync_standby) + return self.patch_or_create(self.sync_path, sync_state, index, False) + + def delete_sync_state(self, index: Optional[str] = None) -> bool: + """Patch annotations of $SCOPE-sync Endpoint or ConfigMap with empty values. + + Effectively it removes "leader" and "sync_standby" annotations from the object. + :param index: last known `resource_version` for conditional update of the object + :returns: `True` if "delete" was successful + """ + return self.write_sync_state(None, None, index=index) def watch(self, leader_index, timeout): if self.__do_not_watch: diff --git a/patroni/ha.py b/patroni/ha.py index ebb69d03..444e8141 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -13,6 +13,7 @@ from typing import List, Optional, Union from . import psycopg from .async_executor import AsyncExecutor, CriticalTask +from .collections import CaseInsensitiveSet from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException from .postgresql.callback_executor import CallbackAction from .postgresql.misc import postgres_version_to_int @@ -472,7 +473,7 @@ class Ha(object): node_to_follow = self._get_node_to_follow(self.cluster) if self.is_synchronous_mode(): - self.state_handler.sync_handler.set_synchronous_standby_names([]) + self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet()) elif self.has_lock(): msg = "starting as readonly because i had the session lock" node_to_follow = None @@ -568,7 +569,7 @@ class Ha(object): """:returns: `True` if failsafe_mode is enabled in global configuration.""" return self.global_config.check_mode('failsafe_mode') - def process_sync_replication(self): + def process_sync_replication(self) -> None: """Process synchronous standby beahvior. Synchronous standbys are registered in two places postgresql.conf and DCS. The order of updating them must @@ -578,36 +579,34 @@ class Ha(object): promoting standbys that were guaranteed to be replicating synchronously. """ if self.is_synchronous_mode(): - sync_node_count = self.global_config.synchronous_node_count - sync_node_maxlag = self.global_config.maximum_lag_on_syncnode - current = [] if self.cluster.sync.is_empty else self.cluster.sync.members - picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count, - sync_node_maxlag) - if set(picked) != set(current): + current = CaseInsensitiveSet(self.cluster.sync.members) + picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster) + + if picked != current: # update synchronous standby list in dcs temporarily to point to common nodes in current and picked - sync_common = list(set(current).intersection(set(allow_promote))) - if set(sync_common) != set(current): - logger.info("Updating synchronous privilege temporarily from %s to %s", current, sync_common) - if not self.dcs.write_sync_state(self.state_handler.name, - sync_common or None, + sync_common = current & allow_promote + if sync_common != current: + logger.info("Updating synchronous privilege temporarily from %s to %s", + list(current), list(sync_common)) + if not self.dcs.write_sync_state(self.state_handler.name, sync_common, index=self.cluster.sync.index): logger.info('Synchronous replication key updated by someone else.') return - # Update db param and wait for x secs + # When strict mode and no suitable replication connections put "*" to synchronous_standby_names if self.global_config.is_synchronous_mode_strict and not picked: - picked = ['*'] + picked = CaseInsensitiveSet('*') logger.warning("No standbys available!") + # Update postgresql.conf and wait 2 secs for changes to become active logger.info("Assigning synchronous standby status to %s", picked) self.state_handler.sync_handler.set_synchronous_standby_names(picked) - if picked and picked[0] != '*' and set(allow_promote) != set(picked) and not allow_promote: + if picked and picked != CaseInsensitiveSet('*') and allow_promote != picked and not allow_promote: # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby time.sleep(2) - _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count, - sync_node_maxlag) - if allow_promote and set(allow_promote) != set(sync_common): + _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster) + if allow_promote and allow_promote != sync_common: try: cluster = self.dcs.get_cluster() except DCSError: @@ -618,11 +617,11 @@ class Ha(object): if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, index=cluster.sync.index): logger.info("Synchronous replication key updated by someone else") return - logger.info("Synchronous standby status assigned to %s", allow_promote) + logger.info("Synchronous standby status assigned to %s", list(allow_promote)) else: if not self.cluster.sync.is_empty and self.dcs.delete_sync_state(index=self.cluster.sync.index): logger.info("Disabled synchronous replication") - self.state_handler.sync_handler.set_synchronous_standby_names([]) + self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet()) def is_sync_standby(self, cluster: Cluster) -> bool: """:returns: `True` if the current node is a synchronous standby.""" @@ -734,7 +733,7 @@ class Ha(object): # promotion until next cycle. TODO: trigger immediate retry of run_cycle return 'Postponing promotion because synchronous replication state was updated by somebody else' self.state_handler.sync_handler.set_synchronous_standby_names( - ['*'] if self.global_config.is_synchronous_mode_strict else []) + CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet()) if self.state_handler.role not in ('master', 'promoted', 'primary'): def on_success(): self._rewind.reset_state() @@ -1088,7 +1087,7 @@ class Ha(object): node_to_follow, leader = None, None if self.is_synchronous_mode(): - self.state_handler.sync_handler.set_synchronous_standby_names([]) + self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet()) # FIXME: with mode offline called from DCS exception handler and handle_long_action_in_progress # there could be an async action already running, calling follow from here will lead diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 472d0adb..a80b76e3 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -12,7 +12,7 @@ from datetime import datetime from dateutil import tz from psutil import TimeoutExpired from threading import current_thread, Lock -from typing import Optional, Union, TYPE_CHECKING +from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING from .bootstrap import Bootstrap from .callback_executor import CallbackAction, CallbackExecutor @@ -30,7 +30,7 @@ from ..exceptions import PostgresConnectionException from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int if TYPE_CHECKING: # pragma: no cover - from .config import GlobalConfig + from ..config import GlobalConfig logger = logging.getLogger(__name__) @@ -157,6 +157,11 @@ class Postgresql(object): def lsn_name(self): return 'lsn' if self._major_version >= 100000 else 'location' + @property + def supports_multiple_sync(self) -> bool: + """:returns: `True` if Postgres version supports more than one synchronous node.""" + return self._major_version >= 90600 + @property def cluster_info_query(self): """Returns the monitoring query with a fixed number of fields. @@ -419,13 +424,16 @@ class Postgresql(object): def received_timeline(self): return self._cluster_info_state_get('received_tli') - def synchronous_commit(self): + def synchronous_commit(self) -> str: + """:returns: "synchronous_commit" GUC value.""" return self._cluster_info_state_get('synchronous_commit') - def synchronous_standby_names(self): + def synchronous_standby_names(self) -> str: + """:returns: "synchronous_standby_names" GUC value.""" return self._cluster_info_state_get('synchronous_standby_names') - def pg_stat_replication(self): + def pg_stat_replication(self) -> List[Dict[str, Any]]: + """:returns: a result set of 'SELECT * FROM pg_stat_replication'.""" return self._cluster_info_state_get('pg_stat_replication') or [] def is_leader(self): @@ -924,7 +932,8 @@ class Postgresql(object): self._cached_replica_timeline = self.get_replica_timeline() return self._cached_replica_timeline - def get_primary_timeline(self): + def get_primary_timeline(self) -> int: + """:returns: current timeline if postgres is running as a primary or 0.""" return self._cluster_info_state_get('timeline') def get_history(self, timeline): diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 43c3d9f2..7d252b96 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -3,9 +3,13 @@ import re import time from copy import deepcopy +from typing import Any, Collection, Dict, Tuple, TYPE_CHECKING from ..collections import CaseInsensitiveDict, CaseInsensitiveSet +from ..dcs import Cluster from ..psycopg import quote_ident as _quote_ident +if TYPE_CHECKING: # pragma: no cover + from . import Postgresql logger = logging.getLogger(__name__) @@ -26,12 +30,12 @@ SYNC_REP_PARSER_RE = re.compile(r""" _EMPTY_SSN = {'type': 'off', 'num': 0, 'members': CaseInsensitiveSet()} -def quote_ident(value): - """Very simplified version of quote_ident""" +def quote_ident(value: str) -> str: + """Very simplified version of `psycopg` :func:`quote_ident` function.""" return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value) -def parse_sync_standby_names(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' @@ -137,7 +141,7 @@ class SyncHandler(object): and the `current_state()` method will count newly added names as "sync" only when they reached memorized LSN and also reported as "sync" by `pg_stat_replication`""" - def __init__(self, postgresql): + def __init__(self, postgresql: 'Postgresql') -> None: self._postgresql = postgresql self._synchronous_standby_names = '' # last known value of synchronous_standby_names self._ssn_data = deepcopy(_EMPTY_SSN) @@ -169,18 +173,21 @@ class SyncHandler(object): self._postgresql.query('SELECT pg_catalog.txid_current()') # Ensure some WAL traffic to move replication self._postgresql.reset_cluster_info_state(None) # Reset internal cache to query fresh values - def current_state(self, cluster, sync_node_count=1, sync_node_maxlag=-1): + def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]: """Finds best candidates to be the synchronous standbys. Current synchronous standby is always preferred, unless it has disconnected or does not want to be a synchronous standby any longer. - Parameter sync_node_maxlag(maximum_lag_on_syncnode) would help swapping unhealthy sync replica in case - if it stops responding (or hung). Please set the value high enough so it won't unncessarily swap sync - standbys during high loads. Any less or equal of 0 value keep the behavior backward compatible and - will not swap. Please note that it will not also swap sync standbys in case where all replicas are hung. - :returns: tuple of candidates list and synchronous standby list.""" + Standbys are selected based on values from the global configuration: + - `maximum_lag_on_syncnode`: would help swapping unhealthy sync replica in case if it stops + responding (or hung). Please set the value high enough so it won't unncessarily swap sync + standbys during high loads. Any value less or equal of 0 keeps the behavior backward compatible. + Please note that it will not also swap sync standbys in case where all replicas are hung. + - `synchronous_node_count`: controlls how many nodes should be set as synchronous. + :returns: tuple of candidates :class:`CaseInsensitiveSet` and synchronous standbys :class:`CaseInsensitiveSet`. + """ self._handle_synchronous_standby_names_change() # Pick candidates based on who has higher replay/remote_write/flush lsn. @@ -209,11 +216,13 @@ class SyncHandler(object): max_lsn = max(replica_list, key=lambda x: x[3])[3]\ if len(replica_list) > 1 else self._postgresql.last_operation() - if self._postgresql.major_version < 90600: - sync_node_count = 1 + 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 - candidates = [] - sync_nodes = [] + 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 @@ -223,28 +232,32 @@ class SyncHandler(object): self._ready_replicas[app_name] = pid if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag: - candidates.append(app_name) + candidates.add(app_name) if sync_state == 'sync' and app_name in self._ready_replicas: - sync_nodes.append(app_name) + sync_nodes.add(app_name) if len(candidates) >= sync_node_count: break return candidates, sync_nodes - def set_synchronous_standby_names(self, value): - """Constructs and sets `synchronous_standby_names` value. + def set_synchronous_standby_names(self, sync: Collection[str]) -> None: + """Constructs and sets "synchronous_standby_names" GUC value. - :param value: list[str] - the list of wanted sync members""" - if value and value != ['*']: - value = [quote_ident(x) for x in value] - - if self._postgresql.major_version >= 90600 and len(value) > 1: - sync_param = '{0} ({1})'.format(len(value), ','.join(value)) + :param sync: set of nodes to sync to + """ + has_asterisk = '*' in sync + if has_asterisk: + sync = ['*'] else: - sync_param = next(iter(value), None) + sync = [quote_ident(x) for x in sync] + + if self._postgresql.supports_multiple_sync and len(sync) > 1: + sync_param = '{0} ({1})'.format(len(sync), ','.join(sync)) + else: + sync_param = next(iter(sync), None) if not (self._postgresql.config.set_synchronous_standby_names(sync_param) - and self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']: + and self._postgresql.state == 'running' and self._postgresql.is_leader()) or has_asterisk: return time.sleep(0.1) # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms diff --git a/tests/test_ha.py b/tests/test_ha.py index bd7516ec..067fd891 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -4,6 +4,7 @@ import os import sys from mock import Mock, MagicMock, PropertyMock, patch, mock_open +from patroni.collections import CaseInsensitiveSet from patroni.config import Config from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory from patroni.dcs.etcd import AbstractEtcdClientWithFailover @@ -795,7 +796,7 @@ class TestHa(PostgresInit): # manual failover when the `other` node isn't available but our name is in the /sync key self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None), sync=('leader1', 'postgresql0')) - self.p.sync_handler.current_state = Mock(return_value=([], [])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet())) self.ha.dcs.write_sync_state = true self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') @@ -811,7 +812,8 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None), sync=('leader1', 'other')) self.p.set_role('replica') - self.p.sync_handler.current_state = Mock(return_value=(['leader1'], ['leader1'])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['leader1']), + CaseInsensitiveSet(['leader1']))) self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') def test_manual_failover_process_no_leader_in_pause(self): @@ -1112,7 +1114,7 @@ class TestHa(PostgresInit): with patch.object(self.ha.dcs, 'delete_sync_state') as mock_delete_sync: self.ha.run_cycle() mock_delete_sync.assert_called_once() - mock_set_sync.assert_called_once_with([]) + mock_set_sync.assert_called_once_with(CaseInsensitiveSet()) mock_set_sync.reset_mock() # Test sync key not touched when not there @@ -1120,14 +1122,15 @@ class TestHa(PostgresInit): with patch.object(self.ha.dcs, 'delete_sync_state') as mock_delete_sync: self.ha.run_cycle() mock_delete_sync.assert_not_called() - mock_set_sync.assert_called_once_with([]) + mock_set_sync.assert_called_once_with(CaseInsensitiveSet()) mock_set_sync.reset_mock() self.ha.is_synchronous_mode = true # Test sync standby not touched when picking the same node - self.p.sync_handler.current_state = Mock(return_value=(['other'], ['other'])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other']), + CaseInsensitiveSet(['other']))) self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) self.ha.run_cycle() mock_set_sync.assert_not_called() @@ -1135,17 +1138,18 @@ class TestHa(PostgresInit): mock_set_sync.reset_mock() # Test sync standby is replaced when switching standbys - self.p.sync_handler.current_state = Mock(return_value=(['other2'], [])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2']), CaseInsensitiveSet())) self.ha.dcs.write_sync_state = Mock(return_value=True) self.ha.run_cycle() - mock_set_sync.assert_called_once_with(['other2']) + mock_set_sync.assert_called_once_with(CaseInsensitiveSet(['other2'])) # Test sync standby is replaced when new standby is joined - self.p.sync_handler.current_state = Mock(return_value=(['other2', 'other3'], ['other2'])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2', 'other3']), + CaseInsensitiveSet(['other2']))) self.ha.dcs.write_sync_state = Mock(return_value=True) self.ha.run_cycle() - self.assertEqual(mock_set_sync.call_args_list[0][0], (['other2'],)) - self.assertEqual(mock_set_sync.call_args_list[1][0], (['other2', 'other3'],)) + self.assertEqual(mock_set_sync.call_args_list[0][0], (CaseInsensitiveSet(['other2']),)) + self.assertEqual(mock_set_sync.call_args_list[1][0], (CaseInsensitiveSet(['other2', 'other3']),)) mock_set_sync.reset_mock() # Test sync standby is not disabled when updating dcs fails @@ -1158,7 +1162,8 @@ class TestHa(PostgresInit): self.ha.dcs.write_sync_state = Mock(return_value=True) self.ha.dcs.get_cluster = Mock(return_value=get_cluster_initialized_with_leader(sync=('leader', 'other'))) # self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) - self.p.sync_handler.current_state = Mock(return_value=(['other2'], ['other2'])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2']), + CaseInsensitiveSet(['other2']))) self.ha.run_cycle() self.ha.dcs.get_cluster.assert_called_once() self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2) @@ -1180,10 +1185,10 @@ class TestHa(PostgresInit): # Test sync set to '*' when synchronous_mode_strict is enabled mock_set_sync.reset_mock() - self.p.sync_handler.current_state = Mock(return_value=([], [])) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet())) with patch('patroni.config.GlobalConfig.is_synchronous_mode_strict', PropertyMock(return_value=True)): self.ha.run_cycle() - mock_set_sync.assert_called_once_with(['*']) + mock_set_sync.assert_called_once_with(CaseInsensitiveSet('*')) def test_sync_replication_become_primary(self): self.ha.is_synchronous_mode = true @@ -1198,7 +1203,7 @@ class TestHa(PostgresInit): # When we just became primary nobody is sync self.assertEqual(self.ha.enforce_primary_role('msg', 'promote msg'), 'promote msg') - mock_set_sync.assert_called_once_with([]) + mock_set_sync.assert_called_once_with(CaseInsensitiveSet()) mock_write_sync.assert_called_once_with('leader', None, index=0) mock_set_sync.reset_mock() diff --git a/tests/test_sync.py b/tests/test_sync.py index 7e6aea47..0cfba7f8 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -2,6 +2,8 @@ import os from mock import Mock, patch +from patroni.collections import CaseInsensitiveSet +from patroni.config import GlobalConfig from patroni.dcs import Cluster, SyncState from patroni.postgresql import Postgresql @@ -20,6 +22,7 @@ class TestSync(BaseTestPostgresql): def setUp(self): super(TestSync, self).setUp() self.p.config.write_postgresql_conf() + self.p._global_config = GlobalConfig({'synchronous_mode': True}) self.s = self.p.sync_handler @patch.object(Postgresql, 'last_operation', Mock(return_value=1)) @@ -35,32 +38,34 @@ class TestSync(BaseTestPostgresql): # sync node is a bit behind of async, but we prefer it anyway with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[self.leadermem.name, 'on', pg_stat_replication]): - self.assertEqual(self.s.current_state(cluster), ([self.leadermem.name], [self.leadermem.name])) + self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.leadermem.name]), + CaseInsensitiveSet([self.leadermem.name]))) # prefer node with sync_state='potential', even if it is slightly behind of async pg_stat_replication[0]['sync_state'] = 'potential' for r in pg_stat_replication: r['write_lsn'] = r.pop('flush_lsn') with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['', 'remote_write', pg_stat_replication]): - self.assertEqual(self.s.current_state(cluster), ([self.leadermem.name], [])) + self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.leadermem.name]), + CaseInsensitiveSet())) # when there are no sync or potential candidates we pick async with the minimal replication lag for i, r in enumerate(pg_stat_replication): r.update(replay_lsn=3 - i, application_name=r['application_name'].upper()) missing = pg_stat_replication.pop(0) with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['', 'remote_apply', pg_stat_replication]): - self.assertEqual(self.s.current_state(cluster), ([self.me.name], [])) + self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet())) # unknown sync node is ignored missing.update(application_name='missing', sync_state='sync') pg_stat_replication.insert(0, missing) with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['', 'remote_apply', pg_stat_replication]): - self.assertEqual(self.s.current_state(cluster), ([self.me.name], [])) + self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet())) # invalid synchronous_standby_names and empty pg_stat_replication with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['a b', 'remote_apply', None]): self.p._major_version = 90400 - self.assertEqual(self.s.current_state(cluster), ([], [])) + self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet(), CaseInsensitiveSet())) def test_set_sync_standby(self): def value_in_conf(): @@ -70,20 +75,26 @@ class TestSync(BaseTestPostgresql): return line.strip() mock_reload = self.p.reload = Mock() - self.s.set_synchronous_standby_names(['n1']) + self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1'])) self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'") mock_reload.assert_called() mock_reload.reset_mock() - self.s.set_synchronous_standby_names(['n1']) + self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1'])) mock_reload.assert_not_called() self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'") - self.s.set_synchronous_standby_names(['n1', 'n2']) + self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1', 'n2'])) mock_reload.assert_called() self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'") mock_reload.reset_mock() - self.s.set_synchronous_standby_names([]) + self.s.set_synchronous_standby_names(CaseInsensitiveSet([])) mock_reload.assert_called() self.assertEqual(value_in_conf(), None) + + mock_reload.reset_mock() + self.p._global_config = GlobalConfig({'synchronous_mode': True}) + self.s.set_synchronous_standby_names(CaseInsensitiveSet('*')) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = '*'")