consumer: remove pods for deleted workflows

closes reanahub#437

Author: Vladyslav Moisieienkov

reana_workflow_controller/consumer.py

@@ -31,7 +31,7 @@
     build_unique_component_name,
 )
 from reana_db.database import Session
-from reana_db.models import Job, JobCache, Workflow, RunStatus
+from reana_db.models import Job, JobCache, Workflow, RunStatus, JobStatus
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm.attributes import flag_modified
 
@@ -118,11 +118,40 @@ def on_message(self, body, message):
                     )
             elif workflow and workflow.status not in ALIVE_STATUSES:
                 logging.warning(
-                    f"Event for not alive workflow {workflow.id_} with DB status {workflow.status} received:\n"
-                    f"{body}\nIgnoring..."
+                    f"Event for not alive workflow {workflow.id_} with DB status {workflow.status} received."
+                )
+                try:
+                    _delete_workflow_job(workflow)
+                    logging.info(
+                        f"Remove batch-pod for not alive {workflow.id_} workflow."
+                    )
+                except REANAWorkflowControllerError as exception:
+                    logging.error(
+                        f"Could not clean up not alive workflow {workflow.id_} batch pod for workflow."
+                        f" Error: {exception}"
+                    )
+                jobs = Session.query(Job).filter(
+                    Job.workflow_uuid == workflow_uuid,
                 )
+                for job in jobs:
+                    job_name = build_unique_component_name("run-job", job.id_)
+                    if job.status in [
+                        JobStatus.running,
+                        JobStatus.queued,
+                        JobStatus.created,
+                    ]:
+                        try:
+                            current_k8s_batchv1_api_client.delete_namespaced_job(
+                                name=job_name,
+                                namespace=REANA_RUNTIME_KUBERNETES_NAMESPACE,
+                                propagation_policy="Background",
+                            )
+                        except ApiException as e:
+                            logging.error(
+                                f"run-job pod {job_name} for {workflow.id_} could not be deleted. Error: {e}"
+                            )
             else:
-                logging.warning(
+                logging.error(
                     f"Event for workflow {workflow_uuid} that doesn't exist in DB received:\n"
                     f"{body}\nIgnoring..."
                 )