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 the derived work dir and coordination dir; the job store
+    creates itself.
+
+    :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)
+    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}")
+        )
+    if work_dir is None:
+        work_dir = os.path.join(run_dir, "work")
+        os.makedirs(work_dir, exist_ok=True)
+    if coordination_dir is None:
+        coordination_dir = os.path.join(run_dir, "coordination")
+        os.makedirs(coordination_dir, exist_ok=True)
+    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,24 @@ 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. The
+            # jobStore-derivation branch inside this is only reachable for
+            # direct callers of the flag-based parser (jobstore_as_flag=True)
+            # that leave --jobStore unset. The CWL/WDL runners fill in their
+            # own jobStore default before calling setOptions, and the plain
+            # `toil` entry point requires jobStore as a positional argument,
+            # so neither ever reaches it in practice.
+            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 +1206,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; the batch system
+        isn't built yet here, and grid systems log their own dir on startup.
+        """
+        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 +1790,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 +1804,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,
@@ -53,6 +56,7 @@
     TemporaryID,
 )
 from toil.jobStores.abstractJobStore import AbstractJobStore, NoSuchJobException, TOIL_WORKER_NO_JOB_STORE_EXIT_CODE
+from toil.lib.io import ensure_dir_exists
 from toil.lib.throttle import LocalThrottle
 from toil.provisioners.abstractProvisioner import AbstractProvisioner
 from toil.provisioners.clusterScaler import ScalerThread, NonScalableBatchSystemError
@@ -262,6 +266,22 @@ def run(self) -> Any:
 
         :return: The return value of the root job's run function.
         """
+
+        # Toil.getToilWorkDir and get_local_workflow_coordination_dir already
+        # create these directories during Toil.__enter__, before the Leader
+        # exists. These calls are a defensive backstop for any code path that
+        # reaches Leader.run() without having gone through Toil.__enter__ first.
+        ensure_dir_exists(self.config.workDir, "--workDir")
+        ensure_dir_exists(self.config.coordination_dir, "--coordinationDir")
+        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,12 +1613,11 @@ 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,
-                    )
 
+                # 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 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
@@ -1620,6 +1639,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 +1675,32 @@ def process_finished_job_description(
                                         % log_file
                                     )
 
+                if (
+                    replacement_job.logJobStoreFileID is None
+                    and not found_batch_system_log
+                ):
+                    if batch_system_id is None:
+                        logger.warning(
+                            "No log file is present, despite job failing: %s. "
+                            "Toil does not retain worker logs by default; rerun with "
+                            "--writeLogs=PATH or --writeLogsGzip=PATH to save failed "
+                            "jobs' logs to disk. Toil was not able to look for logs "
+                            "from the batch system for this job; check the batch "
+                            "system's own tools or logs directly.",
+                            replacement_job,
+                        )
+                    else:
+                        logger.warning(
+                            "No log file is present, despite job failing: %s. "
+                            "Toil does not retain worker logs by default; rerun with "
+                            "--writeLogs=PATH or --writeLogsGzip=PATH to save failed "
+                            "jobs' logs to disk. Toil looked for the batch system's "
+                            "own logs (see --batchLogsDir) but found none; check the "
+                            "batch system's own tools or logs directly if you are "
+                            "running on a grid engine.",
+                            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",