From 96b75fa7cbf43f185642cf7d5e5b6b57ecf231bc Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 12 May 2022 07:45:49 +0200 Subject: [PATCH] Special handling of check_recovery_conf for v12+ (#2292) When starting as a replica it may take some time before Postgres starts accepting new connections, but meanwhile, it could happen that the leader transitioned to a different member and the `primary_conninfo` must be updated. On pre v12 Patroni regularly checks `recovery.conf` in order to check that recovery parameters match the expectation. Starting from v12 recovery parameters were converted to GUC's and Patroni gets current values from the `pg_settings` view. The last one creates a problem when it takes more than a minute for Postgres to start accepting new connections. Since Patroni attempts to execute at least `pg_is_in_recovery()` every HA loop, and it is raising at exception, the `check_recovery_conf()` effectively wasn't reachable until recovery is finished, but it changed when #2082 was introduced. As a result of #2082 we got the following behavior: 1. Up to v12 (not including) everything was working as expected 2. v12 and v13 - Patroni restarting Postgres after 1m of recovery 3. v14+ - the `check_recovery_conf()` is not executed because the `replay_paused()` method raising an exception. In order to properly handle changes of recovery parameters or leader transitioned to a different node on v12+, we will rely on the cached values of recovery parameters until Postgres becomes ready to execute queries. Close https://github.com/zalando/patroni/issues/2289 --- patroni/ha.py | 7 +++-- patroni/postgresql/__init__.py | 2 +- patroni/postgresql/config.py | 51 ++++++++++++++++++++++------------ tests/test_postgresql.py | 4 ++- 4 files changed, 42 insertions(+), 22 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 7ec0d6ce..0bd2747a 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -423,9 +423,10 @@ class Ha(object): self.state_handler.get_history(self._leader_timeline + 1): self._rewind.trigger_check_diverged_lsn() - msg = self._handle_rewind_or_reinitialize() - if msg: - return msg + if not self.state_handler.is_starting(): + msg = self._handle_rewind_or_reinitialize() + if msg: + return msg if not self.is_paused(): self.state_handler.handle_parameter_change() diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 3b9f2507..c52c397b 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -387,7 +387,7 @@ class Postgresql(object): self._query('SELECT pg_catalog.pg_{0}_replay_resume()'.format(self.wal_name)) def handle_parameter_change(self): - if self.major_version >= 140000 and self.replay_paused(): + if self.major_version >= 140000 and not self.is_starting() and self.replay_paused(): logger.info('Resuming paused WAL replay for PostgreSQL 14+') self.resume_wal_replay() diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 6afb9e77..93143152 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -577,6 +577,9 @@ class ConfigHandler(object): return self._RECOVERY_PARAMETERS - skip_params def _read_recovery_params(self): + if self._postgresql.is_starting(): + return None, False + pg_conf_mtime = mtime(self._postgresql_conf) auto_conf_mtime = mtime(self._auto_conf) passfile_mtime = mtime(self._passfile) if self._passfile else False @@ -650,16 +653,17 @@ class ConfigHandler(object): elif not primary_conninfo: return False - wal_receiver_primary_conninfo = self._postgresql.primary_conninfo() - if wal_receiver_primary_conninfo: - wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo) - # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison + if not self._postgresql.is_starting(): + wal_receiver_primary_conninfo = self._postgresql.primary_conninfo() if wal_receiver_primary_conninfo: - primary_conninfo = wal_receiver_primary_conninfo - # There could be no password in the primary_conninfo or it is masked. - # Just copy the "desired" value in order to make comparison succeed. - if 'password' in wanted_primary_conninfo: - primary_conninfo['password'] = wanted_primary_conninfo['password'] + wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo) + # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison + if wal_receiver_primary_conninfo: + primary_conninfo = wal_receiver_primary_conninfo + # There could be no password in the primary_conninfo or it is masked. + # Just copy the "desired" value in order to make comparison succeed. + if 'password' in wanted_primary_conninfo: + primary_conninfo['password'] = wanted_primary_conninfo['password'] if 'passfile' in primary_conninfo and 'password' not in primary_conninfo \ and 'password' in wanted_primary_conninfo: @@ -668,7 +672,7 @@ class ConfigHandler(object): else: return False - return all(primary_conninfo.get(p) == str(v) for p, v in wanted_primary_conninfo.items() if v is not None) + return all(str(primary_conninfo.get(p)) == str(v) for p, v in wanted_primary_conninfo.items() if v is not None) def check_recovery_conf(self, member): """Returns a tuple. The first boolean element indicates that recovery params don't match @@ -704,16 +708,19 @@ class ConfigHandler(object): else: # empty string, primary_conninfo is not in the config primary_conninfo[0] = {} - # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver - wal_receiver_primary_slot_name = self._postgresql.primary_slot_name() - if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo(): - wal_receiver_primary_slot_name = '' - if wal_receiver_primary_slot_name is not None: - self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name + if not self._postgresql.is_starting(): + # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver + wal_receiver_primary_slot_name = self._postgresql.primary_slot_name() + if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo(): + wal_receiver_primary_slot_name = '' + if wal_receiver_primary_slot_name is not None: + self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name # Increment the 'reload' to enforce write of postgresql.conf when joining the running postgres required = {'restart': 0, - 'reload': int(not self._postgresql.cb_called and self._postgresql.major_version >= 120000)} + 'reload': int(self._postgresql.major_version >= 120000 + and not self._postgresql.cb_called + and not self._postgresql.is_starting())} def record_missmatch(mtype): required['restart' if mtype else 'reload'] += 1 @@ -780,6 +787,15 @@ class ConfigHandler(object): else: self._remove_file_if_exists(self._standby_signal) open(self._recovery_signal, 'w').close() + + def restart_required(name): + if self._postgresql.major_version >= 140000: + return False + return name == 'restore_command' or (self._postgresql.major_version < 130000 + and name in ('primary_conninfo', 'primary_slot_name')) + + self._current_recovery_params = {n: [v, restart_required(n), self._postgresql_conf] + for n, v in recovery_params.items()} else: with ConfigWriter(self._recovery_conf) as f: os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD) @@ -789,6 +805,7 @@ class ConfigHandler(object): for name in (self._recovery_conf, self._standby_signal, self._recovery_signal): self._remove_file_if_exists(name) self._recovery_params = {} + self._current_recovery_params = None def _sanitize_auto_conf(self): overwrite = False diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 523072c2..3f9bd738 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -95,7 +95,7 @@ class TestPostgresql(BaseTestPostgresql): @patch('subprocess.call', Mock(return_value=0)) @patch('os.rename', Mock()) @patch('patroni.postgresql.CallbackExecutor', Mock()) - @patch.object(Postgresql, 'get_major_version', Mock(return_value=130000)) + @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) def setUp(self): super(TestPostgresql, self).setUp() @@ -287,6 +287,8 @@ class TestPostgresql(BaseTestPostgresql): mock_get_pg_settings.side_effect = Exception with patch('patroni.postgresql.config.mtime', mock_mtime): self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) + with patch.object(Postgresql, 'is_starting', Mock(return_value=True)): + self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) @patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000)) @patch.object(Postgresql, 'primary_conninfo', Mock(return_value='host=1'))