Release the leader lock when pg_controldata reports "shut down" (#2067)

Due to different reasons, it could happen that WAL archiving on the primary stuck or significantly delayed. If we try to do a switchover or shut it down, the shutdown will take forever and will not finish until the whole backlog of WALs is processed.
In the meantime, Patroni keeps updating the leader lock, which prevents other nodes from starting the leader race even if it is known that they received/applied all changes.

The `Database cluster state:` is changed to `"shut down"` after:
- all data is fsynced to disk and the latest checkpoint is written to WAL
- all streaming replicas confirmed that they received all changes (including the latest checkpoint)
- at the same time, the archiver process continues to do its job and the postmaster process is still running.

In order to solve this problem and make the switchover more reliable/fast in a case when `archive_command` is slow/failing, Patroni will remove the leader key immediately after `pg_controldata` started reporting PGDATA as `"shut down"` cleanly and it verified that there is at least one replica that received all changes. If there are no replicas that fulfill the condition the leader key isn't removed and the old behavior is retained, i.e. Patroni will keep updating it.

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

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:

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

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