Add dedicated schema for Replicator worker connection info.

coriolis/providers/replicator.py

@@ -459,13 +459,17 @@ class Replicator(object):
         port = conn_info.get('port', 22)
         password = conn_info.get('password', None)
         pkey = conn_info.get('pkey', None)
-        for i in required:
-            if conn_info.get(i) is None:
-                raise exception.CoriolisException(
-                    "missing required field: %s" % i)
+        missing = [
+            field for field in required
+            if field not in conn_info]
+        if missing:
+            raise exception.CoriolisException(
+                "Missing some required fields from source replication "
+                "worker VM connection info: %s" % missing)
         if any([password, pkey]) is False:
             raise exception.CoriolisException(
-                "Either password or pkey is required")
+                "Either 'password' or 'pkey' for source worker VM is required "
+                "to initialize the Coriolis replicator.")
 
         if pkey:
             if type(pkey) is str:

coriolis/schemas.py

@@ -38,6 +38,8 @@ _CORIOLIS_DISK_SYNC_RESOURCES_INFO_SCHEMA_NAME = (
     "disk_sync_resources_info_schema.json")
 _CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA_NAME = (
     "disk_sync_resources_conn_info_schema.json")
+_CORIOLIS_REPLICATION_WORKER_CONN_INFO_SCHEMA_NAME = (
+    "replication_worker_conn_info_schema.json")
 
 
 def get_schema(package_name, schema_name,
@@ -128,3 +130,6 @@ CORIOLIS_DISK_SYNC_RESOURCES_INFO_SCHEMA = get_schema(
 
 CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA = get_schema(
     __name__, _CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA_NAME)
+
+CORIOLIS_REPLICATION_WORKER_CONN_INFO_SCHEMA = get_schema(
+    __name__, _CORIOLIS_REPLICATION_WORKER_CONN_INFO_SCHEMA_NAME)

coriolis/schemas/replication_worker_conn_info_schema.json

@@ -0,0 +1,32 @@
+{
+  "$schema": "http://cloudbase.it/coriolis/schemas/replication_worker_conn_info#",
+  "type": "object",
+  "description": "Object defining SSH connection parameters for the source-side temporary minion VM which will carry out disk exports using Coriolis' in-built data replication engine.",
+  "properties": {
+    "ip": {
+      "type": "string"
+    },
+    "port": {
+      "type": "integer"
+    },
+    "username": {
+      "type": "string"
+    },
+    "password": {
+      "$ref": "#/definitions/nullableString"
+    },
+    "pkey": {
+      "$ref": "#/definitions/nullableString"
+    }
+  },
+  "required": ["ip", "port", "username"],
+  "definitions": {
+    "nullableString": {
+      "oneOf": [{
+        "type": "string"
+      }, {
+        "type": "null"
+      }]
+    }
+  }
+}

coriolis/tasks/base.py

@@ -129,21 +129,27 @@ def get_connection_info(ctxt, data):
     return utils.get_secret_connection_info(ctxt, connection_info)
 
 
-def marshal_migr_conn_info(migr_connection_info):
-    if migr_connection_info and "pkey" in migr_connection_info:
+def marshal_migr_conn_info(
+        migr_connection_info, private_key_field_name="pkey"):
+    if migr_connection_info and (
+            private_key_field_name in migr_connection_info):
         migr_connection_info = migr_connection_info.copy()
-        pkey = migr_connection_info["pkey"]
+        pkey = migr_connection_info[private_key_field_name]
         if isinstance(pkey, str) is False:
-            migr_connection_info["pkey"] = utils.serialize_key(
-                pkey, CONF.serialization.temp_keypair_password)
+            migr_connection_info[private_key_field_name] = (
+                utils.serialize_key(
+                    pkey, CONF.serialization.temp_keypair_password))
     return migr_connection_info
 
 
-def unmarshal_migr_conn_info(migr_connection_info):
-    if migr_connection_info and "pkey" in migr_connection_info:
+def unmarshal_migr_conn_info(
+        migr_connection_info, private_key_field_name="pkey"):
+    if migr_connection_info and (
+            private_key_field_name in migr_connection_info):
         migr_connection_info = migr_connection_info.copy()
-        pkey_str = migr_connection_info["pkey"]
+        pkey_str = migr_connection_info[private_key_field_name]
         if isinstance(pkey_str, paramiko.rsakey.RSAKey) is False:
-            migr_connection_info["pkey"] = utils.deserialize_key(
-                pkey_str, CONF.serialization.temp_keypair_password)
+            migr_connection_info[private_key_field_name] = (
+                utils.deserialize_key(
+                    pkey_str, CONF.serialization.temp_keypair_password))
     return migr_connection_info

coriolis/tasks/replica_tasks.py

@@ -148,11 +148,18 @@ class ReplicateDisksTask(base.TaskRunner):
         if migr_source_conn_info:
             schemas.validate_value(
                 migr_source_conn_info,
-                schemas.CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA)
-        migr_source_conn_info = base.unmarshal_migr_conn_info(
-            migr_source_conn_info)
+                schemas.CORIOLIS_REPLICATION_WORKER_CONN_INFO_SCHEMA)
+            migr_source_conn_info = base.unmarshal_migr_conn_info(
+                migr_source_conn_info)
 
         migr_target_conn_info = task_info["target_resources_connection_info"]
+        if migr_target_conn_info:
+            schemas.validate_value(
+                migr_target_conn_info,
+                schemas.CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA)
+            migr_target_conn_info['connection_details'] = (
+                base.unmarshal_migr_conn_info(
+                    migr_target_conn_info['connection_details']))
         incremental = task_info.get("incremental", True)
 
         source_environment = task_info['source_environment']
@@ -282,7 +289,7 @@ class DeployReplicaSourceResourcesTask(base.TaskRunner):
                     migr_connection_info)
                 schemas.validate_value(
                     migr_connection_info,
-                    schemas.CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA,
+                    schemas.CORIOLIS_REPLICATION_WORKER_CONN_INFO_SCHEMA,
                     # NOTE: we avoid raising so that the cleanup task
                     # can [try] to deal with the temporary resources.
                     raise_on_error=False)
@@ -390,8 +397,10 @@ class DeployReplicaTargetResourcesTask(base.TaskRunner):
                         str(err)))
 
             if migr_connection_info:
-                migr_connection_info = base.marshal_migr_conn_info(
-                    migr_connection_info)
+                if 'connection_details' in migr_connection_info:
+                    migr_connection_info['connection_details'] = (
+                        base.marshal_migr_conn_info(
+                            migr_connection_info['connection_details']))
                 schemas.validate_value(
                     migr_connection_info,
                     schemas.CORIOLIS_DISK_SYNC_RESOURCES_CONN_INFO_SCHEMA,