diff --git a/patroni/ha.py b/patroni/ha.py index d4102e61..0019253d 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -92,8 +92,19 @@ class Ha: def recover(self): has_lock = self.has_lock() - self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader) - if not 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 and\ + pg_controldata.get('Database cluster state', '') == 'in production': # crashed master + self.state_handler.require_rewind() + + # XXX: follow the leader calls stop, which might take quite some time. + # perhaps we should run sync asynchronously + # (we still need the exit code from follow_the_leader) + ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True) + if not ret: if not has_lock: return 'failed to start postgres' self.dcs.delete_leader() diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 673f16c3..fc7956e9 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -48,6 +48,7 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] + 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 @@ -67,6 +68,7 @@ class Postgresql: self._connection = None self._cursor_holder = None + self._need_rewind = False self.replication_slots = [] # list of already existing replication slots self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException) @@ -79,6 +81,33 @@ class Postgresql: self._state = 'running' self._role = 'master' if self.is_leader() else 'replica' + @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: + 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(',') local_address = listen_addresses[0].strip() # take first address from listen_addresses @@ -135,16 +164,19 @@ class Postgresql: def delete_trigger_file(self): os.path.exists(self.trigger_file) and os.unlink(self.trigger_file) - def sync_from_leader(self, leader): - r = parseurl(leader.conn_url) - + def write_pgpass(self, record): pgpass = 'pgpass' with open(pgpass, 'w') as f: os.fchmod(f.fileno(), 0o600) - f.write('{host}:{port}:*:{user}:{password}\n'.format(**r)) - + f.write('{host}:{port}:*:{user}:{password}\n'.format(**record)) env = os.environ.copy() env['PGPASSFILE'] = pgpass + return env + + def sync_from_leader(self, leader): + r = parseurl(leader.conn_url) + + env = self.write_pgpass(r) return self.create_replica(r, env) == 0 @staticmethod @@ -308,10 +340,7 @@ class Postgresql: with open(self.recovery_conf, 'r') as f: for line in f: if line.startswith('primary_conninfo'): - if not pattern: - return False - return pattern in line - + return pattern and (pattern in line) return not pattern def write_recovery_conf(self, leader): @@ -326,12 +355,120 @@ recovery_target_timeline = 'latest' for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) - def follow_the_leader(self, leader): - if not self.check_recovery_conf(leader): + def rewind(self, leader): + # prepare pg_rewind connection + r = parseurl(leader.conn_url) + r.update(self.pg_rewind) + r['user'] = r['username'] + env = self.write_pgpass(r) + pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r) + logger.info("running pg_rewind from {}".format(pc)) + pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc] + try: + ret = (subprocess.call(pg_rewind, env=env) == 0) + except: + ret = False + if ret: self.write_recovery_conf(leader) - run_callback = self.role == 'master' - self.restart() - run_callback and self.call_nowait(ACTION_ON_ROLE_CHANGE) + return ret + + def controldata(self): + """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ + result = {} + try: + 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 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.islink(path): + os.unlink(path) + elif os.path.isfile(path): + os.remove(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) + 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.islink(self.recovery_conf): + os.unlink(self.recovery_conf) + else: + os.remove(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: + logger.error("unable to rewind the former master") + self.remove_data_directory() + ret = True + self._need_rewind = False + change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE) + return ret + else: + return True def save_configuration_files(self): """ @@ -355,6 +492,8 @@ recovery_target_timeline = 'latest' ret = subprocess.call(self._pg_ctl + ['promote']) == 0 if ret: self.set_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/patroni/scripts/aws.py b/patroni/scripts/aws.py index b172a8c3..a0622f15 100755 --- a/patroni/scripts/aws.py +++ b/patroni/scripts/aws.py @@ -1,4 +1,4 @@ -#!/usr/bin/python +#!/usr/bin/env python import logging import requests diff --git a/patroni/scripts/restore.py b/patroni/scripts/restore.py index 4ac091f3..6b20e3e8 100755 --- a/patroni/scripts/restore.py +++ b/patroni/scripts/restore.py @@ -1,4 +1,4 @@ -#!/usr/bin/python +#!/usr/bin/env python # arguments are: # - cluster scope # - cluster role diff --git a/postgres0.yml b/postgres0.yml index f800183a..a155b1cd 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -34,6 +34,9 @@ postgresql: data_dir: data/postgresql0 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True + pg_rewind: + username: postgres + password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -42,6 +45,7 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: + username: postgres password: zalando admin: username: admin @@ -62,3 +66,4 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" + wal_log_hints: "on" diff --git a/postgres1.yml b/postgres1.yml index e1c3e663..94e33a42 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -34,6 +34,9 @@ postgresql: data_dir: data/postgresql1 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True + pg_rewind: + username: postgres + password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -42,6 +45,7 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: + user: postgres password: zalando admin: username: admin @@ -62,3 +66,4 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" + wal_log_hints: "on" diff --git a/tests/test_ha.py b/tests/test_ha.py index 0a986280..38bb07a6 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -117,13 +117,14 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'started as a secondary') def test_recover_replica_failed(self): + self.p.controldata = lambda: {'Database cluster state': 'in production'} self.p.is_healthy = false - self.p.start = false + self.p.follow_the_leader = false self.assertEquals(self.ha.run_cycle(), 'failed to start postgres') def test_recover_master_failed(self): + self.p.follow_the_leader = false self.p.is_healthy = false - self.p.start = false self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'removed leader key after trying and failing to start postgres') diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 007310a5..e5e2dfd4 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,14 +1,22 @@ +import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psycopg2 import shutil import unittest -from mock import Mock, patch +from sys import version_info +if version_info.major == 2: + import __builtin__ as builtins +else: + import builtins + +from mock import Mock, MagicMock, PropertyMock, patch, mock_open from patroni.dcs import Cluster, Leader, Member from patroni.exceptions import PostgresException, PostgresConnectionException from patroni.postgresql import Postgresql from patroni.utils import RetryFailedError from test_ha import false +import subprocess class MockCursor: @@ -77,6 +85,68 @@ class MockConnect(Mock): pass +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() @@ -94,6 +164,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'}, @@ -142,11 +213,48 @@ class TestPostgresql(unittest.TestCase): def test_sync_from_leader(self): self.assertTrue(self.p.sync_from_leader(self.leader)) - def test_follow_the_leader(self): - self.p.follow_the_leader(self.leader) + @patch('subprocess.call', side_effect=Exception("Test")) + def test_pg_rewind(self, mock_call): + self.assertTrue(self.p.rewind(self.leader)) + subprocess.call = mock_call + self.assertFalse(self.p.rewind(self.leader)) + + @patch('patroni.postgresql.Postgresql.rewind', return_value=False) + @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True)) + @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1)) + def test_follow_the_leader(self, mock_pg_rewind): + self.p.demote() + self.p.follow_the_leader(None) self.p.demote() self.p.follow_the_leader(self.leader) self.p.follow_the_leader(Leader(-1, 28, self.other)) + self.p.rewind = mock_pg_rewind + self.p.follow_the_leader(self.leader) + self.p.require_rewind() + with mock.patch('os.path.islink', MagicMock(return_value=True)): + with mock.patch('os.unlink', MagicMock(return_value=True)): + with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): + self.p.follow_the_leader(self.leader, recovery=True) + self.p.require_rewind() + with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): + self.p.rewind.return_value = True + self.p.follow_the_leader(self.leader, recovery=True) + self.p.rewind.return_value = False + self.p.follow_the_leader(self.leader, recovery=True) + + def test_can_rewind(self): + tmp = self.p.pg_rewind + self.p.pg_rewind = None + self.assertFalse(self.p.can_rewind) + self.p.pg_rewind = tmp + with mock.patch('subprocess.call', MagicMock(return_value=1)): + self.assertFalse(self.p.can_rewind) + with mock.patch('subprocess.call', side_effect=OSError("foo")): + self.assertFalse(self.p.can_rewind) + tmp = self.p.controldata() + self.p.controldata = lambda: {'wal_log_hints setting': 'on'} + self.assertTrue(self.p.can_rewind) + self.p.controldata = tmp def test_create_replica(self): self.p.delete_trigger_file = Mock(side_effect=OSError()) @@ -236,3 +344,88 @@ 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.assertEquals(data, dict()) + + subprocess.check_output = check_output_generic_exception + self.assertRaises(Exception, self.p.controldata()) + + def test_read_postmaster_opts(self): + m = 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") + mock_link.side_effect = Exception("foo") + self.p.cleanup_archive_status() + mock_unlink.assert_not_called() + mock_remove.assert_not_called()