diff --git a/sentieon_cli/__init__.py b/sentieon_cli/__init__.py index 3fcb34b..1de7890 100644 --- a/sentieon_cli/__init__.py +++ b/sentieon_cli/__init__.py @@ -2,6 +2,7 @@ from .dnascope import DNAscopePipeline from .dnascope_hybrid import DNAscopeHybridPipeline from .dnascope_longread import DNAscopeLRPipeline +from .job import Job from .sentieon_pangenome import SentieonPangenome from .util import __version__ @@ -16,7 +17,7 @@ def main(): action="store_const", dest="loglevel", const="INFO", - default="WARNING", + default="INFO", ) parser.add_argument( "-d", @@ -59,6 +60,9 @@ def main(): dnascope_pangenome_subparser.set_defaults(pipeline=pipeline.main) args = parser.parse_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() args.pipeline(args) diff --git a/sentieon_cli/base_pangenome.py b/sentieon_cli/base_pangenome.py index b15dfc3..7284de6 100644 --- a/sentieon_cli/base_pangenome.py +++ b/sentieon_cli/base_pangenome.py @@ -129,6 +129,7 @@ def build_kmc_job( ), "kmc", job_threads, + task_name="kmer-counting", ) return kmc_job @@ -150,6 +151,7 @@ def build_ploidy_job( ), "estimate-ploidy", 0, + task_name="ploidy", ) return ploidy_job diff --git a/sentieon_cli/dnascope.py b/sentieon_cli/dnascope.py index c4fa230..70a2e10 100644 --- a/sentieon_cli/dnascope.py +++ b/sentieon_cli/dnascope.py @@ -517,6 +517,7 @@ def sr_align_inputs(self) -> Tuple[List[pathlib.Path], Set[Job], Job]: ), f"bam-align-{i}", self.cores, + task_name="alignment", ) res.append(out_aln) jobs.add(job) @@ -532,6 +533,7 @@ def sr_align_inputs(self) -> Tuple[List[pathlib.Path], Set[Job], Job]: ), "rm-bam-aln", 0, + task_name="cleanup", ) return (res, jobs, rm_job) @@ -613,6 +615,7 @@ def sr_align_fastq( f"bam-align-{i}-{j}", split_cores, resources={f"node{j}": 1}, + task_name="alignment", ) res.append(out_aln) jobs.add(job) @@ -628,6 +631,7 @@ def sr_align_fastq( ), "rm-fq-aln", 0, + task_name="cleanup", ) return (res, jobs, rm_job) @@ -716,6 +720,7 @@ def dedup_and_metrics( Pipeline(Command(*driver.build_cmd())), "locuscollector", self.cores, + task_name="dedup", ) if self.sr_duplicate_marking == "none": @@ -743,7 +748,10 @@ def dedup_and_metrics( ) ) dedup_job = Job( - Pipeline(Command(*driver.build_cmd())), "dedup", self.cores + Pipeline(Command(*driver.build_cmd())), + "dedup", + self.cores, + task_name="dedup", ) if self.skip_metrics: @@ -767,7 +775,10 @@ def dedup_and_metrics( driver.add_algo(HsMetricAlgo(hs_metrics, self.bed, self.bed)) driver.add_algo(InsertSizeMetricAlgo(is_metrics)) metrics_job = Job( - Pipeline(Command(*driver.build_cmd())), "metrics", 0 + Pipeline(Command(*driver.build_cmd())), + "metrics", + 0, + task_name="metrics", ) # Run metrics in the background # Run WgsMetricsAlgo after duplicate marking to account for @@ -779,7 +790,10 @@ def dedup_and_metrics( ) driver.add_algo(CoverageMetrics(coverage_metrics)) metrics_job = Job( - Pipeline(Command(*driver.build_cmd())), "metrics", 0 + Pipeline(Command(*driver.build_cmd())), + "metrics", + 0, + task_name="metrics", ) # Run metrics in the background # Rehead WGS metrics so they are recognized by MultiQC @@ -801,6 +815,7 @@ def dedup_and_metrics( ), "Rehead metrics", 0, + task_name="metrics", ) return ([deduped], lc_job, dedup_job, metrics_job, rehead_job) @@ -874,6 +889,7 @@ def sr_call_variants( Pipeline(Command(*driver.build_cmd())), "variant-calling", self.cores, + task_name="variant-calling", ) # Genotyping and filtering with DNAModelApply @@ -889,12 +905,20 @@ def sr_call_variants( ) ) apply_job = Job( - Pipeline(Command(*driver.build_cmd())), "model-apply", self.cores + Pipeline(Command(*driver.build_cmd())), + "model-apply", + self.cores, + task_name="model-apply", ) # Remove the tmp_vcf rm_cmd = ["rm", str(tmp_vcf), str(tmp_vcf) + ".tbi"] - rm_job = Job(Pipeline(Command(*rm_cmd, fail_ok=True)), "rm-tmp-vcf", 0) + rm_job = Job( + Pipeline(Command(*rm_cmd, fail_ok=True)), + "rm-tmp-vcf", + 0, + task_name="cleanup", + ) # Genotype gVCFs gvcftyper_job = None @@ -911,7 +935,10 @@ def sr_call_variants( ) ) gvcftyper_job = Job( - Pipeline(Command(*driver.build_cmd())), "gvcftyper", self.cores + Pipeline(Command(*driver.build_cmd())), + "gvcftyper", + self.cores, + task_name="gvcftyper", ) # Call SVs @@ -930,7 +957,9 @@ def sr_call_variants( ) ) svsolver_job = Job( - Pipeline(Command(*driver.build_cmd())), "svsolver" + Pipeline(Command(*driver.build_cmd())), + "svsolver", + task_name="sv-calling", ) sv_rm_job = Job( Pipeline( @@ -943,6 +972,7 @@ def sr_call_variants( ), "rm-tmp-sv", 0, + task_name="cleanup", ) return ( @@ -991,7 +1021,10 @@ def call_cnvs( ) ) cnvscope_job = Job( - Pipeline(Command(*driver.build_cmd())), "CNVscope", cores + Pipeline(Command(*driver.build_cmd())), + "CNVscope", + cores, + task_name="cnv", ) driver = Driver( @@ -1009,6 +1042,7 @@ def call_cnvs( Pipeline(Command(*driver.build_cmd())), "CNVModelApply", cores, + task_name="cnv", ) return (cnvscope_job, cnvmodelapply_job) diff --git a/sentieon_cli/dnascope_hybrid.py b/sentieon_cli/dnascope_hybrid.py index e1a9e28..6be5285 100644 --- a/sentieon_cli/dnascope_hybrid.py +++ b/sentieon_cli/dnascope_hybrid.py @@ -764,7 +764,10 @@ def call_variants( ) ) call_job = Job( - Pipeline(Command(*driver.build_cmd())), "calling-1", self.cores + Pipeline(Command(*driver.build_cmd())), + "calling-1", + self.cores, + task_name="variant-calling", ) # Region selection @@ -782,6 +785,7 @@ def call_variants( ), "hybrid-select", 0, + task_name="region-selection", ) mapq0_bed = self.tmp_dir.joinpath("hybrid_mapq0.bed") @@ -802,6 +806,7 @@ def call_variants( Pipeline(Command(*driver.build_cmd())), "mapq0-bed", self.cores, + task_name="region-selection", ) mapq0_slop_bed = self.tmp_dir.joinpath("hybrid_mapq0.ex1000.bed") @@ -814,6 +819,7 @@ def call_variants( ), "mapq0-bed-slop", 0, + task_name="region-selection", ) diff_bed = self.tmp_dir.joinpath("merged_diff.bed") @@ -825,9 +831,15 @@ def call_variants( ), "concat-merge-bed", 0, + task_name="region-selection", ) rm_cmd = ["rm", str(selected_bed), str(mapq0_slop_bed)] - rm_job1 = Job(Pipeline(Command(*rm_cmd, fail_ok=True)), "rm-tmp1", 0) + rm_job1 = Job( + Pipeline(Command(*rm_cmd, fail_ok=True)), + "rm-tmp1", + 0, + task_name="cleanup", + ) stage1_ins_fa = self.tmp_dir.joinpath("stage1_ins.fa") stage1_ins_bed = self.tmp_dir.joinpath("stage1_ins.bed") @@ -881,6 +893,7 @@ def call_variants( ), "first-stage", self.cores, + task_name="hybrid-realignment", ) rm_cmd = [ "rm", @@ -888,7 +901,12 @@ def call_variants( str(stage1_ins_bed), str(stage1_hap_vcf), ] - rm_job2 = Job(Pipeline(Command(*rm_cmd, fail_ok=True)), "rm-tmp2", 0) + rm_job2 = Job( + Pipeline(Command(*rm_cmd, fail_ok=True)), + "rm-tmp2", + 0, + task_name="cleanup", + ) stage2_bed = self.tmp_dir.joinpath("hybrid_stage2.bed") stage2_unmap_bam = self.tmp_dir.joinpath("hybrid_stage2_unmap.bam") @@ -911,10 +929,16 @@ def call_variants( Pipeline(Command(*driver.build_cmd())), "second-stage", self.cores, + task_name="hybrid-realignment", ) rm_cmd = ["rm", str(stage1_bam), str(stage1_hap_bam)] - rm_job3 = Job(Pipeline(Command(*rm_cmd, fail_ok=True)), "rm-tmp3", 0) + rm_job3 = Job( + Pipeline(Command(*rm_cmd, fail_ok=True)), + "rm-tmp3", + 0, + task_name="cleanup", + ) suffix = "bam" if self.bam_format else "cram" stage3_aln = pathlib.Path( @@ -943,9 +967,15 @@ def call_variants( ), "third-stage", self.cores, + task_name="hybrid-realignment", ) rm_cmd = ["rm", str(stage2_unmap_bam), str(stage2_alt_bam)] - rm_job4 = Job(Pipeline(Command(*rm_cmd, fail_ok=True)), "rm-tmp4", 0) + rm_job4 = Job( + Pipeline(Command(*rm_cmd, fail_ok=True)), + "rm-tmp4", + 0, + task_name="cleanup", + ) # pass 2 of variant calling pass2_vcf = self.tmp_dir.joinpath("hybrid_pass2.vcf.gz") @@ -967,7 +997,10 @@ def call_variants( ) ) call2_job = Job( - Pipeline(Command(*driver.build_cmd())), "call2", self.cores + Pipeline(Command(*driver.build_cmd())), + "call2", + self.cores, + task_name="variant-calling", ) # Merge and normalize the VCFs @@ -981,6 +1014,7 @@ def call_variants( ), "subset-calls", 0, + task_name="vcf-merge", ) concat_job = Job( cmds.bcftools_concat( @@ -989,9 +1023,15 @@ def call_variants( ), "concat-calls", 0, + task_name="vcf-merge", ) rm_cmd = ["rm", str(combined_vcf), str(subset_vcf), str(pass2_vcf)] - rm_job5 = Job(Pipeline(Command(*rm_cmd, fail_ok=True)), "rm-tmp5", 0) + rm_job5 = Job( + Pipeline(Command(*rm_cmd, fail_ok=True)), + "rm-tmp5", + 0, + task_name="cleanup", + ) # Annotate the output VCF hybrid_anno = pathlib.Path( @@ -1011,6 +1051,7 @@ def call_variants( ), "anno-calls", 0, + task_name="annotation", ) transfer_jobs: Optional[List[Job]] = None @@ -1077,7 +1118,10 @@ def call_variants( ) ) apply_job = Job( - Pipeline(Command(*driver.build_cmd())), "model-apply", self.cores + Pipeline(Command(*driver.build_cmd())), + "model-apply", + self.cores, + task_name="model-apply", ) # Final normalize @@ -1090,6 +1134,7 @@ def call_variants( ), "final-norm", 0, + task_name="vcf-norm", ) return ( call_job, diff --git a/sentieon_cli/dnascope_longread.py b/sentieon_cli/dnascope_longread.py index c256cd0..ca4aeb0 100644 --- a/sentieon_cli/dnascope_longread.py +++ b/sentieon_cli/dnascope_longread.py @@ -672,6 +672,7 @@ def lr_align_inputs(self) -> Tuple[List[pathlib.Path], Set[Job]]: ), f"bam-realign-{i}", self.cores, + task_name="alignment", ) ) res.append(out_aln) @@ -725,8 +726,9 @@ def lr_align_fastq(self) -> Tuple[List[pathlib.Path], Set[Job]]: self.minimap2_args, self.util_sort_args, ), - "align-{i}", + f"align-{i}", self.cores, + task_name="alignment", ) ) res.append(out_aln) @@ -764,6 +766,7 @@ def mosdepth(self, sample_input: List[pathlib.Path]) -> Set[Job]: ), f"mosdepth-{i}", 0, # Run in background + task_name="metrics", ) ) return mosdepth_jobs @@ -791,6 +794,7 @@ def merge_input_files( Pipeline(Command(*driver.build_cmd())), "merge-bam", 0, + task_name="alignment", ) return (merged_bam, merge_job) @@ -823,6 +827,7 @@ def pbsv( ), "pbsv-discover", 0, + task_name="sv-calling", ) # pbsv call @@ -840,6 +845,7 @@ def pbsv( Pipeline(Command(*call_cmd)), "pbsv-call", self.cores, + task_name="sv-calling", ) return (pbsv_discover, pbsv_call) @@ -875,7 +881,10 @@ def hificnv( if self.cnv_excluded_regions: hificnv_cmd.extend(["--exclude", str(self.cnv_excluded_regions)]) hificnv_job = Job( - Pipeline(Command(*hificnv_cmd)), "hificnv", self.cores + Pipeline(Command(*hificnv_cmd)), + "hificnv", + self.cores, + task_name="cnv", ) return hificnv_job @@ -935,7 +944,10 @@ def lr_call_variants( ) ) first_calling_job = Job( - Pipeline(Command(*driver.build_cmd())), "first-pass", self.cores + Pipeline(Command(*driver.build_cmd())), + "first-pass", + self.cores, + task_name="variant-calling", ) # Transfer annotations to the tmp vcf @@ -970,6 +982,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "first-modelapply", self.cores, + task_name="model-apply", ) # Phasing and RepeatModel @@ -998,7 +1011,10 @@ def lr_call_variants( ) ) phaser_job = Job( - Pipeline(Command(*driver.build_cmd())), "variantphaser", self.cores + Pipeline(Command(*driver.build_cmd())), + "variantphaser", + self.cores, + task_name="phasing", ) bcftools_subset_phased_job = None @@ -1022,6 +1038,7 @@ def lr_call_variants( ), "bcftools-subset-phased", 0, + task_name="phasing", ) fai_to_bed_job = None @@ -1036,12 +1053,14 @@ def lr_call_variants( ), "fai-to-bed", 0, + task_name="phasing", ) bcftools_subtract_job = Job( cmds.cmd_bedtools_subtract(bed, phased_bed, unphased_bed), "bedtools-subtract", 0, + task_name="phasing", ) repeatmodel_job = None @@ -1068,6 +1087,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "repeatmodel", self.cores, + task_name="repeat-model", ) bcftools_subset_unphased_job = Job( @@ -1085,6 +1105,7 @@ def lr_call_variants( ), "bcftools-subset-unphased", 0, + task_name="phasing", ) # Second pass - phased variants @@ -1127,6 +1148,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "second-pass", self.cores, + task_name="variant-calling", ) ) @@ -1153,6 +1175,7 @@ def lr_call_variants( ), "patch", self.cores, + task_name="variant-patch", ) # Transfer annotations to the patched VCFs @@ -1198,6 +1221,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "second-modelapply", self.cores, + task_name="model-apply", ) ) @@ -1223,6 +1247,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "calling-unphased", self.cores, + task_name="variant-calling", ) # Patch DNA and DNAHP variants @@ -1236,7 +1261,9 @@ def lr_call_variants( self.cores, kwargs, ) - diploid_patch_job = Job(cmd, "diploid-patch", self.cores) + diploid_patch_job = Job( + cmd, "diploid-patch", self.cores, task_name="variant-patch" + ) # Transfer annotations to the diploid VCF unphased_transfer_jobs: List[Job] = [] @@ -1270,6 +1297,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "modelapply-unphased", self.cores, + task_name="model-apply", ) # merge calls to create the output @@ -1290,6 +1318,7 @@ def lr_call_variants( ), "merge", self.cores, + task_name="vcf-merge", ) gvcf_combine_job = None @@ -1304,6 +1333,7 @@ def lr_call_variants( ), "gvcf-combine", 0, + task_name="gvcf", ) haploid_calling_job = None @@ -1352,6 +1382,7 @@ def lr_call_variants( Pipeline(Command(*driver.build_cmd())), "haploid-calling", self.cores, + task_name="variant-calling", ) haploid_patch2_job = Job( @@ -1364,6 +1395,7 @@ def lr_call_variants( ), "haploid-patch2", self.cores, + task_name="variant-patch", ) haploid_concat_job = Job( cmds.bcftools_concat( @@ -1372,6 +1404,7 @@ def lr_call_variants( ), "haploid-diploid-concat", 0, + task_name="vcf-merge", ) if self.gvcf: @@ -1394,6 +1427,7 @@ def lr_call_variants( ), "haploid-gvcf-combine", 0, + task_name="gvcf", ) haploid_gvcf_concat_job = Job( cmds.bcftools_concat( @@ -1402,6 +1436,7 @@ def lr_call_variants( ), "haploid-gvcf-concat", 0, + task_name="gvcf", ) return LRCallVariantsResult( first_calling_job, @@ -1466,6 +1501,9 @@ def call_svs( ) ) longreadsv_job = Job( - Pipeline(Command(*driver.build_cmd())), "LongReadSV", self.cores + Pipeline(Command(*driver.build_cmd())), + "LongReadSV", + self.cores, + task_name="sv-calling", ) return longreadsv_job diff --git a/sentieon_cli/executor.py b/sentieon_cli/executor.py index 66b1aaf..3b9ad58 100644 --- a/sentieon_cli/executor.py +++ b/sentieon_cli/executor.py @@ -2,20 +2,46 @@ import asyncio import contextlib +import os +import pathlib import signal import sys import threading import time from abc import ABC, abstractmethod -from typing import Any, Callable, Dict, List, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple from .job import Job from .logging import get_logger +from .run_logs import JobLogSink, RunLogs from .scheduler import BaseScheduler -from .shell_pipeline import Context +from .shell_pipeline import Command, Context logger = get_logger(__name__) +# Lines of a failing process's log to quote in the failure report +TAIL_LINES = 20 +# Only the tail of a log is read; tool logs can be arbitrarily large +TAIL_BYTES = 64 * 1024 + + +def _log_tail(path: pathlib.Path, lines: int = TAIL_LINES) -> List[str]: + """Return the last ``lines`` lines of a log file. + + Reads at most ``TAIL_BYTES`` from the end, so quoting the tail of a + multi-gigabyte log is cheap. A line split by the seek is dropped with the + rest of the truncated prefix. + """ + try: + with open(path, "rb") as handle: + size = handle.seek(0, os.SEEK_END) + handle.seek(max(0, size - TAIL_BYTES)) + data = handle.read() + except OSError as exc: + logger.debug("could not read %s: %s", path, exc) + return [] + return data.decode("utf-8", errors="replace").splitlines()[-lines:] + def _signal_procs(context: Context, sig: int) -> None: """Send a signal to every live sub-process in a context.""" @@ -202,12 +228,14 @@ def __init__( *, install_signal_handlers: bool = False, shutdown_grace_period: float = 10.0, + run_logs: Optional[RunLogs] = None, ) -> None: super().__init__( scheduler, install_signal_handlers=install_signal_handlers, ) self.shutdown_grace_period = shutdown_grace_period + self.run_logs = run_logs self.running: List[ Tuple[ Job, @@ -221,13 +249,19 @@ async def run_job(self, job: Job) -> None: """Run a job""" cmd = job.shell logger.info("Running: %s", cmd) - context = Context() + sink = self.run_logs.job_sink(job) if self.run_logs else None + context = Context() if sink is None else Context(log_sink=sink) start_time = time.monotonic_ns() try: - proc = await cmd.run( - context, - stderr=sys.stderr, - ) + if sink is None: + # No log directory: the processes inherit our stderr + proc = await cmd.run( + context, + stderr=sys.stderr, + ) + else: + # Unset streams are resolved against the sink + proc = await cmd.run(context) self.running.append( ( job, @@ -242,6 +276,8 @@ async def run_job(self, job: Job) -> None: # the caller's pipeline and is left to propagate. logger.error("failed to start command: %s", job.shell) logger.error("Error: %s", str(e)) + # Stages that did spawn have logs; keep them and point at them. + self._report_logs(job, sink) self.jobs_with_errors.append(job) self.start_new_jobs = False # A later pipeline stage can fail to spawn after earlier stages @@ -252,6 +288,47 @@ async def run_job(self, job: Job) -> None: await self._terminate_context(context) await _cleanup_quietly(context) + def _report_logs(self, job: Job, sink: Optional[JobLogSink]) -> None: + """Point at the logs of a job that was aborted rather than reaped.""" + if sink is None: + return + paths = sink.log_paths() + if paths: + logger.error( + "Logs from %s: %s", + job, + ", ".join(str(path) for path in paths), + ) + + def _report_failure( + self, + job: Job, + subcommand: Command, + ret: int, + sink: Optional[JobLogSink], + ) -> None: + """Report a failed sub-command with its pid, log path and log tail. + + Without a log -- no log directory, or a process that never spawned -- + the caller's messages are the whole report. + """ + log_path = sink.stderr_log_for(subcommand) if sink else None + if log_path is None: + return + pid = subcommand.proc.pid if subcommand.proc else None + report = [ + f"Failed sub-command of {job}: {subcommand}", + f" exit code: {ret}, pid: {pid}", + f" log: {log_path}", + ] + tail = _log_tail(log_path) + if tail: + report.append(f" last {len(tail)} line(s) of the log:") + report.extend(f" {line}" for line in tail) + else: + report.append(" the log is empty") + logger.error("\n".join(report)) + async def _terminate_context(self, context: Context) -> None: """Tear down processes already spawned in an aborted context. @@ -310,6 +387,7 @@ async def _drive(self, stop_event: asyncio.Event) -> None: # Check if the command failed cmd_failed = False + failures: List[Tuple[Command, int]] = [] for subcommand in context.commands: if not subcommand.proc: logger.error( @@ -337,6 +415,7 @@ async def _drive(self, stop_event: asyncio.Event) -> None: f"{subcommand}" ) cmd_failed = True + failures.append((subcommand, ret)) if ret == -9: logger.error( "Sub-command received SIGKILL. " @@ -355,8 +434,18 @@ async def _drive(self, stop_event: asyncio.Event) -> None: logger.error("Error: %s", str(exc)) cmd_failed = True + sink = context.log_sink if cmd_failed: logger.error("Command failure: %s", job.shell) + # The sink's files are closed by cleanup(), so their + # tails can be read now. + for failed, code in failures: + self._report_failure(job, failed, code, sink) + if self.run_logs is not None: + logger.error( + "Task logs for debugging: %s", + self.run_logs.task_logs, + ) self.jobs_with_errors.append(job) self.start_new_jobs = False else: @@ -364,6 +453,8 @@ async def _drive(self, stop_event: asyncio.Event) -> None: f"Finished command in " f"{total_seconds:.2f}: {job.shell}" ) + if sink is not None: + sink.finalize(success=not cmd_failed) if not self.start_new_jobs: # Don't start new jobs @@ -406,8 +497,11 @@ async def _shutdown(self) -> None: if tasks: await asyncio.wait(tasks) - for _job, context, _task, _start in running: + for job, context, _task, _start in running: # A proc-sub inner launch failure can surface from cleanup() as we # tear down after an interrupt; log it rather than let it escape, # and keep releasing the remaining contexts. await _cleanup_quietly(context) + # An interrupted job keeps all of its logs; they explain how far + # it got. + self._report_logs(job, context.log_sink) diff --git a/sentieon_cli/job.py b/sentieon_cli/job.py index 3e36fa2..bb893bb 100644 --- a/sentieon_cli/job.py +++ b/sentieon_cli/job.py @@ -16,22 +16,46 @@ class Job: * ``name`` -- a human-readable label (not part of identity). * ``threads`` -- CPU threads the job needs (a local scheduling budget). * ``resources`` -- named resource counts (e.g. NUMA-node tokens). + * ``task_name`` -- the pipeline stage this job belongs to; it groups the + job's log files, so every shard of an operation shares one value. + * ``job_id`` -- ``{name}-{n}``, unique across a run (not part of + identity). A job's identity is keyed only on ``shell``: two jobs with the same pipeline are equal (and collide in a DAG) regardless of the other fields. + So two Job objects built from identical pipelines are "the same job" to + the DAG even though each carries its own ``job_id``. """ + # Per-name counters backing ``job_id``; see ``reset_ids``. + _id_counters: Dict[str, int] = {} + def __init__( self, pipeline: Pipeline, name: str, threads: int = 1, resources: Optional[Dict[str, int]] = None, + *, + task_name: str, ) -> None: self.shell = pipeline self.name = name 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 + self.job_id = f"{name}-{count}" + + @classmethod + def reset_ids(cls) -> None: + """Restart id numbering. + + Ids must stay unique for a whole run, which can execute more than one + DAG, so this is called once per CLI invocation -- never between DAGs. + """ + cls._id_counters.clear() def __hash__(self) -> int: return hash(self.shell) @@ -45,7 +69,7 @@ def __ne__(self, other: object) -> bool: return not self == other def __repr__(self) -> str: - return f"Job({self.name})" + return f"Job({self.job_id})" def __str__(self) -> str: - return f"Job({self.name})" + return f"Job({self.job_id})" diff --git a/sentieon_cli/logging.py b/sentieon_cli/logging.py index faca6d0..df0a48a 100644 --- a/sentieon_cli/logging.py +++ b/sentieon_cli/logging.py @@ -1,17 +1,59 @@ +""" +Logging configuration + +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. +""" + import logging +import os +from typing import Optional + +LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s" + +_package_logger = logging.getLogger(__package__) +_package_logger.setLevel(logging.DEBUG) +_package_logger.propagate = False # avoid duplicates through the root logger + +_console_handler = logging.StreamHandler() +_console_handler.setFormatter(logging.Formatter(LOG_FORMAT)) +_console_handler.setLevel(logging.INFO) +_package_logger.addHandler(_console_handler) -handler = logging.StreamHandler() -handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s")) +# The file handler installed by `add_file_handler`, kept so repeated setup in +# a single process replaces it rather than accumulating handlers. +_file_handler: Optional[logging.FileHandler] = None def get_logger(name: str) -> logging.Logger: - """Return a logger with a StreamHandler.""" - logger = logging.getLogger(name) - logger.addHandler(handler) - logger.propagate = False - return logger + """Return a module logger propagating to the package logger.""" + return logging.getLogger(name) + + +def set_console_level(level: int | str) -> None: + """Set the verbosity of the console handler.""" + _console_handler.setLevel(level) + + +def add_file_handler(path: "os.PathLike[str] | str") -> logging.FileHandler: + """Attach a DEBUG file handler, replacing any previously added one.""" + global _file_handler + if _file_handler is not None: + remove_file_handler(_file_handler) + handler = logging.FileHandler(path, mode="w") + handler.setFormatter(logging.Formatter(LOG_FORMAT)) + handler.setLevel(logging.DEBUG) + _package_logger.addHandler(handler) + _file_handler = handler + return handler -def set_level(level: int | str) -> None: - """Set the level of the package logger.""" - logging.getLogger(__package__).setLevel(level) +def remove_file_handler(handler: logging.FileHandler) -> None: + """Detach and close a handler added by ``add_file_handler``.""" + global _file_handler + _package_logger.removeHandler(handler) + handler.close() + if _file_handler is handler: + _file_handler = None diff --git a/sentieon_cli/pipeline.py b/sentieon_cli/pipeline.py index be82e0d..65231aa 100644 --- a/sentieon_cli/pipeline.py +++ b/sentieon_cli/pipeline.py @@ -9,6 +9,7 @@ import pathlib import shutil import sys +import time from typing import Any, Dict, List, Optional import packaging.version @@ -18,7 +19,8 @@ from .exceptions import DagExecutionError from .executor import BaseExecutor, DryRunExecutor, LocalExecutor from .job import Job -from .logging import get_logger, set_level +from .logging import get_logger, set_console_level +from .run_logs import RunLogs from .scheduler import ThreadScheduler from .util import __version__, check_version, path_arg, tmp @@ -53,6 +55,13 @@ class BasePipeline(ABC): "help": "Print the commands without running them.", "action": "store_true", }, + "log_dir": { + "help": ( + "Directory for the run's log files. Defaults to the output " + "VCF with the '.vcf.gz' suffix replaced by '_logs'." + ), + "type": path_arg(), + }, # Hidden arguments "retain_tmpdir": { "help": argparse.SUPPRESS, @@ -105,11 +114,6 @@ def handle_arguments(self, args: argparse.Namespace): if k in args.__dict__: setattr(self, k, getattr(args, k)) - def setup_logging(self, args: argparse.Namespace) -> None: - self.logger = get_logger(__name__) - set_level(args.loglevel) - self.logger.info("Starting sentieon-cli version: %s", __version__) - def __init__(self) -> None: self.reference: Optional[pathlib.Path] = None self.cores = mp.cpu_count() @@ -117,25 +121,71 @@ def __init__(self) -> None: self.dry_run = False self.retain_tmpdir = False self.skip_version_check = False + self.log_dir: Optional[pathlib.Path] = None self.output_vcf: Optional[pathlib.Path] = 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""" + self.logger = get_logger(__name__) + set_console_level(args.loglevel) + + # 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() + + # 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) + + def log_completion(self, success: bool, start_time: float) -> None: + """Report the outcome and duration of the run""" + self.logger.info( + "Finished sentieon-cli (status: %s, elapsed: %.1fs)", + "succeeded" if success else "failed", + time.monotonic() - start_time, + ) + if not success and self.run_logs: + self.logger.info( + "Logs from this run are in: %s", self.run_logs.log_dir + ) def main(self, args: argparse.Namespace) -> None: """Run the DNAscope pipeline""" self.handle_arguments(args) self.setup_logging(args) - self.validate() - self.configure() + start_time = time.monotonic() + success = False + try: + self.validate() + self.configure() - tmp_dir_str = tmp() - self.tmp_dir = pathlib.Path(tmp_dir_str) + tmp_dir_str = tmp() + self.tmp_dir = pathlib.Path(tmp_dir_str) - try: - dag = self.build_dag() - executor = self.run(dag) - self.check_execution(dag, executor) + try: + dag = self.build_dag() + executor = self.run(dag) + self.check_execution(dag, executor) + finally: + if not self.retain_tmpdir: + shutil.rmtree(tmp_dir_str) + success = True finally: - if not self.retain_tmpdir: - shutil.rmtree(tmp_dir_str) + self.log_completion(success, start_time) def check_execution( self, @@ -145,7 +195,10 @@ def check_execution( """Check the DAG and executor after a run""" if executor.jobs_with_errors: failed = ", ".join(str(job) for job in executor.jobs_with_errors) - raise DagExecutionError(f"Execution failed for jobs: {failed}") + message = f"Execution failed for jobs: {failed}" + if self.run_logs: + message += f"\nTask logs are in: {self.run_logs.task_logs}" + raise DagExecutionError(message) if len(dag.waiting_jobs) > 0 or len(dag.ready_jobs) > 0: raise DagExecutionError( @@ -158,10 +211,14 @@ def check_execution( def validate(self) -> None: pass - def validate_output_vcf(self) -> None: + def validate_output_suffix(self) -> None: + """Confirm the output VCF file name ends in '.vcf.gz'""" if not str(self.output_vcf).endswith(".vcf.gz"): self.logger.error("The output file should end with '.vcf.gz'") sys.exit(2) + + def validate_output_vcf(self) -> None: + self.validate_output_suffix() assert self.output_vcf is not None parent = self.output_vcf.resolve().parent if not parent.is_dir(): @@ -234,6 +291,7 @@ def multiqc(self) -> Optional[Job]: ), "multiqc", 0, + task_name="multiqc", ) return multiqc_job @@ -256,7 +314,11 @@ def run(self, dag: DAG) -> BaseExecutor: else: # Handle Ctrl-C/SIGTERM by terminating running jobs gracefully; # the handlers are installed only for the duration of the run. - executor = LocalExecutor(scheduler, install_signal_handlers=True) + executor = LocalExecutor( + scheduler, + install_signal_handlers=True, + run_logs=self.run_logs, + ) self.logger.info("Starting execution") executor.execute() diff --git a/sentieon_cli/run_logs.py b/sentieon_cli/run_logs.py new file mode 100644 index 0000000..5a7f7c3 --- /dev/null +++ b/sentieon_cli/run_logs.py @@ -0,0 +1,191 @@ +""" +The log directory for a single run +""" + +from __future__ import annotations + +import dataclasses +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__ + +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""" + + command: Command + stage_index: int + path: pathlib.Path + kind: str + header_bytes: int + + +class JobLogSink: + """The per-process log files of a single job. + + Attached to the job's ``Context``, so every process spawned for the job -- + including process substitutions -- writes its stderr (and its stdout, when + that would otherwise be inherited) to its own file. + """ + + def __init__(self, task_logs: pathlib.Path, job: Job) -> None: + self.job_id = job.job_id + self.name = job.name + self.task_name = job.task_name + self.log_dir = task_logs / sanitize(job.task_name) + self._prefix = sanitize(job.job_id) + self._stage_index = 0 # Next unallocated stage index + self._logs: List[_ProcessLog] = [] + + def open_stderr(self, command: Command) -> IO[bytes]: + """Open the stderr log of the next process to spawn""" + return self._open(command, _STDERR) + + def open_stdout(self, command: Command) -> IO[bytes]: + """Open the stdout log of the next process to spawn""" + return self._open(command, _STDOUT) + + def _stage_for(self, command: Command) -> int: + """The stage index of a command, allocated on first use. + + Both streams of one process share its index, so an already-seen + command reuses it. Matched by identity: equal-but-distinct commands + are distinct processes. + """ + for log in self._logs: + if log.command is command: + return log.stage_index + stage_index = self._stage_index + self._stage_index += 1 + return stage_index + + def _open(self, command: Command, kind: str) -> IO[bytes]: + self.log_dir.mkdir(parents=True, exist_ok=True) + stage_index = self._stage_for(command) + suffix = ".log" if kind == _STDERR else ".stdout.log" + path = self.log_dir / f"{self._prefix}.{stage_index}{suffix}" + handle = open(path, "wb") + header_bytes = 0 + if kind == _STDERR: + # The child inherits the file offset, so the header must be on + # disk before the FD is handed over or it would interleave with + # (or land after) the process's own output. Stdout logs get no + # header: they hold exactly what the tool would have printed. + header = self._header(command, stage_index) + handle.write(header) + handle.flush() + header_bytes = len(header) + self._logs.append( + _ProcessLog(command, stage_index, path, kind, header_bytes) + ) + return handle + + def _header(self, command: Command, stage_index: int) -> bytes: + timestamp = datetime.datetime.now().astimezone() + return ( + f"# timestamp: {timestamp.isoformat(timespec='seconds')}\n" + f"# task: {self.task_name}\n" + f"# job: {self.name} ({self.job_id})\n" + f"# stage: {stage_index}\n" + f"# command: {command}\n" + ).encode() + + def stderr_log_for(self, command: Command) -> Optional[pathlib.Path]: + """The stderr log of a command, matched by identity. + + Equal-but-distinct ``Command`` objects can share a context, so the + lookup must not fall back to ``__eq__``. + """ + for log in self._logs: + if log.kind == _STDERR and log.command is command: + return log.path + return None + + def log_paths(self) -> List[pathlib.Path]: + """Every log file opened for this job""" + return [log.path for log in self._logs] + + def finalize(self, success: bool) -> None: + """Drop the logs of a successful job that hold nothing but a header. + + Empty logs are common (cleanup jobs, unused stdout) and only add + clutter. After a failure everything is kept -- an empty log is itself + informative. + """ + if not success: + return + for log in self._logs: + try: + if log.path.stat().st_size <= log.header_bytes: + log.path.unlink() + except OSError: + pass + + +class RunLogs: + """The log directory of a single sentieon-cli invocation""" + + def __init__(self, log_dir: pathlib.Path) -> None: + self.log_dir = log_dir + self.run_log = log_dir / "run.log" + self.command_txt = log_dir / "command.txt" + self.task_logs = log_dir / "task_logs" + self.file_handler: Optional[logging.FileHandler] = None + + def setup(self) -> None: + """Prepare the log directory and start writing `run.log`""" + self.create_dirs() + self.write_command() + self.file_handler = add_file_handler(self.run_log) + + def create_dirs(self) -> None: + """Create the log directory, clearing logs from any previous run""" + # The log directory itself is never removed - the user may point + # `--log_dir` at an existing directory holding unrelated files. + self.log_dir.mkdir(parents=True, exist_ok=True) + if self.task_logs.exists(): + shutil.rmtree(self.task_logs) + self.task_logs.mkdir(parents=True) + + def job_sink(self, job: Job) -> JobLogSink: + """Create the log sink for a job""" + return JobLogSink(self.task_logs, job) + + def write_command(self) -> None: + """Record the invocation so the run can be reproduced""" + timestamp = datetime.datetime.now().astimezone() + self.command_txt.write_text( + f"command: {shlex.join(sys.argv)}\n" + f"version: {__version__}\n" + f"directory: {pathlib.Path.cwd()}\n" + f"timestamp: {timestamp.isoformat(timespec='seconds')}\n" + ) + + def close(self) -> None: + """Stop writing `run.log`""" + if self.file_handler is not None: + remove_file_handler(self.file_handler) + self.file_handler = None diff --git a/sentieon_cli/sentieon_pangenome.py b/sentieon_cli/sentieon_pangenome.py index 6d73fcf..50f23af 100644 --- a/sentieon_cli/sentieon_pangenome.py +++ b/sentieon_cli/sentieon_pangenome.py @@ -8,6 +8,7 @@ import pathlib import shutil import sys +import time from typing import Dict, List, Optional, Set, Tuple, Union import packaging.version @@ -269,34 +270,42 @@ def main(self, args: argparse.Namespace) -> None: """Run the pipeline""" self.handle_arguments(args) self.setup_logging(args) - self.validate_ref() - - self.fai_data = parse_fai(pathlib.Path(str(self.reference) + ".fai")) - self.pop_vcf_contigs: Dict[str, Optional[int]] = {} - if self.pop_vcf: - self.pop_vcf_contigs = vcf_contigs(self.pop_vcf, self.dry_run) - self.logger.debug("VCF contigs are: %s", self.pop_vcf_contigs) - - self.validate() - self.shards = determine_shards_from_fai( - self.fai_data, 10 * 1000 * 1000 - ) + start_time = time.monotonic() + success = False + try: + self.validate_ref() - tmp_dir_str = tmp() - self.tmp_dir = pathlib.Path(tmp_dir_str) + self.fai_data = parse_fai( + pathlib.Path(str(self.reference) + ".fai") + ) + self.pop_vcf_contigs: Dict[str, Optional[int]] = {} + if self.pop_vcf: + self.pop_vcf_contigs = vcf_contigs(self.pop_vcf, self.dry_run) + self.logger.debug("VCF contigs are: %s", self.pop_vcf_contigs) + + self.validate() + self.shards = determine_shards_from_fai( + self.fai_data, 10 * 1000 * 1000 + ) - dag = self.build_first_dag() - executor = self.run(dag) - self.check_execution(dag, executor) + tmp_dir_str = tmp() + self.tmp_dir = pathlib.Path(tmp_dir_str) - if self.expansion_catalog or self.segdup_caller is not None: - self.get_sex(self.ploidy_json) - dag = self.build_second_dag() + dag = self.build_first_dag() executor = self.run(dag) self.check_execution(dag, executor) - if not self.retain_tmpdir: - shutil.rmtree(tmp_dir_str) + if self.expansion_catalog or self.segdup_caller is not None: + self.get_sex(self.ploidy_json) + dag = self.build_second_dag() + executor = self.run(dag) + self.check_execution(dag, executor) + + if not self.retain_tmpdir: + shutil.rmtree(tmp_dir_str) + success = True + finally: + self.log_completion(success, start_time) def validate(self) -> None: """Validate pipeline inputs""" @@ -696,6 +705,7 @@ def build_first_dag(self) -> DAG: Pipeline(Command("ln", "-sf", "/dev/stdout", str(rw_bam))), "extract-kmc-symlink", 1, + task_name="read-extraction", ) dag.add_job(ln_job) @@ -712,6 +722,7 @@ def build_first_dag(self) -> DAG: ), "extract-kmc", self.cores, + task_name="read-extraction", ) dag.add_job(extract_kmc_job, {ln_job}) haplotype_dependencies.add(extract_kmc_job) @@ -960,6 +971,7 @@ def build_alignment_job( ), "bwa-extract", self.cores, + task_name="alignment", ) return bwa_job @@ -990,6 +1002,7 @@ def build_haplotypes_job( ), "vg-haplotypes", self.cores, + task_name="pangenome", ) return haplotypes_job @@ -1006,6 +1019,7 @@ def build_gfa_job( ), "vg-convert-gfa", 0, + task_name="pangenome", ) return gfa_job @@ -1020,6 +1034,7 @@ def build_fasta_job( ), "vg-paths-fasta", 0, + task_name="pangenome", ) return fasta_job @@ -1064,6 +1079,7 @@ def build_minimap2_lift_job( ), "mm2-lift", self.cores, + task_name="pangenome-alignment", ) return mm2_job @@ -1098,6 +1114,7 @@ def build_dedup_job( Pipeline(Command(*driver.build_cmd())), f"locuscollector-{tag}", self.cores, + task_name="dedup", ) driver2 = Driver( @@ -1112,6 +1129,7 @@ def build_dedup_job( Pipeline(Command(*driver2.build_cmd())), f"dedup-{tag}", self.cores, + task_name="dedup", ) return lc_job, dedup_job @@ -1167,7 +1185,12 @@ def build_metrics_job( driver.add_algo(WgsMetricsAlgo(wgs_metrics, include_unpaired="true")) driver.add_algo(CoverageMetrics(coverage_metrics)) - metrics_job = Job(Pipeline(Command(*driver.build_cmd())), "metrics", 0) + metrics_job = Job( + Pipeline(Command(*driver.build_cmd())), + "metrics", + 0, + task_name="metrics", + ) rehead_script = pathlib.Path( str( @@ -1185,6 +1208,7 @@ def build_metrics_job( ), "Rehead metrics", 0, + task_name="metrics", ) return (metrics_job, rehead_job) @@ -1233,6 +1257,7 @@ def build_dnascope_job( Pipeline(Command(*driver.build_cmd())), "dnascope-raw", self.cores, + task_name="variant-calling", ) def build_dnamodelapply_job( @@ -1259,6 +1284,7 @@ def build_dnamodelapply_job( Pipeline(Command(*driver.build_cmd())), "model-apply", self.cores, + task_name="model-apply", ) def build_gvcftyper_job( @@ -1286,6 +1312,7 @@ def build_gvcftyper_job( Pipeline(Command(*driver.build_cmd())), "gvcftyper", self.cores, + task_name="gvcftyper", ) def build_segdup_job( @@ -1324,6 +1351,7 @@ def build_segdup_job( ), "segdup-caller", self.cores, + task_name="segdup", ) def build_second_dag(self) -> DAG: @@ -1388,6 +1416,7 @@ def build_t1k_jobs( Pipeline(Command(*driver.build_cmd())), f"t1k-{tag}-extract", self.cores, + task_name="t1k", ) t1k_job = Job( @@ -1401,6 +1430,7 @@ def build_t1k_jobs( ), f"t1k-{tag}", self.cores, + task_name="t1k", ) return (extract_job, t1k_job) @@ -1426,6 +1456,7 @@ def build_expansion_job( ), "expansion-hunter", self.cores, + task_name="expansion-hunter", ) def _add_cnv_jobs( @@ -1497,6 +1528,7 @@ def _build_cnvscope_job( Pipeline(Command(*driver.build_cmd())), "cnvscope", self.cores, + task_name="cnv", ) def _build_cnv_model_apply_job( @@ -1521,6 +1553,7 @@ def _build_cnv_model_apply_job( Pipeline(Command(*driver.build_cmd())), "cnv-model-apply", self.cores, + task_name="cnv", ) def _build_indel2cnv_job( @@ -1547,6 +1580,7 @@ def _build_indel2cnv_job( ), "indel2cnv", 0, + task_name="cnv", ) def _build_combine_cnv_job( @@ -1574,4 +1608,5 @@ def _build_combine_cnv_job( ), "combine-cnv", 0, + task_name="cnv", ) diff --git a/sentieon_cli/shell_pipeline.py b/sentieon_cli/shell_pipeline.py index 1d6d26d..fa6aa7a 100644 --- a/sentieon_cli/shell_pipeline.py +++ b/sentieon_cli/shell_pipeline.py @@ -8,16 +8,18 @@ import asyncio import dataclasses import fcntl -import io import os import pathlib import tempfile import shlex from abc import ABC, abstractmethod -from typing import Any, Dict, IO, List, Optional, Union +from typing import Any, Dict, IO, List, Optional, TYPE_CHECKING, Union from .logging import get_logger +if TYPE_CHECKING: + from .run_logs import JobLogSink + logger = get_logger(__name__) @@ -64,7 +66,7 @@ class _ProcSubWait: class Context: """Holds shared resources.""" - def __init__(self) -> None: + def __init__(self, log_sink: Optional[JobLogSink] = None) -> None: # This directory is already unique per pipeline run self.temp_dir = tempfile.TemporaryDirectory() # Background tasks to wait for (e.g., proc sub processes) @@ -72,7 +74,9 @@ def __init__(self) -> None: # Hold the commands run in this context self.commands: List[Command] = [] self._counter = 0 # Monotonic counter - self.file_handles: List[io.IOBase] = [] + self.file_handles: List[IO[Any]] = [] + # Per-process log files; without a sink, stdout/stderr are inherited + self.log_sink = log_sink # Set by cleanup(); a proc sub whose FIFO open was unblocked by # cleanup must not spawn its inner command. self.closing: bool = False @@ -169,7 +173,20 @@ async def run( else: final_args.append(str(arg)) - # 2. Run the process + # 2. Capture the streams the caller left unset + # An explicit stream (a pipe between stages, a file_output, a proc-sub + # FIFO) always wins; only an inherited stream goes to the sink. + if context.log_sink is not None: + if stderr is None: + stderr_log = context.log_sink.open_stderr(self) + context.file_handles.append(stderr_log) + stderr = stderr_log + if stdout is None: + stdout_log = context.log_sink.open_stdout(self) + context.file_handles.append(stdout_log) + stdout = stdout_log + + # 3. Run the process # We allow the caller to wait for this process self.proc = await asyncio.create_subprocess_exec( *final_args, diff --git a/sentieon_cli/transfer.py b/sentieon_cli/transfer.py index a5fed93..ed082fb 100644 --- a/sentieon_cli/transfer.py +++ b/sentieon_cli/transfer.py @@ -89,6 +89,7 @@ def build_transfer_jobs( ), "merge-trim-extra", 1, + task_name="annotation-transfer", ) sharded_merge_jobs.append(view_job) sharded_vcfs.append(subset_vcf) @@ -123,6 +124,7 @@ def build_transfer_jobs( ), f"merge-trim-{i}", 1, + task_name="annotation-transfer", ) sharded_merge_jobs.append(merge_job) sharded_vcfs.append(shard_vcf) @@ -136,5 +138,6 @@ def build_transfer_jobs( ), "merge-trim-concat", cores, + task_name="annotation-transfer", ) return (sharded_merge_jobs, concat_job) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..0436aa7 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,23 @@ +""" +Shared integration-test fixtures +""" + +import pytest + +from sentieon_cli.job import Job + +# `pytest --doctest-modules` collects .py files as modules, and two +# conftest.py files outside a package collide on the module name. This one +# holds no doctests, so skip collecting it. +collect_ignore = ["conftest.py"] + + +@pytest.fixture(autouse=True) +def reset_job_ids(): + """Restart job id numbering for every test. + + ``job_id`` comes from a class-level per-name counter that the CLI resets + once per invocation, so without this a test's ids would depend on which + tests ran before it. + """ + Job.reset_ids() diff --git a/tests/integration/test_executor.py b/tests/integration/test_executor.py index 63c3779..41ce529 100644 --- a/tests/integration/test_executor.py +++ b/tests/integration/test_executor.py @@ -40,6 +40,7 @@ def test_local_executor_simple_job(): job = Job( Pipeline(Command("echo", "hello executor"), file_output=cmd_out), "echo-job", + task_name="test", ) dag.add_job(job) @@ -61,7 +62,7 @@ def test_local_executor_pipeline_job(): Command("cat"), file_output=cmd_out, ) - job = Job(pipeline, "pipeline-job") + job = Job(pipeline, "pipeline-job", task_name="test") dag.add_job(job) scheduler = ThreadScheduler(dag, 1) @@ -78,7 +79,11 @@ def test_local_executor_failing_job(): cmd_in = pathlib.Path(tmp_dir_str) / "test_in.txt" dag = DAG() # This command will fail: the input file does not exist - job = Job(Pipeline(Command("cat", str(cmd_in))), "failing-job") + job = Job( + Pipeline(Command("cat", str(cmd_in))), + "failing-job", + task_name="test", + ) dag.add_job(job) scheduler = ThreadScheduler(dag, 1) @@ -99,7 +104,11 @@ def test_local_executor_proc_sub_job(): InputProcSub(Pipeline(Command("echo", "a"))), InputProcSub(Pipeline(Command("echo", "b"))), ) - job = Job(Pipeline(command, file_output=cmd_out), "proc-sub-job") + job = Job( + Pipeline(command, file_output=cmd_out), + "proc-sub-job", + task_name="test", + ) dag.add_job(job) # Need at least 2 threads for the two proc subs @@ -125,6 +134,7 @@ def test_dry_run_executor_prints_without_running(capsys): job = Job( Pipeline(Command("echo", "dryrun"), file_output=cmd_out), "dry-job", + task_name="test", ) dag.add_job(job) @@ -142,8 +152,8 @@ def test_dry_run_executor_prints_without_running(capsys): def test_dry_run_executor_drains_dependencies(capsys): """DryRunExecutor walks the whole DAG, including dependent jobs""" dag = DAG() - a = Job(Pipeline(Command("echo", "a")), "a") - b = Job(Pipeline(Command("echo", "b")), "b") + a = Job(Pipeline(Command("echo", "a")), "a", task_name="test") + b = Job(Pipeline(Command("echo", "b")), "b", task_name="test") dag.add_job(a) dag.add_job(b, {a}) @@ -172,6 +182,7 @@ def test_fail_ok_subcommand_does_not_fail_job(): file_output=out, ), "tolerant-job", + task_name="test", ) dag.add_job(job) @@ -189,8 +200,20 @@ def test_independent_jobs_all_complete(): o1 = tmp_dir / "o1.txt" o2 = tmp_dir / "o2.txt" dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "1"), file_output=o1), "j1")) - dag.add_job(Job(Pipeline(Command("echo", "2"), file_output=o2), "j2")) + dag.add_job( + Job( + Pipeline(Command("echo", "1"), file_output=o1), + "j1", + task_name="test", + ) + ) + dag.add_job( + Job( + Pipeline(Command("echo", "2"), file_output=o2), + "j2", + task_name="test", + ) + ) scheduler = ThreadScheduler(dag, 2) executor = LocalExecutor(scheduler) @@ -208,8 +231,16 @@ def test_dependent_job_runs_after_dependency(): first = tmp_dir / "first.txt" second = tmp_dir / "second.txt" dag = DAG() - a = Job(Pipeline(Command("echo", "a"), file_output=first), "a") - b = Job(Pipeline(Command("echo", "b"), file_output=second), "b") + a = Job( + Pipeline(Command("echo", "a"), file_output=first), + "a", + task_name="test", + ) + b = Job( + Pipeline(Command("echo", "b"), file_output=second), + "b", + task_name="test", + ) dag.add_job(a) dag.add_job(b, {a}) @@ -225,7 +256,7 @@ def test_dependent_job_runs_after_dependency(): def test_job_that_fails_to_start_is_recorded(): """A job whose command cannot be spawned is recorded as an error.""" dag = DAG() - job = Job(Pipeline(Command("no_such_cmd_zzz")), "ghost") + job = Job(Pipeline(Command("no_such_cmd_zzz")), "ghost", task_name="test") dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) executor.execute() @@ -239,7 +270,9 @@ def test_launch_loop_stops_after_a_failed_launch(): for i in range(5): # Distinct args -> distinct pipeline identities (else the DAG # rejects them as duplicates). - dag.add_job(Job(Pipeline(Command("true", str(i))), f"j{i}")) + dag.add_job( + Job(Pipeline(Command("true", str(i))), f"j{i}", task_name="test") + ) executor = LocalExecutor(ThreadScheduler(dag, 5)) launched = [] @@ -259,14 +292,18 @@ async def fake_run_job(job): def test_infeasible_job_raises_instead_of_deadlocking(): """A job larger than the budget raises rather than stalling silently.""" dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "hi")), "big", 100)) + dag.add_job( + Job(Pipeline(Command("echo", "hi")), "big", 100, task_name="test") + ) with pytest.raises(DagExecutionError): LocalExecutor(ThreadScheduler(dag, 2)).execute() def test_dry_run_also_rejects_infeasible_job(): dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "hi")), "big", 100)) + dag.add_job( + Job(Pipeline(Command("echo", "hi")), "big", 100, task_name="test") + ) with pytest.raises(DagExecutionError): DryRunExecutor(ThreadScheduler(dag, 2)).execute() @@ -279,7 +316,7 @@ async def boom(self, *args, **kwargs): monkeypatch.setattr("sentieon_cli.shell_pipeline.Pipeline.run", boom) dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "x")), "j")) + dag.add_job(Job(Pipeline(Command("echo", "x")), "j", task_name="test")) with pytest.raises(RuntimeError): LocalExecutor(ThreadScheduler(dag, 1)).execute() @@ -325,6 +362,7 @@ def test_procsub_job_with_early_exiting_outer_does_not_hang(): Command("false", InputProcSub(Pipeline(Command("echo", "x")))) ), "early-exit", + task_name="test", ) dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) @@ -344,6 +382,7 @@ def test_procsub_job_with_unspawnable_outer_does_not_hang(): ) ), "ghost-with-procsub", + task_name="test", ) dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) @@ -364,6 +403,7 @@ def test_procsub_inner_that_fails_to_launch_is_recorded(): ) ), "inner-fail", + task_name="test", ) dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) @@ -384,8 +424,9 @@ def test_procsub_inner_failure_does_not_skip_sibling_bookkeeping(): ) ), "inner-fail", + task_name="test", ) - good = Job(Pipeline(Command("true")), "sibling") + good = Job(Pipeline(Command("true")), "sibling", task_name="test") dag.add_job(bad) dag.add_job(good) executor = LocalExecutor(ThreadScheduler(dag, 2)) @@ -419,6 +460,7 @@ def __init__(self) -> None: Command("no_such_cmd_zzz"), ), "leak", + task_name="test", ) dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) @@ -446,6 +488,7 @@ def test_sigpipe_producer_does_not_fail_job(tmp_path): file_output=out, ), "sigpipe", + task_name="test", ) dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) @@ -458,7 +501,11 @@ def test_nonzero_producer_still_fails_job(): """SIGPIPE forgiveness must not mask a producer that exits non-zero for another reason: `false | cat` still fails the job.""" dag = DAG() - job = Job(Pipeline(Command("false"), Command("cat")), "real-fail") + job = Job( + Pipeline(Command("false"), Command("cat")), + "real-fail", + task_name="test", + ) dag.add_job(job) executor = LocalExecutor(ThreadScheduler(dag, 2)) _execute_bounded(executor) @@ -471,6 +518,7 @@ class _FakeContext: def __init__(self, cleanup_error=None): self.commands = [] self.cleaned = False + self.log_sink = None self._error = cleanup_error async def cleanup(self): @@ -510,13 +558,13 @@ async def _done(): executor.running = [ ( - Job(Pipeline(Command("true")), "j1"), + Job(Pipeline(Command("true")), "j1", task_name="test"), bad, asyncio.create_task(_done()), 0, ), ( - Job(Pipeline(Command("false")), "j2"), + Job(Pipeline(Command("false")), "j2", task_name="test"), good, asyncio.create_task(_done()), 0, diff --git a/tests/integration/test_job_log_capture.py b/tests/integration/test_job_log_capture.py new file mode 100644 index 0000000..0940cec --- /dev/null +++ b/tests/integration/test_job_log_capture.py @@ -0,0 +1,292 @@ +""" +Integration tests for capturing job output to per-process log files. +""" + +import asyncio +import logging +import os +import sys +from typing import List + +import pytest + +# Add the parent directory to the path +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 LocalExecutor # noqa: E402 +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 ( # noqa: E402 + Command, + Context, + InputProcSub, + OutputProcSub, + Pipeline, +) + + +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(dag: DAG, log_dir, cores: int = 2): + """Execute a DAG with its output captured under ``log_dir``""" + run_logs = RunLogs(log_dir) + run_logs.create_dirs() + executor = LocalExecutor(ThreadScheduler(dag, cores), run_logs=run_logs) + executor.execute() + return executor, run_logs + + +def _task_dir(run_logs: RunLogs, task_name: str): + return run_logs.task_logs / task_name + + +def test_each_pipeline_stage_gets_its_own_stderr_log(tmp_path): + """Every stage's stderr is captured; only uncaptured stdout is.""" + dag = DAG() + job = Job( + Pipeline( + Command("sh", "-c", "echo payload; echo first >&2"), + Command("sh", "-c", "cat; echo second >&2"), + ), + "staged", + task_name="two-stage", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert executor.jobs_with_errors == [] + task_dir = _task_dir(run_logs, "two-stage") + assert sorted(p.name for p in task_dir.iterdir()) == [ + "staged-1.0.log", + "staged-1.1.log", + "staged-1.1.stdout.log", + ] + assert "first" in (task_dir / "staged-1.0.log").read_text() + assert "second" in (task_dir / "staged-1.1.log").read_text() + # The last stage's stdout was inherited, so it is captured verbatim + # (sharing the stage index of that stage's stderr log); the stdout + # piped between the stages is not. + assert (task_dir / "staged-1.1.stdout.log").read_text() == "payload\n" + + +def test_a_file_output_wins_over_the_sink(tmp_path): + """An explicit destination is never redirected into the log dir.""" + out = tmp_path / "out.txt" + dag = DAG() + job = Job( + Pipeline(Command("echo", "hello"), file_output=out), + "redirected", + task_name="file-out", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert executor.jobs_with_errors == [] + assert out.read_text() == "hello\n" + # The command was silent, so its header-only stderr log was pruned. + assert list(_task_dir(run_logs, "file-out").iterdir()) == [] + + +def test_input_proc_sub_stderr_is_captured(tmp_path): + """An inner <(...) command writes to the job's sink, not the terminal.""" + out = tmp_path / "out.txt" + dag = DAG() + job = Job( + Pipeline( + Command( + "cat", + InputProcSub( + Pipeline( + Command("sh", "-c", "echo inner >&2; echo payload") + ) + ), + ), + file_output=out, + ), + "reader", + task_name="proc-sub", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert executor.jobs_with_errors == [] + assert out.read_text() == "payload\n" + logs = list(_task_dir(run_logs, "proc-sub").iterdir()) + assert [p.name for p in logs] == ["reader-1.1.log"] + assert "inner" in logs[0].read_text() + + +def test_output_proc_sub_stderr_is_captured(tmp_path): + """An inner >(...) command writes to the job's sink too.""" + dag = DAG() + job = Job( + Pipeline( + Command( + "sh", + "-c", + 'echo payload > "$1"', + "sh", + OutputProcSub( + Pipeline(Command("sh", "-c", "cat >&2; echo done >&2")) + ), + ) + ), + "writer", + task_name="proc-sub", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert executor.jobs_with_errors == [] + contents = [ + path.read_text() for path in _task_dir(run_logs, "proc-sub").iterdir() + ] + assert any("payload" in text and "done" in text for text in contents) + + +def test_a_successful_job_prunes_its_empty_logs(tmp_path): + dag = DAG() + dag.add_job(Job(Pipeline(Command("true")), "quiet", task_name="cleanup")) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert executor.jobs_with_errors == [] + assert list(_task_dir(run_logs, "cleanup").iterdir()) == [] + + +def test_a_failing_job_is_reported_with_its_log_and_tail(tmp_path, messages): + dag = DAG() + job = Job( + Pipeline(Command("sh", "-c", "echo boom >&2; exit 3")), + "boomer", + task_name="failing", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert job in executor.jobs_with_errors + log_path = _task_dir(run_logs, "failing") / "boomer-1.0.log" + assert log_path.is_file() # nothing is pruned after a failure + assert "boom" in log_path.read_text() + + report = "\n".join(messages) + pid = job.shell.nodes[0].proc.pid + assert f"Failed sub-command of {job}" in report + assert f"exit code: 3, pid: {pid}" in report + assert str(log_path) in report + assert " boom" in report + assert f"Task logs for debugging: {run_logs.task_logs}" in report + + +def test_a_failing_stage_reports_its_own_log(tmp_path, messages): + """The tail comes from the failing process, not a sibling stage.""" + dag = DAG() + job = Job( + Pipeline( + Command("sh", "-c", "echo upstream-noise >&2"), + Command("sh", "-c", "cat; echo downstream-boom >&2; exit 4"), + ), + "mixed", + task_name="failing", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert job in executor.jobs_with_errors + report = "\n".join(messages) + # Quoted log lines are indented; the sibling's output is not quoted. + assert " downstream-boom" in report + assert " upstream-noise" not in report + assert str(_task_dir(run_logs, "failing") / "mixed-1.1.log") in report + + +def test_a_job_that_cannot_start_points_at_its_partial_logs( + tmp_path, messages +): + dag = DAG() + job = Job( + Pipeline( + Command("sh", "-c", "echo started >&2; sleep 5"), + Command("no_such_cmd_zzz"), + ), + "unstartable", + task_name="launch-failure", + ) + dag.add_job(job) + executor, run_logs = _run(dag, tmp_path / "logs") + + assert job in executor.jobs_with_errors + report = "\n".join(messages) + # The stage that did spawn keeps its log, and the report names it. + log_path = _task_dir(run_logs, "launch-failure") / "unstartable-1.0.log" + assert log_path.is_file() + assert f"Logs from {job}" in report + assert str(log_path) in report + + +@pytest.mark.asyncio +async def test_an_interrupted_job_keeps_and_reports_its_logs( + tmp_path, messages +): + """Shutdown after an interrupt prunes nothing and names the logs.""" + run_logs = RunLogs(tmp_path / "logs") + run_logs.create_dirs() + job = Job( + Pipeline(Command("sleep", "30")), + "sleeper", + task_name="interrupted", + ) + sink = run_logs.job_sink(job) + context = Context(log_sink=sink) + context.file_handles.append(sink.open_stderr(Command("sleep", "30"))) + + async def _done() -> int: + return 0 + + executor = LocalExecutor(ThreadScheduler(DAG(), 1), run_logs=run_logs) + executor.running = [(job, context, asyncio.create_task(_done()), 0)] + await executor._shutdown() + + log_path = sink.log_paths()[0] + assert log_path.is_file() + assert str(log_path) in "\n".join(messages) + + +def test_without_run_logs_no_files_are_written(tmp_path): + """An executor with no log directory behaves exactly as before.""" + out = tmp_path / "out.txt" + dag = DAG() + dag.add_job( + Job( + Pipeline(Command("echo", "hi"), file_output=out), + "plain", + task_name="no-logs", + ) + ) + executor = LocalExecutor(ThreadScheduler(dag, 1)) + executor.execute() + + assert executor.jobs_with_errors == [] + assert [p.name for p in tmp_path.iterdir()] == ["out.txt"] diff --git a/tests/integration/test_signals.py b/tests/integration/test_signals.py index 8aa2042..665a935 100644 --- a/tests/integration/test_signals.py +++ b/tests/integration/test_signals.py @@ -35,7 +35,11 @@ def _echo_dag(out): dag = DAG() - job = Job(Pipeline(Command("echo", "hi"), file_output=out), "echo") + job = Job( + Pipeline(Command("echo", "hi"), file_output=out), + "echo", + task_name="test", + ) dag.add_job(job) return dag @@ -103,7 +107,9 @@ async def cleanup(self): monkeypatch.setattr(executor_mod, "Context", SpyContext) dag = DAG() - dag.add_job(Job(Pipeline(Command("sleep", "30")), "sleeper")) + dag.add_job( + Job(Pipeline(Command("sleep", "30")), "sleeper", task_name="test") + ) executor = LocalExecutor( ThreadScheduler(dag, 1), install_signal_handlers=True, @@ -134,7 +140,11 @@ def test_grace_period_escalates_to_sigkill(): "time.sleep(30)" ) dag = DAG() - job = Job(Pipeline(Command(sys.executable, "-c", script)), "stubborn") + job = Job( + Pipeline(Command(sys.executable, "-c", script)), + "stubborn", + task_name="test", + ) dag.add_job(job) executor = LocalExecutor( ThreadScheduler(dag, 1), @@ -172,6 +182,7 @@ def test_sigint_with_blocked_procsub_cleans_up(): ) ), "blocked-procsub", + task_name="test", ) dag.add_job(job) executor = LocalExecutor( diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000..226413c --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,43 @@ +""" +Shared unit-test fixtures +""" + +import logging + +import pytest + +from sentieon_cli import logging as cli_logging +from sentieon_cli.job import Job + + +@pytest.fixture(autouse=True) +def reset_job_ids(): + """Restart job id numbering for every test. + + ``job_id`` comes from a class-level per-name counter that the CLI resets + once per invocation, so without this a test's ids would depend on which + tests ran before it. + """ + Job.reset_ids() + + +@pytest.fixture(autouse=True) +def no_handler_leaks(): + """Drop any handler a test leaves on the package logger. + + `setup_logging` attaches a `run.log` FileHandler to the package logger, + so any test that reaches it with an output VCF (or log_dir) set would + otherwise leak that handler into later tests. + """ + package_logger = logging.getLogger("sentieon_cli") + before = list(package_logger.handlers) + console_level = cli_logging._console_handler.level + yield + for handler in list(package_logger.handlers): + if handler in before: + continue + if isinstance(handler, logging.FileHandler): + cli_logging.remove_file_handler(handler) + else: + package_logger.removeHandler(handler) + cli_logging.set_console_level(console_level) diff --git a/tests/unit/test_dag.py b/tests/unit/test_dag.py index aa9d565..014f945 100644 --- a/tests/unit/test_dag.py +++ b/tests/unit/test_dag.py @@ -21,15 +21,17 @@ def _job(name, threads=1, resources=None): """A trivial job whose identity is its (unique) command name.""" - return Job(Pipeline(Command(name)), name, threads, resources) + return Job( + Pipeline(Command(name)), name, threads, resources, task_name="test" + ) def test_job_identity_is_the_pipeline(): # Two jobs with the same pipeline are equal even if named # differently; a DAG deduplicates on this identity. - j1 = Job(Pipeline(Command("echo", "x")), "name-a") - j2 = Job(Pipeline(Command("echo", "x")), "name-b") - j3 = Job(Pipeline(Command("echo", "y")), "name-c") + j1 = Job(Pipeline(Command("echo", "x")), "name-a", task_name="test") + j2 = Job(Pipeline(Command("echo", "x")), "name-b", task_name="test") + j3 = Job(Pipeline(Command("echo", "y")), "name-c", task_name="test") assert j1 == j2 assert hash(j1) == hash(j2) assert j1 != j3 @@ -101,17 +103,29 @@ def test_multilevel_chain(self): def test_add_duplicate_pipeline_raises(self): dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "x")), "step-1")) + dag.add_job( + Job(Pipeline(Command("echo", "x")), "step-1", task_name="test") + ) # A second job with an identical pipeline collides on identity. with pytest.raises(ValueError): - dag.add_job(Job(Pipeline(Command("echo", "x")), "step-2")) + dag.add_job( + Job(Pipeline(Command("echo", "x")), "step-2", task_name="test") + ) def test_jobs_differing_only_in_exec_kwargs_are_not_duplicates(self): # A command's env/cwd is part of its identity, so two jobs that # differ only there are distinct, not rejected as duplicates. dag = DAG() - j1 = Job(Pipeline(Command("run", exec_kwargs={"cwd": "/a"})), "a") - j2 = Job(Pipeline(Command("run", exec_kwargs={"cwd": "/b"})), "b") + j1 = Job( + Pipeline(Command("run", exec_kwargs={"cwd": "/a"})), + "a", + task_name="test", + ) + j2 = Job( + Pipeline(Command("run", exec_kwargs={"cwd": "/b"})), + "b", + task_name="test", + ) dag.add_job(j1) dag.add_job(j2) # must not raise "already in the DAG" assert j1 in dag.ready_jobs diff --git a/tests/unit/test_dag_construction.py b/tests/unit/test_dag_construction.py index e9bacdf..c0df2cf 100644 --- a/tests/unit/test_dag_construction.py +++ b/tests/unit/test_dag_construction.py @@ -5,6 +5,7 @@ import json import pathlib import pytest +import re import tempfile import shlex import sys @@ -20,6 +21,17 @@ from sentieon_cli.job import Job from sentieon_cli.shell_pipeline import Pipeline, Command +# task_name becomes a log directory name, so it must be kebab-case. +TASK_NAME_RE = re.compile(r"[a-z0-9]+(-[a-z0-9]+)*") + + +def assert_task_names(dag): + """Every job in a built DAG carries a kebab-case task_name""" + all_jobs = list(dag.waiting_jobs.keys()) + list(dag.ready_jobs.keys()) + assert all_jobs + for job in all_jobs: + assert TASK_NAME_RE.fullmatch(job.task_name), job.name + class TestDAGConstruction: """Test DAG construction for pipelines""" @@ -187,6 +199,18 @@ def test_gvcf_mode_dag_differences(self, mock_lib_preloaded): assert not any("gvcftyper" in name for name in job_names1) assert any("gvcftyper" in name for name in job_names2) + @patch('sentieon_cli.util.library_preloaded') + def test_jobs_have_task_names(self, mock_lib_preloaded): + """Every job built by the pipeline groups under a task_name""" + mock_lib_preloaded.return_value = True + + pipeline = self.create_basic_dnascope_pipeline() + pipeline.gvcf = True + pipeline.validate() + pipeline.configure() + + assert_task_names(pipeline.build_dag()) + class TestDAGJobProperties: """Test properties of individual jobs in the DAG""" @@ -198,6 +222,7 @@ def test_job_thread_allocation(self): Pipeline(Command(*shlex.split("sentieon driver --algo DNAscope"))), "variant-calling", 8, + task_name="variant-calling", ) assert job.threads == 8 @@ -206,6 +231,7 @@ def test_job_thread_allocation(self): Pipeline(Command("rm", "temp_file.vcf")), "cleanup", 0, + task_name="cleanup", ) assert job.threads == 0 @@ -217,6 +243,7 @@ def test_job_resource_requirements(self): "alignment", 4, resources={"node0": 1}, + task_name="alignment", ) assert job.resources == {"node0": 1} @@ -225,6 +252,7 @@ def test_job_resource_requirements(self): Pipeline(Command("simple-command")), "simple", 2, + task_name="test", ) assert job.resources == {} @@ -235,6 +263,7 @@ def test_job_failure_tolerance(self): Pipeline(Command("rm", "temp_files", fail_ok=True)), "cleanup", 0, + task_name="cleanup", ) assert job.shell.nodes[0].fail_ok is True @@ -243,6 +272,7 @@ def test_job_failure_tolerance(self): Pipeline(Command("sentieon", "driver")), "variant-calling", 4, + task_name="variant-calling", ) assert job.shell.nodes[0].fail_ok is False @@ -328,6 +358,9 @@ def test_longread_dag_complexity(self, mock_lib_preloaded): # Should have multiple phases of variant calling assert total_jobs > 10 # Rough estimate for complex phased calling + # Every job groups under a task_name, phasing jobs included + assert_task_names(dag) + @patch('sentieon_cli.util.library_preloaded') def test_ont_vs_hifi_dag_differences(self, mock_lib_preloaded): """Test DAG differences between ONT and HiFi technologies""" @@ -393,9 +426,9 @@ def test_dag_has_no_cycles(self): dag = DAG() # Create test jobs - job1 = Job(Pipeline(Command("command1")), "job1") - job2 = Job(Pipeline(Command("command2")), "job2") - job3 = Job(Pipeline(Command("command3")), "job3") + job1 = Job(Pipeline(Command("command1")), "job1", task_name="test") + job2 = Job(Pipeline(Command("command2")), "job2", task_name="test") + job3 = Job(Pipeline(Command("command3")), "job3", task_name="test") # Add jobs with dependencies: job1 -> job2 -> job3 dag.add_job(job1) @@ -416,9 +449,9 @@ def test_dag_execution_simulation(self): dag = DAG() # Create test jobs - job1 = Job(Pipeline(Command("command1")), "job1") - job2 = Job(Pipeline(Command("command2")), "job2") - job3 = Job(Pipeline(Command("command3")), "job3") + job1 = Job(Pipeline(Command("command1")), "job1", task_name="test") + job2 = Job(Pipeline(Command("command2")), "job2", task_name="test") + job3 = Job(Pipeline(Command("command3")), "job3", task_name="test") # Add jobs: job1 and job2 can run in parallel, then job3 dag.add_job(job1) diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py new file mode 100644 index 0000000..4b37a58 --- /dev/null +++ b/tests/unit/test_job.py @@ -0,0 +1,104 @@ +""" +Unit tests for Job identity: task_name, job_id, and id numbering. +""" + +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.dag import DAG # noqa: E402 +from sentieon_cli.job import Job # noqa: E402 +from sentieon_cli.shell_pipeline import Command, Pipeline # noqa: E402 + + +def _job(name, arg, task_name="test"): + """A job whose identity is its (unique) command argument.""" + return Job(Pipeline(Command("echo", arg)), name, task_name=task_name) + + +def test_task_name_is_required(): + with pytest.raises(TypeError): + Job( # type: ignore[call-arg] + Pipeline(Command("echo", "x")), "no-task" + ) + + +def test_ids_are_numbered_per_name(): + jobs = [_job("shard", str(i)) for i in range(3)] + assert [job.job_id for job in jobs] == ["shard-1", "shard-2", "shard-3"] + + +def test_each_name_has_its_own_counter(): + first = _job("dedup", "a") + other = _job("metrics", "b") + second = _job("dedup", "c") + + assert first.job_id == "dedup-1" + assert other.job_id == "metrics-1" + assert second.job_id == "dedup-2" + + +def test_reset_ids_restarts_the_sequence(): + assert _job("multiqc", "a").job_id == "multiqc-1" + assert _job("multiqc", "b").job_id == "multiqc-2" + + Job.reset_ids() + + assert _job("multiqc", "c").job_id == "multiqc-1" + + +def test_ids_keep_counting_across_two_dags(): + # The pangenome pipeline executes two DAGs in one process; ids must stay + # unique for the whole run, so they are never reset between DAGs. + def build_dag(tag): + dag = DAG() + for i in range(2): + dag.add_job(_job("calling", f"{tag}-{i}")) + return dag + + first = build_dag("a") + second = build_dag("b") + + ids = sorted( + job.job_id + for dag in (first, second) + for job in list(dag.ready_jobs) + list(dag.waiting_jobs) + ) + assert ids == ["calling-1", "calling-2", "calling-3", "calling-4"] + + +def test_task_name_and_job_id_are_not_part_of_identity(): + j1 = Job(Pipeline(Command("echo", "x")), "a", task_name="alignment") + j2 = Job(Pipeline(Command("echo", "x")), "b", task_name="cleanup") + + assert j1.job_id != j2.job_id + assert j1.task_name != j2.task_name + assert j1 == j2 + assert hash(j1) == hash(j2) + + +def test_jobs_with_different_task_names_still_collide_in_a_dag(): + dag = DAG() + dag.add_job(Job(Pipeline(Command("echo", "x")), "a", task_name="dedup")) + with pytest.raises(ValueError): + dag.add_job( + Job(Pipeline(Command("echo", "x")), "b", task_name="metrics") + ) + + +def test_repr_and_str_report_the_job_id(): + job = _job("locuscollector", "x") + assert repr(job) == "Job(locuscollector-1)" + assert str(job) == "Job(locuscollector-1)" + + +def test_task_name_is_stored(): + job = _job("dnascope", "x", task_name="variant-calling") + assert job.task_name == "variant-calling" diff --git a/tests/unit/test_job_logs.py b/tests/unit/test_job_logs.py new file mode 100644 index 0000000..e438533 --- /dev/null +++ b/tests/unit/test_job_logs.py @@ -0,0 +1,203 @@ +""" +Unit tests for the per-process job log files +""" + +import os +import sys + +# 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.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.shell_pipeline import ( # noqa: E402 + Command, + Context, + Pipeline, +) + + +def _sink(tmp_path, name: str = "bwa", task_name: str = "alignment"): + job = Job(Pipeline(Command("true")), name, task_name=task_name) + return RunLogs(tmp_path / "logs").job_sink(job) + + +def test_sanitize_keeps_safe_characters(): + assert sanitize("bwa-1.0_x") == "bwa-1.0_x" + assert sanitize("call vars/chr1:1-2") == "call-vars-chr1-1-2" + + +def test_stage_indices_follow_spawn_order_across_both_streams(tmp_path): + sink = _sink(tmp_path) + for handle in ( + sink.open_stderr(Command("a")), + sink.open_stdout(Command("b")), + sink.open_stderr(Command("c")), + ): + handle.close() + + assert [path.name for path in sink.log_paths()] == [ + "bwa-1.0.log", + "bwa-1.1.stdout.log", + "bwa-1.2.log", + ] + + +def test_both_streams_of_one_process_share_a_stage_index(tmp_path): + sink = _sink(tmp_path) + first = Command("a") + second = Command("b") + for handle in ( + sink.open_stderr(first), + sink.open_stdout(first), + sink.open_stderr(second), + sink.open_stdout(second), + ): + handle.close() + + assert [path.name for path in sink.log_paths()] == [ + "bwa-1.0.log", + "bwa-1.0.stdout.log", + "bwa-1.1.log", + "bwa-1.1.stdout.log", + ] + + +def test_the_task_directory_is_created_lazily(tmp_path): + sink = _sink(tmp_path) + assert not sink.log_dir.exists() + + sink.open_stderr(Command("true")).close() + + assert sink.log_dir == tmp_path / "logs" / "task_logs" / "alignment" + assert sink.log_dir.is_dir() + + +def test_path_components_are_sanitized(tmp_path): + sink = _sink(tmp_path, name="call vars/2", task_name="variant calling") + sink.open_stderr(Command("true")).close() + + path = sink.log_paths()[0] + assert path.parent.name == "variant-calling" + assert path.name == "call-vars-2-1.0.log" + + +def test_the_stderr_log_starts_with_a_header(tmp_path): + sink = _sink(tmp_path) + handle = sink.open_stderr(Command("echo", "hello world")) + handle.write(b"from the child\n") + handle.close() + + contents = sink.log_paths()[0].read_text() + assert "# timestamp: " in contents + assert "# task: alignment" in contents + assert "# job: bwa (bwa-1)" in contents + assert "# stage: 0" in contents + assert "# command: echo 'hello world'" in contents + # The header is flushed before the child inherits the FD, so child + # output always lands after it. + assert contents.splitlines()[-1] == "from the child" + + +def test_the_stdout_log_has_no_header(tmp_path): + sink = _sink(tmp_path) + handle = sink.open_stdout(Command("echo", "hello")) + handle.write(b"payload\n") + handle.close() + + assert sink.log_paths()[0].read_bytes() == b"payload\n" + + +def test_logs_are_matched_to_commands_by_identity(tmp_path): + sink = _sink(tmp_path) + first = Command("echo", "same") + second = Command("echo", "same") + assert first == second # equal, but two distinct processes + + sink.open_stderr(first).close() + sink.open_stderr(second).close() + + assert sink.stderr_log_for(first).name == "bwa-1.0.log" + assert sink.stderr_log_for(second).name == "bwa-1.1.log" + assert sink.stderr_log_for(Command("elsewhere")) is None + + +def test_stdout_logs_are_not_returned_as_stderr_logs(tmp_path): + sink = _sink(tmp_path) + command = Command("echo", "hi") + sink.open_stdout(command).close() + + assert sink.stderr_log_for(command) is None + + +def test_finalize_prunes_uninformative_logs_after_a_success(tmp_path): + sink = _sink(tmp_path) + header_only = Command("quiet") + empty_stdout = Command("quiet") + noisy = Command("loud") + sink.open_stderr(header_only).close() + sink.open_stdout(empty_stdout).close() + handle = sink.open_stderr(noisy) + handle.write(b"something happened\n") + handle.close() + + sink.finalize(success=True) + + assert not sink.stderr_log_for(header_only).exists() + assert not sink.log_paths()[1].exists() + assert sink.stderr_log_for(noisy).exists() + + +def test_finalize_keeps_every_log_after_a_failure(tmp_path): + sink = _sink(tmp_path) + sink.open_stderr(Command("quiet")).close() + sink.open_stdout(Command("quiet")).close() + + sink.finalize(success=False) + + assert all(path.exists() for path in sink.log_paths()) + + +def test_a_context_without_a_sink_captures_nothing(): + assert Context().log_sink is None + + +def test_log_tail_returns_the_last_lines(tmp_path): + log = tmp_path / "big.log" + log.write_text("".join(f"line {i}\n" for i in range(100))) + + tail = _log_tail(log) + + assert len(tail) == TAIL_LINES + assert tail[0] == "line 80" + assert tail[-1] == "line 99" + + +def test_log_tail_returns_a_short_file_whole(tmp_path): + log = tmp_path / "small.log" + log.write_text("one\ntwo\nthree\n") + + assert _log_tail(log) == ["one", "two", "three"] + + +def test_log_tail_only_reads_the_end_of_a_large_file(tmp_path): + log = tmp_path / "huge.log" + padding = "x" * 200 + with open(log, "w") as handle: + for i in range(2000): + handle.write(f"line {i} {padding}\n") + assert log.stat().st_size > 64 * 1024 + + tail = _log_tail(log) + + assert len(tail) == TAIL_LINES + assert tail[-1].startswith("line 1999 ") + assert not any(line.startswith("line 0 ") for line in tail) + + +def test_log_tail_of_a_missing_file_is_empty(tmp_path): + assert _log_tail(tmp_path / "gone.log") == [] diff --git a/tests/unit/test_pipeline_lifecycle.py b/tests/unit/test_pipeline_lifecycle.py index b9659c1..593a03f 100644 --- a/tests/unit/test_pipeline_lifecycle.py +++ b/tests/unit/test_pipeline_lifecycle.py @@ -131,9 +131,10 @@ def __init__(self, jobs_with_errors=None): def test_check_execution_names_the_failed_jobs(): pipeline = _DummyPipeline() pipeline.setup_logging(argparse.Namespace(loglevel="WARNING")) - job = Job(Pipeline(Command("false")), "broken-step") + job = Job(Pipeline(Command("false")), "broken-step", task_name="test") - with pytest.raises(DagExecutionError, match="broken-step"): + # Jobs report themselves by job_id, not by name. + with pytest.raises(DagExecutionError, match=r"Job\(broken-step-1\)"): pipeline.check_execution(DAG(), _StubExecutor([job])) @@ -141,7 +142,9 @@ def test_check_execution_flags_unexecuted_jobs(): pipeline = _DummyPipeline() pipeline.setup_logging(argparse.Namespace(loglevel="WARNING")) dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "x")), "unexecuted")) + dag.add_job( + Job(Pipeline(Command("echo", "x")), "unexecuted", task_name="test") + ) - with pytest.raises(DagExecutionError, match="unexecuted"): + with pytest.raises(DagExecutionError, match=r"Job\(unexecuted-1\)"): pipeline.check_execution(dag, _StubExecutor()) diff --git a/tests/unit/test_run_logs.py b/tests/unit/test_run_logs.py new file mode 100644 index 0000000..02b8eed --- /dev/null +++ b/tests/unit/test_run_logs.py @@ -0,0 +1,261 @@ +""" +Unit tests for the run log directory and the logging plumbing +""" + +import argparse +import logging +import os +import pathlib +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 import logging as cli_logging # noqa: E402 +from sentieon_cli.dag import DAG # noqa: E402 +from sentieon_cli.exceptions import DagExecutionError # noqa: E402 +from sentieon_cli.job import Job # noqa: E402 +from sentieon_cli.pipeline import BasePipeline # noqa: E402 +from sentieon_cli.run_logs import RunLogs # noqa: E402 +from sentieon_cli.shell_pipeline import Command, Pipeline # noqa: E402 +from sentieon_cli.util import __version__ # noqa: E402 + +PACKAGE_LOGGER = "sentieon_cli" + + +class _DummyPipeline(BasePipeline): + """A minimal concrete pipeline for exercising the logging plumbing.""" + + def validate(self) -> None: + pass + + def configure(self) -> None: + pass + + def build_dag(self) -> DAG: + return DAG() + + +class _FailingPipeline(_DummyPipeline): + """A pipeline whose DAG construction raises.""" + + def build_dag(self) -> DAG: + raise RuntimeError("boom") + + +class _ErroredExecutor: + """A stand-in executor that finished with failed jobs.""" + + def __init__(self, jobs: List[Job]) -> None: + self.jobs_with_errors = jobs + + +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(PACKAGE_LOGGER) + package_logger.addHandler(handler) + yield handler.messages + package_logger.removeHandler(handler) + + +def _args(loglevel: str = "INFO") -> argparse.Namespace: + return argparse.Namespace(loglevel=loglevel) + + +def test_module_loggers_delegate_to_the_package_logger(): + module_logger = cli_logging.get_logger("sentieon_cli.example") + + assert module_logger.handlers == [] + assert module_logger.propagate + assert not logging.getLogger(PACKAGE_LOGGER).propagate + + +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()) + + assert pipeline.run_logs is not None + assert pipeline.run_logs.log_dir == tmp_path / "sample_logs" + + +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()) + + assert pipeline.run_logs.log_dir == tmp_path / "a.vcf.gz.rerun_logs" + + +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()) + + assert pipeline.run_logs.log_dir == tmp_path / "elsewhere" + assert (tmp_path / "elsewhere" / "run.log").is_file() + + +def test_invalid_output_suffix_exits_before_creating_a_log_dir(tmp_path): + pipeline = _DummyPipeline() + pipeline.output_vcf = tmp_path / "sample.bcf" + + with pytest.raises(SystemExit) as excinfo: + pipeline.setup_logging(_args()) + + assert excinfo.value.code == 2 + assert list(tmp_path.iterdir()) == [] + + +def test_setup_wipes_stale_task_logs_but_keeps_the_log_dir(tmp_path): + log_dir = tmp_path / "logs" + stale = log_dir / "task_logs" / "alignment" + stale.mkdir(parents=True) + (stale / "bwa-1.0.log").write_text("from the previous run") + (log_dir / "unrelated.txt").write_text("keep me") + + RunLogs(log_dir).create_dirs() + + assert (log_dir / "unrelated.txt").read_text() == "keep me" + assert (log_dir / "task_logs").is_dir() + assert list((log_dir / "task_logs").iterdir()) == [] + + +def test_command_txt_records_the_invocation(tmp_path, monkeypatch): + monkeypatch.setattr( + sys, "argv", ["sentieon-cli", "dnascope", "a b.vcf.gz"] + ) + run_logs = RunLogs(tmp_path / "logs") + run_logs.create_dirs() + run_logs.write_command() + + contents = run_logs.command_txt.read_text() + assert "command: sentieon-cli dnascope 'a b.vcf.gz'" in contents + assert f"version: {__version__}" in contents + assert f"directory: {pathlib.Path.cwd()}" in contents + assert "timestamp: " in contents + + +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")) + logging.getLogger("sentieon_cli.example").debug("a debug record") + run_log = pipeline.run_logs.run_log + pipeline.run_logs.close() + + contents = run_log.read_text() + assert "a debug record" in contents + assert "Starting sentieon-cli version" in contents + assert f"Writing logs to: {pipeline.run_logs.log_dir}" in contents + assert cli_logging._console_handler.level == logging.INFO + + +def test_repeated_setup_does_not_accumulate_handlers(tmp_path): + package_logger = logging.getLogger(PACKAGE_LOGGER) + before = len(package_logger.handlers) + + pipeline = None + for i in range(3): + pipeline = _DummyPipeline() + pipeline.output_vcf = tmp_path / f"sample{i}.vcf.gz" + pipeline.setup_logging(_args()) + assert len(package_logger.handlers) == before + 1 + + pipeline.run_logs.close() + assert len(package_logger.handlers) == before + + +def test_rerun_truncates_the_previous_run_log(tmp_path): + log_dir = tmp_path / "logs" + pipeline = None + for _ in range(2): + pipeline = _DummyPipeline() + pipeline.log_dir = log_dir + pipeline.setup_logging(_args()) + run_log = pipeline.run_logs.run_log + pipeline.run_logs.close() + + banner = "Starting sentieon-cli version" + assert run_log.read_text().count(banner) == 1 + + +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()) + + assert pipeline.run_logs is None + assert list(tmp_path.iterdir()) == [] + + +def test_bare_pipeline_setup_logging_creates_nothing(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + pipeline = _DummyPipeline() + pipeline.setup_logging(_args()) + + assert pipeline.run_logs is None + assert list(tmp_path.iterdir()) == [] + + +def test_check_execution_names_the_task_log_dir(tmp_path): + pipeline = _DummyPipeline() + pipeline.log_dir = tmp_path / "logs" + pipeline.setup_logging(_args()) + job = Job(Pipeline(Command("false")), "boom", task_name="failing") + + with pytest.raises(DagExecutionError) as excinfo: + pipeline.check_execution(DAG(), _ErroredExecutor([job])) + pipeline.run_logs.close() + + message = str(excinfo.value) + assert "Job(boom-1)" in message + assert str(tmp_path / "logs" / "task_logs") in message + + +def test_end_of_run_message_on_success(tmp_path, monkeypatch, messages): + monkeypatch.setenv("SENTIEON_TMPDIR", str(tmp_path)) + pipeline = _DummyPipeline() + pipeline.dry_run = True + + pipeline.main(_args()) + + assert any("status: succeeded" in msg for msg in messages) + + +def test_end_of_run_message_on_failure(tmp_path, monkeypatch, messages): + monkeypatch.setenv("SENTIEON_TMPDIR", str(tmp_path)) + pipeline = _FailingPipeline() + pipeline.log_dir = tmp_path / "logs" + + with pytest.raises(RuntimeError, match="boom"): + pipeline.main(_args()) + + run_log = pipeline.run_logs.run_log + pipeline.run_logs.close() + + assert any("status: failed" in msg for msg in messages) + assert any(str(tmp_path / "logs") in msg for msg in messages) + assert "status: failed" in run_log.read_text() diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index aaecf0e..20327ae 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -21,7 +21,9 @@ def _job(name, threads=1, resources=None): - return Job(Pipeline(Command(name)), name, threads, resources) + return Job( + Pipeline(Command(name)), name, threads, resources, task_name="test" + ) def test_schedules_independent_jobs_together(): @@ -91,14 +93,22 @@ def test_resource_requirements_ignored_when_unmanaged(): def test_oversized_thread_request_is_rejected(): dag = DAG() - dag.add_job(Job(Pipeline(Command("echo", "hi")), "big", 100)) + dag.add_job( + Job(Pipeline(Command("echo", "hi")), "big", 100, task_name="test") + ) with pytest.raises(DagExecutionError): ThreadScheduler(dag, threads=2).start() def test_oversized_resource_request_is_rejected(): dag = DAG() - job = Job(Pipeline(Command("echo", "hi")), "greedy", 1, {"node0": 3}) + job = Job( + Pipeline(Command("echo", "hi")), + "greedy", + 1, + {"node0": 3}, + task_name="test", + ) dag.add_job(job) with pytest.raises(DagExecutionError): ThreadScheduler(dag, threads=8, resources={"node0": 1}).start() diff --git a/tests/unit/test_scheduler_contract.py b/tests/unit/test_scheduler_contract.py index 0460403..4454575 100644 --- a/tests/unit/test_scheduler_contract.py +++ b/tests/unit/test_scheduler_contract.py @@ -31,7 +31,9 @@ def _job(name, threads=1, resources=None): - return Job(Pipeline(Command(name)), name, threads, resources) + return Job( + Pipeline(Command(name)), name, threads, resources, task_name="test" + ) def drive(scheduler):