diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 481f0b7a..6a9b6057 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -1,6 +1,8 @@ import logging import os +import re import shlex +import shutil import six import subprocess @@ -277,28 +279,37 @@ class Rewind(object): def checkpoint_after_promote(self): return self._state == REWIND_STATUS.CHECKPOINT - def _fetch_missing_wal(self, restore_command, wal_filename): + def _buid_archiver_command(self, command, wal_filename): + """Replace placeholders in the given archiver command's template. + Applicable for archive_command and restore_command. + Can also be used for archive_cleanup_command and recovery_end_command, + however %r value is always set to 000000010000000000000001.""" cmd = '' - length = len(restore_command) + length = len(command) i = 0 while i < length: - if restore_command[i] == '%' and i + 1 < length: + if command[i] == '%' and i + 1 < length: i += 1 - if restore_command[i] == 'p': + if command[i] == 'p': cmd += os.path.join(self._postgresql.wal_dir, wal_filename) - elif restore_command[i] == 'f': + elif command[i] == 'f': cmd += wal_filename - elif restore_command[i] == 'r': + elif command[i] == 'r': cmd += '000000010000000000000001' - elif restore_command[i] == '%': + elif command[i] == '%': cmd += '%' else: cmd += '%' i -= 1 else: - cmd += restore_command[i] + cmd += command[i] i += 1 + return cmd + + def _fetch_missing_wal(self, restore_command, wal_filename): + cmd = self._buid_archiver_command(restore_command, wal_filename) + logger.info('Trying to fetch the missing wal: %s', cmd) return self._postgresql.cancellable.call(shlex.split(cmd)) == 0 @@ -315,6 +326,41 @@ class Rewind(object): if waldir.endswith('/pg_' + self._postgresql.wal_name) and len(wal_filename) == 24: return wal_filename + def _archive_ready_wals(self): + """Try to archive WALs that have .ready files just in case + archive_mode was not set to 'always' before promote, while + after it the WALs were recycled on the promoted replica. + With this we prevent the entire loss of such WALs and the + consequent old leader's start failure.""" + archive_mode = self._postgresql.get_guc_value('archive_mode') + archive_cmd = self._postgresql.get_guc_value('archive_command') + if archive_mode not in ('on', 'always') or not archive_cmd: + return + + walseg_regex = re.compile(r'^[0-9A-F]{24}(\.partial){0,1}\.ready$') + status_dir = os.path.join(self._postgresql.wal_dir, 'archive_status') + try: + wals_to_archive = [f[:-6] for f in os.listdir(status_dir) if walseg_regex.match(f)] + except OSError as e: + return logger.error('Unable to list %s: %r', status_dir, e) + + # skip fsync, as postgres --single or pg_rewind will anyway run it + for wal in sorted(wals_to_archive): + if os.path.isfile(os.path.join(self._postgresql.wal_dir, wal)): + cmd = self._buid_archiver_command(archive_cmd, wal) + # it is the author of archive_command, who is responsible + # for not overriding the WALs already present in archive + logger.info('Trying to archive %s: %s', wal, cmd) + if self._postgresql.cancellable.call(shlex.split(cmd)) == 0: + old_name = os.path.join(status_dir, wal + '.ready') + new_name = os.path.join(status_dir, wal + '.done') + try: + shutil.move(old_name, new_name) + except Exception as e: + logger.error('Unable to rename %s to %s: %r', old_name, new_name, e) + else: + logger.info('Failed to archive WAL segment %s', wal) + def pg_rewind(self, r): # prepare pg_rewind connection env = self._postgresql.config.write_pgpass(r) @@ -367,6 +413,8 @@ class Rewind(object): if self._postgresql.is_running() and not self._postgresql.stop(checkpoint=False): return logger.warning('Can not run pg_rewind because postgres is still running') + self._archive_ready_wals() + # prepare pg_rewind connection r = self._conn_kwargs(leader, self._postgresql.config.rewind_credentials) @@ -465,6 +513,7 @@ class Rewind(object): logger.exception('Unable to list %s', status_dir) def ensure_clean_shutdown(self): + self._archive_ready_wals() self.cleanup_archive_status() # Start in a single user mode and stop to produce a clean shutdown diff --git a/tests/test_rewind.py b/tests/test_rewind.py index 2d510b33..b9398205 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -218,6 +218,62 @@ class TestRewind(BaseTestPostgresql): self.r.cleanup_archive_status() self.r.cleanup_archive_status() + @patch('os.path.isfile', Mock(return_value=True)) + @patch('shutil.move', Mock(side_effect=OSError)) + @patch('patroni.postgresql.rewind.logger.info') + def test_archive_ready_wals(self, mock_logger_info): + with patch('os.listdir', Mock(side_effect=OSError)), \ + patch.object(Postgresql, 'get_guc_value', Mock(side_effect=['on', 'command %f'])): + self.r._archive_ready_wals() + mock_logger_info.assert_not_called() + + # each assert_not_called() calls get_guc_value('archive_mode') + get_guc_value('archive_command') + get_guc_value_res = [ + '', 'command %f', + 'on', '', + ] + with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=get_guc_value_res)): + for _ in range(len(get_guc_value_res)//2): + self.r._archive_ready_wals() + mock_logger_info.assert_not_called() + + with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])): + # successful archive_command call + with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)): + get_guc_value_res = [ + 'on', 'command %f', + 'always', 'command %f', + ] + with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=get_guc_value_res)): + for _ in range(len(get_guc_value_res)//2): + self.r._archive_ready_wals() + mock_logger_info.assert_called_once() + self.assertEqual(('Trying to archive %s: %s', + '000000000000000000000000', 'command 000000000000000000000000'), + mock_logger_info.call_args[0]) + mock_logger_info.reset_mock() + + # failed archive_command call + with patch.object(CancellableSubprocess, 'call', Mock(return_value=1)): + with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=['on', 'command %f'])): + self.r._archive_ready_wals() + self.assertEqual(('Trying to archive %s: %s', + '000000000000000000000000', 'command 000000000000000000000000'), + mock_logger_info.call_args_list[0][0]) + self.assertEqual(('Failed to archive WAL segment %s', '000000000000000000000000'), + mock_logger_info.call_args_list[1][0]) + mock_logger_info.reset_mock() + + wal_files_to_skip = [ + '000000000000000000000000.done', + '000000000000000000000001.partial.done', + '002.ready', + 'U00000000000000000000001.ready', + ] + with patch('os.listdir', Mock(return_value=wal_files_to_skip)): + self.r._archive_ready_wals() + mock_logger_info.assert_not_called() + @patch('os.unlink', Mock()) @patch('os.listdir', Mock(return_value=[])) @patch('os.path.isfile', Mock(return_value=True))