Merged in alexpilotti/coriolis/bp/gui-support (pull request #41)

Adds endpoints

coriolis/api/common.py

@@ -0,0 +1,12 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis import utils
+
+
+def get_paging_params(req):
+    marker = req.GET.get("marker")
+    limit = req.GET.get("limit")
+    if limit is not None:
+        limit = utils.parse_int_value(limit)
+    return marker, limit

coriolis/api/v1/endpoint_instances.py

@@ -0,0 +1,30 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from oslo_log import log as logging
+
+from coriolis.api import common
+from coriolis.api import wsgi as api_wsgi
+from coriolis.api.v1.views import endpoint_instance_view
+from coriolis.endpoint_instances import api
+
+LOG = logging.getLogger(__name__)
+
+
+class EndpointInstanceController(api_wsgi.Controller):
+    def __init__(self):
+        self._instance_api = api.API()
+        super(EndpointInstanceController, self).__init__()
+
+    def index(self, req, endpoint_id):
+        marker, limit = common.get_paging_params(req)
+        instance_name_pattern = req.GET.get("name")
+
+        return endpoint_instance_view.collection(
+            req, self._instance_api.get_endpoint_instances(
+                req.environ['coriolis.context'], endpoint_id, marker, limit,
+                instance_name_pattern))
+
+
+def create_resource():
+    return api_wsgi.Resource(EndpointInstanceController())

coriolis/api/v1/endpoints.py

@@ -0,0 +1,66 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from webob import exc
+
+from oslo_log import log as logging
+
+from coriolis.api import wsgi as api_wsgi
+from coriolis.api.v1.views import endpoint_view
+from coriolis import exception
+from coriolis.endpoints import api
+
+LOG = logging.getLogger(__name__)
+
+
+class EndpointController(api_wsgi.Controller):
+    def __init__(self):
+        self._endpoint_api = api.API()
+        super(EndpointController, self).__init__()
+
+    def show(self, req, id):
+        endpoint = self._endpoint_api.get_endpoint(
+            req.environ["coriolis.context"], id)
+        if not endpoint:
+            raise exc.HTTPNotFound()
+
+        return endpoint_view.single(req, endpoint)
+
+    def index(self, req):
+        return endpoint_view.collection(
+            req, self._endpoint_api.get_endpoints(
+                req.environ['coriolis.context']))
+
+    def _validate_create_body(self, body):
+        try:
+            endpoint = body["endpoint"]
+            name = endpoint["name"]
+            description = endpoint.get("description")
+            endpoint_type = endpoint["type"]
+            connection_info = endpoint["connection_info"]
+            return name, endpoint_type, description, connection_info
+        except Exception as ex:
+            LOG.exception(ex)
+            if hasattr(ex, "message"):
+                msg = ex.message
+            else:
+                msg = str(ex)
+            raise exception.InvalidInput(msg)
+
+    def create(self, req, body):
+        (name, endpoint_type, description,
+         connection_info) = self._validate_create_body(body)
+        return endpoint_view.single(req, self._endpoint_api.create(
+            req.environ['coriolis.context'], name, endpoint_type, description,
+            connection_info))
+
+    def delete(self, req, id):
+        try:
+            self._endpoint_api.delete(req.environ['coriolis.context'], id)
+            raise exc.HTTPNoContent()
+        except exception.NotFound as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+
+
+def create_resource():
+    return api_wsgi.Resource(EndpointController())

coriolis/api/v1/migrations.py

@@ -40,23 +40,13 @@ class MigrationController(api_wsgi.Controller):
 
     def _validate_migration_input(self, migration):
         try:
-            origin = migration["origin"]
-            destination = migration["destination"]
+            origin_endpoint_id = migration["origin_endpoint_id"]
+            destination_endpoint_id = migration["destination_endpoint_id"]
+            destination_environment = migration.get("destination_environment")
+            instances = migration["instances"]
 
-            export_provider = factory.get_provider(
-                origin["type"], constants.PROVIDER_TYPE_EXPORT, None)
-            export_provider.validate_connection_info(
-                origin.get("connection_info", {}))
-
-            import_provider = factory.get_provider(
-                destination["type"], constants.PROVIDER_TYPE_IMPORT, None)
-            import_provider.validate_connection_info(
-                destination.get("connection_info", {}))
-
-            import_provider.validate_target_environment(
-                destination.get("target_environment", {}))
-
-            return origin, destination, migration["instances"]
+            return (origin_endpoint_id, destination_endpoint_id,
+                    destination_environment, instances)
         except Exception as ex:
             LOG.exception(ex)
             if hasattr(ex, "message"):
@@ -78,10 +68,14 @@ class MigrationController(api_wsgi.Controller):
             migration = self._migration_api.deploy_replica_instances(
                 context, replica_id, clone_disks, force)
         else:
-            origin, destination, instances = self._validate_migration_input(
+            (origin_endpoint_id,
+             destination_endpoint_id,
+             destination_environment,
+             instances) = self._validate_migration_input(
                 migration_body)
             migration = self._migration_api.migrate_instances(
-                context, origin, destination, instances)
+                context, origin_endpoint_id, destination_endpoint_id,
+                destination_environment, instances)
 
         return migration_view.single(req, migration)
 

coriolis/api/v1/replicas.py

@@ -44,23 +44,13 @@ class ReplicaController(api_wsgi.Controller):
         try:
             replica = body["replica"]
 
-            origin = replica["origin"]
-            destination = replica["destination"]
+            origin_endpoint_id = replica["origin_endpoint_id"]
+            destination_endpoint_id = replica["destination_endpoint_id"]
+            destination_environment = replica.get("destination_environment")
+            instances = replica["instances"]
 
-            export_provider = factory.get_provider(
-                origin["type"], constants.PROVIDER_TYPE_EXPORT, None)
-            export_provider.validate_connection_info(
-                origin.get("connection_info", {}))
-
-            import_provider = factory.get_provider(
-                destination["type"], constants.PROVIDER_TYPE_IMPORT, None)
-            import_provider.validate_connection_info(
-                destination.get("connection_info", {}))
-
-            import_provider.validate_target_environment(
-                destination.get("target_environment", {}))
-
-            return origin, destination, replica["instances"]
+            return (origin_endpoint_id, destination_endpoint_id,
+                    destination_environment, instances)
         except Exception as ex:
             LOG.exception(ex)
             if hasattr(ex, "message"):
@@ -70,9 +60,12 @@ class ReplicaController(api_wsgi.Controller):
             raise exception.InvalidInput(msg)
 
     def create(self, req, body):
-        origin, destination, instances = self._validate_create_body(body)
+        (origin_endpoint_id, destination_endpoint_id,
+         destination_environment, instances) = self._validate_create_body(body)
+
         return replica_view.single(req, self._replica_api.create(
-            req.environ['coriolis.context'], origin, destination, instances))
+            req.environ['coriolis.context'], origin_endpoint_id,
+            destination_endpoint_id, destination_environment, instances))
 
     def delete(self, req, id):
         try:

coriolis/api/v1/router.py

@@ -4,6 +4,8 @@
 from oslo_log import log as logging
 
 from coriolis import api
+from coriolis.api.v1 import endpoints
+from coriolis.api.v1 import endpoint_instances
 from coriolis.api.v1 import migrations
 from coriolis.api.v1 import migration_actions
 from coriolis.api.v1 import replica_actions
@@ -28,6 +30,17 @@ class APIRouter(api.APIRouter):
     def _setup_routes(self, mapper, ext_mgr):
         mapper.redirect("", "/")
 
+        self.resources['endpoints'] = endpoints.create_resource()
+        mapper.resource('endpoint', 'endpoints',
+                        controller=self.resources['endpoints'],
+                        collection={'detail': 'GET'},
+                        member={'action': 'POST'})
+
+        self.resources['endpoint_instances'] = \
+            endpoint_instances.create_resource()
+        mapper.resource('instance', 'endpoints/{endpoint_id}/instances',
+                        controller=self.resources['endpoint_instances'])
+
         self.resources['migrations'] = migrations.create_resource()
         mapper.resource('migration', 'migrations',
                         controller=self.resources['migrations'],

coriolis/api/v1/views/endpoint_instance_view.py

@@ -0,0 +1,24 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import itertools
+
+
+def _format_instance(req, instance, keys=None):
+    def transform(key, value):
+        if keys and key not in keys:
+            return
+        yield (key, value)
+
+    return dict(itertools.chain.from_iterable(
+        transform(k, v) for k, v in instance.items()))
+
+
+def single(req, instance):
+    return {"instance": _format_instance(req, instance)}
+
+
+def collection(req, instances):
+    formatted_instances = [_format_instance(req, m)
+                           for m in instances]
+    return {'instances': formatted_instances}

coriolis/api/v1/views/endpoint_view.py

@@ -0,0 +1,24 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import itertools
+
+
+def _format_endpoint(req, endpoint, keys=None):
+    def transform(key, value):
+        if keys and key not in keys:
+            return
+        yield (key, value)
+
+    return dict(itertools.chain.from_iterable(
+        transform(k, v) for k, v in endpoint.items()))
+
+
+def single(req, endpoint):
+    return {"endpoint": _format_endpoint(req, endpoint)}
+
+
+def collection(req, endpoints):
+    formatted_endpoints = [_format_endpoint(req, m)
+                           for m in endpoints]
+    return {'endpoints': formatted_endpoints}

coriolis/conductor/rpc/client.py

@@ -13,6 +13,33 @@ class ConductorClient(object):
         target = messaging.Target(topic='coriolis_conductor', version=VERSION)
         self._client = rpc.get_client(target)
 
+    def create_endpoint(self, ctxt, name, endpoint_type, description,
+                        connection_info):
+        return self._client.call(
+            ctxt, 'create_endpoint', name=name, endpoint_type=endpoint_type,
+            description=description, connection_info=connection_info)
+
+    def get_endpoints(self, ctxt):
+        return self._client.call(
+            ctxt, 'get_endpoints')
+
+    def get_endpoint(self, ctxt, endpoint_id):
+        return self._client.call(
+            ctxt, 'get_endpoint', endpoint_id=endpoint_id)
+
+    def delete_endpoint(self, ctxt, endpoint_id):
+        return self._client.call(
+            ctxt, 'delete_endpoint', endpoint_id=endpoint_id)
+
+    def get_endpoint_instances(self, ctxt, endpoint_id, marker=None,
+                               limit=None, instance_name_pattern=None):
+        return self._client.call(
+            ctxt, 'get_endpoint_instances',
+            endpoint_id=endpoint_id,
+            marker=marker,
+            limit=limit,
+            instance_name_pattern=instance_name_pattern)
+
     def execute_replica_tasks(self, ctxt, replica_id,
                               shutdown_instances=False):
         return self._client.call(
@@ -42,10 +69,16 @@ class ConductorClient(object):
             ctxt, 'cancel_replica_tasks_execution', replica_id=replica_id,
             execution_id=execution_id, force=force)
 
-    def create_instances_replica(self, ctxt, origin, destination, instances):
+    def create_instances_replica(self, ctxt, origin_endpoint_id,
+                                 destination_endpoint_id,
+                                 destination_environment,
+                                 instances):
         return self._client.call(
-            ctxt, 'create_instances_replica', origin=origin,
-            destination=destination, instances=instances)
+            ctxt, 'create_instances_replica',
+            origin_endpoint_id=origin_endpoint_id,
+            destination_endpoint_id=destination_endpoint_id,
+            destination_environment=destination_environment,
+            instances=instances)
 
     def get_replicas(self, ctxt, include_tasks_executions=False):
         return self._client.call(
@@ -72,9 +105,14 @@ class ConductorClient(object):
         return self._client.call(
             ctxt, 'get_migration', migration_id=migration_id)
 
-    def migrate_instances(self, ctxt, origin, destination, instances):
+    def migrate_instances(self, ctxt, origin_endpoint_id,
+                          destination_endpoint_id, destination_environment,
+                          instances):
         return self._client.call(
-            ctxt, 'migrate_instances', origin=origin, destination=destination,
+            ctxt, 'migrate_instances',
+            origin_endpoint_id=origin_endpoint_id,
+            destination_endpoint_id=destination_endpoint_id,
+            destination_environment=destination_environment,
             instances=instances)
 
     def deploy_replica_instances(self, ctxt, replica_id, clone_disks=False,

coriolis/conductor/rpc/server.py

@@ -12,6 +12,8 @@ from coriolis.db import api as db_api
 from coriolis.db.sqlalchemy import models
 from coriolis import exception
 from coriolis import keystone
+from coriolis.providers import factory as providers_factory
+from coriolis import schemas
 from coriolis import utils
 from coriolis.worker.rpc import client as rpc_worker_client
 
@@ -20,6 +22,16 @@ VERSION = "1.0"
 LOG = logging.getLogger(__name__)
 
 
+def endpoint_synchronized(func):
+    @functools.wraps(func)
+    def wrapper(self, ctxt, endpoint_id, *args, **kwargs):
+        @lockutils.synchronized(endpoint_id)
+        def inner():
+            return func(self, ctxt, endpoint_id, *args, **kwargs)
+        return inner()
+    return wrapper
+
+
 def replica_synchronized(func):
     @functools.wraps(func)
     def wrapper(self, ctxt, replica_id, *args, **kwargs):
@@ -64,6 +76,51 @@ class ConductorServerEndpoint(object):
     def __init__(self):
         self._rpc_worker_client = rpc_worker_client.WorkerClient()
 
+    def create_endpoint(self, ctxt, name, endpoint_type, description,
+                        connection_info):
+        endpoint = models.Endpoint()
+        endpoint.name = name
+        endpoint.type = endpoint_type
+        endpoint.description = description
+        endpoint.connection_info = connection_info
+
+        db_api.add_endpoint(ctxt, endpoint)
+        LOG.info("Endpoint created: %s", endpoint.id)
+        return self.get_endpoint(ctxt, endpoint.id)
+
+    def get_endpoints(self, ctxt):
+        return db_api.get_endpoints(ctxt)
+
+    @endpoint_synchronized
+    def get_endpoint(self, ctxt, endpoint_id):
+        endpoint = db_api.get_endpoint(ctxt, endpoint_id)
+        if not endpoint:
+            raise exception.NotFound("Endpoint not found")
+        return endpoint
+
+    @endpoint_synchronized
+    def delete_endpoint(self, ctxt, endpoint_id):
+        db_api.delete_endpoint(ctxt, endpoint_id)
+
+    def get_endpoint_instances(self, ctxt, endpoint_id, marker, limit,
+                               instance_name_pattern):
+        endpoint = self.get_endpoint(ctxt, endpoint_id)
+
+        export_provider = providers_factory.get_provider(
+            endpoint.type, constants.PROVIDER_TYPE_ENDPOINT, None)
+
+        connection_info = utils.get_secret_connection_info(
+            ctxt, endpoint.connection_info)
+
+        instances_info = export_provider.get_instances(
+            ctxt, connection_info, last_seen_id=marker, limit=limit,
+            instance_name_pattern=instance_name_pattern)
+        for instance_info in instances_info:
+            schemas.validate_value(
+                instance_info, schemas.CORIOLIS_VM_INSTANCE_INFO_SCHEMA)
+
+        return instances_info
+
     @staticmethod
     def _create_task(instance, task_type, execution, depends_on=None,
                      on_error=False):
@@ -87,9 +144,27 @@ class ConductorServerEndpoint(object):
                         break
         return task
 
+    def _get_task_origin(self, ctxt, action):
+        endpoint = self.get_endpoint(ctxt, action.origin_endpoint_id)
+        return {
+            "connection_info": endpoint.connection_info,
+            "type": endpoint.type
+        }
+
+    def _get_task_destination(self, ctxt, action):
+        endpoint = self.get_endpoint(ctxt, action.destination_endpoint_id)
+        return {
+            "connection_info": endpoint.connection_info,
+            "type": endpoint.type,
+            "target_environment": action.destination_environment
+        }
+
     def _begin_tasks(self, ctxt, execution, task_info={}):
         keystone.create_trust(ctxt)
 
+        origin = self._get_task_origin(ctxt, execution.action)
+        destination = self._get_task_destination(ctxt, execution.action)
+
         for task in execution.tasks:
             if (not task.depends_on and
                     task.status == constants.TASK_STATUS_PENDING):
@@ -97,8 +172,8 @@ class ConductorServerEndpoint(object):
                     ctxt, server=None,
                     task_id=task.id,
                     task_type=task.task_type,
-                    origin=execution.action.origin,
-                    destination=execution.action.destination,
+                    origin=origin,
+                    destination=destination,
                     instance=task.instance,
                     task_info=task_info.get(task.instance, {}))
 
@@ -243,18 +318,24 @@ class ConductorServerEndpoint(object):
         return self.get_replica_tasks_execution(ctxt, replica_id, execution.id)
 
     @staticmethod
-    def _check_endpoints(ctxt, origin, destination):
+    def _check_endpoints(ctxt, origin_endpoint, destination_endpoint):
         # TODO(alexpilotti): check Barbican secrets content as well
-        if origin.get("connection_info") == destination.get("connection_info"):
+        if (origin_endpoint.connection_info ==
+                destination_endpoint.connection_info):
             raise exception.SameDestination()
 
-    def create_instances_replica(self, ctxt, origin, destination, instances):
-        self._check_endpoints(ctxt, origin, destination)
+    def create_instances_replica(self, ctxt, origin_endpoint_id,
+                                 destination_endpoint_id,
+                                 destination_environment, instances):
+        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)
 
         replica = models.Replica()
         replica.id = str(uuid.uuid4())
-        replica.origin = origin
-        replica.destination = destination
+        replica.origin_endpoint = origin_endpoint
+        replica.destination_endpoint = destination_endpoint
+        replica.destination_environment = destination_environment
         replica.instances = instances
         replica.executions = []
         replica.info = {}
@@ -331,8 +412,9 @@ class ConductorServerEndpoint(object):
 
         migration = models.Migration()
         migration.id = str(uuid.uuid4())
-        migration.origin = replica.origin
-        migration.destination = replica.destination
+        migration.origin_endpoint_id = replica.origin_endpoint_id
+        migration.destination_endpoint_id = replica.destination_endpoint_id
+        migration.destination_environment = replica.destination_environment
         migration.instances = instances
         migration.replica = replica
         migration.info = replica.info
@@ -387,13 +469,18 @@ class ConductorServerEndpoint(object):
 
         return self.get_migration(ctxt, migration.id)
 
-    def migrate_instances(self, ctxt, origin, destination, instances):
-        self._check_endpoints(ctxt, origin, destination)
+    def migrate_instances(self, ctxt, origin_endpoint_id,
+                          destination_endpoint_id, destination_environment,
+                          instances):
+        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)
 
         migration = models.Migration()
         migration.id = str(uuid.uuid4())
-        migration.origin = origin
-        migration.destination = destination
+        migration.origin_endpoint = origin_endpoint
+        migration.destination_endpoint = destination_endpoint
+        migration.destination_environment = destination_environment
         execution = models.TasksExecution()
         execution.status = constants.EXECUTION_STATUS_RUNNING
         execution.number = 1
@@ -467,23 +554,32 @@ class ConductorServerEndpoint(object):
                     ctxt, task.id, constants.TASK_STATUS_CANCELED)
 
         if not has_running_tasks:
-            for task in execution.tasks:
-                if task.status in [constants.TASK_STATUS_PENDING,
-                                   constants.TASK_STATUS_ON_ERROR_ONLY]:
-                    if task.on_error:
-                        action = db_api.get_action(ctxt, execution.action_id)
-                        task_info = action.info.get(task.instance, {})
-
-                        self._rpc_worker_client.begin_task(
-                            ctxt, server=None,
-                            task_id=task.id,
-                            task_type=task.task_type,
-                            origin=action.origin,
-                            destination=action.destination,
-                            instance=task.instance,
-                            task_info=task_info)
-
-                        has_running_tasks = True
+            try:
+                origin = self._get_task_origin(ctxt, execution.action)
+                destination = self._get_task_destination(
+                    ctxt, execution.action)
+
+                for task in execution.tasks:
+                    if task.status in [constants.TASK_STATUS_PENDING,
+                                       constants.TASK_STATUS_ON_ERROR_ONLY]:
+                        if task.on_error:
+                            action = db_api.get_action(
+                                ctxt, execution.action_id)
+                            task_info = action.info.get(task.instance, {})
+
+                            self._rpc_worker_client.begin_task(
+                                ctxt, server=None,
+                                task_id=task.id,
+                                task_type=task.task_type,
+                                origin=origin,
+                                destination=destination,
+                                instance=task.instance,
+                                task_info=task_info)
+
+                            has_running_tasks = True
+            except exception.NotFound as ex:
+                LOG.error("A required endpoint could not be found")
+                LOG.exception(ex)
 
         if not has_running_tasks:
             self._set_tasks_execution_status(
@@ -503,6 +599,9 @@ class ConductorServerEndpoint(object):
             ctxt, task_id, constants.TASK_STATUS_RUNNING)
 
     def _start_pending_tasks(self, ctxt, execution, parent_task, task_info):
+        origin = self._get_task_origin(ctxt, execution.action)
+        destination = self._get_task_destination(ctxt, execution.action)
+
         for task in execution.tasks:
             if task.status == constants.TASK_STATUS_PENDING:
                 if task.depends_on and parent_task.id in task.depends_on:
@@ -521,13 +620,12 @@ class ConductorServerEndpoint(object):
                                 constants.TASK_TYPE_IMPORT_INSTANCE):
                             server = parent_task.host
 
-                        action = execution.action
                         self._rpc_worker_client.begin_task(
                             ctxt, server=server,
                             task_id=task.id,
                             task_type=task.task_type,
-                            origin=action.origin,
-                            destination=action.destination,
+                            origin=origin,
+                            destination=destination,
                             instance=task.instance,
                             task_info=task_info)
 

coriolis/constants.py

@@ -41,6 +41,7 @@ PROVIDER_TYPE_IMPORT = 1
 PROVIDER_TYPE_EXPORT = 2
 PROVIDER_TYPE_REPLICA_IMPORT = 4
 PROVIDER_TYPE_REPLICA_EXPORT = 8
+PROVIDER_TYPE_ENDPOINT = 16
 
 DISK_FORMAT_VMDK = 'vmdk'
 DISK_FORMAT_RAW = 'raw'

coriolis/db/api.py

@@ -58,6 +58,36 @@ def _soft_delete_aware_query(context, *args, **kwargs):
     return query
 
 
+@enginefacade.reader
+def get_endpoints(context):
+    q = _soft_delete_aware_query(context, models.Endpoint)
+    return q.filter(
+        models.Replica.project_id == context.tenant).all()
+
+
+@enginefacade.reader
+def get_endpoint(context, endpoint_id):
+    q = _soft_delete_aware_query(context, models.Endpoint)
+    return q.filter(
+        models.Endpoint.project_id == context.tenant,
+        models.Endpoint.id == endpoint_id).first()
+
+
+@enginefacade.writer
+def add_endpoint(context, endpoint):
+    endpoint.user_id = context.user
+    endpoint.project_id = context.tenant
+    context.session.add(endpoint)
+
+
+@enginefacade.writer
+def delete_endpoint(context, endpoint_id):
+    count = _soft_delete_aware_query(context, models.Endpoint).filter_by(
+        project_id=context.tenant, id=endpoint_id).soft_delete()
+    if count == 0:
+        raise exception.NotFound("0 entries were soft deleted")
+
+
 @enginefacade.reader
 def get_replica_tasks_executions(context, replica_id, include_tasks=False):
     q = _soft_delete_aware_query(context, models.TasksExecution)
@@ -173,10 +203,11 @@ def get_migrations(context, include_tasks=False):
 
 def _get_tasks_with_details_options(query):
     return query.options(
-        orm.joinedload("tasks").
-        joinedload("progress_updates")).options(
+        orm.joinedload("action")).options(
             orm.joinedload("tasks").
-            joinedload("events"))
+            joinedload("progress_updates")).options(
+                orm.joinedload("tasks").
+                joinedload("events"))
 
 
 def _get_migration_task_query_options(query):

coriolis/db/sqlalchemy/migrate_repo/versions/002_adds_endpoints.py

@@ -0,0 +1,65 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import uuid
+
+import sqlalchemy
+
+
+def upgrade(migrate_engine):
+    meta = sqlalchemy.MetaData()
+    meta.bind = migrate_engine
+
+    endpoint = sqlalchemy.Table(
+        'endpoint', meta,
+        sqlalchemy.Column('id', sqlalchemy.String(36), primary_key=True,
+                          default=lambda: str(uuid.uuid4())),
+        sqlalchemy.Column('created_at', sqlalchemy.DateTime),
+        sqlalchemy.Column('updated_at', sqlalchemy.DateTime),
+        sqlalchemy.Column('deleted_at', sqlalchemy.DateTime),
+        sqlalchemy.Column('deleted', sqlalchemy.String(36)),
+        sqlalchemy.Column("user_id", sqlalchemy.String(255), nullable=False),
+        sqlalchemy.Column("project_id", sqlalchemy.String(255),
+                          nullable=False),
+        sqlalchemy.Column("connection_info", sqlalchemy.Text, nullable=False),
+        sqlalchemy.Column("type", sqlalchemy.String(255), nullable=False),
+        sqlalchemy.Column("name", sqlalchemy.String(255), nullable=False),
+        sqlalchemy.Column("description", sqlalchemy.Text),
+        mysql_engine='InnoDB',
+        mysql_charset='utf8'
+    )
+
+    tables = (
+        endpoint,
+    )
+
+    for index, table in enumerate(tables):
+        try:
+            table.create()
+        except Exception:
+            # If an error occurs, drop all tables created so far to return
+            # to the previously existing state.
+            meta.drop_all(tables=tables[:index])
+            raise
+
+    base_transfer_action = sqlalchemy.Table(
+        'base_transfer_action', meta, autoload=True)
+
+    # NOTE(alexpilotti) delete all records in base_transfer_action
+    # before performing this migration
+    origin_endpoint_id = sqlalchemy.Column(
+        "origin_endpoint_id", sqlalchemy.String(36),
+        sqlalchemy.ForeignKey('endpoint.id'), nullable=False)
+    base_transfer_action.create_column(origin_endpoint_id)
+
+    destination_endpoint_id = sqlalchemy.Column(
+        "destination_endpoint_id", sqlalchemy.String(36),
+        sqlalchemy.ForeignKey('endpoint.id'), nullable=False)
+    base_transfer_action.create_column(destination_endpoint_id)
+
+    destination_environment = sqlalchemy.Column(
+        "destination_environment", sqlalchemy.Text, nullable=True)
+    base_transfer_action.create_column(destination_environment)
+
+    base_transfer_action.drop_column("origin")
+    base_transfer_action.drop_column("destination")

coriolis/db/sqlalchemy/models.py

@@ -95,8 +95,7 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
                                 primary_key=True)
     user_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
     project_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
-    origin = sqlalchemy.Column(types.Json, nullable=False)
-    destination = sqlalchemy.Column(types.Json, nullable=False)
+    destination_environment = sqlalchemy.Column(types.Json, nullable=True)
     type = sqlalchemy.Column(sqlalchemy.String(50))
     executions = orm.relationship(TasksExecution, cascade="all,delete",
                                   backref=orm.backref('action'),
@@ -105,6 +104,12 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
                                   "TasksExecution.deleted=='0')")
     instances = sqlalchemy.Column(types.List, nullable=False)
     info = sqlalchemy.Column(types.Json, nullable=False)
+    origin_endpoint_id = sqlalchemy.Column(
+        sqlalchemy.String(36),
+        sqlalchemy.ForeignKey('endpoint.id'), nullable=False)
+    destination_endpoint_id = sqlalchemy.Column(
+        sqlalchemy.String(36),
+        sqlalchemy.ForeignKey('endpoint.id'), nullable=False)
 
     __mapper_args__ = {
         'polymorphic_identity': 'base_transfer_action',
@@ -141,3 +146,26 @@ class Migration(BaseTransferAction):
     __mapper_args__ = {
         'polymorphic_identity': 'migration',
     }
+
+
+class Endpoint(BASE, models.TimestampMixin, models.ModelBase,
+               models.SoftDeleteMixin):
+    __tablename__ = 'endpoint'
+
+    id = sqlalchemy.Column(sqlalchemy.String(36),
+                           default=lambda: str(uuid.uuid4()),
+                           primary_key=True)
+    user_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    project_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    connection_info = sqlalchemy.Column(types.Json, nullable=False)
+    type = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    description = sqlalchemy.Column(sqlalchemy.String(1024), nullable=True)
+    origin_actions = orm.relationship(
+        BaseTransferAction, backref=orm.backref('origin_endpoint'),
+        primaryjoin="and_(BaseTransferAction.origin_endpoint_id==Endpoint.id, "
+        "BaseTransferAction.deleted=='0')")
+    destination_actions = orm.relationship(
+        BaseTransferAction, backref=orm.backref('destination_endpoint'),
+        primaryjoin="and_(BaseTransferAction.destination_endpoint_id=="
+        "Endpoint.id, BaseTransferAction.deleted=='0')")

coriolis/endpoint_instances/api.py

@@ -0,0 +1,14 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.conductor.rpc import client as rpc_client
+
+
+class API(object):
+    def __init__(self):
+        self._rpc_client = rpc_client.ConductorClient()
+
+    def get_endpoint_instances(self, ctxt, endpoint_id, marker=None,
+                               limit=None, instance_name_pattern=None):
+        return self._rpc_client.get_endpoint_instances(
+            ctxt, endpoint_id, marker, limit, instance_name_pattern)

coriolis/endpoints/api.py

@@ -0,0 +1,23 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.conductor.rpc import client as rpc_client
+
+
+class API(object):
+    def __init__(self):
+        self._rpc_client = rpc_client.ConductorClient()
+
+    def create(self, ctxt, name, endpoint_type, description,
+               connection_info):
+        return self._rpc_client.create_endpoint(
+            ctxt, name, endpoint_type, description, connection_info)
+
+    def delete(self, ctxt, endpoint_id):
+        self._rpc_client.delete_endpoint(ctxt, endpoint_id)
+
+    def get_endpoints(self, ctxt):
+        return self._rpc_client.get_endpoints(ctxt)
+
+    def get_endpoint(self, ctxt, endpoint_id):
+        return self._rpc_client.get_endpoint(ctxt, endpoint_id)

coriolis/keystone.py

@@ -105,19 +105,28 @@ def create_keystone_session(ctxt, connection_info={}):
     verify = not allow_untrusted
 
     username = connection_info.get("username")
+    auth = None
 
     if not username:
         # Using directly the caller's token is not feasible for long running
         # tasks as once it expires it cannot be automatically renewed. This is
         # solved by using a Keystone trust, which must have been set priorly.
-        if not ctxt.trust_id:
-            raise exception.InvalidConfigurationValue(
-                "Trust id not set in context")
-
-        auth = _get_trusts_auth_plugin(ctxt.trust_id)
-        session = ks_session.Session(auth=auth, verify=verify)
+        if ctxt.trust_id:
+            auth = _get_trusts_auth_plugin(ctxt.trust_id)
+        else:
+            plugin_name = "token"
+            plugin_args = {
+                "token": ctxt.auth_token
+            }
     else:
+        plugin_name = "password"
         password = connection_info.get("password")
+        plugin_args = {
+            "username": username,
+            "password": password,
+        }
+
+    if not auth:
         project_name = connection_info.get("project_name", ctxt.project_name)
 
         auth_url = connection_info.get("auth_url", CONF.keystone.auth_url)
@@ -127,14 +136,10 @@ def create_keystone_session(ctxt, connection_info={}):
                 '"auth_url" in group "[openstack_migration_provider]" '
                 'not set')
 
-        plugin_name = "password"
-
-        plugin_args = {
+        plugin_args.update({
             "auth_url": auth_url,
             "project_name": project_name,
-            "username": username,
-            "password": password,
-        }
+        })
 
         keystone_version = connection_info.get(
             "identity_api_version", CONF.keystone.identity_api_version)
@@ -152,6 +157,5 @@ def create_keystone_session(ctxt, connection_info={}):
 
         loader = loading.get_plugin_loader(plugin_name)
         auth = loader.load_from_options(**plugin_args)
-        session = ks_session.Session(auth=auth, verify=verify)
 
-    return session
+    return ks_session.Session(auth=auth, verify=verify)

coriolis/migrations/api.py

@@ -8,9 +8,12 @@ class API(object):
     def __init__(self):
         self._rpc_client = rpc_client.ConductorClient()
 
-    def migrate_instances(self, ctxt, origin, destination, instances):
+    def migrate_instances(self, ctxt, origin_endpoint_id,
+                          destination_endpoint_id, destination_environment,
+                          instances):
         return self._rpc_client.migrate_instances(
-            ctxt, origin, destination, instances)
+            ctxt, origin_endpoint_id, destination_endpoint_id,
+            destination_environment, instances)
 
     def deploy_replica_instances(self, ctxt, replica_id, clone_disks=False,
                                  force=False):

coriolis/providers/base.py

@@ -40,6 +40,17 @@ class BaseProvider(object):
         raise exception.OSMorphingToolsNotFound()
 
 
+class BaseEndpointProvider(BaseProvider):
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    def get_instances(self, ctxt, connection_info, limit=None,
+                      last_seen_id=None, instance_name_pattern=None):
+        """ Returns a list of instances
+        """
+        pass
+
+
 class BaseImportProvider(BaseProvider):
     __metaclass__ = abc.ABCMeta
 

coriolis/providers/factory.py

@@ -22,6 +22,7 @@ PROVIDER_TYPE_MAP = {
     constants.PROVIDER_TYPE_REPLICA_EXPORT: base.BaseReplicaExportProvider,
     constants.PROVIDER_TYPE_IMPORT: base.BaseImportProvider,
     constants.PROVIDER_TYPE_REPLICA_IMPORT: base.BaseReplicaImportProvider,
+    constants.PROVIDER_TYPE_ENDPOINT: base.BaseEndpointProvider,
 }
 
 

coriolis/replicas/api.py

@@ -8,9 +8,11 @@ class API(object):
     def __init__(self):
         self._rpc_client = rpc_client.ConductorClient()
 
-    def create(self, ctxt, origin, destination, instances):
+    def create(self, ctxt, origin_endpoint_id, destination_endpoint_id,
+               destination_environment, instances):
         return self._rpc_client.create_instances_replica(
-            ctxt, origin, destination, instances)
+            ctxt, origin_endpoint_id, destination_endpoint_id,
+            destination_environment, instances)
 
     def delete(self, ctxt, replica_id):
         self._rpc_client.delete_replica(ctxt, replica_id)

coriolis/schemas.py

@@ -19,6 +19,10 @@ PROVIDER_CONNECTION_INFO_SCHEMA_NAME = "connection_info_schema.json"
 
 PROVIDER_TARGET_ENVIRONMENT_SCHEMA_NAME = "target_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_VM_IMPORT_INFO_SCHEMA_NAME = "vm_import_info_schema.json"
+
 
 def get_schema(package_name, schema_name,
                schemas_directory=DEFAULT_SCHEMAS_DIRECTORY):
@@ -54,11 +58,12 @@ def validate_string(json_string, schema):
     validate_value(json.loads(json_string), schema)
 
 
-# Global schemas:
-CORIOLIS_VM_EXPORT_INFO_SCHEMA_NAME = "vm_export_info_schema.json"
+# Global schemas
 CORIOLIS_VM_EXPORT_INFO_SCHEMA = get_schema(
-    __name__, CORIOLIS_VM_EXPORT_INFO_SCHEMA_NAME)
+    __name__, _CORIOLIS_VM_EXPORT_INFO_SCHEMA_NAME)
+
+CORIOLIS_VM_IMPORT_INFO_SCHEMA = get_schema(
+    __name__, _CORIOLIS_VM_IMPORT_INFO_SCHEMA_NAME)
 
-CORIOLIS_IMPORT_INFO_SCHEMA_NAME = "import_info_schema.json"
-CORIOLIS_IMPORT_INFO_SCHEMA = get_schema(
-    __name__, CORIOLIS_IMPORT_INFO_SCHEMA_NAME)
+CORIOLIS_VM_INSTANCE_INFO_SCHEMA = get_schema(
+    __name__, _CORIOLIS_VM_INSTANCE_INFO_SCHEMA_NAME)

coriolis/schemas/vm_instance_info_schema.json

@@ -0,0 +1,51 @@
+{
+  "$schema": "http://cloudbase.it/coriolis/schemas/vm_export_info#",
+  "type": "object",
+  "properties": {
+    "num_cpu": {
+      "type": "integer",
+      "description": "Number of CPUs of the VM."
+    },
+    "num_cores_per_socket": {
+      "type": "integer",
+      "description": "Number of CPU cores per socket, if applicable."
+    },
+    "memory_mb": {
+      "type": "integer",
+      "description": "Memory of the VM in MegaBytes."
+    },
+    "name": {
+      "type": "string",
+      "description": "Human-readable name of th VM."
+    },
+    "id": {
+      "type": "string",
+      "description": "Unique identifier of the VM."
+    },
+    "os_type": {
+      "type": "string",
+      "description": "The generic type of the operating system installed on the VM.",
+      "enum": ["bsd", "linux", "osx", "solaris", "windows"]
+    },
+    "firmware_type": {
+      "type": "string",
+      "description": "The type of firmware of the VM.",
+      "enum": ["BIOS", "EFI"]
+    },
+    "guest_id": {
+      "type": "string",
+      "description": "Extra ID field for added categorisation."
+    },
+    "flavor_name": {
+      "type": "string",
+      "description": "Name of the exported VM's flavor."
+    }
+  },
+  "required": [
+    "id",
+    "name",
+    "num_cpu",
+    "memory_mb",
+    "os_type"
+  ]
+}

coriolis/tasks/base.py

@@ -3,7 +3,6 @@
 
 import abc
 
-from coriolis import secrets
 from coriolis import utils
 
 from oslo_config import cfg
@@ -29,11 +28,7 @@ class TaskRunner(metaclass=abc.ABCMeta):
 
 def get_connection_info(ctxt, data):
     connection_info = data.get("connection_info") or {}
-    secret_ref = connection_info.get("secret_ref")
-    if secret_ref:
-        LOG.info("Retrieving connection info from secret: %s", secret_ref)
-        connection_info = secrets.get_secret(ctxt, secret_ref)
-    return connection_info
+    return utils.get_secret_connection_info(ctxt, connection_info)
 
 
 def marshal_migr_conn_info(migr_connection_info):

coriolis/tasks/migration_tasks.py

@@ -51,7 +51,7 @@ class ImportInstanceTask(base.TaskRunner):
             import_info["osmorphing_connection_info"])
 
         schemas.validate_value(
-            task_info, schemas.CORIOLIS_IMPORT_INFO_SCHEMA)
+            task_info, schemas.CORIOLIS_VM_IMPORT_INFO_SCHEMA)
 
         task_info["origin_provider_type"] = constants.PROVIDER_TYPE_EXPORT
         task_info["destination_provider_type"] = constants.PROVIDER_TYPE_IMPORT

coriolis/utils.py

@@ -21,6 +21,7 @@ import paramiko
 
 from coriolis import constants
 from coriolis import exception
+from coriolis import secrets
 
 opts = [
     cfg.StrOpt('qemu_img_path',
@@ -342,3 +343,18 @@ def check_md5(data, md5):
     new_md5 = m.hexdigest()
     if new_md5 != md5:
         raise exception.CoriolisException("MD5 check failed")
+
+
+def get_secret_connection_info(ctxt, connection_info):
+    secret_ref = connection_info.get("secret_ref")
+    if secret_ref:
+        LOG.info("Retrieving connection info from secret: %s", secret_ref)
+        connection_info = secrets.get_secret(ctxt, secret_ref)
+    return connection_info
+
+
+def parse_int_value(value):
+    try:
+        return int(str(value))
+    except ValueError:
+        raise exception.InvalidInput("Invalid integer: %s" % value)