Adds replica id to tasks execution actions

coriolis/api/v1/replica_tasks_execution_actions.py

@@ -19,7 +19,7 @@ class ReplicaTasksExecutionActionsController(api_wsgi.Controller):
             force = (body["cancel"] or {}).get("force", False)
 
             self._replica_tasks_execution_api.cancel(
-                req.environ['coriolis.context'], id, force)
+                req.environ['coriolis.context'], replica_id, id, force)
             raise exc.HTTPNoContent()
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)

coriolis/api/v1/replica_tasks_executions.py

@@ -16,7 +16,7 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
 
     def show(self, req, replica_id, id):
         execution = self._replica_tasks_execution_api.get_execution(
-            req.environ["coriolis.context"], id)
+            req.environ["coriolis.context"], replica_id, id)
         if not execution:
             raise exc.HTTPNotFound()
 
@@ -48,7 +48,7 @@ class ReplicaTasksExecutionController(api_wsgi.Controller):
     def delete(self, req, replica_id, id):
         try:
             self._replica_tasks_execution_api.delete(
-                req.environ['coriolis.context'], id)
+                req.environ['coriolis.context'], replica_id, id)
             raise exc.HTTPNoContent()
         except exception.NotFound as ex:
             raise exc.HTTPNotFound(explanation=ex.msg)

coriolis/conductor/rpc/client.py

@@ -26,19 +26,20 @@ class ConductorClient(object):
             replica_id=replica_id,
             include_tasks=include_tasks)
 
-    def get_replica_tasks_execution(self, ctxt, execution_id):
+    def get_replica_tasks_execution(self, ctxt, replica_id, execution_id):
         return self._client.call(
-            ctxt, 'get_replica_tasks_execution',
+            ctxt, 'get_replica_tasks_execution', replica_id=replica_id,
             execution_id=execution_id)
 
-    def delete_replica_tasks_execution(self, ctxt, execution_id):
+    def delete_replica_tasks_execution(self, ctxt, replica_id, execution_id):
         return self._client.call(
-            ctxt, 'delete_replica_tasks_execution',
+            ctxt, 'delete_replica_tasks_execution', replica_id=replica_id,
             execution_id=execution_id)
 
-    def cancel_replica_tasks_execution(self, ctxt, execution_id, force):
+    def cancel_replica_tasks_execution(self, ctxt, replica_id, execution_id,
+                                       force):
         return self._client.call(
-            ctxt, 'cancel_replica_tasks_execution',
+            ctxt, 'cancel_replica_tasks_execution', replica_id=replica_id,
             execution_id=execution_id, force=force)
 
     def create_instances_replica(self, ctxt, origin, destination, instances):

coriolis/conductor/rpc/server.py

@@ -52,10 +52,10 @@ def migration_synchronized(func):
 
 def tasks_execution_synchronized(func):
     @functools.wraps(func)
-    def wrapper(self, ctxt, execution_id, *args, **kwargs):
+    def wrapper(self, ctxt, replica_id, execution_id, *args, **kwargs):
         @lockutils.synchronized(execution_id)
         def inner():
-            return func(self, ctxt, execution_id, *args, **kwargs)
+            return func(self, ctxt, replica_id, execution_id, *args, **kwargs)
         return inner()
     return wrapper
 
@@ -154,7 +154,7 @@ class ConductorServerEndpoint(object):
         LOG.info("Replica tasks execution created: %s", execution.id)
 
         self._begin_tasks(ctxt, execution, replica.info)
-        return self.get_replica_tasks_execution(ctxt, execution.id)
+        return self.get_replica_tasks_execution(ctxt, replica_id, execution.id)
 
     @replica_synchronized
     def get_replica_tasks_executions(self, ctxt, replica_id,
@@ -163,31 +163,32 @@ class ConductorServerEndpoint(object):
             ctxt, replica_id, include_tasks)
 
     @tasks_execution_synchronized
-    def get_replica_tasks_execution(self, ctxt, execution_id):
+    def get_replica_tasks_execution(self, ctxt, replica_id, execution_id):
         return self._get_replica_tasks_execution(
-            ctxt, execution_id)
+            ctxt, replica_id, execution_id)
 
     @tasks_execution_synchronized
-    def delete_replica_tasks_execution(self, ctxt, execution_id):
+    def delete_replica_tasks_execution(self, ctxt, replica_id, execution_id):
         execution = self._get_replica_tasks_execution(
-            ctxt, execution_id)
+            ctxt, replica_id, execution_id)
         if execution.status == constants.EXECUTION_STATUS_RUNNING:
             raise exception.InvalidMigrationState(
                 "Cannot delete a running replica tasks execution")
         db_api.delete_replica_tasks_execution(ctxt, execution_id)
 
     @tasks_execution_synchronized
-    def cancel_replica_tasks_execution(self, ctxt, execution_id, force):
+    def cancel_replica_tasks_execution(self, ctxt, replica_id, execution_id,
+                                       force):
         execution = self._get_replica_tasks_execution(
-            ctxt, execution_id)
+            ctxt, replica_id, execution_id)
         if execution.status != constants.EXECUTION_STATUS_RUNNING:
             raise exception.InvalidReplicaState(
                 "The replica tasks execution is not running")
         self._cancel_tasks_execution(ctxt, execution, force)
 
-    def _get_replica_tasks_execution(self, ctxt, execution_id):
+    def _get_replica_tasks_execution(self, ctxt, replica_id, execution_id):
         execution = db_api.get_replica_tasks_execution(
-            ctxt, execution_id)
+            ctxt, replica_id, execution_id)
         if not execution:
             raise exception.NotFound("Tasks execution not found")
         return execution
@@ -234,7 +235,7 @@ class ConductorServerEndpoint(object):
         LOG.info("Replica tasks execution created: %s", execution.id)
 
         self._begin_tasks(ctxt, execution, replica.info)
-        return self.get_replica_tasks_execution(ctxt, execution.id)
+        return self.get_replica_tasks_execution(ctxt, replica_id, execution.id)
 
     def create_instances_replica(self, ctxt, origin, destination, instances):
         replica = models.Replica()

coriolis/db/api.py

@@ -70,12 +70,13 @@ def get_replica_tasks_executions(context, replica_id, include_tasks=False):
 
 
 @enginefacade.reader
-def get_replica_tasks_execution(context, execution_id):
+def get_replica_tasks_execution(context, replica_id, execution_id):
     q = _soft_delete_aware_query(context, models.TasksExecution).join(
         models.Replica)
     q = _get_tasks_with_details_options(q)
     return q.filter(
         models.Replica.project_id == context.tenant,
+        models.Replica.id == replica_id,
         models.TasksExecution.id == execution_id).first()
 
 

coriolis/replica_tasks_executions/api.py

@@ -12,18 +12,18 @@ class API(object):
         return self._rpc_client.execute_replica_tasks(
             ctxt, replica_id, shutdown_instances)
 
-    def delete(self, ctxt, execution_id):
+    def delete(self, ctxt, replica_id, execution_id):
         self._rpc_client.delete_replica_tasks_execution(
-            ctxt, execution_id)
+            ctxt, replica_id, execution_id)
 
-    def cancel(self, ctxt, execution_id, force):
+    def cancel(self, ctxt, replica_id, execution_id, force):
         self._rpc_client.cancel_replica_tasks_execution(
-            ctxt, execution_id, force)
+            ctxt, replica_id, execution_id, force)
 
     def get_executions(self, ctxt, replica_id, include_tasks=False):
         return self._rpc_client.get_replica_tasks_executions(
             ctxt, replica_id, include_tasks)
 
-    def get_execution(self, ctxt, execution_id):
+    def get_execution(self, ctxt, replica_id, execution_id):
         return self._rpc_client.get_replica_tasks_execution(
-            ctxt, execution_id)
+            ctxt, replica_id, execution_id)