WIP recover from interrupted workflows

FNNDSC · Feb 1, 2024 · de6b5f2 · de6b5f2
1 parent b6e77ce
commit de6b5f2
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 6 deletions.
diff --git a/app/controllers/subprocesses/status.py b/app/controllers/subprocesses/status.py
@@ -94,7 +94,7 @@ def __progress_percent(curr_state: int, total_states: int, state_progress: int)
 
 
 def is_status_subprocess_running(workflow: WorkflowDBSchema):
-    proc_count = get_process_count("status", args.data)
+    proc_count = get_process_count("app/controllers/subprocesses/status.py", args.data)
 
     if not workflow.stale:
         return True

diff --git a/app/controllers/subprocesses/wf_manager.py b/app/controllers/subprocesses/wf_manager.py
@@ -52,7 +52,7 @@ def manage_workflow(db_key: str, test: bool):
 
     workflow = analysis_retry(workflow)
 
-    if workflow.started or not workflow.response.status or test:
+    if not workflow.response.status or test:
         # Do nothing and return
         reason = f"Workflow request failed. {workflow.response.error}." if not workflow.response.status \
                  else f"Workflow already started."
@@ -118,6 +118,9 @@ def manage_workflow(db_key: str, test: bool):
                         workflow.response.error = Error.analysis.value + str(ex)
                         workflow.response.status = False
                         update_workflow(key, workflow)
+            case State.COMPLETED:
+                logger.info(f"Request is now complete. Exiting while loop. ",extra=d)
+                return
 
         logger.info(f"Calling status update subprocess.", extra=d)
         update_status(request)
@@ -166,7 +169,7 @@ def update_status(request: WorkflowRequestSchema):
     """
     d_data = request_to_dict(request)
     str_data = json.dumps(d_data)
-    proc_count = get_process_count("status", str_data)
+    proc_count = get_process_count("app/controllers/subprocesses/status.py", str_data)
     logger.debug(f"{proc_count} subprocess of status manager running on the system.", extra=d)
     if proc_count > 0:
         logger.info(f"No new status subprocess started.", extra=d)
@@ -283,7 +286,7 @@ def do_cube_create_feed(request: WorkflowRequestSchema, cube_url: str, retries:
     data_path = client.getSwiftPath(pacs_details)
     logger.info(f"Received data path: {data_path}", extra=d)
     if retries < 5:
-        feed_name = feed_name + f"retry#{retries}"
+        feed_name = feed_name + f"-retry#{retries}"
 
     # Get plugin Id
     plugin_search_params = {"name": "pl-dircopy"}

diff --git a/app/controllers/workflow.py b/app/controllers/workflow.py
@@ -177,7 +177,7 @@ def manage_workflow(str_data: str, mode: str):
     """
     Manage a workflow request in a separate subprocess
     """
-    proc_count = get_process_count("wf_manager", str_data)
+    proc_count = get_process_count("app/controllers/subprocesses/wf_manager.py", str_data)
     logger.debug(f"{proc_count} subprocess of workflow manager running on the system.", extra=d)
     if proc_count > 0:
         logger.info(f"No new manager subprocess started.", extra=d)
@@ -196,7 +196,7 @@ def update_workflow_status(str_data: str, mode: str):
     """
     Update the current status of a workflow request in a separate process
     """
-    proc_count = get_process_count("status", str_data)
+    proc_count = get_process_count("app/controllers/subprocesses/status.py", str_data)
     logger.debug(f"{proc_count} subprocess of status manager running on the system.", extra=d)
     if proc_count > 0:
         logger.info(f"No new status subprocess started.", extra=d)