diff --git a/patroni/ha.py b/patroni/ha.py index 0ebb8508..3d6c6d1d 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -3,6 +3,7 @@ import functools import json import logging import psycopg2 +import six import sys import time import uuid @@ -687,7 +688,7 @@ class Ha(object): logger.info('Ignoring the former leader being ahead of us') return True - def is_failover_possible(self, members, check_synchronous=True): + def is_failover_possible(self, members, check_synchronous=True, cluster_lsn=None): ret = False cluster_timeline = self.cluster.timeline members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url] @@ -698,7 +699,10 @@ class Ha(object): not_allowed_reason = st.failover_limitation() if not_allowed_reason: logger.info('Member %s is %s', st.member.name, not_allowed_reason) - elif self.is_lagging(st.wal_position): + elif not isinstance(st.wal_position, six.integer_types): + logger.info('Member %s does not report wal_position', st.member.name) + elif cluster_lsn and st.wal_position < cluster_lsn or\ + not cluster_lsn and self.is_lagging(st.wal_position): logger.info('Member %s exceeds maximum replication lag', st.member.name) elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline): logger.info('Timeline %s of member %s is behind the cluster timeline %s', @@ -836,16 +840,32 @@ class Ha(object): logger.info('Demoting self (%s)', mode) self._rewind.trigger_check_diverged_lsn() + + status = {'released': False} + + def on_shutdown(checkpoint_location): + # Postmaster is still running, but pg_control already reports clean "shut down". + # It could happen if Postgres is still archiving the backlog of WAL files. + # If we know that there are replicas that received the shutdown checkpoint + # location, we can remove the leader key and allow them to start leader race. + if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location): + self.state_handler.set_role('demoted') + with self._async_executor: + self.release_leader_key_voluntarily(checkpoint_location) + status['released'] = True + self.state_handler.stop(mode_control['stop'], checkpoint=mode_control['checkpoint'], on_safepoint=self.watchdog.disable if self.watchdog.is_running else None, + on_shutdown=on_shutdown if mode_control['release'] else None, stop_timeout=self.master_stop_timeout()) self.state_handler.set_role('demoted') self.set_is_leader(False) if mode_control['release']: - checkpoint_location = self.state_handler.latest_checkpoint_location() if mode == 'graceful' else None - with self._async_executor: - self.release_leader_key_voluntarily(checkpoint_location) + if not status['released']: + checkpoint_location = self.state_handler.latest_checkpoint_location() if mode == 'graceful' else None + with self._async_executor: + self.release_leader_key_voluntarily(checkpoint_location) time.sleep(2) # Give a time to somebody to take the leader lock if mode_control['offline']: node_to_follow, leader = None, None @@ -1495,10 +1515,27 @@ class Ha(object): # This might not be the desired behavior of users, as a graceful shutdown of the host can mean lost data. # We probably need to something smarter here. disable_wd = self.watchdog.disable if self.watchdog.is_running else None + + status = {'deleted': False} + + def _on_shutdown(checkpoint_location): + if self.is_leader(): + # Postmaster is still running, but pg_control already reports clean "shut down". + # It could happen if Postgres is still archiving the backlog of WAL files. + # If we know that there are replicas that received the shutdown checkpoint + # location, we can remove the leader key and allow them to start leader race. + if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location): + self.dcs.delete_leader(checkpoint_location) + status['deleted'] = True + else: + self.dcs.write_leader_optime(checkpoint_location) + + on_shutdown = _on_shutdown if self.is_leader() else None self.while_not_sync_standby(lambda: self.state_handler.stop(checkpoint=False, on_safepoint=disable_wd, + on_shutdown=on_shutdown, stop_timeout=self.master_stop_timeout())) if not self.state_handler.is_running(): - if self.is_leader(): + if self.is_leader() and not status['deleted']: checkpoint_location = self.state_handler.latest_checkpoint_location() self.dcs.delete_leader(checkpoint_location) self.touch_member() diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 6cfd42c3..4949b456 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -598,7 +598,8 @@ class Postgresql(object): logger.exception('Exception during CHECKPOINT') return 'not accessible or not healty' - def stop(self, mode='fast', block_callbacks=False, checkpoint=None, on_safepoint=None, stop_timeout=None): + def stop(self, mode='fast', block_callbacks=False, checkpoint=None, + on_safepoint=None, on_shutdown=None, stop_timeout=None): """Stop PostgreSQL Supports a callback when a safepoint is reached. A safepoint is when no user backend can return a successful @@ -606,11 +607,12 @@ class Postgresql(object): could be added. :param on_safepoint: This callback is called when no user backends are running. + :param on_shutdown: is called when pg_controldata starts reporting `Database cluster state: shut down` """ if checkpoint is None: checkpoint = False if mode == 'immediate' else True - success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint, stop_timeout) + success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint, on_shutdown, stop_timeout) if success: # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones @@ -623,7 +625,7 @@ class Postgresql(object): self.set_state('stop failed') return success - def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint, stop_timeout): + def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint, on_shutdown, stop_timeout): postmaster = self.is_running() if not postmaster: if on_safepoint: @@ -650,6 +652,22 @@ class Postgresql(object): postmaster.wait_for_user_backends_to_close() on_safepoint() + if on_shutdown and mode in ('fast', 'smart'): + i = 0 + # Wait for pg_controldata `Database cluster state:` to change to "shut down" + while postmaster.is_running(): + data = self.controldata() + if data.get('Database cluster state', '') == 'shut down': + on_shutdown(int(self.latest_checkpoint_location())) + break + elif data.get('Database cluster state', '').startswith('shut down'): # shut down in recovery + break + elif stop_timeout and i >= stop_timeout: + stop_timeout = 0 + break + time.sleep(STOP_POLLING_INTERVAL) + i += STOP_POLLING_INTERVAL + try: postmaster.wait(timeout=stop_timeout) except TimeoutExpired: diff --git a/tests/test_ha.py b/tests/test_ha.py index 5c198e4a..945a083c 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1103,6 +1103,14 @@ class TestHa(PostgresInit): def test_shutdown(self): self.p.is_running = false self.ha.is_leader = true + + def stop(*args, **kwargs): + kwargs['on_shutdown'](123) + + self.p.stop = stop + self.ha.shutdown() + + self.ha.is_failover_possible = true self.ha.shutdown() @patch('time.sleep', Mock()) @@ -1190,3 +1198,8 @@ class TestHa(PostgresInit): self.ha.cluster.is_unlocked = false self.p.is_leader = false self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots')) + + def test_is_failover_possible(self): + self.ha.fetch_node_status = Mock(return_value=_MemberStatus(self.ha.cluster.members[0], + True, True, 0, 2, None, {}, False)) + self.assertFalse(self.ha.is_failover_possible(self.ha.cluster.members)) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 96f0b5bc..884fbbc8 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -203,6 +203,21 @@ class TestPostgresql(BaseTestPostgresql): mock_postmaster.signal_stop.side_effect = [None, True] self.assertTrue(self.p.stop(on_safepoint=mock_callback, stop_timeout=30)) + @patch('time.sleep', Mock()) + @patch.object(Postgresql, 'is_running', MockPostmaster) + @patch.object(Postgresql, '_wait_for_connection_close', Mock()) + @patch.object(Postgresql, 'latest_checkpoint_location', Mock(return_value='7')) + def test__do_stop(self): + mock_callback = Mock() + with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down'})): + self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3)) + mock_callback.assert_called() + with patch.object(Postgresql, 'controldata', + Mock(return_value={'Database cluster state': 'shut down in recovery'})): + self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3)) + with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shutting down'})): + self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3)) + def test_restart(self): self.p.start = Mock(return_value=False) self.assertFalse(self.p.restart())