diff --git a/README.md b/README.md index b51d89f..cccd5fc 100644 --- a/README.md +++ b/README.md @@ -55,8 +55,12 @@ sentieon-cli ... ## Global arguments The `sentieon-cli` supports the following global arguments: -- `--verbose`: verbose logging. -- `--debug`: debugging mode for more verbose logging. +- `--verbose` (`-v`): verbose logging. This is the default. +- `--quiet` (`-q`): only log warnings and errors. +- `--debug` (`-d`): debugging mode for more verbose logging. Takes precedence over `--verbose` and `--quiet`. + +## Logging +Each run writes its log files to a directory next to the output VCF, named after the output file with the `.vcf.gz` suffix replaced by `_logs`, so `sample.vcf.gz` produces `sample_logs/`. The directory records the invocation in `command.txt`, the pipeline's own messages in `run.log`, and the output of each tool under `task_logs/`. Every pipeline accepts a `--log_dir` argument to write these files elsewhere. Rerunning a pipeline with the same output overwrites the logs of the previous run. ## Supported pipelines - [**DNAscope**](https://support.sentieon.com/docs/sentieon_cli/#dnascope) - DNAscope pipeline implementation for germline SNV and indel calling from short read data. diff --git a/sentieon_cli/__init__.py b/sentieon_cli/__init__.py index 1de7890..59c9971 100644 --- a/sentieon_cli/__init__.py +++ b/sentieon_cli/__init__.py @@ -1,3 +1,5 @@ +import argparse + from . import argh_parser from .dnascope import DNAscopePipeline from .dnascope_hybrid import DNAscopeHybridPipeline @@ -7,26 +9,41 @@ from .util import __version__ -def main(): - """main entry point for this project""" - parser = argh_parser.CustomArgparseParser() +def add_logging_args(parser: argparse.ArgumentParser) -> None: + """Add the console verbosity flags""" parser.add_argument( "-v", "--verbose", - help="Verbose logging", - action="store_const", - dest="loglevel", - const="INFO", - default="INFO", + help="Verbose logging (the default)", + action="store_true", + ) + parser.add_argument( + "-q", + "--quiet", + help="Only log warnings and errors", + action="store_true", ) parser.add_argument( "-d", "--debug", help="Print debugging info", - action="store_const", - dest="loglevel", - const="DEBUG", + action="store_true", ) + + +def resolve_loglevel(args: argparse.Namespace) -> str: + """The console log level implied by the verbosity flags""" + if args.debug: + return "DEBUG" + if args.quiet and not args.verbose: + return "WARNING" + return "INFO" + + +def main(): + """main entry point for this project""" + parser = argh_parser.CustomArgparseParser() + add_logging_args(parser) parser.add_argument( "--version", action="version", @@ -60,6 +77,7 @@ def main(): dnascope_pangenome_subparser.set_defaults(pipeline=pipeline.main) args = parser.parse_args() + args.loglevel = resolve_loglevel(args) # Job ids must be unique for the whole run, which may execute more than # one DAG, so numbering restarts here rather than per DAG. Job.reset_ids() diff --git a/sentieon_cli/executor.py b/sentieon_cli/executor.py index 3b9ad58..2fecf6f 100644 --- a/sentieon_cli/executor.py +++ b/sentieon_cli/executor.py @@ -1,9 +1,11 @@ """Execute jobs""" import asyncio +import concurrent.futures import contextlib import os import pathlib +import resource import signal import sys import threading @@ -43,6 +45,22 @@ def _log_tail(path: pathlib.Path, lines: int = TAIL_LINES) -> List[str]: return data.decode("utf-8", errors="replace").splitlines()[-lines:] +def _maxrss_bytes(ru: resource.struct_rusage) -> int: + """Return a child's peak resident set size in bytes. + + ``ru_maxrss`` is already bytes on macOS; every other platform we run on + reports kibibytes. + """ + if sys.platform == "darwin": + return int(ru.ru_maxrss) + return int(ru.ru_maxrss) * 1024 + + +def _mib(nbytes: int) -> str: + """Render a byte count as MiB for the human-readable log lines.""" + return f"{nbytes / (1024 * 1024):.1f} MiB" + + def _signal_procs(context: Context, sig: int) -> None: """Send a signal to every live sub-process in a context.""" for subcommand in context.commands: @@ -126,6 +144,10 @@ class AsyncExecutor(BaseExecutor, ABC): Set ``install_signal_handlers=True`` to install handlers (on the main thread only) for the duration of a run; the previous handlers are restored afterwards, so a caller's signal state is left untouched. + + ``thread_pool_size`` sizes the loop's default thread pool for the run; + ``None`` leaves the loop's own default alone, which is what an executor + that spawns no processes wants. """ def __init__( @@ -133,9 +155,11 @@ def __init__( scheduler: BaseScheduler, *, install_signal_handlers: bool = False, + thread_pool_size: Optional[int] = None, ) -> None: super().__init__(scheduler) self.install_signal_handlers = install_signal_handlers + self.thread_pool_size = thread_pool_size self.start_new_jobs = True def _install_signal_handlers( @@ -201,10 +225,34 @@ async def _execute(self) -> None: """Execute jobs from the DAG""" self.jobs_with_errors = [] loop = asyncio.get_running_loop() + if self.thread_pool_size is not None: + # Threads are created lazily, so a generous cap costs nothing + # unless it is used, and asyncio.run() shuts the executor down on + # exit. + loop.set_default_executor( + concurrent.futures.ThreadPoolExecutor( + max_workers=self.thread_pool_size, + thread_name_prefix="sentieon-wait", + ) + ) stop_event = asyncio.Event() restore = self._install_signal_handlers(loop, stop_event) try: - await self._drive(stop_event) + try: + await self._drive(stop_event) + except BaseException: + # Tear the children down before the exception unwinds into + # asyncio.run's Runner.close, which joins the wait threads -- + # and those threads are blocked in wait4 on live children. + # BaseException so cancellation and KeyboardInterrupt are + # covered too. + self.start_new_jobs = False + try: + await self._shutdown() + except Exception as exc: + # A failed teardown must not mask the original error. + logger.error("teardown after failure also failed: %s", exc) + raise if stop_event.is_set(): await self._shutdown() finally: @@ -220,7 +268,14 @@ async def _shutdown(self) -> None: class LocalExecutor(AsyncExecutor): - """Run jobs locally as async subprocesses.""" + """Run jobs locally as sub-processes, driven by an asyncio loop. + + ``thread_pool_size`` defaults high because the loop's default pool serves + two kinds of blocking work at once: one thread per in-flight ``wait()`` + and the proc-sub FIFO opens. A FIFO open queued behind saturated wait + threads would deadlock the run, so the pool must comfortably exceed the + number of processes a run keeps in flight. + """ def __init__( self, @@ -229,10 +284,12 @@ def __init__( install_signal_handlers: bool = False, shutdown_grace_period: float = 10.0, run_logs: Optional[RunLogs] = None, + thread_pool_size: int = 512, ) -> None: super().__init__( scheduler, install_signal_handlers=install_signal_handlers, + thread_pool_size=thread_pool_size, ) self.shutdown_grace_period = shutdown_grace_period self.run_logs = run_logs @@ -266,7 +323,7 @@ async def run_job(self, job: Job) -> None: ( job, context, - asyncio.create_task(proc.wait()), + asyncio.create_task(proc.async_wait()), start_time, ) ) @@ -346,7 +403,7 @@ async def _terminate_context(self, context: Context) -> None: if not live: return _signal_procs(context, signal.SIGTERM) - waits = [asyncio.create_task(proc.wait()) for proc in live] + waits = [asyncio.create_task(proc.async_wait()) for proc in live] await asyncio.wait(waits, timeout=self.shutdown_grace_period) _kill_survivors(context) await asyncio.wait(waits) @@ -388,6 +445,14 @@ async def _drive(self, stop_event: asyncio.Event) -> None: # Check if the command failed cmd_failed = False failures: List[Tuple[Command, int]] = [] + # Kernel resource usage, aggregated over the job's + # processes. RSS is the per-process peak rather than a + # sum: concurrent pipeline stages peak at different + # times, so a sum would overstate the job. + have_rusage = False + job_utime = 0.0 + job_stime = 0.0 + job_maxrss = 0 for subcommand in context.commands: if not subcommand.proc: logger.error( @@ -397,8 +462,36 @@ async def _drive(self, stop_event: asyncio.Event) -> None: cmd_failed = True continue ret = ( - await subcommand.proc.wait() + await subcommand.proc.async_wait() ) # Wait on all sub-commands + # Reported for every reaped process, whatever the + # job's outcome; signal-killed children have rusage + # too. + rusage = subcommand.proc.rusage + if rusage is not None: + rss = _maxrss_bytes(rusage) + have_rusage = True + job_utime += rusage.ru_utime + job_stime += rusage.ru_stime + job_maxrss = max(job_maxrss, rss) + logger.debug( + "rusage for %s [pid %d, %s]: utime=%.2fs " + "stime=%.2fs maxrss=%.1fMiB minflt=%d " + "majflt=%d inblock=%d oublock=%d " + "nvcsw=%d nivcsw=%d", + job, + subcommand.proc.pid, + subcommand, + rusage.ru_utime, + rusage.ru_stime, + rss / (1024 * 1024), + rusage.ru_minflt, + rusage.ru_majflt, + rusage.ru_inblock, + rusage.ru_oublock, + rusage.ru_nvcsw, + rusage.ru_nivcsw, + ) # A subcommand killed by SIGPIPE is not a failure: # it was writing to a pipe whose reader exited early # (e.g. `... | head`), which is normal in default @@ -448,6 +541,16 @@ async def _drive(self, stop_event: asyncio.Event) -> None: ) self.jobs_with_errors.append(job) self.start_new_jobs = False + elif have_rusage: + logger.info( + "Finished command in %.2fs (user %.1fs, " + "sys %.1fs, max proc RSS %s): %s", + total_seconds, + job_utime, + job_stime, + _mib(job_maxrss), + job.shell, + ) else: logger.info( f"Finished command in " diff --git a/sentieon_cli/job.py b/sentieon_cli/job.py index bb893bb..a6a15f7 100644 --- a/sentieon_cli/job.py +++ b/sentieon_cli/job.py @@ -5,6 +5,7 @@ from typing import Dict, Optional from .shell_pipeline import Pipeline +from .util import sanitize class Job: @@ -44,8 +45,11 @@ def __init__( self.threads = threads self.resources = {} if resources is None else resources self.task_name = task_name - count = Job._id_counters.get(name, 0) + 1 - Job._id_counters[name] = count + # Log file names are sanitized, so ids must stay unique after + # sanitization; the id itself keeps the readable name. + key = sanitize(name) + count = Job._id_counters.get(key, 0) + 1 + Job._id_counters[key] = count self.job_id = f"{name}-{count}" @classmethod diff --git a/sentieon_cli/logging.py b/sentieon_cli/logging.py index df0a48a..fe12b53 100644 --- a/sentieon_cli/logging.py +++ b/sentieon_cli/logging.py @@ -4,7 +4,8 @@ Handlers live only on the package logger; the loggers returned by ``get_logger`` are plain children that propagate to it. The package logger is always at DEBUG so verbosity can be enforced per handler: the console handler -follows ``-v``/``-d`` while the ``run.log`` file handler always records DEBUG. +follows ``-v``/``-q``/``-d`` while the ``run.log`` file handler always records +DEBUG. """ import logging diff --git a/sentieon_cli/pipeline.py b/sentieon_cli/pipeline.py index 65231aa..55f267a 100644 --- a/sentieon_cli/pipeline.py +++ b/sentieon_cli/pipeline.py @@ -126,30 +126,43 @@ def __init__(self) -> None: self.run_logs: Optional[RunLogs] = None def setup_logging(self, args: argparse.Namespace) -> None: - """Configure the console handler and the run's log directory""" + """Configure console logging""" self.logger = get_logger(__name__) set_console_level(args.loglevel) + def start_run_logs(self) -> None: + """Create the run's log directory and start writing `run.log`. + + Called after `validate`, so a run rejected for an invalid output path + neither creates directories nor clobbers a previous run's logs. + """ # File logging is skipped for dry-runs and when there is nothing to # derive a log directory from (a bare pipeline, as used by tests). - if not self.dry_run and ( - self.log_dir is not None or self.output_vcf is not None - ): - log_dir = self.log_dir - if log_dir is None: - # Check the suffix before deriving the directory name so an - # invalid output path cannot create a garbage-named log dir. - self.validate_output_suffix() - log_dir = pathlib.Path( - str(self.output_vcf).removesuffix(".vcf.gz") + "_logs" - ) - self.run_logs = RunLogs(log_dir) - self.run_logs.setup() + if self.dry_run or (self.log_dir is None and self.output_vcf is None): + self.logger.info("Starting sentieon-cli version: %s", __version__) + return + + log_dir = self.log_dir + if log_dir is None: + # Defensive: `validate` has already checked the output path, but + # not every pipeline's validation covers the suffix. + self.validate_output_suffix() + log_dir = pathlib.Path( + str(self.output_vcf).removesuffix(".vcf.gz") + "_logs" + ) + run_logs = RunLogs(log_dir) + try: + run_logs.setup() + except OSError as exc: + self.logger.error( + "Could not prepare the log directory %s: %s", log_dir, exc + ) + sys.exit(2) + self.run_logs = run_logs # After the file handler is attached, so the banner reaches run.log self.logger.info("Starting sentieon-cli version: %s", __version__) - if self.run_logs: - self.logger.info("Writing logs to: %s", self.run_logs.log_dir) + self.logger.info("Writing logs to: %s", self.run_logs.log_dir) def log_completion(self, success: bool, start_time: float) -> None: """Report the outcome and duration of the run""" @@ -171,6 +184,7 @@ def main(self, args: argparse.Namespace) -> None: success = False try: self.validate() + self.start_run_logs() self.configure() tmp_dir_str = tmp() diff --git a/sentieon_cli/run_logs.py b/sentieon_cli/run_logs.py index 5a7f7c3..adee9db 100644 --- a/sentieon_cli/run_logs.py +++ b/sentieon_cli/run_logs.py @@ -8,30 +8,22 @@ import datetime import logging import pathlib -import re import shlex import shutil import sys from typing import IO, List, Optional, TYPE_CHECKING from .logging import add_file_handler, remove_file_handler -from .util import __version__ +from .util import __version__, sanitize if TYPE_CHECKING: from .job import Job from .shell_pipeline import Command -_UNSAFE = re.compile(r"[^A-Za-z0-9._-]") - _STDERR = "stderr" _STDOUT = "stdout" -def sanitize(component: str) -> str: - """Restrict a path component to filesystem-safe characters""" - return _UNSAFE.sub("-", component) - - @dataclasses.dataclass class _ProcessLog: """A log file opened for one spawned process""" diff --git a/sentieon_cli/sentieon_pangenome.py b/sentieon_cli/sentieon_pangenome.py index 50f23af..bb985db 100644 --- a/sentieon_cli/sentieon_pangenome.py +++ b/sentieon_cli/sentieon_pangenome.py @@ -284,6 +284,7 @@ def main(self, args: argparse.Namespace) -> None: self.logger.debug("VCF contigs are: %s", self.pop_vcf_contigs) self.validate() + self.start_run_logs() self.shards = determine_shards_from_fai( self.fai_data, 10 * 1000 * 1000 ) diff --git a/sentieon_cli/shell_pipeline.py b/sentieon_cli/shell_pipeline.py index fa6aa7a..d2d1277 100644 --- a/sentieon_cli/shell_pipeline.py +++ b/sentieon_cli/shell_pipeline.py @@ -10,10 +10,21 @@ import fcntl import os import pathlib +import resource +import subprocess import tempfile import shlex from abc import ABC, abstractmethod -from typing import Any, Dict, IO, List, Optional, TYPE_CHECKING, Union +from typing import ( + Any, + Dict, + IO, + List, + Optional, + Tuple, + TYPE_CHECKING, + Union, +) from .logging import get_logger @@ -124,6 +135,14 @@ async def cleanup(self) -> None: finally: for fd in unblock_fds: os.close(fd) + # asyncio's child watcher reaped children nobody awaited; Popen does + # not, so sweep up already-exited processes here to avoid zombies for + # callers that only await the final pipeline stage. poll() is + # non-blocking: a process still running is left to its owner. + for command in self.commands: + proc = command.proc + if proc is not None and proc.returncode is None: + proc.poll() for fh in self.file_handles: fh.close() self.temp_dir.cleanup() @@ -135,6 +154,40 @@ async def cleanup(self) -> None: raise exc +class RusagePopen(subprocess.Popen[bytes]): + """A Popen that records the child's rusage when it is reaped. + + Only the ``wait``/``async_wait`` path reaps through ``_try_wait`` and + sets ``rusage``; a reap via ``poll`` -- or ``send_signal``/``kill``, + which poll internally -- goes through the base class's machinery and + leaves ``rusage`` None. The executor waits every process it reports + metrics for, so those paths only collect processes whose rusage is + never read. + """ + + rusage: Optional[resource.struct_rusage] = None + + def _try_wait(self, wait_flags: int) -> Tuple[int, int]: + # Mirrors the base class, substituting wait4 for waitpid so the + # child's rusage is captured at reap time. Callers hold + # self._waitpid_lock (base-class contract). + try: + pid, sts, res = os.wait4(self.pid, wait_flags) + except ChildProcessError: + pid, sts = self.pid, 0 + else: + if pid == self.pid: + self.rusage = res + return (pid, sts) + + async def async_wait(self) -> int: + """Await the child's exit without blocking the event loop.""" + if self.returncode is not None: + # Already reaped; no need for a pool thread. + return self.returncode + return await asyncio.to_thread(self.wait) + + class ShellNode(ABC): """Abstract base class for any node in the shell syntax tree.""" @@ -152,7 +205,7 @@ def __init__( self.executable = executable self.args = list(args) self.fail_ok = fail_ok - self.proc: Union[asyncio.subprocess.Process, None] = None + self.proc: Optional[RusagePopen] = None self.exec_kwargs: Dict[str, Any] = exec_kwargs if exec_kwargs else {} async def run( @@ -161,7 +214,7 @@ async def run( stdin: Union[IO[Any], int, None] = None, stdout: Union[IO[Any], int, None] = None, stderr: Union[IO[Any], int, None] = None, - ) -> asyncio.subprocess.Process: + ) -> RusagePopen: # 1. Resolve Arguments (Handle Process Substitutions) final_args = [self.executable] @@ -188,8 +241,8 @@ async def run( # 3. Run the process # We allow the caller to wait for this process - self.proc = await asyncio.create_subprocess_exec( - *final_args, + self.proc = RusagePopen( + final_args, stdin=stdin, stdout=stdout, stderr=stderr, @@ -266,7 +319,7 @@ async def run( stdin: Union[IO[Any], int, None] = None, stdout: Union[IO[Any], int, None] = None, stderr: Union[IO[Any], int, None] = None, - ) -> asyncio.subprocess.Process: + ) -> RusagePopen: # We cannot have handles from both files and to run(). if self.file_input is not None and stdin is not None: raise ValueError( @@ -429,7 +482,7 @@ async def run_inner() -> None: context, stdout=write_handle, ) - await proc.wait() + await proc.async_wait() finally: write_handle.close() @@ -467,7 +520,7 @@ async def run_inner() -> None: if context.closing: return # cleanup unblocked us; do not spawn proc = await self.node.run(context, stdin=read_handle) - await proc.wait() + await proc.async_wait() finally: read_handle.close() diff --git a/sentieon_cli/util.py b/sentieon_cli/util.py index 68021ea..fe9cc73 100644 --- a/sentieon_cli/util.py +++ b/sentieon_cli/util.py @@ -31,6 +31,13 @@ NUMA_NODE_PAT = re.compile(r"^NUMA node. CPU\(s\):\s+(?P.*)$") READ_LENGTH_PAT = re.compile(r"SN\taverage length:\t(?P\d*)$") +_UNSAFE = re.compile(r"[^A-Za-z0-9._-]") + + +def sanitize(component: str) -> str: + """Restrict a path component to filesystem-safe characters""" + return _UNSAFE.sub("-", component) + def tmp(): """Create a temporary directory for the current process.""" diff --git a/tests/integration/test_executor.py b/tests/integration/test_executor.py index 41ce529..d0ea7f5 100644 --- a/tests/integration/test_executor.py +++ b/tests/integration/test_executor.py @@ -8,6 +8,7 @@ import sys import tempfile import threading +import time import pytest @@ -475,6 +476,58 @@ def __init__(self) -> None: assert proc.returncode is not None +def test_error_in_the_run_loop_terminates_running_jobs(monkeypatch): + """An exception escaping the run loop must tear the children down first. + + Otherwise it unwinds into asyncio.run's shutdown, which joins the wait + threads -- and those are blocked in wait4 on children nobody signalled, + so the run hangs until they exit on their own (regression: a `sleep 30` + held the CLI for 30s and was never terminated).""" + import sentieon_cli.executor as executor_mod + + contexts = [] + real_context = executor_mod.Context + + class RecordingContext(real_context): # type: ignore[valid-type,misc] + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + contexts.append(self) + + monkeypatch.setattr(executor_mod, "Context", RecordingContext) + + dag = DAG() + dag.add_job(Job(Pipeline(Command("sleep", "30")), "slow", task_name="t")) + dag.add_job(Job(Pipeline(Command("true")), "quick", task_name="t")) + + scheduler = ThreadScheduler(dag, 2) + + def boom(job): + # Fires when the quick job finishes, with the slow one still running + raise RuntimeError("scheduler exploded") + + monkeypatch.setattr(scheduler, "job_finished", boom) + + executor = LocalExecutor(scheduler, shutdown_grace_period=0.5) + start = time.monotonic() + with pytest.raises(RuntimeError): + executor.execute() + elapsed = time.monotonic() - start + + assert elapsed < 15, "the run waited for the sleep instead of killing it" + spawned = [sub.proc for c in contexts for sub in c.commands if sub.proc] + assert spawned, "expected both jobs to have spawned" + for proc in spawned: + assert proc.returncode is not None + + +def test_thread_pool_size_is_configurable(): + """The wait/FIFO-open pool cap is a parameter, not a hidden constant.""" + scheduler = ThreadScheduler(DAG(), 1) + + assert LocalExecutor(scheduler).thread_pool_size == 512 + assert LocalExecutor(scheduler, thread_pool_size=8).thread_pool_size == 8 + + def test_sigpipe_producer_does_not_fail_job(tmp_path): """A producer killed by SIGPIPE when a consumer exits early (e.g. `seq | head -1`) produces the right output and must not fail the job diff --git a/tests/unit/test_cli_args.py b/tests/unit/test_cli_args.py new file mode 100644 index 0000000..28909ed --- /dev/null +++ b/tests/unit/test_cli_args.py @@ -0,0 +1,40 @@ +""" +Unit tests for the top-level CLI arguments +""" + +import argparse +import os +import sys + +import pytest + +# Add the parent directory to the path so we can import sentieon_cli +sys.path.insert( + 0, + os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")), +) + +from sentieon_cli import add_logging_args, resolve_loglevel # noqa: E402 + + +def _loglevel(argv): + parser = argparse.ArgumentParser() + add_logging_args(parser) + return resolve_loglevel(parser.parse_args(argv)) + + +@pytest.mark.parametrize( + "argv,expected", + [ + ([], "INFO"), + (["-v"], "INFO"), + (["-q"], "WARNING"), + (["-d"], "DEBUG"), + # `-d` wins wherever it appears; it used to be undone by a later + # `-v` because both flags wrote to one destination. + (["-d", "-v"], "DEBUG"), + (["-q", "-v"], "INFO"), + ], +) +def test_the_verbosity_flags_resolve_to_a_console_level(argv, expected): + assert _loglevel(argv) == expected diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py index 4b37a58..f77205e 100644 --- a/tests/unit/test_job.py +++ b/tests/unit/test_job.py @@ -16,6 +16,7 @@ from sentieon_cli.dag import DAG # noqa: E402 from sentieon_cli.job import Job # noqa: E402 from sentieon_cli.shell_pipeline import Command, Pipeline # noqa: E402 +from sentieon_cli.util import sanitize # noqa: E402 def _job(name, arg, task_name="test"): @@ -45,6 +46,17 @@ def test_each_name_has_its_own_counter(): assert second.job_id == "dedup-2" +def test_names_differing_only_in_unsafe_characters_share_a_counter(): + # Log file names are sanitized, so ids that only differ in unsafe + # characters would name the same file and truncate each other. + first = _job("a b", "x") + second = _job("a-b", "y") + + assert (first.job_id, second.job_id) == ("a b-1", "a-b-2") + assert sanitize(first.job_id) == "a-b-1" + assert sanitize(second.job_id) == "a-b-2" + + def test_reset_ids_restarts_the_sequence(): assert _job("multiqc", "a").job_id == "multiqc-1" assert _job("multiqc", "b").job_id == "multiqc-2" diff --git a/tests/unit/test_job_logs.py b/tests/unit/test_job_logs.py index e438533..29395bc 100644 --- a/tests/unit/test_job_logs.py +++ b/tests/unit/test_job_logs.py @@ -13,12 +13,13 @@ from sentieon_cli.executor import TAIL_LINES, _log_tail # noqa: E402 from sentieon_cli.job import Job # noqa: E402 -from sentieon_cli.run_logs import RunLogs, sanitize # noqa: E402 +from sentieon_cli.run_logs import RunLogs # noqa: E402 from sentieon_cli.shell_pipeline import ( # noqa: E402 Command, Context, Pipeline, ) +from sentieon_cli.util import sanitize # noqa: E402 def _sink(tmp_path, name: str = "bwa", task_name: str = "alignment"): diff --git a/tests/unit/test_run_logs.py b/tests/unit/test_run_logs.py index 02b8eed..a9c84ff 100644 --- a/tests/unit/test_run_logs.py +++ b/tests/unit/test_run_logs.py @@ -49,6 +49,13 @@ def build_dag(self) -> DAG: raise RuntimeError("boom") +class _ValidatingPipeline(_DummyPipeline): + """A pipeline that checks its output path, as the real ones do.""" + + def validate(self) -> None: + self.validate_output_vcf() + + class _ErroredExecutor: """A stand-in executor that finished with failed jobs.""" @@ -81,6 +88,12 @@ def _args(loglevel: str = "INFO") -> argparse.Namespace: return argparse.Namespace(loglevel=loglevel) +def _start(pipeline: BasePipeline, loglevel: str = "INFO") -> None: + """Set up logging the way `main` does, minus the run itself.""" + pipeline.setup_logging(_args(loglevel)) + pipeline.start_run_logs() + + def test_module_loggers_delegate_to_the_package_logger(): module_logger = cli_logging.get_logger("sentieon_cli.example") @@ -92,7 +105,7 @@ def test_module_loggers_delegate_to_the_package_logger(): def test_default_log_dir_is_derived_from_the_output_vcf(tmp_path): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "sample.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs is not None assert pipeline.run_logs.log_dir == tmp_path / "sample_logs" @@ -102,7 +115,7 @@ def test_default_log_dir_only_strips_the_trailing_suffix(tmp_path): # `str.replace` would mangle a name with '.vcf.gz' in the middle. pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "a.vcf.gz.rerun.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs.log_dir == tmp_path / "a.vcf.gz.rerun_logs" @@ -111,7 +124,7 @@ def test_explicit_log_dir_skips_the_output_suffix_check(tmp_path): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "sample.bcf" pipeline.log_dir = tmp_path / "elsewhere" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs.log_dir == tmp_path / "elsewhere" assert (tmp_path / "elsewhere" / "run.log").is_file() @@ -122,10 +135,64 @@ def test_invalid_output_suffix_exits_before_creating_a_log_dir(tmp_path): pipeline.output_vcf = tmp_path / "sample.bcf" with pytest.raises(SystemExit) as excinfo: - pipeline.setup_logging(_args()) + _start(pipeline) + + assert excinfo.value.code == 2 + assert list(tmp_path.iterdir()) == [] + + +def test_an_invalid_output_path_creates_no_log_dir(tmp_path, messages): + # `validate` runs first, so a typo in the output path is rejected before + # anything is written -- including the log directory next to it. + pipeline = _ValidatingPipeline() + pipeline.output_vcf = tmp_path / "typo" / "sample.vcf.gz" + + with pytest.raises(SystemExit) as excinfo: + pipeline.main(_args()) assert excinfo.value.code == 2 + assert pipeline.run_logs is None assert list(tmp_path.iterdir()) == [] + assert any("status: failed" in msg for msg in messages) + + +def test_a_rejected_rerun_keeps_the_previous_runs_logs(tmp_path): + log_dir = tmp_path / "logs" + first = _ValidatingPipeline() + first.output_vcf = tmp_path / "sample.vcf.gz" + first.log_dir = log_dir + _start(first) + stale = first.run_logs.task_logs / "alignment" + stale.mkdir(parents=True) + (stale / "bwa-1.0.log").write_text("from the previous run") + first.run_logs.close() + run_log = (log_dir / "run.log").read_text() + + second = _ValidatingPipeline() + second.output_vcf = tmp_path / "typo" / "sample.vcf.gz" + second.log_dir = log_dir + with pytest.raises(SystemExit): + second.main(_args()) + + assert (stale / "bwa-1.0.log").read_text() == "from the previous run" + assert (log_dir / "run.log").read_text() == run_log + + +def test_a_log_dir_colliding_with_a_file_exits_cleanly(tmp_path, messages): + collision = tmp_path / "logs" + collision.write_text("not a directory") + pipeline = _DummyPipeline() + pipeline.log_dir = collision + + with pytest.raises(SystemExit) as excinfo: + _start(pipeline) + + assert excinfo.value.code == 2 + assert pipeline.run_logs is None + assert collision.read_text() == "not a directory" + assert any( + "Could not prepare the log directory" in msg for msg in messages + ) def test_setup_wipes_stale_task_logs_but_keeps_the_log_dir(tmp_path): @@ -160,7 +227,7 @@ def test_command_txt_records_the_invocation(tmp_path, monkeypatch): def test_run_log_captures_debug_while_the_console_stays_at_info(tmp_path): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "sample.vcf.gz" - pipeline.setup_logging(_args("INFO")) + _start(pipeline) logging.getLogger("sentieon_cli.example").debug("a debug record") run_log = pipeline.run_logs.run_log pipeline.run_logs.close() @@ -180,7 +247,7 @@ def test_repeated_setup_does_not_accumulate_handlers(tmp_path): for i in range(3): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / f"sample{i}.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert len(package_logger.handlers) == before + 1 pipeline.run_logs.close() @@ -193,7 +260,7 @@ def test_rerun_truncates_the_previous_run_log(tmp_path): for _ in range(2): pipeline = _DummyPipeline() pipeline.log_dir = log_dir - pipeline.setup_logging(_args()) + _start(pipeline) run_log = pipeline.run_logs.run_log pipeline.run_logs.close() @@ -205,7 +272,7 @@ def test_dry_run_skips_file_logging(tmp_path): pipeline = _DummyPipeline() pipeline.dry_run = True pipeline.output_vcf = tmp_path / "sample.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs is None assert list(tmp_path.iterdir()) == [] @@ -214,7 +281,7 @@ def test_dry_run_skips_file_logging(tmp_path): def test_bare_pipeline_setup_logging_creates_nothing(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) pipeline = _DummyPipeline() - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs is None assert list(tmp_path.iterdir()) == [] @@ -223,7 +290,7 @@ def test_bare_pipeline_setup_logging_creates_nothing(tmp_path, monkeypatch): def test_check_execution_names_the_task_log_dir(tmp_path): pipeline = _DummyPipeline() pipeline.log_dir = tmp_path / "logs" - pipeline.setup_logging(_args()) + _start(pipeline) job = Job(Pipeline(Command("false")), "boom", task_name="failing") with pytest.raises(DagExecutionError) as excinfo: diff --git a/tests/unit/test_rusage_metrics.py b/tests/unit/test_rusage_metrics.py new file mode 100644 index 0000000..a7c0b53 --- /dev/null +++ b/tests/unit/test_rusage_metrics.py @@ -0,0 +1,99 @@ +""" +Unit tests for the per-process resource-usage metrics +""" + +import logging +import os +import pathlib +import resource +import sys +from typing import List + +import pytest + +# Add the parent directory to the path so we can import sentieon_cli +sys.path.insert( + 0, + os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")), +) + +from sentieon_cli.dag import DAG # noqa: E402 +from sentieon_cli.executor import ( # noqa: E402 + LocalExecutor, + _maxrss_bytes, +) +from sentieon_cli.job import Job # noqa: E402 +from sentieon_cli.run_logs import RunLogs # noqa: E402 +from sentieon_cli.scheduler import ThreadScheduler # noqa: E402 +from sentieon_cli.shell_pipeline import Command, Pipeline # noqa: E402 + + +class _RecordingHandler(logging.Handler): + """Collects formatted messages from the package logger.""" + + def __init__(self) -> None: + super().__init__(logging.DEBUG) + self.messages: List[str] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.messages.append(record.getMessage()) + + +@pytest.fixture +def messages(): + """Capture the package logger, which does not propagate to the root.""" + handler = _RecordingHandler() + package_logger = logging.getLogger("sentieon_cli") + package_logger.addHandler(handler) + yield handler.messages + package_logger.removeHandler(handler) + + +def _run_one_job(log_dir: pathlib.Path) -> None: + """Run a single trivial job with its output captured under ``log_dir``""" + dag = DAG() + dag.add_job(Job(Pipeline(Command("true")), "metrics", task_name="metrics")) + run_logs = RunLogs(log_dir) + run_logs.create_dirs() + LocalExecutor(ThreadScheduler(dag, 1), run_logs=run_logs).execute() + + +def test_each_process_of_a_job_reports_its_rusage(tmp_path, messages): + _run_one_job(tmp_path / "logs") + + rusage_lines = [m for m in messages if m.startswith("rusage for ")] + assert len(rusage_lines) == 1 + line = rusage_lines[0] + assert "Job(metrics-1)" in line + for field in ("utime=", "stime=", "maxrss=", "minflt=", "nivcsw="): + assert field in line + + +def test_a_finished_job_reports_its_aggregated_metrics(tmp_path, messages): + _run_one_job(tmp_path / "logs") + + finished = [m for m in messages if m.startswith("Finished command in")] + assert len(finished) == 1 + assert "user " in finished[0] + assert "sys " in finished[0] + assert "max proc RSS" in finished[0] + + +def test_maxrss_is_reported_in_bytes(): + ru = resource.getrusage(resource.RUSAGE_SELF) + + maxrss = _maxrss_bytes(ru) + + assert isinstance(maxrss, int) + assert maxrss > 0 + + +def test_maxrss_units_follow_the_platform(monkeypatch): + """macOS reports bytes; everywhere else reports kibibytes.""" + ru = resource.getrusage(resource.RUSAGE_SELF) + + monkeypatch.setattr(sys, "platform", "darwin") + assert _maxrss_bytes(ru) == ru.ru_maxrss + + monkeypatch.setattr(sys, "platform", "linux") + assert _maxrss_bytes(ru) == ru.ru_maxrss * 1024 diff --git a/tests/unit/test_shell_pipeline.py b/tests/unit/test_shell_pipeline.py index 38ec2bd..a6d704c 100644 --- a/tests/unit/test_shell_pipeline.py +++ b/tests/unit/test_shell_pipeline.py @@ -6,6 +6,7 @@ import fcntl import os import pathlib +import signal import sys import tempfile @@ -34,7 +35,7 @@ async def test_simple_command(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await cmd.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() stdout_file.seek(0) output = stdout_file.read().strip() @@ -44,6 +45,29 @@ async def test_simple_command(): assert output == "hello world" assert proc.returncode == 0 + # wait4 captured the child's resource usage as it was reaped + assert proc.rusage is not None + assert proc.rusage.ru_maxrss > 0 + + +@pytest.mark.asyncio +async def test_a_signalled_command_reports_its_signal_and_its_rusage(): + """A child killed by a signal is still reaped through ``wait4``. + + This pins ``RusagePopen._try_wait``: if a future CPython changes that + private method's shape, the override stops being called and ``rusage`` + silently stays None while the negative return code still works. + """ + cmd = Command("sleep", "5") + context = Context() + + proc = await cmd.run(context) + proc.send_signal(signal.SIGTERM) + + assert await proc.async_wait() == -signal.SIGTERM + assert proc.rusage is not None + + await context.cleanup() @pytest.mark.asyncio @@ -56,7 +80,7 @@ async def test_simple_pipeline(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await pipeline.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() stdout_file.seek(0) output = stdout_file.read().strip() @@ -89,7 +113,7 @@ async def test_pipeline_with_file_io(): ) proc = await pipeline.run(context) - await proc.wait() + await proc.async_wait() with open(outfile_path, "r") as f: output = f.read().strip() @@ -114,7 +138,7 @@ async def test_input_process_substitution(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await cmd.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() stdout_file.seek(0) output = stdout_file.read().strip() @@ -149,7 +173,7 @@ async def test_output_process_substitution(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await main_pipeline.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() # Wait for background tasks from process substitution to finish await asyncio.gather(*context.tasks) @@ -198,10 +222,10 @@ async def test_pipe_size_enlarges_internal_pipes(tmp_path): pipe_size=target, ) proc = await pipeline.run(context) - await proc.wait() + await proc.async_wait() for sub in context.commands: if sub.proc: - await sub.proc.wait() + await sub.proc.async_wait() await context.cleanup() assert int(report.read_text()) == target @@ -347,7 +371,7 @@ async def test_cleanup_after_outer_exits_without_opening(): context = Context() cmd = Command("false", InputProcSub(Pipeline(Command("echo", "x")))) proc = await cmd.run(context) - await proc.wait() + await proc.async_wait() await asyncio.wait_for(context.cleanup(), timeout=10) @@ -364,7 +388,7 @@ async def test_cleanup_unblocks_multiple_procsubs(): OutputProcSub(Pipeline(Command("cat"))), ) proc = await cmd.run(context) - await proc.wait() + await proc.async_wait() await asyncio.wait_for(context.cleanup(), timeout=10) @@ -380,7 +404,7 @@ async def test_cleanup_still_raises_inner_launch_failure(): "cat", InputProcSub(Pipeline(Command("no_such_cmd_zzz"))) ) proc = await cmd.run(context) - await proc.wait() + await proc.async_wait() with pytest.raises(FileNotFoundError): await asyncio.wait_for(context.cleanup(), timeout=10) @@ -408,7 +432,7 @@ async def test_cleanup_does_not_stall_running_inner_writer(): await asyncio.sleep(0.01) await asyncio.wait_for(context.cleanup(), timeout=10) - assert await proc.wait() == 0 + assert await proc.async_wait() == 0 @pytest.mark.asyncio @@ -417,7 +441,7 @@ async def test_cleanup_twice_is_safe(): context = Context() cmd = Command("cat", InputProcSub(Pipeline(Command("echo", "hello")))) proc = await cmd.run(context, stdout=asyncio.subprocess.DEVNULL) - await proc.wait() + await proc.async_wait() await asyncio.wait_for(context.cleanup(), timeout=10) await asyncio.wait_for(context.cleanup(), timeout=10)