Repository: zalando/patroni
Pull request #2944 (2023-11-09 09:49:02)

Fix pg_rewind behavior with Postgres v16+
The error message format was changed in
https://github.com/postgres/postgres/commit/4ac30ba4f29d4b586b131404b0d514f16501272a, what caused `pg_rewind` being called by Patroni even when it was not necessary.

coveralls: ## Pull Request Test Coverage Report for [Build 6810059436](https://coveralls.io/builds/63852676)


* **7** of **7**   **(100.0%)**  changed or added relevant lines in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.858%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/63852676/badge)](https://coveralls.io/builds/63852676) |
| :-- | --: |
| Change from base [Build 6798679992](https://coveralls.io/builds/63829063): |  0.0% |
| Covered Lines: | 13342 |
| Relevant Lines: | 13361 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 389cae86..9c2f7335 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -1564,7 +1564,7 @@ class AbstractDCS(abc.ABC):
         """
 
     @abc.abstractmethod
-    def _citus_cluster_loader(self, path: Any) -> Union[Cluster, Dict[int, Cluster]]:
+    def _citus_cluster_loader(self, path: Any) -> Dict[int, Cluster]:
         """Load and build all Patroni clusters from a single Citus cluster.
 
         :param path: the path in DCS where to load Cluster(s) from.
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 58e20030..1324e606 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -422,7 +422,7 @@ class Consul(AbstractDCS):
     def _cluster_loader(self, path: str) -> Cluster:
         _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         if results is None:
-            raise NotFound
+            return Cluster.empty()
         nodes = {}
         for node in results:
             node['Value'] = (node['Value'] or b'').decode('utf-8')
@@ -445,8 +445,6 @@ class Consul(AbstractDCS):
     ) -> Union[Cluster, Dict[int, Cluster]]:
         try:
             return loader(path)
-        except NotFound:
-            return Cluster.empty()
         except Exception:
             logger.exception('get_cluster')
             raise ConsulError('Consul is not responding properly')
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index f242a6b2..3be699a6 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -710,13 +710,20 @@ class Etcd(AbstractEtcd):
         return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe)
 
     def _cluster_loader(self, path: str) -> Cluster:
-        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
+        try:
+            result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
+        except etcd.EtcdKeyNotFound:
+            return Cluster.empty()
         nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves}
         return self._cluster_from_nodes(result.etcd_index, nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
+        try:
+            result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
+        except etcd.EtcdKeyNotFound:
+            return {}
+
         clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict)
-        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         for node in result.leaves:
             key = node.key[len(result.key):].lstrip('/').split('/', 1)
             if len(key) == 2 and citus_group_re.match(key[0]):
@@ -729,8 +736,6 @@ class Etcd(AbstractEtcd):
         cluster = None
         try:
             cluster = loader(path)
-        except etcd.EtcdKeyNotFound:
-            cluster = Cluster.empty()
         except Exception as e:
             self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
         self._has_failed = False
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index 6e1aab88..4a5283f7 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -101,12 +101,26 @@ class Rewind(object):
             return 'not accessible or not healty'
 
     def _get_checkpoint_end(self, timeline: int, lsn: int) -> int:
-        """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 will 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."""
+        """Get the end of checkpoint record from WAL.
 
+        .. note::
+            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, is to 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 will fail to read the second record with message:
+
+                fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: wanted 24, got 0; or
+
+                fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: expected at least 24, got 0
+
+            The error message contains information about LSN of the next record, which is exactly where checkpoint ends.
+
+        :param timeline: the checkpoint *timeline* from ``pg_controldata``.
+        :param lsn: the checkpoint *location* as :class:`int` from ``pg_controldata``.
+
+        :returns: the end of checkpoint record as :class:`int` or ``0`` if failed to parse ``pg_waldump`` output.
+        """
         lsn8 = format_lsn(lsn, True)
         lsn_str = format_lsn(lsn)
         out, err = self._postgresql.waldump(timeline, lsn_str, 2)
@@ -117,12 +131,17 @@ class Rewind(object):
 
             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)
+                # Message format depends on the major version:
+                # * expected at least -- starting from v16
+                # * wanted -- before v16
+                # We will simply check all possible combinations.
+                for pattern in (': expected at least ', ': wanted '):
+                    j = err[0].find(pattern, 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 pg_%sdump output', self._postgresql.wal_name)
             logger.error(' stdout=%s', '\n'.join(out))
             logger.error(' stderr=%s', '\n'.join(err))


Test patch:
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index 90402b5f..874aac5c 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -274,6 +274,8 @@ class TestEtcd(unittest.TestCase):
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
+        self.etcd._base_path = '/service/nocluster'
+        self.assertTrue(self.etcd.get_cluster().is_empty())
 
     def test_touch_member(self):
         self.assertFalse(self.etcd.touch_member(''))
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index a54c27e9..af8e7d97 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -180,6 +180,16 @@ class TestRewind(BaseTestPostgresql):
             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))
+
+            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 '
+                b'length at 0/402DD98: expected at least 24, got 0\n'
+            )
+            self.r.reset_state()
+            self.r.trigger_check_diverged_lsn()
+            self.assertFalse(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