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

Compare all recovery parameters (#1208)

Browse Source 
Previously check_recovery_conf() function was only checking whether primary_conninfo has changed and never taking into account all other recovery parameters.

Fixes https://github.com/zalando/patroni/issues/1201
This commit is contained in:
Alexander Kukushkin
2019-10-30 12:30:09 +01:00
committed by GitHub
parent 9e87b00d36
commit 29ac77b6e7
2 changed files with 128 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/postgresql/config.py
+96 -58
View File
@@ -323,7 +323,7 @@ class ConfigHandler(object):
self._passfile_mtime = None
self._synchronous_standby_names = None
self._postmaster_ctime = None
self._primary_conninfo = None
self._current_recovery_params = None
self._config = {}
self._recovery_params = {}
self.reload_config(config)
@@ -515,7 +515,20 @@ class ConfigHandler(object):
return os.path.exists(self._standby_signal) or os.path.exists(self._recovery_signal)
return os.path.exists(self._recovery_conf)
def _read_primary_conninfo(self):
@property
def _triggerfile_good_name(self):
return 'trigger_file' if self._postgresql.major_version < 120000 else 'promote_trigger_file'
@property
def _triggerfile_wrong_name(self):
return 'trigger_file' if self._postgresql.major_version >= 120000 else 'promote_trigger_file'
@property
def _recovery_parameters_to_compare(self):
skip_params = {'recovery_target_inclusive', 'recovery_target_action', self._triggerfile_wrong_name}
return self._RECOVERY_PARAMETERS - skip_params
def _read_recovery_params(self):
pg_conf_mtime = mtime(self._postgresql_conf)
auto_conf_mtime = mtime(self._auto_conf)
passfile_mtime = mtime(self._passfile) if self._passfile else False
@@ -528,35 +541,43 @@ class ConfigHandler(object):
return None, False
try:
primary_conninfo = self._postgresql.query('SHOW primary_conninfo').fetchone()[0]
values = {p[0]: p[1] for p in self._get_pg_settings(self._recovery_parameters_to_compare).values()}
self._postgresql_conf_mtime = pg_conf_mtime
self._auto_conf_mtime = auto_conf_mtime
self._postmaster_ctime = postmaster_ctime
except Exception:
primary_conninfo = None
return primary_conninfo, True
values = None
return values, True
def _read_primary_conninfo_pre_v12(self):
def _read_recovery_params_pre_v12(self):
recovery_conf_mtime = mtime(self._recovery_conf)
passfile_mtime = mtime(self._passfile) if self._passfile else False
if recovery_conf_mtime == self._recovery_conf_mtime and passfile_mtime == self._passfile_mtime:
return None, False
primary_conninfo = ''
values = {}
with open(self._recovery_conf, 'r') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
value = None
match = PARAMETER_RE.match(line)
if match and match.group(1) == 'primary_conninfo':
primary_conninfo = read_recovery_param_value(line[match.end():])
if match:
value = read_recovery_param_value(line[match.end():])
if value is None:
return None, True
values[match.group(1)] = value
self._recovery_conf_mtime = recovery_conf_mtime
return primary_conninfo, True
values.setdefault('recovery_min_apply_delay', '0')
values.update({param: '' for param in self._recovery_parameters_to_compare if param not in values})
return values, True
def _check_passfile(self, wanted_primary_conninfo):
def _check_passfile(self, passfile, wanted_primary_conninfo):
# If there is a passfile in the primary_conninfo try to figure out that
# the passfile contains the line allowing connection to the given node.
# We assume that the passfile was created by Patroni and therefore doing
# the full match and not covering cases when host, port or user are set to '*'
passfile = self._primary_conninfo['passfile']
passfile_mtime = mtime(passfile)
if passfile_mtime:
try:
@@ -564,7 +585,6 @@ class ConfigHandler(object):
wanted_line = self._pgpass_line(wanted_primary_conninfo).strip()
for raw_line in f:
if raw_line.strip() == wanted_line:
self._primary_conninfo['password'] = wanted_primary_conninfo['password']
self._passfile = passfile
self._passfile_mtime = passfile_mtime
return True
@@ -572,52 +592,69 @@ class ConfigHandler(object):
logger.info('Failed to read %s', passfile)
return False
def check_recovery_conf(self, member): # Name is confusing. In fact it checks the value of primary_conninfo
def _check_primary_conninfo(self, primary_conninfo, wanted_primary_conninfo):
# first we will cover corner cases, when we are replicating from somewhere while shouldn't
# or there is no primary_conninfo but we should replicate from some specific node.
if not wanted_primary_conninfo:
return not primary_conninfo
elif not primary_conninfo:
return False
if 'passfile' in primary_conninfo and 'password' not in primary_conninfo \
and 'password' in wanted_primary_conninfo:
if self._check_passfile(primary_conninfo['passfile'], wanted_primary_conninfo):
primary_conninfo['password'] = wanted_primary_conninfo['password']
else:
return False
return all(primary_conninfo.get(p) == str(v) for p, v in wanted_primary_conninfo.items())
def check_recovery_conf(self, member):
"""Returns a tuple. The first boolean element indicates that recovery params don't match
and the second is set to `True` if the restart is required in order to apply new values"""
# TODO: recovery.conf could be stale, would be nice to detect that.
if self._postgresql.major_version >= 120000:
if not os.path.exists(self._standby_signal):
return False
_read_primary_conninfo = self._read_primary_conninfo
_read_recovery_params = self._read_recovery_params
else:
if not self.recovery_conf_exists():
return False
_read_primary_conninfo = self._read_primary_conninfo_pre_v12
_read_recovery_params = self._read_recovery_params_pre_v12
primary_conninfo, updated = _read_primary_conninfo()
# updated indicates that mtime of postgresql.conf, postgresql.auto.conf, or recovery.conf was changed
# and the primary_conninfo value was read either from config or from the database connection.
params, updated = _read_recovery_params()
# updated indicates that mtime of postgresql.conf, postgresql.auto.conf, or recovery.conf
# was changed and params were read either from the config or from the database connection.
if updated:
# primary_conninfo is one of:
# - None (exception or unparsable config)
# - '' (not in config)
# - or the actual dsn value
self._primary_conninfo = primary_conninfo
if primary_conninfo:
# We will cache parsed value until the next config change.
self._primary_conninfo = parse_dsn(primary_conninfo)
# If we failed to parse non-empty connection string this indicates that config if broken.
if not self._primary_conninfo:
return False
elif primary_conninfo is not None:
self._primary_conninfo = {}
else: # primary_conninfo is None, config is probably broken
if params is None: # exception or unparsable config
return False
wanted_primary_conninfo = self.primary_conninfo_params(member)
# first we will cover corner cases, when we are replicating from somewhere while shouldn't
# or there is no primary_conninfo but we should replicate from some specific node.
if not wanted_primary_conninfo:
return not self._primary_conninfo
elif not self._primary_conninfo:
return False
# We will cache parsed value until the next config change.
self._current_recovery_params = params
if params['primary_conninfo']:
params['primary_conninfo'] = parse_dsn(params['primary_conninfo'])
# If we failed to parse non-empty connection string this indicates that config if broken.
if not params['primary_conninfo']:
return False
else: # empty string, primary_conninfo is not in the config
params['primary_conninfo'] = {}
if 'passfile' in self._primary_conninfo and 'password' in wanted_primary_conninfo and \
'password' not in self._primary_conninfo and not self._check_passfile(wanted_primary_conninfo):
return False
return all(self._primary_conninfo.get(p) == str(v) for p, v in wanted_primary_conninfo.items())
ret = True
wanted_recovery_params = self.build_recovery_params(member)
for param, value in self._current_recovery_params.items():
if param == 'recovery_min_apply_delay':
if not compare_values('integer', 'ms', value, wanted_recovery_params.get(param, 0)):
ret = False
elif param == 'primary_conninfo':
if not self._check_primary_conninfo(value, wanted_recovery_params.get('primary_conninfo', {})):
ret = False
elif (param != 'primary_slot_name' or wanted_recovery_params.get('primary_conninfo')) \
and str(value) != str(wanted_recovery_params.get(param, '')):
ret = False
return ret
@staticmethod
def _remove_file_if_exists(name):
@@ -696,13 +733,9 @@ class ConfigHandler(object):
self._config['recovery_conf'] = recovery_conf
if self.get('recovery_conf'):
good_name, bad_name = 'trigger_file', 'promote_trigger_file'
if self._postgresql.major_version >= 120000:
good_name, bad_name = bad_name, good_name
value = self._config['recovery_conf'].pop(bad_name, None)
if good_name not in self._config['recovery_conf'] and value:
self._config['recovery_conf'][good_name] = value
value = self._config['recovery_conf'].pop(self._triggerfile_wrong_name, None)
if self._triggerfile_good_name not in self._config['recovery_conf'] and value:
self._config['recovery_conf'][self._triggerfile_good_name] = value
def get_server_parameters(self, config):
parameters = config['parameters'].copy()
@@ -780,6 +813,12 @@ class ConfigHandler(object):
self._postgresql.set_connection_kwargs(self.local_connect_kwargs)
def _get_pg_settings(self, names):
return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context '
+ ' FROM pg_catalog.pg_settings ' +
' WHERE pg_catalog.lower(name) = ANY(%s)'),
[n.lower() for n in names])}
@staticmethod
def _handle_wal_buffers(old_values, changes):
wal_block_size = parse_int(old_values['wal_block_size'][1])
@@ -802,17 +841,16 @@ class ConfigHandler(object):
conf_changed = hba_changed = ident_changed = local_connection_address_changed = pending_restart = False
if self._postgresql.state == 'running':
changes = CaseInsensitiveDict({p: v for p, v in server_parameters.items() if '.' not in p})
changes.update({p: None for p in self._server_parameters.keys() if not ('.' in p or p in changes)})
changes = CaseInsensitiveDict({p: v for p, v in server_parameters.items()
if '.' not in p and p.lower() not in self._RECOVERY_PARAMETERS})
changes.update({p: None for p in self._server_parameters.keys()
if not ('.' in p or p in changes or p.lower() in self._RECOVERY_PARAMETERS)})
if changes:
if 'wal_buffers' in changes: # we need to calculate the default value of wal_buffers
undef = [p for p in ('shared_buffers', 'wal_segment_size', 'wal_block_size') if p not in changes]
changes.update({p: None for p in undef})
# XXX: query can raise an exception
old_values = {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context '
+ 'FROM pg_catalog.pg_settings ' +
' WHERE pg_catalog.lower(name) = ANY(%s)'),
[k.lower() for k in changes.keys()])}
old_values = self._get_pg_settings(changes.keys())
if 'wal_buffers' in changes:
self._handle_wal_buffers(old_values, changes)
for p in undef:
tests/test_postgresql.py
+32 -22
View File
@@ -202,43 +202,53 @@ class TestPostgresql(BaseTestPostgresql):
self.assertEqual(self.p.checkpoint(), 'not accessible or not healty')
@patch('patroni.postgresql.config.mtime', mock_mtime)
@patch.object(MockCursor, 'fetchone')
def test_check_recovery_conf(self, mock_fetchone):
mock_fetchone.side_effect = [('foo=bar',), ('',), ('foo',), ('host=1 passfile=' + self.p.config._pgpass,)]
@patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
def test_check_recovery_conf(self, mock_get_pg_settings):
mock_get_pg_settings.return_value = {
'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster'],
'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup']
}
self.assertFalse(self.p.config.check_recovery_conf(None))
self.p.config.write_recovery_conf({'standby_mode': 'on'})
self.assertFalse(self.p.config.check_recovery_conf(None))
mock_get_pg_settings.return_value['primary_conninfo'][1] = ''
mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
self.assertFalse(self.p.config.check_recovery_conf(None))
mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
self.assertTrue(self.p.config.check_recovery_conf(None))
conninfo = {'host': '1', 'password': 'bar'}
for version in (120000, 100000):
with patch.object(Postgresql, 'major_version', PropertyMock(return_value=version)):
self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'sslmode': 'prefer'}})
self.assertFalse(self.p.config.check_recovery_conf(None))
self.p.config.write_recovery_conf({'primary_conninfo': {'sslmode': 'prefer'}})
self.p.config.write_postgresql_conf()
self.assertFalse(self.p.config.check_recovery_conf(None))
self.p.config.write_recovery_conf({'standby_mode': 'on'})
self.assertTrue(self.p.config.check_recovery_conf(None))
with patch('patroni.postgresql.config.ConfigHandler.primary_conninfo_params',
Mock(return_value=conninfo.copy())):
self.assertFalse(self.p.config.check_recovery_conf(None))
self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': conninfo.copy()})
self.p.config.check_recovery_conf(None)
with patch('patroni.postgresql.config.ConfigHandler.primary_conninfo_params', Mock(return_value=conninfo)):
mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
self.assertFalse(self.p.config.check_recovery_conf(None))
mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 passfile=' + self.p.config._pgpass
mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
self.assertFalse(self.p.config.check_recovery_conf(None))
self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': conninfo.copy()})
self.p.config.write_postgresql_conf()
self.assertTrue(self.p.config.check_recovery_conf(None))
@patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000))
@patch.object(Postgresql, 'is_running', MockPostmaster)
@patch.object(MockPostmaster, 'create_time', Mock(return_value=1234567), create=True)
@patch.object(MockCursor, 'fetchone', Mock(return_value=('',)))
def test__read_primary_conninfo(self):
@patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
def test__read_recovery_params(self, mock_get_pg_settings):
mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string', 'postmaster']}
self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
self.p.config.write_postgresql_conf()
self.assertTrue(self.p.config.check_recovery_conf(None))
self.assertTrue(self.p.config.check_recovery_conf(None))
with patch.object(Postgresql, 'query', Mock(side_effect=Exception)),\
patch('patroni.postgresql.config.mtime', mock_mtime):
mock_get_pg_settings.side_effect = Exception
with patch('patroni.postgresql.config.mtime', mock_mtime):
self.assertFalse(self.p.config.check_recovery_conf(None))
@patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000))
def test__read_primary_conninfo_pre_v12(self):
def test__read_recovery_params_pre_v12(self):
self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
self.assertFalse(self.p.config.check_recovery_conf(None))
self.assertFalse(self.p.config.check_recovery_conf(None))
self.p.config.write_recovery_conf({'standby_mode': '\n'})
with patch('patroni.postgresql.config.mtime', mock_mtime):
self.assertFalse(self.p.config.check_recovery_conf(None))
def test_write_postgresql_and_sanitize_auto_conf(self):
read_data = 'primary_conninfo = foo\nfoo = bar\n'