diff --git a/patroni/ha.py b/patroni/ha.py index 05081fb8..6b95c6d0 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -458,7 +458,7 @@ class Ha(object): node_to_follow = self._get_node_to_follow(self.cluster) if self.is_synchronous_mode(): - self.state_handler.config.set_synchronous_standby([]) + self.state_handler.sync_handler.set_synchronous_standby_names([]) elif self.has_lock(): msg = "starting as readonly because i had the session lock" node_to_follow = None @@ -568,9 +568,9 @@ class Ha(object): if self.is_synchronous_mode(): sync_node_count = self.patroni.config['synchronous_node_count'] current = self.cluster.sync.leader and self.cluster.sync.members or [] - picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster, sync_node_count, - self.patroni.config[ - 'maximum_lag_on_syncnode']) + picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count, + self.patroni.config[ + 'maximum_lag_on_syncnode']) if set(picked) != set(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))) @@ -588,15 +588,15 @@ class Ha(object): logger.warning("No standbys available!") logger.info("Assigning synchronous standby status to %s", picked) - self.state_handler.config.set_synchronous_standby(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: # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby time.sleep(2) - _, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster, - sync_node_count, - self.patroni.config[ - 'maximum_lag_on_syncnode']) + _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, + sync_node_count, + self.patroni.config[ + 'maximum_lag_on_syncnode']) if allow_promote and set(allow_promote) != set(sync_common): try: cluster = self.dcs.get_cluster() @@ -612,7 +612,7 @@ class Ha(object): else: if self.cluster.sync.leader and self.dcs.delete_sync_state(index=self.cluster.sync.index): logger.info("Disabled synchronous replication") - self.state_handler.config.set_synchronous_standby([]) + self.state_handler.sync_handler.set_synchronous_standby_names([]) def is_sync_standby(self, cluster): return cluster.leader and cluster.sync.leader == cluster.leader.name \ @@ -721,7 +721,8 @@ class Ha(object): # Somebody else updated sync state, it may be due to us losing the lock. To be safe, postpone # 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.config.set_synchronous_standby(['*'] if self.is_synchronous_mode_strict() else []) + self.state_handler.sync_handler.set_synchronous_standby_names( + ['*'] if self.is_synchronous_mode_strict() else []) if self.state_handler.role != 'master': def on_success(): self._rewind.reset_state() @@ -1044,7 +1045,7 @@ class Ha(object): node_to_follow, leader = None, None if self.is_synchronous_mode(): - self.state_handler.config.set_synchronous_standby([]) + self.state_handler.sync_handler.set_synchronous_standby_names([]) # 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 6793fd84..afd40e11 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -22,7 +22,7 @@ from .connection import Connection, get_connection_cursor from .misc import parse_history, parse_lsn, postgres_major_version_to_int from .postmaster import PostmasterProcess from .slots import SlotsHandler -from .validator import CaseInsensitiveDict +from .sync import SyncHandler from .. import psycopg from ..exceptions import PostgresConnectionException from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int @@ -85,6 +85,7 @@ class Postgresql(object): self.__thread_ident = current_thread().ident self.slots_handler = SlotsHandler(self) + self.sync_handler = SyncHandler(self) self._callback_executor = CallbackExecutor() self.__cb_called = False @@ -176,12 +177,12 @@ class Postgresql(object): extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), " + "pg_catalog.current_setting('synchronous_standby_names'), " - "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT application_name, sync_state," + + "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT w.pid as pid, application_name, sync_state," + " pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn," + " pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn," + " pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn " + "FROM pg_catalog.pg_stat_get_wal_senders() w," + - " pg_catalog.pg_stat_get_activity(pid)" + + " pg_catalog.pg_stat_get_activity(w.pid)" + " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name) if self._is_synchronous_mode and self.role == 'master' else "'on', '', NULL") @@ -403,6 +404,15 @@ class Postgresql(object): def received_timeline(self): return self._cluster_info_state_get('received_tli') + def synchronous_commit(self): + return self._cluster_info_state_get('synchronous_commit') + + def synchronous_standby_names(self): + return self._cluster_info_state_get('synchronous_standby_names') + + def pg_stat_replication(self): + return self._cluster_info_state_get('pg_stat_replication') or [] + def is_leader(self): try: return bool(self._cluster_info_state_get('timeline')) @@ -1112,54 +1122,6 @@ class Postgresql(object): logger.exception('Could not remove data directory %s', self._data_dir) self.move_data_directory() - def pick_synchronous_standby(self, cluster, sync_node_count=1, sync_node_maxlag=-1): - """Finds the best candidate to be the synchronous standby. - - 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. - """ - if self._major_version < 90600: - sync_node_count = 1 - members = CaseInsensitiveDict({m.name: m for m in cluster.members}) - candidates = [] - sync_nodes = [] - replica_list = [] - # Pick candidates based on who has higher replay/remote_write/flush lsn. - synchronous_commit = self._cluster_info_state_get('synchronous_commit') - sort_col = {'remote_apply': 'replay', 'remote_write': 'write'}.get(synchronous_commit, 'flush') + '_lsn' - pg_stat_replication = [(r['application_name'], r['sync_state'], r[sort_col]) - for r in self._cluster_info_state_get('pg_stat_replication') or [] - if r[sort_col] is not None] - # 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 app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: (r[1], r[2]), reverse=True): - member = members.get(app_name) - if member and member.is_running and not member.tags.get('nosync', False): - replica_list.append((member.name, sync_state, replica_lsn, bool(member.nofailover))) - - max_lsn = max(replica_list, key=lambda x: x[2])[2] if len(replica_list) > 1 else self.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, _ in sorted(replica_list, key=lambda x: x[3]): - if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag: - candidates.append(app_name) - if sync_state == 'sync': - sync_nodes.append(app_name) - if len(candidates) >= sync_node_count: - break - - return candidates, sync_nodes - def schedule_sanity_checks_after_pause(self): """ After coming out of pause we have to: diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index b732d01f..3a438c43 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -12,21 +12,14 @@ from .validator import CaseInsensitiveDict, recovery_parameters,\ transform_postgresql_parameter_value, transform_recovery_parameter_value from ..dcs import slot_name_from_member_name, RemoteMember from ..exceptions import PatroniFatalException -from ..psycopg import quote_ident as _quote_ident from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \ validate_directory, is_subpath logger = logging.getLogger(__name__) -SYNC_STANDBY_NAME_RE = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$') PARAMETER_RE = re.compile(r'([a-z_]+)\s*=\s*') -def quote_ident(value): - """Very simplified version of quote_ident""" - return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value) - - def conninfo_uri_parse(dsn): ret = {} r = urlparse(dsn) @@ -1044,23 +1037,19 @@ class ConfigHandler(object): else: logger.info('No PostgreSQL configuration items changed, nothing to reload.') - def set_synchronous_standby(self, sync_members): - """Sets a node to be synchronous standby and if changed does a reload for PostgreSQL.""" - if sync_members and sync_members != ['*']: - sync_members = [quote_ident(x) for x in sync_members] - if self._postgresql.major_version >= 90600 and len(sync_members) > 1: - sync_param = '{0} ({1})'.format(len(sync_members), ','.join(sync_members)) - else: - sync_param = next(iter(sync_members), None) - if sync_param != self._synchronous_standby_names: - if sync_param is None: + def set_synchronous_standby_names(self, value): + """Updates synchronous_standby_names and reloads if necessary. + :returns: True if value was updated.""" + if value != self._synchronous_standby_names: + if value is None: self._server_parameters.pop('synchronous_standby_names', None) else: - self._server_parameters['synchronous_standby_names'] = sync_param - self._synchronous_standby_names = sync_param + self._server_parameters['synchronous_standby_names'] = value + self._synchronous_standby_names = value if self._postgresql.state == 'running': self.write_postgresql_conf() self._postgresql.reload() + return True @property def effective_configuration(self): diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py new file mode 100644 index 00000000..bbc30044 --- /dev/null +++ b/patroni/postgresql/sync.py @@ -0,0 +1,258 @@ +import logging +import re +import time + +from copy import deepcopy + +from .validator import CaseInsensitiveDict +from ..psycopg import quote_ident as _quote_ident + +logger = logging.getLogger(__name__) + +SYNC_STANDBY_NAME_RE = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$') +SYNC_REP_PARSER_RE = re.compile(r""" + (?P [fF][iI][rR][sS][tT] ) + | (?P [aA][nN][yY] ) + | (?P \s+ ) + | (?P [A-Za-z_][A-Za-z_0-9\$]* ) + | (?P " (?: [^"]+ | "" )* " ) + | (?P [*] ) + | (?P \d+ ) + | (?P , ) + | (?P \( ) + | (?P \) ) + | (?P . ) + """, re.X) +_EMPTY_SSN = {'type': 'off', 'num': 0, 'members': CaseInsensitiveDict({})} + + +def quote_ident(value): + """Very simplified version of quote_ident""" + return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value) + + +def parse_sync_standby_names(value): + """Parse postgresql synchronous_standby_names to constituent parts. + Returns dict with the following keys: + * type: 'quorum'|'priority' + * num: int + * members: CaseInsensitiveDict, with names as keys + * has_star: bool - Present if true + If the configuration value can not be parsed, raises a ValueError. + + >>> parse_sync_standby_names('')['type'] + 'off' + + >>> parse_sync_standby_names('FiRsT')['type'] + 'priority' + + >>> parse_sync_standby_names('FiRsT')['members'] + {'FiRsT': True} + + >>> parse_sync_standby_names('"1"')['members'] + {'1': True} + + >>> parse_sync_standby_names(' a , b ')['members'] + {'a': True, 'b': True} + + >>> parse_sync_standby_names(' a , b ')['num'] + 1 + + >>> parse_sync_standby_names('ANY 4("a",*,b)')['has_star'] + True + + >>> parse_sync_standby_names('ANY 4("a",*,b)')['num'] + 4 + + >>> parse_sync_standby_names('1') # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + ValueError: Unparseable synchronous_standby_names value + + >>> parse_sync_standby_names('a,') # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + ValueError: Unparseable synchronous_standby_names value + + >>> parse_sync_standby_names('ANY 4("a" b,"c c")') # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + ValueError: Unparseable synchronous_standby_names value + + >>> parse_sync_standby_names('FIRST 4("a",)') # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + ValueError: Unparseable synchronous_standby_names value + + >>> parse_sync_standby_names('2 (,)') # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + ValueError: Unparseable synchronous_standby_names value + """ + tokens = [(m.lastgroup, m.group(0), m.start()) + for m in SYNC_REP_PARSER_RE.finditer(value) + if m.lastgroup != 'space'] + if not tokens: + 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])} + 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])} + 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])} + synclist = tokens[2:-1] + else: + result = {'type': 'priority', 'num': 1} + synclist = tokens + result['members'] = CaseInsensitiveDict({}) + 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 + 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': + 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'][a_value] = True + elif a_type == 'star': + result['members'][a_value] = True + result['has_star'] = True + elif a_type == 'dquot': + result['members'][a_value[1:-1].replace('""', '"')] = True + else: + raise ValueError("Unparseable synchronous_standby_names value %r: Unexpected token %s %r at %d" % + (value, a_type, a_value, a_pos)) + return result + + +class SyncHandler(object): + """Class responsible for working with the `synchronous_standby_names`. + + Sync standbys are chosen based on their state in `pg_stat_replication`. + When `synchronous_standby_names` is changed we memorize the `_primary_flush_lsn` + 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): + self._postgresql = postgresql + self._synchronous_standby_names = '' # last known value of synchronous_standby_names + self._ssn_data = deepcopy(_EMPTY_SSN) + self._primary_flush_lsn = 0 + # "sync" replication connections, that were verified to reach self._primary_flush_lsn at some point + self._ready_replicas = CaseInsensitiveDict({}) # keys: member names, values: connection pids + + def _handle_synchronous_standby_names_change(self): + """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.""" + synchronous_standby_names = self._postgresql.synchronous_standby_names() + if synchronous_standby_names == self._synchronous_standby_names: + return False + + self._synchronous_standby_names = synchronous_standby_names + try: + self._ssn_data = parse_sync_standby_names(synchronous_standby_names) + except ValueError as e: + logger.warning('%s', e) + self._ssn_data = deepcopy(_EMPTY_SSN) + + # Invalidate cache of "sync" connections + for app_name in list(self._ready_replicas.keys()): + 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 + self._primary_flush_lsn = self._postgresql.last_operation() + 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): + """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.""" + + 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' + + 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 = [] + # 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() + + if self._postgresql.major_version < 90600: + sync_node_count = 1 + + candidates = [] + sync_nodes = [] + # 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 + # ig 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 and app_name in cluster.sync.members or + sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn): + self._ready_replicas[app_name] = pid + + if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag: + candidates.append(app_name) + if sync_state == 'sync' and app_name in self._ready_replicas: + sync_nodes.append(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. + + :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)) + else: + sync_param = next(iter(value), None) + + if not (self._postgresql.config.set_synchronous_standby_names(sync_param) and + self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']: + return + + time.sleep(0.1) # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms + + # Reset internal cache to query fresh values + self._postgresql.reset_cluster_info_state(None) + + # timeline == 0 -- indicates that this is the replica, shoudn't ever happen + if self._postgresql.get_master_timeline() > 0: + self._handle_synchronous_standby_names_change() diff --git a/tests/__init__.py b/tests/__init__.py index 286bafbc..91cc99df 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -101,7 +101,9 @@ class MockCursor(object): elif sql.startswith('WITH slots AS (SELECT slot_name, active'): self.results = [(False, True)] elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'): - self.results = [(1, 2, 1, 0, False, 1, 1, None, None, [{"slot_name": "ls", "confirmed_flush_lsn": 12345}])] + self.results = [(1, 2, 1, 0, False, 1, 1, None, None, + [{"slot_name": "ls", "confirmed_flush_lsn": 12345}], + 'on', 'n1', None)] elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'): self.results = [(False, 2)] elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'): diff --git a/tests/test_ha.py b/tests/test_ha.py index 66be6a71..c88fc774 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -757,7 +757,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.pick_synchronous_standby = Mock(return_value=([], [])) + self.p.sync_handler.current_state = Mock(return_value=([], [])) self.ha.dcs.write_sync_state = true self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') @@ -773,7 +773,7 @@ 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.pick_synchronous_standby = Mock(return_value=(['leader1'], ['leader1'])) + self.p.sync_handler.current_state = Mock(return_value=(['leader1'], ['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): @@ -1062,7 +1062,7 @@ class TestHa(PostgresInit): def test_process_sync_replication(self): self.ha.has_lock = true - mock_set_sync = self.p.config.set_synchronous_standby = Mock() + mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock() self.p.name = 'leader' # Test sync key removed when sync mode disabled @@ -1085,7 +1085,7 @@ class TestHa(PostgresInit): self.ha.is_synchronous_mode = true # Test sync standby not touched when picking the same node - self.p.pick_synchronous_standby = Mock(return_value=(['other'], ['other'])) + self.p.sync_handler.current_state = Mock(return_value=(['other'], ['other'])) self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) self.ha.run_cycle() mock_set_sync.assert_not_called() @@ -1093,13 +1093,13 @@ class TestHa(PostgresInit): mock_set_sync.reset_mock() # Test sync standby is replaced when switching standbys - self.p.pick_synchronous_standby = Mock(return_value=(['other2'], [])) + self.p.sync_handler.current_state = Mock(return_value=(['other2'], [])) self.ha.dcs.write_sync_state = Mock(return_value=True) self.ha.run_cycle() mock_set_sync.assert_called_once_with(['other2']) # Test sync standby is replaced when new standby is joined - self.p.pick_synchronous_standby = Mock(return_value=(['other2', 'other3'], ['other2'])) + self.p.sync_handler.current_state = Mock(return_value=(['other2', 'other3'], ['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'],)) @@ -1116,7 +1116,7 @@ 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.pick_synchronous_standby = Mock(return_value=(['other2'], ['other2'])) + self.p.sync_handler.current_state = Mock(return_value=(['other2'], ['other2'])) self.ha.run_cycle() self.ha.dcs.get_cluster.assert_called_once() self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2) @@ -1139,14 +1139,14 @@ class TestHa(PostgresInit): # Test sync set to '*' when synchronous_mode_strict is enabled mock_set_sync.reset_mock() self.ha.is_synchronous_mode_strict = true - self.p.pick_synchronous_standby = Mock(return_value=([], [])) + self.p.sync_handler.current_state = Mock(return_value=([], [])) self.ha.run_cycle() mock_set_sync.assert_called_once_with(['*']) def test_sync_replication_become_master(self): self.ha.is_synchronous_mode = true - mock_set_sync = self.p.config.set_synchronous_standby = Mock() + mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock() self.p.is_leader = false self.p.set_role('replica') self.ha.has_lock = true diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 8e5c1fa9..b8725218 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -10,7 +10,7 @@ from mock import Mock, MagicMock, PropertyMock, patch, mock_open import patroni.psycopg as psycopg from patroni.async_executor import CriticalTask -from patroni.dcs import Cluster, RemoteMember, SyncState +from patroni.dcs import RemoteMember from patroni.exceptions import PostgresConnectionException, PatroniException from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE from patroni.postgresql.bootstrap import Bootstrap @@ -640,79 +640,12 @@ class TestPostgresql(BaseTestPostgresql): self.p._state = 'starting' self.assertIsNone(self.p.wait_for_startup()) - @patch.object(Postgresql, 'last_operation', Mock(return_value=2)) - def test_pick_sync_standby(self): - cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, - SyncState(0, self.me.name, self.leadermem.name), None, None, None) - - with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ - 'on', - [{'application_name': self.leadermem.name, 'sync_state': 'sync', 'flush_lsn': 1}, - {'application_name': self.me.name, 'sync_state': 'async', 'flush_lsn': 2}, - {'application_name': self.other.name, 'sync_state': 'async', 'flush_lsn': 2}] - ]): - - self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [self.leadermem.name])) - - with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ - 'remote_write', - [{'application_name': self.leadermem.name, 'sync_state': 'potential', 'write_lsn': 1}, - {'application_name': self.me.name, 'sync_state': 'async', 'write_lsn': 2}, - {'application_name': self.other.name, 'sync_state': 'async', 'write_lsn': 2}] - ]): - self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [])) - - with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ - 'remote_apply', - [{'application_name': self.me.name.upper(), 'sync_state': 'async', 'replay_lsn': 2}, - {'application_name': self.other.name, 'sync_state': 'async', 'replay_lsn': 1}] - ]): - self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], [])) - - with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ - 'remote_apply', - [{'application_name': 'missing', 'sync_state': 'sync', 'replay_lsn': 3}, - {'application_name': self.me.name, 'sync_state': 'async', 'replay_lsn': 2}, - {'application_name': self.other.name, 'sync_state': 'async', 'replay_lsn': 1}] - ]): - self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], [])) - - with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['remote_apply', []]): - self.p._major_version = 90400 - self.assertEqual(self.p.pick_synchronous_standby(cluster), ([], [])) - - def test_set_sync_standby(self): - def value_in_conf(): - with open(os.path.join(self.p.data_dir, 'postgresql.conf')) as f: - for line in f: - if line.startswith('synchronous_standby_names'): - return line.strip() - - mock_reload = self.p.reload = Mock() - self.p.config.set_synchronous_standby(['n1']) - self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'") - mock_reload.assert_called() - - mock_reload.reset_mock() - self.p.config.set_synchronous_standby(['n1']) - mock_reload.assert_not_called() - self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'") - - self.p.config.set_synchronous_standby(['n1', 'n2']) - mock_reload.assert_called() - self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'") - - mock_reload.reset_mock() - self.p.config.set_synchronous_standby([]) - mock_reload.assert_called() - self.assertEqual(value_in_conf(), None) - def test_get_server_parameters(self): config = {'synchronous_mode': True, 'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'} self.p.config.get_server_parameters(config) config['synchronous_mode_strict'] = True self.p.config.get_server_parameters(config) - self.p.config.set_synchronous_standby('foo') + self.p.config.set_synchronous_standby_names('foo') self.assertTrue(str(self.p.config.get_server_parameters(config)).startswith('{')) @patch('time.sleep', Mock()) diff --git a/tests/test_sync.py b/tests/test_sync.py new file mode 100644 index 00000000..7e6aea47 --- /dev/null +++ b/tests/test_sync.py @@ -0,0 +1,89 @@ +import os + +from mock import Mock, patch + +from patroni.dcs import Cluster, SyncState +from patroni.postgresql import Postgresql + +from . import BaseTestPostgresql, psycopg_connect + + +@patch('subprocess.call', Mock(return_value=0)) +@patch('patroni.psycopg.connect', psycopg_connect) +class TestSync(BaseTestPostgresql): + + @patch('subprocess.call', Mock(return_value=0)) + @patch('os.rename', Mock()) + @patch('patroni.postgresql.CallbackExecutor', Mock()) + @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)) + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + def setUp(self): + super(TestSync, self).setUp() + self.p.config.write_postgresql_conf() + self.s = self.p.sync_handler + + @patch.object(Postgresql, 'last_operation', Mock(return_value=1)) + def test_pick_sync_standby(self): + cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, + SyncState(0, self.me.name, self.leadermem.name), None, None, None) + + pg_stat_replication = [ + {'pid': 100, 'application_name': self.leadermem.name, 'sync_state': 'sync', 'flush_lsn': 1}, + {'pid': 101, 'application_name': self.me.name, 'sync_state': 'async', 'flush_lsn': 2}, + {'pid': 102, 'application_name': self.other.name, 'sync_state': 'async', '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=[self.leadermem.name, + 'on', pg_stat_replication]): + self.assertEqual(self.s.current_state(cluster), ([self.leadermem.name], [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], [])) + + # 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], [])) + + # 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], [])) + + # 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), ([], [])) + + def test_set_sync_standby(self): + def value_in_conf(): + with open(os.path.join(self.p.data_dir, 'postgresql.conf')) as f: + for line in f: + if line.startswith('synchronous_standby_names'): + return line.strip() + + mock_reload = self.p.reload = Mock() + self.s.set_synchronous_standby_names(['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']) + mock_reload.assert_not_called() + self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'") + + self.s.set_synchronous_standby_names(['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([]) + mock_reload.assert_called() + self.assertEqual(value_in_conf(), None)