diff --git a/patroni/ha.py b/patroni/ha.py index 3f342c62..43d14a94 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -207,6 +207,16 @@ class Ha(object): msg = "starting as a secondary" node_to_follow = self._get_node_to_follow(self.cluster) + # once we already tried to start postgres but failed, single user mode is a rescue in this case + if self.recovering and not self.state_handler.rewind_executed and self.state_handler.can_rewind: + data = self.state_handler.controldata() + if data.get('Database cluster state') not in ('shut down', 'shut down in recovery'): + self.recovering = False + msg = 'fixing cluster state in a single user mode' + self._async_executor.schedule(msg) + self._async_executor.run_async(self.state_handler.fix_cluster_state) + return msg + self.recovering = True self._async_executor.schedule('restarting after failure') diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 356def42..edd43f63 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -1261,12 +1261,14 @@ class Postgresql(object): else: # otherwise analyze pg_controldata output data = self.controldata() try: + if data.get('Database cluster state') == 'shut down in recovery': + lsn = data.get('Minimum recovery ending location') + timeline = int(data.get("Min recovery ending loc's timeline")) + if lsn == '0/0' or timeline == 0: # it was a master when it crashed + data['Database cluster state'] = 'shut down' if data.get('Database cluster state') == 'shut down': lsn = data.get('Latest checkpoint location') timeline = int(data.get("Latest checkpoint's TimeLineID")) - elif data.get('Database cluster state') == 'shut down in recovery': - lsn = data.get('Minimum recovery ending location') - timeline = int(data.get("Min recovery ending loc's timeline")) except (TypeError, ValueError): logger.exception('Failed to get local timeline and lsn from pg_controldata output') logger.info('Local timeline=%s lsn=%s', timeline, lsn) @@ -1744,3 +1746,57 @@ $$""".format(name, ' '.join(options)), name, password, password) 90600 """ return Postgresql.postgres_version_to_int(pg_version + '.0') + + def read_postmaster_opts(self): + """returns the list of option names/values from postgres.opts, Empty dict if read failed or no file""" + result = {} + try: + with open(os.path.join(self._data_dir, 'postmaster.opts')) as f: + data = f.read() + for opt in data.split('" "'): + if '=' in opt and opt.startswith('--'): + name, val = opt.split('=', 1) + result[name.strip('-')] = val.rstrip('"\n') + except IOError: + logger.exception('Error when reading postmaster.opts') + return result + + def single_user_mode(self, command=None, options=None): + """run a given command in a single-user mode. If the command is empty - then just start and stop""" + cmd = [self._pgcommand('postgres'), '--single', '-D', self._data_dir] + for opt, val in sorted((options or {}).items()): + cmd.extend(['-c', '{0}={1}'.format(opt, val)]) + # need a database name to connect + cmd.append(self._database) + p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if p: + if command: + p.communicate('{0}\n'.format(command)) + p.stdin.close() + return p.wait() + return 1 + + def cleanup_archive_status(self): + status_dir = os.path.join(self._data_dir, 'pg_' + self.wal_name, 'archive_status') + try: + for f in os.listdir(status_dir): + path = os.path.join(status_dir, f) + try: + if os.path.islink(path): + os.unlink(path) + elif os.path.isfile(path): + os.remove(path) + except OSError: + logger.exception('Unable to remove %s', path) + except OSError: + logger.exception('Unable to list %s', status_dir) + + def fix_cluster_state(self): + self.cleanup_archive_status() + + # Start in a single user mode and stop to produce a clean shutdown + opts = self.read_postmaster_opts() + opts.update({'archive_mode': 'on', 'archive_command': 'false'}) + if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf): + os.unlink(self._recovery_conf) + return self.single_user_mode(options=opts) == 0 or None diff --git a/tests/test_ha.py b/tests/test_ha.py index 645886f5..6162ef2c 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -193,6 +193,15 @@ class TestHa(unittest.TestCase): self.ha.cluster = get_cluster_initialized_with_leader() self.assertEquals(self.ha.run_cycle(), 'running pg_rewind from leader') + @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True)) + @patch.object(Postgresql, 'fix_cluster_state', Mock()) + def test_single_user_after_recover_failed(self): + self.p.controldata = lambda: {'Database cluster state': 'in production'} + self.p.is_running = false + self.p.follow = false + self.assertEquals(self.ha.run_cycle(), 'starting as a secondary') + self.assertEquals(self.ha.run_cycle(), 'fixing cluster state in a single user mode') + @patch('sys.exit', return_value=1) @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True)) def test_sysid_no_match(self, exit_mock): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index ceb5d2fe..94491e9f 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -327,10 +327,10 @@ class TestPostgresql(unittest.TestCase): @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True)) def test__get_local_timeline_lsn(self): self.p.trigger_check_diverged_lsn() - with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down'})): - self.p.rewind_needed_and_possible(self.leader) with patch.object(Postgresql, 'controldata', - Mock(return_value={'Database cluster state': 'shut down in recovery'})): + Mock(return_value={'Database cluster state': 'shut down in recovery', + 'Minimum recovery ending location': '0/0', + "Min recovery ending loc's timeline": '0'})): self.p.rewind_needed_and_possible(self.leader) with patch.object(Postgresql, 'is_running', Mock(return_value=True)): with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(False, ), Exception])): @@ -883,3 +883,40 @@ class TestPostgresql(unittest.TestCase): def test_terminate_starting_postmaster(self): self.p.terminate_starting_postmaster(123) self.p.terminate_starting_postmaster(123) + + def test_read_postmaster_opts(self): + m = mock_open(read_data='/usr/lib/postgres/9.6/bin/postgres "-D" "data/postgresql0" \ +"--listen_addresses=127.0.0.1" "--port=5432" "--hot_standby=on" "--wal_level=hot_standby" \ +"--wal_log_hints=on" "--max_wal_senders=5" "--max_replication_slots=5"\n') + with patch.object(builtins, 'open', m): + data = self.p.read_postmaster_opts() + self.assertEquals(data['wal_level'], 'hot_standby') + self.assertEquals(int(data['max_replication_slots']), 5) + self.assertEqual(data.get('D'), None) + + m.side_effect = IOError + data = self.p.read_postmaster_opts() + self.assertEqual(data, dict()) + + @patch('subprocess.Popen') + @patch.object(builtins, 'open', Mock(return_value=42)) + def test_single_user_mode(self, subprocess_popen_mock): + subprocess_popen_mock.return_value.wait.return_value = 0 + self.assertEquals(self.p.single_user_mode(command="CHECKPOINT"), 0) + subprocess_popen_mock.return_value = None + self.assertEquals(self.p.single_user_mode(), 1) + + @patch('os.listdir', Mock(side_effect=[OSError, ['a', 'b']])) + @patch('os.unlink', Mock(side_effect=OSError)) + @patch('os.remove', Mock()) + @patch('os.path.islink', Mock(side_effect=[True, False])) + @patch('os.path.isfile', Mock(return_value=True)) + def test_cleanup_archive_status(self): + self.p.cleanup_archive_status() + self.p.cleanup_archive_status() + + @patch('os.unlink', Mock()) + @patch('os.path.isfile', Mock(return_value=True)) + @patch.object(Postgresql, 'single_user_mode', Mock(return_value=0)) + def test_fix_cluster_state(self): + self.assertTrue(self.p.fix_cluster_state())