From a5ff7870740668c41a12b810a1933ffd9cf0460e Mon Sep 17 00:00:00 2001 From: Adrian Vladu Date: Tue, 20 Oct 2026 15:51:59 +0300 Subject: [PATCH] Add parallel metadata loading Change-Id: Icaf88755c78b983f2c40144251c6b635b3a10cd8 Signed-off-by: Adrian Vladu --- cloudbaseinit/conf/default.py | 11 ++ cloudbaseinit/metadata/factory.py | 56 ++++++++- cloudbaseinit/tests/metadata/test_factory.py | 125 +++++++++++++++++++ 3 files changed, 191 insertions(+), 1 deletion(-) diff --git a/cloudbaseinit/conf/default.py b/cloudbaseinit/conf/default.py index 619096011..7aaf48f50 100644 --- a/cloudbaseinit/conf/default.py +++ b/cloudbaseinit/conf/default.py @@ -157,6 +157,11 @@ def __init__(self, config): 'the password is a clear text password, coming from the ' 'metadata. The last option is `no`, when the user is ' 'never forced to change the password.'), + cfg.IntOpt( + 'parallel_metadata_workers', default=4, + help='Maximum number of metadata services probed at the same ' + 'time when "--parallel" is enabled. Set to 0 or 1 to ' + 'disable concurrency'), cfg.ListOpt( 'metadata_services', default=[ @@ -298,6 +303,12 @@ def __init__(self, config): ] self._cli_options = [ + cfg.BoolOpt( + 'parallel', default=False, + help='Search the metadata sources listed in "metadata_' + 'services" concurrently instead of one by one. The ' + 'first available service is still picked according to ' + 'the order in which the services are listed.'), cfg.BoolOpt( 'reset_service_password', default=True, help='If set to True, the service user password will be ' diff --git a/cloudbaseinit/metadata/factory.py b/cloudbaseinit/metadata/factory.py index aa483274a..9624a9306 100644 --- a/cloudbaseinit/metadata/factory.py +++ b/cloudbaseinit/metadata/factory.py @@ -12,6 +12,8 @@ # License for the specific language governing permissions and limitations # under the License. +import concurrent.futures + from oslo_log import log as oslo_logging from cloudbaseinit import conf as cloudbaseinit_conf @@ -23,7 +25,24 @@ LOG = oslo_logging.getLogger(__name__) -def get_metadata_service(): +def _load_metadata_service(classloader, class_path): + """Load and probe a single metadata service. + + Returns None when the class cannot be loaded or the provider is not + available; errors are logged instead of propagated so that a single + misbehaving provider cannot break the discovery of the other ones. + """ + try: + service = classloader.load_class(class_path)() + if service.load(): + return service + except Exception as ex: + LOG.error("Failed to load metadata service '%s'" % class_path) + LOG.exception(ex) + return None + + +def get_metadata_service_sequential(): # Return the first service that loads correctly cl = classloader.ClassLoader() for class_path in CONF.metadata_services: @@ -35,3 +54,38 @@ def get_metadata_service(): LOG.error("Failed to load metadata service '%s'" % class_path) LOG.exception(ex) raise exception.MetadataNotFoundException("No available service found") + + +def get_metadata_service_parallel(): + """Return the first available service, testing providers concurrently. + + Every provider listed in `metadata_services` is probed at the same time, + which helps when several of them poll unreachable endpoints. The winner + is still picked by the position in the list, so the outcome matches what + the sequential search would have returned. A provider that raises while + loading only disables itself; it does not affect the other probes. + """ + cl = classloader.ClassLoader() + with concurrent.futures.ThreadPoolExecutor( + max_workers=max(CONF.parallel_metadata_workers, 1)) as pool: + futures = [pool.submit(_load_metadata_service, cl, class_path) + for class_path in CONF.metadata_services] + + # Futures keep the order of `metadata_services`. Waiting for the + # probes of the higher priority providers before accepting a result + # is what makes the winner independent from the probing speed; + # without it a later provider could win just because it responded + # first. + for future in futures: + service = future.result() + if service is not None: + return service + + raise exception.MetadataNotFoundException( + "No available service found") + + +def get_metadata_service(): + if CONF.parallel: + return get_metadata_service_parallel() + return get_metadata_service_sequential() diff --git a/cloudbaseinit/tests/metadata/test_factory.py b/cloudbaseinit/tests/metadata/test_factory.py index a827f9d09..45b7ca75b 100644 --- a/cloudbaseinit/tests/metadata/test_factory.py +++ b/cloudbaseinit/tests/metadata/test_factory.py @@ -12,14 +12,19 @@ # License for the specific language governing permissions and limitations # under the License. +import time import unittest import unittest.mock as mock +from cloudbaseinit import conf as cloudbaseinit_conf from cloudbaseinit import exception from cloudbaseinit.metadata import factory from cloudbaseinit.tests import testutils +CONF = cloudbaseinit_conf.CONF + + class MetadataServiceFactoryTests(unittest.TestCase): @mock.patch('cloudbaseinit.utils.classloader.ClassLoader.load_class') @@ -50,3 +55,123 @@ def test_get_metadata_service_load_exception(self): with testutils.LogSnatcher('cloudbaseinit.metadata.' 'factory'): self._test_get_metadata_service(load_exception=True) + + @mock.patch('cloudbaseinit.metadata.factory.classloader.ClassLoader') + def _test_parallel_get_metadata_service(self, mock_class_loader, + available_indexes=(1,), + raising_indexes=(), + class_paths=None): + if class_paths is None: + class_paths = ['fake.service.%s' % index + for index in range(3)] + CONF.metadata_services = class_paths + loaded_services = [mock.MagicMock() for _ in class_paths] + # `load_class(path)()` is what the factory calls, so each provider + # must be handed out through a callable. + mock_class_loader.return_value.load_class.side_effect = [ + lambda service=service: service for service in loaded_services] + for index, service in enumerate(loaded_services): + if index in raising_indexes: + service.load.side_effect = Exception + else: + service.load.return_value = index in available_indexes + + response = factory.get_metadata_service_parallel() + expected_index = next((index for index in range(len(class_paths)) + if index in available_indexes and + index not in raising_indexes), None) + if expected_index is None: + self.assertIsNone(response) + else: + self.assertEqual(loaded_services[expected_index], response) + + def test_get_metadata_service_parallel_first_available(self): + self._test_parallel_get_metadata_service(available_indexes=(0,)) + + def test_get_metadata_service_parallel_last_available(self): + self._test_parallel_get_metadata_service(available_indexes=(2,)) + + def test_get_metadata_service_parallel_prefers_listed_order(self): + # All providers report success; the earliest listed one must win. + self._test_parallel_get_metadata_service( + available_indexes=(0, 1, 2)) + + def test_get_metadata_service_parallel_none_available(self): + with self.assertRaises(exception.MetadataNotFoundException): + self._test_parallel_get_metadata_service(available_indexes=()) + + @mock.patch('cloudbaseinit.metadata.factory.classloader.ClassLoader') + def _test_parallel_slow_high_priority_provider(self, mock_class_loader): + """A slow high-priority provider must still beat a fast one. + + The probes run concurrently, so the winner has to be decided by the + position in `metadata_services` and not by which probe happened to + answer first. + """ + class_paths = ['fake.service.%s' % index for index in range(3)] + CONF.metadata_services = class_paths + loaded_services = [mock.MagicMock() for _ in class_paths] + mock_class_loader.return_value.load_class.side_effect = [ + lambda service=service: service for service in loaded_services] + + def load_slow(): + time.sleep(0.2) + return True + + loaded_services[0].load.side_effect = load_slow + for service in loaded_services[1:]: + service.load.return_value = True + + response = factory.get_metadata_service_parallel() + self.assertEqual(loaded_services[0], response) + + def test_get_metadata_service_parallel_raising_provider_is_skipped(self): + # A provider that raises must not hide a later available one. + self._test_parallel_get_metadata_service( + available_indexes=(1, 2), raising_indexes=(0,)) + + @mock.patch('cloudbaseinit.metadata.factory.classloader.ClassLoader') + def _test_parallel_get_metadata_service_unloadable(self, + mock_class_loader): + """A provider whose class cannot be loaded must not break the run.""" + class_paths = ['fake.service.unloadable', 'fake.service.available'] + CONF.metadata_services = list(class_paths) + available_service = mock.MagicMock() + available_service.load.return_value = True + + def load_class(path): + if path == class_paths[0]: + raise exception.CloudbaseInitException('cannot import') + return lambda: available_service + + mock_class_loader.return_value.load_class.side_effect = load_class + + response = factory.get_metadata_service_parallel() + self.assertEqual(available_service, response) + + def test_get_metadata_service_parallel_load_exception(self): + with testutils.LogSnatcher('cloudbaseinit.metadata.factory') \ + as snatcher: + self._test_parallel_get_metadata_service_unloadable() + self.assertEqual( + ["Failed to load metadata service " + "'fake.service.unloadable'"], snatcher.output[:1]) + + def test_get_metadata_service_parallel_slow_provider_priority(self): + self._test_parallel_slow_high_priority_provider() + + @testutils.ConfPatcher('parallel', True) + def test_get_metadata_service_parallel_enabled(self): + with mock.patch('cloudbaseinit.metadata.factory.' + 'get_metadata_service_parallel') \ + as mock_parallel: + factory.get_metadata_service() + mock_parallel.assert_called_once_with() + + @testutils.ConfPatcher('parallel', False) + def test_get_metadata_service_parallel_disabled(self): + with mock.patch('cloudbaseinit.metadata.factory.' + 'get_metadata_service_sequential') \ + as mock_sequential: + factory.get_metadata_service() + mock_sequential.assert_called_once_with()