From 29ac77b6e7216857875bfeccc8702c0c53631e1d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 30 Oct 2019 12:30:09 +0100 Subject: [PATCH] Compare all recovery parameters (#1208) 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 --- patroni/postgresql/config.py | 154 ++++++++++++++++++++++------------- tests/test_postgresql.py | 54 +++++++----- 2 files changed, 128 insertions(+), 80 deletions(-) diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 9951148c..7094d957 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -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: diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index be96422f..39b89f03 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -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'