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

Try to cover as much as possible pg_rewind corner-cases

Browse Source 
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.
This commit is contained in:
Alexander Kukushkin
2016-06-29 14:29:31 +02:00
parent 0c6a1307da
commit 4b67008488
2 changed files with 87 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/postgresql.py
+48 -16
View File
@@ -88,6 +88,7 @@ class Postgresql(object):
self.resolve_connection_addresses() self.resolve_connection_addresses()
self._use_pg_rewind = config.get('use_pg_rewind', False) self._use_pg_rewind = config.get('use_pg_rewind', False)
self._need_rewind = False
self._use_slots = config.get('use_slots', True) self._use_slots = config.get('use_slots', True)
self._version_file = os.path.join(self._data_dir, 'PG_VERSION') self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
self._major_version = self.get_major_version() self._major_version = self.get_major_version()
@@ -537,6 +538,7 @@ class Postgresql(object):
return ret return ret
def checkpoint(self, connect_kwargs=None): def checkpoint(self, connect_kwargs=None):
check_not_is_in_recovery = connect_kwargs is not None
connect_kwargs = connect_kwargs or self._connect_kwargs connect_kwargs = connect_kwargs or self._connect_kwargs
for p in ['connect_timeout', 'options']: for p in ['connect_timeout', 'options']:
connect_kwargs.pop(p, None) connect_kwargs.pop(p, None)
@@ -545,7 +547,12 @@ class Postgresql(object):
conn.autocommit = True conn.autocommit = True
with conn.cursor() as cur: with conn.cursor() as cur:
cur.execute("SET statement_timeout = 0") 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') cur.execute('CHECKPOINT')
return True
except psycopg2.Error: except psycopg2.Error:
logging.exception('Exception during CHECKPOINT') logging.exception('Exception during CHECKPOINT')
@@ -571,6 +578,7 @@ class Postgresql(object):
# block_callbacks is used during restart to avoid # block_callbacks is used during restart to avoid
# running start/stop callbacks in addition to restart ones # running start/stop callbacks in addition to restart ones
if not ret: if not ret:
logger.warning('pg_ctl stop failed')
self.set_state('stop failed') self.set_state('stop failed')
elif not block_callbacks: elif not block_callbacks:
self.set_state('stopped') 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'): if name not in ('standby_mode', 'recovery_target_timeline', 'primary_conninfo', 'primary_slot_name'):
f.write("{0} = '{1}'\n".format(name, value)) f.write("{0} = '{1}'\n".format(name, value))
def rewind(self, leader): def rewind(self, r):
# prepare pg_rewind connection # prepare pg_rewind connection
r = get_conn_kwargs(leader.conn_url, self._superuser)
env = self.write_pgpass(r) env = self.write_pgpass(r)
pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r) dsn = 'user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1'.format(**r)
# first run a checkpoint on a promoted master in order logger.info('running pg_rewind from %s', dsn)
# 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]
try: 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: except OSError:
return False return False
@@ -724,14 +727,36 @@ class Postgresql(object):
def follow(self, member, leader, recovery=False): def follow(self, member, leader, recovery=False):
if self.check_recovery_conf(member) and not recovery: if self.check_recovery_conf(member) and not recovery:
return True return True
change_role = self.role == 'master' change_role = self.role == 'master'
need_rewind = change_role and self.can_rewind self._need_rewind = self._need_rewind or change_role and self.can_rewind
if need_rewind:
if self._need_rewind:
logger.info("set the rewind flag after demote") 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(): if self.is_running():
self.stop() stopped = self.stop()
self.set_role('unknown') 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 ([email protected])
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 # 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 not shutdown in recovery. We have to remove the recovery.conf if present
# and start/shutdown in a single user mode to emulate this. # and start/shutdown in a single user mode to emulate this.
@@ -740,24 +765,30 @@ class Postgresql(object):
os.unlink(self._recovery_conf) os.unlink(self._recovery_conf)
elif os.path.isfile(self._recovery_conf): elif os.path.isfile(self._recovery_conf):
os.remove(self._recovery_conf) os.remove(self._recovery_conf)
# Archived segments might be useful to pg_rewind, # Archived segments might be useful to pg_rewind,
# clean the flags that tell we should remove them. # clean the flags that tell we should remove them.
self.cleanup_archive_status() self.cleanup_archive_status()
# Start in a single user mode and stop to produce a clean shutdown # Start in a single user mode and stop to produce a clean shutdown
opts = self.read_postmaster_opts() opts = self.read_postmaster_opts()
opts.update({'archive_mode': 'on', 'archive_command': 'false'}) opts.update({'archive_mode': 'on', 'archive_command': 'false'})
self.single_user_mode(options=opts) self.single_user_mode(options=opts)
if self.rewind(leader):
if self.rewind(r):
self.write_recovery_conf(member) self.write_recovery_conf(member)
ret = self.start() ret = self.start()
else: else:
logger.error("unable to rewind the former master") logger.error('unable to rewind the former master')
self.remove_data_directory() self.remove_data_directory()
self.set_role('uninitialized')
ret = True ret = True
else: # do not rewind until the leader becomes available self._need_rewind = False
else:
self.write_recovery_conf(member) self.write_recovery_conf(member)
ret = self.restart() ret = self.restart()
self.set_role('replica') self.set_role('replica')
if change_role: if change_role:
self.call_nowait(ACTION_ON_ROLE_CHANGE) self.call_nowait(ACTION_ON_ROLE_CHANGE)
return ret return ret
@@ -791,6 +822,7 @@ class Postgresql(object):
if ret: if ret:
self.set_role('master') self.set_role('master')
logger.info("cleared rewind flag after becoming the leader") logger.info("cleared rewind flag after becoming the leader")
self._need_rewind = False
self.call_nowait(ACTION_ON_ROLE_CHANGE) self.call_nowait(ACTION_ON_ROLE_CHANGE)
return ret return ret
tests/test_postgresql.py
+39 -26
View File
@@ -230,41 +230,54 @@ class TestPostgresql(unittest.TestCase):
def test_write_pgpass(self): def test_write_pgpass(self):
self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'}) 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('subprocess.call', side_effect=OSError)
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
def test_pg_rewind(self, mock_call): 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 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('os.unlink', Mock(return_value=True))
@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('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)) @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, '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): def test_follow(self, mock_pg_rewind):
with patch('patroni.postgresql.Postgresql.restart', Mock(return_value=False)): with patch.object(Postgresql, 'check_recovery_conf', Mock(return_value=True)):
self.p.follow(None, None) 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.set_role('master')
self.p.follow(self.leader, self.leader)
self.p.follow(Leader(-1, 28, self.other), self.leader) self.p.follow(self.leader, self.me) # Can not rewind from myself
self.p.rewind = mock_pg_rewind
self.p.follow(self.leader, self.leader) with patch.object(Postgresql, 'stop', Mock(return_value=False)):
self.p.set_role('master') self.p.follow(self.leader, self.leader) # failed to stop postgres
with mock.patch('os.path.islink', MagicMock(return_value=True)):
with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): self.p.follow(self.leader, None) # Leader unknown, can not rewind
with mock.patch('os.unlink', MagicMock(return_value=True)):
self.p.follow(self.leader, self.leader, recovery=True) self.p.follow(self.leader, self.leader) # "leader" is not accessible or is_in_recovery
self.p.set_role('master')
with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): with patch.object(Postgresql, 'checkpoint', Mock(return_value=True)):
self.p.rewind.return_value = True with patch('os.path.islink', Mock(return_value=True)):
self.p.follow(self.leader, self.leader, recovery=True) self.p.follow(self.leader, self.leader)
self.p.set_role('master') self.p.set_role('master')
self.p.rewind.return_value = False mock_pg_rewind.return_value = True
self.p.follow(self.leader, self.leader, recovery=True) self.p.follow(self.leader, self.leader)
with mock.patch('patroni.postgresql.Postgresql.check_recovery_conf', MagicMock(return_value=True)):
self.assertTrue(self.p.follow(None, None)) self.assertTrue(self.p.follow(None, None)) # check_recovery_conf...
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)) @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
def test_can_rewind(self): def test_can_rewind(self):