diff --git a/patroni/ha.py b/patroni/ha.py index dec4bdf4..8afe7aa2 100644 --- a/patroni/ha.py +++ b/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) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 4d196911..ce69c7fa 100644 --- a/patroni/postgresql/__init__.py +++ b/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: diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 06a0e49c..b90cf038 100644 --- a/patroni/postgresql/rewind.py +++ b/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 diff --git a/tests/test_ha.py b/tests/test_ha.py index 9dfa438d..c99152eb 100644 --- a/tests/test_ha.py +++ b/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) diff --git a/tests/test_rewind.py b/tests/test_rewind.py index 343de8b1..0f5c2805 100644 --- a/tests/test_rewind.py +++ b/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()