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

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

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

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={}))

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