Repository: zalando/patroni
Pull request #2943 (2023-11-08 11:26:10)

Fix Etcd v2 with Citus
When deploying a new Citus cluster with Etcd v2 Patroni was failing to start with the following exception:
```python
2023-11-09 10:51:41,246 INFO: Selected new etcd server http://localhost:2379
Traceback (most recent call last):
  File "/home/akukushkin/git/patroni/./patroni.py", line 6, in <module>
    main()
  File "/home/akukushkin/git/patroni/patroni/__main__.py", line 343, in main
    return patroni_main(args.configfile)
  File "/home/akukushkin/git/patroni/patroni/__main__.py", line 237, in patroni_main
    abstract_main(Patroni, configfile)
  File "/home/akukushkin/git/patroni/patroni/daemon.py", line 172, in abstract_main
    controller = cls(config)
  File "/home/akukushkin/git/patroni/patroni/__main__.py", line 66, in __init__
    self.ensure_unique_name()
  File "/home/akukushkin/git/patroni/patroni/__main__.py", line 112, in ensure_unique_name
    cluster = self.dcs.get_cluster()
  File "/home/akukushkin/git/patroni/patroni/dcs/__init__.py", line 1654, in get_cluster
    cluster = self._get_citus_cluster() if self.is_citus_coordinator() else self.__get_patroni_cluster()
  File "/home/akukushkin/git/patroni/patroni/dcs/__init__.py", line 1638, in _get_citus_cluster
    cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
AttributeError: 'Cluster' object has no attribute 'pop'
```

It is broken since #2909.

In addition to that fix `_citus_cluster_loader()` interface by allowing it to return only dict obj.

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


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

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/63826689/badge)](https://coveralls.io/builds/63826689) |
| :-- | --: |
| Change from base [Build 6782917032](https://coveralls.io/builds/63796745): |  0.0% |
| Covered Lines: | 13347 |
| Relevant Lines: | 13366 |

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


Patch:
diff --git a/patroni/__main__.py b/patroni/__main__.py
index 02ba56da..229ccfb9 100644
--- a/patroni/__main__.py
+++ b/patroni/__main__.py
@@ -107,8 +107,6 @@ class Patroni(AbstractPatroniDaemon, Tags):
 
     def ensure_unique_name(self) -> None:
         """A helper method to prevent splitbrain from operator naming error."""
-        from urllib.parse import urlparse
-        from urllib3.connection import HTTPConnection
         from patroni.dcs import Member
 
         cluster = self.dcs.get_cluster()
@@ -118,14 +116,14 @@ class Patroni(AbstractPatroniDaemon, Tags):
         if not isinstance(member, Member):
             return
         try:
-            parts = urlparse(member.api_url)
-            if isinstance(parts.hostname, str):
-                connection = HTTPConnection(parts.hostname, port=parts.port or 80, timeout=3)
-                connection.connect()
-                logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
-                sys.exit(1)
+            # Silence annoying WARNING: Retrying (...) messages when Patroni is quickly restarted.
+            # At this moment we don't have custom log levels configured and hence shouldn't lose anything useful.
+            self.logger.update_loggers({'urllib3.connectionpool': 'ERROR'})
+            _ = self.request(member, endpoint="/liveness", timeout=3)
+            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+            sys.exit(1)
         except Exception:
-            return
+            self.logger.update_loggers({})
 
     def _get_tags(self) -> Dict[str, Any]:
         """Get tags configured for this node, if any.
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/log.py b/patroni/log.py
index 09d73883..6ac67a17 100644
--- a/patroni/log.py
+++ b/patroni/log.py
@@ -202,24 +202,37 @@ class PatroniLogger(Thread):
         self._proxy_handler = ProxyHandler(self)
         self._root_logger.addHandler(self._proxy_handler)
 
-    def update_loggers(self) -> None:
-        """Configure loggers' log level as defined in ``log.loggers`` section of Patroni configuration.
+    def update_loggers(self, config: Dict[str, Any]) -> None:
+        """Configure custom loggers' log levels.
 
         .. note::
             It creates logger objects that are not defined yet in the log manager.
+
+        :param config: :class:`dict` object with custom loggers configuration, is set either from:
+
+                       * ``log.loggers`` section of Patroni configuration; or
+
+                       * from the method that is trying to make sure that the node name
+                         isn't duplicated (to silence annoying ``urllib3`` WARNING's).
+
+        :Example:
+
+            .. code-block:: python
+
+                update_loggers({'urllib3.connectionpool': 'WARNING'})
         """
-        loggers = deepcopy((self._config or {}).get('loggers') or {})
+        loggers = deepcopy(config)
         for name, logger in self._root_logger.manager.loggerDict.items():
             # ``Placeholder`` is a node in the log manager for which no logger has been defined. We are interested only
             # in the ones that were defined
             if not isinstance(logger, logging.PlaceHolder):
-                # if this logger is present in ``log.loggers`` Patroni configuration, use the configured level,
-                # otherwise use ``logging.NOTSET``, which means it will inherit the level from any parent node up to
-                # the root for which log level is defined.
+                # if this logger is present in *config*, use the configured level, otherwise
+                # use ``logging.NOTSET``, which means it will inherit the level
+                # from any parent node up to the root for which log level is defined.
                 level = loggers.pop(name, logging.NOTSET)
                 logger.setLevel(level)
 
-        # define loggers that do not exist yet and set level as configured in ``log.loggers`` section of configuration.
+        # define loggers that do not exist yet and set level as configured in the *config*
         for name, level in loggers.items():
             logger = self._root_logger.manager.getLogger(name)
             logger.setLevel(level)
@@ -274,7 +287,7 @@ class PatroniLogger(Thread):
                     self.log_handler = new_handler
 
             self._config = config.copy()
-            self.update_loggers()
+            self.update_loggers(config.get('loggers') or {})
 
     def _close_old_handlers(self) -> None:
         """Close old log handlers.


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_patroni.py b/tests/test_patroni.py
index 19497ab5..bf9e2871 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -45,7 +45,7 @@ class MockFrozenImporter(object):
 @patch('time.sleep', Mock())
 @patch('subprocess.call', Mock(return_value=0))
 @patch('patroni.psycopg.connect', psycopg_connect)
-@patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
+@patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -69,7 +69,7 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, _main)
 
     @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()]))
-    @patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
+    @patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
     @patch('sys.frozen', Mock(return_value=True), create=True)
     @patch.object(HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)
@@ -273,8 +273,8 @@ class TestPatroni(unittest.TestCase):
         )
         with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)):
             # If the api of the running node cannot be reached, this implies unique name
-            with patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=ConnectionError)):
+            with patch('urllib3.PoolManager.request', Mock(side_effect=ConnectionError)):
                 self.assertIsNone(self.p.ensure_unique_name())
             # Only if the api of the running node is reachable do we throw an error
-            with patch('urllib3.connection.HTTPConnection.connect', Mock()):
+            with patch('urllib3.PoolManager.request', Mock()):
                 self.assertRaises(SystemExit, self.p.ensure_unique_name)