Fix unhandled DCSError during startup phase (#3149)

Ensure DCS connectivity before we check node uniqueness or load dynamic configuration.

patroni/__main__.py

@@ -18,6 +18,7 @@ from patroni.tags import Tags
 
 if TYPE_CHECKING:  # pragma: no cover
     from .config import Config
+    from .dcs import Cluster
 
 logger = logging.getLogger(__name__)
 
@@ -63,10 +64,11 @@ class Patroni(AbstractPatroniDaemon, Tags):
         self.dcs = get_dcs(self.config)
         self.request = PatroniRequest(self.config, True)
 
-        self.ensure_unique_name()
+        cluster = self.ensure_dcs_access()
+        self.ensure_unique_name(cluster)
 
         self.watchdog = Watchdog(self.config)
-        self.load_dynamic_configuration()
+        self.apply_dynamic_configuration(cluster)
 
         self.postgresql = Postgresql(self.config['postgresql'], self.dcs.mpp)
         self.api = RestApiServer(self, self.config['restapi'])
@@ -76,40 +78,49 @@ class Patroni(AbstractPatroniDaemon, Tags):
         self.next_run = time.time()
         self.scheduled_restart: Dict[str, Any] = {}
 
-    def load_dynamic_configuration(self) -> None:
-        """Load Patroni dynamic configuration.
+    def ensure_dcs_access(self, sleep_time: int = 5) -> 'Cluster':
+        """Continuously attempt to retrieve cluster from DCS with delay.
 
-        Load dynamic configuration from the DCS, if `/config` key is available in the DCS, otherwise fall back to
+        :param sleep_time: seconds to wait between retry attempts after dcs connection raise :exc:`DCSError`.
+
+        :returns: a PostgreSQL or MPP implementation of :class:`Cluster`.
+        """
+        from patroni.exceptions import DCSError
+
+        while True:
+            try:
+                return self.dcs.get_cluster()
+            except DCSError:
+                logger.warning('Can not get cluster from dcs')
+                time.sleep(sleep_time)
+
+    def apply_dynamic_configuration(self, cluster: 'Cluster') -> None:
+        """Apply Patroni dynamic configuration.
+
+        Apply dynamic configuration from the DCS, if `/config` key is available in the DCS, otherwise fall back to
         ``bootstrap.dcs`` section from the configuration file.
 
-        If the DCS connection fails returning the exception :class:`~patroni.exceptions.DCSError` an attempt will be
-        remade every 5 seconds.
-
         .. note::
             This method is called only once, at the time when Patroni is started.
-        """
-        from patroni.exceptions import DCSError
-        while True:
-            try:
-                cluster = self.dcs.get_cluster()
-                if cluster and cluster.config and cluster.config.data:
-                    if self.config.set_dynamic_configuration(cluster.config):
-                        self.dcs.reload_config(self.config)
-                        self.watchdog.reload_config(self.config)
-                elif not self.config.dynamic_configuration and 'bootstrap' in self.config:
-                    if self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']):
-                        self.dcs.reload_config(self.config)
-                        self.watchdog.reload_config(self.config)
-                break
-            except DCSError:
-                logger.warning('Can not get cluster from dcs')
-                time.sleep(5)
 
-    def ensure_unique_name(self) -> None:
-        """A helper method to prevent splitbrain from operator naming error."""
+        :param cluster: a PostgreSQL or MPP implementation of :class:`Cluster`.
+        """
+        if cluster and cluster.config and cluster.config.data:
+            if self.config.set_dynamic_configuration(cluster.config):
+                self.dcs.reload_config(self.config)
+                self.watchdog.reload_config(self.config)
+        elif not self.config.dynamic_configuration and 'bootstrap' in self.config:
+            if self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']):
+                self.dcs.reload_config(self.config)
+                self.watchdog.reload_config(self.config)
+
+    def ensure_unique_name(self, cluster: 'Cluster') -> None:
+        """A helper method to prevent splitbrain from operator naming error.
+
+        :param cluster: a PostgreSQL or MPP implementation of :class:`Cluster`.
+        """
         from patroni.dcs import Member
 
-        cluster = self.dcs.get_cluster()
         if not cluster:
             return
         member = cluster.get_member(self.config['name'], False)

tests/test_patroni.py

@@ -15,7 +15,7 @@ import patroni.config as config
 from patroni.__main__ import check_psycopg, main as _main, Patroni
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
-from patroni.dcs import Cluster, Member
+from patroni.dcs import Cluster, ClusterConfig, Member
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
@@ -90,11 +90,22 @@ class TestPatroni(unittest.TestCase):
     def tearDown(self):
         logging.getLogger().handlers[:] = self._handlers
 
-    @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(side_effect=[None, DCSError('foo'), None]))
-    def test_load_dynamic_configuration(self):
+    def test_apply_dynamic_configuration(self):
+        empty_cluster = Cluster.empty()
         self.p.config._dynamic_configuration = {}
-        self.p.load_dynamic_configuration()
-        self.p.load_dynamic_configuration()
+        self.p.apply_dynamic_configuration(empty_cluster)
+        self.assertEqual(self.p.config._dynamic_configuration['ttl'], 30)
+
+        without_config = empty_cluster._asdict()
+        del without_config['config']
+        cluster = Cluster(
+            config=ClusterConfig(version=1, modify_version=1, data={"ttl": 40}),
+            **without_config
+        )
+        self.p.config._dynamic_configuration = {}
+        self.p.apply_dynamic_configuration(cluster)
+        self.assertEqual(self.p.config._dynamic_configuration['ttl'], 40)
+
 
     @patch('sys.argv', ['patroni.py', 'postgres0.yml'])
     @patch('time.sleep', Mock(side_effect=SleepException))
@@ -276,11 +287,9 @@ class TestPatroni(unittest.TestCase):
 
     def test_ensure_unique_name(self):
         # None/empty cluster implies unique name
-        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=None)):
-            self.assertIsNone(self.p.ensure_unique_name())
+        self.assertIsNone(self.p.ensure_unique_name(None))
         empty_cluster = Cluster.empty()
-        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=empty_cluster)):
-            self.assertIsNone(self.p.ensure_unique_name())
+        self.assertIsNone(self.p.ensure_unique_name(empty_cluster))
         without_members = empty_cluster._asdict()
         del without_members['members']
 
@@ -289,8 +298,7 @@ class TestPatroni(unittest.TestCase):
             members=[Member(version=1, name="distinct", session=1, data={})],
             **without_members
         )
-        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=okay_cluster)):
-            self.assertIsNone(self.p.ensure_unique_name())
+        self.assertIsNone(self.p.ensure_unique_name(okay_cluster))
 
         # Cluster with a member with the same name that is running
         bad_cluster = Cluster(
@@ -299,10 +307,17 @@ class TestPatroni(unittest.TestCase):
             })],
             **without_members
         )
-        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.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.PoolManager.request', Mock()):
-                self.assertRaises(SystemExit, self.p.ensure_unique_name)
+        # If the api of the running node cannot be reached, this implies unique name
+        with patch('urllib3.PoolManager.request', Mock(side_effect=ConnectionError)):
+            self.assertIsNone(self.p.ensure_unique_name(bad_cluster))
+        # Only if the api of the running node is reachable do we throw an error
+        with patch('urllib3.PoolManager.request', Mock()):
+            self.assertRaises(SystemExit, self.p.ensure_unique_name, bad_cluster)
+
+    @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(side_effect=[DCSError('foo'), DCSError('foo'), None]))
+    def test_ensure_dcs_access(self):
+        with patch('patroni.__main__.logger.warning') as mock_logger:
+            result = self.p.ensure_dcs_access()
+            self.assertEqual(result, None)
+            self.assertEqual(mock_logger.call_count, 2)
+