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

@@ -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

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'))
 

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):

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))

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'):

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)

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)