Merge pull request #228 from zalando/bugfix/always-expose-role

bugfix: api must report role=master during pg_ctl stop

docs/SETTINGS.rst

@@ -69,6 +69,7 @@ PostgreSQL
 -  **pgpass**: path to the `.pgpass <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ 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

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)

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()

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")

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)):