From 0318749b5624911fb34817191447785cab53f1f9 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 28 Jun 2016 12:02:38 +0200 Subject: [PATCH] bugfix: api must report role=master during pg_ctl stop In addition for that make pg_ctl --timeout option configurable. If the stop or start didn't succeeded during given timeout when demoting master, role will be forcibly changed to 'unknown' and all needed callbacks executed. --- docs/SETTINGS.rst | 1 + patroni/api.py | 2 +- patroni/ha.py | 13 +++---------- patroni/postgresql.py | 33 +++++++++++++++++++++++++-------- tests/test_postgresql.py | 9 +++++++-- 5 files changed, 37 insertions(+), 21 deletions(-) diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 88be5031..1b548057 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -69,6 +69,7 @@ PostgreSQL - **pgpass**: path to the `.pgpass `__ password file. Patroni creates this file before executing pg\_basebackup and under some other circumstances. The location must be writable by Patroni. - **recovery\_conf**: additional configuration settings written to recovery.conf when configuring follower. - **parameters**: list of configuration settings for Postgres. Many of these are required for replication to work. +- **pg\_ctl\_timeout**: How long should pg_ctl wait when doing ``start``, ``stop`` or ``restart``. Default value is 60 seconds. - **replica\_method** for each create_replica_method other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value". REST API diff --git a/patroni/api.py b/patroni/api.py index b87711b2..5dfa2942 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -331,7 +331,7 @@ class RestApiHandler(BaseHTTPRequestHandler): if state == 'running': logger.exception('get_postgresql_status') state = 'unknown' - return {'state': state} + return {'state': state, 'role': self.server.patroni.postgresql.role} def log_message(self, fmt, *args): logger.debug("API thread: %s - - [%s] %s", self.client_address[0], self.log_date_time_string(), fmt % args) diff --git a/patroni/ha.py b/patroni/ha.py index 0352b270..8a2f9405 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -165,9 +165,8 @@ class Ha(object): logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content) json = response.json() is_master = json['role'] == 'master' - xlog_location = json['xlog']['location' if is_master else 'replayed_location'] - tags = json.get('tags', dict()) - return (member, True, not is_master, xlog_location, tags) + xlog_location = None if is_master else json['xlog']['replayed_location'] + return (member, True, not is_master, xlog_location, json.get('tags', {})) except: logging.exception('request failed: GET %s', member.api_url) return (member, False, None, 0, {}) @@ -182,12 +181,6 @@ class Ha(object): def _is_healthiest_node(self, members, check_replication_lag=True): """This method tries to determine whether I am healthy enough to became a new leader candidate or not.""" - if self.state_handler.is_leader(): - return True - - if self.patroni.nofailover is True: - return False - if check_replication_lag and not self.state_handler.check_replication_lag(self.cluster.last_leader_operation): return False # Too far behind last reported xlog location on master @@ -259,7 +252,6 @@ class Ha(object): return self._is_healthiest_node(members, check_replication_lag=False) def is_healthiest_node(self): - if self.state_handler.is_leader(): # leader is always the healthiest return True @@ -276,6 +268,7 @@ class Ha(object): def demote(self, delete_leader=True): if delete_leader: self.state_handler.stop() + self.state_handler.set_role('unknown') self.dcs.delete_leader() self.touch_member() self.dcs.reset_cluster() diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 4016c6d7..25bb6f84 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -106,8 +106,6 @@ class Postgresql(object): self._trigger_file = config.get('recovery_conf', {}).get('trigger_file') or 'promote' self._trigger_file = os.path.abspath(os.path.join(self._data_dir, self._trigger_file)) - self._pg_ctl = ['pg_ctl', '-w', '-D', self._data_dir] - self._connection = None self._cursor_holder = None self._sysid = None @@ -152,6 +150,22 @@ class Postgresql(object): self.connection_string = 'postgres://{connect_address}/{database}'.format( connect_address=self._connect_address or self._local_address, database=self._database) + def pg_ctl(self, cmd, *args, **kwargs): + """Builds and executes pg_ctl command + + :returns: `!True` when return_code == 0, otherwise `!False`""" + + pg_ctl = ['pg_ctl', cmd] + if cmd in ('start', 'stop', 'restart'): + pg_ctl += ['-w'] + timeout = self.config.get('pg_ctl_timeout') + if timeout: + try: + pg_ctl += ['-t', str(int(timeout))] + except Exception: + logger.error('Bad value of pg_ctl_timeout: %s', timeout) + return subprocess.call(pg_ctl + ['-D', self._data_dir] + list(args), **kwargs) == 0 + def reload_config(self, config): server_parameters = self.get_server_parameters(config) @@ -339,8 +353,9 @@ class Postgresql(object): os.write(fd, self._superuser['password'].encode('utf-8')) os.close(fd) options.append('--pwfile={0}'.format(pwfile)) + options = ['-o', ' '.join(options)] if options else [] - ret = subprocess.call(self._pg_ctl + ['initdb'] + (['-o', ' '.join(options)] if options else [])) == 0 + ret = self.pg_ctl('initdb', *options) if pwfile: os.remove(pwfile) if ret: @@ -508,7 +523,7 @@ class Postgresql(object): options = ' '.join("--{0}='{1}'".format(p, self._server_parameters[p]) for p, v in self.CMDLINE_OPTIONS.items() if self._major_version >= v[2]) - ret = subprocess.call(self._pg_ctl + ['start', '-o', options], env=env, preexec_fn=os.setsid) == 0 + ret = self.pg_ctl('start', '-o', options, env=env, preexec_fn=os.setsid) self._pending_restart = False self.set_state('running' if ret else 'start failed') @@ -552,7 +567,7 @@ class Postgresql(object): if not block_callbacks: self.set_state('stopping') - ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0 + ret = self.pg_ctl('stop', '-m', mode) # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones if not ret: @@ -563,7 +578,7 @@ class Postgresql(object): return ret def reload(self): - ret = subprocess.call(self._pg_ctl + ['reload']) == 0 + ret = self.pg_ctl('reload') if ret: self.call_nowait(ACTION_ON_RELOAD) return ret @@ -716,6 +731,7 @@ class Postgresql(object): if leader and leader.name != self.name and need_rewind: # we have a leader and need to rewind if self.is_running(): self.stop() + self.set_role('unknown') # at present, pg_rewind only runs when the cluster is shut down cleanly # and not shutdown in recovery. We have to remove the recovery.conf if present # and start/shutdown in a single user mode to emulate this. @@ -741,7 +757,8 @@ class Postgresql(object): else: # do not rewind until the leader becomes available self.write_recovery_conf(member) ret = self.restart() - if change_role and ret: + self.set_role('replica') + if change_role: self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret @@ -770,7 +787,7 @@ class Postgresql(object): def promote(self): if self.role == 'master': return True - ret = subprocess.call(self._pg_ctl + ['promote']) == 0 + ret = self.pg_ctl('promote') if ret: self.set_role('master') logger.info("cleared rewind flag after becoming the leader") diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 42dcab0a..6e0c563f 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -171,7 +171,7 @@ class TestPostgresql(unittest.TestCase): 'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432', 'authentication': {'superuser': {'username': 'test', 'password': 'test'}, 'replication': {'username': 'replicator', 'password': 'rep-pass'}}, - 'use_pg_rewind': True, + 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'parameters': self._PARAMETERS, 'recovery_conf': {'foo': 'bar'}, 'callbacks': {'on_start': 'true', 'on_stop': 'true', @@ -244,18 +244,23 @@ class TestPostgresql(unittest.TestCase): @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) def test_follow(self, mock_pg_rewind): - self.p.follow(None, None) + with patch('patroni.postgresql.Postgresql.restart', Mock(return_value=False)): + self.p.follow(None, None) + self.p.set_role('master') self.p.follow(self.leader, self.leader) self.p.follow(Leader(-1, 28, self.other), self.leader) self.p.rewind = mock_pg_rewind self.p.follow(self.leader, self.leader) + self.p.set_role('master') with mock.patch('os.path.islink', MagicMock(return_value=True)): with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): with mock.patch('os.unlink', MagicMock(return_value=True)): self.p.follow(self.leader, self.leader, recovery=True) + self.p.set_role('master') with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): self.p.rewind.return_value = True self.p.follow(self.leader, self.leader, recovery=True) + self.p.set_role('master') self.p.rewind.return_value = False self.p.follow(self.leader, self.leader, recovery=True) with mock.patch('patroni.postgresql.Postgresql.check_recovery_conf', MagicMock(return_value=True)):