Add `source_environment` column to `base_transfer_action`.

coriolis/api/v1/migrations.py

@@ -54,6 +54,10 @@ class MigrationController(api_wsgi.Controller):
             notes = migration.get("notes")
             skip_os_morphing = migration.get("skip_os_morphing", False)
 
+            source_environment = migration.get("source_environment", {})
+            self._endpoints_api.validate_source_environment(
+                context, origin_endpoint_id, source_environment)
+
             network_map = migration.get("network_map", {})
             api_utils.validate_network_map(network_map)
             destination_environment['network_map'] = network_map
@@ -62,13 +66,8 @@ class MigrationController(api_wsgi.Controller):
             # import provider before appending the 'storage_mappings' parameter
             # for plugins with strict property name checks which do not yet
             # support storage mapping features:
-            is_valid, message = (
-                self._endpoints_api.validate_target_environment(
-                    context, destination_endpoint_id, destination_environment))
-            if not is_valid:
-                raise exc.HTTPBadRequest(
-                    explanation="Invalid destination "
-                                "environment: %s" % message)
+            self._endpoints_api.validate_target_environment(
+                context, destination_endpoint_id, destination_environment)
 
             # TODO(aznashwan): until the provider plugin interface is updated
             # to have separate 'network_map' and 'storage_mappings' fields,
@@ -78,8 +77,8 @@ class MigrationController(api_wsgi.Controller):
             destination_environment['storage_mappings'] = storage_mappings
 
             return (origin_endpoint_id, destination_endpoint_id,
-                    destination_environment, instances, notes,
-                    skip_os_morphing, network_map, storage_mappings)
+                    source_environment, destination_environment, instances,
+                    notes, skip_os_morphing, network_map, storage_mappings)
         except Exception as ex:
             LOG.exception(ex)
             msg = getattr(ex, "message", str(ex))
@@ -104,17 +103,17 @@ class MigrationController(api_wsgi.Controller):
         else:
             (origin_endpoint_id,
              destination_endpoint_id,
+             source_environment,
              destination_environment,
              instances,
              notes,
              skip_os_morphing, network_map,
              storage_mappings) = self._validate_migration_input(
                 context, migration_body)
-
             migration = self._migration_api.migrate_instances(
                 context, origin_endpoint_id, destination_endpoint_id,
-                destination_environment, instances, network_map,
-                storage_mappings, notes, skip_os_morphing)
+                source_environment, destination_environment, instances,
+                network_map, storage_mappings, notes, skip_os_morphing)
 
         return migration_view.single(req, migration)
 

coriolis/api/v1/replicas.py

@@ -55,6 +55,10 @@ class ReplicaController(api_wsgi.Controller):
             instances = replica["instances"]
             notes = replica.get("notes")
 
+            source_environment = replica.get("source_environment", {})
+            self._endpoints_api.validate_source_environment(
+                context, origin_endpoint_id, source_environment)
+
             network_map = replica.get("network_map", {})
             api_utils.validate_network_map(network_map)
             destination_environment['network_map'] = network_map
@@ -63,14 +67,8 @@ class ReplicaController(api_wsgi.Controller):
             # import provider before appending the 'storage_mappings' parameter
             # for plugins with strict property name checks which do not yet
             # support storage mapping features:
-            is_valid, message = (
-                self._endpoints_api.validate_target_environment(
-                    context, destination_endpoint_id,
-                    destination_environment))
-            if not is_valid:
-                raise exc.HTTPBadRequest(
-                    explanation="Invalid destination "
-                                "environment: %s" % message)
+            self._endpoints_api.validate_target_environment(
+                context, destination_endpoint_id, destination_environment)
 
             storage_mappings = replica.get("storage_mappings", {})
             api_utils.validate_storage_mappings(storage_mappings)
@@ -81,8 +79,8 @@ class ReplicaController(api_wsgi.Controller):
             destination_environment['storage_mappings'] = storage_mappings
 
             return (origin_endpoint_id, destination_endpoint_id,
-                    destination_environment, instances, network_map,
-                    storage_mappings, notes)
+                    source_environment, destination_environment, instances,
+                    network_map, storage_mappings, notes)
         except Exception as ex:
             LOG.exception(ex)
             msg = getattr(ex, "message", str(ex))
@@ -93,13 +91,13 @@ class ReplicaController(api_wsgi.Controller):
         context.can(replica_policies.get_replicas_policy_label("create"))
 
         (origin_endpoint_id, destination_endpoint_id,
-         destination_environment, instances, network_map,
+         source_environment, destination_environment, instances, network_map,
          storage_mappings, notes) = self._validate_create_body(context, body)
 
         return replica_view.single(req, self._replica_api.create(
             context, origin_endpoint_id, destination_endpoint_id,
-            destination_environment, instances, network_map,
-            storage_mappings, notes))
+            source_environment, destination_environment, instances,
+            network_map, storage_mappings, notes))
 
     def delete(self, req, id):
         context = req.environ["coriolis.context"]

coriolis/api/v1/utils.py

@@ -1,6 +1,7 @@
 # Copyright 2018 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+
 from oslo_log import log as logging
 from webob import exc
 
@@ -29,3 +30,14 @@ def validate_storage_mappings(storage_mappings):
     except exception.SchemaValidationException as ex:
         raise exc.HTTPBadRequest(
             explanation="Invalid storage_mappings: %s" % str(ex))
+
+
+def bad_request_on_error(error_message):
+    def _bad_request_on_error(func):
+        def wrapper(*args, **kwargs):
+            (is_valid, message) = func(*args, **kwargs)
+            if not is_valid:
+                raise exc.HTTPBadRequest(explanation=(error_message % message))
+            return (is_valid, message)
+        return wrapper
+    return _bad_request_on_error

coriolis/conductor/rpc/client.py

@@ -94,6 +94,12 @@ class ConductorClient(object):
             ctxt, 'validate_endpoint_target_environment',
             endpoint_id=endpoint_id, target_env=target_env)
 
+    def validate_endpoint_source_environment(
+            self, ctxt, endpoint_id, source_env):
+        return self._client.call(
+            ctxt, 'validate_endpoint_source_environment',
+            endpoint_id=endpoint_id, source_env=source_env)
+
     def get_available_providers(self, ctxt):
         return self._client.call(
             ctxt, 'get_available_providers')
@@ -135,8 +141,8 @@ class ConductorClient(object):
 
     def create_instances_replica(self, ctxt, origin_endpoint_id,
                                  destination_endpoint_id,
-                                 destination_environment, instances,
-                                 network_map, storage_mappings,
+                                 source_environment, destination_environment,
+                                 instances, network_map, storage_mappings,
                                  notes=None):
         return self._client.call(
             ctxt, 'create_instances_replica',
@@ -146,7 +152,8 @@ class ConductorClient(object):
             instances=instances,
             notes=notes,
             network_map=network_map,
-            storage_mappings=storage_mappings)
+            storage_mappings=storage_mappings,
+            source_environment=source_environment)
 
     def get_replicas(self, ctxt, include_tasks_executions=False):
         return self._client.call(
@@ -174,8 +181,9 @@ class ConductorClient(object):
             ctxt, 'get_migration', migration_id=migration_id)
 
     def migrate_instances(self, ctxt, origin_endpoint_id,
-                          destination_endpoint_id, destination_environment,
-                          instances, network_map, storage_mappings, notes=None,
+                          destination_endpoint_id, source_environment,
+                          destination_environment, instances, network_map,
+                          storage_mappings, notes=None,
                           skip_os_morphing=False):
         return self._client.call(
             ctxt, 'migrate_instances',
@@ -186,7 +194,8 @@ class ConductorClient(object):
             notes=notes,
             skip_os_morphing=skip_os_morphing,
             network_map=network_map,
-            storage_mappings=storage_mappings)
+            storage_mappings=storage_mappings,
+            source_environment=source_environment)
 
     def deploy_replica_instances(self, ctxt, replica_id, clone_disks=False,
                                  force=False, skip_os_morphing=False):

coriolis/conductor/rpc/server.py

@@ -168,6 +168,12 @@ class ConductorServerEndpoint(object):
         return self._rpc_worker_client.validate_endpoint_target_environment(
             ctxt, endpoint.type, target_env)
 
+    def validate_endpoint_source_environment(
+            self, ctxt, endpoint_id, source_env):
+        endpoint = self.get_endpoint(ctxt, endpoint_id)
+        return self._rpc_worker_client.validate_endpoint_source_environment(
+            ctxt, endpoint.type, source_env)
+
     def get_available_providers(self, ctxt):
         return self._rpc_worker_client.get_available_providers(ctxt)
 
@@ -202,7 +208,8 @@ class ConductorServerEndpoint(object):
         endpoint = self.get_endpoint(ctxt, action.origin_endpoint_id)
         return {
             "connection_info": endpoint.connection_info,
-            "type": endpoint.type
+            "type": endpoint.type,
+            "source_environment": action.source_environment
         }
 
     def _get_task_destination(self, ctxt, action):
@@ -385,7 +392,7 @@ class ConductorServerEndpoint(object):
             raise exception.SameDestination()
 
     def create_instances_replica(self, ctxt, origin_endpoint_id,
-                                 destination_endpoint_id,
+                                 destination_endpoint_id, source_environment,
                                  destination_environment, instances,
                                  network_map, storage_mappings, notes=None):
         origin_endpoint = self.get_endpoint(ctxt, origin_endpoint_id)
@@ -397,6 +404,7 @@ class ConductorServerEndpoint(object):
         replica.origin_endpoint = origin_endpoint
         replica.destination_endpoint = destination_endpoint
         replica.destination_environment = destination_environment
+        replica.source_environment = source_environment
         replica.instances = instances
         replica.executions = []
         replica.info = {}
@@ -495,6 +503,7 @@ class ConductorServerEndpoint(object):
         migration.origin_endpoint_id = replica.origin_endpoint_id
         migration.destination_endpoint_id = replica.destination_endpoint_id
         migration.destination_environment = replica.destination_environment
+        migration.source_environment = replica.source_environment
         migration.network_map = replica.network_map
         migration.storage_mappings = replica.storage_mappings
         migration.instances = instances
@@ -584,9 +593,10 @@ class ConductorServerEndpoint(object):
         return self.get_migration(ctxt, migration.id)
 
     def migrate_instances(self, ctxt, origin_endpoint_id,
-                          destination_endpoint_id, destination_environment,
-                          instances, network_map, storage_mappings,
-                          skip_os_morphing=False, notes=None):
+                          destination_endpoint_id, source_environment,
+                          destination_environment, instances, network_map,
+                          storage_mappings, skip_os_morphing=False,
+                          notes=None):
         origin_endpoint = self.get_endpoint(ctxt, origin_endpoint_id)
         destination_endpoint = self.get_endpoint(ctxt, destination_endpoint_id)
         self._check_endpoints(ctxt, origin_endpoint, destination_endpoint)
@@ -599,6 +609,7 @@ class ConductorServerEndpoint(object):
         migration.origin_endpoint = origin_endpoint
         migration.destination_endpoint = destination_endpoint
         migration.destination_environment = destination_environment
+        migration.source_environment = source_environment
         migration.network_map = network_map
         migration.storage_mappings = storage_mappings
         execution = models.TasksExecution()

coriolis/db/sqlalchemy/migrate_repo/versions/008_adds_source_environment.py

@@ -0,0 +1,17 @@
+# Copyright 2018 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sqlalchemy
+
+
+def upgrade(migrate_engine):
+    meta = sqlalchemy.MetaData()
+    meta.bind = migrate_engine
+
+    # add 'source_environment' column to 'base_transfer_action':
+    base_transfer_action = sqlalchemy.Table(
+        'base_transfer_action', meta, autoload=True)
+
+    source_environment = sqlalchemy.Column(
+        "source_environment", sqlalchemy.Text, nullable=True)
+    base_transfer_action.create_column(source_environment)

coriolis/db/sqlalchemy/models.py

@@ -114,6 +114,7 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
     transfer_result = sqlalchemy.Column(types.Json, nullable=True)
     network_map = sqlalchemy.Column(types.Json, nullable=True)
     storage_mappings = sqlalchemy.Column(types.Json, nullable=True)
+    source_environment = sqlalchemy.Column(types.Json, nullable=True)
 
     __mapper_args__ = {
         'polymorphic_identity': 'base_transfer_action',

coriolis/endpoints/api.py

@@ -2,6 +2,7 @@
 # All Rights Reserved.
 
 from coriolis.conductor.rpc import client as rpc_client
+from coriolis.api.v1 import utils as api_utils
 
 
 class API(object):
@@ -33,6 +34,12 @@ class API(object):
         return self._rpc_client.validate_endpoint_connection(
             ctxt, endpoint_id)
 
+    @api_utils.bad_request_on_error("Invalid destination environment: %s")
     def validate_target_environment(self, ctxt, endpoint_id, target_env):
         return self._rpc_client.validate_endpoint_target_environment(
             ctxt, endpoint_id, target_env)
+
+    @api_utils.bad_request_on_error("Invalid source environment: %s")
+    def validate_source_environment(self, ctxt, endpoint_id, source_env):
+        return self._rpc_client.validate_endpoint_source_environment(
+            ctxt, endpoint_id, source_env)

coriolis/migrations/api.py

@@ -9,13 +9,14 @@ class API(object):
         self._rpc_client = rpc_client.ConductorClient()
 
     def migrate_instances(self, ctxt, origin_endpoint_id,
-                          destination_endpoint_id, destination_environment,
-                          instances, network_map, storage_mappings, notes=None,
+                          destination_endpoint_id, source_environment,
+                          destination_environment, instances, network_map,
+                          storage_mappings, notes=None,
                           skip_os_morphing=False):
         return self._rpc_client.migrate_instances(
             ctxt, origin_endpoint_id, destination_endpoint_id,
-            destination_environment, instances, network_map, storage_mappings,
-            notes, skip_os_morphing)
+            source_environment, destination_environment, instances,
+            network_map, storage_mappings, notes, skip_os_morphing)
 
     def deploy_replica_instances(self, ctxt, replica_id, clone_disks=False,
                                  force=False, skip_os_morphing=False):

coriolis/providers/base.py

@@ -309,8 +309,8 @@ class BaseReplicaImportProvider(BaseImportInstanceProvider):
 class BaseExportProvider(BaseInstanceProvider):
 
     @abc.abstractmethod
-    def export_instance(self, ctxt, connection_info, instance_name,
-                        export_path):
+    def export_instance(self, ctxt, connection_info, source_environment,
+                        instance_name, export_path):
         """Exports the given instance.
 
          Exports the instance given by its name from the given source cloud
@@ -318,19 +318,26 @@ class BaseExportProvider(BaseInstanceProvider):
         """
         pass
 
+    @abc.abstractmethod
+    def get_source_environment_schema(self):
+        pass
+
 
 class BaseReplicaExportProvider(BaseInstanceProvider):
 
     @abc.abstractmethod
-    def get_replica_instance_info(self, ctxt, connection_info, instance_name):
+    def get_replica_instance_info(self, ctxt, connection_info,
+                                  source_environment, instance_name):
         pass
 
     @abc.abstractmethod
-    def deploy_replica_source_resources(self, ctxt, connection_info):
+    def deploy_replica_source_resources(self, ctxt, connection_info,
+                                        source_environment):
         pass
 
     @abc.abstractmethod
-    def delete_replica_source_resources(self, ctxt, connection_info,
+    def delete_replica_source_resources(self, ctxt, source_environment,
+                                        connection_info,
                                         migr_resources_dict):
         pass
 
@@ -341,7 +348,8 @@ class BaseReplicaExportProvider(BaseInstanceProvider):
         pass
 
     @abc.abstractmethod
-    def shutdown_instance(self, ctxt, connection_info, instance_name):
+    def shutdown_instance(self, ctxt, connection_info, source_environment,
+                          instance_name):
         pass
 
 

coriolis/replicas/api.py

@@ -9,12 +9,12 @@ class API(object):
         self._rpc_client = rpc_client.ConductorClient()
 
     def create(self, ctxt, origin_endpoint_id, destination_endpoint_id,
-               destination_environment, instances, network_map,
-               storage_mappings, notes=None):
+               source_environment, destination_environment, instances,
+               network_map, storage_mappings, notes=None):
         return self._rpc_client.create_instances_replica(
             ctxt, origin_endpoint_id, destination_endpoint_id,
-            destination_environment, instances, network_map, storage_mappings,
-            notes)
+            source_environment, destination_environment, instances,
+            network_map, storage_mappings, notes)
 
     def delete(self, ctxt, replica_id):
         self._rpc_client.delete_replica(ctxt, replica_id)

coriolis/schemas.py

@@ -19,6 +19,7 @@ DEFAULT_SCHEMAS_DIRECTORY = "schemas"
 PROVIDER_CONNECTION_INFO_SCHEMA_NAME = "connection_info_schema.json"
 
 PROVIDER_TARGET_ENVIRONMENT_SCHEMA_NAME = "target_environment_schema.json"
+PROVIDER_SOURCE_ENVIRONMENT_SCHEMA_NAME = "source_environment_schema.json"
 
 _CORIOLIS_VM_EXPORT_INFO_SCHEMA_NAME = "vm_export_info_schema.json"
 _CORIOLIS_VM_INSTANCE_INFO_SCHEMA_NAME = "vm_instance_info_schema.json"

coriolis/tasks/migration_tasks.py

@@ -22,8 +22,9 @@ class ExportInstanceTask(base.TaskRunner):
         connection_info = base.get_connection_info(ctxt, origin)
         export_path = task_info["export_path"]
 
+        source_environment = origin.get('source_environment') or {}
         export_info = provider.export_instance(
-            ctxt, connection_info, instance, export_path)
+            ctxt, connection_info, source_environment, instance, export_path)
 
         # Validate the output
         schemas.validate_value(

coriolis/tasks/replica_tasks.py

@@ -29,8 +29,9 @@ class GetInstanceInfoTask(base.TaskRunner):
             event_handler)
         connection_info = base.get_connection_info(ctxt, origin)
 
+        source_environment = origin.get('source_environment') or {}
         export_info = provider.get_replica_instance_info(
-            ctxt, connection_info, instance)
+            ctxt, connection_info, source_environment, instance)
 
         # Validate the output
         schemas.validate_value(
@@ -48,7 +49,9 @@ class ShutdownInstanceTask(base.TaskRunner):
             event_handler)
         connection_info = base.get_connection_info(ctxt, origin)
 
-        provider.shutdown_instance(ctxt, connection_info, instance)
+        source_environment = origin.get('source_environment') or {}
+        provider.shutdown_instance(ctxt, connection_info, source_environment,
+                                   instance)
 
         return task_info
 
@@ -128,8 +131,9 @@ class DeployReplicaSourceResourcesTask(base.TaskRunner):
             event_handler)
         connection_info = base.get_connection_info(ctxt, origin)
 
+        source_environment = origin.get('source_environment') or {}
         replica_resources_info = provider.deploy_replica_source_resources(
-            ctxt, connection_info)
+            ctxt, connection_info, source_environment)
 
         task_info["migr_source_resources"] = replica_resources_info[
             "migr_resources"]
@@ -150,9 +154,11 @@ class DeleteReplicaSourceResourcesTask(base.TaskRunner):
 
         migr_resources = task_info.get("migr_source_resources")
 
+        source_environment = origin.get("source_environment", {})
+
         if migr_resources:
             provider.delete_replica_source_resources(
-                ctxt, connection_info, migr_resources)
+                ctxt, source_environment, connection_info, migr_resources)
 
         task_info["migr_source_resources"] = None
         task_info["migr_source_connection_info"] = None

coriolis/worker/rpc/client.py

@@ -91,6 +91,13 @@ class WorkerClient(object):
             platform_name=platform_name,
             target_env=target_env)
 
+    def validate_endpoint_source_environment(
+            self, ctxt, platform_name, source_env):
+        return self._client.call(
+            ctxt, 'validate_endpoint_source_environment',
+            platform_name=platform_name,
+            source_env=source_env)
+
     def get_endpoint_storage(self, ctxt, platform_name, connection_info, env):
         return self._client.call(
             ctxt, 'get_endpoint_storage',

coriolis/worker/rpc/server.py

@@ -21,6 +21,7 @@ from coriolis.providers import factory as providers_factory
 from coriolis import schemas
 from coriolis.tasks import factory as task_runners_factory
 from coriolis import utils
+from coriolis.api.v1 import utils as api_utils
 
 
 worker_opts = [
@@ -294,6 +295,7 @@ class WorkerServerEndpoint(object):
     def get_available_providers(self, ctxt):
         return providers_factory.get_available_providers()
 
+    @api_utils.bad_request_on_error("Invalid destination environment: %s")
     def validate_endpoint_target_environment(
             self, ctxt, platform_name, target_env):
         provider = providers_factory.get_provider(
@@ -310,6 +312,23 @@ class WorkerServerEndpoint(object):
 
         return (is_valid, message)
 
+    @api_utils.bad_request_on_error("Invalid source environment: %s")
+    def validate_endpoint_source_environment(
+            self, ctxt, platform_name, source_env):
+        provider = providers_factory.get_provider(
+            platform_name, constants.PROVIDER_TYPE_EXPORT, None)
+        source_env_schema = provider.get_source_environment_schema()
+
+        is_valid = True
+        message = None
+        try:
+            schemas.validate_value(source_env, source_env_schema)
+        except exception.SchemaValidationException as ex:
+            is_valid = False
+            message = str(ex)
+
+        return (is_valid, message)
+
     def validate_endpoint_connection(self, ctxt, platform_name,
                                      connection_info):
         provider = providers_factory.get_provider(
@@ -354,6 +373,11 @@ class WorkerServerEndpoint(object):
             schema = provider.get_target_environment_schema()
             schemas["destination_environment_schema"] = schema
 
+        if provider_type in [constants.PROVIDER_TYPE_EXPORT,
+                             constants.PROVIDER_TYPE_REPLICA_EXPORT]:
+            schema = provider.get_source_environment_schema()
+            schemas["source_environment_schema"] = schema
+
         return schemas