Repository: zalando/patroni
Pull request #3072 (2024-05-31 14:06:06)

Fix infinite recursion in in replicatefrom tags
Besides that:
1. fix problem with is_physical_slot() methods, it was returning false positives for logical slots.
2. Fix a little issue with replicatefrom docs.

Close https://github.com/zalando/patroni/issues/3068

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


### Details

* **14** of **14**   **(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.862%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/67816938/badge)](https://coveralls.io/builds/67816938) |
| :-- | --: |
| Change from base [Build 9220595698](https://coveralls.io/builds/67684598): |  0.0% |
| Covered Lines: | 13764 |
| Relevant Lines: | 13783 |

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


Patch:
diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst
index 4b45798f..d6c40116 100644
--- a/docs/ENVIRONMENT.rst
+++ b/docs/ENVIRONMENT.rst
@@ -80,7 +80,7 @@ Etcdv3
 Environment names for Etcdv3 are similar as for Etcd, you just need to use ``ETCD3`` instead of ``ETCD`` in the variable name. Example: ``PATRONI_ETCD3_HOST``, ``PATRONI_ETCD3_CACERT``, and so on.
 
 .. warning::
-    Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from Etcd to Etcdv3 just by updating Patroni configuration.
+    Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from Etcd to Etcdv3 just by updating Patroni configuration. In addition, Patroni uses Etcd's gRPC-gateway (proxy) to communicate with the V3 API, which means that TLS common name authentication is not possible.
 
 
 ZooKeeper
diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst
index 0929404d..adedc00f 100644
--- a/docs/yaml_configuration.rst
+++ b/docs/yaml_configuration.rst
@@ -144,7 +144,7 @@ Etcdv3
 If you want that Patroni works with Etcd cluster via protocol version 3, you need to use the ``etcd3`` section in the Patroni configuration file. All configuration parameters are the same as for ``etcd``.
 
 .. warning::
-    Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from ``etcd`` to ``etcd3`` just by updating Patroni config file.
+    Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from ``etcd`` to ``etcd3`` just by updating Patroni config file. In addition, Patroni uses Etcd's gRPC-gateway (proxy) to communicate with the V3 API, which means that TLS common name authentication is not possible.
 
 
 ZooKeeper
@@ -395,7 +395,7 @@ Tags
 ----
 -  **clonefrom**: ``true`` or ``false``. If set to ``true`` other nodes might prefer to use this node for bootstrap (take ``pg_basebackup`` from). If there are several nodes with ``clonefrom`` tag set to ``true`` the node to bootstrap from will be chosen randomly. The default value is ``false``.
 -  **noloadbalance**: ``true`` or ``false``. If set to ``true`` the node will return HTTP Status Code 503 for the ``GET /replica`` REST API health-check and therefore will be excluded from the load-balancing. Defaults to ``false``.
--  **replicatefrom**: The IP address/hostname of another replica. Used to support cascading replication.
+-  **replicatefrom**: The name of another replica to replicate from. Used to support cascading replication.
 -  **nosync**: ``true`` or ``false``. If set to ``true`` the node will never be selected as a synchronous replica.
 -  **nofailover**: ``true`` or ``false``, controls whether this node is allowed to participate in the leader race and become a leader. Defaults to ``false``, meaning this node _can_ participate in leader races. 
 -  **failover_priority**: integer, controls the priority that this node should have during failover. Nodes with higher priority will be preferred over lower priority nodes if they received/replayed the same amount of WAL. However, nodes with higher values of receive/replay LSN are preferred regardless of their priority. If the ``failover_priority`` is 0 or negative - such node is not allowed to participate in the leader race and to become a leader (similar to ``nofailover: true``).
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 971cc3c9..33a39464 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -10,7 +10,7 @@ from copy import deepcopy
 from random import randint
 from threading import Event, Lock
 from typing import Any, Callable, Collection, Dict, Iterator, List, \
-    NamedTuple, Optional, Tuple, Type, TYPE_CHECKING, Union
+    NamedTuple, Optional, Set, Tuple, Type, TYPE_CHECKING, Union
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 import dateutil.parser
@@ -913,7 +913,9 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: ``True`` if *value* is a physical replication slot, otherwise ``False``.
         """
-        return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'
+        return not value \
+            or (isinstance(value, dict) and not Cluster.is_logical_slot(value)
+                and value.get('type', 'physical') == 'physical')
 
     @staticmethod
     def is_logical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
@@ -1179,6 +1181,10 @@ class Cluster(NamedTuple('Cluster',
 
         if global_config.use_slots:
             name = member.name if isinstance(member, Member) else postgresql.name
+
+            if not self.get_slot_name_on_primary(name, member):
+                return False
+
             members = [m for m in self.members if m.replicatefrom == name and m.name != self.leader_name]
             return any(self.should_enforce_hot_standby_feedback(postgresql, m) for m in members)
         return False
@@ -1197,11 +1203,18 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: the slot name on the primary that is in use for physical replication on this node.
         """
-        if tags.nostream:
-            return None
-        replicatefrom = self.get_member(tags.replicatefrom, False) if tags.replicatefrom else None
-        return self.get_slot_name_on_primary(replicatefrom.name, replicatefrom) \
-            if isinstance(replicatefrom, Member) else slot_name_from_member_name(name)
+        seen_nodes: Set[str] = set()
+        while True:
+            seen_nodes.add(name)
+            if tags.nostream:
+                return None
+            replicatefrom = self.get_member(tags.replicatefrom, False) \
+                if tags.replicatefrom and tags.replicatefrom != name else None
+            if not isinstance(replicatefrom, Member):
+                return slot_name_from_member_name(name)
+            if replicatefrom.name in seen_nodes:
+                return None
+            name, tags = replicatefrom.name, replicatefrom
 
     @property
     def timeline(self) -> int:
diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py
index 0f3c4ba5..47dbacfe 100644
--- a/patroni/postgresql/bootstrap.py
+++ b/patroni/postgresql/bootstrap.py
@@ -216,15 +216,13 @@ class Bootstrap(object):
         cmd = config.get('post_bootstrap') or config.get('post_init')
         if cmd:
             r = self._postgresql.connection_pool.conn_kwargs
-            connstring = self._postgresql.config.format_dsn(r, True)
-            if 'host' not in r:
-                # https://www.postgresql.org/docs/current/static/libpq-pgpass.html
-                # A host name of localhost matches both TCP (host name localhost) and Unix domain socket
-                # (pghost empty or the default socket directory) connections coming from the local machine.
-                r['host'] = 'localhost'  # set it to localhost to write into pgpass
 
-            env = self._postgresql.config.write_pgpass(r)
+            # https://www.postgresql.org/docs/current/static/libpq-pgpass.html
+            # A host name of localhost matches both TCP (host name localhost) and Unix domain socket
+            # (pghost empty or the default socket directory) connections coming from the local machine.
+            env = self._postgresql.config.write_pgpass({'host': 'localhost', **r})
             env['PGOPTIONS'] = '-c synchronous_commit=local -c statement_timeout=0'
+            connstring = self._postgresql.config.format_dsn({**r, 'password': None})
 
             try:
                 ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env)
@@ -258,7 +256,7 @@ class Bootstrap(object):
             r = clone_member.conn_kwargs(self._postgresql.config.replication)
             # add the credentials to connect to the replica origin to pgpass.
             env = self._postgresql.config.write_pgpass(r)
-            connstring = self._postgresql.config.format_dsn(r, True)
+            connstring = self._postgresql.config.format_dsn({**r, 'password': None})
         else:
             connstring = ''
             env = os.environ.copy()
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 3755a258..ef93e1f3 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -574,27 +574,26 @@ class ConfigHandler(object):
             del ret['dbname']
         return ret
 
-    def format_dsn(self, params: Dict[str, Any], include_dbname: bool = False) -> str:
+    def format_dsn(self, params: Dict[str, Any]) -> str:
+        """Format connection string from connection parameters.
+
+        .. note::
+            only parameters from the below list are considered and values are escaped.
+
+        :param params: :class:`dict` object with connection parameters.
+
+        :returns: a connection string in a format "key1=value2 key2=value2"
+        """
         # A list of keywords that can be found in a conninfo string. Follows what is acceptable by libpq
         keywords = ('dbname', 'user', 'passfile' if params.get('passfile') else 'password', 'host', 'port',
                     'sslmode', 'sslcompression', 'sslcert', 'sslkey', 'sslpassword', 'sslrootcert', 'sslcrl',
                     'sslcrldir', 'application_name', 'krbsrvname', 'gssencmode', 'channel_binding',
                     'target_session_attrs')
-        if include_dbname:
-            params = params.copy()
-            if 'dbname' not in params:
-                params['dbname'] = self._postgresql.database
-            # we are abusing information about the necessity of dbname
-            # dsn should contain passfile or password only if there is no dbname in it (it is used in recovery.conf)
-            skip = {'passfile', 'password'}
-        else:
-            skip = {'dbname'}
 
         def escape(value: Any) -> str:
             return re.sub(r'([\'\\ ])', r'\\\1', str(value))
 
-        return ' '.join('{0}={1}'.format(kw, escape(params[kw])) for kw in keywords
-                        if kw not in skip and params.get(kw) is not None)
+        return ' '.join('{0}={1}'.format(kw, escape(params[kw])) for kw in keywords if params.get(kw) is not None)
 
     def _write_recovery_params(self, fd: ConfigWriter, recovery_params: CaseInsensitiveDict) -> None:
         if self._postgresql.major_version >= 90500:
@@ -606,8 +605,7 @@ class ConfigHandler(object):
                 recovery_params.setdefault('pause_at_recovery_target', 'false')
         for name, value in sorted(recovery_params.items()):
             if name == 'primary_conninfo':
-                if 'password' in value and self._postgresql.major_version >= 100000:
-                    self.write_pgpass(value)
+                if self._postgresql.major_version >= 100000 and 'PGPASSFILE' in self.write_pgpass(value):
                     value['passfile'] = self._passfile = self._pgpass
                     self._passfile_mtime = mtime(self._pgpass)
                 value = self.format_dsn(value)
@@ -754,7 +752,7 @@ class ConfigHandler(object):
         if passfile_mtime:
             try:
                 with open(passfile) as f:
-                    wanted_lines = (self._pgpass_line(wanted_primary_conninfo) or '').splitlines()
+                    wanted_lines = (self._pgpass_content(wanted_primary_conninfo) or '').splitlines()
                     file_lines = f.read().splitlines()
                     if set(wanted_lines) == set(file_lines):
                         self._passfile = passfile
@@ -873,27 +871,38 @@ class ConfigHandler(object):
             os.unlink(name)
 
     @staticmethod
-    def _pgpass_line(record: Dict[str, Any]) -> Optional[str]:
+    def _pgpass_content(record: Dict[str, Any]) -> Optional[str]:
+        """Generate content of `pgpassfile` based on connection parameters.
+
+        .. note::
+            In case if ``host`` is a comma separated string we generate one line per host.
+
+        :param record: :class:`dict` object with connection parameters.
+        :returns: a string with generated content of pgpassfile or ``None`` if there is no ``password``.
+        """
         if 'password' in record:
             def escape(value: Any) -> str:
                 return re.sub(r'([:\\])', r'\\\1', str(value))
 
-            record = {n: escape(record.get(n) or '*') for n in ('host', 'port', 'user', 'password')}
-            # 'host' could be several comma-separated hostnames, in this case
-            # we need to write on pgpass line per host
-            line = ''
-            for hostname in record['host'].split(','):
-                line += hostname + ':{port}:*:{user}:{password}'.format(**record) + '\n'
-            return line.rstrip()
+            # 'host' could be several comma-separated hostnames, in this case we need to write on pgpass line per host
+            hosts = map(escape, filter(None, map(str.strip, (record.get('host') or '*').split(','))))
+            record = {n: escape(record.get(n) or '*') for n in ('port', 'user', 'password')}
+            return '\n'.join('{host}:{port}:*:{user}:{password}'.format(**record, host=host) for host in hosts)
 
     def write_pgpass(self, record: Dict[str, Any]) -> Dict[str, str]:
-        line = self._pgpass_line(record)
-        if not line:
+        """Maybe creates :attr:`_passfile` based on connection parameters.
+
+        :param record: :class:`dict` object with connection parameters.
+
+        :returns: a copy of environment variables, that will include ``PGPASSFILE`` in case if the file was written.
+        """
+        content = self._pgpass_content(record)
+        if not content:
             return os.environ.copy()
 
         with open(self._pgpass, 'w') as f:
             os.chmod(self._pgpass, stat.S_IWRITE | stat.S_IREAD)
-            f.write(line)
+            f.write(content)
 
         return {**os.environ, 'PGPASSFILE': self._pgpass}
 
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index c3461630..91cddd59 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -412,11 +412,22 @@ class Rewind(object):
             except Exception as e:
                 logger.warning('Unable to clean %s: %r', replslot_dir, e)
 
-    def pg_rewind(self, r: Dict[str, Any]) -> bool:
-        # prepare pg_rewind connection
-        env = self._postgresql.config.write_pgpass(r)
+    def pg_rewind(self, conn_kwargs: Dict[str, Any]) -> bool:
+        """Do pg_rewind.
+
+        .. note::
+            If ``pg_rewind`` doesn't support ``--restore-target-wal`` parameter and exited with non zero code,
+            Patroni will parse stderr/stdout to figure out if it failed due to a missing WAL file and will
+            repeat an attempt after downloading the missing file using ``restore_command``.
+
+        :param conn_kwargs: :class:`dict` object with connection parameters.
+
+        :returns: ``True`` if ``pg_rewind`` finished successfully, ``False`` otherwise.
+        """
+        # prepare pg_rewind connection string
+        env = self._postgresql.config.write_pgpass(conn_kwargs)
         env.update(LANG='C', LC_ALL='C', PGOPTIONS='-c statement_timeout=0')
-        dsn = self._postgresql.config.format_dsn(r, True)
+        dsn = self._postgresql.config.format_dsn({**conn_kwargs, 'password': None})
         logger.info('running pg_rewind from %s', dsn)
 
         restore_command = (self._postgresql.config.get('recovery_conf') or EMPTY_DICT).get('restore_command') \


Test patch:
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index e7ef224f..ebfa1ebe 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -103,7 +103,10 @@ class TestRewind(BaseTestPostgresql):
     @patch.object(Postgresql, 'stop', Mock(return_value=False))
     @patch.object(Postgresql, 'start', Mock())
     def test_execute(self, mock_checkpoint):
-        self.r.execute(self.leader)
+        with patch('patroni.postgresql.rewind.logger.info') as mock_logger:
+            self.r.execute(self.leader)
+            self.assertEqual(mock_logger.call_args_list[0][0],
+                             ('running pg_rewind from %s', 'dbname=postgres user=foo host=127.0.0.1 port=5435'))
         with patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000)):
             self.r.execute(self.leader)
             with patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
diff --git a/tests/test_slots.py b/tests/test_slots.py
index a7112f81..71758581 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -186,6 +186,32 @@ class TestSlotsHandler(BaseTestPostgresql):
             cluster.get_slot_name_on_primary(self.p.name, stream_node),
             'test_4')
 
+    def test_get_slot_name_on_primary(self):
+        node1 = Member(0, 'node1', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres',
+            'tags': {'replicatefrom': 'node2'}
+        })
+        node2 = Member(0, 'node2', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres',
+            'tags': {'replicatefrom': 'node1'}
+        })
+        cluster = Cluster(True, None, self.leader, Status.empty(), [self.leadermem, node1, node2],
+                          None, SyncState.empty(), None, None)
+        self.assertIsNone(cluster.get_slot_name_on_primary('node1', node1))
+
+    def test_should_enforce_hot_standby_feedback(self):
+        node1 = Member(0, 'postgresql0', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres',
+            'tags': {'replicatefrom': 'postgresql1'}
+        })
+        node2 = Member(0, 'postgresql1', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres',
+            'tags': {'replicatefrom': 'postgresql0'}
+        })
+        cluster = Cluster(True, None, self.leader, Status.empty(), [self.leadermem, node1, node2],
+                          None, SyncState.empty(), None, None)
+        self.assertFalse(cluster.should_enforce_hot_standby_feedback(self.p, node1))
+
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')