fix(k8s): check readiness probe for Dask service (#626)

Closes #625

reana_workflow_controller/k8s.py

@@ -1,5 +1,5 @@
 # This file is part of REANA.
-# Copyright (C) 2019, 2020, 2021, 2022, 2024 CERN.
+# Copyright (C) 2019, 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.
@@ -487,10 +487,32 @@ def delete_dask_dashboard_ingress(cluster_name, workflow_id):
     )
 
 
+def check_pod_readiness_by_prefix(
+    pod_name_prefix, namespace=REANA_RUNTIME_KUBERNETES_NAMESPACE
+):
+    """Check the readiness of a Pod in the given namespace whose name starts with the specified prefix. We assume that there exists 0 or 1 pod with a given prefix."""
+    try:
+        pods = current_k8s_corev1_api_client.list_namespaced_pod(namespace=namespace)
+
+        for pod in pods.items:
+            if pod.metadata.name.startswith(pod_name_prefix):
+                # Check the pod's readiness condition
+                if pod.status.conditions:
+                    for condition in pod.status.conditions:
+                        if condition.type == "Ready":
+                            return (
+                                "Ready" if condition.status == "True" else "Not Ready"
+                            )
+                return "Not Ready"
+        return "Not Found"
+    except ApiException as e:
+        return f"Error: {e.reason}"
+
+
 def check_pod_status_by_prefix(
     pod_name_prefix, namespace=REANA_RUNTIME_KUBERNETES_NAMESPACE
 ):
-    """Check if there is a Pod in the given namespace whose name starts with the specified prefix. We assume that there exists 0 or 1 pod with a given prefix."""
+    """Check the pod status of a Pod in the given namespace whose name starts with the specified prefix. We assume that there exists 0 or 1 pod with a given prefix."""
     try:
         pods = current_k8s_corev1_api_client.list_namespaced_pod(namespace=namespace)
 

reana_workflow_controller/rest/workflows.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2020, 2021, 2022, 2023 CERN.
+# Copyright (C) 2020, 2021, 2022, 2023, 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.
@@ -59,7 +59,10 @@
     use_paginate_args,
 )
 
-from reana_workflow_controller.k8s import check_pod_status_by_prefix
+from reana_workflow_controller.k8s import (
+    check_pod_status_by_prefix,
+    check_pod_readiness_by_prefix,
+)
 from reana_workflow_controller.dask import requires_dask
 
 START = "start"
@@ -412,11 +415,11 @@ def get_workflows(args, paginate=None):  # noqa
 
                 dask_service = workflow.services.first()
                 if dask_service and dask_service.status == ServiceStatus.created:
-                    pod_status = check_pod_status_by_prefix(
+                    pod_readiness = check_pod_readiness_by_prefix(
                         pod_name_prefix=f"reana-run-dask-{workflow.id_}"
                     )
 
-                    if pod_status == "Running":
+                    if pod_readiness == "Ready":
                         dask_service.status = ServiceStatus.running
                         db_session = Session.object_session(dask_service)
                         db_session.commit()