Add Minion Pool Executions CRUDS.

coriolis/api/v1/minion_pool_actions.py

@@ -0,0 +1,46 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from webob import exc
+
+from coriolis import exception
+from coriolis.api.v1.views import minion_pool_tasks_execution_view
+from coriolis.api import wsgi as api_wsgi
+from coriolis.policies import minion_pools as minion_pool_policies
+from coriolis.minion_pools import api
+
+
+class MinionPoolActionsController(api_wsgi.Controller):
+    def __init__(self):
+        self.minion_pool_api = api.API()
+        super(MinionPoolActionsController, self).__init__()
+
+    @api_wsgi.action('allocate')
+    def _allocate_pool(self, req, id, body):
+        context = req.environ['coriolis.context']
+        context.can(
+            minion_pool_policies.get_minion_pools_policy_label("allocate"))
+        try:
+            return minion_pool_tasks_execution_view.single(
+                req, self.minion_pool_api.allocate(context, id))
+        except exception.NotFound as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+        except exception.InvalidParameterValue as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+
+    @api_wsgi.action('deallocate')
+    def _deallocate_pool(self, req, id, body):
+        context = req.environ['coriolis.context']
+        context.can(
+            minion_pool_policies.get_minion_pools_policy_label("deallocate"))
+        try:
+            return minion_pool_tasks_execution_view.single(
+                req, self.minion_pool_api.deallocate(context, id))
+        except exception.NotFound as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+        except exception.InvalidParameterValue as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+
+
+def create_resource():
+    return api_wsgi.Resource(MinionPoolActionsController())

coriolis/api/v1/minion_pool_tasks_execution_actions.py

@@ -0,0 +1,37 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from webob import exc
+
+from coriolis import exception
+from coriolis.api import wsgi as api_wsgi
+from coriolis.policies \
+    import minion_pool_tasks_executions as pool_execution_policies
+from coriolis.minion_pool_tasks_executions import api
+
+
+class MinionPoolTasksExecutionActionsController(api_wsgi.Controller):
+    def __init__(self):
+        self._minion_pool_tasks_executions_api = api.API()
+        super(MinionPoolTasksExecutionActionsController, self).__init__()
+
+    @api_wsgi.action('cancel')
+    def _cancel(self, req, minion_pool_id, id, body):
+        context = req.environ['coriolis.context']
+        context.can(
+            pool_execution_policies.get_minion_pool_executions_policy_label(
+                'cancel'))
+        try:
+            force = (body["cancel"] or {}).get("force", False)
+
+            self._minion_pool_tasks_executions_api.cancel(
+                context, minion_pool_id, id, force)
+            raise exc.HTTPNoContent()
+        except exception.NotFound as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+        except exception.InvalidParameterValue as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+
+
+def create_resource():
+    return api_wsgi.Resource(MinionPoolTasksExecutionActionsController())

coriolis/api/v1/minion_pool_tasks_executions.py

@@ -0,0 +1,65 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from webob import exc
+
+from coriolis.api import wsgi as api_wsgi
+from coriolis.api.v1.views import minion_pool_tasks_execution_view
+from coriolis import exception
+from coriolis.minion_pool_tasks_executions import api
+from coriolis.policies \
+    import minion_pool_tasks_executions as pool_execution_policies
+
+
+class MinionPoolTasksExecutionController(api_wsgi.Controller):
+    def __init__(self):
+        self._pool_tasks_execution_api = api.API()
+        super(MinionPoolTasksExecutionController, self).__init__()
+
+    def show(self, req, minion_pool_id, id):
+        context = req.environ["coriolis.context"]
+        context.can(
+            pool_execution_policies.get_minion_pool_executions_policy_label(
+                "show"))
+        execution = self._pool_tasks_execution_api.get(
+            context, minion_pool_id, id)
+        if not execution:
+            raise exc.HTTPNotFound()
+
+        return minion_pool_tasks_execution_view.single(req, execution)
+
+    def index(self, req, minion_pool_id):
+        context = req.environ["coriolis.context"]
+        context.can(
+            pool_execution_policies.get_minion_pool_executions_policy_label(
+                "list"))
+
+        return minion_pool_tasks_execution_view.collection(
+            req, self._pool_tasks_execution_api.list(
+                context, minion_pool_id, include_tasks=False))
+
+    def detail(self, req, minion_pool_id):
+        context = req.environ["coriolis.context"]
+        context.can(
+            pool_execution_policies.get_minion_pool_executions_policy_label(
+                "show"))
+        return minion_pool_tasks_execution_view.collection(
+            req, self._pool_tasks_execution_api.list(
+                req.environ['coriolis.context'], minion_pool_id,
+                include_tasks=True))
+
+    def delete(self, req, minion_pool_id, id):
+        context = req.environ["coriolis.context"]
+        context.can(
+            pool_execution_policies.get_minion_pool_executions_policy_label(
+                "delete"))
+
+        try:
+            self._pool_tasks_execution_api.delete(context, minion_pool_id, id)
+            raise exc.HTTPNoContent()
+        except exception.NotFound as ex:
+            raise exc.HTTPNotFound(explanation=ex.msg)
+
+
+def create_resource():
+    return api_wsgi.Resource(MinionPoolTasksExecutionController())

coriolis/api/v1/minion_pools.py

@@ -6,6 +6,7 @@ from webob import exc
 
 from coriolis import exception
 from coriolis.api.v1.views import minion_pool_view
+from coriolis.api.v1.views import minion_pool_tasks_execution_view
 from coriolis.api import wsgi as api_wsgi
 from coriolis.policies import minion_pools as pools_policies
 from coriolis.minion_pools import api
@@ -87,8 +88,9 @@ class MinionPoolController(api_wsgi.Controller):
         context = req.environ["coriolis.context"]
         context.can(pools_policies.get_minion_pools_policy_label("update"))
         updated_values = self._validate_update_body(body)
-        return minion_pool_view.single(req, self._minion_pool_api.update(
-            req.environ['coriolis.context'], id, updated_values))
+        return minion_pool_tasks_execution_view.single(
+            req, self._minion_pool_api.update(
+                req.environ['coriolis.context'], id, updated_values))
 
     def delete(self, req, id):
         context = req.environ["coriolis.context"]

coriolis/api/v1/replica_tasks_executions.py

@@ -36,10 +36,13 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
                 context, replica_id, include_tasks=False))
 
     def detail(self, req, replica_id):
+        context = req.environ["coriolis.context"]
+        context.can(
+            executions_policies.get_replica_executions_policy_label("show"))
+
         return replica_tasks_execution_view.collection(
             req, self._replica_tasks_execution_api.get_executions(
-                req.environ['coriolis.context'], replica_id,
-                include_tasks=True))
+                context, replica_id, include_tasks=True))
 
     def create(self, req, replica_id, body):
         context = req.environ["coriolis.context"]

coriolis/api/v1/router.py

@@ -15,6 +15,8 @@ from coriolis.api.v1 import endpoints
 from coriolis.api.v1 import migration_actions
 from coriolis.api.v1 import migrations
 from coriolis.api.v1 import minion_pools
+from coriolis.api.v1 import minion_pool_actions
+from coriolis.api.v1 import minion_pool_tasks_executions
 from coriolis.api.v1 import provider_schemas
 from coriolis.api.v1 import providers
 from coriolis.api.v1 import regions
@@ -67,6 +69,22 @@ class APIRouter(api.APIRouter):
                         controller=self.resources['minion_pools'],
                         collection={'detail': 'GET'})
 
+        minion_pool_actions_resource = minion_pool_actions.create_resource()
+        self.resources['minion_pool_actions'] = minion_pool_actions_resource
+        minion_pool_path = '/{project_id}/minion_pools/{id}'
+        mapper.connect('minion_pool_actions',
+                       minion_pool_path + '/actions',
+                       controller=self.resources['minion_pool_actions'],
+                       action='action',
+                       conditions={'method': 'POST'})
+
+        self.resources['minion_pool_tasks_executions'] = \
+            minion_pool_tasks_executions.create_resource()
+        mapper.resource('minion_pools', 'minion_pools/{minion_pool_id}/executions',
+                        controller=self.resources['minion_pool_tasks_executions'],
+                        collection={'detail': 'GET'},
+                        member={'action': 'POST'})
+
         endpoint_actions_resource = endpoint_actions.create_resource()
         self.resources['endpoint_actions'] = endpoint_actions_resource
         endpoint_path = '/{project_id}/endpoints/{id}'

coriolis/api/v1/views/minion_pool_tasks_execution_view.py

@@ -0,0 +1,45 @@
+# Copyright 2016 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import itertools
+
+from coriolis import constants
+from coriolis import utils
+
+
+def _sort_tasks(tasks, filter_error_only_tasks=True):
+    """ Sorts the given list of dicts representing tasks.
+    Tasks are sorted primarily based on their index.
+    """
+    if filter_error_only_tasks:
+        tasks = [
+            t for t in tasks
+            if t['status'] != (
+                constants.TASK_STATUS_ON_ERROR_ONLY)]
+    return sorted(
+        tasks, key=lambda t: t.get('index', 0))
+
+
+def format_minion_pool_tasks_execution(req, execution, keys=None):
+    def transform(key, value):
+        if keys and key not in keys:
+            return
+        yield (key, value)
+
+    if "tasks" in execution:
+        execution["tasks"] = _sort_tasks(execution["tasks"])
+
+    execution_dict = dict(itertools.chain.from_iterable(
+        transform(k, v) for k, v in execution.items()))
+
+    return execution_dict
+
+
+def single(req, execution):
+    return {"execution": format_minion_pool_tasks_execution(req, execution)}
+
+
+def collection(req, executions):
+    formatted_executions = [
+        format_minion_pool_tasks_execution(req, m) for m in executions]
+    return {'executions': formatted_executions}

coriolis/conductor/rpc/client.py

@@ -384,6 +384,14 @@ class ConductorClient(object):
             minion_max_idle_time=minion_max_idle_time,
             minion_retention_strategy=minion_retention_strategy)
 
+    def allocate_minion_pool(self, ctxt, minion_pool_id):
+        return self._client.call(
+            ctxt, "allocate_minion_pool", minion_pool_id=minion_pool_id)
+
+    def deallocate_minion_pool(self, ctxt, minion_pool_id):
+        return self._client.call(
+            ctxt, "deallocate_minion_pool", minion_pool_id=minion_pool_id)
+
     def get_minion_pools(self, ctxt):
         return self._client.call(ctxt, 'get_minion_pools')
 
@@ -399,3 +407,28 @@ class ConductorClient(object):
     def delete_minion_pool(self, ctxt, minion_pool_id):
         return self._client.call(
             ctxt, 'delete_minion_pool', minion_pool_id=minion_pool_id)
+
+    def get_minion_pool_lifecycle_executions(
+            self, ctxt, minion_pool_id, include_tasks=False):
+        return self._client.call(
+            ctxt, 'get_minion_pool_lifecycle_executions',
+            minion_pool_id=minion_pool_id, include_tasks=include_tasks)
+
+    def get_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id):
+        return self._client.call(
+            ctxt, 'get_minion_pool_lifecycle_execution',
+            minion_pool_id=minion_pool_id, execution_id=execution_id)
+
+    def delete_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id):
+        return self._client.call(
+            ctxt, 'delete_minion_pool_lifecycle_execution',
+            minion_pool_id=minion_pool_id, execution_id=execution_id)
+
+    def cancel_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id, force):
+        return self._client.call(
+            ctxt, 'cancel_minion_pool_lifecycle_execution',
+            minion_pool_id=minion_pool_id, execution_id=execution_id,
+            force=force)

coriolis/conductor/rpc/server.py

@@ -2916,81 +2916,224 @@ class ConductorServerEndpoint(object):
             minion_retention_strategy):
         endpoint = db_api.get_endpoint(ctxt, endpoint_id)
 
-        minion_pool = models.MinionPool()
+        minion_pool = models.MinionPoolLifecycle()
         minion_pool.id = str(uuid.uuid4())
         minion_pool.name = name
-        minion_pool.endpoint_id = endpoint_id
-        minion_pool.environment_options = environment_options
+        minion_pool.pool_status = constants.MINION_POOL_STATUS_UNINITIALIZED
         minion_pool.minimum_minions = minimum_minions
         minion_pool.maximum_minions = maximum_minions
         minion_pool.minion_max_idle_time = minion_max_idle_time
         minion_pool.minion_retention_strategy = minion_retention_strategy
 
-        db_api.add_minion_pool(ctxt, minion_pool)
+        # TODO(aznashwan): These field redundancies should be
+        # eliminated once the DB model hirearchy is overhauled:
+        minion_pool.origin_endpoint_id = endpoint_id
+        minion_pool.destination_endpoint_id = endpoint_id
+        minion_pool.source_environment = environment_options
+        minion_pool.destination_environment = minimum_minions
+        minion_pool.instances = []
+        minion_pool.info = {}
+
+        db_api.add_minion_pool_lifecycle(ctxt, minion_pool)
         return self.get_minion_pool(ctxt, minion_pool.id)
 
     def get_minion_pools(self, ctxt):
-        return db_api.get_minion_pools(ctxt)
+        return db_api.get_minion_pool_lifecycles(
+            ctxt, include_tasks_executions=False)
 
-    @minion_pool_synchronized
-    def get_minion_pool(self, ctxt, minion_pool_id):
-        minion_pool = db_api.get_minion_pool(ctxt, minion_pool_id)
+    def _get_minion_pool(self, ctxt, minion_pool_id):
+        minion_pool = db_api.get_minion_pool_lifecycle(ctxt, minion_pool_id)
         if not minion_pool:
             raise exception.NotFound(
                 "minion_pool with ID '%s' not found." % minion_pool_id)
         return minion_pool
 
+    @minion_pool_synchronized
+    def allocate_minion_pool(self, ctxt, minion_pool_id):
+        LOG.info("Attempting to allocate Minion Pool '%s'.", minion_pool_id)
+        minion_pool = db_api.get_minion_pool_lifecycle(ctxt, minion_pool_id)
+        if minion_pool.status not in (
+                constants.MINION_POOL_STATUS_UNINITIALIZED,
+                constants.MINION_POOL_STATUS_DEALLOCATED):
+            raise exception.InvalidMinionPoolState(
+                "Minion Pool '%s' cannot be started as it is in '%s' state" % (
+                    minion_pool_id, minion_pool.status))
+
+        execution = models.TasksExecution()
+        execution.id = str(uuid.uuid4())
+        execution.action = minion_pool
+        execution.status = constants.EXECUTION_STATUS_UNEXECUTED
+        execution.type = constants.EXECUTION_TYPE_MINION_POOL_MAINTENANCE
+
+        minion_machine_ids = []
+        try:
+            for i in range(minion_pool.minimum_minions):
+                minion_machine = models.MinionMachine()
+                minion_machine.id = str(uuid.uuid4())
+                minion_machine.pool_id = minion_pool.id
+                minion_machine.status = (
+                    constants.MINION_MACHINE_STATUS_UNINITIALIZED)
+                minion_machine.connection_info = {}
+                minion_machine.provider_properties = {}
+                db_api.add_minion_machine(ctxt, minion_machine)
+
+                minion_machine_ids.append(minion_machine.id)
+        except Exception as ex:
+            LOG.warn(
+                "Failed to create minion machine DB entry for pool with "
+                "ID '%s'. Cleaning up already created machines: %s" % (
+                    minion_pool_id, minion_machine_ids))
+            for minion_machine_id in minion_machine_ids:
+                utils.ignore_exceptions(
+                    db_api.delete_minion_machine(ctxt, minion_machine_id))
+            raise
+
+        # TODO(aznashwan): add shared pool resources setup tasks:
+        for minion_machine_id in minion_machine_ids:
+            minion_pool.info[minion_machine_id] = {
+                "pool_environment_options": minion_pool.source_environment}
+
+            validate_minions_option_task = self._create_task(
+                minion_machine_id,
+                constants.TASK_TYPE_VALIDATE_MINION_POOL_OPTIONS,
+                execution)
+
+            create_minion_task = self._create_task(
+                minion_machine_id,
+                constants.TASK_TYPE_CREATE_MINION,
+                execution, depends_on=[validate_minions_option_task.id])
+
+            self._create_task(
+                minion_machine_id,
+                constants.TASK_TYPE_DELETE_MINION,
+                execution, on_error_only=True,
+                depends_on=[create_minion_task.id])
+
+        self._check_execution_tasks_sanity(execution, minion_pool.info)
+
+        # update the action info for all of the pool's minions:
+        for minion_machine_id in minion_machine_ids:
+            db_api.update_transfer_action_info_for_instance(
+                ctxt, minion_pool.id, minion_machine_id,
+                minion_pool.info[minion_machine_id])
+
+        # add new execution to DB:
+        db_api.add_minion_pool_lifecycle_execution(ctxt, execution)
+        LOG.info("Minion pool allocation execution created: %s", execution.id)
+
+        self._begin_tasks(ctxt, execution, task_info=minion_pool.info)
+        return self.get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution.id)
+
+    @minion_pool_synchronized
+    def deallocate_minion_pool(self, ctxt, minion_pool_id):
+        LOG.info("Attempting to deallocate Minion Pool '%s'.", minion_pool_id)
+        minion_pool = db_api.get_minion_pool_lifecycle(ctxt, minion_pool_id)
+        if minion_pool.status not in (
+                constants.MINION_POOL_STATUS_AVAILABLE):
+            raise exception.InvalidMinionPoolState(
+                "Minion Pool '%s' cannot be started as it is in '%s' state" % (
+                    minion_pool_id, minion_pool.status))
+
+        # TODO(aznashwan): check minion pool running
+        # executions/allocated machines
+
+        execution = models.TasksExecution()
+        execution.id = str(uuid.uuid4())
+        execution.action = minion_pool
+        execution.status = constants.EXECUTION_STATUS_UNEXECUTED
+        execution.type = constants.EXECUTION_TYPE_MINION_POOL_MAINTENANCE
+
+        # TODO(aznashwan): add shared pool resources setup tasks:
+        for minion_machine_id in minion_pool.instances:
+            minion_pool.info[minion_machine_id] = {
+                "pool_environment_options": minion_pool.source_environment}
+            self._create_task(
+                minion_machine_id, constants.TASK_TYPE_DELETE_MINION,
+                execution)
+
+        self._check_execution_tasks_sanity(execution, minion_pool.info)
+
+        # update the action info for all of the pool's minions:
+        for minion_machine_id in minion_pool.instances:
+            db_api.update_transfer_action_info_for_instance(
+                ctxt, minion_pool.id, minion_machine_id,
+                minion_pool.info[minion_machine_id])
+
+        # add new execution to DB:
+        db_api.add_minion_pool_lifecycle_execution(ctxt, execution)
+        LOG.info("Minion pool allocation execution created: %s", execution.id)
+
+        self._begin_tasks(ctxt, execution, task_info=minion_pool.info)
+        return self.get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution.id)
+
+    @minion_pool_synchronized
+    def get_minion_pool(self, ctxt, minion_pool_id):
+        return self._get_minion_pool(ctxt, minion_pool_id)
+
     @minion_pool_synchronized
     def update_minion_pool(self, ctxt, minion_pool_id, updated_values):
         LOG.info(
             "Attempting to update minion_pool '%s' with payload: %s",
             minion_pool_id, updated_values)
-        db_api.update_minion_pool(ctxt, minion_pool_id, updated_values)
+        db_api.update_minion_pool_lifecycle(ctxt, minion_pool_id, updated_values)
         LOG.info("Minion Pool '%s' successfully updated", minion_pool_id)
-        return db_api.get_minion_pool(ctxt, minion_pool_id)
+        return db_api.get_minion_pool_lifecycle(ctxt, minion_pool_id)
 
     @minion_pool_synchronized
     def delete_minion_pool(self, ctxt, minion_pool_id):
         # TODO(aznashwan): add checks for endpoints/services
         # associated to the minion_pool before deletion:
-        db_api.delete_minion_pool(ctxt, minion_pool_id)
+        db_api.delete_minion_pool_lifecycle(ctxt, minion_pool_id)
 
-    @replica_synchronized
+    @minion_pool_synchronized
     def get_minion_pool_lifecycle_executions(
             self, ctxt, minion_pool_id, include_tasks=False):
-        return db_api.get_replica_tasks_executions(
-            ctxt, replica_id, include_tasks)
+        return db_api.get_minion_pool_lifecycle_executions(
+            ctxt, minion_pool_id, include_tasks)
+
+    def _get_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id):
+        execution = db_api.get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id)
+        if not execution:
+            raise exception.NotFound(
+                "Execution with ID '%s' for Minion Pool '%s' not found." % (
+                    execution_id, minion_pool_id))
+        return execution
 
     @tasks_execution_synchronized
-    def get_replica_tasks_execution(self, ctxt, replica_id, execution_id):
-        return self._get_replica_tasks_execution(
-            ctxt, replica_id, execution_id)
+    def get_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id):
+        return self._get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id)
 
     @tasks_execution_synchronized
-    def delete_replica_tasks_execution(self, ctxt, replica_id, execution_id):
-        execution = self._get_replica_tasks_execution(
-            ctxt, replica_id, execution_id)
+    def delete_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id):
+        execution = self._get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id)
         if execution.status in constants.ACTIVE_EXECUTION_STATUSES:
             raise exception.InvalidMigrationState(
-                "Cannot delete execution '%s' for Replica '%s' as it is "
+                "Cannot delete execution '%s' for Minion pool '%s' as it is "
                 "currently in '%s' state." % (
-                    execution_id, replica_id, execution.status))
-        db_api.delete_replica_tasks_execution(ctxt, execution_id)
+                    execution_id, minion_pool_id, execution.status))
+        db_api.delete_minion_pool_lifecycle_execution(ctxt, execution_id)
 
     @tasks_execution_synchronized
-    def cancel_replica_tasks_execution(self, ctxt, replica_id, execution_id,
-                                       force):
-        execution = self._get_replica_tasks_execution(
-            ctxt, replica_id, execution_id)
+    def cancel_minion_pool_lifecycle_execution(
+            self, ctxt, minion_pool_id, execution_id, force):
+        execution = self._get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id)
         if execution.status not in constants.ACTIVE_EXECUTION_STATUSES:
-            raise exception.InvalidReplicaState(
-                "Replica '%s' has no running execution to cancel." % (
-                    replica_id))
+            raise exception.InvalidMinionPoolState(
+                "Minion pool '%s' has no running execution to cancel." % (
+                    minion_pool_id))
         if execution.status == constants.EXECUTION_STATUS_CANCELLING and (
                 not force):
-            raise exception.InvalidReplicaState(
-                "Replica '%s' is already being cancelled. Please use the "
-                "force option if you'd like to force-cancel it." % (
-                    replica_id))
+            raise exception.InvalidMinionPoolState(
+                "Execution for Minion Pool '%s' is already being cancelled. "
+                "Please use the force option if you'd like to force-cancel "
+                "it." % (minion_pool_id))
         self._cancel_tasks_execution(ctxt, execution, force=force)
-

coriolis/constants.py

@@ -130,6 +130,19 @@ TASK_TYPE_VALIDATE_REPLICA_DEPLOYMENT_INPUTS = (
 TASK_TYPE_UPDATE_SOURCE_REPLICA = "UPDATE_SOURCE_REPLICA"
 TASK_TYPE_UPDATE_DESTINATION_REPLICA = "UPDATE_DESTINATION_REPLICA"
 
+TASK_TYPE_VALIDATE_MINION_POOL_OPTIONS = (
+    "VALIDATE_MINION_POOL_ENVIRONMENT_OPTIONS")
+TASK_TYPE_CREATE_MINION = "CREATE_MINION"
+TASK_TYPE_DELETE_MINION = "DELETE_MINION"
+TASK_TYPE_STOP_MINION = "STOP_MINION"
+TASK_TYPE_START_MINION = "START_MINION"
+TASK_TYPE_ATTACH_DISK_TO_MINION = "ATTACH_DISK_TO_MINION"
+TASK_TYPE_DETACH_DISK_FROM_MINION = "DETACH_DISK_FROM_MINION"
+TASK_TYPE_SET_UP_SHARED_POOL_RESOURCES = "SET_UP_POOL_RESOURCES"
+TASK_TYPE_TEAR_DOWN_SHARED_POOL_RESOURCES = (
+    "TEAR_DOWN_SHARED_POOL_RESOURCES")
+
+
 TASK_PLATFORM_SOURCE = "source"
 TASK_PLATFORM_DESTINATION = "destination"
 TASK_PLATFORM_BILATERAL = "bilateral"
@@ -156,6 +169,7 @@ PROVIDER_TYPE_ENDPOINT_STORAGE = 32768
 PROVIDER_TYPE_SOURCE_REPLICA_UPDATE = 65536
 PROVIDER_TYPE_SOURCE_ENDPOINT_OPTIONS = 131072
 PROVIDER_TYPE_DESTINATION_REPLICA_UPDATE = 262144
+PROVIDER_TYPE_MINION_POOL = 524288
 
 DISK_FORMAT_VMDK = 'vmdk'
 DISK_FORMAT_RAW = 'raw'
@@ -205,6 +219,7 @@ EXECUTION_TYPE_REPLICA_DISKS_DELETE = "replica_disks_delete"
 EXECUTION_TYPE_REPLICA_DEPLOY = "replica_deploy"
 EXECUTION_TYPE_MIGRATION = "migration"
 EXECUTION_TYPE_REPLICA_UPDATE = "replica_update"
+EXECUTION_TYPE_MINION_POOL_MAINTENANCE = "minion_pool_maintenance"
 
 TASK_LOCK_NAME_FORMAT = "task-%s"
 EXECUTION_LOCK_NAME_FORMAT = "execution-%s"
@@ -221,7 +236,8 @@ EXECUTION_TYPE_TO_ACTION_LOCK_NAME_FORMAT_MAP = {
     EXECUTION_TYPE_REPLICA_EXECUTION: REPLICA_LOCK_NAME_FORMAT,
     EXECUTION_TYPE_REPLICA_DEPLOY: REPLICA_LOCK_NAME_FORMAT,
     EXECUTION_TYPE_REPLICA_UPDATE: REPLICA_LOCK_NAME_FORMAT,
-    EXECUTION_TYPE_REPLICA_DISKS_DELETE: REPLICA_LOCK_NAME_FORMAT
+    EXECUTION_TYPE_REPLICA_DISKS_DELETE: REPLICA_LOCK_NAME_FORMAT,
+    EXECUTION_TYPE_MINION_POOL_MAINTENANCE: MINION_POOL_LOCK_NAME_FORMAT
 }
 
 SERVICE_STATUS_UP = "UP"
@@ -233,3 +249,16 @@ CONDUCTOR_MAIN_MESSAGING_TOPIC = "coriolis_conductor"
 WORKER_MAIN_MESSAGING_TOPIC = "coriolis_worker"
 SCHEDULER_MAIN_MESSAGING_TOPIC = "coriolis_scheduler"
 REPLICA_CRON_MAIN_MESSAGING_TOPIC = "coriolis_replica_cron_worker"
+
+MINION_POOL_STATUS_UNKNOWN = "UNKNOWN"
+MINION_POOL_STATUS_UNINITIALIZED = "UNINITIALIZED"
+MINION_POOL_STATUS_AVAILABLE = "AVAILABLE"
+MINION_POOL_STATUS_DEALLOCATED = "DEALLOCATED"
+MINION_POOL_STATUS_RECONFIGURING = "RECONFIGURING"
+
+MINION_MACHINE_IDENTIFIER_FORMAT = "coriolis-pool-%(pool_id)s-minion-%(minion_id)s"
+MINION_MACHINE_STATUS_UNKNOWN = "UNKNOWN"
+MINION_MACHINE_STATUS_UNINITIALIZED = "UNINITIALIZED"
+MINION_MACHINE_STATUS_RECONFIGURING = "RECONFIGURING"
+MINION_MACHINE_STATUS_AVAILABLE = "AVAILABLE"
+MINION_MACHINE_STATUS_ALLOCATED = "ALLOCATED"

coriolis/db/api.py

@@ -1089,56 +1089,10 @@ def get_mapped_services_for_region(context, region_id):
     return q.all()
 
 
-@enginefacade.writer
-def add_minion_pool(context, minion_pool):
-    minion_pool.user_id = context.user
-    minion_pool.project_id = context.tenant
-    _session(context).add(minion_pool)
-
-
-@enginefacade.reader
-def get_minion_pools(context):
-    q = _soft_delete_aware_query(context, models.MinionPool)
-    q = q.options(orm.joinedload('endpoint'))
-    return q.all()
-
-
-@enginefacade.reader
-def get_minion_pool(context, minion_pool_id):
-    q = _soft_delete_aware_query(context, models.MinionPool)
-    q = q.options(orm.joinedload('endpoint'))
-    return q.filter(
-        models.MinionPool.id == minion_pool_id).first()
-
-
-@enginefacade.writer
-def update_minion_pool(context, minion_pool_id, updated_values):
-    if not minion_pool_id:
-        raise exception.InvalidInput(
-            "No minion_pool ID specified for updating.")
-    minion_pool = get_minion_pool(context, minion_pool_id)
-    if not minion_pool:
-        raise exception.NotFound(
-            "MinionPool with ID '%s' does not exist." % minion_pool_id)
-
-    updateable_fields = [
-        "name", "environment_options", "minimum_minions", "maximum_minions",
-        "minion_max_idle_time", "minion_retention_strategy"]
-    _update_sqlalchemy_object_fields(
-        minion_pool, updateable_fields, updated_values)
-
-
-@enginefacade.writer
-def delete_minion_pool(context, minion_pool_id):
-    minion_pool = get_minion_pool(context, minion_pool_id)
-    count = _soft_delete_aware_query(context, models.MinionPool).filter_by(
-        id=minion_pool_id).soft_delete()
-    if count == 0:
-        raise exception.NotFound("0 minion_pool entries were soft deleted")
-
-
 @enginefacade.writer
 def add_minion_machine(context, minion_machine):
+    minion_machine.user_id = context.user
+    minion_machine.project_id = context.tenant
     _session(context).add(minion_machine)
 
 
@@ -1183,10 +1137,16 @@ def delete_minion_machine(context, minion_machine_id):
 
 
 @enginefacade.writer
-def add_minion_pool_lifecycle(context, lifecycle):
-    lifecycle.user_id = context.user
-    lifecycle.project_id = context.tenant
-    _session(context).add(lifecycle)
+def add_minion_pool_lifecycle(context, minion_pool_lifecycle):
+    minion_pool_lifecycle.user_id = context.user
+    minion_pool_lifecycle.project_id = context.tenant
+    _session(context).add(minion_pool_lifecycle)
+
+
+@enginefacade.writer
+def delete_minion_pool_lifecycle(context, minion_pool_id):
+    _delete_transfer_action(
+        context, models.MinionPoolLifecycle, minion_pool_id)
 
 
 @enginefacade.reader
@@ -1200,6 +1160,25 @@ def get_minion_pool_lifecycle(context, minion_pool_id):
         models.MinionPoolLifecycle.id == minion_pool_id).first()
 
 
+@enginefacade.reader
+def get_minion_pool_lifecycles(
+        context, include_tasks_executions=False, include_info=False,
+        to_dict=True):
+    q = _soft_delete_aware_query(context, models.MinionPoolLifecycle)
+    if include_tasks_executions:
+        q = q.options(orm.joinedload(models.MinionPoolLifecycle.executions))
+    if include_info is False:
+        q = q.options(orm.defer('info'))
+    q = q.filter()
+    if is_user_context(context):
+        q = q.filter(
+            models.Replica.project_id == context.tenant)
+    db_result = q.all()
+    if to_dict:
+        return [i.to_dict(include_info=include_info) for i in db_result]
+    return db_result
+
+
 @enginefacade.writer
 def add_minion_pool_lifecycle_execution(context, execution):
     if is_user_context(context):
@@ -1223,15 +1202,26 @@ def update_minion_pool_lifecycle(context, minion_pool_id, updated_values):
             "Minion pool '%s' not found" % minion_pool_id)
 
     updateable_fields = [
-        "source_environment", "destination_environment",
         "minimum_minions", "maximum_minions", "minion_max_idle_time",
-        "minion_retention_strategy"]
+        "minion_retention_strategy", "environment_options"]
+    # TODO(aznashwan): this should no longer be required when the
+    # transfer action class hirearchy is to be overhauled:
+    redundancies = {
+        "environment_options": [
+            "source_environment", "destination_environment"]}
     for field in updateable_fields:
         if field in updated_values:
-            LOG.debug(
-                "Updating the '%s' field of Minion Pool '%s' to: '%s'",
-                field, minion_pool_id, updated_values[field])
-            setattr(lifecycle, field, updated_values[field])
+            if field in redundancies:
+                for old_key in redundancies["field"]:
+                    LOG.debug(
+                        "Updating the '%s' field of Minion Pool '%s' to: '%s'",
+                        old_key, minion_pool_id, updated_values[field])
+                    setattr(lifecycle, old_key, updated_values[field])
+            else:
+                LOG.debug(
+                    "Updating the '%s' field of Minion Pool '%s' to: '%s'",
+                    field, minion_pool_id, updated_values[field])
+                setattr(lifecycle, field, updated_values[field])
 
     non_updateable_fields = set(
         updated_values.keys()).difference(updateable_fields)

coriolis/db/sqlalchemy/migrate_repo/versions/015_adds_minion_vm_pools.py → coriolis/db/sqlalchemy/migrate_repo/versions/016_adds_minion_vm_pools.py

@@ -26,6 +26,9 @@ def upgrade(migrate_engine):
                 "id", sqlalchemy.String(36),
                 sqlalchemy.ForeignKey('base_transfer_action.base_id'),
                 primary_key=True),
+            sqlalchemy.Column(
+                "pool_status", sqlalchemy.String(255), nullable=False,
+                default=lambda: "UNKNOWN"),
             sqlalchemy.Column(
                 'minimum_minions', sqlalchemy.Integer, nullable=False),
             sqlalchemy.Column(
@@ -51,7 +54,8 @@ def upgrade(migrate_engine):
                 'pool_id', sqlalchemy.String(36),
                 sqlalchemy.ForeignKey('minion_pool.id'), nullable=False),
             sqlalchemy.Column(
-                'status', sqlalchemy.String(255), nullable=False),
+                'status', sqlalchemy.String(255), nullable=False,
+                default=lambda: "UNKNOWN"),
             sqlalchemy.Column('connection_info', sqlalchemy.Text),
             sqlalchemy.Column('provider_properties', sqlalchemy.Text)))
 

coriolis/db/sqlalchemy/models.py

@@ -200,10 +200,14 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
         sqlalchemy.String(36),
         sqlalchemy.ForeignKey('minion_pool_lifecycle.id'),
         nullable=True, default=lambda: None)
+    source_minion_pool = orm.relationship(
+        "minion_pool_lifecycle", foreign_keys=[source_minion_pool_id])
     destination_minion_pool_id = sqlalchemy.Column(
         sqlalchemy.String(36),
         sqlalchemy.ForeignKey('minion_pool_lifecycle.id'),
         nullable=True, default=lambda: None)
+    destination_minion_pool = orm.relationship(
+        "minion_pool_lifecycle", foreign_keys=[destination_minion_pool_id])
 
     __mapper_args__ = {
         'polymorphic_identity': 'base_transfer_action',
@@ -250,12 +254,16 @@ class MinionPoolLifecycle(BaseTransferAction):
     id = sqlalchemy.Column(
         sqlalchemy.String(36),
         sqlalchemy.ForeignKey(
-            'base_transfer_action.base_id'), primary_key=True)
+            'base_transfer_action.base_id'),
+        primary_key=True)
 
     name = sqlalchemy.Column(
         sqlalchemy.String(255),
         nullable=False)
 
+    pool_status = sqlalchemy.Column(
+        sqlalchemy.String(255), nullable=False,
+        default=lambda: constants.MINION_POOL_STATUS_UNKNOWN)
     minimum_minions = sqlalchemy.Column(
         sqlalchemy.Integer, nullable=False)
     maximum_minions = sqlalchemy.Column(
@@ -272,6 +280,17 @@ class MinionPoolLifecycle(BaseTransferAction):
         base = super(MinionPoolLifecycle, self).to_dict(
             include_info=include_info)
         base.update({"id": self.id})
+        # TODO(aznashwan): these nits should be avoided by splitting the
+        # BaseTransferAction class into a more specialized hireachy:
+        redundancies = {
+            "environment_options": [
+                "source_environment", "destination_environment"],
+            "endpoint_id": [
+                "origin_endpoint_id", "destination_endpoint_id"]}
+        for new_key, old_keys in redundancies.items():
+            for old_key in old_keys:
+                if old_key in base:
+                    base[new_key] = base.pop(old_key)
         return base
 
 
@@ -499,7 +518,8 @@ class MinionMachine(BASE, models.TimestampMixin, models.ModelBase,
         foreign_keys=[pool_id])
 
     status = sqlalchemy.Column(
-        sqlalchemy.String(255), nullable=False)
+        sqlalchemy.String(255), nullable=False,
+        default=lambda: constants.MINION_MACHINE_STATUS_UNKNOWN)
 
     connection_info = sqlalchemy.Column(types.Json)
 

coriolis/exception.py

@@ -202,6 +202,11 @@ class InvalidTaskState(Invalid):
         'Task "%(task_id)s" in in an invalid state: %(task_state)s')
 
 
+class InvalidMinionPoolState(Invalid):
+    message = _(
+        'Minion pool "%(pool_id)s" in in an invalid state: %(pool_state)s')
+
+
 class TaskIsCancelling(InvalidTaskState):
     message = _(TASK_ALREADY_CANCELLING_EXCEPTION_FMT)
 

coriolis/minion_pool_tasks_executions/api.py

@@ -0,0 +1,26 @@
+# Copyright 2020 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis import utils
+from coriolis.conductor.rpc import client as rpc_client
+
+
+class API(object):
+    def __init__(self):
+        self._rpc_client = rpc_client.ConductorClient()
+
+    def list(self, ctxt, minion_pool_id, include_tasks=False):
+        return self._rpc_client.get_minion_pool_lifecycle_executions(
+            ctxt, minion_pool_id, include_tasks=include_tasks)
+
+    def get(self, ctxt, minion_pool_id, execution_id):
+        return self._rpc_client.get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id)
+
+    def cancel(self, ctxt, minion_pool_id, execution_id, force):
+        return self._rpc_client.cancel_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id, force)
+
+    def delete(self, ctxt, minion_pool_id, execution_id):
+        return self._rpc_client.delete_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution_id)

coriolis/minion_pools/api.py

@@ -29,3 +29,9 @@ class API(object):
 
     def get_minion_pool(self, ctxt, minion_pool_id):
         return self._rpc_client.get_minion_pool(ctxt, minion_pool_id)
+
+    def allocate(self, ctxt, minion_pool_id):
+        return self._rpc_client.allocate_minion_pool(ctxt, minion_pool_id)
+
+    def deallocate(self, ctxt, minion_pool_id):
+        return self._rpc_client.deallocate_minion_pool(ctxt, minion_pool_id)

coriolis/policies/minion_pool_tasks_executions.py

@@ -0,0 +1,83 @@
+# Copyright 2018 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from oslo_policy import policy
+
+from coriolis.policies import base
+
+
+MINION_POOL_EXECUTIONS_POLICY_PREFIX = "%s:minion_pool_executions" % (
+    base.CORIOLIS_POLICIES_PREFIX)
+MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULE = "rule:admin_or_owner"
+
+
+def get_minion_pool_executions_policy_label(rule_label):
+    return "%s:%s" % (
+        MINION_POOL_EXECUTIONS_POLICY_PREFIX, rule_label)
+
+
+MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULES = [
+    policy.DocumentedRuleDefault(
+        get_minion_pool_executions_policy_label('create'),
+        MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULE,
+        "Create a new execution for a given Minion Pool",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/executions",
+                "method": "POST"
+            }
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        get_minion_pool_executions_policy_label('list'),
+        MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULE,
+        "List Executions for a given Minion Pool",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/executions",
+                "method": "GET"
+            }
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        get_minion_pool_executions_policy_label('show'),
+        MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULE,
+        "Show details for Minion Pool execution",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/executions/{execution_id}",
+                "method": "GET"
+            }
+        ]
+    ),
+    # TODO(aznashwan): minion pool execution actions should ideally be
+    # declared in a separate module
+    policy.DocumentedRuleDefault(
+        get_minion_pool_executions_policy_label('cancel'),
+        MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULE,
+        "Cancel a Minion Pool execution",
+        [
+            {
+                "path": (
+                    "/minion_pools/{minion_pool_id}/executions/"
+                    "{execution_id}/actions"),
+                "method": "POST"
+            }
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        get_minion_pool_executions_policy_label('delete'),
+        MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULE,
+        "Delete an execution for a given Minion Pool",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/executions/{execution_id}",
+                "method": "DELETE"
+            }
+        ]
+    )
+]
+
+
+def list_rules():
+    return MINION_POOL_EXECUTIONS_POLICY_DEFAULT_RULES

coriolis/policies/minion_pools.py

@@ -71,6 +71,28 @@ MINION_POOLS_DEFAULT_RULES = [
                 "method": "DELETE"
             }
         ]
+    ),
+    policy.DocumentedRuleDefault(
+        get_minion_pools_policy_label('allocate'),
+        MINION_POOLS_DEFAULT_RULE,
+        "Allocate Minion Pool",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/actions",
+                "method": "POST"
+            }
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        get_minion_pools_policy_label('deallocate'),
+        MINION_POOLS_DEFAULT_RULE,
+        "Deallocate Minion Pool",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/actions",
+                "method": "POST"
+            }
+        ]
     )
 ]
 

coriolis/policy.py

@@ -15,6 +15,7 @@ from coriolis.policies import endpoints
 from coriolis.policies import general
 from coriolis.policies import migrations
 from coriolis.policies import minion_pools
+from coriolis.policies import minion_pool_tasks_executions
 from coriolis.policies import regions
 from coriolis.policies import replicas
 from coriolis.policies import replica_schedules
@@ -29,7 +30,8 @@ _ENFORCER = None
 
 DEFAULT_POLICIES_MODULES = [
     base, endpoints, general, migrations, replicas, replica_schedules,
-    replica_tasks_executions, diagnostics, regions, services, minion_pools]
+    replica_tasks_executions, diagnostics, regions, services, minion_pools,
+    minion_pool_tasks_executions]
 
 
 def reset():

coriolis/providers/base.py

@@ -532,3 +532,87 @@ class BaseUpdateDestinationReplicaProvider(
         having been executed or the replica disks having been deleted, this
         method should simply return the empty `volumes_info` it was given.
         """
+
+
+class BaseMinionPoolProvider(
+        object, with_metaclass(abc.ABCMeta)):
+    """ Class for providers which offer Minion Pool management functionality.
+    """
+
+    @abc.abstractmethod
+    def get_minion_pool_environment_schema(self):
+        """ Returns the schema for the minion pool options. """
+        pass
+
+    @abc.abstractmethod
+    def get_minion_pool_options(self, ctxt, environment_options):
+        """ Returns possible environment options for minion pools. """
+        pass
+
+    @abc.abstractmethod
+    def validate_minion_compatibility_for_transfer(
+            self, ctxt, environment_options,
+            transfer_options, storage_mappings):
+        """ Validates compatibility between the pool's options and the options
+        selected for a given transfer. Should raise if any options related to
+        the minions in the pool might be deemed incompatible with the desited
+        transfer options.
+        """
+        pass
+
+    @abc.abstractmethod
+    def validate_pool_options(
+            self, ctxt, connection_info, environment_options):
+        """ Validates the provided pool options. """
+        pass
+
+    @abc.abstractmethod
+    def setup_pool_supporting_resources(
+            self, ctxt, connection_info, environment_options):
+        """ Sets up supporting resources which can be re-used amongst the
+        machines which will be spawned within the pool (e.g. a shared network)
+        """
+        pass
+
+    @abc.abstractmethod
+    def teardown_pool_supporting_resources(
+            self, ctxt, connection_info, environment_options,
+            pool_resource_info):
+        """ Tears down all pool supporting resources. """
+        pass
+
+    @abc.abstractmethod
+    def create_minion(
+            self, ctxt, connection_info, environment_options,
+            new_minion_identifier):
+        pass
+
+    @abc.abstractmethod
+    def delete_minion(
+            self, ctxt, connection_info, environment_options,
+            minion_properties):
+        pass
+
+    @abc.abstractmethod
+    def shutdown_minion(
+            self, ctxt, connection_info, environment_options,
+            minion_properties):
+        pass
+
+    @abc.abstractmethod
+    def start_minion(
+            self, ctxt, connection_info, environment_options,
+            minion_properties):
+        pass
+
+    @abc.abstractmethod
+    def attach_volume_to_minion(
+            self, ctxt, connection_info, environment_options,
+            minion_properties, volume_info):
+        pass
+
+    @abc.abstractmethod
+    def detach_volume_from_minion(
+            self, ctxt, connection_info, environment_options,
+            minion_properties, volume_info):
+        pass

coriolis/providers/factory.py

@@ -49,7 +49,9 @@ PROVIDER_TYPE_MAP = {
     constants.PROVIDER_TYPE_DESTINATION_REPLICA_UPDATE: (
         base.BaseUpdateDestinationReplicaProvider),
     constants.PROVIDER_TYPE_SOURCE_ENDPOINT_OPTIONS: (
-        base.BaseEndpointSourceOptionsProvider)
+        base.BaseEndpointSourceOptionsProvider),
+    constants.PROVIDER_TYPE_MINION_POOL: (
+        base.BaseMinionPoolProvider)
 }
 
 

coriolis/schemas.py

@@ -10,7 +10,7 @@ import jsonschema
 from oslo_log import log as logging
 
 from coriolis import exception
-from coriolis import utils 
+from coriolis import utils
 
 
 LOG = logging.getLogger(__name__)
@@ -22,6 +22,8 @@ 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"
+PROVIDER_MINION_POOL_ENVIRONMENT_SCHEMA_NAME = (
+    "minion_pool_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/factory.py

@@ -4,6 +4,7 @@
 from coriolis import constants
 from coriolis import exception
 from coriolis.tasks import migration_tasks
+from coriolis.tasks import minion_pool_tasks
 from coriolis.tasks import osmorphing_tasks
 from coriolis.tasks import replica_tasks
 
@@ -81,7 +82,13 @@ _TASKS_MAP = {
     constants.TASK_TYPE_UPDATE_SOURCE_REPLICA:
         replica_tasks.UpdateSourceReplicaTask,
     constants.TASK_TYPE_UPDATE_DESTINATION_REPLICA:
-        replica_tasks.UpdateDestinationReplicaTask
+        replica_tasks.UpdateDestinationReplicaTask,
+    constants.TASK_TYPE_VALIDATE_MINION_POOL_OPTIONS:
+        minion_pool_tasks.ValidateMinionPoolOptionsTask,
+    constants.TASK_TYPE_CREATE_MINION:
+        minion_pool_tasks.CreateMinionTask,
+    constants.TASK_TYPE_DELETE_MINION:
+        minion_pool_tasks.DeleteMinionTask
 }
 
 

coriolis/tasks/minion_pool_tasks.py

@@ -0,0 +1,151 @@
+# 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.providers import factory as providers_factory
+from coriolis.tasks import base
+
+
+LOG = logging.getLogger(__name__)
+
+
+
+class ValidateMinionPoolOptionsTask(base.TaskRunner):
+
+    @classmethod
+    def get_required_platform(cls):
+        # TODO(aznashwan): this is only used to determined the Worker Service
+        # region of which endpoint to aim the Scheduler towards during normal
+        # transfer actions. Once the DB model hirearchy for transfer actions
+        # gets overhauled and this will be redundant, it should be removed.
+        return constants.TASK_PLATFORM_DESTINATION
+
+    @classmethod
+    def get_required_task_info_properties(cls):
+        return ["pool_environment_options"]
+
+    @classmethod
+    def get_returned_task_info_properties(cls):
+        return []
+
+    @classmethod
+    def get_required_provider_types(cls):
+        return {
+            # TODO(aznashwan): remove redundant doubling after
+            # transfer action DB model overhaul:
+            constants.PROVIDER_PLATFORM_SOURCE: [
+                constants.PROVIDER_TYPE_MINION_POOL],
+            constants.PROVIDER_PLATFORM_DESTINATION: [
+                constants.PROVIDER_TYPE_MINION_POOL],
+        }
+
+    def _run(self, ctxt, minion_pool_machine_id, origin, destination,
+             task_info, event_handler):
+
+        # NOTE: both origin or target endpoints would work:
+        connection_info = base.get_connection_info(ctxt, destination)
+        provider = providers_factory.get_provider(
+            destination["type"], constants.PROVIDER_TYPE_MINION_POOL,
+            event_handler)
+
+        environment_options = task_info['pool_environment_options']
+        provider.validate_pool_options(
+            ctxt, connection_info, environment_options)
+
+        return {}
+
+
+class CreateMinionTask(base.TaskRunner):
+
+    @classmethod
+    def get_required_platform(cls):
+        # TODO(aznashwan): this is only used to determined the Worker Service
+        # region of which endpoint to aim the Scheduler towards during normal
+        # transfer actions. Once the DB model hirearchy for transfer actions
+        # gets overhauled and this will be redundant, it should be removed.
+        return constants.TASK_PLATFORM_DESTINATION
+
+    @classmethod
+    def get_required_task_info_properties(cls):
+        return ["pool_environment_options"]
+
+    @classmethod
+    def get_returned_task_info_properties(cls):
+        return ["minion_provider_properties"]
+
+    @classmethod
+    def get_required_provider_types(cls):
+        return {
+            # TODO(aznashwan): remove redundant doubling after
+            # transfer action DB model overhaul:
+            constants.PROVIDER_PLATFORM_SOURCE: [
+                constants.PROVIDER_TYPE_MINION_POOL],
+            constants.PROVIDER_PLATFORM_DESTINATION: [
+                constants.PROVIDER_TYPE_MINION_POOL],
+        }
+
+    def _run(self, ctxt, minion_pool_machine_id, origin, destination,
+             task_info, event_handler):
+
+        # NOTE: both origin or target endpoints would work:
+        connection_info = base.get_connection_info(ctxt, destination)
+        provider = providers_factory.get_provider(
+            destination["type"], constants.PROVIDER_TYPE_MINION_POOL,
+            event_handler)
+
+        environment_options = task_info['pool_environment_options']
+        minion_provider_properties = provider.create_minion(
+            ctxt, connection_info, environment_options, minion_pool_machine_id)
+
+        return {"minion_provider_properties": minion_provider_properties}
+
+
+class DeleteMinionTask(base.TaskRunner):
+
+    @classmethod
+    def get_required_platform(cls):
+        # TODO(aznashwan): this is only used to determined the Worker Service
+        # region of which endpoint to aim the Scheduler towards during normal
+        # transfer actions. Once the DB model hirearchy for transfer actions
+        # gets overhauled and this will be redundant, it should be removed.
+        return constants.TASK_PLATFORM_DESTINATION
+
+    @classmethod
+    def get_required_task_info_properties(cls):
+        return ["pool_environment_options", "minion_provider_properties"]
+
+    @classmethod
+    def get_returned_task_info_properties(cls):
+        return ["minion_provider_properties"]
+
+    @classmethod
+    def get_required_provider_types(cls):
+        return {
+            # TODO(aznashwan): remove redundant doubling after
+            # transfer action DB model overhaul:
+            constants.PROVIDER_PLATFORM_SOURCE: [
+                constants.PROVIDER_TYPE_MINION_POOL],
+            constants.PROVIDER_PLATFORM_DESTINATION: [
+                constants.PROVIDER_TYPE_MINION_POOL],
+        }
+
+    def _run(self, ctxt, minion_pool_machine_id, origin, destination,
+             task_info, event_handler):
+
+        # NOTE: both origin or target endpoints would work:
+        connection_info = base.get_connection_info(ctxt, destination)
+        provider = providers_factory.get_provider(
+            destination["type"], constants.PROVIDER_TYPE_MINION_POOL,
+            event_handler)
+
+        environment_options = task_info['pool_environment_options']
+        minion_provider_properties = task_info['minion_provider_properties']
+        provider.delete_minion(
+            ctxt, connection_info, environment_options,
+            minion_provider_properties)
+
+        return {"minion_provider_properties": None}

coriolis/worker/rpc/server.py

@@ -506,6 +506,10 @@ class WorkerServerEndpoint(object):
             schema = provider.get_source_environment_schema()
             schemas["source_environment_schema"] = schema
 
+        if provider_type == constants.PROVIDER_TYPE_MINION_POOL:
+            schema = provider.get_minion_pool_environment_schema()
+            schemas["minion_pool_environment_schema"] = schema
+
         return schemas
 
     def get_diagnostics(self, ctxt):