Added more explicit checks on return results of some tasks.

coriolis/exception.py

@@ -193,6 +193,10 @@ class TaskIsCancelling(InvalidTaskState):
     message = _(TASK_ALREADY_CANCELLING_EXCEPTION_FMT)
 
 
+class InvalidTaskResult(InvalidTaskState):
+    message = _('Task returned an invalid result.')
+
+
 class InvalidActionTasksExecutionState(Invalid):
     message = _("Invalid tasks execution state: %(reason)s")
 

coriolis/tasks/osmorphing_tasks.py

@@ -1,13 +1,19 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+from oslo_log import log as logging
+
 from coriolis import constants
+from coriolis import exception
+from coriolis import schemas
 from coriolis.osmorphing import manager as osmorphing_manager
 from coriolis.providers import factory as providers_factory
-from coriolis import schemas
 from coriolis.tasks import base
 
 
+LOG = logging.getLogger(__name__)
+
+
 class OSMorphingTask(base.TaskRunner):
 
     @property
@@ -79,20 +85,38 @@ class DeployOSMorphingResourcesTask(base.TaskRunner):
         import_info = provider.deploy_os_morphing_resources(
             ctxt, connection_info, instance_deployment_info)
 
-        result = {}
-        result["os_morphing_resources"] = import_info.get(
-            "os_morphing_resources")
-        result["osmorphing_info"] = import_info.get("osmorphing_info", {})
-        result["osmorphing_connection_info"] = base.marshal_migr_conn_info(
-            import_info["osmorphing_connection_info"])
-
         schemas.validate_value(
-            task_info, schemas.CORIOLIS_OS_MORPHING_RESOURCES_SCHEMA,
+            import_info, schemas.CORIOLIS_OS_MORPHING_RESOURCES_SCHEMA,
             # NOTE: we avoid raising so that the cleanup task
             # can [try] to deal with the temporary resources.
             raise_on_error=False)
 
-        return result
+        os_morphing_resources = import_info.get('os_morphing_resources')
+        if not os_morphing_resources:
+            raise exception.InvalidTaskResult(
+                "Target provider for '%s' did NOT return any "
+                "'os_morphing_resources'." % (
+                    destination["type"]))
+
+        osmorphing_connection_info = import_info.get(
+            'osmorphing_connection_info')
+        if not osmorphing_connection_info:
+            raise exception.InvalidTaskResult(
+                "Target provider '%s' did NOT return any "
+                "'osmorphing_connection_info'." % (
+                    destination["type"]))
+
+        os_morphing_info = import_info.get("osmorphing_info", {})
+        if not os_morphing_info:
+            LOG.warn(
+                "Target provider for '%s' did NOT return any "
+                "'osmorphing_info'. Defaulting to %s",
+                destination["type"], os_morphing_info)
+
+        return {
+            "os_morphing_resources": os_morphing_resources,
+            "osmorphing_connection_info": osmorphing_connection_info,
+            "osmorphing_info": os_morphing_info}
 
 
 class DeleteOSMorphingResourcesTask(base.TaskRunner):

coriolis/tasks/replica_tasks.py

@@ -37,11 +37,11 @@ def _check_ensure_volumes_info_ordering(export_info, volumes_info):
         matching_volumes = [
             vol for vol in volumes_info if vol['disk_id'] == disk_id]
         if not matching_volumes:
-            raise exception.CoriolisException(
+            raise exception.InvalidActionTasksExecutionState(
                 "Could not find source disk '%s' (ID '%s') in Replica "
                 "volumes info: %s" % (disk, disk_id, volumes_info))
         elif len(matching_volumes) > 1:
-            raise exception.CoriolisException(
+            raise exception.InvalidActionTasksExecutionState(
                 "Multiple disks with ID '%s' foind in Replica "
                 "volumes info: %s" % (disk_id, volumes_info))
 
@@ -272,19 +272,41 @@ class DeployReplicaSourceResourcesTask(base.TaskRunner):
         replica_resources_info = provider.deploy_replica_source_resources(
             ctxt, connection_info, source_environment)
 
-        migr_connection_info = replica_resources_info.get("connection_info")
-        if migr_connection_info:
-            migr_connection_info = base.marshal_migr_conn_info(
-                migr_connection_info)
-            schemas.validate_value(
-                migr_connection_info,
-                schemas.CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA,
-                # NOTE: we avoid raising so that the cleanup task
-                # can [try] to deal with the temporary resources.
-                raise_on_error=False)
+        migr_connection_info = replica_resources_info.get(
+            "connection_info", {})
+        if 'connection_info' not in replica_resources_info:
+            LOG.warn(
+                "Replica source provider for '%s' did NOT return any "
+                "'connection_info'. Defaulting to '%s'",
+                origin["type"], migr_connection_info)
+        else:
+            migr_connection_info = replica_resources_info['connection_info']
+            if migr_connection_info:
+                migr_connection_info = base.marshal_migr_conn_info(
+                    migr_connection_info)
+                schemas.validate_value(
+                    migr_connection_info,
+                    schemas.CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA,
+                    # NOTE: we avoid raising so that the cleanup task
+                    # can [try] to deal with the temporary resources.
+                    raise_on_error=False)
+            else:
+                LOG.warn(
+                    "Replica source provider for '%s' returned empty "
+                    "'connection_info' in source resources deployment: %s",
+                    origin["type"], migr_connection_info)
+
+        migr_resources = {}
+        if 'migr_resources' not in replica_resources_info:
+            LOG.warn(
+                "Replica source provider for '%s' did NOT return any "
+                "'migr_resources'. Defaulting to %s",
+                origin["type"], migr_resources)
+        else:
+            migr_resources = replica_resources_info['migr_resources']
 
         return {
-            "source_resources": replica_resources_info["migr_resources"],
+            "source_resources": migr_resources,
             "source_resources_connection_info": migr_connection_info}
 
 
@@ -350,21 +372,43 @@ class DeployReplicaTargetResourcesTask(base.TaskRunner):
             # can [try] to deal with the temporary resources.
             raise_on_error=False)
 
-        volumes_info = _check_ensure_volumes_info_ordering(
-            export_info, replica_resources_info["volumes_info"])
+        if "volumes_info" in replica_resources_info:
+            volumes_info = replica_resources_info["volumes_info"]
+            volumes_info = _check_ensure_volumes_info_ordering(
+                export_info, volumes_info)
+        else:
+            LOG.warn(
+                "Replica target provider for '%s' did not return any "
+                "'volumes_info'. Using the previous value of it.")
+
+        migr_connection_info = {}
+        if 'connection_info' in replica_resources_info:
+            migr_connection_info = replica_resources_info['connection_info']
+            try:
+                backup_writers.BackupWritersFactory(
+                    migr_connection_info, None).get_writer()
+            except Exception as err:
+                LOG.warn(
+                    "Seemingly invalid connection info. Replica will likely "
+                    "fail during disk Replication. Error is: %s" % str(err))
+        else:
+            LOG.warn(
+                "Replica target provider for '%s' did NOT return any "
+                "'connection_info'. Defaulting to %s",
+                destination["type"], migr_connection_info)
 
-        migr_connection_info = replica_resources_info["connection_info"]
-        try:
-            backup_writers.BackupWritersFactory(
-                migr_connection_info, None).get_writer()
-        except BaseException as err:
+        target_resources = {}
+        if 'migr_resources' not in replica_resources_info:
             LOG.warn(
-                "Seemingly invalid connection info. Replica will likely "
-                "fail during disk Replication. Error is: %s" % err)
+                "Replica target provider for '%s' did NOT return any "
+                "'migr_resources'. Defaulting to %s",
+                destination["type"], target_resources)
+        else:
+            target_resources = replica_resources_info["migr_resources"]
 
         return {
             "volumes_info": volumes_info,
-            "target_resources": replica_resources_info["migr_resources"],
+            "target_resources": target_resources,
             "target_resources_connection_info": migr_connection_info}
 
 
@@ -648,7 +692,7 @@ class ValidateReplicaExecutionDestinationInputsTask(base.TaskRunner):
 
         export_info = task_info.get("export_info")
         if not export_info:
-            raise exception.CoriolisException(
+            raise exception.InvalidActionTasksExecutionState(
                 "Instance export info is not set. Cannot perform "
                 "Replica Import validation for destination platform "
                 "'%s'" % destination_type)
@@ -727,7 +771,7 @@ class UpdateSourceReplicaTask(base.TaskRunner):
             origin["type"], constants.PROVIDER_TYPE_SOURCE_REPLICA_UPDATE,
             event_handler, raise_if_not_found=False)
         if not source_provider:
-            raise exception.CoriolisException(
+            raise exception.InvalidActionTasksExecutionState(
                 "Replica source provider plugin for '%s' does not support"
                 " updating Replicas" % origin["type"])
 
@@ -775,7 +819,7 @@ class UpdateDestinationReplicaTask(base.TaskRunner):
             constants.PROVIDER_TYPE_DESTINATION_REPLICA_UPDATE,
             event_handler, raise_if_not_found=False)
         if not destination_provider:
-            raise exception.CoriolisException(
+            raise exception.InvalidActionTasksExecutionState(
                 "Replica destination provider plugin for '%s' does not "
                 "support updating Replicas" % destination["type"])