feat(dask): update service status upon cluster deletion (reanahub#648)

Mark the Dask service as deleted instead of removing it when the Dask
cluster terminates.

Author: Alputer

reana_workflow_controller/consumer.py

@@ -29,7 +29,6 @@
     calculate_hash_of_dir,
     calculate_job_input_hash,
     build_unique_component_name,
-    get_dask_component_name,
 )
 from reana_db.database import Session
 from reana_db.models import Job, JobCache, Workflow, RunStatus, Service

reana_workflow_controller/dask.py

@@ -12,10 +12,9 @@
 from flask import current_app
 
 from kubernetes import client
-from kubernetes.client.exceptions import ApiException
 
 from reana_db.database import Session
-from reana_db.models import Service
+from reana_db.models import Service, ServiceStatus
 from reana_db.utils import _get_workflow_with_uuid_or_name
 from reana_commons.config import (
     K8S_CERN_EOS_AVAILABLE,
@@ -677,16 +676,16 @@ def delete_dask_cluster(workflow_id, user_id) -> None:
         )
         if dask_service:
             workflow = _get_workflow_with_uuid_or_name(str(workflow_id), user_id)
-            workflow.services.remove(dask_service)
-            Session.delete(dask_service)
+            dask_service.status = ServiceStatus.deleted
             Session.object_session(workflow).commit()
+
             logging.info(
-                f"Dask service model for workflow {workflow_id} deleted successfully from database."
+                f"Dask service model for workflow {workflow_id} status updated to 'deleted' in database."
             )
 
     except Exception as e:
         errors.append(
-            f"Error deleting Dask Service model from database of the workflow: {workflow_id}: {e}"
+            f"Error updating Dask Service model status in database for workflow {workflow_id}: {e}"
         )
 
     # Raise collected errors if any

reana_workflow_controller/rest/utils.py

@@ -249,6 +249,13 @@ def build_workflow_logs(workflow, steps=None, paginate=None):
     return all_logs
 
 
+def build_service_logs(workflow):
+    """Return the logs for all services of a workflow."""
+    # Currently we have only Dask cluster service, need a more generic implementation if we introduce additional services.
+
+    return json.loads(workflow.services[0].logs)
+
+
 def remove_workflow_jobs_from_cache(workflow):
     """Remove any cached jobs from given workflow.
 

reana_workflow_controller/rest/workflows_status.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2020, 2021, 2022, 2024 CERN.
+# Copyright (C) 2020, 2021, 2022, 2024, 2025 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -27,6 +27,7 @@
 )
 from reana_workflow_controller.rest.utils import (
     build_workflow_logs,
+    build_service_logs,
     delete_workflow,
     get_workflow_name,
     get_workflow_progress,
@@ -150,6 +151,7 @@ def get_workflow_logs(workflow_id_or_name, paginate=None, **kwargs):  # noqa
         steps = None
         if request.is_json:
             steps = request.json
+
         if steps:
             workflow_logs = {
                 "workflow_logs": None,
@@ -172,20 +174,27 @@ def get_workflow_logs(workflow_id_or_name, paginate=None, **kwargs):  # noqa
             workflow_logs = {
                 "workflow_logs": logs or workflow.logs,
                 "job_logs": build_workflow_logs(workflow, paginate=paginate),
+                "service_logs": {},
                 "engine_specific": workflow.engine_specific,
             }
-        return (
-            jsonify(
-                {
-                    "workflow_id": workflow.id_,
-                    "workflow_name": get_workflow_name(workflow),
-                    "logs": json.dumps(workflow_logs),
-                    "user": user_uuid,
-                    "live_logs_enabled": REANA_OPENSEARCH_ENABLED,
-                }
-            ),
-            200,
-        )
+
+            # Get all services logs
+            workflow_logs["service_logs"] = {
+                s.name: [log.log for log in s.logs] for s in workflow.services
+            }
+
+            return (
+                jsonify(
+                    {
+                        "workflow_id": workflow.id_,
+                        "workflow_name": get_workflow_name(workflow),
+                        "logs": json.dumps(workflow_logs),
+                        "user": user_uuid,
+                        "live_logs_enabled": REANA_OPENSEARCH_ENABLED,
+                    }
+                ),
+                200,
+            )
 
     except ValueError:
         return (