Repository: zalando/patroni
Pull request #2067 (2021-09-24 14:35:47)

Release the leader lock when pg_controldata reports "shut down"
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.

FxKu: :+1: 

Patch:
diff --git a/patroni/ha.py b/patroni/ha.py
index 793c5d4f..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
@@ -317,7 +318,7 @@ 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, True):
+                if self.is_failover_possible(self.cluster.members):
                     logger.info("Master crashed. Failing over.")
                     self.demote('immediate')
                     return 'stopped PostgreSQL to fail over after a crash'
@@ -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',
@@ -782,6 +786,10 @@ class Ha(object):
             return False
 
         if self.cluster.failover:
+            # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race
+            if self.is_synchronous_mode() and self.cluster.failover.leader and \
+                    self.cluster.failover.candidate and not self.cluster.sync.matches(self.state_handler.name):
+                return False
             return self.manual_failover_process_no_leader()
 
         if not self.watchdog.is_healthy:
@@ -832,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
@@ -925,7 +949,7 @@ class Ha(object):
                     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
+                    if self.is_failover_possible(members, False):  # check that there are healthy members
                         ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',))
                         return ret or 'manual failover: demoting myself'
                     else:
@@ -1179,7 +1203,7 @@ class Ha(object):
         if self.has_lock() and self.update_lock():
             if self._async_executor.scheduled_action == 'doing crash recovery in a single user mode':
                 time_left = self.patroni.config['master_start_timeout'] - (time.time() - self._crash_recovery_started)
-                if time_left <= 0 and self.is_failover_possible(self.cluster.members, True):
+                if time_left <= 0 and self.is_failover_possible(self.cluster.members):
                     logger.info("Demoting self because crash recovery is taking too long")
                     self.state_handler.cancellable.cancel(True)
                     self.demote('immediate')
@@ -1286,7 +1310,7 @@ class Ha(object):
             time_left = timeout - self.state_handler.time_in_state()
 
             if time_left <= 0:
-                if self.is_failover_possible(self.cluster.members, True):
+                if self.is_failover_possible(self.cluster.members):
                     logger.info("Demoting self because master startup is taking too long")
                     self.demote('immediate')
                     return 'stopped PostgreSQL because of startup timeout'
@@ -1491,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:


Test patch:
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 0f30a110..945a083c 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -624,6 +624,11 @@ class TestHa(PostgresInit):
         # same as previous, but set the current member to nofailover. In no case it should be elected as a leader
         self.ha.patroni.nofailover = True
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')
+        # in sync mode only the sync node is allowed to take over
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'other', None))
+        self.ha.patroni.nofailover = False
+        self.ha.is_synchronous_mode = true
+        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
     def test_manual_failover_process_no_leader_in_pause(self):
         self.ha.is_paused = true
@@ -1098,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())
@@ -1185,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())