diff --git a/patroni/config.py b/patroni/config.py index 4d3eea8e..4486fdd3 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -77,10 +77,15 @@ class GlobalConfig(object): """:returns: `True` if cluster is in maintenance mode.""" return self.check_mode('pause') + @property + def is_quorum_commit_mode(self) -> bool: + """:returns: `True` if quorum commit replication is requested""" + return str(self.get('synchronous_mode')).lower() == 'quorum' + @property def is_synchronous_mode(self) -> bool: """:returns: `True` if synchronous replication is requested.""" - return self.check_mode('synchronous_mode') + return self.check_mode('synchronous_mode') is True or self.is_quorum_commit_mode @property def is_synchronous_mode_strict(self) -> bool: diff --git a/patroni/ha.py b/patroni/ha.py index 611d945f..68e07bb2 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -586,7 +586,9 @@ class Ha(object): promoting standbys that were guaranteed to be replicating synchronously. """ if self.is_synchronous_mode(): - picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster) + current_state = self.state_handler.sync_handler.current_state(self.cluster) + picked = current_state.active + allow_promote = current_state.sync voters = CaseInsensitiveSet(self.cluster.sync.voters) if picked != voters: @@ -612,7 +614,7 @@ class Ha(object): if picked and picked != CaseInsensitiveSet('*') and allow_promote != picked: # 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) + allow_promote = self.state_handler.sync_handler.current_state(self.cluster).sync if allow_promote and allow_promote != sync_common: if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, index=sync.index): return logger.info("Synchronous replication key updated by someone else") diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 7f792826..b5245bfb 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -162,6 +162,11 @@ class Postgresql(object): def lsn_name(self) -> str: return 'lsn' if self._major_version >= 100000 else 'location' + @property + def supports_quorum_commit(self) -> bool: + """:returns: `True` if quorum commit is supported by Postgres.""" + return self._major_version >= 100000 + @property def supports_multiple_sync(self) -> bool: """:returns: `True` if Postgres version supports more than one synchronous node.""" diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 78e7b427..edff92c5 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, Tuple, TYPE_CHECKING +from typing import Collection, List, NamedTuple, Optional, Tuple, TYPE_CHECKING from ..collections import CaseInsensitiveDict, CaseInsensitiveSet from ..dcs import Cluster @@ -153,6 +153,14 @@ def parse_sync_standby_names(value: str) -> _SSN: return _SSN(sync_type, has_star, num, members) +class _SyncState(NamedTuple): + sync_type: str + numsync: int + numsync_confirmed: int + sync: CaseInsensitiveSet + active: CaseInsensitiveSet + + class SyncHandler(object): """Class responsible for working with the `synchronous_standby_names`. @@ -196,7 +204,7 @@ 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: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]: + def current_state(self, cluster: Cluster) -> _SyncState: """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 @@ -209,7 +217,8 @@ class SyncHandler(object): 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`. + :param cluster: current cluster topology from DCS + :returns: current synchronous replication state as a :class:`_SyncState` object """ self._handle_synchronous_standby_names_change() @@ -245,41 +254,79 @@ class SyncHandler(object): if self._postgresql.supports_multiple_sync else 1 sync_node_maxlag = self._postgresql._global_config.maximum_lag_on_syncnode - candidates = CaseInsensitiveSet() + 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, _ 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 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 + 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 if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag: - candidates.add(app_name) - if sync_state == 'sync' and app_name in self._ready_replicas: - sync_nodes.add(app_name) - if len(candidates) >= sync_node_count: - break + if self._postgresql._global_config.is_quorum_commit_mode: + # add nodes with nofailover tag only to get enough "active" nodes + if not nofailover or len(active) < sync_node_count: + if app_name in self._ready_replicas: + numsync_confirmed += 1 + active.add(app_name) + else: + active.add(app_name) + if sync_state == 'sync' and app_name in self._ready_replicas: + sync_nodes.add(app_name) + numsync_confirmed += 1 + if len(active) >= sync_node_count: + break - return candidates, sync_nodes + if self._postgresql._global_config.is_quorum_commit_mode: + sync_nodes = CaseInsensitiveSet() if self._ssn_data.has_star else self._ssn_data.members - def set_synchronous_standby_names(self, sync: Collection[str]) -> None: + return _SyncState( + self._ssn_data.sync_type, + 0 if self._ssn_data.has_star else self._ssn_data.num, + numsync_confirmed, + sync_nodes, + active) + + def set_synchronous_standby_names(self, sync: Collection[str], num: Optional[int] = None) -> None: """Constructs and sets "synchronous_standby_names" GUC value. :param sync: set of nodes to sync to + :param num: specifies number of nodes to sync to. The *num* is set only in case if quorum commit is enabled """ - has_asterisk = '*' in sync + # Special case. If sync nodes set is empty but requested num of sync nodes >= 1 + # we want to set synchronous_standby_names to '*' + has_asterisk = '*' in sync or num and num >= 1 and not sync if has_asterisk: sync = ['*'] else: - sync = [quote_ident(x) for x in sync] + sync = [quote_ident(x) for x in sorted(sync)] if self._postgresql.supports_multiple_sync and len(sync) > 1: - sync_param = '{0} ({1})'.format(len(sync), ','.join(sync)) + if num is None: + num = len(sync) + sync_param = ','.join(sync) else: sync_param = next(iter(sync), None) + if TYPE_CHECKING: # pragma: no cover + assert self._postgresql._global_config is not None + + if self._postgresql._global_config.is_quorum_commit_mode and sync or\ + self._postgresql.supports_multiple_sync and len(sync) > 1: + prefix = 'ANY ' if self._postgresql._global_config.is_quorum_commit_mode\ + and self._postgresql.supports_quorum_commit else '' + sync_param = '{0}{1} ({2})'.format(prefix, num, sync_param) + if not (self._postgresql.config.set_synchronous_standby_names(sync_param) and self._postgresql.state == 'running' and self._postgresql.is_leader()) or has_asterisk: return diff --git a/tests/test_ha.py b/tests/test_ha.py index db11706d..59015784 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -17,6 +17,7 @@ from patroni.postgresql.config import ConfigHandler from patroni.postgresql.postmaster import PostmasterProcess from patroni.postgresql.rewind import Rewind from patroni.postgresql.slots import SlotsHandler +from patroni.postgresql.sync import _SyncState from patroni.utils import tzutc from patroni.watchdog import Watchdog @@ -1126,8 +1127,9 @@ class TestHa(PostgresInit): 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=(CaseInsensitiveSet(['other']), - CaseInsensitiveSet(['other']))) + self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1, + 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,14 +1137,16 @@ 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=(CaseInsensitiveSet(['other2']), CaseInsensitiveSet())) + self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0, CaseInsensitiveSet(), + CaseInsensitiveSet(['other2']))) self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) self.ha.run_cycle() 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=(CaseInsensitiveSet(['other2', 'other3']), - CaseInsensitiveSet(['other2']))) + self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1, + CaseInsensitiveSet(['other2']), + CaseInsensitiveSet(['other2', 'other3']))) self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) self.ha.run_cycle() self.assertEqual(mock_set_sync.call_args_list[0][0], (CaseInsensitiveSet(['other2']),)) @@ -1159,8 +1163,9 @@ class TestHa(PostgresInit): self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) 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=(CaseInsensitiveSet(['other2']), - CaseInsensitiveSet(['other2']))) + self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1, + CaseInsensitiveSet(['other2']), + CaseInsensitiveSet(['other2']))) self.ha.run_cycle() self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2) @@ -1182,7 +1187,8 @@ 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=(CaseInsensitiveSet(), CaseInsensitiveSet())) + self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0, 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(CaseInsensitiveSet('*')) diff --git a/tests/test_sync.py b/tests/test_sync.py index a93b98f3..9a2a9d12 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -38,7 +38,8 @@ 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), (CaseInsensitiveSet([self.leadermem.name]), + self.assertEqual(self.s.current_state(cluster), ('priority', 1, 1, + CaseInsensitiveSet([self.leadermem.name]), CaseInsensitiveSet([self.leadermem.name]))) # prefer node with sync_state='potential', even if it is slightly behind of async @@ -46,26 +47,46 @@ class TestSync(BaseTestPostgresql): 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), (CaseInsensitiveSet([self.leadermem.name]), - CaseInsensitiveSet())) + self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(), + CaseInsensitiveSet([self.leadermem.name]))) # 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), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet())) + self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(), + CaseInsensitiveSet([self.me.name]))) # 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), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet())) + self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(), + CaseInsensitiveSet([self.me.name]))) # 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), (CaseInsensitiveSet(), CaseInsensitiveSet())) + self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(), CaseInsensitiveSet())) + + @patch.object(Postgresql, 'last_operation', Mock(return_value=1)) + def test_current_state_quorum(self): + self.p._global_config = GlobalConfig({'synchronous_mode': 'quorum'}) + cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, + SyncState(0, self.me.name, self.leadermem.name, 0), None, None, None) + + pg_stat_replication = [ + {'pid': 100, 'application_name': self.leadermem.name, 'sync_state': 'quorum', 'flush_lsn': 1}, + {'pid': 101, 'application_name': self.other.name, 'sync_state': 'quorum', 'flush_lsn': 2}] + + # sync node is a bit behind of async, but we prefer it anyway + with patch.object(Postgresql, "_cluster_info_state_get", + side_effect=['ANY 1 ({0},"{1}")'.format(self.leadermem.name, self.other.name), + 'on', pg_stat_replication]): + self.assertEqual(self.s.current_state(cluster), + ('quorum', 1, 2, CaseInsensitiveSet([self.other.name, self.leadermem.name]), + CaseInsensitiveSet([self.leadermem.name, self.other.name]))) def test_set_sync_standby(self): def value_in_conf(): @@ -84,6 +105,7 @@ class TestSync(BaseTestPostgresql): mock_reload.assert_not_called() self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'") + mock_reload.reset_mock() self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1', 'n2'])) mock_reload.assert_called() self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'") @@ -98,3 +120,35 @@ class TestSync(BaseTestPostgresql): self.s.set_synchronous_standby_names(CaseInsensitiveSet('*')) mock_reload.assert_called() self.assertEqual(value_in_conf(), "synchronous_standby_names = '*'") + + self.p._global_config = GlobalConfig({'synchronous_mode': 'quorum'}) + mock_reload.reset_mock() + self.s.set_synchronous_standby_names([], 1) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = 'ANY 1 (*)'") + + mock_reload.reset_mock() + self.s.set_synchronous_standby_names(['a', 'b'], 1) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = 'ANY 1 (a,b)'") + + mock_reload.reset_mock() + self.s.set_synchronous_standby_names(['a', 'b'], 3) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = 'ANY 3 (a,b)'") + + self.p._major_version = 90601 + mock_reload.reset_mock() + self.s.set_synchronous_standby_names([], 1) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = '1 (*)'") + + mock_reload.reset_mock() + self.s.set_synchronous_standby_names(['a', 'b'], 1) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = '1 (a,b)'") + + mock_reload.reset_mock() + self.s.set_synchronous_standby_names(['a', 'b'], 3) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = '3 (a,b)'")