Master stop timeout (#1445)

## Feature: Postgres stop timeout

Switchover/Failover operation hangs on signal_stop (or checkpoint) call when postmaster doesn't respond or  hangs for some reason(Issue described in [1371](https://github.com/zalando/patroni/issues/1371)). This is leading to service loss for an extended period of time until the hung postmaster starts responding or it is killed by some other actor.

### master_stop_timeout

The number of seconds Patroni is allowed to wait when stopping Postgres and effective only when synchronous_mode is enabled. When set to > 0 and the synchronous_mode is enabled, Patroni sends SIGKILL to the postmaster if the stop operation is running for more than the value set by master_stop_timeout. Set the value according to your durability/availability tradeoff. If the parameter is not set or set <= 0, master_stop_timeout does not apply.

docs/SETTINGS.rst

@@ -16,6 +16,7 @@ Dynamic configuration is stored in the DCS (Distributed Configuration Store) and
 -  **retry\_timeout**: timeout for DCS and PostgreSQL operation retries (in seconds). DCS or network issues shorter than this will not cause Patroni to demote the leader. Default value: 10
 -  **maximum\_lag\_on\_failover**: the maximum bytes a follower may lag to be able to participate in leader election.
 -  **master\_start\_timeout**: the amount of time a master is allowed to recover from failures before failover is triggered (in seconds). Default is 300 seconds. When set to 0 failover is done immediately after a crash is detected if possible. When using asynchronous replication a failover can cause lost transactions. Worst case failover time for master failure is: loop\_wait + master\_start\_timeout + loop\_wait, unless master\_start\_timeout is zero, in which case it's just loop\_wait. Set the value according to your durability/availability tradeoff.
+- **master\_stop\_timeout**: The number of seconds Patroni is allowed to wait when stopping Postgres and effective only when synchronous_mode is enabled. When set to > 0 and the synchronous_mode is enabled, Patroni sends SIGKILL to the postmaster if the stop operation is running for more than the value set by master_stop_timeout. Set the value according to your durability/availability tradeoff. If the parameter is not set or set <= 0, master_stop_timeout does not apply.
 -  **synchronous\_mode**: turns on synchronous replication mode. In this mode a replica will be chosen as synchronous and only the latest leader and synchronous replica are able to participate in leader election. Synchronous mode makes sure that successfully committed transactions will not be lost at failover, at the cost of losing availability for writes when Patroni cannot ensure transaction durability. See :ref:`replication modes documentation <replication_modes>` for details.
 -  **synchronous\_mode\_strict**: prevents disabling synchronous replication if no synchronous replicas are available, blocking all client writes to the master. See :ref:`replication modes documentation <replication_modes>` for details.
 -  **postgresql**:

patroni/config.py

@@ -59,6 +59,7 @@ class Config(object):
         'maximum_lag_on_failover': 1048576,
         'check_timeline': False,
         'master_start_timeout': 300,
+        'master_stop_timeout': 0,
         'synchronous_mode': False,
         'synchronous_mode_strict': False,
         'standby_cluster': {

patroni/ha.py

@@ -14,7 +14,7 @@ from patroni.exceptions import DCSError, PostgresConnectionException, PatroniExc
 from patroni.postgresql import ACTION_ON_START, ACTION_ON_ROLE_CHANGE
 from patroni.postgresql.misc import postgres_version_to_int
 from patroni.postgresql.rewind import Rewind
-from patroni.utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster
+from patroni.utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int
 from patroni.dcs import RemoteMember
 from threading import RLock
 
@@ -94,6 +94,11 @@ class Ha(object):
         else:
             return self.patroni.config.check_mode(mode)
 
+    def master_stop_timeout(self):
+        """ Master stop timeout """
+        ret = parse_int(self.patroni.config['master_stop_timeout'])
+        return ret if ret and ret > 0 and self.is_synchronous_mode() else None
+
     def is_paused(self):
         return self.check_mode('pause')
 
@@ -772,7 +777,8 @@ class Ha(object):
 
         self._rewind.trigger_check_diverged_lsn()
         self.state_handler.stop(mode_control['stop'], checkpoint=mode_control['checkpoint'],
-                                on_safepoint=self.watchdog.disable if self.watchdog.is_running else None)
+                                on_safepoint=self.watchdog.disable if self.watchdog.is_running else None,
+                                stop_timeout=self.master_stop_timeout())
         self.state_handler.set_role('demoted')
         self.set_is_leader(False)
 
@@ -1083,7 +1089,7 @@ class Ha(object):
                 return (False, 'restart failed')
 
     def _do_reinitialize(self, cluster):
-        self.state_handler.stop('immediate')
+        self.state_handler.stop('immediate', stop_timeout=self.patroni.config['retry_timeout'])
         # Commented redundant data directory cleanup here
         # self.state_handler.remove_data_directory()
 
@@ -1156,7 +1162,7 @@ class Ha(object):
     def cancel_initialization(self):
         logger.info('removing initialize key after failed attempt to bootstrap the cluster')
         self.dcs.cancel_initialization()
-        self.state_handler.stop('immediate')
+        self.state_handler.stop('immediate', stop_timeout=self.patroni.config['retry_timeout'])
         self.state_handler.move_data_directory()
         raise PatroniException('Failed to bootstrap cluster')
 
@@ -1279,7 +1285,7 @@ class Ha(object):
             # is data directory empty?
             if self.state_handler.data_directory_empty():
                 self.state_handler.set_role('uninitialized')
-                self.state_handler.stop('immediate')
+                self.state_handler.stop('immediate', stop_timeout=self.patroni.config['retry_timeout'])
                 # In case datadir went away while we were master.
                 self.watchdog.disable()
 
@@ -1373,7 +1379,8 @@ 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
-            self.while_not_sync_standby(lambda: self.state_handler.stop(checkpoint=False, on_safepoint=disable_wd))
+            self.while_not_sync_standby(lambda: self.state_handler.stop(checkpoint=False, on_safepoint=disable_wd,
+                                                                        stop_timeout=self.master_stop_timeout()))
             if not self.state_handler.is_running():
                 if self.has_lock():
                     self.dcs.delete_leader()

patroni/postgresql/__init__.py

@@ -19,6 +19,7 @@ from patroni.postgresql.slots import SlotsHandler
 from patroni.exceptions import PostgresConnectionException
 from patroni.utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty
 from threading import current_thread, Lock
+from psutil import TimeoutExpired
 
 
 logger = logging.getLogger(__name__)
@@ -462,11 +463,13 @@ class Postgresql(object):
         else:
             return None
 
-    def checkpoint(self, connect_kwargs=None):
+    def checkpoint(self, connect_kwargs=None, timeout=None):
         check_not_is_in_recovery = connect_kwargs is not None
         connect_kwargs = connect_kwargs or self.config.local_connect_kwargs
         for p in ['connect_timeout', 'options']:
             connect_kwargs.pop(p, None)
+        if timeout:
+            connect_kwargs['connect_timeout'] = timeout
         try:
             with get_connection_cursor(**connect_kwargs) as cur:
                 cur.execute("SET statement_timeout = 0")
@@ -479,7 +482,7 @@ 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):
+    def stop(self, mode='fast', block_callbacks=False, checkpoint=None, on_safepoint=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
@@ -491,7 +494,7 @@ class Postgresql(object):
         if checkpoint is None:
             checkpoint = False if mode == 'immediate' else True
 
-        success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint)
+        success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint, stop_timeout)
         if success:
             # block_callbacks is used during restart to avoid
             # running start/stop callbacks in addition to restart ones
@@ -504,7 +507,7 @@ class Postgresql(object):
             self.set_state('stop failed')
         return success
 
-    def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint):
+    def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint, stop_timeout):
         postmaster = self.is_running()
         if not postmaster:
             if on_safepoint:
@@ -512,7 +515,7 @@ class Postgresql(object):
             return True, False
 
         if checkpoint and not self.is_starting():
-            self.checkpoint()
+            self.checkpoint(timeout=stop_timeout)
 
         if not block_callbacks:
             self.set_state('stopping')
@@ -531,10 +534,28 @@ class Postgresql(object):
             postmaster.wait_for_user_backends_to_close()
             on_safepoint()
 
-        postmaster.wait()
+        try:
+            postmaster.wait(timeout=stop_timeout)
+        except TimeoutExpired:
+            logger.warning("Timeout during postmaster stop, aborting Postgres.")
+            if not self.terminate_postmaster(postmaster, mode, stop_timeout):
+                postmaster.wait()
 
         return True, True
 
+    def terminate_postmaster(self, postmaster, mode, stop_timeout):
+        if mode in ['fast', 'smart']:
+            try:
+                success = postmaster.signal_stop('immediate', self.pgcommand('pg_ctl'))
+                if success:
+                    return True
+                postmaster.wait(timeout=stop_timeout)
+                return True
+            except TimeoutExpired:
+                pass
+        logger.warning("Sending SIGKILL to Postmaster and its children")
+        return postmaster.signal_kill()
+
     def terminate_starting_postmaster(self, postmaster):
         """Terminates a postmaster that has not yet opened ports or possibly even written a pid file. Blocks
         until the process goes away."""

patroni/postgresql/postmaster.py

@@ -105,6 +105,42 @@ class PostmasterProcess(psutil.Process):
         except psutil.NoSuchProcess:
             return None
 
+    def signal_kill(self):
+        """to suspend and kill postmaster and all children
+
+        :returns True if postmaster and children are killed, False if error
+        """
+        try:
+            self.suspend()
+        except psutil.NoSuchProcess:
+            return True
+        except psutil.Error as e:
+            logger.warning('Failed to suspend postmaster: %s', e)
+
+        try:
+            children = self.children(recursive=True)
+        except psutil.NoSuchProcess:
+            return True
+        except psutil.Error as e:
+            logger.warning('Failed to get a list of postmaster children: %s', e)
+            children = []
+
+        try:
+            self.kill()
+        except psutil.NoSuchProcess:
+            return True
+        except psutil.Error as e:
+            logger.warning('Could not kill postmaster: %s', e)
+            return False
+
+        for child in children:
+            try:
+                child.kill()
+            except psutil.Error:
+                pass
+        psutil.wait_procs(children + [self])
+        return True
+
     def signal_stop(self, mode, pg_ctl='pg_ctl'):
         """Signal postmaster process to stop
 

tests/__init__.py

@@ -66,6 +66,7 @@ class MockPostmaster(object):
         self.wait_for_user_backends_to_close = Mock()
         self.signal_stop = Mock(return_value=None)
         self.wait = Mock()
+        self.signal_kill = Mock(return_value=False)
 
 
 class MockCursor(object):

tests/test_ha.py

@@ -817,6 +817,17 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'stopped PostgreSQL to fail over after a crash')
         demote.assert_called_once()
 
+    def test_master_stop_timeout(self):
+        self.assertEqual(self.ha.master_stop_timeout(), None)
+        self.ha.patroni.config.set_dynamic_configuration({'master_stop_timeout': 30})
+        with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
+            self.assertEqual(self.ha.master_stop_timeout(), 30)
+        self.ha.patroni.config.set_dynamic_configuration({'master_stop_timeout': 30})
+        with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=False)):
+            self.assertEqual(self.ha.master_stop_timeout(), None)
+            self.ha.patroni.config.set_dynamic_configuration({'master_stop_timeout': None})
+            self.assertEqual(self.ha.master_stop_timeout(), None)
+
     @patch('patroni.postgresql.Postgresql.follow')
     def test_demote_immediate(self, follow):
         self.ha.has_lock = true

tests/test_postgresql.py

@@ -1,5 +1,6 @@
 import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
+import psutil
 import psycopg2
 import re
 import subprocess
@@ -177,6 +178,17 @@ class TestPostgresql(BaseTestPostgresql):
         mock_callback.assert_called()
         mock_postmaster.signal_stop.assert_called()
 
+        # Timed out waiting for fast shutdown triggers immediate shutdown
+        mock_postmaster.wait.side_effect = [psutil.TimeoutExpired(30), psutil.TimeoutExpired(30), Mock()]
+        mock_callback.reset_mock()
+        self.assertTrue(self.p.stop(on_safepoint=mock_callback, stop_timeout=30))
+        mock_callback.assert_called()
+        mock_postmaster.signal_stop.assert_called()
+
+        # Immediate shutdown succeeded
+        mock_postmaster.wait.side_effect = [psutil.TimeoutExpired(30), Mock()]
+        self.assertTrue(self.p.stop(on_safepoint=mock_callback, stop_timeout=30))
+
         # Stop signal failed
         mock_postmaster.signal_stop.return_value = False
         self.assertFalse(self.p.stop())
@@ -187,6 +199,11 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertTrue(self.p.stop(on_safepoint=mock_callback))
         mock_callback.assert_called()
 
+        # Fast shutdown is timed out but when immediate postmaster is already gone
+        mock_postmaster.wait.side_effect = [psutil.TimeoutExpired(30), Mock()]
+        mock_postmaster.signal_stop.side_effect = [None, True]
+        self.assertTrue(self.p.stop(on_safepoint=mock_callback, stop_timeout=30))
+
     def test_restart(self):
         self.p.start = Mock(return_value=False)
         self.assertFalse(self.p.restart())
@@ -203,7 +220,7 @@ class TestPostgresql(BaseTestPostgresql):
             self.assertEqual(self.p.checkpoint({'user': 'postgres'}), 'is_in_recovery=true')
         with patch.object(MockCursor, 'execute', Mock(return_value=None)):
             self.assertIsNone(self.p.checkpoint())
-        self.assertEqual(self.p.checkpoint(), 'not accessible or not healty')
+        self.assertEqual(self.p.checkpoint(timeout=10), 'not accessible or not healty')
 
     @patch('patroni.postgresql.config.mtime', mock_mtime)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')

tests/test_postmaster.py

@@ -63,6 +63,37 @@ class TestPostmasterProcess(unittest.TestCase):
         mock_init.side_effect = None
         self.assertNotEqual(PostmasterProcess.from_pid(123), None)
 
+    @patch('psutil.Process.__init__', Mock())
+    @patch('psutil.wait_procs', Mock())
+    @patch('psutil.Process.suspend')
+    @patch('psutil.Process.children')
+    @patch('psutil.Process.kill')
+    def test_signal_kill(self, mock_kill, mock_children, mock_suspend):
+        proc = PostmasterProcess(123)
+
+        # all processes successfully stopped
+        mock_children.return_value = [Mock()]
+        mock_children.return_value[0].kill.side_effect = psutil.Error
+        self.assertTrue(proc.signal_kill())
+
+        # postmaster has gone before suspend
+        mock_suspend.side_effect = psutil.NoSuchProcess(123)
+        self.assertTrue(proc.signal_kill())
+
+        # postmaster has gone before we got a list of children
+        mock_suspend.side_effect = psutil.Error()
+        mock_children.side_effect = psutil.NoSuchProcess(123)
+        self.assertTrue(proc.signal_kill())
+
+        # postmaster has gone after we got a list of children
+        mock_children.side_effect = psutil.Error()
+        mock_kill.side_effect = psutil.NoSuchProcess(123)
+        self.assertTrue(proc.signal_kill())
+
+        # failed to kill postmaster
+        mock_kill.side_effect = psutil.AccessDenied(123)
+        self.assertFalse(proc.signal_kill())
+
     @patch('psutil.Process.__init__', Mock())
     @patch('psutil.Process.send_signal')
     @patch('psutil.Process.pid', Mock(return_value=123))