Merged in aznashwan/coriolis-core/worker-process-presetup (pull request #122)

Add support for adding custom library paths to worker processes.

coriolis/constants.py

@@ -56,6 +56,7 @@ PROVIDER_TYPE_OS_MORPHING = 64
 PROVIDER_TYPE_ENDPOINT_NETWORKS = 128
 PROVIDER_TYPE_INSTANCE_FLAVOR = 256
 PROVIDER_TYPE_ENDPOINT_OPTIONS = 512
+PROVIDER_TYPE_SETUP_LIBS = 1024
 
 DISK_FORMAT_VMDK = 'vmdk'
 DISK_FORMAT_RAW = 'raw'

coriolis/providers/base.py

@@ -54,6 +54,18 @@ class BaseEndpointNetworksProvider(object, with_metaclass(abc.ABCMeta)):
         raise NotImplementedError()
 
 
+class BaseProviderSetupExtraLibsMixin(object, with_metaclass(abc.ABCMeta)):
+    """ ABC mixin for providers which require extra libraries loaded. """
+
+    @abc.abstractmethod
+    def get_shared_library_directories(self, ctxt, connection_info):
+        """ Should return a list of string paths to directories somewhere in
+        the worker filesystem where extra libraries required for the provider
+        are located.
+        """
+        return []
+
+
 class BaseEndpointDestinationOptionsProvider(
         object, with_metaclass(abc.ABCMeta)):
     @abc.abstractmethod

coriolis/providers/factory.py

@@ -31,6 +31,7 @@ PROVIDER_TYPE_MAP = {
         base.BaseEndpointNetworksProvider,
     constants.PROVIDER_TYPE_OS_MORPHING: base.BaseImportInstanceProvider,
     constants.PROVIDER_TYPE_INSTANCE_FLAVOR: base.BaseInstanceFlavorProvider,
+    constants.PROVIDER_TYPE_SETUP_LIBS: base.BaseProviderSetupExtraLibsMixin
 }
 
 
@@ -51,13 +52,17 @@ def get_available_providers():
     return providers
 
 
-def get_provider(platform_name, provider_type, event_handler):
+def get_provider(
+        platform_name, provider_type, event_handler, raise_if_not_found=True):
     for provider in CONF.providers:
         cls = utils.load_class(provider)
         if (cls.platform == platform_name and
                 issubclass(cls, PROVIDER_TYPE_MAP[provider_type])):
             return cls(event_handler)
 
-    raise exception.NotFound(
-        "Provider not found for: %(platform_name)s, %(provider_type)s" %
-        {"platform_name": platform_name, "provider_type": provider_type})
+    if raise_if_not_found:
+        raise exception.NotFound(
+            "Provider not found for: %(platform_name)s, %(provider_type)s" %
+            {"platform_name": platform_name, "provider_type": provider_type})
+
+    return None

coriolis/tasks/base.py

@@ -7,7 +7,9 @@ from oslo_config import cfg
 from oslo_log import log as logging
 from six import with_metaclass
 
+from coriolis import constants
 from coriolis import utils
+from coriolis.providers import factory as providers_factory
 
 serialization_opts = [
     cfg.StrOpt('temp_keypair_password',
@@ -21,6 +23,33 @@ LOG = logging.getLogger(__name__)
 
 
 class TaskRunner(with_metaclass(abc.ABCMeta)):
+
+    def get_shared_libs_for_providers(
+            self, ctxt, origin, destination, event_handler):
+        """ Returns a list of directories containing libraries needed
+        for both the source and destination providers. """
+        required_libs = []
+
+        origin_provider = providers_factory.get_provider(
+            origin["type"], constants.PROVIDER_TYPE_SETUP_LIBS, event_handler,
+            raise_if_not_found=False)
+        if origin_provider:
+            conn_info = get_connection_info(ctxt, origin)
+            required_libs.extend(
+                origin_provider.get_shared_library_directories(
+                    ctxt, conn_info))
+
+        destination_provider = providers_factory.get_provider(
+            destination["type"], constants.PROVIDER_TYPE_SETUP_LIBS,
+            event_handler, raise_if_not_found=False)
+        if destination_provider:
+            conn_info = get_connection_info(ctxt, destination)
+            required_libs.extend(
+                destination_provider.get_shared_library_directories(
+                    ctxt, conn_info))
+
+        return required_libs
+
     @abc.abstractmethod
     def run(self, ctxt, instance, origin, destination, task_info,
             event_handler):

coriolis/worker/rpc/server.py

@@ -109,6 +109,46 @@ class WorkerServerEndpoint(object):
                 if not p.is_alive():
                     break
 
+    def _start_process_with_custom_library_paths(
+            self, process, extra_library_paths):
+        """ Given a process instance, this method will add any shared libs
+        needed by the origin/destination provider plugins to the
+        'LD_LIBRARY_PATH' env variable and start the process.
+        This method will always restore the 'LD_LIBRARY_PATH' to the
+        original value for the parent process.
+        param process: multiprocessing.Process: Process instance to run
+        with the modified 'LD_LIBRARY_PATH'
+        param extra_library_paths: list(str): list of paths with extra
+        libraries which should be available to the worker process.
+        """
+        original_ld_path = os.environ.get('LD_LIBRARY_PATH', "")
+        new_ld_path = None
+        extra_libdirs = ":".join(extra_library_paths)
+        if not original_ld_path:
+            new_ld_path = extra_libdirs
+        else:
+            new_ld_path = "%s:%s" % (original_ld_path, extra_libdirs)
+
+        LOG.debug(
+            "Starting new worker process with extra libraries: '%s'",
+            extra_library_paths)
+        try:
+            os.environ['LD_LIBRARY_PATH'] = new_ld_path
+            process.start()
+        finally:
+            os.environ['LD_LIBRARY_PATH'] = original_ld_path
+
+    def _get_extra_library_paths_for_providers(
+            self, ctxt, task_id, task_type, origin, destination):
+        """ Returns a list of strings with paths on the worker with shared
+        libraries needed by the source/destination providers.
+        """
+        event_handler = _ConductorProviderEventHandler(ctxt, task_id)
+        task_runner = task_runners_factory.get_task_runner(task_type)
+
+        return task_runner.get_shared_libs_for_providers(
+            ctxt, origin, destination, event_handler)
+
     def _exec_task_process(self, ctxt, task_id, task_type, origin, destination,
                            instance, task_info):
         mp_ctx = multiprocessing.get_context('spawn')
@@ -119,7 +159,10 @@ class WorkerServerEndpoint(object):
             args=(ctxt, task_id, task_type, origin, destination, instance,
                   task_info, mp_q, mp_log_q))
 
-        p.start()
+        extra_library_paths = self._get_extra_library_paths_for_providers(
+            ctxt, task_id, task_type, origin, destination)
+
+        self._start_process_with_custom_library_paths(p, extra_library_paths)
         LOG.info("Task process started: %s", task_id)
         self._rpc_conductor_client.set_task_host(
             ctxt, task_id, self._server, p.pid)