diff --git a/beeflow/client/bee_client.py b/beeflow/client/bee_client.py index 431d243b7..9e8b6e4ab 100644 --- a/beeflow/client/bee_client.py +++ b/beeflow/client/bee_client.py @@ -38,8 +38,8 @@ from beeflow.common.object_models import generate_workflow_id from beeflow.client import remote_client from beeflow.wf_manager.models import ( - CopyWorkflowRequest, - CopyWorkflowResponse, + ResubmitWorkflowRequest, + ResubmitWorkflowResponse, ListWorkflowsResponse, SubmitWorkflowRequest, ModifyWorkflowRequest, @@ -456,15 +456,15 @@ def submit( # pylint:disable=R0915 ..., help="the workflow name" ), wf_path: pathlib.Path = typer.Argument( - ..., help="path to the workflow .tgz or dir" + None, help="path to the workflow .tgz or dir" ), main_cwl: str = typer.Argument( - ..., + None, help="filename of main CWL (if using CWL tarball), " + "path of main CWL (if using CWL directory)", ), yaml_file: str = typer.Argument( - ..., + None, help="filename of yaml file (if using CWL tarball), " + "path of yaml file (if using CWL directory)", ), @@ -476,7 +476,7 @@ def submit( # pylint:disable=R0915 False, "--no-start", "-n", help="do not start the workflow" ), ): - """Submit a new workflow.""" + """Submit a new workflow or resubmit a failed workflow.""" def is_parent(parent, path): """Return true if the path is a child of the other path.""" @@ -860,23 +860,19 @@ def cancel( @app.command() -def copy(wf_id: str = typer.Argument(..., callback=match_short_id)): - """Copy an archived workflow.""" +def resubmit(wf_id: str = typer.Argument(..., callback=match_short_id)): + """Resubmit a failed archived workflow.""" long_wf_id = wf_id try: conn = _wfm_conn() resp = conn.patch( - _url(), json=CopyWorkflowRequest(wf_id=long_wf_id).model_dump(), timeout=60 + _url(), json=ResubmitWorkflowRequest(wf_id=long_wf_id).model_dump(), timeout=60 ) except requests.exceptions.ConnectionError: error_exit("Could not reach WF Manager.") if resp.status_code != requests.codes.okay: # pylint: disable=no-member - error_exit("WF Manager could not copy workflow.") - archive_info = CopyWorkflowResponse.model_validate(resp.json()) - archive_file = jsonpickle.decode(archive_info.archive_file_pickle) - archive_filename = archive_info.archive_filename - logging.info(f"Copy workflow: {resp.text}") - return archive_file, archive_filename + error_exit("WF Manager could not resubmit workflow.") + logging.info(f"Resubmit workflow: {resp.text}") @app.command() diff --git a/beeflow/common/db/gdb_db.py b/beeflow/common/db/gdb_db.py index 9f0de5428..1d4fd530b 100644 --- a/beeflow/common/db/gdb_db.py +++ b/beeflow/common/db/gdb_db.py @@ -261,6 +261,16 @@ def set_task_state(self, task_id: str, state: str): WHERE id = :task_id;""" bdb.run(self.db_file, set_task_state_query, {'task_id': task_id, 'state': state}) + def reset_failed_tasks(self, workflow_id: str): + """Reset failed tasks.""" + placeholders = ", ".join("?" for _ in failed_task_states) + query = f""" + UPDATE task + SET state = 'WAITING' + WHERE workflow_id = :workflow_id + AND state IN ({placeholders}); + """ + bdb.run(self.db_file, query, [workflow_id, *failed_task_states]) def add_dependencies(self, task: Task, old_task: Task=None, restarted_task=False): """Add dependencies for a task based on its inputs and outputs.""" diff --git a/beeflow/common/gdb/gdb_driver.py b/beeflow/common/gdb/gdb_driver.py index a8e813369..bd1e1d985 100644 --- a/beeflow/common/gdb/gdb_driver.py +++ b/beeflow/common/gdb/gdb_driver.py @@ -83,6 +83,17 @@ def restart_task(self, old_task, new_task): :type new_task: Task """ + @abstractmethod + def reset_failed_tasks(self, workflow_id): + """Restart a failed task. + + Create a Task node for new_task with state 'RESTARTED' and an edge + to indicate that it is the child of the Task node of old_task. + + :rtype: Workflow + """ + + @abstractmethod def finalize_task(self, task): """Set task state to 'COMPLETED' and set inputs from source. diff --git a/beeflow/common/gdb/sqlite3_driver.py b/beeflow/common/gdb/sqlite3_driver.py index e9cce21dc..40790567f 100644 --- a/beeflow/common/gdb/sqlite3_driver.py +++ b/beeflow/common/gdb/sqlite3_driver.py @@ -103,6 +103,12 @@ def restart_task(self, old_task, new_task): self.db.set_task_state(new_task.id, 'WAITING') self.db.add_dependencies(new_task, old_task=old_task, restarted_task=True) + def reset_failed_tasks(self, workflow_id): + """Set failed tasks to 'WAITING'. + + Used after resubmiting a wokrflow. + """ + self.db.reset_failed_tasks(workflow_id) def finalize_task(self, task): """Set task state to 'COMPLETED' and set inputs from source. @@ -139,6 +145,12 @@ def get_workflow_description(self, workflow_id): """ return self.db.get_workflow(workflow_id) + def get_workflow_workdir(self, workflow_id): + """Return the workdir for the specified workflow. + + :rtype: str + """ + def get_workflow_state(self, workflow_id): """Return the current state of the workflow. @@ -281,7 +293,7 @@ def set_task_input(self, task_id, input_id, value): """Set the value of a task input. :param task_id: the ID of the task whose input to set - :type task_id: str + :type task_id: stsr :param input_id: the ID of the input :type input_id: str :param value: str or int or float diff --git a/beeflow/common/wf_interface.py b/beeflow/common/wf_interface.py index eba91a38e..6f6fb641e 100644 --- a/beeflow/common/wf_interface.py +++ b/beeflow/common/wf_interface.py @@ -73,6 +73,12 @@ def reset_workflow(self, workflow_id): self._workflow_id = workflow_id self._gdb_driver.set_workflow_state(self._workflow_id, 'SUBMITTED') + def reset_failed_workflow(self, wf_id): + """Reset the execution state and ID of a BEE workflow.""" + self._gdb_driver.reset_failed_tasks(self._workflow_id) + #self._workflow_id = workflow_id + self._gdb_driver.set_workflow_state(self._workflow_id, 'RESTARTED') + def add_task(self, task): """Add a new task to a BEE workflow. diff --git a/beeflow/wf_manager/models.py b/beeflow/wf_manager/models.py index 047a7199c..f212cf98f 100644 --- a/beeflow/wf_manager/models.py +++ b/beeflow/wf_manager/models.py @@ -30,14 +30,14 @@ class SubmitWorkflowResponse(BaseModel): status: str wf_id: Optional[str] = None -class CopyWorkflowRequest(BaseModel): +class ResubmitWorkflowRequest(BaseModel): """Request model for copying a workflow.""" wf_id: str -class CopyWorkflowResponse(BaseModel): +class ResubmitWorkflowResponse(BaseModel): """Response model for workflow copy.""" - archive_file_pickle: str - archive_filename: str + msg: str + status: str class TaskStateUpdate(BaseModel): """Information about a task state update.""" diff --git a/beeflow/wf_manager/resources/wf_list.py b/beeflow/wf_manager/resources/wf_list.py index 54912907c..d6ac60cd4 100644 --- a/beeflow/wf_manager/resources/wf_list.py +++ b/beeflow/wf_manager/resources/wf_list.py @@ -1,6 +1,6 @@ """The workflow list module. -This contains endpoints forsubmitting, starting, and reexecuting workflows. +This module contains endpoints for submitting, starting, and reexecuting workflows. """ import base64 @@ -20,11 +20,11 @@ # from beeflow.common.wf_profiler import WorkflowProfiler from beeflow.wf_manager.models import ( - CopyWorkflowRequest, - CopyWorkflowResponse, ListWorkflowsResponse, SubmitWorkflowRequest, SubmitWorkflowResponse, + ResubmitWorkflowRequest, + ResubmitWorkflowResponse, ) from beeflow.wf_manager.resources import wf_utils @@ -82,7 +82,7 @@ def get(self): return ListWorkflowsResponse(workflow_info_list=info).model_dump(), 200 def post(self): - """Upload a workflown and start.""" + """Upload a workflow and start.""" try: data = SubmitWorkflowRequest.model_validate(request.json) except ValidationError as e: @@ -115,16 +115,16 @@ def post(self): ) def patch(self): - """Copy workflow archive.""" - wf_id = CopyWorkflowRequest.model_validate(request.json).wf_id - archive_dir = bc.get("DEFAULT", "bee_archive_dir") - archive_path = os.path.join(archive_dir, wf_id + ".tgz") - with open(archive_path, "rb") as archive: - archive_file = jsonpickle.encode(archive.read()) - archive_filename = os.path.basename(archive_path) - return ( - CopyWorkflowResponse( - archive_file_pickle=archive_file, archive_filename=archive_filename - ).model_dump(), - 200, - ) + """Resubmit failed workflow.""" + try: + data = ResubmitWorkflowRequest.model_validate(request.json) + except ValidationError as e: + log.error(f"Error parsing request data: {e}") + return ( + ResubmitWorkflowResponse( + msg="Invalid request data", status="error", wf_id=None + ).model_dump(), + 400, + ) + wf_id = data.wf_id + wf_utils.restart_workflow(wf_id) diff --git a/beeflow/wf_manager/resources/wf_utils.py b/beeflow/wf_manager/resources/wf_utils.py index 2db37a535..d98827c05 100644 --- a/beeflow/wf_manager/resources/wf_utils.py +++ b/beeflow/wf_manager/resources/wf_utils.py @@ -222,6 +222,13 @@ def setup_workflow(wf_id, wf_name, wf_dir, wf_workdir, no_start, workflow=None, log.info("Starting workflow") start_workflow.delay(wf_id) +def restart_workflow(wf_id): + """Restart a failed workflow.""" + wfi = get_workflow_interface(wf_id) + wfi.reset_failed_workflow(wf_id) + update_wf_status(wf_id, "Starting") + log.info("Reset failed worflow tasks.") + def export_dag(wf_id, output_dir, graphmls_dir, no_dag_dir, workflow_dir=None): """Export the DAG of the workflow."""