Improve timeline divergence check (#1563)

We don't need to rewind when:
1. replayed location for the former replica is not ahead of switchpoint
2. end of checkpoint record for the former primary is the same as switchpoint

In order to get the end of checkpoint record we use the `pg_waldump` and parse its output.

Close https://github.com/zalando/patroni/issues/1493

patroni/ha.py

@@ -421,6 +421,7 @@ class Ha(object):
                                                        self.state_handler.follow, args=(node_to_follow, role))
                 else:
                     self.state_handler.follow(node_to_follow, role, do_reload=True)
+                self._rewind.trigger_check_diverged_lsn()
             elif role == 'standby_leader' and self.state_handler.role != role:
                 self.state_handler.set_role(role)
                 self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)

patroni/postgresql/__init__.py

@@ -726,18 +726,13 @@ class Postgresql(object):
         with get_connection_cursor(**conn_kwargs) as cur:
             yield cur
 
-    def get_local_timeline_lsn_from_replication_connection(self):
-        timeline = lsn = None
+    def get_replica_timeline(self):
         try:
             with self.get_replication_connection_cursor(**self.config.local_replication_address) as cur:
                 cur.execute('IDENTIFY_SYSTEM')
-                timeline, lsn = cur.fetchone()[1:3]
+                return cur.fetchone()[1]
         except Exception:
             logger.exception('Can not fetch local timeline and lsn from replication connection')
-        return timeline, lsn
-
-    def get_replica_timeline(self):
-        return self.get_local_timeline_lsn_from_replication_connection()[0]
 
     def replica_cached_timeline(self, master_timeline):
         if not self._cached_replica_timeline or not master_timeline or self._cached_replica_timeline != master_timeline:

patroni/postgresql/rewind.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import six
 import subprocess
 
 from threading import Lock, Thread
@@ -69,29 +70,81 @@ class Rewind(object):
         except Exception:
             return logger.exception('Exception when working with leader')
 
+    def _get_checkpoint_end(self, timeline, lsn):
+        """The checkpoint record size in WAL depends on postgres major version and platform (memory alignment).
+        Hence, the only reliable way to figure out where it ends, read the record from file with the help of pg_waldump
+        and parse the output. We are trying to read two records, and expect that it wil fail to read the second one:
+        `pg_waldump: fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: wanted 24, got 0`
+        The error message contains information about LSN of the next record, which is exactly where checkpoint ends."""
+
+        cmd = self._postgresql.pgcommand('pg_{0}dump'.format(self._postgresql.wal_name))
+        lsn8 = format_lsn(lsn, True)
+        lsn = format_lsn(lsn)
+        env = os.environ.copy()
+        env.update(LANG='C', LC_ALL='C', PGDATA=self._postgresql.data_dir)
+        try:
+            waldump = subprocess.Popen([cmd, '-t', str(timeline), '-s', lsn, '-n', '2'],
+                                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+            out, err = waldump.communicate()
+            waldump.wait()
+        except Exception as e:
+            logger.error('Failed to execute `%s -t %s -s %s -n 2`: %r', cmd, timeline, lsn, e)
+        else:
+            out = out.decode('utf-8').rstrip().split('\n')
+            err = err.decode('utf-8').rstrip().split('\n')
+            pattern = 'error in WAL record at {0}: invalid record length at '.format(lsn)
+
+            if len(out) == 1 and len(err) == 1 and ', lsn: {0}, prev '.format(lsn8) in out[0] and pattern in err[0]:
+                i = err[0].find(pattern) + len(pattern)
+                j = err[0].find(": wanted ", i)
+                if j > -1:
+                    try:
+                        return parse_lsn(err[0][i:j])
+                    except Exception as e:
+                        logger.error('Failed to parse lsn %s: %r', err[0][i:j], e)
+            logger.error('Failed to parse `%s -t %s -s %s -n 2` output', cmd, timeline, lsn)
+            logger.error(' stdout=%s', '\n'.join(out))
+            logger.error(' stderr=%s', '\n'.join(err))
+
+        return 0
+
     def _get_local_timeline_lsn_from_controldata(self):
-        timeline = lsn = None
+        in_recovery = timeline = lsn = None
         data = self._postgresql.controldata()
         try:
             if data.get('Database cluster state') == 'shut down in recovery':
+                in_recovery = True
                 lsn = data.get('Minimum recovery ending location')
                 timeline = int(data.get("Min recovery ending loc's timeline"))
                 if lsn == '0/0' or timeline == 0:  # it was a master when it crashed
                     data['Database cluster state'] = 'shut down'
             if data.get('Database cluster state') == 'shut down':
+                in_recovery = False
                 lsn = data.get('Latest checkpoint location')
                 timeline = int(data.get("Latest checkpoint's TimeLineID"))
         except (TypeError, ValueError):
             logger.exception('Failed to get local timeline and lsn from pg_controldata output')
-        return timeline, lsn
+
+        if lsn is not None:
+            try:
+                lsn = parse_lsn(lsn)
+            except (IndexError, ValueError) as e:
+                logger.error('Exception when parsing lsn %s: %r', lsn, e)
+                lsn = None
+
+        return in_recovery, timeline, lsn
 
     def _get_local_timeline_lsn(self):
-        if self._postgresql.is_running():  # if postgres is running - get timeline and lsn from replication connection
-            timeline, lsn = self._postgresql.get_local_timeline_lsn_from_replication_connection()
+        if self._postgresql.is_running():  # if postgres is running - get timeline from replication connection
+            in_recovery = True
+            timeline = self._postgresql.received_timeline() or self._postgresql.get_replica_timeline()
+            lsn = self._postgresql.replayed_location()
         else:  # otherwise analyze pg_controldata output
-            timeline, lsn = self._get_local_timeline_lsn_from_controldata()
-        logger.info('Local timeline=%s lsn=%s', timeline, lsn)
-        return timeline, lsn
+            in_recovery, timeline, lsn = self._get_local_timeline_lsn_from_controldata()
+
+        log_lsn = format_lsn(lsn) if isinstance(lsn, six.integer_types) else lsn
+        logger.info('Local timeline=%s lsn=%s', timeline, log_lsn)
+        return in_recovery, timeline, lsn
 
     @staticmethod
     def _log_master_history(history, i):
@@ -112,7 +165,7 @@ class Rewind(object):
         logger.info('master: history=%s', '\n'.join(history_show))
 
     def _check_timeline_and_lsn(self, leader):
-        local_timeline, local_lsn = self._get_local_timeline_lsn()
+        in_recovery, local_timeline, local_lsn = self._get_local_timeline_lsn()
         if local_timeline is None or local_lsn is None:
             return
 
@@ -144,10 +197,15 @@ class Rewind(object):
             history = list(parse_history(history))
             for i, (parent_timeline, switchpoint, _) in enumerate(history):
                 if parent_timeline == local_timeline:
-                    try:
-                        need_rewind = parse_lsn(local_lsn) >= switchpoint
-                    except (IndexError, ValueError):
-                        logger.exception('Exception when parsing lsn')
+                    # We don't need to rewind when:
+                    # 1. for replica: replayed location is not ahead of switchpoint
+                    # 2. for the former primary: end of checkpoint record is the same as switchpoint
+                    if in_recovery:
+                        need_rewind = local_lsn > switchpoint
+                    elif local_lsn >= switchpoint:
+                        need_rewind = True
+                    else:
+                        need_rewind = switchpoint != self._get_checkpoint_end(local_timeline, local_lsn)
                     break
                 elif parent_timeline > local_timeline:
                     break

tests/test_ha.py

@@ -167,7 +167,7 @@ def run_async(self, func, args=()):
 @patch.object(Postgresql, 'query', Mock())
 @patch.object(Postgresql, 'checkpoint', Mock())
 @patch.object(CancellableSubprocess, 'call', Mock(return_value=0))
-@patch.object(Postgresql, 'get_local_timeline_lsn_from_replication_connection', Mock(return_value=[2, 10]))
+@patch.object(Postgresql, 'get_replica_timeline', Mock(return_value=2))
 @patch.object(Postgresql, 'get_master_timeline', Mock(return_value=2))
 @patch.object(ConfigHandler, 'restore_configuration_files', Mock())
 @patch.object(etcd.Client, 'write', etcd_write)

tests/test_rewind.py

@@ -53,11 +53,12 @@ class TestRewind(BaseTestPostgresql):
         with patch.object(Postgresql, 'controldata',
                           Mock(return_value={'Database cluster state': 'shut down in recovery',
                                              'Minimum recovery ending location': '0/0',
-                                             "Min recovery ending loc's timeline": '0'})):
+                                             "Min recovery ending loc's timeline": '0',
+                                             'Latest checkpoint location': '0/'})):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
         with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
-            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(False, ), Exception])):
+            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1,), Exception])):
                 self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
     @patch.object(CancellableSubprocess, 'call', Mock(return_value=0))
@@ -95,9 +96,10 @@ class TestRewind(BaseTestPostgresql):
 
     @patch.object(Postgresql, 'start', Mock())
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
-    @patch.object(Rewind, '_get_local_timeline_lsn', Mock(return_value=(2, '40159C1')))
+    @patch.object(Rewind, '_get_local_timeline_lsn')
     @patch.object(Rewind, 'check_leader_is_not_in_recovery')
-    def test__check_timeline_and_lsn(self, mock_check_leader_is_not_in_recovery):
+    def test__check_timeline_and_lsn(self, mock_check_leader_is_not_in_recovery, mock_get_local_timeline_lsn):
+        mock_get_local_timeline_lsn.return_value = (True, 2, 67197377)
         mock_check_leader_is_not_in_recovery.return_value = False
         self.r.trigger_check_diverged_lsn()
         self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
@@ -113,11 +115,30 @@ class TestRewind(BaseTestPostgresql):
             self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
         self.r.trigger_check_diverged_lsn()
         with patch.object(MockCursor, 'fetchone', Mock(return_value=('', 1, '0/0'))):
-            with patch.object(Rewind, '_get_local_timeline_lsn', Mock(return_value=(1, '0/0'))):
+            with patch.object(Rewind, '_get_local_timeline_lsn', Mock(return_value=(True, 1, '0/0'))):
                 self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
             self.r.trigger_check_diverged_lsn()
             self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
 
+        self.r.reset_state()
+        self.r.trigger_check_diverged_lsn()
+        mock_get_local_timeline_lsn.return_value = (False, 2, 67296664)
+        self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+
+        with patch('subprocess.Popen') as mock_popen:
+            mock_popen.return_value.communicate.return_value = (
+                b'0, lsn: 0/040159C1, prev 0/\n',
+                b'pg_waldump: fatal: error in WAL record at 0/40159C1: invalid record length at /: wanted 24, got 0\n'
+            )
+            self.r.reset_state()
+            self.r.trigger_check_diverged_lsn()
+            mock_get_local_timeline_lsn.return_value = (False, 2, 67197377)
+            self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+            self.r.reset_state()
+            self.r.trigger_check_diverged_lsn()
+            mock_popen.side_effect = Exception
+            self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+
     @patch.object(MockCursor, 'fetchone', Mock(side_effect=[(True,), Exception]))
     def test_check_leader_is_not_in_recovery(self):
         self.r.check_leader_is_not_in_recovery()