Add metrics logging

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.

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()

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,16 +144,22 @@ 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__(
         self,
         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 "

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

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

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()