From 0ceb59b49d677e1cc2621ee7d2e9b3c22a473bcb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 5 Jul 2021 09:29:39 +0200 Subject: [PATCH] Write prev LSN to before checkpoint to optime if wal_achive=on (#1889) The #1527 introduced a feature of updating `/optime/leader` with the location of the last checkpoint after the Postgres was shutdown cleanly. If wal archiving is enabled, Postgres always switching the WAL file before writing the checkpoint shutdown record. Normally it is not an issue, but for databases without too much write activity it could lead to the situation that the visible replication lag becomes equal to the size of a single WAL file. In fact, the previous WAL file is mostly empty and contains only a few records. Therefore it should be safe to report the LSN of the SWITCH record before the shutdown checkpoint. In order to do that, Patroni first gets the output of the pg_controldata and based on it calls pg_waldump two times: * The first call reads the checkpoint record (and verifies that this is really the shutdown checkpoint). * The next call reads the previous record and in case if it is the 'xlog switch' (for 9.3 and 9.4) or 'SWITCH' (for 9.5+), the LSN of the SWITCH record is written to the `/optime/leader`. In case of any mismatch, failure to call pg_waldump or parse its output, the old behavior is retained, i.e. `Latest checkpoint location` from the pg_controldata is used. Close https://github.com/zalando/patroni/issues/1860 --- patroni/postgresql/__init__.py | 57 +++++++++++++++++++++++++++++----- patroni/postgresql/rewind.py | 15 ++------- tests/test_ha.py | 3 +- tests/test_postgresql.py | 35 ++++++++++++++++++--- 4 files changed, 85 insertions(+), 25 deletions(-) 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())