From ad5c686c11df6631bea3d5b88ee2b55460cbdb76 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 15 May 2020 18:04:24 +0200 Subject: [PATCH] Take advantage of pg_stat_wal_recevier (#1513) So far Patroni was parsing `recovery.conf` or querying `pg_settings` in order to get the current values of recovery parameters. On PostgreSQL earlier than 12 it could easily happen that the value of `primary_conninfo` in the `recovery.conf` has nothing to do with reality. Luckily for us, on PostgreSQL 9.6+ there is a `pg_stat_wal_receiver` view, which contains current values of `primary_conninfo` and `primary_slot_name`. The password field is masked through, but this is fine, because authentication happens only during opening the connection. All other parameters we compare as usual. Another advantage of `pg_stat_wal_recevier` - it contains the current timeline, therefore on 9.6+ we don't need to use the replication connection trick if walreceiver process is alive. If there is no walreceiver process available or it is not streaming we will stick to old methods. --- patroni/ha.py | 2 ++ patroni/postgresql/__init__.py | 28 +++++++++++++++++++++++----- patroni/postgresql/config.py | 18 ++++++++++++++++++ patroni/postgresql/rewind.py | 4 ++-- tests/__init__.py | 2 +- tests/test_ha.py | 1 + tests/test_postgresql.py | 11 +++++++++++ 7 files changed, 58 insertions(+), 8 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 3219a18e..953c9636 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -198,6 +198,8 @@ class Ha(object): try: timeline, wal_position, pg_control_timeline = self.state_handler.timeline_wal_position() data['xlog_location'] = wal_position + if not timeline: # try pg_stat_wal_receiver to get the timeline + timeline = self.state_handler.received_timeline() if not timeline: # So far the only way to get the current timeline on the standby is from # the replication connection. In order to avoid opening the replication diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 8fd158b9..ff3efa24 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -135,17 +135,25 @@ class Postgresql(object): @property def cluster_info_query(self): - pg_control_timeline = 'timeline_id FROM pg_catalog.pg_control_checkpoint()' \ - if self._major_version >= 90600 and self.role == 'standby_leader' else '0' + if self._major_version >= 90600: + extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END," + " slot_name, conninfo FROM pg_catalog.pg_stat_get_wal_receiver()") + if self.role == 'standby_leader': + extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()" + else: + extra = "0" + extra + else: + extra = "0, NULL, NULL, NULL" + return ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name(" "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, " "CASE WHEN pg_catalog.pg_is_in_recovery() THEN GREATEST(" " pg_catalog.pg_{0}_{1}_diff(COALESCE(" "pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint," - " pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint)" + " pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint) " "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint " - "END, {2}").format(self.wal_name, self.lsn_name, pg_control_timeline) + "END, {2}").format(self.wal_name, self.lsn_name, extra) def _version_file_exists(self): return not self.data_directory_empty() and os.path.isfile(self._version_file) @@ -290,7 +298,8 @@ class Postgresql(object): if not self._cluster_info_state: try: result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone() - self._cluster_info_state = dict(zip(['timeline', 'wal_position', 'pg_control_timeline'], result)) + self._cluster_info_state = dict(zip(['timeline', 'wal_position', 'pg_control_timeline', + 'received_tli', 'slot_name', 'conninfo'], result)) except RetryFailedError as e: # SELECT failed two times self._cluster_info_state = {'error': str(e)} if not self.is_starting() and self.pg_isready() == STATE_REJECT: @@ -301,6 +310,15 @@ class Postgresql(object): return self._cluster_info_state.get(name) + def primary_slot_name(self): + return self._cluster_info_state_get('slot_name') + + def primary_conninfo(self): + return self._cluster_info_state_get('conninfo') + + def received_timeline(self): + return self._cluster_info_state_get('received_tli') + def is_leader(self): return bool(self._cluster_info_state_get('timeline')) diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index fcab640e..2e16e1e4 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -652,6 +652,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 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: if self._check_passfile(primary_conninfo['passfile'], wanted_primary_conninfo): @@ -695,6 +706,13 @@ 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 + required = {'restart': 0, 'reload': 0} def record_missmatch(mtype): diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 1e6cd621..06a0e49c 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -99,8 +99,8 @@ class Rewind(object): end = None if i + 4 >= len(history) else i + 2 history_show = [] - def format_history_line(l): - return '{0}\t{1}\t{2}'.format(l[0], format_lsn(l[1]), l[2]) + def format_history_line(line): + return '{0}\t{1}\t{2}'.format(line[0], format_lsn(line[1]), line[2]) for line in history[start:end]: history_show.append(format_history_line(line)) diff --git a/tests/__init__.py b/tests/__init__.py index 2bf33b3d..ab67116b 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -88,7 +88,7 @@ class MockCursor(object): elif sql.startswith('SELECT slot_name'): self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b')] elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'): - self.results = [(1, 2, 1)] + self.results = [(1, 2, 1, 1, None, None)] elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'): self.results = [(False, 2)] elif sql.startswith('SELECT pg_catalog.to_char'): diff --git a/tests/test_ha.py b/tests/test_ha.py index 2597d416..a35641f3 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -202,6 +202,7 @@ class TestHa(PostgresInit): self.p.last_operation = Mock(side_effect=PostgresConnectionException('')) self.assertTrue(self.ha.update_lock(True)) + @patch.object(Postgresql, 'received_timeline', Mock(return_value=None)) def test_touch_member(self): self.p.timeline_wal_position = Mock(return_value=(0, 1, 0)) self.p.replica_cached_timeline = Mock(side_effect=Exception) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 9f52c0b9..adb92a32 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -250,6 +250,10 @@ class TestPostgresql(BaseTestPostgresql): self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': conninfo.copy()}) self.p.config.write_postgresql_conf() self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) + with patch.object(Postgresql, 'primary_conninfo', Mock(return_value='host=1')): + mock_get_pg_settings.return_value['primary_slot_name'] = [ + 'primary_slot_name', '', '', 'string', 'postmaster', self.p.config._postgresql_conf] + self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) @patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)) @patch.object(Postgresql, 'is_running', MockPostmaster) @@ -267,6 +271,7 @@ class TestPostgresql(BaseTestPostgresql): self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) @patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000)) + @patch.object(Postgresql, 'primary_conninfo', Mock(return_value='host=1')) def test__read_recovery_params_pre_v12(self): self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}}) self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) @@ -695,3 +700,9 @@ class TestPostgresql(BaseTestPostgresql): @patch('os.path.isfile', Mock(return_value=False)) def test_pgpass_is_dir(self): self.assertRaises(PatroniException, self.setUp) + + @patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))) + def test_received_timeline(self): + self.p.set_role('standby_leader') + self.p.reset_cluster_info_state() + self.assertRaises(PostgresConnectionException, self.p.received_timeline)