Add parallel metadata loading

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 '

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()

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 @@
         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()