diff --git a/patroni/ha.py b/patroni/ha.py index 0d95d372..41a2a56a 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -66,8 +66,15 @@ class Ha: if self.state_handler.is_healthy(): return False has_lock = self.has_lock() - self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader) - self.state_handler.start() + + # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind + # in order to join the former standby being promoted. + pg_controldata = self.state_handler.controldata() + if not has_lock and pg_controldata.get('Database cluster state', '') == 'in production': # crashed master + self.state_handler.require_rewind() + + # XXX: should we call ha.follow_the_leader here instead? + ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True) if has_lock: logger.info('started as readonly because i had the session lock') self.load_cluster_from_dcs() diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 4fd0e13d..59ee4936 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -47,7 +47,7 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] - self._pg_rewind = config.get('pg_rewind', {}) + self.pg_rewind = config.get('pg_rewind', {}) self.callback = config.get('callbacks', {}) self.use_slots = config.get('use_slots', True) self.schedule_load_slots = self.use_slots @@ -68,23 +68,36 @@ class Postgresql: self._connection = None self._cursor_holder = None + self._need_rewind = False self.members = [] # list of already existing replication slots self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException) - self.init_pg_rewind() - def init_pg_rewind(self): + @property + def can_rewind(self): + """ check if pg_rewind executable is there and that pg_controldata indicates + we have either wal_log_hints or checksums turned on + """ + # low-hanging fruit: check if pg_rewind configuration is there + if not self.pg_rewind or\ + not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')): + return False + + cmd = ['pg_rewind', '--help'] try: - self._pg_rewind_present = ('username' in self._pg_rewind and - subprocess.call(['pg_rewind', - '--version'], - stdout=open(os.devnull, 'w'), - stderr=subprocess.STDOUT) == 0) - if self._pg_rewind_present: - self._pg_rewind['user'] = self._pg_rewind['username'] - except: - self._pg_rewind_present = False - if self._pg_rewind and not self._pg_rewind_present: - logger.warning("pg_rewind support is disabled") + ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if ret != 0: # pg_rewind is not there, close up the shop and go home + return False + except OSError: + return False + # check if the cluster's configuration permits pg_rewind + data = self.controldata() + if data: + return data.get('wal_log_hints setting', 'off') == 'on' or\ + data.get('Data page checksum version', '0') != '0' + return False + + def require_rewind(self): + self._need_rewind = True def get_local_address(self): listen_addresses = self.listen_addresses.split(',') @@ -209,6 +222,8 @@ class Postgresql: return ret def stop(self, mode='fast', block_callbacks=False): + if not self.is_running(): + return True if block_callbacks: try: self.query('SET statement_timeout TO 0') @@ -315,10 +330,11 @@ recovery_target_timeline = 'latest' for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) - def pg_rewind(self, leader): + def rewind(self, leader): # prepare pg_rewind connection r = parseurl(leader.conn_url) - r.update(self._pg_rewind) + r.update(self.pg_rewind) + r['user'] = r['username'] env = self.write_pgpass(r, append=True) pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r) logger.info("running pg_rewind from {}".format(pc)) @@ -331,30 +347,99 @@ recovery_target_timeline = 'latest' self.write_recovery_conf(leader) return ret - def pg_rewind_verify_cluster(self): - """ check that pg_rewind can be used with the cluster """ + def controldata(self): + """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ + result = None try: - return self.query("""SELECT bool_or(setting::boolean) - FROM pg_settings - WHERE name IN ( 'data_checksums', 'wal_log_hints')""").fetchone()[0] - except: - return False + data = subprocess.check_output(['pg_controldata', self.data_dir]) + if data: + data = data.splitlines() + result = {l.split(':')[0]: l.split(':')[1].strip() for l in data if l} + except subprocess.CalledProcessError: + logger.exception("Error when calling pg_controldata") + finally: + return result - def follow_the_leader(self, leader): - if not self.check_recovery_conf(leader): + 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() + opts = [opt.strip('"\n') for opt in data.split(' "')] + for opt in opts: + if '=' in opt and opt.startswith('--'): + name, val = opt.split('=', 1) + name = name.strip('-') + result[name] = val + except IOError: + logger.exception('Error when reading postmaster.opts') + finally: + return result + + def single_user_mode(self, command=None, options={}): + """ run a given command in a single-user mode. If the command is empty - then just start and stop """ + cmd = ['postgres', '--single', '-D', self.data_dir] + for opt in sorted(options): + cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])]) + # need a database name to connect + cmd.append('postgres') + p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if p: + command and p.communicate('{}\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_xlog', 'archive_status') + if os.path.isdir(status_dir): + for f in os.listdir(status_dir): + path = os.path.join(status_dir, f) + try: + if os.path.isfile(path): + os.remove(path) + elif os.path.islink(path): # should not happen, but just in case + os.unlink(path) + except: + logger.exception("Unable to remove {}".format(path)) + + def follow_the_leader(self, leader, recovery=False): + if not self.check_recovery_conf(leader) or recovery: + change_role = (self.role == 'master') + + self._need_rewind = (self._need_rewind or change_role) and self.can_rewind + if self._need_rewind: + logger.info("set the rewind flag after demote") self.write_recovery_conf(leader) - change_role = self.role == 'master' - - if leader and change_role and self._pg_rewind_present and self.pg_rewind_verify_cluster(): - self.stop() - if self.pg_rewind(leader): + if not leader or not self._need_rewind: # do not rewind until the leader becomes available + ret = self.restart() + else: # we have a leader and need to rewind + if self.is_running(): + self.stop() + # at present, pg_rewind only runs when the cluster is shut down cleanly + # and not shutdown in recovery. We have to remove the recovery.conf if present + # and start/shutdown in a single user mode to emulate this. + # XXX: if recovery.conf is linked, it will be written anew as a normal file. + if os.path.isfile(self.recovery_conf): + os.remove(self.recovery_conf) + else: + os.unlink(self.recovery_conf) + # Archived segments might be useful to pg_rewind, + # clean the flags that tell we should remove them. + self.cleanup_archive_status() + # Start in a single user mode and stop to produce a clean shutdown + opts = self.read_postmaster_opts() + opts['archive_mode'] = 'on' + opts['archive_command'] = 'false' + self.single_user_mode(options=opts) + if self.rewind(leader): ret = self.start() else: - ret = False + logger.error("unable to rewind the former master") self.remove_data_directory() - logger.error("unable to rewind the former leader") - else: - ret = self.restart() + ret = True + self._need_rewind = False change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE) def save_configuration_files(self): @@ -379,6 +464,8 @@ recovery_target_timeline = 'latest' ret = subprocess.call(self._pg_ctl + ['promote']) == 0 if ret: self._role = 'master' + logger.info("cleared rewind flag after becoming the leader") + self._need_rewind = False self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 8551e02e..96b1cbe4 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,8 +1,15 @@ +import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psycopg2 import shutil import unittest +from sys import version_info +if version_info.major == 2: + import __builtin__ as builtins +else: + import builtins + from mock import Mock, MagicMock, patch from patroni.dcs import Cluster, Leader, Member from patroni.exceptions import PostgresException, PostgresConnectionException @@ -81,6 +88,68 @@ class MockConnect(Mock): return MockCursor(self) +def pg_controldata_string(*args, **kwargs): + return """ +pg_control version number: 942 +Catalog version number: 201509161 +Database system identifier: 6200971513092291716 +Database cluster state: shut down in recovery +pg_control last modified: Fri Oct 2 10:57:06 2015 +Latest checkpoint location: 0/30000C8 +Prior checkpoint location: 0/2000060 +Latest checkpoint's REDO location: 0/3000090 +Latest checkpoint's REDO WAL file: 000000020000000000000003 +Latest checkpoint's TimeLineID: 2 +Latest checkpoint's PrevTimeLineID: 2 +Latest checkpoint's full_page_writes: on +Latest checkpoint's NextXID: 0/943 +Latest checkpoint's NextOID: 24576 +Latest checkpoint's NextMultiXactId: 1 +Latest checkpoint's NextMultiOffset: 0 +Latest checkpoint's oldestXID: 931 +Latest checkpoint's oldestXID's DB: 1 +Latest checkpoint's oldestActiveXID: 943 +Latest checkpoint's oldestMultiXid: 1 +Latest checkpoint's oldestMulti's DB: 1 +Latest checkpoint's oldestCommitTs: 0 +Latest checkpoint's newestCommitTs: 0 +Time of latest checkpoint: Fri Oct 2 10:56:54 2015 +Fake LSN counter for unlogged rels: 0/1 +Minimum recovery ending location: 0/30241F8 +Min recovery ending loc's timeline: 2 +Backup start location: 0/0 +Backup end location: 0/0 +End-of-backup record required: no +wal_level setting: hot_standby +wal_log_hints setting: on +max_connections setting: 100 +max_worker_processes setting: 8 +max_prepared_xacts setting: 0 +max_locks_per_xact setting: 64 +track_commit_timestamp setting: off +Maximum data alignment: 8 +Database block size: 8192 +Blocks per segment of large relation: 131072 +WAL block size: 8192 +Bytes per WAL segment: 16777216 +Maximum length of identifiers: 64 +Maximum columns in an index: 32 +Maximum size of a TOAST chunk: 1996 +Size of a large-object chunk: 2048 +Date/time type storage: 64-bit integers +Float4 argument passing: by value +Float8 argument passing: by value +Data page checksum version: 0 +""" + + +def postmaster_opts_string(*args, **kwargs): + return '/usr/local/pgsql/bin/postgres "-D" "data/postgresql0" "--listen_addresses=127.0.0.1" "--port=5432"'\ + ' "--hot_standby=on" "--wal_keep_segments=8" "--wal_level=hot_standby" "--archive_command=mkdir -p ../wal_archive \n'\ + '&& cp %p ../wal_archive/%f" "--wal_log_hints=on" "--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on"'\ + ' "--max_replication_slots=5"\n' + + def psycopg2_connect(*args, **kwargs): return MockConnect() @@ -96,6 +165,7 @@ class TestPostgresql(unittest.TestCase): 'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'], 'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'}, + 'pg_rewind': {'username': 'admin', 'password': 'admin'}, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, @@ -133,32 +203,22 @@ class TestPostgresql(unittest.TestCase): def test_sync_from_leader(self): self.assertTrue(self.p.sync_from_leader(self.leader)) - @patch('os.system', side_effect=Exception("Test")) - def test_init_pg_rewind(self, mock_system): - self.p.init_pg_rewind() - # prepare parameters for pg_rewind - self.p._pg_rewind = {'username': 'foo'} - self.p.config['parameters']['data_checksums'] = 1 - os.system = mock_system - self.p.init_pg_rewind() - @patch('subprocess.call', side_effect=Exception("Test")) def test_pg_rewind(self, mock_call): - self.assertTrue(self.p.pg_rewind(self.leader)) + self.assertTrue(self.p.rewind(self.leader)) self.p subprocess.call = mock_call - self.assertFalse(self.p.pg_rewind(self.leader)) + self.assertFalse(self.p.rewind(self.leader)) - @patch('patroni.postgresql.Postgresql.pg_rewind', return_value=False) + @patch('patroni.postgresql.Postgresql.rewind', return_value=False) @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True)) def test_follow_the_leader(self, mock_pg_rewind): self.p.demote(self.leader) self.p.follow_the_leader(None) - self.p._pg_rewind_present = True self.p.demote(self.leader) self.p.follow_the_leader(self.leader) self.p.follow_the_leader(Leader(-1, None, 28, self.other)) - self.p.pg_rewind = mock_pg_rewind + self.p.rewind = mock_pg_rewind self.p.follow_the_leader(self.leader) def test_create_replica(self): @@ -248,3 +308,85 @@ class TestPostgresql(unittest.TestCase): with patch('os.unlink', Mock(side_effect=Exception)): self.p.remove_data_directory() self.p.remove_data_directory() + + @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string)) + @patch('subprocess.check_output', side_effect=subprocess.CalledProcessError) + @patch('subprocess.check_output', side_effect=Exception('Failed')) + def test_controldata(self, check_output_call_error, check_output_generic_exception): + data = self.p.controldata() + self.assertEquals(len(data), 50) + self.assertEquals(data['Database cluster state'], 'shut down in recovery') + self.assertEquals(data['wal_log_hints setting'], 'on') + self.assertEquals(int(data['Database block size']), 8192) + + subprocess.check_output = check_output_call_error + data = self.p.controldata() + self.assertIsNone(data) + + subprocess.check_output = check_output_generic_exception + self.assertRaises(Exception, self.p.controldata()) + + def test_read_postmaster_opts(self): + m = mock.mock_open(read_data=postmaster_opts_string()) + 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("foo") + data = self.p.read_postmaster_opts() + self.assertEqual(data, dict()) + + m.side_effect = Exception("foo") + self.assertRaises(Exception, self.p.read_postmaster_opts()) + + @patch('subprocess.Popen') + @patch.object(builtins, 'open', MagicMock(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(options=dict(archive_mode='on', archive_command='false')), 0) + subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir, + '-c', 'archive_command=false', '-c', 'archive_mode=on', + 'postgres'], stdin=subprocess.PIPE, + stdout=42, + stderr=subprocess.STDOUT) + subprocess_popen_mock.reset_mock() + self.assertEquals(self.p.single_user_mode(command="CHECKPOINT"), 0) + subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir, + 'postgres'], stdin=subprocess.PIPE, + stdout=42, + stderr=subprocess.STDOUT) + subprocess_popen_mock.return_value = None + self.assertEquals(self.p.single_user_mode(), 1) + + def fake_listdir(path): + if path.endswith(os.path.join('pg_xlog', 'archive_status')): + return ["a", "b", "c"] + return [] + + @patch('os.listdir', MagicMock(side_effect=fake_listdir)) + @patch('os.path.isdir', MagicMock(return_value=True)) + @patch('os.unlink', return_value=True) + @patch('os.remove', return_value=True) + @patch('os.path.islink', return_value=False) + @patch('os.path.isfile', return_value=True) + def test_cleanup_archive_status(self, mock_file, mock_link, mock_remove, mock_unlink): + ap = os.path.join(self.p.data_dir, 'pg_xlog', 'archive_status/') + self.p.cleanup_archive_status() + mock_remove.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')]) + mock_unlink.assert_not_called() + + mock_remove.reset_mock() + mock_file.return_value = False + mock_link.return_value = True + self.p.cleanup_archive_status() + mock_unlink.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')]) + mock_remove.assert_not_called() + + mock_unlink.reset_mock() + mock_remove.reset_mock() + mock_file.side_effect = Exception("foo") + self.p.cleanup_archive_status() + mock_unlink.assert_not_called() + mock_remove.assert_not_called()