Apply master_start_timeout when executing crash recovery (#1720)

It is not very common, but the master Postgres might "crash" due to different reasons, like OOM, or out of disk space. Of course, there are chances that the current node holds some unreplicated data and therefore Patroni by default prefers to start Postgres on the leader node rather than doing a failover.

In order to be on the safe side Patroni always starts Postgres in recovery no matter whether the current node owns the leader lock or not. If the Postgres wasn't shut down cleanly, starting in recovery might fail, therefore in some cases as a workaround Patroni is executing a crash recovery by starting the postgres up in the single-user mode.

A few times we end up in the situation:
1. Master postgres crashed due to the out of disk space
2. Patroni starts crash recovery in a single-user mode
3. While doing crash-recovery Patroni keeps updating the leader lock

It makes Patroni stuck on step 3 and the manual intervention is required for recovering the cluster.

Patroni already has the option `master_start_timeout`, which controls for how long we let postgres stay in the `starting` state and after that Patroni might decide to release the leader lock if there are healthy replicas available which could take it over.

This PR makes the `master_start_timeout` option also work for crash recovery.

patroni/ha.py

@@ -72,6 +72,7 @@ class Ha(object):
         self.recovering = False
         self._async_response = CriticalTask()
         self._crash_recovery_executed = False
+        self._crash_recovery_started = None
         self._start_timeout = None
         self._async_executor = AsyncExecutor(self.state_handler.cancellable, self.wakeup)
         self.watchdog = patroni.watchdog
@@ -333,6 +334,7 @@ class Ha(object):
                 and not self._crash_recovery_executed and \
                 (self.cluster.is_unlocked() or self._rewind.can_rewind):
             self._crash_recovery_executed = True
+            self._crash_recovery_started = time.time()
             msg = 'doing crash recovery in a single user mode'
             return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
 
@@ -1171,6 +1173,14 @@ class Ha(object):
         Figure out what to do with the task AsyncExecutor is performing.
         """
         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):
+                    logger.info("Demoting self because crash recovery is taking too long")
+                    self.state_handler.cancellable.cancel(True)
+                    self.demote('immediate')
+                    return 'terminated crash recovery because of startup timeout'
+
             return 'updated leader lock during ' + self._async_executor.scheduled_action
         elif not self.state_handler.bootstrapping:
             # Don't have lock, make sure we are not promoting or starting up a master in the background

patroni/postgresql/cancellable.py

@@ -114,7 +114,7 @@ class CancellableSubprocess(CancellableExecutor):
         with self._lock:
             return self._is_cancelled
 
-    def cancel(self):
+    def cancel(self, kill=False):
         with self._lock:
             self._is_cancelled = True
             if self._process is None or not self._process.is_running():
@@ -127,5 +127,7 @@ class CancellableSubprocess(CancellableExecutor):
             with self._lock:
                 if self._process is None or not self._process.is_running():
                     return
+            if kill:
+                break
 
         self._kill_process()

patroni/postgresql/rewind.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import shlex
 import six
 import subprocess
 
@@ -273,7 +274,7 @@ class Rewind(object):
             i += 1
 
         logger.info('Trying to fetch the missing wal: %s', cmd)
-        return self._postgresql.cancellable.call(cmd, shell=True) == 0
+        return self._postgresql.cancellable.call(shlex.split(cmd)) == 0
 
     def _find_missing_wal(self, data):
         # could not open file "$PGDATA/pg_wal/0000000A00006AA100000068": No such file or directory
@@ -427,4 +428,4 @@ class Rewind(object):
         opts = self.read_postmaster_opts()
         opts.update({'archive_mode': 'on', 'archive_command': 'false'})
         self._postgresql.config.remove_recovery_conf()
-        return self.single_user_mode(options=opts) == 0 or None
+        return self.single_user_mode(communicate={}, options=opts) == 0 or None

tests/test_ha.py

@@ -261,9 +261,17 @@ class TestHa(PostgresInit):
 
     @patch.object(Rewind, 'ensure_clean_shutdown', Mock())
     def test_crash_recovery(self):
+        self.ha.has_lock = true
         self.p.is_running = false
         self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
         self.assertEqual(self.ha.run_cycle(), 'doing crash recovery in a single user mode')
+        with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)),\
+                patch.object(Ha, 'check_timeline', Mock(return_value=False)):
+            self.ha._async_executor.schedule('doing crash recovery in a single user mode')
+            self.ha.state_handler.cancellable._process = Mock()
+            self.ha._crash_recovery_started -= 600
+            self.ha.patroni.config.set_dynamic_configuration({'maximum_lag_on_failover': 10})
+            self.assertEqual(self.ha.run_cycle(), 'terminated crash recovery because of startup timeout')
 
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))