From 57ed40f66cc9273caa8bf02575a8bb7a39f720db Mon Sep 17 00:00:00 2001 From: Waynerv Date: Thu, 12 Sep 2024 14:55:05 +0800 Subject: [PATCH] Fix unhandled DCSError during startup phase (#3149) Ensure DCS connectivity before we check node uniqueness or load dynamic configuration. --- patroni/__main__.py | 67 +++++++++++++++++++++++++------------------ tests/test_patroni.py | 51 ++++++++++++++++++++------------ 2 files changed, 72 insertions(+), 46 deletions(-) diff --git a/patroni/__main__.py b/patroni/__main__.py index 169a39cd..e3d8a3c4 100644 --- a/patroni/__main__.py +++ b/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) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index abf86f45..ae017c47 100644 --- a/tests/test_patroni.py +++ b/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) +