Add Minion Pool initialization and allocation execution.

coriolis/api/v1/minion_pool_actions.py

@@ -15,6 +15,19 @@ class MinionPoolActionsController(api_wsgi.Controller):
         self.minion_pool_api = api.API()
         super(MinionPoolActionsController, self).__init__()
 
+    @api_wsgi.action('initialize')
+    def _initialize_pool(self, req, id, body):
+        context = req.environ['coriolis.context']
+        context.can(
+            minion_pool_policies.get_minion_pools_policy_label("initialize"))
+        try:
+            return minion_pool_tasks_execution_view.single(
+                req, self.minion_pool_api.initialize(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('allocate')
     def _allocate_pool(self, req, id, body):
         context = req.environ['coriolis.context']

coriolis/api/v1/minion_pools.py

@@ -37,7 +37,7 @@ class MinionPoolController(api_wsgi.Controller):
     def _validate_create_body(self, body):
         try:
             minion_pool = body["minion_pool"]
-            name = minion_pool["name"]
+            name = minion_pool["pool_name"]
             endpoint_id = minion_pool["endpoint_id"]
             # TODO(aznashwan): validate pool schema:
             environment_options = minion_pool["environment_options"]
@@ -47,10 +47,11 @@ class MinionPoolController(api_wsgi.Controller):
                 "minion_max_idle_time", 1)
             minion_retention_strategy = minion_pool.get(
                 "minion_retention_strategy")
+            notes = minion_pool.get("notes")
             return (
                 name, endpoint_id, environment_options, minimum_minions,
                 maximum_minions, minion_max_idle_time,
-                minion_retention_strategy)
+                minion_retention_strategy, notes)
         except Exception as ex:
             LOG.exception(ex)
             if hasattr(ex, "message"):
@@ -63,11 +64,13 @@ class MinionPoolController(api_wsgi.Controller):
         context = req.environ["coriolis.context"]
         context.can(pools_policies.get_minion_pools_policy_label("create"))
         (name, endpoint_id, environment_options, minimum_minions,
-         maximum_minions, minion_max_idle_time, minion_retention_strategy) = (
+         maximum_minions, minion_max_idle_time, minion_retention_strategy,
+         notes) = (
             self._validate_create_body(body))
         return minion_pool_view.single(req, self._minion_pool_api.create(
             context, name, endpoint_id, environment_options, minimum_minions,
-            maximum_minions, minion_max_idle_time, minion_retention_strategy))
+            maximum_minions, minion_max_idle_time, minion_retention_strategy,
+            notes=notes))
 
     def _validate_update_body(self, body):
         try:
@@ -75,7 +78,7 @@ class MinionPoolController(api_wsgi.Controller):
             return {k: minion_pool[k] for k in minion_pool.keys() &
                     {"name", "environment_options", "minimum_minions",
                      "maximum_minions", "minion_max_idle_time",
-                     "minion_retention_strategy"}}
+                     "minion_retention_strategy", "notes"}}
         except Exception as ex:
             LOG.exception(ex)
             if hasattr(ex, "message"):

coriolis/api/v1/views/minion_pool_view.py

@@ -13,6 +13,19 @@ def _format_minion_pool(req, minion_pool, keys=None):
     minion_pool_dict = dict(itertools.chain.from_iterable(
         transform(k, v) for k, v in minion_pool.items()))
 
+    # TODO(aznashwan): remove these redundancies once the base
+    # DB action model hirearchy will be overhauled:
+    for key in ["origin_endpoint_id", "destination_endpoint_id"]:
+        if key in minion_pool_dict:
+            minion_pool_dict["endpoint_id"] = minion_pool_dict.pop(key)
+    for key in ["source_environment", "destination_environment"]:
+        if key in minion_pool_dict:
+            minion_pool_dict["environment_options"] = minion_pool_dict.pop(key)
+
+    pool_machines = minion_pool_dict.get('minion_machines', [])
+    minion_pool_dict['minion_machines'] = [
+        machine['id'] for machine in pool_machines]
+
     return minion_pool_dict
 
 

coriolis/conductor/rpc/client.py

@@ -376,13 +376,18 @@ class ConductorClient(object):
     def create_minion_pool(
             self, ctxt, name, endpoint_id, environment_options,
             minimum_minions, maximum_minions, minion_max_idle_time,
-            minion_retention_strategy):
+            minion_retention_strategy, notes=None):
         return self._client.call(
             ctxt, 'create_minion_pool', name=name, endpoint_id=endpoint_id,
             environment_options=environment_options,
             minimum_minions=minimum_minions, maximum_minions=maximum_minions,
             minion_max_idle_time=minion_max_idle_time,
-            minion_retention_strategy=minion_retention_strategy)
+            minion_retention_strategy=minion_retention_strategy,
+            notes=notes)
+
+    def initialize_minion_pool(self, ctxt, minion_pool_id):
+        return self._client.call(
+            ctxt, "initialize_minion_pool", minion_pool_id=minion_pool_id)
 
     def allocate_minion_pool(self, ctxt, minion_pool_id):
         return self._client.call(

coriolis/conductor/rpc/server.py

@@ -143,6 +143,18 @@ def tasks_execution_synchronized(func):
     return wrapper
 
 
+def minion_pool_tasks_execution_synchronized(func):
+    @functools.wraps(func)
+    def wrapper(self, ctxt, minion_pool_id, execution_id, *args, **kwargs):
+        @lockutils.synchronized(
+            constants.EXECUTION_LOCK_NAME_FORMAT % execution_id,
+            external=True)
+        def inner():
+            return func(self, ctxt, minion_pool_id, execution_id, *args, **kwargs)
+        return inner()
+    return wrapper
+
+
 def region_synchronized(func):
     @functools.wraps(func)
     def wrapper(self, ctxt, region_id, *args, **kwargs):
@@ -169,12 +181,12 @@ def service_synchronized(func):
 
 def minion_pool_synchronized(func):
     @functools.wraps(func)
-    def wrapper(self, ctxt, pool_id, *args, **kwargs):
+    def wrapper(self, ctxt, minion_pool_id, *args, **kwargs):
         @lockutils.synchronized(
-            constants.MINION_POOL_LOCK_NAME_FORMAT % pool_id,
+            constants.MINION_POOL_LOCK_NAME_FORMAT % minion_pool_id,
             external=True)
         def inner():
-            return func(self, ctxt, pool_id, *args, **kwargs)
+            return func(self, ctxt, minion_pool_id, *args, **kwargs)
         return inner()
     return wrapper
 
@@ -764,7 +776,7 @@ class ConductorServerEndpoint(object):
                             if missing_deps:
                                 raise exception.CoriolisException(
                                     "Task '%s' (type '%s') for instance '%s' "
-                                    "has non-exitent tasks referenced as "
+                                    "has non-existent tasks referenced as "
                                     "dependencies: %s" % (
                                         task.id, task.task_type,
                                         instance, missing_deps))
@@ -2198,6 +2210,16 @@ class ConductorServerEndpoint(object):
     def _handle_post_task_actions(self, ctxt, task, execution, task_info):
         task_type = task.task_type
 
+        def _check_other_tasks_running(execution, current_task):
+            still_running = False
+            for other_task in execution.tasks:
+                if other_task.id == current_task.id:
+                    continue
+                if other_task.status in constants.ACTIVE_TASK_STATUSES:
+                    still_running = True
+                    break
+            return still_running
+
         if task_type == constants.TASK_TYPE_RESTORE_REPLICA_DISK_SNAPSHOTS:
 
             # When restoring a snapshot in some import providers (OpenStack),
@@ -2278,13 +2300,7 @@ class ConductorServerEndpoint(object):
 
             if task_type == constants.TASK_TYPE_UPDATE_DESTINATION_REPLICA:
                 # check if this was the last task in the update execution:
-                still_running = False
-                for other_task in execution.tasks:
-                    if other_task.id == task.id:
-                        continue
-                    if other_task.status in constants.ACTIVE_TASK_STATUSES:
-                        still_running = True
-                        break
+                still_running = _check_other_tasks_running(execution, task)
                 if not still_running:
                     # it means this was the last update task in the Execution
                     # and we may safely update the params of the Replica
@@ -2299,6 +2315,70 @@ class ConductorServerEndpoint(object):
                     # update task finishes last:
                     db_api.update_replica(
                         ctxt, execution.action_id, task_info)
+
+        elif task_type == constants.TASK_TYPE_SET_UP_SHARED_POOL_RESOURCES:
+            still_running = _check_other_tasks_running(execution, task)
+            if not still_running:
+                LOG.info(
+                    "Updating 'pool_supporting_resources' for pool %s after "
+                    "completion of task '%s' (type '%s').",
+                    execution.action_id, task.id, task_type)
+                db_api.update_minion_pool_lifecycle(
+                    ctxt, execution.action_id, {
+                        "pool_supporting_resources": task_info.get(
+                            "pool_supporting_resources", {})})
+                db_api.set_minion_pool_lifecycle_status(
+                    ctxt, execution.action_id,
+                    constants.MINION_POOL_STATUS_DEALLOCATED)
+
+        elif task_type == constants.TASK_TYPE_TEAR_DOWN_SHARED_POOL_RESOURCES:
+            still_running = _check_other_tasks_running(execution, task)
+            if not still_running:
+                LOG.info(
+                    "Clearing 'pool_supporting_resources' for pool %s following "
+                    "completion of task '%s' (type %s)",
+                    execution.action_id, task.id, task_type)
+                db_api.update_minion_pool_lifecycle(
+                    ctxt, execution.action_id, {
+                        "pool_supporting_resources": {}})
+                db_api.set_minion_pool_lifecycle_status(
+                    ctxt, execution.action_id,
+                    constants.MINION_POOL_STATUS_UNINITIALIZED)
+
+        elif task_type == constants.TASK_TYPE_CREATE_MINION:
+            LOG.info(
+                "Updating properties for Minion Machine '%s' of pool %s "
+                "following completion of task '%s' (type %s).",
+                task.instance, execution.action_id, task.id, task_type)
+            db_api.update_minion_machine(
+                # NOTE: we used the Minion Machine ID as the tasks' instance.
+                ctxt, task.instance, {
+                    "status": constants.MINION_MACHINE_STATUS_AVAILABLE,
+                    "provider_properties": task_info[
+                        "minion_provider_properties"],
+                    "connection_info": task_info[
+                        "minion_connection_info"]})
+
+            still_running = _check_other_tasks_running(execution, task)
+            if not still_running:
+                db_api.set_minion_pool_lifecycle_status(
+                    ctxt, execution.action_id,
+                    constants.MINION_POOL_STATUS_AVAILABLE)
+
+        elif task_type == constants.TASK_TYPE_DELETE_MINION:
+            LOG.info(
+                "Clearing properties for Minion Machine '%s' of pool %s "
+                "following completion of task '%s' (type %s).",
+                task.instance, execution.action_id, task.id, task_type)
+            db_api.update_minion_machine(
+                # NOTE: we used the Minion Machine ID as the tasks' instance.
+                ctxt, task.instance, {
+                    "status": constants.MINION_MACHINE_STATUS_UNINITIALIZED,
+                    "provider_properties": task_info[
+                        "minion_provider_properties"],
+                    "connection_info": task_info[
+                        "minion_connection_info"]})
+
         else:
             LOG.debug(
                 "No post-task actions required for task '%s' of type '%s'",
@@ -2913,12 +2993,13 @@ class ConductorServerEndpoint(object):
     def create_minion_pool(
             self, ctxt, name, endpoint_id, environment_options,
             minimum_minions, maximum_minions, minion_max_idle_time,
-            minion_retention_strategy):
+            minion_retention_strategy, notes=None):
         endpoint = db_api.get_endpoint(ctxt, endpoint_id)
 
         minion_pool = models.MinionPoolLifecycle()
         minion_pool.id = str(uuid.uuid4())
-        minion_pool.name = name
+        minion_pool.pool_name = name
+        minion_pool.notes = notes
         minion_pool.pool_status = constants.MINION_POOL_STATUS_UNINITIALIZED
         minion_pool.minimum_minions = minimum_minions
         minion_pool.maximum_minions = maximum_minions
@@ -2930,7 +3011,7 @@ class ConductorServerEndpoint(object):
         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.destination_environment = environment_options
         minion_pool.instances = []
         minion_pool.info = {}
 
@@ -2945,19 +3026,68 @@ class ConductorServerEndpoint(object):
         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)
+                "Minion pool with ID '%s' not found." % minion_pool_id)
         return minion_pool
 
+    @minion_pool_synchronized
+    def initialize_minion_pool(self, ctxt, minion_pool_id):
+        LOG.info("Attempting to initialize Minion Pool '%s'.", minion_pool_id)
+        minion_pool = db_api.get_minion_pool_lifecycle(ctxt, minion_pool_id)
+        if minion_pool.pool_status != constants.MINION_POOL_STATUS_UNINITIALIZED:
+            raise exception.InvalidMinionPoolState(
+                "Minion Pool '%s' cannot be initialized as it is in '%s' state "
+                "instead of the expected %s."% (
+                    minion_pool_id, minion_pool.pool_status,
+                    constants.MINION_POOL_STATUS_UNINITIALIZED))
+
+        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_INITIALIZATION
+
+        minion_pool.info[minion_pool_id] = {
+            "pool_identifier": minion_pool.id,
+            # TODO(aznashwan): remove redundancy once transfer
+            # action DB models have been overhauled:
+            "pool_environment_options": minion_pool.source_environment}
+
+        validate_pool_options_task = self._create_task(
+            minion_pool.id,
+            constants.TASK_TYPE_VALIDATE_MINION_POOL_OPTIONS,
+            execution)
+
+        setup_pool_resources_task = self._create_task(
+            minion_pool.id,
+            constants.TASK_TYPE_SET_UP_SHARED_POOL_RESOURCES,
+            execution,
+            depends_on=[validate_pool_options_task.id])
+
+        self._check_execution_tasks_sanity(execution, minion_pool.info)
+
+        # update the action info for the pool's instance:
+        db_api.update_transfer_action_info_for_instance(
+            ctxt, minion_pool.id, minion_pool.id,
+            minion_pool.info[minion_pool.id])
+
+        # add new execution to DB:
+        db_api.add_minion_pool_lifecycle_execution(ctxt, execution)
+        LOG.info("Minion pool initialization 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).to_dict()
+
     @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):
+        minion_pool = self._get_minion_pool(ctxt, minion_pool_id)
+        if minion_pool.pool_status != 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))
+                "Minion Pool '%s' cannot be started as it is in '%s' state "
+                "instead of the expected %s."% (
+                    minion_pool_id, minion_pool.pool_status,
+                    constants.MINION_POOL_STATUS_DEALLOCATED))
 
         execution = models.TasksExecution()
         execution.id = str(uuid.uuid4())
@@ -3022,18 +3152,18 @@ class ConductorServerEndpoint(object):
         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)
+        return self._get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution.id).to_dict()
 
     @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 (
+        if minion_pool.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))
+                    minion_pool_id, minion_pool.pool_status))
 
         # TODO(aznashwan): check minion pool running
         # executions/allocated machines
@@ -3044,8 +3174,8 @@ class ConductorServerEndpoint(object):
         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:
+        for minion_machine in minion_pool.minion_machines:
+            minion_machine_id = minion_machine.id
             minion_pool.info[minion_machine_id] = {
                 "pool_environment_options": minion_pool.source_environment}
             self._create_task(
@@ -3065,8 +3195,8 @@ class ConductorServerEndpoint(object):
         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)
+        return self._get_minion_pool_lifecycle_execution(
+            ctxt, minion_pool_id, execution.id).to_dict()
 
     @minion_pool_synchronized
     def get_minion_pool(self, ctxt, minion_pool_id):
@@ -3103,13 +3233,13 @@ class ConductorServerEndpoint(object):
                     execution_id, minion_pool_id))
         return execution
 
-    @tasks_execution_synchronized
+    @minion_pool_tasks_execution_synchronized
     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)
+            ctxt, minion_pool_id, execution_id).to_dict()
 
-    @tasks_execution_synchronized
+    @minion_pool_tasks_execution_synchronized
     def delete_minion_pool_lifecycle_execution(
             self, ctxt, minion_pool_id, execution_id):
         execution = self._get_minion_pool_lifecycle_execution(
@@ -3121,7 +3251,7 @@ class ConductorServerEndpoint(object):
                     execution_id, minion_pool_id, execution.status))
         db_api.delete_minion_pool_lifecycle_execution(ctxt, execution_id)
 
-    @tasks_execution_synchronized
+    @minion_pool_tasks_execution_synchronized
     def cancel_minion_pool_lifecycle_execution(
             self, ctxt, minion_pool_id, execution_id, force):
         execution = self._get_minion_pool_lifecycle_execution(

coriolis/constants.py

@@ -219,7 +219,8 @@ 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"
+EXECUTION_TYPE_MINION_POOL_MAINTENANCE = "pool_maintenance"
+EXECUTION_TYPE_MINION_POOL_INITIALIZATION = "pool_initialization"
 
 TASK_LOCK_NAME_FORMAT = "task-%s"
 EXECUTION_LOCK_NAME_FORMAT = "execution-%s"
@@ -237,7 +238,8 @@ EXECUTION_TYPE_TO_ACTION_LOCK_NAME_FORMAT_MAP = {
     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_MINION_POOL_MAINTENANCE: MINION_POOL_LOCK_NAME_FORMAT
+    EXECUTION_TYPE_MINION_POOL_MAINTENANCE: MINION_POOL_LOCK_NAME_FORMAT,
+    EXECUTION_TYPE_MINION_POOL_INITIALIZATION: MINION_POOL_LOCK_NAME_FORMAT
 }
 
 SERVICE_STATUS_UP = "UP"

coriolis/db/api.py

@@ -615,26 +615,28 @@ def update_transfer_action_info_for_instance(
 
     # Copy is needed, otherwise sqlalchemy won't save the changes
     action_info = action.info.copy()
+    instance_info_old = {}
     if instance in action_info:
         instance_info_old = action_info[instance]
-        old_keys = set(instance_info_old.keys())
-        new_keys = set(new_instance_info.keys())
-        overwritten_keys = old_keys.intersection(new_keys)
-        if overwritten_keys:
-            LOG.debug(
-                "Overwriting the values of the following keys for info of "
-                "instance '%s' of action with ID '%s': %s",
-                instance, action_id, overwritten_keys)
-        newly_added_keys = new_keys.difference(old_keys)
-        if newly_added_keys:
-            LOG.debug(
-                "The following new keys will be added for info of instance "
-                "'%s' in action with ID '%s': %s",
-                instance, action_id, newly_added_keys)
 
-        instance_info_old_copy = instance_info_old.copy()
-        instance_info_old_copy.update(new_instance_info)
-        action_info[instance] = instance_info_old_copy
+    old_keys = set(instance_info_old.keys())
+    new_keys = set(new_instance_info.keys())
+    overwritten_keys = old_keys.intersection(new_keys)
+    if overwritten_keys:
+        LOG.debug(
+            "Overwriting the values of the following keys for info of "
+            "instance '%s' of action with ID '%s': %s",
+            instance, action_id, overwritten_keys)
+    newly_added_keys = new_keys.difference(old_keys)
+    if newly_added_keys:
+        LOG.debug(
+            "The following new keys will be added for info of instance "
+            "'%s' in action with ID '%s': %s",
+            instance, action_id, newly_added_keys)
+
+    instance_info_old_copy = instance_info_old.copy()
+    instance_info_old_copy.update(new_instance_info)
+    action_info[instance] = instance_info_old_copy
     action.info = action_info
 
     return action_info[instance]
@@ -1099,15 +1101,12 @@ def add_minion_machine(context, minion_machine):
 @enginefacade.reader
 def get_minion_machines(context):
     q = _soft_delete_aware_query(context, models.MinionMachine)
-    q = q.options(orm.joinedload('mapped_services'))
     return q.all()
 
 
 @enginefacade.reader
 def get_minion_machine(context, minion_machine_id):
     q = _soft_delete_aware_query(context, models.MinionMachine)
-    q = q.options(orm.joinedload('mapped_endpoints'))
-    q = q.options(orm.joinedload('mapped_services'))
     return q.filter(
         models.MinionMachine.id == minion_machine_id).first()
 
@@ -1122,7 +1121,7 @@ def update_minion_machine(context, minion_machine_id, updated_values):
         raise exception.NotFound(
             "MinionMachine with ID '%s' does not exist." % minion_machine_id)
 
-    updateable_fields = ["connection_info"]
+    updateable_fields = ["connection_info", "provider_properties", "status"]
     _update_sqlalchemy_object_fields(
         minion_machine, updateable_fields, updated_values)
 
@@ -1150,9 +1149,14 @@ def delete_minion_pool_lifecycle(context, minion_pool_id):
 
 
 @enginefacade.reader
-def get_minion_pool_lifecycle(context, minion_pool_id):
+def get_minion_pool_lifecycle(
+        context, minion_pool_id, include_tasks_executions=True,
+        include_machines=True):
     q = _soft_delete_aware_query(context, models.MinionPoolLifecycle)
-    q = q.options(orm.joinedload(models.MinionPoolLifecycle.executions))
+    if include_tasks_executions:
+        q = q.options(orm.joinedload(models.MinionPoolLifecycle.executions))
+    if include_machines:
+        q = q.options(orm.joinedload('minion_machines'))
     if is_user_context(context):
         q = q.filter(
             models.MinionPoolLifecycle.project_id == context.tenant)
@@ -1173,6 +1177,7 @@ def get_minion_pool_lifecycles(
     if is_user_context(context):
         q = q.filter(
             models.Replica.project_id == context.tenant)
+    q = q.options(orm.joinedload('minion_machines'))
     db_result = q.all()
     if to_dict:
         return [i.to_dict(include_info=include_info) for i in db_result]
@@ -1194,16 +1199,31 @@ def add_minion_pool_lifecycle_execution(context, execution):
     _session(context).add(execution)
 
 
+@enginefacade.writer
+def set_minion_pool_lifecycle_status(context, minion_pool_id, status):
+    pool = get_minion_pool_lifecycle(
+        context, minion_pool_id, include_tasks_executions=False,
+        include_machines=False)
+    LOG.debug(
+        "Transitioning minion pool '%s' from status '%s' to '%s'in DB",
+        minion_pool_id, pool.pool_status, status)
+    pool.pool_status = status
+    setattr(pool, 'updated_at', timeutils.utcnow())
+
+
 @enginefacade.writer
 def update_minion_pool_lifecycle(context, minion_pool_id, updated_values):
-    lifecycle = get_minion_pool_lifecycle(context, minion_pool_id)
+    lifecycle = get_minion_pool_lifecycle(
+        context, minion_pool_id, include_tasks_executions=False,
+        include_machines=False)
     if not lifecycle:
         raise exception.NotFound(
             "Minion pool '%s' not found" % minion_pool_id)
 
     updateable_fields = [
         "minimum_minions", "maximum_minions", "minion_max_idle_time",
-        "minion_retention_strategy", "environment_options"]
+        "minion_retention_strategy", "environment_options",
+        "pool_supporting_resources", "notes", "pool_name"]
     # TODO(aznashwan): this should no longer be required when the
     # transfer action class hirearchy is to be overhauled:
     redundancies = {
@@ -1249,7 +1269,7 @@ def get_minion_pool_lifecycle_executions(
 @enginefacade.reader
 def get_minion_pool_lifecycle_execution(context, lifecycle_id, execution_id):
     q = _soft_delete_aware_query(context, models.TasksExecution).join(
-        models.Replica)
+        models.MinionPoolLifecycle)
     q = _get_tasks_with_details_options(q)
     if is_user_context(context):
         q = q.filter(models.MinionPoolLifecycle.project_id == context.tenant)

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

@@ -26,9 +26,13 @@ def upgrade(migrate_engine):
                 "id", sqlalchemy.String(36),
                 sqlalchemy.ForeignKey('base_transfer_action.base_id'),
                 primary_key=True),
+            sqlalchemy.Column(
+                "pool_name", sqlalchemy.String(255), nullable=False),
             sqlalchemy.Column(
                 "pool_status", sqlalchemy.String(255), nullable=False,
                 default=lambda: "UNKNOWN"),
+            sqlalchemy.Column(
+                "pool_supporting_resources", sqlalchemy.Text, nullable=True),
             sqlalchemy.Column(
                 'minimum_minions', sqlalchemy.Integer, nullable=False),
             sqlalchemy.Column(
@@ -46,13 +50,17 @@ def upgrade(migrate_engine):
             meta,
             sqlalchemy.Column('id', sqlalchemy.String(36), primary_key=True,
                               default=lambda: str(uuid.uuid4())),
+            sqlalchemy.Column(
+                "user_id", sqlalchemy.String(255), nullable=False),
+            sqlalchemy.Column(
+                "project_id", sqlalchemy.String(255), nullable=False),
             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(
                 'pool_id', sqlalchemy.String(36),
-                sqlalchemy.ForeignKey('minion_pool.id'), nullable=False),
+                sqlalchemy.ForeignKey('minion_pool_lifecycle.id'), nullable=False),
             sqlalchemy.Column(
                 'status', sqlalchemy.String(255), nullable=False,
                 default=lambda: "UNKNOWN"),
@@ -67,14 +75,3 @@ def upgrade(migrate_engine):
             # to the previously existing state.
             meta.drop_all(tables=tables[:index])
             raise
-
-    # update base_transfer_action:
-    columns = [
-        sqlalchemy.Column(
-            "source_minion_pool_id", sqlalchemy.String(36),
-            sqlalchemy.ForeignKey('minion_pool_lifecycle.id'), nullable=True),
-        sqlalchemy.Column(
-            "destination_minion_pool_id", sqlalchemy.String(36),
-            sqlalchemy.ForeignKey('minion_pool_lifecycle.id'), nullable=True)]
-    for col in columns:
-        base_transfer_action.create_column(col)

coriolis/db/sqlalchemy/models.py

@@ -143,7 +143,7 @@ class TasksExecution(BASE, models.TimestampMixin, models.ModelBase,
                              backref=orm.backref('execution'))
     status = sqlalchemy.Column(sqlalchemy.String(100), nullable=False)
     number = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
-    type = sqlalchemy.Column(sqlalchemy.String(20))
+    type = sqlalchemy.Column(sqlalchemy.String(255))
 
     def to_dict(self):
         result = {
@@ -196,18 +196,6 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
     network_map = sqlalchemy.Column(types.Json, nullable=True)
     storage_mappings = sqlalchemy.Column(types.Json, nullable=True)
     source_environment = sqlalchemy.Column(types.Json, nullable=True)
-    source_minion_pool_id = sqlalchemy.Column(
-        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',
@@ -245,55 +233,6 @@ class BaseTransferAction(BASE, models.TimestampMixin, models.ModelBase,
         return result
 
 
-class MinionPoolLifecycle(BaseTransferAction):
-    # TODO(aznashwan): this class inherits numerous redundant fields from
-    # BaseTransferAction. Ideally, the upper hirearchy should be split into a
-    # BaseAction, and a separate inheriting BaseTransferAction.
-    __tablename__ = 'minion_pool_lifecycle'
-
-    id = sqlalchemy.Column(
-        sqlalchemy.String(36),
-        sqlalchemy.ForeignKey(
-            '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(
-        sqlalchemy.Integer, nullable=False)
-    minion_max_idle_time = sqlalchemy.Column(
-        sqlalchemy.Integer, nullable=False)
-    minion_retention_strategy = sqlalchemy.Column(
-        sqlalchemy.String(255), nullable=False)
-
-    __mapper_args__ = {
-        'polymorphic_identity': 'minion_pool_lifecycle'}
-
-    def to_dict(self, include_info=True):
-        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
-
-
 class Replica(BaseTransferAction):
     __tablename__ = 'replica'
 
@@ -513,9 +452,6 @@ class MinionMachine(BASE, models.TimestampMixin, models.ModelBase,
         sqlalchemy.String(36),
         sqlalchemy.ForeignKey('minion_pool_lifecycle.id'),
         nullable=False)
-    pool = orm.relationship(
-        MinionPoolLifecycle, backref=orm.backref("minion_machines"),
-        foreign_keys=[pool_id])
 
     status = sqlalchemy.Column(
         sqlalchemy.String(255), nullable=False,
@@ -524,3 +460,66 @@ class MinionMachine(BASE, models.TimestampMixin, models.ModelBase,
     connection_info = sqlalchemy.Column(types.Json)
 
     provider_properties = sqlalchemy.Column(types.Json)
+
+
+
+class MinionPoolLifecycle(BaseTransferAction):
+    # TODO(aznashwan): this class inherits numerous redundant fields from
+    # BaseTransferAction. Ideally, the upper hirearchy should be split into a
+    # BaseAction, and a separate inheriting BaseTransferAction.
+    __tablename__ = 'minion_pool_lifecycle'
+
+    id = sqlalchemy.Column(
+        sqlalchemy.String(36),
+        sqlalchemy.ForeignKey(
+            'base_transfer_action.base_id'),
+        primary_key=True)
+
+    pool_name = sqlalchemy.Column(
+        sqlalchemy.String(255),
+        nullable=False)
+    pool_status = sqlalchemy.Column(
+        sqlalchemy.String(255), nullable=False,
+        default=lambda: constants.MINION_POOL_STATUS_UNKNOWN)
+    pool_supporting_resources = sqlalchemy.Column(
+        types.Json, nullable=True)
+    minimum_minions = sqlalchemy.Column(
+        sqlalchemy.Integer, nullable=False)
+    maximum_minions = sqlalchemy.Column(
+        sqlalchemy.Integer, nullable=False)
+    minion_max_idle_time = sqlalchemy.Column(
+        sqlalchemy.Integer, nullable=False)
+    minion_retention_strategy = sqlalchemy.Column(
+        sqlalchemy.String(255), nullable=False)
+    minion_machines = orm.relationship(
+        MinionMachine, backref=orm.backref('minion_pool'),
+        primaryjoin="and_(MinionMachine.pool_id==MinionPoolLifecycle.id, "
+                    "MinionMachine.deleted=='0')")
+
+    __mapper_args__ = {
+        'polymorphic_identity': 'minion_pool_lifecycle'}
+
+    def to_dict(self, include_info=True):
+        base = super(MinionPoolLifecycle, self).to_dict(
+            include_info=include_info)
+        base.update({
+            "id": self.id,
+            "pool_name": self.pool_name,
+            "pool_supporting_resources": self.pool_supporting_resources,
+            "pool_status": self.pool_status,
+            "minimum_minions": self.minimum_minions,
+            "maximum_minions": self.maximum_minions,
+            "minion_max_idle_time": self.minion_max_idle_time,
+            "minion_retention_strategy": self.minion_retention_strategy})
+        # 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

coriolis/minion_pools/api.py

@@ -12,10 +12,11 @@ class API(object):
     def create(
             self, ctxt, name, endpoint_id, environment_options,
             minimum_minions, maximum_minions, minion_max_idle_time,
-            minion_retention_strategy):
+            minion_retention_strategy, notes=None):
         return self._rpc_client.create_minion_pool(
             ctxt, name, endpoint_id, environment_options, minimum_minions,
-            maximum_minions, minion_max_idle_time, minion_retention_strategy)
+            maximum_minions, minion_max_idle_time, minion_retention_strategy,
+            notes=notes)
 
     def update(self, ctxt, minion_pool_id, updated_values):
         return self._rpc_client.update_minion_pool(
@@ -30,6 +31,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 initialize(self, ctxt, minion_pool_id):
+        return self._rpc_client.initialize_minion_pool(ctxt, minion_pool_id)
+
     def allocate(self, ctxt, minion_pool_id):
         return self._rpc_client.allocate_minion_pool(ctxt, minion_pool_id)
 

coriolis/policies/minion_pools.py

@@ -72,6 +72,18 @@ MINION_POOLS_DEFAULT_RULES = [
             }
         ]
     ),
+    policy.DocumentedRuleDefault(
+        get_minion_pools_policy_label('initialize'),
+        MINION_POOLS_DEFAULT_RULE,
+        "Initialize Minion Pool",
+        [
+            {
+                "path": "/minion_pools/{minion_pool_id}/actions",
+                "method": "POST"
+            }
+        ]
+    ),
+
     policy.DocumentedRuleDefault(
         get_minion_pools_policy_label('allocate'),
         MINION_POOLS_DEFAULT_RULE,

coriolis/providers/base.py

@@ -568,7 +568,7 @@ class BaseMinionPoolProvider(
 
     @abc.abstractmethod
     def setup_pool_supporting_resources(
-            self, ctxt, connection_info, environment_options):
+            self, ctxt, connection_info, environment_options, pool_identifier):
         """ Sets up supporting resources which can be re-used amongst the
         machines which will be spawned within the pool (e.g. a shared network)
         """
@@ -577,7 +577,7 @@ class BaseMinionPoolProvider(
     @abc.abstractmethod
     def teardown_pool_supporting_resources(
             self, ctxt, connection_info, environment_options,
-            pool_resource_info):
+            pool_supporting_resources):
         """ Tears down all pool supporting resources. """
         pass
 

coriolis/tasks/factory.py

@@ -88,7 +88,11 @@ _TASKS_MAP = {
     constants.TASK_TYPE_CREATE_MINION:
         minion_pool_tasks.CreateMinionTask,
     constants.TASK_TYPE_DELETE_MINION:
-        minion_pool_tasks.DeleteMinionTask
+        minion_pool_tasks.DeleteMinionTask,
+    constants.TASK_TYPE_SET_UP_SHARED_POOL_RESOURCES:
+        minion_pool_tasks.SetupPoolSupportingResourcesTask,
+    constants.TASK_TYPE_TEAR_DOWN_SHARED_POOL_RESOURCES:
+        minion_pool_tasks.TeardownPoolSupportingResourcesTask
 }
 
 

coriolis/tasks/minion_pool_tasks.py

@@ -4,8 +4,6 @@
 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
 
@@ -75,7 +73,7 @@ class CreateMinionTask(base.TaskRunner):
 
     @classmethod
     def get_returned_task_info_properties(cls):
-        return ["minion_provider_properties"]
+        return ["minion_provider_properties", "minion_connection_info"]
 
     @classmethod
     def get_required_provider_types(cls):
@@ -98,10 +96,14 @@ class CreateMinionTask(base.TaskRunner):
             event_handler)
 
         environment_options = task_info['pool_environment_options']
-        minion_provider_properties = provider.create_minion(
+        minion_properties = provider.create_minion(
             ctxt, connection_info, environment_options, minion_pool_machine_id)
 
-        return {"minion_provider_properties": minion_provider_properties}
+        return {
+            "minion_connection_info": minion_properties[
+                "minion_connection_info"],
+            "minion_provider_properties": minion_properties[
+                "minion_provider_properties"]}
 
 
 class DeleteMinionTask(base.TaskRunner):
@@ -120,7 +122,7 @@ class DeleteMinionTask(base.TaskRunner):
 
     @classmethod
     def get_returned_task_info_properties(cls):
-        return ["minion_provider_properties"]
+        return ["minion_provider_properties", "minion_connection_info"]
 
     @classmethod
     def get_required_provider_types(cls):
@@ -148,4 +150,99 @@ class DeleteMinionTask(base.TaskRunner):
             ctxt, connection_info, environment_options,
             minion_provider_properties)
 
-        return {"minion_provider_properties": None}
+        return {
+            "minion_provider_properties": None,
+            "minion_connection_info": None}
+
+
+class SetupPoolSupportingResourcesTask(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", "pool_identifier"]
+
+    @classmethod
+    def get_returned_task_info_properties(cls):
+        return ["pool_supporting_resources"]
+
+    @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)
+
+        pool_identifier = task_info['pool_identifier']
+        environment_options = task_info['pool_environment_options']
+        pool_supporting_resources = provider.setup_pool_supporting_resources(
+            ctxt, connection_info, environment_options, pool_identifier)
+
+        return {"pool_supporting_resources": pool_supporting_resources}
+
+
+class TeardownPoolSupportingResourcesTask(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", "pool_supporting_resources"]
+
+    @classmethod
+    def get_returned_task_info_properties(cls):
+        return ["pool_supporting_resources"]
+
+    @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']
+        pool_supporting_resources = task_info['pool_supporting_resources']
+        provider.teardown_pool_supporting_resources(
+            ctxt, connection_info, environment_options,
+            pool_supporting_resources)
+
+        return {"pool_supporting_resources": None}