submit_workflow

Runner backend subsystem root V2

This module is the single entry point to the runner backend subsystem V2. Other subsystems should only import this module and not its submodules or the individual backends.

submit_workflow(*, workflow_id, dataset_id, job_id, user_id, user_cache_dir, resource, profile, worker_init=None, fractal_ssh=None)

Prepares a workflow and applies it to a dataset

This function wraps the process_workflow one, which is different for each backend (e.g. local or slurm backend).

PARAMETER	DESCRIPTION
workflow_id	ID of the workflow being applied TYPE: int
dataset_id	Dataset ID TYPE: int
job_id	Id of the job record which stores the state for the current workflow application. TYPE: int
user_id	User ID. TYPE: int
worker_init	Custom executor parameters that get parsed before the execution of each task. TYPE: str | None DEFAULT: None
user_cache_dir	Cache directory (namely a path where the user can write). For slurm_sudo backend, this is both a base directory for job.working_dir_user. For slurm_sudo and slurm_ssh backends, this is used for user_local_exports. TYPE: str
resource	Computational resource to be used for this job (e.g. a SLURM cluster). TYPE: Resource
profile	Computational profile to be used for this job. TYPE: Profile
fractal_ssh	SSH object, for when resource.type = "slurm_ssh". TYPE: FractalSSH | None DEFAULT: None

Source code in fractal_server/runner/v2/submit_workflow.py

def submit_workflow(
    *,
    workflow_id: int,
    dataset_id: int,
    job_id: int,
    user_id: int,
    user_cache_dir: str,
    resource: Resource,
    profile: Profile,
    worker_init: str | None = None,
    fractal_ssh: FractalSSH | None = None,
) -> None:
    """
    Prepares a workflow and applies it to a dataset

    This function wraps the process_workflow one, which is different for each
    backend (e.g. local or slurm backend).

    Args:
        workflow_id:
            ID of the workflow being applied
        dataset_id:
            Dataset ID
        job_id:
            Id of the job record which stores the state for the current
            workflow application.
        user_id:
            User ID.
        worker_init:
            Custom executor parameters that get parsed before the execution of
            each task.
        user_cache_dir:
            Cache directory (namely a path where the user can write). For
            `slurm_sudo` backend, this is both a base directory for
            `job.working_dir_user`. For `slurm_sudo` and `slurm_ssh` backends,
            this is used for `user_local_exports`.
        resource:
            Computational resource to be used for this job (e.g. a SLURM
            cluster).
        profile:
            Computational profile to be used for this job.
        fractal_ssh: SSH object, for when `resource.type = "slurm_ssh"`.
    """
    # Declare runner backend and set `process_workflow` function
    logger_name = f"WF{workflow_id}_job{job_id}"
    logger = set_logger(logger_name=logger_name)

    with next(DB.get_sync_db()) as db_sync:
        try:
            job = db_sync.get_one(JobV2, job_id)
        except NoResultFound:
            logger.error(f"JobV2 {job_id} not found.")
            reset_logger_handlers(logger)
            return

        try:
            dataset = db_sync.get_one(DatasetV2, dataset_id)
            workflow = db_sync.get_one(WorkflowV2, workflow_id)
        except NoResultFound:
            log_msg = (
                f"Cannot fetch dataset {dataset_id} "
                f"and/or workflow {workflow_id} "
                f"(as part of job {job_id})."
            )
            logger.error(log_msg)
            fail_job(
                db=db_sync,
                job=job,
                log_msg=log_msg,
                logger_name=logger_name,
                emit_log=False,
            )
            return

        try:
            # Define local/remote folders, and create local folder
            local_job_dir = Path(job.working_dir)
            remote_job_dir = Path(job.working_dir_user)
            match resource.type:
                case ResourceType.LOCAL:
                    local_job_dir.mkdir(parents=True, exist_ok=False)
                case ResourceType.SLURM_SUDO:
                    original_umask = os.umask(0)
                    local_job_dir.mkdir(
                        parents=True, mode=0o755, exist_ok=False
                    )
                    os.umask(original_umask)
                case ResourceType.SLURM_SSH:
                    local_job_dir.mkdir(parents=True, exist_ok=False)

        except Exception as e:
            error_type = type(e).__name__
            fail_job(
                db=db_sync,
                job=job,
                log_msg=(
                    f"{error_type} error while creating local job folder."
                    f" Original error: {str(e)}"
                ),
                logger_name=logger_name,
                emit_log=True,
            )
            return

        # After Session.commit() is called, either explicitly or when using a
        # context manager, all objects associated with the Session are expired.
        # https://docs.sqlalchemy.org/en/14/orm/
        #   session_basics.html#opening-and-closing-a-session
        # https://docs.sqlalchemy.org/en/14/orm/
        #   session_state_management.html#refreshing-expiring

        # See issue #928:
        # https://github.com/fractal-analytics-platform/
        #   fractal-server/issues/928

        db_sync.refresh(dataset)
        db_sync.refresh(workflow)
        for wftask in workflow.task_list:
            db_sync.refresh(wftask)

        # Write logs
        log_file_path = local_job_dir / WORKFLOW_LOG_FILENAME
        logger = set_logger(
            logger_name=logger_name,
            log_file_path=log_file_path,
        )
        logger.info(
            f'Start execution of workflow "{workflow.name}"; '
            f"more logs at {str(log_file_path)}"
        )
        logger.debug(f"Resource name: {resource.name}")
        logger.debug(f"Profile name: {profile.name}")
        logger.debug(f"Username: {profile.username}")
        if resource.type in [ResourceType.SLURM_SUDO, ResourceType.SLURM_SSH]:
            logger.debug(f"slurm_account: {job.slurm_account}")
            logger.debug(f"worker_init: {worker_init}")
        logger.debug(f"job.id: {job.id}")
        logger.debug(f"job.working_dir: {job.working_dir}")
        logger.debug(f"job.working_dir_user: {job.working_dir_user}")
        logger.debug(f"job.first_task_index: {job.first_task_index}")
        logger.debug(f"job.last_task_index: {job.last_task_index}")
        logger.debug(f'START workflow "{workflow.name}"')
        job_working_dir = job.working_dir

    try:
        process_workflow: ProcessWorkflowType
        match resource.type:
            case ResourceType.LOCAL:
                process_workflow = local_process_workflow
            case ResourceType.SLURM_SUDO:
                process_workflow = slurm_sudo_process_workflow
            case ResourceType.SLURM_SSH:
                process_workflow = slurm_ssh_process_workflow

        process_workflow(
            workflow=workflow,
            dataset=dataset,
            job_id=job_id,
            user_id=user_id,
            workflow_dir_local=local_job_dir,
            workflow_dir_remote=remote_job_dir,
            logger_name=logger_name,
            worker_init=worker_init,
            first_task_index=job.first_task_index,
            last_task_index=job.last_task_index,
            job_attribute_filters=job.attribute_filters,
            job_type_filters=job.type_filters,
            resource=resource,
            profile=profile,
            user_cache_dir=user_cache_dir,
            fractal_ssh=fractal_ssh,
            slurm_account=job.slurm_account,
        )

        logger.info(
            f'End execution of workflow "{workflow.name}"; '
            f"more logs at {str(log_file_path)}"
        )
        logger.debug(f'END workflow "{workflow.name}"')

        # Update job DB entry
        with next(DB.get_sync_db()) as db_sync:
            job = db_sync.get_one(JobV2, job_id)
            job.status = JobStatusType.DONE
            job.end_timestamp = get_timestamp()
            with log_file_path.open("r") as f:
                logs = f.read()
            job.log = logs
            db_sync.merge(job)
            db_sync.commit()

    except JobExecutionError as e:
        logger.debug(f'FAILED workflow "{workflow.name}", JobExecutionError.')
        logger.info(f'Workflow "{workflow.name}" failed (JobExecutionError).')
        with next(DB.get_sync_db()) as db_sync:
            job = db_sync.get_one(JobV2, job_id)
            fail_job(
                db=db_sync,
                job=job,
                log_msg=(
                    f"JOB ERROR in Fractal job {job.id}:\n"
                    f"TRACEBACK:\n{e.assemble_error()}"
                ),
                logger_name=logger_name,
            )

    except Exception:
        logger.debug(f'FAILED workflow "{workflow.name}", unknown error.')
        logger.info(f'Workflow "{workflow.name}" failed (unkwnon error).')

        current_traceback = traceback.format_exc()
        with next(DB.get_sync_db()) as db_sync:
            job = db_sync.get_one(JobV2, job_id)
            fail_job(
                db=db_sync,
                job=job,
                log_msg=(
                    f"UNKNOWN ERROR in Fractal job {job.id}\n"
                    f"TRACEBACK:\n{current_traceback}"
                ),
                logger_name=logger_name,
            )

    finally:
        reset_logger_handlers(logger)
        _zip_folder_to_file_and_remove(folder=job_working_dir)