Add --runDir option; make --workDir/--coordinationDir create-if-missing (#5516)

src/toil/common.py

@@ -83,7 +83,7 @@
 from toil.lib.retry import retry
 from toil.lib.threading import ensure_filesystem_lockable
 from toil.lib.url import URLAccess
-from toil.options.common import JOBSTORE_HELP, add_base_toil_options
+from toil.options.common import JOBSTORE_HELP, add_base_toil_options, parse_jobstore
 from toil.options.cwl import add_cwl_options
 from toil.options.runner import add_runner_options
 from toil.options.wdl import add_wdl_options
@@ -123,6 +123,42 @@ def get_default_config_path() -> str:
     return os.path.join(get_toil_home(), "default.yaml")
 
 
+def derive_run_dir_defaults(
+    run_dir: str,
+    job_store: str | None,
+    work_dir: str | None,
+    coordination_dir: str | None,
+) -> tuple[str, str, str, str]:
+    """
+    Given --runDir and the current values of --jobStore, --workDir, and
+    --coordinationDir, fill in defaults for any of them left unset,
+    derived from --runDir. Explicit values always win.
+
+    Creates run_dir on disk; work_dir, coordination_dir, and the job
+    store are only computed here and create themselves later.
+
+    :return: (run_dir, job_store, work_dir, coordination_dir), with
+             run_dir made absolute and the other three either the
+             original explicit value or the runDir-derived default.
+    """
+    run_dir = os.path.abspath(run_dir)
+    os.makedirs(run_dir, exist_ok=True)
+    if job_store is None:
+        # Give each invocation its own job store directory under runDir,
+        # so concurrent workflows sharing one --runDir don't collide
+        # trying to create the same job store. The exact path is logged
+        # at startup (see Toil._log_resolved_paths) for --restart.
+        job_store = parse_jobstore(
+            os.path.join(run_dir, f"jobstore-{uuid.uuid4().hex}")
+        )
+        # The job store must not already exist, so it isn't created here.
+    if work_dir is None:
+        work_dir = os.path.join(run_dir, "work")
+    if coordination_dir is None:
+        coordination_dir = os.path.join(run_dir, "coordination")
+    return run_dir, job_store, work_dir, coordination_dir
+
+
 class Config:
     """Class to represent configuration operations for a toil workflow run."""
 
@@ -173,6 +209,7 @@ class Config:
     colored_logs: bool
     workDir: str | None
     coordination_dir: str | None
+    runDir: str | None
     noStdOutErr: bool
     stats: bool
 
@@ -353,9 +390,18 @@ def set_option(option_name: str, old_names: list[str] | None = None) -> None:
 
         # Core options
         set_option("jobStore")
-        # TODO: LOG LEVEL STRING
         set_option("workDir")
         set_option("coordination_dir")
+        set_option("runDir")
+
+        if self.runDir is not None:
+            # A single --runDir was given. Derive defaults for anything the
+            # user didn't set explicitly; explicit flags always win.
+            self.runDir, self.jobStore, self.workDir, self.coordination_dir = (
+                derive_run_dir_defaults(
+                    self.runDir, self.jobStore, self.workDir, self.coordination_dir
+                )
+            )
 
         set_option("noStdOutErr")
         set_option("stats")
@@ -1154,11 +1200,32 @@ def __enter__(self) -> "Toil":
         self._start_time = time.time()
         self._inContextManager = True
 
+        self._log_resolved_paths(config)
+
         # This will make sure `self.__exit__()` is called when we get a SIGTERM signal.
         signal.signal(signal.SIGTERM, lambda *_: sys.exit(1))
 
         return self
 
+    def _log_resolved_paths(self, config: "Config") -> None:
+        """
+        Log the resolved job store, work dir, and coordination dir paths
+        at INFO, once, so users can find them without tracing flag
+        precedence themselves. Skips the batch logs dir, since the batch
+        system isn't built yet here; see Leader.run for that.
+        """
+        assert config.workflowID is not None
+        work_dir = self.getLocalWorkflowDir(config.workflowID, config.workDir)
+        coordination_dir = self.get_local_workflow_coordination_dir(
+            config.workflowID, config.workDir, config.coordination_dir
+        )
+        logger.info(
+            "Resolved Toil run paths: job store: %s, work dir: %s, coordination dir: %s",
+            self.canonical_locator(config.jobStore),
+            work_dir,
+            coordination_dir,
+        )
+
     def __exit__(
         self,
         exc_type: type[BaseException] | None,
@@ -1717,8 +1784,9 @@ def getToilWorkDir(configWorkDir: str | None = None) -> str:
         Return a path to a writable directory under which per-workflow directories exist.
 
         This directory is always required to exist on a machine, even if the Toil
-        worker has not run yet.  If your workers and leader have different temp
-        directories, you may need to set TOIL_WORKDIR.
+        worker has not run yet, and is created if it does not already exist. If
+        your workers and leader have different temp directories, you may need to
+        set TOIL_WORKDIR.
 
         :param configWorkDir: Value passed to the program using the --workDir flag
         :return: Path to the Toil work directory, constant across all machines
@@ -1730,9 +1798,13 @@ def getToilWorkDir(configWorkDir: str | None = None) -> str:
             or tempfile.gettempdir()
         )
         if not os.path.exists(workDir):
-            raise RuntimeError(
-                f"The directory specified by --workDir or TOIL_WORKDIR ({workDir}) does not exist."
-            )
+            try:
+                os.makedirs(workDir, exist_ok=True)
+            except OSError as e:
+                raise RuntimeError(
+                    f"The directory specified by --workDir or TOIL_WORKDIR "
+                    f"({workDir}) does not exist and could not be created: {e}"
+                )
         return workDir
 
     @classmethod

src/toil/cwl/cwltoil.py

@@ -111,7 +111,13 @@
 
 from toil.batchSystems.abstractBatchSystem import InsufficientSystemResources
 from toil.batchSystems.registry import DEFAULT_BATCH_SYSTEM
-from toil.common import Config, Toil, addOptions, InconsistentConfigurationError
+from toil.common import (
+    Config,
+    Toil,
+    addOptions,
+    derive_run_dir_defaults,
+    InconsistentConfigurationError,
+)
 from toil.cwl import check_cwltool_version
 from toil.lib.directory import DirectoryContents, decode_directory, encode_directory
 from toil.lib.interpreter import (
@@ -4748,6 +4754,25 @@ def main(args: list[str] | None = None, stdout: TextIO = sys.stdout) -> int:
     tmp_outdir_prefix = options.tmp_outdir_prefix or tmpdir_prefix
     # tmpdir_prefix and tmp_outdir_prefix must not be checked for existence as they may exist on a worker only path
     # See https://github.com/DataBiosphere/toil/issues/5310
+
+    if options.runDir is not None:
+        # A single --runDir was given. Derive defaults for the job store,
+        # work dir, coordination dir, and cachedir from it, for anything
+        # not set explicitly. This has to happen here, before the
+        # fallbacks below run, because by the time Toil's own
+        # Config.setOptions sees these options, options.jobStore is
+        # never None (the fallback below always fills it in first).
+        options.runDir, options.jobStore, options.workDir, options.coordination_dir = (
+            derive_run_dir_defaults(
+                options.runDir,
+                options.jobStore,
+                options.workDir,
+                options.coordination_dir,
+            )
+        )
+        if options.cachedir is None:
+            options.cachedir = os.path.join(options.runDir, "image-cache")
+
     workdir = options.workDir or tmp_outdir_prefix
 
     if options.jobStore is None:
@@ -4856,6 +4881,15 @@ def main(args: list[str] | None = None, stdout: TextIO = sys.stdout) -> int:
             expected_config = Config()
             expected_config.setOptions(options)
 
+            if expected_config.runDir is not None and "CWL_SINGULARITY_CACHE" not in os.environ:
+                # try_prepull() and cwltool's own Singularity execution code
+                # read this straight from the environment, so default it
+                # here rather than on an options/config attribute.
+                os.environ["CWL_SINGULARITY_CACHE"] = os.path.join(
+                    expected_config.runDir, "image-cache", "singularity"
+                )
+                os.makedirs(os.environ["CWL_SINGULARITY_CACHE"], exist_ok=True)
+
             # Before showing the options to any cwltool stuff that wants to
             # load the workflow, transform options.cwltool, where our
             # argument for what to run is, to handle Dockstore workflows.

src/toil/leader.py

@@ -34,6 +34,9 @@
     BatchJobExitReason,
     UpdatedBatchJobInfo,
 )
+from toil.batchSystems.abstractGridEngineBatchSystem import (
+    AbstractGridEngineBatchSystem,
+)
 from toil.bus import (
     JobCompletedMessage,
     JobFailedMessage,
@@ -262,6 +265,16 @@ def run(self) -> Any:
 
         :return: The return value of the root job's run function.
         """
+
+        if isinstance(self.batchSystem, AbstractGridEngineBatchSystem):
+            # The batch system isn't available yet when Toil logs the other
+            # resolved run paths (see Toil._log_resolved_paths), so log this
+            # one here instead, now that it exists. Only grid batch systems
+            # actually write their own logs to this directory.
+            logger.info(
+                "Resolved batch logs dir: %s", self.batchSystem.get_batch_logs_dir()
+            )
+
         self.jobStore.write_kill_flag(kill=False)
 
         with enlighten.get_manager(
@@ -1593,19 +1606,21 @@ def process_finished_job_description(
                 # If the batch system returned a non-zero exit code then the worker
                 # is assumed not to have captured the failure of the job, so we
                 # reduce the try count here.
-                if replacement_job.logJobStoreFileID is None:
-                    logger.warning(
-                        "No log file is present, despite job failing: %s",
-                        replacement_job,
-                    )
 
-                if batch_system_id is not None:
+                # Search for the batch system's own logs first, so the
+                # "no log file" warning below is only shown when Toil
+                # genuinely found nothing, and can say so specifically.
+                found_batch_system_log = False
+                if (
+                    isinstance(self.batchSystem, AbstractGridEngineBatchSystem)
+                    and batch_system_id is not None
+                ):
                     # Look for any standard output/error files created by the batch system.
                     # They will only appear if the batch system actually supports
                     # returning logs to the machine that submitted jobs, or if
                     # --workDir / TOIL_WORKDIR is on a shared file system.
-                    # They live directly in the Toil work directory because that is
-                    # guaranteed to exist on the leader and workers.
+                    # They live in --batchLogsDir, or the Toil work directory
+                    # if that isn't set.
                     file_list = glob.glob(
                         self.batchSystem.format_std_out_err_glob(batch_system_id)
                     )
@@ -1620,6 +1635,7 @@ def process_finished_job_description(
                         else:
                             with log_stream:
                                 if os.path.getsize(log_file) > 0:
+                                    found_batch_system_log = True
                                     StatsAndLogging.logWithFormatting(
                                         f'Log from job "{job_store_id}"',
                                         log_stream,
@@ -1655,6 +1671,27 @@ def process_finished_job_description(
                                         % log_file
                                     )
 
+                if (
+                    replacement_job.logJobStoreFileID is None
+                    and not found_batch_system_log
+                ):
+                    # Alert the user that the worker failed to report in
+                    # like it was supposed to. Only mention batch system
+                    # logs for batch systems that actually use them.
+                    if isinstance(self.batchSystem, AbstractGridEngineBatchSystem):
+                        logger.warning(
+                            "No log file is present, despite job failing: %s. "
+                            "Toil looked for the batch system's own logs "
+                            "(see --batchLogsDir) but found none; check the "
+                            "batch system's own tools or logs directly.",
+                            replacement_job,
+                        )
+                    else:
+                        logger.warning(
+                            "No log file is present, despite job failing: %s.",
+                            replacement_job,
+                        )
+
                 # Tell the job to reset itself after a failure.
                 # It needs to know the failure reason if available; some are handled specially.
                 replacement_job.setupJobAfterFailure(

src/toil/lib/io.py

@@ -365,6 +365,34 @@ def try_path(path: str, min_size: int = 100 * 1024 * 1024) -> str | None:
 
     return path
 
+
+def ensure_dir_exists(path: str | None, arg_name: str) -> None:
+    """
+    Ensure that the given directory exists, creating it if necessary.
+
+    Logs a critical error and exits if the directory does not exist and
+    cannot be created.
+
+    :param arg_name: Name of the flag or option that provided the path, used
+           to make the error message actionable.
+    """
+
+    if path is None:
+        return
+    if not os.path.exists(path):
+        try:
+            os.makedirs(path, exist_ok=True)
+        except OSError as e:
+            logger.critical(
+                "The path provided to %s (%s) does not exist and could not "
+                "be created: %s",
+                arg_name,
+                path,
+                e,
+            )
+            sys.exit(1)
+
+
 def path_union(first_path: str, second_path: str | None) -> str:
     """
     Union two os.pathsep-separated PATH environment variables.

src/toil/options/common.py

@@ -257,10 +257,6 @@ def __call__(
             workDir = values
             if workDir is not None:
                 workDir = os.path.abspath(workDir)
-                if not os.path.exists(workDir):
-                    raise RuntimeError(
-                        f"The path provided to --workDir ({workDir}) does not exist."
-                    )
 
                 if len(workDir) > 80:
                     logger.warning(
@@ -281,10 +277,6 @@ def __call__(
             coordination_dir = values
             if coordination_dir is not None:
                 coordination_dir = os.path.abspath(coordination_dir)
-                if not os.path.exists(coordination_dir):
-                    raise RuntimeError(
-                        f"The path provided to --coordinationDir ({coordination_dir}) does not exist."
-                    )
             setattr(namespace, self.dest, coordination_dir)
 
     def make_closed_interval_action(
@@ -357,6 +349,16 @@ def is_within(x: int | float) -> bool:
         "When sharing a cache between containers on a host, this directory must be "
         "shared between the containers.",
     )
+    core_options.add_argument(
+        "--runDir",
+        dest="runDir",
+        default=None,
+        env_var="TOIL_RUN_DIR",
+        metavar="PATH",
+        help="Directory under which to place the job store, work dir, coordination "
+        "dir, and CWL/WDL image caches for this run, unless overridden individually. "
+        "Toil creates the directory and its subdirectories.",
+    )
     core_options.add_argument(
         "--noStdOutErr",
         dest="noStdOutErr",

src/toil/test/cwl/cwlTest.py

@@ -250,6 +250,26 @@ def test_cwl_cmdline_input(self) -> None:
             # If the workflow runs, it must have had options
             cwltoil.main(args, stdout=st)
 
+    def test_cwl_run_dir(self, tmp_path: Path) -> None:
+        """
+        Test that --runDir derives the CWL image cache location.
+        """
+        from toil.cwl import cwltoil
+
+        run_dir = tmp_path / "rundir"
+        with get_data("test/cwl/echo.cwl") as cwlfile:
+            args = [
+                f"--runDir={run_dir}",
+                "--outdir",
+                str(tmp_path / "out"),
+                str(cwlfile),
+                "--message",
+                "str",
+            ]
+            st = StringIO()
+            cwltoil.main(args, stdout=st)
+        assert (run_dir / "image-cache").is_dir()
+
     def _tester(
         self,
         cwlfile: Path,