Issues/5477 cpu usage docker cwl

src/toil/cwl/cwltoil.py

@@ -31,13 +31,21 @@
 import pprint
 import shutil
 import stat
+import subprocess
 import sys
 import textwrap
 import uuid
 
 # This is also in configargparse but MyPy doesn't know it
 from argparse import RawDescriptionHelpFormatter
-from collections.abc import Callable, Iterator, Mapping, MutableMapping, MutableSequence
+from collections.abc import (
+    Callable,
+    Generator,
+    Iterator,
+    Mapping,
+    MutableMapping,
+    MutableSequence,
+)
 from tempfile import NamedTemporaryFile, TemporaryFile, gettempdir
 from threading import Thread
 from typing import IO, Any, Literal, Optional, Protocol, TextIO, TypeVar, Union, cast
@@ -67,6 +75,8 @@
     fill_in_defaults,
     shortname,
 )
+from cwltool.docker import DockerCommandLineJob, PodmanCommandLineJob
+from cwltool.job import ContainerCommandLineJob
 from cwltool.secrets import SecretStore
 from cwltool.singularity import SingularityCommandLineJob
 from cwltool.software_requirements import (
@@ -102,6 +112,12 @@
 from toil.common import Config, Toil, addOptions
 from toil.cwl import check_cwltool_version
 from toil.lib.directory import DirectoryContents, decode_directory, encode_directory
+from toil.lib.interpreter import (
+    add_injections,
+    command_line_to_shell_script,
+    handle_injection_messages_from_outdir,
+    shell_script_to_command_line,
+)
 from toil.lib.misc import call_command
 from toil.lib.trs import resolve_workflow
 from toil.provisioners.clusterScaler import JobTooBigError
@@ -1086,6 +1102,32 @@ def run_jobs(
         return super().run_jobs(process, job_order_object, logger, runtime_context)
 
 
+class ToilContainerCommandLineJob(ContainerCommandLineJob):
+    """Container job that collects resource stats from injected in-container code."""
+
+    def _execute(
+        self,
+        runtime: list[str],
+        env: MutableMapping[str, str],
+        runtimeContext: cwltool.context.RuntimeContext,
+        monitor_function: Callable[["subprocess.Popen[str]"], None] | None = None,
+    ) -> None:
+        super()._execute(runtime, env, runtimeContext, monitor_function)
+        handle_injection_messages_from_outdir(self.outdir)
+
+
+class ToilDockerCommandLineJob(ToilContainerCommandLineJob, DockerCommandLineJob):
+    """Docker container job with Toil runtime injection support."""
+
+
+class ToilPodmanCommandLineJob(ToilContainerCommandLineJob, PodmanCommandLineJob):
+    """Podman container job with Toil runtime injection support."""
+
+
+class ToilSingularityCommandLineJob(ToilContainerCommandLineJob, SingularityCommandLineJob):
+    """Singularity container job with Toil runtime injection support."""
+
+
 class ToilTool:
     """Mixin to hook Toil into a cwltool tool type."""
 
@@ -1141,7 +1183,64 @@ def __str__(self) -> str:
 
 
 class ToilCommandLineTool(ToilTool, cwltool.command_line_tool.CommandLineTool):
-    """Subclass the cwltool command line tool to provide the custom ToilPathMapper."""
+    """Subclass the cwltool command line tool to provide the custom ToilPathMapper
+    and add the monitoring code to the job's container command line."""
+
+    def make_job_runner(
+        self, runtimeContext: cwltool.context.RuntimeContext
+    ) -> type[cwltool.job.JobBase]:
+        """Use Toil container job classes that collect injected runtime messages."""
+        parent_class = super().make_job_runner(runtimeContext)
+        if parent_class is DockerCommandLineJob:
+            return ToilDockerCommandLineJob
+        if parent_class is PodmanCommandLineJob:
+            return ToilPodmanCommandLineJob
+        if parent_class is SingularityCommandLineJob:
+            return ToilSingularityCommandLineJob
+        return parent_class
+
+    def _uses_container(self, runtimeContext: cwltool.context.RuntimeContext) -> bool:
+        if not runtimeContext.use_container:
+            return False
+        docker_req, _ = self.get_requirement("DockerRequirement")
+        if docker_req is not None:
+            return True
+        if runtimeContext.find_default_container is not None:
+            return runtimeContext.find_default_container(self) is not None
+        return runtimeContext.default_container is not None
+
+    @staticmethod
+    def _file_mounts_from_pathmapper(
+        job: ContainerCommandLineJob,
+    ) -> list[tuple[str, str]]:
+        file_mounts: list[tuple[str, str]] = []
+        for location in job.pathmapper.files():
+            ent = job.pathmapper.mapper(location)
+            if ent.type == "File" and not ent.resolved.startswith("_:"):
+                file_mounts.append((ent.resolved, ent.target))
+        return file_mounts
+
+    def job(
+        self,
+        job_order: CWLObjectType,
+        output_callbacks: Callable[[CWLObjectType | None, str], None],
+        runtimeContext: cwltool.context.RuntimeContext,
+    ) -> Generator[
+        cwltool.job.JobBase | cwltool.command_line_tool.CallbackJob, None, None
+    ]:
+        """
+        Override the job method to inject resource (cpu & memory) monitoring code into
+        the job's container command line.
+        """
+        for job in super().job(job_order, output_callbacks, runtimeContext):
+            if isinstance(job, ContainerCommandLineJob) and self._uses_container(
+                runtimeContext
+            ):
+                file_mounts = self._file_mounts_from_pathmapper(job)
+                script = command_line_to_shell_script(job.command_line)
+                script = add_injections(script, file_mounts)
+                job.command_line = shell_script_to_command_line(script)
+            yield job
 
     def _initialworkdir(
         self, j: cwltool.job.JobBase | None, builder: cwltool.builder.Builder

src/toil/job.py

@@ -3304,20 +3304,8 @@ def _executor(self, stats: StatsDict, fileStore: AbstractFileStore) -> Iterator[
             )
             job_time = time.time() - startTime
             job_cpu_time = totalCpuTime - startClock
-            allocated_cpu_time = job_time * self.cores
-
-            if job_cpu_time > allocated_cpu_time and allocated_cpu_time > 0:
-                # Too much CPU was used by this job! Maybe we're using a batch
-                # system that doesn't/can't sandbox us and we started too many
-                # threads. Complain to the user!
-                excess_factor = job_cpu_time / allocated_cpu_time
-                fileStore.log_to_leader(
-                    f"Job {self.description} used {excess_factor:.2f}x more "
-                    f"CPU than the requested {self.cores} cores. Consider "
-                    f"increasing the job's required CPU cores or limiting the "
-                    f"number of processes/threads launched.",
-                    level=logging.WARNING,
-                )
+
+            self._check_cpu_usage(job_time, job_cpu_time, fileStore)
 
             # Finish up the stats
             if stats is not None:
@@ -3337,6 +3325,46 @@ def _executor(self, stats: StatsDict, fileStore: AbstractFileStore) -> Iterator[
                         succeeded=str(succeeded),
                     )
                 )
+
+
+    def _check_cpu_usage(
+        self,
+        job_time: float,
+        job_cpu_time: float,
+        fileStore: AbstractFileStore,
+        threshold_factor: float = 1.05,
+    ) -> None:
+        """
+        Log a warning when a job consumes more CPU time than its allocation.
+
+        If *job_time* is non-positive, returns immediately (wall-clock time may
+        lack resolution for very short jobs). If *cores* is non-positive, logs
+        a warning that overuse cannot be assessed and returns, avoiding
+        division by zero. Otherwise, compares *job_cpu_time* to
+        ``job_time * cores * threshold_factor``.
+        """
+        if job_time <= 0:
+            # The job may have run too quickly for wall-clock resolution.
+            return
+        if self.cores <= 0:
+            # Jobs should not have non-positive core requests, but avoid divide-by-zero.
+            fileStore.log_to_leader(
+                f"Job {self.description} has invalid requested cores value "
+                f"{self.cores}; cannot evaluate CPU overuse.",
+                level=logging.WARNING,
+            )
+            return
+        allocated_cpu_time = job_time * self.cores
+
+        excess_factor = job_cpu_time / allocated_cpu_time
+        if excess_factor > threshold_factor:
+            fileStore.log_to_leader(
+                f"Job {self.description} used {excess_factor:.2f}x more "
+                f"CPU than the requested {self.cores} cores. Consider "
+                f"increasing the job's required CPU cores or limiting the "
+                f"number of processes/threads launched.",
+                level=logging.WARNING,
+            )
 
     def _runner(
         self,