mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Move ensure_clean_shutdown into rewind module (#1528)
Logically fits there better
This commit is contained in:
+1
-1
@@ -323,7 +323,7 @@ class Ha(object):
|
||||
(self.cluster.is_unlocked() or self._rewind.can_rewind):
|
||||
self._crash_recovery_executed = True
|
||||
msg = 'doing crash recovery in a single user mode'
|
||||
return self._async_executor.try_run_async(msg, self.state_handler.fix_cluster_state) or msg
|
||||
return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
|
||||
|
||||
self.load_cluster_from_dcs()
|
||||
|
||||
|
||||
@@ -887,53 +887,6 @@ class Postgresql(object):
|
||||
return candidates[0], False
|
||||
return None, False
|
||||
|
||||
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()
|
||||
for opt in data.split('" "'):
|
||||
if '=' in opt and opt.startswith('--'):
|
||||
name, val = opt.split('=', 1)
|
||||
result[name.strip('-')] = val.rstrip('"\n')
|
||||
except IOError:
|
||||
logger.exception('Error when reading postmaster.opts')
|
||||
return result
|
||||
|
||||
def single_user_mode(self, command=None, options=None):
|
||||
"""run a given command in a single-user mode. If the command is empty - then just start and stop"""
|
||||
cmd = [self.pgcommand('postgres'), '--single', '-D', self._data_dir]
|
||||
for opt, val in sorted((options or {}).items()):
|
||||
cmd.extend(['-c', '{0}={1}'.format(opt, val)])
|
||||
# need a database name to connect
|
||||
cmd.append(self._database)
|
||||
return self.cancellable.call(cmd, communicate_input=command)
|
||||
|
||||
def cleanup_archive_status(self):
|
||||
status_dir = os.path.join(self._data_dir, 'pg_' + self.wal_name, 'archive_status')
|
||||
try:
|
||||
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 OSError:
|
||||
logger.exception('Unable to remove %s', path)
|
||||
except OSError:
|
||||
logger.exception('Unable to list %s', status_dir)
|
||||
|
||||
def fix_cluster_state(self):
|
||||
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.config.remove_recovery_conf()
|
||||
return self.single_user_mode(options=opts) == 0 or None
|
||||
|
||||
def schedule_sanity_checks_after_pause(self):
|
||||
"""
|
||||
After coming out of pause we have to:
|
||||
|
||||
@@ -262,3 +262,50 @@ class Rewind(object):
|
||||
@property
|
||||
def failed(self):
|
||||
return self._state == REWIND_STATUS.FAILED
|
||||
|
||||
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._postgresql.data_dir, 'postmaster.opts')) as f:
|
||||
data = f.read()
|
||||
for opt in data.split('" "'):
|
||||
if '=' in opt and opt.startswith('--'):
|
||||
name, val = opt.split('=', 1)
|
||||
result[name.strip('-')] = val.rstrip('"\n')
|
||||
except IOError:
|
||||
logger.exception('Error when reading postmaster.opts')
|
||||
return result
|
||||
|
||||
def single_user_mode(self, command=None, options=None):
|
||||
"""run a given command in a single-user mode. If the command is empty - then just start and stop"""
|
||||
cmd = [self._postgresql.pgcommand('postgres'), '--single', '-D', self._postgresql.data_dir]
|
||||
for opt, val in sorted((options or {}).items()):
|
||||
cmd.extend(['-c', '{0}={1}'.format(opt, val)])
|
||||
# need a database name to connect
|
||||
cmd.append('template1')
|
||||
return self._postgresql.cancellable.call(cmd, communicate_input=command)
|
||||
|
||||
def cleanup_archive_status(self):
|
||||
status_dir = os.path.join(self._postgresql.data_dir, 'pg_' + self._postgresql.wal_name, 'archive_status')
|
||||
try:
|
||||
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 OSError:
|
||||
logger.exception('Unable to remove %s', path)
|
||||
except OSError:
|
||||
logger.exception('Unable to list %s', status_dir)
|
||||
|
||||
def ensure_clean_shutdown(self):
|
||||
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._postgresql.config.remove_recovery_conf()
|
||||
return self.single_user_mode(options=opts) == 0 or None
|
||||
|
||||
+1
-1
@@ -257,7 +257,7 @@ class TestHa(PostgresInit):
|
||||
self.ha.cluster = get_cluster_initialized_with_leader()
|
||||
self.assertEqual(self.ha.run_cycle(), 'starting as readonly because i had the session lock')
|
||||
|
||||
@patch.object(Postgresql, 'fix_cluster_state', Mock())
|
||||
@patch.object(Rewind, 'ensure_clean_shutdown', Mock())
|
||||
def test_crash_recovery(self):
|
||||
self.p.is_running = false
|
||||
self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
|
||||
|
||||
@@ -674,42 +674,6 @@ class TestPostgresql(BaseTestPostgresql):
|
||||
mock_postmaster.signal_stop.assert_called()
|
||||
mock_postmaster.wait.assert_called()
|
||||
|
||||
def test_read_postmaster_opts(self):
|
||||
m = mock_open(read_data='/usr/lib/postgres/9.6/bin/postgres "-D" "data/postgresql0" \
|
||||
"--listen_addresses=127.0.0.1" "--port=5432" "--hot_standby=on" "--wal_level=hot_standby" \
|
||||
"--wal_log_hints=on" "--max_wal_senders=5" "--max_replication_slots=5"\n')
|
||||
with patch.object(builtins, 'open', m):
|
||||
data = self.p.read_postmaster_opts()
|
||||
self.assertEqual(data['wal_level'], 'hot_standby')
|
||||
self.assertEqual(int(data['max_replication_slots']), 5)
|
||||
self.assertEqual(data.get('D'), None)
|
||||
|
||||
m.side_effect = IOError
|
||||
data = self.p.read_postmaster_opts()
|
||||
self.assertEqual(data, dict())
|
||||
|
||||
@patch('psutil.Popen')
|
||||
def test_single_user_mode(self, subprocess_popen_mock):
|
||||
subprocess_popen_mock.return_value.wait.return_value = 0
|
||||
self.assertEqual(self.p.single_user_mode('CHECKPOINT', {'archive_mode': 'on'}), 0)
|
||||
|
||||
@patch('os.listdir', Mock(side_effect=[OSError, ['a', 'b']]))
|
||||
@patch('os.unlink', Mock(side_effect=OSError))
|
||||
@patch('os.remove', Mock())
|
||||
@patch('os.path.islink', Mock(side_effect=[True, False]))
|
||||
@patch('os.path.isfile', Mock(return_value=True))
|
||||
def test_cleanup_archive_status(self):
|
||||
self.p.cleanup_archive_status()
|
||||
self.p.cleanup_archive_status()
|
||||
|
||||
@patch('os.unlink', Mock())
|
||||
@patch('os.listdir', Mock(return_value=[]))
|
||||
@patch('os.path.isfile', Mock(return_value=True))
|
||||
@patch.object(Postgresql, 'read_postmaster_opts', Mock(return_value={}))
|
||||
@patch.object(Postgresql, 'single_user_mode', Mock(return_value=0))
|
||||
def test_fix_cluster_state(self):
|
||||
self.assertTrue(self.p.fix_cluster_state())
|
||||
|
||||
def test_replica_cached_timeline(self):
|
||||
self.assertEqual(self.p.replica_cached_timeline(2), 3)
|
||||
|
||||
|
||||
+38
-1
@@ -1,8 +1,9 @@
|
||||
from mock import Mock, PropertyMock, patch
|
||||
from mock import Mock, PropertyMock, patch, mock_open
|
||||
|
||||
from patroni.postgresql import Postgresql
|
||||
from patroni.postgresql.cancellable import CancellableSubprocess
|
||||
from patroni.postgresql.rewind import Rewind
|
||||
from six.moves import builtins
|
||||
|
||||
from . import BaseTestPostgresql, MockCursor, psycopg2_connect
|
||||
|
||||
@@ -122,6 +123,42 @@ class TestRewind(BaseTestPostgresql):
|
||||
self.r.check_leader_is_not_in_recovery()
|
||||
self.r.check_leader_is_not_in_recovery()
|
||||
|
||||
def test_read_postmaster_opts(self):
|
||||
m = mock_open(read_data='/usr/lib/postgres/9.6/bin/postgres "-D" "data/postgresql0" \
|
||||
"--listen_addresses=127.0.0.1" "--port=5432" "--hot_standby=on" "--wal_level=hot_standby" \
|
||||
"--wal_log_hints=on" "--max_wal_senders=5" "--max_replication_slots=5"\n')
|
||||
with patch.object(builtins, 'open', m):
|
||||
data = self.r.read_postmaster_opts()
|
||||
self.assertEqual(data['wal_level'], 'hot_standby')
|
||||
self.assertEqual(int(data['max_replication_slots']), 5)
|
||||
self.assertEqual(data.get('D'), None)
|
||||
|
||||
m.side_effect = IOError
|
||||
data = self.r.read_postmaster_opts()
|
||||
self.assertEqual(data, dict())
|
||||
|
||||
@patch('psutil.Popen')
|
||||
def test_single_user_mode(self, subprocess_popen_mock):
|
||||
subprocess_popen_mock.return_value.wait.return_value = 0
|
||||
self.assertEqual(self.r.single_user_mode('CHECKPOINT', {'archive_mode': 'on'}), 0)
|
||||
|
||||
@patch('os.listdir', Mock(side_effect=[OSError, ['a', 'b']]))
|
||||
@patch('os.unlink', Mock(side_effect=OSError))
|
||||
@patch('os.remove', Mock())
|
||||
@patch('os.path.islink', Mock(side_effect=[True, False]))
|
||||
@patch('os.path.isfile', Mock(return_value=True))
|
||||
def test_cleanup_archive_status(self):
|
||||
self.r.cleanup_archive_status()
|
||||
self.r.cleanup_archive_status()
|
||||
|
||||
@patch('os.unlink', Mock())
|
||||
@patch('os.listdir', Mock(return_value=[]))
|
||||
@patch('os.path.isfile', Mock(return_value=True))
|
||||
@patch.object(Rewind, 'read_postmaster_opts', Mock(return_value={}))
|
||||
@patch.object(Rewind, 'single_user_mode', Mock(return_value=0))
|
||||
def test_ensure_clean_shutdown(self):
|
||||
self.assertTrue(self.r.ensure_clean_shutdown())
|
||||
|
||||
@patch('patroni.postgresql.rewind.Thread', MockThread)
|
||||
@patch.object(Postgresql, 'controldata')
|
||||
@patch.object(Postgresql, 'checkpoint')
|
||||
|
||||
Reference in New Issue
Block a user