Allow filtering results by status

We're adding a "status" search option that can be specified when
listing the following resources:

* deployments
    * status -> last_execution_status db field
* transfers
* transfer executions

The filtering will be performed on the db side.

Author: petrutlucian94

coriolis/api/v1/deployments.py

@@ -8,6 +8,7 @@
 from coriolis.api.v1 import utils as api_utils
 from coriolis.api.v1.views import deployment_view
 from coriolis.api import wsgi as api_wsgi
+from coriolis import constants
 from coriolis.deployments import api
 from coriolis.endpoints import api as endpoints_api
 from coriolis import exception
@@ -36,6 +37,18 @@ def show(self, req, id):
 
         return deployment_view.single(deployment)
 
+    def _get_filters(self, req) -> dict:
+        filters = {}
+        # For simplicity and consistency, we'll use "status" to search for a
+        # given "last_execution_status".
+        status = req.GET.get("status")
+        if status is not None:
+            if status not in constants.ALL_EXECUTION_STATUSES:
+                raise exc.HTTPBadRequest(
+                    explanation=f"Unknown deployment status: {status}")
+            filters["status"] = status
+        return filters
+
     def _list(self, req):
         show_deleted = api_utils.get_bool_url_arg(
             req, "show_deleted", default=False)
@@ -47,6 +60,7 @@ def _list(self, req):
 
         marker, limit = common.get_paging_params(req)
         sort_keys, sort_dirs = common.get_sort_params(req)
+        filters = self._get_filters(req)
 
         return deployment_view.collection(
             self._deployment_api.get_deployments(
@@ -55,6 +69,7 @@ def _list(self, req):
                 include_task_info=include_task_info,
                 marker=marker, limit=limit,
                 sort_keys=sort_keys, sort_dirs=sort_dirs,
+                filters=filters,
             ))
 
     def index(self, req):

coriolis/api/v1/transfer_tasks_executions.py

@@ -4,6 +4,7 @@
 from coriolis.api import common
 from coriolis.api.v1.views import transfer_tasks_execution_view
 from coriolis.api import wsgi as api_wsgi
+from coriolis import constants
 from coriolis import exception
 from coriolis.policies import transfer_tasks_executions as executions_policies
 from coriolis.transfer_tasks_executions import api
@@ -27,19 +28,31 @@ def show(self, req, transfer_id, id):
 
         return transfer_tasks_execution_view.single(execution)
 
+    def _get_filters(self, req) -> dict:
+        filters = {}
+        status = req.GET.get("status")
+        if status is not None:
+            if status not in constants.ALL_EXECUTION_STATUSES:
+                raise exc.HTTPBadRequest(
+                    explanation=f"Unknown task execution status: {status}")
+            filters["status"] = status
+        return filters
+
     def index(self, req, transfer_id):
         context = req.environ["coriolis.context"]
         context.can(
             executions_policies.get_transfer_executions_policy_label("list"))
 
         marker, limit = common.get_paging_params(req)
         sort_keys, sort_dirs = common.get_sort_params(req)
+        filters = self._get_filters(req)
 
         return transfer_tasks_execution_view.collection(
             self._transfer_tasks_execution_api.get_executions(
                 context, transfer_id, include_tasks=False,
                 marker=marker, limit=limit,
-                sort_keys=sort_keys, sort_dirs=sort_dirs))
+                sort_keys=sort_keys, sort_dirs=sort_dirs,
+                filters=filters))
 
     def detail(self, req, transfer_id):
         context = req.environ["coriolis.context"]

coriolis/api/v1/transfers.py

@@ -42,6 +42,16 @@ def show(self, req, id):
 
         return transfer_view.single(transfer)
 
+    def _get_filters(self, req) -> dict:
+        filters = {}
+        status = req.GET.get("status")
+        if status is not None:
+            if status not in constants.ALL_TASK_STATUSES:
+                raise exc.HTTPBadRequest(
+                    explanation=f"Unknown task status: {status}")
+            filters["status"] = status
+        return filters
+
     def _list(self, req):
         show_deleted = api_utils.get_bool_url_arg(
             req, "show_deleted", default=False)
@@ -52,13 +62,15 @@ def _list(self, req):
             req, "include_task_info", default=False)
         marker, limit = common.get_paging_params(req)
         sort_keys, sort_dirs = common.get_sort_params(req)
+        filters = self._get_filters(req)
         return transfer_view.collection(
             self._transfer_api.get_transfers(
                 context,
                 include_tasks_executions=include_task_info,
                 include_task_info=include_task_info,
                 marker=marker, limit=limit,
                 sort_keys=sort_keys, sort_dirs=sort_dirs,
+                filters=filters,
             ))
 
     def index(self, req):

coriolis/conductor/rpc/client.py

@@ -147,7 +147,8 @@ def get_transfer_tasks_executions(self, ctxt, transfer_id,
                                       marker=None,
                                       limit=None,
                                       sort_keys=None,
-                                      sort_dirs=None):
+                                      sort_dirs=None,
+                                      filters=None):
         return self._call(
             ctxt, 'get_transfer_tasks_executions',
             transfer_id=transfer_id,
@@ -156,6 +157,7 @@ def get_transfer_tasks_executions(self, ctxt, transfer_id,
             limit=limit,
             sort_keys=sort_keys,
             sort_dirs=sort_dirs,
+            filters=filters,
         )
 
     def get_transfer_tasks_execution(self, ctxt, transfer_id, execution_id,
@@ -208,7 +210,8 @@ def create_instances_transfer(self, ctxt,
     def get_transfers(self, ctxt, include_tasks_executions=False,
                       include_task_info=False,
                       marker=None, limit=None,
-                      sort_keys=None, sort_dirs=None):
+                      sort_keys=None, sort_dirs=None,
+                      filters=None):
         return self._call(
             ctxt, 'get_transfers',
             include_tasks_executions=include_tasks_executions,
@@ -217,6 +220,7 @@ def get_transfers(self, ctxt, include_tasks_executions=False,
             limit=limit,
             sort_keys=sort_keys,
             sort_dirs=sort_dirs,
+            filters=filters,
         )
 
     def get_transfer(self, ctxt, transfer_id, include_task_info=False):
@@ -235,14 +239,16 @@ def delete_transfer_disks(self, ctxt, transfer_id):
     def get_deployments(self, ctxt, include_tasks=False,
                         include_task_info=False,
                         marker=None, limit=None,
-                        sort_keys=None, sort_dirs=None):
+                        sort_keys=None, sort_dirs=None,
+                        filters=None):
         return self._call(
             ctxt, 'get_deployments', include_tasks=include_tasks,
             include_task_info=include_task_info,
             marker=marker,
             limit=limit,
             sort_keys=sort_keys,
             sort_dirs=sort_dirs,
+            filters=filters,
         )
 
     def get_deployment(self, ctxt, deployment_id, include_task_info=False):

coriolis/conductor/rpc/server.py

@@ -1156,14 +1156,16 @@ def get_transfer_tasks_executions(self, ctxt, transfer_id,
                                       marker=None,
                                       limit=None,
                                       sort_keys=None,
-                                      sort_dirs=None):
+                                      sort_dirs=None,
+                                      filters=None):
         return db_api.get_transfer_tasks_executions(
             ctxt, transfer_id, include_tasks,
             include_task_info=include_task_info,
             marker=marker,
             limit=limit,
             sort_keys=sort_keys,
             sort_dirs=sort_dirs,
+            filters=filters,
             to_dict=True)
 
     @tasks_execution_synchronized
@@ -1217,14 +1219,16 @@ def _get_transfer_tasks_execution(ctxt, transfer_id, execution_id,
     def get_transfers(ctxt, include_tasks_executions=False,
                       include_task_info=False,
                       marker=None, limit=None,
-                      sort_keys=None, sort_dirs=None):
+                      sort_keys=None, sort_dirs=None,
+                      filters=None):
         return db_api.get_transfers(
             ctxt, include_tasks_executions=include_tasks_executions,
             include_task_info=include_task_info,
             marker=marker,
             limit=limit,
             sort_keys=sort_keys,
             sort_dirs=sort_dirs,
+            filters=filters,
             to_dict=True)
 
     @transfer_synchronized
@@ -1383,14 +1387,16 @@ def _get_transfer(self, ctxt, transfer_id, include_task_info=False,
     @staticmethod
     def get_deployments(ctxt, include_tasks, include_task_info=False,
                         marker=None, limit=None,
-                        sort_keys=None, sort_dirs=None):
+                        sort_keys=None, sort_dirs=None,
+                        filters=None):
         return db_api.get_deployments(
             ctxt, include_tasks,
             include_task_info=include_task_info,
             marker=marker,
             limit=limit,
             sort_keys=sort_keys,
             sort_dirs=sort_dirs,
+            filters=filters,
             to_dict=True)
 
     @deployment_synchronized

coriolis/constants.py

@@ -33,6 +33,11 @@
     EXECUTION_STATUS_ERROR_ALLOCATING_MINIONS
 ]
 
+ALL_EXECUTION_STATUSES = (
+    ACTIVE_EXECUTION_STATUSES +
+    FINALIZED_EXECUTION_STATUSES
+)
+
 TASK_STATUS_SCHEDULED = "SCHEDULED"
 TASK_STATUS_PENDING = "PENDING"
 TASK_STATUS_STARTING = "STARTING"
@@ -83,6 +88,12 @@
     TASK_STATUS_FAILED_TO_CANCEL
 ]
 
+ALL_TASK_STATUSES = (
+    ACTIVE_TASK_STATUSES +
+    CANCELED_TASK_STATUSES +
+    FINALIZED_TASK_STATUSES
+)
+
 TASK_TYPE_FINALIZE_INSTANCE_DEPLOYMENT = "FINALIZE_INSTANCE_DEPLOYMENT"
 TASK_TYPE_CLEANUP_FAILED_INSTANCE_DEPLOYMENT = (
     "CLEANUP_FAILED_INSTANCE_DEPLOYMENT")

coriolis/db/api.py

@@ -1,6 +1,7 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+import copy
 import uuid
 
 from oslo_config import cfg
@@ -281,6 +282,7 @@ def get_transfer_tasks_executions(context, transfer_id, include_tasks=False,
                                   limit=None,
                                   sort_keys: list[str] | None = None,
                                   sort_dirs: list[str] | None = None,
+                                  filters: dict | None = None,
                                   to_dict=False):
     q = _soft_delete_aware_query(context, models.TasksExecution)
     q = q.join(models.Transfer)
@@ -293,6 +295,13 @@ def get_transfer_tasks_executions(context, transfer_id, include_tasks=False,
 
     q = q.filter(models.Transfer.id == transfer_id)
 
+    filters = copy.deepcopy(filters or {})
+    if "status" in filters:
+        status = filters.pop("status")
+        q = q.filter(models.TasksExecution.status == status)
+    if filters:
+        raise ValueError("Unsupported filters: %s" % filters)
+
     sort_keys, sort_dirs = process_sort_params(
         sort_keys,
         sort_dirs,
@@ -458,6 +467,7 @@ def get_transfers(context,
                   limit=None,
                   sort_keys: list[str] | None = None,
                   sort_dirs: list[str] | None = None,
+                  filters: dict | None = None,
                   to_dict=False):
     q = _soft_delete_aware_query(context, models.Transfer)
     if include_tasks_executions:
@@ -471,6 +481,13 @@ def get_transfers(context,
         q = q.filter(
             models.Transfer.project_id == context.project_id)
 
+    filters = copy.deepcopy(filters or {})
+    if "status" in filters:
+        status = filters.pop("status")
+        q = q.filter(models.Transfer.last_execution_status == status)
+    if filters:
+        raise ValueError("Unsupported filters: %s" % filters)
+
     sort_keys, sort_dirs = process_sort_params(
         sort_keys,
         sort_dirs,
@@ -588,6 +605,7 @@ def get_deployments(context,
                     limit=None,
                     sort_keys: list[str] | None = None,
                     sort_dirs: list[str] | None = None,
+                    filters: dict | None = None,
                     to_dict=False):
     q = _soft_delete_aware_query(context, models.Deployment)
     if include_tasks:
@@ -600,6 +618,13 @@ def get_deployments(context,
     if is_user_context(context):
         q = q.filter_by(project_id=context.project_id)
 
+    filters = copy.deepcopy(filters or {})
+    if "status" in filters:
+        status = filters.pop("status")
+        q = q.filter(models.Deployment.last_execution_status == status)
+    if filters:
+        raise ValueError("Unsupported filters: %s" % filters)
+
     sort_keys, sort_dirs = process_sort_params(
         sort_keys,
         sort_dirs,

coriolis/deployer_manager/rpc/server.py

@@ -123,10 +123,10 @@ def _loop(self):
             try:
                 deployments = self._rpc_conductor_client.get_deployments(
                     self._admin_ctx, include_tasks=False,
-                    include_task_info=False)
-                for d in deployments:
-                    if d['last_execution_status'] == PENDING_STATUS:
-                        self._check_deployer_status(d['id'])
+                    include_task_info=False,
+                    filters={'status': PENDING_STATUS})
+                for pending_deployment in deployments:
+                    self._check_deployer_status(pending_deployment['id'])
             except Exception:
                 LOG.warning(
                     f"Deployer manager failed to list pending deployments. "

coriolis/deployments/api.py

@@ -28,11 +28,14 @@ def cancel(self, ctxt, deployment_id, force):
     def get_deployments(self, ctxt, include_tasks=False,
                         include_task_info=False,
                         marker=None, limit=None,
-                        sort_keys=None, sort_dirs=None):
+                        sort_keys=None, sort_dirs=None,
+                        filters=None):
         return self._rpc_client.get_deployments(
             ctxt, include_tasks, include_task_info=include_task_info,
             marker=marker, limit=limit,
-            sort_keys=sort_keys, sort_dirs=sort_dirs)
+            sort_keys=sort_keys, sort_dirs=sort_dirs,
+            filters=filters,
+        )
 
     def get_deployment(self, ctxt, deployment_id, include_task_info=False):
         return self._rpc_client.get_deployment(

coriolis/tests/api/v1/test_transfer_tasks_executions.py

@@ -95,6 +95,9 @@ def test_index(
             mock.sentinel.marker,
             mock.sentinel.limit,
         )
+        mock_req.GET = {
+            "status": "RUNNING",
+        }
 
         result = self.transfer_api.index(mock_req, transfer_id)
 
@@ -111,6 +114,7 @@ def test_index(
             limit=mock.sentinel.limit,
             sort_keys=mock.sentinel.sort_keys,
             sort_dirs=mock.sentinel.sort_dirs,
+            filters={"status": "RUNNING"},
         )
         mock_collection.assert_called_once_with(
             mock_get_executions.return_value)