From 4b670084888c663fb5c6366edd651d26224fd162 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 29 Jun 2016 14:29:31 +0200 Subject: [PATCH 1/2] Try to cover as much as possible pg_rewind corner-cases rewind is not possible when: 1) trying to rewind from themself 2) leader is not reachable 3) leader is_in_recovery All these cases were leading to removing of data directory... In all cases except 1) it should "retry" when leader will became available and not is_in_recovery. --- patroni/postgresql.py | 64 +++++++++++++++++++++++++++++---------- tests/test_postgresql.py | 65 ++++++++++++++++++++++++---------------- 2 files changed, 87 insertions(+), 42 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 25bb6f84..8e1d7075 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -88,6 +88,7 @@ class Postgresql(object): self.resolve_connection_addresses() self._use_pg_rewind = config.get('use_pg_rewind', False) + self._need_rewind = False self._use_slots = config.get('use_slots', True) self._version_file = os.path.join(self._data_dir, 'PG_VERSION') self._major_version = self.get_major_version() @@ -537,6 +538,7 @@ class Postgresql(object): return ret def checkpoint(self, connect_kwargs=None): + check_not_is_in_recovery = connect_kwargs is not None connect_kwargs = connect_kwargs or self._connect_kwargs for p in ['connect_timeout', 'options']: connect_kwargs.pop(p, None) @@ -545,7 +547,12 @@ class Postgresql(object): conn.autocommit = True with conn.cursor() as cur: cur.execute("SET statement_timeout = 0") + if check_not_is_in_recovery: + cur.execute('SELECT pg_is_in_recovery()') + if cur.fetchone()[0]: + return False cur.execute('CHECKPOINT') + return True except psycopg2.Error: logging.exception('Exception during CHECKPOINT') @@ -571,6 +578,7 @@ class Postgresql(object): # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones if not ret: + logger.warning('pg_ctl stop failed') self.set_state('stop failed') elif not block_callbacks: self.set_state('stopped') @@ -646,18 +654,13 @@ class Postgresql(object): if name not in ('standby_mode', 'recovery_target_timeline', 'primary_conninfo', 'primary_slot_name'): f.write("{0} = '{1}'\n".format(name, value)) - def rewind(self, leader): + def rewind(self, r): # prepare pg_rewind connection - r = get_conn_kwargs(leader.conn_url, self._superuser) env = self.write_pgpass(r) - pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r) - # first run a checkpoint on a promoted master in order - # to make it store the new timeline (5540277D.8020309@iki.fi) - self.checkpoint(r) - logger.info("running pg_rewind from %s", pc) - pg_rewind = ['pg_rewind', '-D', self._data_dir, '--source-server', pc] + dsn = 'user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1'.format(**r) + logger.info('running pg_rewind from %s', dsn) try: - return subprocess.call(pg_rewind, env=env) == 0 + return subprocess.call(['pg_rewind', '-D', self._data_dir, '--source-server', dsn], env=env) == 0 except OSError: return False @@ -724,14 +727,36 @@ class Postgresql(object): def follow(self, member, leader, recovery=False): if self.check_recovery_conf(member) and not recovery: return True + change_role = self.role == 'master' - need_rewind = change_role and self.can_rewind - if need_rewind: + 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") - if leader and leader.name != self.name and need_rewind: # we have a leader and need to rewind + + if leader and leader.name == self.name: + return logger.info('Can not rewind from myself') + if self.is_running(): - self.stop() + stopped = self.stop() self.set_role('unknown') + if not stopped: + return logger.warning('Can not run pg_rewind because posgres is still running') + + if not (leader and leader.conn_url): + return logger.info('Leader unknown, can not rewind') + + # prepare pg_rewind connection + r = get_conn_kwargs(leader.conn_url, self._superuser) + + # first make sure that we are really trying to rewind + # from the master and run a checkpoint on a t in order to + # make it store the new timeline (5540277D.8020309@iki.fi) + leader_status = self.checkpoint(r) + if not leader_status: + return logger.warning('Can not use %s for rewind: %s', leader.name, + 'is_in_recovery=true' if leader_status is False else 'not accessible') + # 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. @@ -740,24 +765,30 @@ class Postgresql(object): os.unlink(self._recovery_conf) elif os.path.isfile(self._recovery_conf): 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.update({'archive_mode': 'on', 'archive_command': 'false'}) self.single_user_mode(options=opts) - if self.rewind(leader): + + if self.rewind(r): self.write_recovery_conf(member) ret = self.start() else: - logger.error("unable to rewind the former master") + logger.error('unable to rewind the former master') self.remove_data_directory() + self.set_role('uninitialized') ret = True - else: # do not rewind until the leader becomes available + self._need_rewind = False + else: self.write_recovery_conf(member) ret = self.restart() self.set_role('replica') + if change_role: self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret @@ -791,6 +822,7 @@ class Postgresql(object): 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/tests/test_postgresql.py b/tests/test_postgresql.py index 6e0c563f..97643aba 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -230,41 +230,54 @@ class TestPostgresql(unittest.TestCase): def test_write_pgpass(self): self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'}) + def test_checkpoint(self): + with patch.object(MockCursor, 'fetchone', Mock(return_value=(True, ))): + self.assertFalse(self.p.checkpoint({'user': 'postgres'})) + with patch.object(MockCursor, 'execute', Mock()): + self.assertTrue(self.p.checkpoint()) + @patch('subprocess.call', side_effect=OSError) @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) def test_pg_rewind(self, mock_call): - self.assertTrue(self.p.rewind(self.leader)) + r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''} + self.assertTrue(self.p.rewind(r)) subprocess.call = mock_call - self.assertFalse(self.p.rewind(self.leader)) + self.assertFalse(self.p.rewind(r)) - @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)) - @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) + @patch('os.unlink', Mock(return_value=True)) @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)) + @patch.object(Postgresql, 'remove_data_directory', Mock(return_value=True)) + @patch.object(Postgresql, 'single_user_mode', Mock(return_value=1)) + @patch.object(Postgresql, 'write_pgpass', Mock(return_value={})) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True)) + @patch.object(Postgresql, 'rewind', return_value=False) def test_follow(self, mock_pg_rewind): - with patch('patroni.postgresql.Postgresql.restart', Mock(return_value=False)): - self.p.follow(None, None) + with patch.object(Postgresql, 'check_recovery_conf', Mock(return_value=True)): + self.assertTrue(self.p.follow(None, None)) # nothing to do, recovery.conf has good primary_conninfo + + with patch.object(Postgresql, 'restart', Mock(return_value=False)): + self.p.set_role('replica') + self.p.follow(None, None) # restart without rewind self.p.set_role('master') - self.p.follow(self.leader, self.leader) - self.p.follow(Leader(-1, 28, self.other), self.leader) - self.p.rewind = mock_pg_rewind - self.p.follow(self.leader, self.leader) - self.p.set_role('master') - with mock.patch('os.path.islink', MagicMock(return_value=True)): - with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): - with mock.patch('os.unlink', MagicMock(return_value=True)): - self.p.follow(self.leader, self.leader, recovery=True) - self.p.set_role('master') - with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): - self.p.rewind.return_value = True - self.p.follow(self.leader, self.leader, recovery=True) - self.p.set_role('master') - self.p.rewind.return_value = False - self.p.follow(self.leader, self.leader, recovery=True) - with mock.patch('patroni.postgresql.Postgresql.check_recovery_conf', MagicMock(return_value=True)): - self.assertTrue(self.p.follow(None, None)) + + self.p.follow(self.leader, self.me) # Can not rewind from myself + + with patch.object(Postgresql, 'stop', Mock(return_value=False)): + self.p.follow(self.leader, self.leader) # failed to stop postgres + + self.p.follow(self.leader, None) # Leader unknown, can not rewind + + self.p.follow(self.leader, self.leader) # "leader" is not accessible or is_in_recovery + + with patch.object(Postgresql, 'checkpoint', Mock(return_value=True)): + with patch('os.path.islink', Mock(return_value=True)): + self.p.follow(self.leader, self.leader) + self.p.set_role('master') + mock_pg_rewind.return_value = True + self.p.follow(self.leader, self.leader) + + self.assertTrue(self.p.follow(None, None)) # check_recovery_conf... @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)) def test_can_rewind(self): From aa10f429132d1cdd90688a6d7856e2067874cf93 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 30 Jun 2016 10:45:54 +0200 Subject: [PATCH 2/2] checkpoint method returns string status message --- patroni/postgresql.py | 17 +++++++---------- tests/test_postgresql.py | 14 +++++++------- 2 files changed, 14 insertions(+), 17 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 8e1d7075..97d91832 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -550,11 +550,11 @@ class Postgresql(object): if check_not_is_in_recovery: cur.execute('SELECT pg_is_in_recovery()') if cur.fetchone()[0]: - return False - cur.execute('CHECKPOINT') - return True + return 'is_in_recovery=true' + return cur.execute('CHECKPOINT') except psycopg2.Error: logging.exception('Exception during CHECKPOINT') + return 'not accessible or not healty' def stop(self, mode='fast', block_callbacks=False, checkpoint=True): # make sure we close all connections established against @@ -741,7 +741,7 @@ class Postgresql(object): stopped = self.stop() self.set_role('unknown') if not stopped: - return logger.warning('Can not run pg_rewind because posgres is still running') + return logger.warning('Can not run pg_rewind because postgres is still running') if not (leader and leader.conn_url): return logger.info('Leader unknown, can not rewind') @@ -753,18 +753,15 @@ class Postgresql(object): # from the master and run a checkpoint on a t in order to # make it store the new timeline (5540277D.8020309@iki.fi) leader_status = self.checkpoint(r) - if not leader_status: - return logger.warning('Can not use %s for rewind: %s', leader.name, - 'is_in_recovery=true' if leader_status is False else 'not accessible') + if leader_status: + return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status) # 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): + if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf): os.unlink(self._recovery_conf) - elif os.path.isfile(self._recovery_conf): - os.remove(self._recovery_conf) # Archived segments might be useful to pg_rewind, # clean the flags that tell we should remove them. diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 97643aba..dde9647c 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -232,9 +232,10 @@ class TestPostgresql(unittest.TestCase): def test_checkpoint(self): with patch.object(MockCursor, 'fetchone', Mock(return_value=(True, ))): - self.assertFalse(self.p.checkpoint({'user': 'postgres'})) - with patch.object(MockCursor, 'execute', Mock()): - self.assertTrue(self.p.checkpoint()) + self.assertEquals(self.p.checkpoint({'user': 'postgres'}), 'is_in_recovery=true') + with patch.object(MockCursor, 'execute', Mock(return_value=None)): + self.assertIsNone(self.p.checkpoint()) + self.assertEquals(self.p.checkpoint(), 'not accessible or not healty') @patch('subprocess.call', side_effect=OSError) @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) @@ -270,10 +271,9 @@ class TestPostgresql(unittest.TestCase): self.p.follow(self.leader, self.leader) # "leader" is not accessible or is_in_recovery - with patch.object(Postgresql, 'checkpoint', Mock(return_value=True)): - with patch('os.path.islink', Mock(return_value=True)): - self.p.follow(self.leader, self.leader) - self.p.set_role('master') + with patch.object(Postgresql, 'checkpoint', Mock(return_value=None)): + self.p.follow(self.leader, self.leader) + self.p.set_role('master') mock_pg_rewind.return_value = True self.p.follow(self.leader, self.leader)