Refactor input params validation task

Instead of having a single task for input parameters validation,
we have two separate atomic tasks to do input parameters
validation at the source and destination cloud providers.

coriolis/conductor/rpc/server.py

@@ -316,15 +316,25 @@ class ConductorServerEndpoint(object):
         execution.action = replica
 
         for instance in execution.action.instances:
-            validate_replica_inputs_task = self._create_task(
-                instance, constants.TASK_TYPE_VALIDATE_REPLICA_INPUTS,
-                execution)
-
             get_instance_info_task = self._create_task(
                 instance, constants.TASK_TYPE_GET_INSTANCE_INFO,
-                execution, depends_on=[validate_replica_inputs_task.id])
+                execution)
+
+            validate_replica_source_inputs_task = self._create_task(
+                instance,
+                constants.TASK_TYPE_VALIDATE_REPLICA_SOURCE_INPUTS,
+                execution,
+                depends_on=[get_instance_info_task.id])
 
-            depends_on = [get_instance_info_task.id]
+            validate_replica_destination_inputs_task = self._create_task(
+                instance,
+                constants.TASK_TYPE_VALIDATE_REPLICA_DESTINATION_INPUTS,
+                execution,
+                depends_on=[get_instance_info_task.id])
+
+            depends_on = [
+                validate_replica_source_inputs_task.id,
+                validate_replica_destination_inputs_task.id]
             if shutdown_instances:
                 shutdown_instance_task = self._create_task(
                     instance, constants.TASK_TYPE_SHUTDOWN_INSTANCE,
@@ -695,13 +705,20 @@ class ConductorServerEndpoint(object):
             migration, licensing_client.RESERVATION_TYPE_MIGRATION)
 
         for instance in instances:
-            task_validate = self._create_task(
-                instance, constants.TASK_TYPE_VALIDATE_MIGRATION_INPUTS,
+            task_validate_source = self._create_task(
+                instance,
+                constants.TASK_TYPE_VALIDATE_MIGRATION_SOURCE_INPUTS,
                 execution)
 
+            task_validate_destination = self._create_task(
+                instance,
+                constants.TASK_TYPE_VALIDATE_MIGRATION_DESTINATION_INPUTS,
+                execution,
+                depends_on=[task_validate_source.id])
+
             task_export = self._create_task(
                 instance, constants.TASK_TYPE_EXPORT_INSTANCE, execution,
-                depends_on=[task_validate.id])
+                depends_on=[task_validate_destination.id])
 
             if (constants.PROVIDER_TYPE_INSTANCE_FLAVOR in
                     destination_provider_types):

coriolis/constants.py

@@ -45,8 +45,13 @@ TASK_TYPE_CREATE_REPLICA_DISK_SNAPSHOTS = "CREATE_REPLICA_DISK_SNAPSHOTS"
 TASK_TYPE_DELETE_REPLICA_DISK_SNAPSHOTS = "DELETE_REPLICA_DISK_SNAPSHOTS"
 TASK_TYPE_RESTORE_REPLICA_DISK_SNAPSHOTS = "RESTORE_REPLICA_DISK_SNAPSHOTS"
 TASK_TYPE_GET_OPTIMAL_FLAVOR = "GET_OPTIMAL_FLAVOR"
-TASK_TYPE_VALIDATE_MIGRATION_INPUTS = "VALIDATE_MIGRATION_INPUTS"
-TASK_TYPE_VALIDATE_REPLICA_INPUTS = "VALIDATE_REPLICA_INPUTS"
+TASK_TYPE_VALIDATE_MIGRATION_SOURCE_INPUTS = (
+    "VALIDATE_MIGRATION_SOURCE_INPUTS")
+TASK_TYPE_VALIDATE_MIGRATION_DESTINATION_INPUTS = (
+    "VALIDATE_MIGRATION_DESTINATION_INPUTS")
+TASK_TYPE_VALIDATE_REPLICA_SOURCE_INPUTS = "VALIDATE_REPLICA_SOURCE_INPUTS"
+TASK_TYPE_VALIDATE_REPLICA_DESTINATION_INPUTS = (
+    "VALIDATE_REPLICA_DESTINATION_INPUTS")
 TASK_TYPE_VALIDATE_REPLICA_DEPLOYMENT_INPUTS = (
     "VALIDATE_REPLICA_DEPLOYMENT_INPUTS")
 TASK_TYPE_UPDATE_REPLICA = "UPDATE_REPLICA"

coriolis/tasks/factory.py

@@ -24,8 +24,10 @@ _TASKS_MAP = {
         migration_tasks.CleanupFailedImportInstanceTask,
     constants.TASK_TYPE_GET_OPTIMAL_FLAVOR:
         migration_tasks.GetOptimalFlavorTask,
-    constants.TASK_TYPE_VALIDATE_MIGRATION_INPUTS:
-        migration_tasks.ValidateMigrationParametersTask,
+    constants.TASK_TYPE_VALIDATE_MIGRATION_SOURCE_INPUTS:
+        migration_tasks.ValidateMigrationSourceInputsTask,
+    constants.TASK_TYPE_VALIDATE_MIGRATION_DESTINATION_INPUTS:
+        migration_tasks.ValidateMigrationDestinationInputsTask,
     constants.TASK_TYPE_DEPLOY_OS_MORPHING_RESOURCES:
         osmorphing_tasks.DeployOSMorphingResourcesTask,
     constants.TASK_TYPE_OS_MORPHING:
@@ -62,8 +64,10 @@ _TASKS_MAP = {
         replica_tasks.DeleteReplicaDiskSnapshotsTask,
     constants.TASK_TYPE_RESTORE_REPLICA_DISK_SNAPSHOTS:
         replica_tasks.RestoreReplicaDiskSnapshotsTask,
-    constants.TASK_TYPE_VALIDATE_REPLICA_INPUTS:
-        replica_tasks.ValidateReplicaExecutionParametersTask,
+    constants.TASK_TYPE_VALIDATE_REPLICA_SOURCE_INPUTS:
+        replica_tasks.ValidateReplicaExecutionSourceInputsTask,
+    constants.TASK_TYPE_VALIDATE_REPLICA_DESTINATION_INPUTS:
+        replica_tasks.ValidateReplicaExecutionDestinationInputsTask,
     constants.TASK_TYPE_VALIDATE_REPLICA_DEPLOYMENT_INPUTS:
         replica_tasks.ValidateReplicaDeploymentParametersTask,
     constants.TASK_TYPE_UPDATE_REPLICA:

coriolis/tasks/migration_tasks.py

@@ -192,28 +192,25 @@ class GetOptimalFlavorTask(base.TaskRunner):
         return task_info
 
 
-class ValidateMigrationParametersTask(base.TaskRunner):
+class ValidateMigrationSourceInputsTask(base.TaskRunner):
     def run(self, ctxt, instance, origin, destination, task_info,
             event_handler):
         event_manager = events.EventManager(event_handler)
-        # validate source params:
-        origin_type = origin["type"]
         origin_connection_info = base.get_connection_info(ctxt, origin)
-        destination_connection_info = base.get_connection_info(
-            ctxt, destination)
-        destination_type = destination["type"]
+        origin_type = origin["type"]
+
         source_provider = providers_factory.get_provider(
             origin_type, constants.PROVIDER_TYPE_VALIDATE_MIGRATION_EXPORT,
             event_handler, raise_if_not_found=False)
         export_info = None
         if source_provider:
             export_info = source_provider.validate_migration_export_input(
-                ctxt, base.get_connection_info(ctxt, origin), instance,
+                ctxt, origin_connection_info, instance,
                 source_environment=origin.get("source_environment", {}))
         else:
             event_manager.progress_update(
-                "Migration Export Provider for platform '%s' does not support "
-                "Migration input validation" % origin_type)
+                "Migration Export Provider for platform '%s' does not "
+                "support Migration input validation" % origin_type)
 
         if export_info is None:
             source_endpoint_provider = providers_factory.get_provider(
@@ -222,9 +219,7 @@ class ValidateMigrationParametersTask(base.TaskRunner):
             if not source_endpoint_provider:
                 event_manager.progress_update(
                     "Migration Export Provider for platform '%s' does not "
-                    "support querying instance export info. Cannot perform "
-                    "Migration Import validation for destination platform "
-                    "'%s'" % (origin_type, destination_type))
+                    "support querying instance export info" % origin_type)
                 return task_info
             export_info = source_endpoint_provider.get_instance(
                 ctxt, origin_connection_info, instance)
@@ -232,25 +227,42 @@ class ValidateMigrationParametersTask(base.TaskRunner):
         # validate Export info:
         schemas.validate_value(
             export_info, schemas.CORIOLIS_VM_EXPORT_INFO_SCHEMA)
-        # NOTE: this export info will get overriden with updated values
+        # NOTE: this export info will get overridden with updated values
         # and disk paths after the ExportInstanceTask.
         task_info["export_info"] = export_info
 
-        # validate destination params:
+        return task_info
+
+
+class ValidateMigrationDestinationInputsTask(base.TaskRunner):
+    def run(self, ctxt, instance, origin, destination, task_info,
+            event_handler):
+        event_manager = events.EventManager(event_handler)
+        destination_type = destination["type"]
+        if task_info.get("export_info") is None:
+            event_manager.progress_update(
+                "Instance export info is not set. Cannot perform Migration "
+                "Import validation for destination platform "
+                "'%s'" % destination_type)
+            return task_info
+
+        destination_connection_info = base.get_connection_info(
+            ctxt, destination)
         destination_provider = providers_factory.get_provider(
             destination_type,
             constants.PROVIDER_TYPE_VALIDATE_MIGRATION_IMPORT, event_handler,
             raise_if_not_found=False)
         if not destination_provider:
             event_manager.progress_update(
-                "Migration Import Provider for platform '%s' does not support "
-                "Migration input validation" % destination_type)
+                "Migration Import Provider for platform '%s' does not "
+                "support Migration input validation" % destination_type)
             return task_info
 
         # NOTE: the target environment JSON schema should have been validated
         # upon accepting the Migration API creation request.
         target_environment = destination.get("target_environment", {})
         destination_provider.validate_migration_import_input(
-            ctxt, destination_connection_info, target_environment, export_info)
+            ctxt, destination_connection_info, target_environment,
+            task_info["export_info"])
 
         return task_info

coriolis/tasks/replica_tasks.py

@@ -339,51 +339,35 @@ class RestoreReplicaDiskSnapshotsTask(base.TaskRunner):
         return task_info
 
 
-class ValidateReplicaExecutionParametersTask(base.TaskRunner):
+class ValidateReplicaExecutionSourceInputsTask(base.TaskRunner):
     def run(self, ctxt, instance, origin, destination, task_info,
             event_handler):
         event_manager = events.EventManager(event_handler)
-        # validate source params:
         origin_type = origin["type"]
-        origin_connection_info = base.get_connection_info(ctxt, origin)
-        destination_connection_info = base.get_connection_info(
-            ctxt, destination)
-        destination_type = destination["type"]
         source_provider = providers_factory.get_provider(
             origin_type, constants.PROVIDER_TYPE_VALIDATE_REPLICA_EXPORT,
             event_handler, raise_if_not_found=False)
-        export_info = None
-        if source_provider:
-            export_info = source_provider.validate_replica_export_input(
-                ctxt, base.get_connection_info(ctxt, origin), instance,
-                source_environment=origin.get("source_environment", {}))
-        else:
+        origin_connection_info = base.get_connection_info(ctxt, origin)
+        if not source_provider:
             event_manager.progress_update(
                 "Replica Export Provider for platform '%s' does not support "
                 "Replica input validation" % origin_type)
+        else:
+            source_provider.validate_replica_export_input(
+                ctxt, origin_connection_info, instance,
+                source_environment=origin.get("source_environment", {}))
 
-        if export_info is None:
-            source_endpoint_provider = providers_factory.get_provider(
-                origin_type, constants.PROVIDER_TYPE_ENDPOINT_INSTANCES,
-                event_handler, raise_if_not_found=False)
-            if not source_endpoint_provider:
-                event_manager.progress_update(
-                    "Replica Export Provider for platform '%s' does not "
-                    "support querying instance export info. Cannot perform "
-                    "Replica Import validation for destination platform "
-                    "'%s'" % (origin_type, destination_type))
-                return task_info
-            export_info = source_endpoint_provider.get_instance(
-                ctxt, origin_connection_info, instance)
+        return task_info
 
-        # validate Export info:
-        schemas.validate_value(
-            export_info, schemas.CORIOLIS_VM_EXPORT_INFO_SCHEMA)
-        # NOTE: this export info will get overriden with updated values
-        # and disk paths after the ExportInstanceTask.
-        task_info["export_info"] = export_info
 
-        # validate destination params:
+class ValidateReplicaExecutionDestinationInputsTask(base.TaskRunner):
+    def run(self, ctxt, instance, origin, destination, task_info,
+            event_handler):
+        event_manager = events.EventManager(event_handler)
+        destination_type = destination["type"]
+
+        destination_connection_info = base.get_connection_info(
+            ctxt, destination)
         destination_provider = providers_factory.get_provider(
             destination_type,
             constants.PROVIDER_TYPE_VALIDATE_REPLICA_IMPORT, event_handler,
@@ -394,11 +378,19 @@ class ValidateReplicaExecutionParametersTask(base.TaskRunner):
                 "Replica input validation" % destination_type)
             return task_info
 
+        export_info = task_info.get("export_info")
+        if not export_info:
+            raise exception.CoriolisException(
+                "Instance export info is not set. Cannot perform "
+                "Replica Import validation for destination platform "
+                "'%s'" % destination_type)
+
         # NOTE: the target environment JSON schema should have been validated
         # upon accepting the Replica API creation request.
         target_environment = destination.get("target_environment", {})
         destination_provider.validate_replica_import_input(
-            ctxt, destination_connection_info, target_environment, export_info)
+            ctxt, destination_connection_info, target_environment,
+            export_info)
 
         return task_info