diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index ab8ea5beb1..f3622e4898 100644 --- a/src/toil/cwl/cwltoil.py +++ b/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 ( @@ -74,6 +84,7 @@ get_container_from_software_requirements, ) from cwltool.stdfsaccess import StdFsAccess, abspath +from cwltool.udocker import UDockerCommandLineJob from cwltool.utils import ( adjustDirObjs, aslist, @@ -81,6 +92,7 @@ get_listing, normalizeFilesDirs, visit_class, + OutputCallbackType, ) from cwl_utils.types import ( CWLDirectoryType, @@ -102,6 +114,13 @@ from toil.common import Config, Toil, addOptions, InconsistentConfigurationError from toil.cwl import check_cwltool_version from toil.lib.directory import DirectoryContents, decode_directory, encode_directory +from toil.lib.interpreter import ( + INJECTED_MESSAGE_DIR, + add_injections, + command_line_to_shell_script, + handle_injection_messages_from, + 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 @@ -1142,9 +1161,134 @@ def __str__(self) -> str: """Return string representation of this tool type.""" return f'{self.__class__.__name__}({repr(getattr(self, "tool", {}).get("id", "???"))})' +# We need to know which cwltool container implementations run the container +# under the calling process's child tree. Unlisted container implementations +# are assumed to run it elsewhere through a deamon. +# +# We have to list it this way around (and not list the implementations that use +# a daemon) because some of these extend DockerCommandLineJob. +CHILD_PROCESS_CONTAINER_JOBS = (PodmanCommandLineJob, SingularityCommandLineJob, UDockerCommandLineJob) + + +class ToilDockerCommandLineJob(DockerCommandLineJob): + """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: + # This job is going to run in a container that makes it not be a + # descendant of our process. So we need to inject code to count its + # resource usage. + + # First we need the command that's actually going to run. To be fully + # spec-compliant, we need to make sure Docker ENTRYPOINT keeps working. + # So we need to fetch it out, put it in the script, and remember to + # override it with a shell later. See + # + + # We know the selected container image is the last element in runtime. + # Inspect it to get its entrypoint list + image_config = json.loads( + subprocess.check_output( + [ + self.docker_exec, + "inspect", + "-f", + "{{json .Config}}", + runtime[-1], + ] + ) + ) + # Null entrypoints are allowed; Docker doesn't seem to put literal + # nulls but who knows how it might change. + entrypoint: list[str] = image_config.get("Entrypoint") or [] + + # We also need to ensure that if the CWL baseCommand is empty, the + # image command list runs. + default_command: list[str] = image_config.get("Cmd") or [] + + # Make a command string that does the expected user work + script = command_line_to_shell_script(entrypoint + (self.command_line or default_command)) + + # Our injected code also checks to make sure files mounted on Docker + # for Mac are intact, so we need to tell it about the file + # mounts we are going to use. + file_mounts = self._file_mounts_from_pathmapper() + + # Since we're not allowed to drop stuff in the working directory, we + # put the resource usage info in the temp directory, which cwltool + # mounts, as divined by Anthropic Claude. See + # + script = add_injections(script, file_mounts, os.path.join(self.CONTAINER_TMPDIR, INJECTED_MESSAGE_DIR)) + self.command_line = shell_script_to_command_line(script) + + # Adjust the runtime to clear the entrypoint. + # See . + # Stick the argument and value in before the last element. + runtime[-1:-1] = ["--entrypoint", ""] + + # By the time the base class _execute() returns, the temp directory + # holding the resource usage information has already been deleted. So + # we need to hook into one of the functions it calls, from somewhere + # where we have access to our tmpdir (like here). + def message_handler_output_callback( + output_callbacks: OutputCallbackType | None, + outputs: CWLObjectType | None, + processStatus: str, + ) -> None: + """ + Output callback that processes resource usage messages. + + Needs to be partial'd into a callback chain. + """ + handle_injection_messages_from(os.path.join(self.tmpdir, INJECTED_MESSAGE_DIR)) + if output_callbacks: + output_callbacks(outputs, processStatus) + # Hook into the output callback chain + self.output_callback = functools.partial(message_handler_output_callback, self.output_callback) + + # Run the container + super()._execute(runtime, env, runtimeContext, monitor_function) + # By the time we get here, the temp directory where we hide the + # resource usage info has already been deleted. + + def _file_mounts_from_pathmapper(self) -> list[tuple[str, str]]: + """ + Get all the container mounts that will be used for a containerized job. + + :returns: a list of (host path, container path) tuples, one per mount. + """ + file_mounts: list[tuple[str, str]] = [] + for location in self.pathmapper.files(): + ent = self.pathmapper.mapper(location) + if ent.type == "File" and not ent.resolved.startswith("_:"): + file_mounts.append((ent.resolved, ent.target)) + return file_mounts + 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 a job class that monitors container resource usage if needed. + + We only act on container systems where the container isn't a descendant + process of us. + """ + parent_class = super().make_job_runner(runtimeContext) + if parent_class is DockerCommandLineJob: + # There's only one daemon-containerized container system: it's + # Docker itself (and not any of the subclasses of this), + return ToilDockerCommandLineJob + return parent_class def _initialworkdir( self, j: cwltool.job.JobBase | None, builder: cwltool.builder.Builder @@ -1186,7 +1330,7 @@ def _initialworkdir( # Notice that we have downloaded our inputs. Explain which files # those are here and what the task will expect to call them. self._toil_job.files_downloaded_hook(host_and_job_paths) - + class ToilExpressionTool(ToilTool, cwltool.command_line_tool.ExpressionTool): """Subclass the cwltool expression tool to provide the custom ToilPathMapper.""" diff --git a/src/toil/jobStores/aws/jobStore.py b/src/toil/jobStores/aws/jobStore.py index 32df188d7e..739f44f375 100644 --- a/src/toil/jobStores/aws/jobStore.py +++ b/src/toil/jobStores/aws/jobStore.py @@ -195,6 +195,11 @@ def initialize(self, config: Config) -> None: Create bucket, raise if it already exists. Set options from config. + + Note that polling for bucket existence only happens once; there's not a + good way to guard against races to create the bucket that also handles + the possibility that a request to AWS could return an error but still + create the bucket. """ logger.debug( f"Instantiating {self.__class__} for region {self.region} with bucket: '{self.bucket_name}'" diff --git a/src/toil/lib/aws/s3.py b/src/toil/lib/aws/s3.py index 9f8d7b4c4c..c8ba5c5ace 100644 --- a/src/toil/lib/aws/s3.py +++ b/src/toil/lib/aws/s3.py @@ -90,6 +90,10 @@ def create_s3_bucket( Create an AWS S3 bucket, using the given Boto3 S3 session, with the given name, in the given region. + If the bucket already exists, uses it. (It is difficult to implement + retries if this function is not idempotent and AWS can successfully create + the bucket but produce an error on the client.) + Supports the us-east-1 region, where bucket creation is special. *ALL* S3 bucket creation should use this function. @@ -107,13 +111,18 @@ def create_s3_bucket( False. """ logger.info("Creating bucket '%s' in region %s.", bucket_name, region) - if region == "us-east-1": # see https://github.com/boto/boto3/issues/125 - bucket = s3_resource.create_bucket(Bucket=bucket_name) - else: - bucket = s3_resource.create_bucket( - Bucket=bucket_name, - CreateBucketConfiguration={"LocationConstraint": region}, - ) + try: + if region == "us-east-1": # see https://github.com/boto/boto3/issues/125 + bucket = s3_resource.create_bucket(Bucket=bucket_name) + else: + bucket = s3_resource.create_bucket( + Bucket=bucket_name, + CreateBucketConfiguration={"LocationConstraint": region}, + ) + except s3_resource.meta.client.exceptions.BucketAlreadyOwnedByYou: + # We could have created the bucket, then failed, then retried. So use + # it. + bucket = s3_resource.Bucket(bucket_name) # wait until the bucket exists before adding tags bucket.wait_until_exists() diff --git a/src/toil/lib/interpreter.py b/src/toil/lib/interpreter.py new file mode 100644 index 0000000000..8041625ed6 --- /dev/null +++ b/src/toil/lib/interpreter.py @@ -0,0 +1,285 @@ +""" +Shared code for workflow language interpreters implemented with Toil. +""" + +# Copyright (C) 2026 Regents of the University of California +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import glob +import logging +import os +import platform +import shlex +import textwrap +from typing import Iterable + +from toil.lib.resources import ResourceMonitor + +logger = logging.getLogger(__name__) + + +#### +# Shell command injection system +# +# When a workflow steps runs inside a container like Docker, where the contained +# process is not a descendant of Toil, the Toil worker process on the host cannot +# see how much CPU and RAM that step actually used. +# +# This system allows Toil to inject code into the container that will colklect +# and export resource usage information back to Toil. +# +# It also allows Toil to do other checks on the container system. +### + +# Code injected into containers container communicates back to the rest of Toil +# through files in a directory named this. +# Where the directory actually is is interpreter-specific. +INJECTED_MESSAGE_DIR = ".toil_runtime" + +# We mostly want to work with shell script strings, but CWL works with command +# argument lists, so we use these functions to convert back and forth. + +def command_line_to_shell_script(command_line: list[str]) -> str: + """ + Extract or synthesize the inner shell script from a cwltool argv list. + + We don't want to disturb user CWL command line binding lists that + explicitly ask for things like ["bash", "-c"]. + """ + return " ".join(shlex.quote(arg) for arg in command_line) + + +def shell_script_to_command_line(script: str) -> list[str]: + """ + Wrap a shell script as a list of arguments for launchign a process. + + The resulting command only requires a POSIX shell as sh. + """ + return ["sh", "-c", script] + +# Main function + +def add_injections( + command_string: str, + file_mounts: Iterable[tuple[str, str]], + message_dir: str, +) -> str: + """ + Add resource usage monitoring and file mount checking code to a command. + + The command is expected to be about to run in a container, under a + container system that does not itself attribute resource usage to the + calling Toil process (such as Docker, which uses a daemon). + + All shell code added is compatible with POSIX sh. + + :param command_string: shell command or script to modify + :param file_mounts: collection of (host path, container path) tuples for + files mounted into the container. Code will be added to require that + the container sees the complete file that the host sees. + :param message_dir: directory, absolute or relative to the working + directory that the command should record resource usage to + + :returns: shell command string (possibly containing multiple commands) that + runs the original command and reports resource usage. + + """ + + parts = [] + # We're running on Docker or another platform where Toil isn't an ancestor + # process, so we need to monitor CPU usage and so on from inside the + # container, since it won't be attributed to Toil child processes in the + # leader's self-monitoring. + + # TODO: Mount this script from a file Toil installs instead or something + # instead of injecting it in every command line, which makes it show up in + # logs. + + # Use https://www.shellcheck.net/ with #!/bin/sh at the top to make sure + # this is all sh-compatible. + + script = textwrap.dedent( + """\ + _toil_resource_monitor () { + # Turn off error checking and echo in here + set +ex + TOIL_MESSAGE_DIR="${1}" + mkdir -p "${TOIL_MESSAGE_DIR}" + + get_field () { + INFILE="${1}" + FIELD_NAME="${2}" + while IFS=' ' read -r KEY VALUE ; do + if [ "${KEY}" = "${FIELD_NAME}" ] ; then + echo "${VALUE}" + return + fi + done < "${INFILE}" + } + + sample_cpu_usec() { + if [ -f /sys/fs/cgroup/cpu.stat ] ; then + get_field /sys/fs/cgroup/cpu.stat usage_usec + elif [ -f /sys/fs/cgroup/cpuacct/cpuacct.stat ] ; then + echo $(( ( $(get_field /sys/fs/cgroup/cpuacct/cpuacct.stat user) + $(get_field /sys/fs/cgroup/cpuacct/cpuacct.stat system) ) * 10000 )) + fi + } + + sample_memory_bytes() { + if [ -f /sys/fs/cgroup/memory.stat ] ; then + get_field /sys/fs/cgroup/memory.stat anon + elif [ -f /sys/fs/cgroup/memory/memory.stat ] ; then + get_field /sys/fs/cgroup/memory/memory.stat total_rss + fi + } + + while true ; do + (printf "CPU\\t" ; sample_cpu_usec ; printf "Memory\\t" ; sample_memory_bytes) >> "${TOIL_MESSAGE_DIR}"/resources.tsv + sleep 1 + done + } + """ + ) + parts.append(script) + + # Launch in a subshell so that it doesn't interfere with "wait" in the + # main shell. + # + # We need another subshell as a background job, so we can get the PID which + # is used as the PGID for the background process group, so we can stop it + # later without job control. + parts.append(f"(set +m; (_toil_resource_monitor {message_dir} &) ) &") + + # We use a loop to try and wait on the whole process group the background + # monitoring is in by polling it, to make sure it's gone before we exit. + # See + # + # TODO: What if the user also traps EXIT? Then this won't work. + # TODO: This also doesn't (always? ever?) work to prevent Docker Swarm from + # reporting a container as running with an exit code. + stopper_script = textwrap.dedent( + """\ + TOIL_BG_PGID="${!}" + wait + trap "while kill -0 -${TOIL_BG_PGID} 2>/dev/null ; do kill -9 -${TOIL_BG_PGID} 2>/dev/null ; done" EXIT + """ + ) + parts.append(stopper_script) + + if platform.system() == "Darwin": + # With gRPC FUSE file sharing, files immediately downloaded before + # being mounted may appear as size 0 in the container due to a race + # condition. Check for this and produce an approperiate error. + + # We can't guarantee that the container has a stat binary, and we can't + # measure the size of a file with pure shell without reading it all, so + # we poll for stat and skip the check if it isn't there. + script = textwrap.dedent( + """\ + _toil_check_size () { + if ! which stat >/dev/null 2>/dev/null ; then + return + fi + TARGET_FILE="${1}" + GOT_SIZE="$(stat -c %s "${TARGET_FILE}")" + EXPECTED_SIZE="${2}" + if [ "${GOT_SIZE}" != "${EXPECTED_SIZE}" ] ; then + echo >&2 "Toil Error:" + echo >&2 "File size visible in container for ${TARGET_FILE} is size ${GOT_SIZE} but should be size ${EXPECTED_SIZE}" + echo >&2 "Are you using gRPC FUSE file sharing in Docker Desktop?" + echo >&2 "It doesn't work: see ." + exit 1 + fi + } + """ + ) + parts.append(script) + for host_path, job_path in file_mounts: + expected_size = os.path.getsize(host_path) + if expected_size != 0: + parts.append(f'_toil_check_size "{job_path}" {expected_size}') + + parts.append(command_string) + + return "\n".join(parts) + +# Helper functions to parse resource usage output + +def handle_message_file(file_path: str) -> None: + """ + Handle a message file received from injected code from :meth:`add_injections()`. + + Records the described usage as if it had occurred within a child process, + by talking to the :class:`toil.lib.resources.ResourceMonitor` system. + + :param file_path: the host-side path of the message file. + """ + logger.debug("Handling message file %s", file_path) + if os.path.basename(file_path) == "resources.tsv": + # This is a TSV of resource usage info. + first_cpu_usec: int | None = None + last_cpu_usec: int | None = None + max_memory_bytes: int | None = None + + for line in open(file_path): + if not line.endswith("\n"): + # Skip partial lines + continue + # For each full line we got + parts = line.strip().split("\t") + if len(parts) != 2: + # Skip odd-shaped lines + continue + if parts[0] == "CPU": + # Parse CPU usage + cpu_usec = int(parts[1]) + # Update summary stats + if first_cpu_usec is None: + first_cpu_usec = cpu_usec + last_cpu_usec = cpu_usec + elif parts[0] == "Memory": + # Parse memory usage + memory_bytes = int(parts[1]) + # Update summary stats + if max_memory_bytes is None or max_memory_bytes < memory_bytes: + max_memory_bytes = memory_bytes + + if max_memory_bytes is not None: + logger.info( + "Container used at about %s bytes of memory at peak", + max_memory_bytes, + ) + # Treat it as if used by a child process + ResourceMonitor.record_extra_memory(max_memory_bytes // 1024) + if last_cpu_usec is not None: + assert first_cpu_usec is not None + cpu_seconds = (last_cpu_usec - first_cpu_usec) / 1000000 + logger.info("Container used about %s seconds of CPU time", cpu_seconds) + # Treat it as if used by a child process + ResourceMonitor.record_extra_cpu(cpu_seconds) + +def handle_injection_messages_from(message_dir: str) -> None: + """ + Handle any message files in the given directory. + + Files would have been left by injected code from :meth:`add_injections()`. + """ + if not os.path.isdir(message_dir): + # It's possible for the container to never have made this directory + logger.debug("No message files found") + return + for filename in os.listdir(message_dir): + file_path = os.path.join(message_dir, filename) + if os.path.isfile(file_path): + handle_message_file(file_path) diff --git a/src/toil/test/cwl/cwlTest.py b/src/toil/test/cwl/cwlTest.py index 712d553eec..db199a283c 100644 --- a/src/toil/test/cwl/cwlTest.py +++ b/src/toil/test/cwl/cwlTest.py @@ -40,6 +40,7 @@ from schema_salad.exceptions import ValidationException +from toil.common import Toil from toil.cwl.utils import ( DirectoryStructure, download_structure, @@ -66,6 +67,7 @@ from toil.test import pneeds_torque as needs_torque from toil.test import pneeds_wes_server as needs_wes_server from toil.test import pslow as slow +from toil.utils.toilStats import get_stats, process_data log = logging.getLogger(__name__) CONFORMANCE_TEST_TIMEOUT = 10000 @@ -482,6 +484,44 @@ def test_s3_as_secondary_file(self, tmp_path: Path) -> None: with open(out["output"]["location"][len("file://") :]) as f: assert f.read().strip() == "When is s4 coming out?" + @needs_docker + @pytest.mark.docker + @pytest.mark.online + @pytest.mark.timeout(180) + def test_cpu_memory_monitoring(self, tmp_path: Path) -> None: + """Container CPU and memory usage is recorded in Toil job stats.""" + from toil.cwl import cwltoil + + with get_data("test/cwl/waste_cpu_memory.cwl") as cwl_file: + with get_data("test/cwl/empty.json") as inputs_file: + job_store = tmp_path / "jobStore" + main_args = [ + "--logDebug", + "--outdir", + str(tmp_path / "output_dir"), + str(cwl_file), + str(inputs_file), + "--jobStore", + str(job_store), + "--stats", + # TODO: this relies on the default container engine for + # toil-cwl-runner being Docker, because we don't have a + # --docker option to make it explicit. + ] + cwltoil.main(main_args) + + resumed_job_store = Toil.resumeJobStore(str(job_store)) + stats = get_stats(resumed_job_store) + collated_stats = process_data(resumed_job_store.config, stats) + + # The tool burns ~30s of CPU and ~1 GiB of RAM inside Docker. + # Per-job stats use KiB for memory and include injected container usage. + assert hasattr(collated_stats, "jobs") + assert hasattr(collated_stats.jobs, "total_clock") + assert hasattr(collated_stats.jobs, "max_memory") + assert collated_stats.jobs.total_clock >= 30, f"Total clock of {collated_stats.jobs.total_clock} is too low" + assert collated_stats.jobs.max_memory >= 1024 * 1024, f"Max memory of {collated_stats.jobs.max_memory} is too low" + @needs_docker @pytest.mark.docker @pytest.mark.online @@ -2299,6 +2339,46 @@ def test_download_structure(tmp_path: Path) -> None: ) +@needs_cwl +@pytest.mark.cwl +@pytest.mark.cwl_small +def test_cwl_resource_message_parsing_records_cpu_and_memory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + Resource monitor messages emitted from a container are translated into + Toil's extra CPU and memory accounting. + """ + from toil.lib import interpreter + from toil.lib.resources import ResourceMonitor + + message_file = tmp_path / "resources.tsv" + message_file.write_text( + "CPU\t1000000\n" + "Memory\t1024\n" + "CPU\t4000000\n" + "Memory\t2048\n", + encoding="utf-8", + ) + + recorded_memory_ki: list[int] = [] + recorded_cpu_seconds: list[float] = [] + monkeypatch.setattr( + ResourceMonitor, + "record_extra_memory", + lambda peak_ki: recorded_memory_ki.append(peak_ki), + ) + monkeypatch.setattr( + ResourceMonitor, + "record_extra_cpu", + lambda seconds: recorded_cpu_seconds.append(seconds), + ) + + interpreter.handle_message_file(str(message_file)) + + assert recorded_memory_ki == [2] + assert recorded_cpu_seconds == [3.0] + @needs_cwl @pytest.mark.cwl @pytest.mark.timeout(300) diff --git a/src/toil/test/cwl/waste_cpu_memory.cwl b/src/toil/test/cwl/waste_cpu_memory.cwl new file mode 100644 index 0000000000..340876f591 --- /dev/null +++ b/src/toil/test/cwl/waste_cpu_memory.cwl @@ -0,0 +1,22 @@ +cwlVersion: v1.2 +class: CommandLineTool + +doc: "Burn ~30s of CPU and allocate ~1 GiB of RAM inside a container, for stats monitoring tests." + +requirements: + - class: DockerRequirement + dockerPull: python:3-slim + +baseCommand: + - python3 + - -c + - | + import time + # Hold ~1 GiB in RAM for the duration of the run. + data = bytearray(1024 * 1024 * 1024) + end = time.monotonic() + 30 + while time.monotonic() < end: + pass + +inputs: [] +outputs: [] diff --git a/src/toil/wdl/wdltoil.py b/src/toil/wdl/wdltoil.py index 2f7010140d..883cc524af 100755 --- a/src/toil/wdl/wdltoil.py +++ b/src/toil/wdl/wdltoil.py @@ -133,6 +133,11 @@ from toil.lib.trs import resolve_workflow from toil.lib.url import URLAccess from toil.provisioners.clusterScaler import JobTooBigError +from toil.lib.interpreter import ( + INJECTED_MESSAGE_DIR, + add_injections, + handle_message_file, +) logger = logging.getLogger(__name__) @@ -3871,98 +3876,6 @@ def __init__( self._cache_key = cache_key self._mount_spec = mount_spec - ### - # Runtime code injection system - ### - - # WDL runtime code injected in the container communicates back to the rest - # of the runtime through files in this directory. - INJECTED_MESSAGE_DIR = ".toil_wdl_runtime" - - def add_injections(self, command_string: str, task_container: TaskContainer) -> str: - """ - Inject extra Bash code from the Toil WDL runtime into the command for the container. - - Currently doesn't implement the MiniWDL plugin system, but does add - resource usage monitoring to Docker containers. - """ - - parts = [] - - if isinstance(task_container, SwarmContainer): - # We're running on Docker Swarm, so we need to monitor CPU usage - # and so on from inside the container, since it won't be attributed - # to Toil child processes in the leader's self-monitoring. - # TODO: Mount this from a file Toil installs instead or something. - script = textwrap.dedent( - """\ - function _toil_resource_monitor () { - # Turn off error checking and echo in here - set +ex - MESSAGE_DIR="${1}" - mkdir -p "${MESSAGE_DIR}" - - function sample_cpu_usec() { - if [[ -f /sys/fs/cgroup/cpu.stat ]] ; then - awk '{ if ($1 == "usage_usec") {print $2} }' /sys/fs/cgroup/cpu.stat - elif [[ -f /sys/fs/cgroup/cpuacct/cpuacct.stat ]] ; then - echo $(( $(head -n 1 /sys/fs/cgroup/cpuacct/cpuacct.stat | cut -f2 -d' ') * 10000 )) - fi - } - - function sample_memory_bytes() { - if [[ -f /sys/fs/cgroup/memory.stat ]] ; then - awk '{ if ($1 == "anon") { print $2 } }' /sys/fs/cgroup/memory.stat - elif [[ -f /sys/fs/cgroup/memory/memory.stat ]] ; then - awk '{ if ($1 == "total_rss") { print $2 } }' /sys/fs/cgroup/memory/memory.stat - fi - } - - while true ; do - printf "CPU\\t" >> ${MESSAGE_DIR}/resources.tsv - sample_cpu_usec >> ${MESSAGE_DIR}/resources.tsv - printf "Memory\\t" >> ${MESSAGE_DIR}/resources.tsv - sample_memory_bytes >> ${MESSAGE_DIR}/resources.tsv - sleep 1 - done - } - """ - ) - parts.append(script) - # Launch in a subshell so that it doesn't interfere with Bash "wait" in the main shell - parts.append(f"(_toil_resource_monitor {self.INJECTED_MESSAGE_DIR} &)") - - if isinstance(task_container, SwarmContainer) and platform.system() == "Darwin": - # With gRPC FUSE file sharing, files immediately downloaded before - # being mounted may appear as size 0 in the container due to a race - # condition. Check for this and produce an approperiate error. - - script = textwrap.dedent( - """\ - function _toil_check_size () { - TARGET_FILE="${1}" - GOT_SIZE="$(stat -c %s "${TARGET_FILE}")" - EXPECTED_SIZE="${2}" - if [[ "${GOT_SIZE}" != "${EXPECTED_SIZE}" ]] ; then - echo >&2 "Toil Error:" - echo >&2 "File size visible in container for ${TARGET_FILE} is size ${GOT_SIZE} but should be size ${EXPECTED_SIZE}" - echo >&2 "Are you using gRPC FUSE file sharing in Docker Desktop?" - echo >&2 "It doesn't work: see ." - exit 1 - fi - } - """ - ) - parts.append(script) - for host_path, job_path in task_container.input_path_map.items(): - expected_size = os.path.getsize(host_path) - if expected_size != 0: - parts.append(f'_toil_check_size "{job_path}" {expected_size}') - - parts.append(command_string) - - return "\n".join(parts) - def handle_injection_messages( self, outputs_library: ToilWDLStdLibTaskOutputs ) -> None: @@ -3971,60 +3884,11 @@ def handle_injection_messages( """ message_files = outputs_library._glob( - WDL.Value.String(os.path.join(self.INJECTED_MESSAGE_DIR, "*")) + WDL.Value.String(os.path.join(INJECTED_MESSAGE_DIR, "*")) ) logger.debug("Handling message files: %s", message_files) for message_file in message_files.value: - self.handle_message_file(message_file.value) - - def handle_message_file(self, file_path: str) -> None: - """ - Handle a message file received from in-container injected code. - - Takes the host-side path of the file. - """ - if os.path.basename(file_path) == "resources.tsv": - # This is a TSV of resource usage info. - first_cpu_usec: int | None = None - last_cpu_usec: int | None = None - max_memory_bytes: int | None = None - - for line in open(file_path): - if not line.endswith("\n"): - # Skip partial lines - continue - # For each full line we got - parts = line.strip().split("\t") - if len(parts) != 2: - # Skip odd-shaped lines - continue - if parts[0] == "CPU": - # Parse CPU usage - cpu_usec = int(parts[1]) - # Update summary stats - if first_cpu_usec is None: - first_cpu_usec = cpu_usec - last_cpu_usec = cpu_usec - elif parts[0] == "Memory": - # Parse memory usage - memory_bytes = int(parts[1]) - # Update summary stats - if max_memory_bytes is None or max_memory_bytes < memory_bytes: - max_memory_bytes = memory_bytes - - if max_memory_bytes is not None: - logger.info( - "Container used at about %s bytes of memory at peak", - max_memory_bytes, - ) - # Treat it as if used by a child process - ResourceMonitor.record_extra_memory(max_memory_bytes // 1024) - if last_cpu_usec is not None: - assert first_cpu_usec is not None - cpu_seconds = (last_cpu_usec - first_cpu_usec) / 1000000 - logger.info("Container used about %s seconds of CPU time", cpu_seconds) - # Treat it as if used by a child process - ResourceMonitor.record_extra_cpu(cpu_seconds) + handle_message_file(message_file.value) ### # Helper functions to work out what containers runtime we can use @@ -4501,8 +4365,17 @@ def get_path_in_container(inode: AnyINode) -> AnyINode | None: .value ) - # Do any command injection we might need to do - command_string = self.add_injections(command_string, task_container) + # Do command injection for docker container resource monitoring + # when a docker container is used + if isinstance(task_container, SwarmContainer): + file_mounts = task_container.input_path_map.items() + # For WDL we're allowed to drop things in the current working + # directory. + command_string = add_injections( + command_string, + file_mounts, + INJECTED_MESSAGE_DIR + ) # Grab the standard out and error paths. MyPy complains if we call # them because in the current MiniWDL version they are untyped. diff --git a/src/toil/worker.py b/src/toil/worker.py index b9529db792..9e866d5b01 100644 --- a/src/toil/worker.py +++ b/src/toil/worker.py @@ -19,6 +19,7 @@ import os import pickle import random +import re import shutil import signal import socket @@ -740,10 +741,24 @@ def blockFn() -> bool: failure_exit_code = e.code else: try: - from WDL.runtime.error import CommandFailed + from WDL.runtime.error import CommandFailed, Interrupted if isinstance(e, CommandFailed): failure_exit_code = e.exit_status + elif isinstance(e, Interrupted): + # Try to work around + # https://github.com/chanzuckerberg/miniwdl/issues/902 at + # least enough to pass the tests. + # If we observed a container supposedly running with a + # nonzero exit code, MiniWDL will bail out with an + # Interrupted with "exit code = {exit_code}" in the + # message. We need to parse it out and use it to at least + # cover the cases where that code was indeed supposed to be + # a failure and needs to make it to the leader. + found = re.search("exit code = ([0-9]+)", str(e)) + if found: + failure_exit_code = int(found.group(1)) + except ImportError: # WDL dependency not available pass