From 23152a7fc4735ebdd7093b11525df150a3825d10 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 24 Aug 2017 07:55:02 +0200 Subject: [PATCH] synchronous_standby_names must be quoted with quote_ident (#505) in addition to that implement additional checks around manual failover and recover when synchronous_mode is enabled * Comparison must be case insensitive --- patroni/api.py | 6 ++++++ patroni/dcs/__init__.py | 6 ++++++ patroni/ha.py | 27 ++++++++++++++++++++------- patroni/postgresql.py | 25 ++++++++++++++++++++----- patroni/utils.py | 6 ------ tests/test_api.py | 7 +++++-- tests/test_ha.py | 17 +++++++++++++++-- 7 files changed, 72 insertions(+), 22 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 3dc7749a..ee8ff2d5 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -293,9 +293,15 @@ class RestApiHandler(BaseHTTPRequestHandler): if leader and (not cluster.leader or cluster.leader.name != leader): return 'leader name does not match' if candidate: + if cluster.is_synchronous_mode() and cluster.sync.sync_standby != candidate: + return 'candidate name does not match with sync_standby' members = [m for m in cluster.members if m.name == candidate] if not members: return 'candidate does not exists' + elif cluster.is_synchronous_mode(): + members = [m for m in cluster.members if m.name == cluster.sync.sync_standby] + if not members: + return 'failover is not possible: can not find sync_standby' else: members = [m for m in cluster.members if m.name != cluster.leader.name and m.api_url] if not members: diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index df8901a2..7afbc995 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -320,6 +320,12 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat def is_paused(self): return self.config and self.config.data.get('pause', False) or False + def is_synchronous_mode(self): + return bool(self.config and self.config.data.get('synchronous_mode')) + + def is_synchronous_mode_strict(self): + return bool(self.config and self.config.data.get('synchronous_mode_strict')) + @six.add_metaclass(abc.ABCMeta) class AbstractDCS(object): diff --git a/patroni/ha.py b/patroni/ha.py index f3134a58..3f342c62 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -184,7 +184,10 @@ class Ha(object): if timeout == 0: # We are requested to prefer failing over to restarting master. But see first if there # is anyone to fail over to. - if self.is_failover_possible(self.cluster.members): + members = self.cluster.members + if self.is_synchronous_mode(): + members = [m for m in members if self.cluster.sync.matches(m.name)] + if self.is_failover_possible(members): logger.info("Master crashed. Failing over.") self.demote('immediate') return 'stopped PostgreSQL to fail over after a crash' @@ -249,10 +252,10 @@ class Ha(object): return follow_reason def is_synchronous_mode(self): - return bool(self.cluster and self.cluster.config and self.cluster.config.data.get('synchronous_mode')) + return bool(self.cluster and self.cluster.is_synchronous_mode()) def is_synchronous_mode_strict(self): - return bool(self.cluster and self.cluster.config and self.cluster.config.data.get('synchronous_mode_strict')) + return bool(self.cluster and self.cluster.is_synchronous_mode_strict()) def process_sync_replication(self): """Process synchronous standby beahvior. @@ -548,8 +551,8 @@ class Ha(object): self.state_handler.set_role('demoted') if mode_control['release']: - self.release_leader_key_voluntarily() - time.sleep(2) # Give a time to somebody to take the leader lock + self.release_leader_key_voluntarily() + time.sleep(2) # Give a time to somebody to take the leader lock if mode_control['offline']: node_to_follow, leader = None, None else: @@ -563,6 +566,8 @@ class Ha(object): self._async_executor.schedule('starting after demotion') self._async_executor.run_async(self.state_handler.follow, (node_to_follow,)) else: + if self.is_synchronous_mode(): + self.state_handler.set_synchronous_standby(None) if self.state_handler.rewind_needed_and_possible(leader): return False # do not start postgres, but run pg_rewind on the next iteration self.state_handler.follow(node_to_follow) @@ -621,8 +626,16 @@ class Ha(object): if not failover.candidate and self.is_paused(): logger.warning('Failover is possible only to a specific candidate in a paused state') else: - members = [m for m in self.cluster.members - if not failover.candidate or m.name == failover.candidate] + if self.is_synchronous_mode(): + if failover.candidate and not self.cluster.sync.matches(failover.candidate): + logger.warning('Failover candidate=%s does not match with sync_standby=%s', + failover.candidate, self.cluster.sync.sync_standby) + members = [] + else: + members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)] + else: + members = [m for m in self.cluster.members + if not failover.candidate or m.name == failover.candidate] if self.is_failover_possible(members): # check that there are healthy members self._async_executor.schedule('manual failover: demote') self._async_executor.run_async(self.demote, ('graceful',)) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index a81b9472..356def42 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -17,7 +17,7 @@ from contextlib import contextmanager from patroni import call_self from patroni.callback_executor import CallbackExecutor from patroni.exceptions import PostgresConnectionException -from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop, null_context +from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop from six import string_types from six.moves.urllib.parse import quote_plus from threading import current_thread, Lock @@ -42,6 +42,12 @@ STOP_SIGNALS = { } STOP_POLLING_INTERVAL = 1 REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5}) +sync_standby_name_re = re.compile('^[A-Za-z_][A-Za-z_0-9\$]*$') + + +def quote_ident(value): + """Very simplified version of quote_ident""" + return value if sync_standby_name_re.match(value) else '"' + value + '"' def slot_name_from_member_name(member_name): @@ -60,6 +66,11 @@ def slot_name_from_member_name(member_name): return slot_name[0:63] +@contextmanager +def null_context(): + yield + + class Postgresql(object): # List of parameters which must be always passed to postmaster as command line options @@ -1652,12 +1663,13 @@ $$""".format(name, ' '.join(options)), name, password, password) :returns tuple of candidate name or None, and bool showing if the member is the active synchronous standby. """ current = cluster.sync.sync_standby - members = {m.name: m for m in cluster.members} + current = current.lower() if current else current + members = {m.name.lower(): m for m in cluster.members} candidates = [] # Pick candidates based on who has flushed WAL farthest. # TODO: for synchronous_commit = remote_write we actually want to order on write_location for app_name, state, sync_state in self.query( - """SELECT application_name, state, sync_state + """SELECT LOWER(application_name), state, sync_state FROM pg_stat_replication ORDER BY flush_{0} DESC""".format(self.lsn_name)): member = members.get(app_name) @@ -1677,14 +1689,17 @@ $$""".format(name, ' '.join(options)), name, password, password) def set_synchronous_standby(self, name): """Sets a node to be synchronous standby and if changed does a reload for PostgreSQL.""" + if name and name != '*': + name = quote_ident(name) if name != self._synchronous_standby_names: if name is None: self._server_parameters.pop('synchronous_standby_names', None) else: self._server_parameters['synchronous_standby_names'] = name self._synchronous_standby_names = name - self._write_postgresql_conf() - self.reload() + if self.state == 'running': + self._write_postgresql_conf() + self.reload() @staticmethod def postgres_version_to_int(pg_version): diff --git a/patroni/utils.py b/patroni/utils.py index 890861f8..3ebd1f7a 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -1,4 +1,3 @@ -import contextlib import random import time import re @@ -281,8 +280,3 @@ def polling_loop(timeout, interval=1): yield iteration iteration += 1 time.sleep(interval) - - -@contextlib.contextmanager -def null_context(): - yield diff --git a/tests/test_api.py b/tests/test_api.py index f5fd0bf6..653958d8 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -281,6 +281,7 @@ class TestRestApiHandler(unittest.TestCase): def test_do_POST_failover(self, dcs): dcs.loop_wait = 10 cluster = dcs.get_cluster.return_value + cluster.is_synchronous_mode.return_value = False post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: ' @@ -292,14 +293,16 @@ class TestRestApiHandler(unittest.TestCase): cluster.leader.name = 'postgresql1' MockRestApiServer(RestApiHandler, request) - MockRestApiServer(RestApiHandler, post + '25\n\n{"leader": "postgresql1"}') + for cluster.is_synchronous_mode.return_value in (True, False): + MockRestApiServer(RestApiHandler, post + '25\n\n{"leader": "postgresql1"}') cluster.leader.name = 'postgresql2' request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}' MockRestApiServer(RestApiHandler, request) cluster.leader.name = 'postgresql1' - MockRestApiServer(RestApiHandler, request) + for cluster.is_synchronous_mode.return_value in (True, False): + MockRestApiServer(RestApiHandler, request) cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}), Member(0, 'postgresql2', 30, {'api_url': 'http'})] diff --git a/tests/test_ha.py b/tests/test_ha.py index 49772774..645886f5 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -158,7 +158,6 @@ class TestHa(unittest.TestCase): self.ha.old_cluster = self.e.get_cluster() self.ha.cluster = get_cluster_not_initialized_without_leader() self.ha.load_cluster_from_dcs = Mock() - self.ha.is_synchronous_mode = false def test_update_lock(self): self.p.last_operation = Mock(side_effect=PostgresConnectionException('')) @@ -438,6 +437,19 @@ class TestHa(unittest.TestCase): self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None)) self.assertEquals('PAUSE: no action. i am the leader with the lock', self.ha.run_cycle()) + @patch('requests.get', requests_get) + def test_manual_failover_from_leader_in_synchronous_mode(self): + self.p.is_leader = true + self.ha.has_lock = true + self.ha.is_synchronous_mode = true + self.ha.is_failover_possible = false + self.ha.process_sync_replication = Mock() + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, None)) + self.assertEquals('no action. i am the leader with the lock', self.ha.run_cycle()) + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, 'a')) + self.ha.is_failover_possible = true + self.assertEquals('manual failover: demoting myself', self.ha.run_cycle()) + @patch('requests.get', requests_get) def test_manual_failover_process_no_leader(self): self.p.is_leader = false @@ -635,7 +647,8 @@ class TestHa(unittest.TestCase): @patch('patroni.ha.Ha.demote') def test_failover_immediately_on_zero_master_start_timeout(self, demote): self.p.is_running = false - self.ha.cluster = get_cluster_initialized_with_leader() + self.ha.cluster = get_cluster_initialized_with_leader(sync=(self.p.name, 'other')) + self.ha.cluster.config.data['synchronous_mode'] = True self.ha.patroni.config.set_dynamic_configuration({'master_start_timeout': 0}) self.ha.has_lock = true self.ha.update_lock = true