gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #230 from zalando/bugfix/pg_rewind

Browse Source 
Try to cover as much as possible pg_rewind corner-cases
This commit is contained in:
Alexander Kukushkin
2016-06-30 12:09:10 +02:00
committed by GitHub
parent 489ed20331 aa10f42913
commit dc27a30800
2 changed files with 87 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/postgresql.py
+49 -20
View File
@@ -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,9 +547,14 @@ class Postgresql(object):
conn.autocommit = True
with conn.cursor() as cur:
cur.execute("SET statement_timeout = 0")
cur.execute('CHECKPOINT')
if check_not_is_in_recovery:
cur.execute('SELECT pg_is_in_recovery()')
if cur.fetchone()[0]:
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
@@ -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 ([email protected])
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,40 +727,65 @@ 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 postgres 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 ([email protected])
leader_status = self.checkpoint(r)
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.
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 +819,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
tests/test_postgresql.py
+38 -25
View File
@@ -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.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()))
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.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=None)):
self.p.follow(self.leader, self.leader)
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))
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):