diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 13055dbe..b7059d5b 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -1,20 +1,22 @@ import logging import os import psycopg2 +import re import shlex import shutil +import six import subprocess import time from contextlib import contextmanager from copy import deepcopy -from dateutil import tz from datetime import datetime +from dateutil import tz from psutil import TimeoutExpired from threading import current_thread, Lock -from .callback_executor import CallbackExecutor from .bootstrap import Bootstrap +from .callback_executor import CallbackExecutor from .cancellable import CancellableSubprocess from .config import ConfigHandler, mtime from .connection import Connection, get_connection_cursor @@ -386,16 +388,41 @@ class Postgresql(object): except (TypeError, ValueError): logger.exception('Failed to parse timeline from pg_controldata output') + def parse_wal_record(self, timeline, lsn): + out, err = self.waldump(timeline, lsn, 1) + if out and not err: + match = re.match(r'^rmgr:\s+(.+?)\s+len \(rec/tot\):\s+\d+/\s+\d+, tx:\s+\d+, ' + r'lsn: ([0-9A-Fa-f]+/[0-9A-Fa-f]+), prev ([0-9A-Fa-f]+/[0-9A-Fa-f]+), ' + r'.*?desc: (.+)', out.decode('utf-8')) + if match: + return match.groups() + return None, None, None, None + def latest_checkpoint_location(self): - """Returns checkpoint location for the cleanly shut down primary""" + """Returns checkpoint location for the cleanly shut down primary. + But, if we know that the checkpoint was written to the new WAL + due to the archive_mode=on, we will return the LSN of prev wal record (SWITCH).""" data = self.controldata() - lsn = data.get('Latest checkpoint location') - if data.get('Database cluster state') == 'shut down' and lsn: + timeline = data.get("Latest checkpoint's TimeLineID") + lsn = checkpoint_lsn = data.get('Latest checkpoint location') + if data.get('Database cluster state') == 'shut down' and lsn and timeline: try: - return str(parse_lsn(lsn)) - except (IndexError, ValueError) as e: - logger.error('Exception when parsing lsn %s: %r', lsn, e) + checkpoint_lsn = parse_lsn(checkpoint_lsn) + rm_name, lsn, prev, desc = self.parse_wal_record(timeline, lsn) + desc = desc.strip().lower() + if rm_name == 'XLOG' and parse_lsn(lsn) == checkpoint_lsn and prev and\ + desc.startswith('checkpoint') and desc.endswith('shutdown'): + _, lsn, _, desc = self.parse_wal_record(timeline, prev) + prev = parse_lsn(prev) + # If the cluster is shutdown with archive_mode=on, WAL is switched before writing the checkpoint. + # In this case we want to take the LSN of previous record (switch) as the last known WAL location. + if parse_lsn(lsn) == prev and desc.strip() in ('xlog switch', 'SWITCH'): + return str(prev) + except Exception as e: + logger.error('Exception when parsing WAL pg_%sdump output: %r', self.wal_name, e) + if isinstance(checkpoint_lsn, six.integer_types): + return str(checkpoint_lsn) def is_running(self): """Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process @@ -767,6 +794,20 @@ class Postgresql(object): logger.exception("Error when calling pg_controldata") return {} + def waldump(self, timeline, lsn, limit): + cmd = self.pgcommand('pg_{0}dump'.format(self.wal_name)) + env = os.environ.copy() + env.update(LANG='C', LC_ALL='C', PGDATA=self._data_dir) + try: + waldump = subprocess.Popen([cmd, '-t', str(timeline), '-s', str(lsn), '-n', str(limit)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + out, err = waldump.communicate() + waldump.wait() + return out, err + except Exception as e: + logger.error('Failed to execute `%s -t %s -s %s -n %s`: %r', cmd, timeline, lsn, limit, e) + return None, None + @contextmanager def get_replication_connection_cursor(self, host='localhost', port=5432, **kwargs): conn_kwargs = self.config.replication.copy() diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 4f40f189..03784428 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -72,19 +72,10 @@ class Rewind(object): `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, err = self._postgresql.waldump(timeline, lsn, 2) + if out is not None and err is not None: 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) @@ -97,7 +88,7 @@ class Rewind(object): 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('Failed to parse pg_%sdump output', self._postgresql.wal_name) logger.error(' stdout=%s', '\n'.join(out)) logger.error(' stderr=%s', '\n'.join(err)) diff --git a/tests/test_ha.py b/tests/test_ha.py index 68a9a4ea..726addc3 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -157,7 +157,8 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'controldata', Mock(return_value={ 'Database system identifier': SYSID, 'Database cluster state': 'shut down', - 'Latest checkpoint location': '0/12345678'})) + 'Latest checkpoint location': '0/12345678', + "Latest checkpoint's TimeLineID": '2'})) @patch.object(SlotsHandler, 'load_replication_slots', Mock(side_effect=Exception)) @patch.object(ConfigHandler, 'append_pg_hba', Mock()) @patch.object(ConfigHandler, 'write_pgpass', Mock(return_value={})) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 54b5c674..96f0b5bc 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -322,10 +322,37 @@ class TestPostgresql(BaseTestPostgresql): with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))): self.assertRaises(PostgresConnectionException, self.p.is_leader) - @patch.object(Postgresql, 'controldata', - Mock(return_value={'Database cluster state': 'shut down', 'Latest checkpoint location': 'X/678'})) - def test_latest_checkpoint_location(self): - self.assertIsNone(self.p.latest_checkpoint_location()) + @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down', + 'Latest checkpoint location': '0/1ADBC18', + "Latest checkpoint's TimeLineID": '1'})) + @patch('subprocess.Popen') + def test_latest_checkpoint_location(self, mock_popen): + mock_popen.return_value.communicate.return_value = (None, None) + self.assertEqual(self.p.latest_checkpoint_location(), '28163096') + # 9.3 and 9.4 format + mock_popen.return_value.communicate.side_effect = [ + (b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' + + b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' + + b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None), + (b'rmgr: Transaction len (rec/tot): 64/ 96, tx: 726, lsn: 0/01ADBBB8, prev 0/01ADBB70, ' + + b'bkp: 0000, desc: commit: 2021-02-26 11:19:37.900918 CET; inval msgs: catcache 11 catcache 10', None)] + self.assertEqual(self.p.latest_checkpoint_location(), '28163096') + mock_popen.return_value.communicate.side_effect = [ + (b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' + + b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' + + b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None), + (b'rmgr: XLOG len (rec/tot): 0/ 32, tx: 0, lsn: 0/01ADBBB8, prev 0/01ADBBA0, ' + + b'bkp: 0000, desc: xlog switch ', None)] + self.assertEqual(self.p.latest_checkpoint_location(), '28163000') + # 9.5+ format + mock_popen.return_value.communicate.side_effect = [ + (b'rmgr: XLOG len (rec/tot): 114/ 114, tx: 0, lsn: 0/01ADBC18, prev 0/018260F8, ' + + b'desc: CHECKPOINT_SHUTDOWN redo 0/1825ED8; tli 1; prev tli 1; fpw true; xid 0:494; oid 16387; multi 1' + + b'; offset 0; oldest xid 479 in DB 1; oldest multi 1 in DB 1; oldest/newest commit timestamp xid: 0/0;' + + b' oldest running xid 0; shutdown', None), + (b'rmgr: XLOG len (rec/tot): 24/ 24, tx: 0, lsn: 0/018260F8, prev 0/01826080, ' + + b'desc: SWITCH ', None)] + self.assertEqual(self.p.latest_checkpoint_location(), '25321720') def test_reload(self): self.assertTrue(self.p.reload())