Pass the genome fasta to ngs-samples; run it and the analysis workflow as one throttled step

Main/lib/perl/WorkflowSteps/MakeDnaSeqMergeExperimentsNextflowConfig.pm

@@ -85,6 +85,7 @@ params {
 // so maxForks only gates the few independent early branches. Serialize them to
 // keep peak memory flat (snpEff JVM, Julia processSeqVars).
 process {
+  beforeScript = 'module load apptainer/1.4.1 && unset LD_LIBRARY_PATH'
   maxForks = 1
 }
 

Main/lib/perl/WorkflowSteps/MakeDnaSeqNextflowConfig.pm

@@ -76,6 +76,7 @@ params {
 }
 
 process {
+  beforeScript = 'module load apptainer/1.4.1 && unset LD_LIBRARY_PATH'
   executor = '$executor'
   queue    = '$queue'
 }

Main/lib/perl/WorkflowSteps/MakeDnaSeqSingleExperimentNextflowConfig.pm

@@ -101,6 +101,7 @@ params {
 }
 
 process {
+  beforeScript = 'module load apptainer/1.4.1 && unset LD_LIBRARY_PATH'
   executor = \'$executor\'
   queue = \'$queue\'
   withName: 'gatk' {

Main/lib/perl/WorkflowSteps/MakeNgsSamplesNextflowConfig.pm

@@ -11,29 +11,38 @@ sub run {
   #NOTE: the subset size here would run "X" number of genomic sequences at a time on the cluster (chromosomes or contigs)
   my $fastaSubsetSize = 5;
 
+  my $queue = $self->getClusterQueue();
+
   my $finalDir = $self->getParamValue("finalDirectory");
   my $resultsDirectory = $self->getParamValue("resultsDirectory");
   my $analysisDirectory = $self->getParamValue("analysisDirectory");
 
   my $nextflowConfigFile = $self->getParamValue("nextflowConfigFile");
   my $sampleSheetName = $self->getParamValue("sampleSheetName");
   my $assayType = $self->getParamValue("assayType");
-  my $organismAbbrev = $self->getParamValue('organismAbbrev');
+  my $genomeFile = $self->getParamValue("genomeFile");
   my $fromSRA = $self->getBooleanParamValue("fromSRA") ? "true" : "false";
 
   # Max SRA file size prefetch will download. If a data load fails because prefetch
   # skipped a run for being "larger than maximum allowed", bump this in the generated
   # config on the cluster and re-run (default sra-tools limit is 20G).
   my $maxDownloadSize = "50G";
 
-  my $gusConfig = $self->getWorkflowDataDir() . "/" . $self->getParamValue('gusConfigFile');
-  my $genomeSize = $self->getGenomeSize($test, $organismAbbrev, $gusConfig);
+  # Depth gate. minOnTargetFraction doubles as the inflation cap: a 0.05 floor means a
+  # contaminated sample is asked for at most 20x the reads a clean one would get.
+  # Below minPlausibleFraction the sample is flagged as probably-wrong-genome, not failed.
+  my $targetCoverage = 60;
+  my $minOnTargetFraction = 0.05;
+  my $minPlausibleFraction = 0.01;
+  my $pilotSize = 100000;
+
   my $workflowDataDir = $self->getWorkflowDataDir();
 
   my $workingDirRelativePath = $self->getParamValue("workingDirRelativePath");
   my $digestedFinalDirPath = $self->relativePathToNextflowClusterPath($workingDirRelativePath, $finalDir);
   my $digestedAnalysisDirPath = $self->relativePathToNextflowClusterPath($workingDirRelativePath, $analysisDirectory);
   my $digestedOutputDir = $self->relativePathToNextflowClusterPath($workingDirRelativePath, $resultsDirectory);
+  my $digestedReferenceFasta = $self->relativePathToNextflowClusterPath($workingDirRelativePath, $genomeFile);
 
   my $clusterServer = $self->getSharedConfig('clusterServer');
   my $clusterWorkflowDataDir = $self->getClusterWorkflowDataDir();
@@ -79,14 +88,18 @@ params {
   samplesheetName = "$sampleSheetName"
   fromSra = $fromSRA
   outDir = "$digestedOutputDir"
-  genomeSize = $genomeSize
+  referenceFasta = "$digestedReferenceFasta"
   assayType = "$assayType"
+  targetCoverage = $targetCoverage
+  minOnTargetFraction = $minOnTargetFraction
+  minPlausibleFraction = $minPlausibleFraction
+  pilotSize = $pilotSize
   maxDownloadSize = "$maxDownloadSize"
 }
 
 process {
-  queue = '$queue'
   maxForks = $maxForks
+  queue = \'$queue\'
 }
 
 includeConfig "$clusterConfigFile"

Main/lib/perl/WorkflowSteps/RunNgsSamplesAndNextflow.pm

@@ -0,0 +1,124 @@
+package ApiCommonWorkflow::Main::WorkflowSteps::RunNgsSamplesAndNextflow;
+
+@ISA = (ReFlow::StepClasses::RunAndMonitorNextflow);
+
+use strict;
+use warnings;
+use ReFlow::StepClasses::RunAndMonitorNextflow;
+
+# ngs-samples-nextflow fetches the reads; the analysis workflow then consumes them.
+# Both run from this single step so the pair holds one throttle slot: as separate
+# steps the analysis had to re-queue behind other experiments' fetches.
+sub ngsSamplesWorkflow { return "VEuPathDB/ngs-samples-nextflow" }
+
+# Each run keeps its own job info, log, trace and stdout files.  Crash recovery
+# depends on that: a re-entered step detects an already-finished run from its log
+# and moves on to the next one.
+sub nextflowRuns {
+  my ($self) = @_;
+
+  my @runs;
+
+  # optional.  analyses with no samples to fetch leave it empty
+  my $ngsSamplesConfigFile = $self->getParamValue("ngsSamplesNextflowConfigFile");
+
+  push(@runs, { label                  => "ngs-samples",
+                workflow               => $self->ngsSamplesWorkflow(),
+                configFile             => $ngsSamplesConfigFile,
+                resultsDir             => $self->getParamValue("ngsSamplesResultsDir"),
+                entry                  => "",
+                clusterJobInfoFileName => "ngs-samples-clusterJobInfo.txt",
+                logFileName            => "ngs-samples-nextflow.log",
+                traceFileName          => "ngs-samples-trace.txt",
+                nextflowStdoutFileName => "ngs-samples-nextflow.txt",
+              }) if $ngsSamplesConfigFile;
+
+  push(@runs, { label                  => $self->getParamValue("nextflowWorkflow"),
+                workflow               => $self->getParamValue("nextflowWorkflow"),
+                configFile             => $self->getParamValue("nextflowConfigFile"),
+                resultsDir             => $self->getParamValue("resultsDir"),
+                entry                  => $self->getParamValue("entry"),
+                clusterJobInfoFileName => $self->clusterJobInfoFileName(),
+                logFileName            => $self->logFileName(),
+                traceFileName          => $self->traceFileName(),
+                nextflowStdoutFileName => $self->nextflowStdoutFileName(),
+              });
+
+  return @runs;
+}
+
+sub run {
+  my ($self, $test, $undo) = @_;
+
+  my $clusterServer = $self->getSharedConfig('clusterServer');
+  my $clusterTransferServer = $self->getSharedConfig('clusterFileTransferServer');
+  my $userName = $self->getSharedConfig("$clusterServer.clusterLogin");
+  my $clusterQueue = $self->getSharedConfig("$clusterServer.clusterQueue");
+  my $maxTimeMins = $self->getSharedConfig("$clusterServer.maxAllowedRuntimeDays") * 24 * 60;
+
+  my $isGitRepo = $self->getBooleanParamValue("isGitRepo");
+
+  my $workingDirRelativePath = $self->getParamValue("workingDirRelativePath");
+  my $clusterWorkingDir = $self->relativePathToNextflowClusterPath($workingDirRelativePath,
+                                                                   $self->getParamValue("workingDir"));
+
+  foreach my $nextflowRun ($self->nextflowRuns()) {
+
+    my $clusterResultsDir = $self->relativePathToNextflowClusterPath($workingDirRelativePath,
+                                                                     $nextflowRun->{resultsDir});
+    my $clusterNextflowConfigFile = $self->relativePathToNextflowClusterPath($workingDirRelativePath,
+                                                                            $nextflowRun->{configFile});
+
+    my $jobInfoFile = "$clusterWorkingDir/" . $nextflowRun->{clusterJobInfoFileName};
+    my $logFile = "$clusterWorkingDir/" . $nextflowRun->{logFileName};
+    my $traceFile = "$clusterWorkingDir/" . $nextflowRun->{traceFileName};
+    my $nextflowStdoutFile = "$clusterWorkingDir/" . $nextflowRun->{nextflowStdoutFileName};
+
+    if ($undo) {
+      $self->runCmdOnClusterTransferServer(0, "rm -fr $clusterWorkingDir/work");
+      $self->runCmdOnClusterTransferServer(0, "rm -fr $clusterResultsDir/*");
+      $self->runCmdOnClusterTransferServer(0, "rm -fr $traceFile");
+      $self->runCmdOnClusterTransferServer(0, "rm -fr $logFile");
+      $self->runCmdOnClusterTransferServer(0, "rm -fr $nextflowStdoutFile");
+      $self->log("Removing log file at: $logFile");
+      next;
+    }
+
+    my $success = $self->runAndMonitor($test, $userName, $clusterServer, $clusterTransferServer,
+                                       $jobInfoFile, $logFile, $nextflowStdoutFile, $clusterWorkingDir,
+                                       $maxTimeMins, $clusterQueue, $nextflowRun->{workflow}, $isGitRepo,
+                                       $clusterNextflowConfigFile, $nextflowRun->{entry});
+
+    $self->error($self->nextflowFailureMessage($nextflowRun->{label}, $logFile, $traceFile)) unless $success;
+
+    # remove the work directory
+    $self->runCmdOnClusterTransferServer(0, "rm -fr $clusterWorkingDir/work");
+  }
+}
+
+sub nextflowFailureMessage {
+  my ($self, $label, $logFile, $traceFile) = @_;
+
+  return
+"
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+The '$label' nextflow run in this step did not successfully run.  Check its task
+log file on the cluster:
+  $logFile
+
+If the task log file ends in a perl error, that suggests an unusual controller failure.  Often those are recoverable by setting the step to ready and trying again.
+
+Otherwise, to diagnose the problem, look in the scheduler and nextflow step logs to see what command is executed on the nodes.  Find those logs at:
+  $traceFile
+
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+";
+}
+
+sub getConfigDeclaration {
+  return (
+         # [name, default, description]
+         );
+}
+
+1;

Main/lib/xml/workflow/bulkrnaseq.xml

@@ -64,7 +64,6 @@
   </step>
 
   <step name="ngsSamplesNextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeNgsSamplesNextflowConfig">
-    <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
     <paramValue name="analysisDirectory">$$analysisDirectory$$</paramValue>
     <paramValue name="finalDirectory">$$finalSymLink$$</paramValue>
     <paramValue name="resultsDirectory">$$analysisDirectory$$/ngs-samples-results</paramValue>
@@ -73,8 +72,9 @@
     <paramValue name="sampleSheetName">samplesheet.csv</paramValue>
     <paramValue name="fromSRA">$$fromSRA$$</paramValue>
     <paramValue name="assayType">RNASeq</paramValue>
-    <paramValue name="organismAbbrev">$$organismAbbrev$$</paramValue>
+    <paramValue name="genomeFile">$$genomeSymLink$$</paramValue>
     <depends name="makeResultDir"/>
+    <depends name="symLinkGenome"/>
   </step>
 
   <step name="nextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeBulkRnaSeqNextflowConfig">
@@ -92,7 +92,7 @@
     <depends name="makeResultDir"/>
   </step>
 
-  <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml">
+  <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml" stepLoadTypes="runClusterTask:rnaseqNextflow">
     <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
     <paramValue name="projectName">$$projectName$$</paramValue>
     <paramValue name="parentDataDir">$$dataDir$$</paramValue>

Main/lib/xml/workflow/longReadRnaSeq.xml

@@ -63,7 +63,6 @@
     </step>
 
     <step name="ngsSamplesNextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeNgsSamplesNextflowConfig">
-        <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
         <paramValue name="analysisDirectory">$$analysisDirectory$$</paramValue>
         <paramValue name="finalDirectory">$$finalSymLink$$</paramValue>
         <paramValue name="resultsDirectory">$$analysisDirectory$$/ngs-samples-results</paramValue>
@@ -72,8 +71,9 @@
         <paramValue name="sampleSheetName">samplesheet.csv</paramValue>
         <paramValue name="fromSRA">$$fromSRA$$</paramValue>
         <paramValue name="assayType">RNASeq</paramValue>
-        <paramValue name="organismAbbrev">$$organismAbbrev$$</paramValue>
+        <paramValue name="genomeFile">$$genomeSymLink$$</paramValue>
         <depends name="makeResultDir"/>
+        <depends name="symLinkGenome"/>
     </step>
 
 
@@ -96,7 +96,7 @@
     </step>
 
 
-    <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml">
+    <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml" stepLoadTypes="runClusterTask:longReadRnaSeqNextflow">
         <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
         <paramValue name="projectName">$$projectName$$</paramValue>
         <paramValue name="parentDataDir">$$dataDir$$</paramValue>

Main/lib/xml/workflow/mapWithBowtie.xml

@@ -17,7 +17,6 @@
     <paramValue name="relativeWebServicesDir">$$relativeWebServicesDir$$</paramValue>
     <paramValue name="genomeExtDbRlsSpec">$$genomeExtDbRlsSpec$$</paramValue>
     <paramValue name="genomeFastaFile">$$genomeFastaFile$$</paramValue>
-    <depends name="makeDataDir"/>
   </subgraph>
 
 
@@ -57,7 +56,6 @@
     <paramValue name="relativeWebServicesDir">$$relativeWebServicesDir$$</paramValue>
     <paramValue name="genomeExtDbRlsSpec">$$genomeExtDbRlsSpec$$</paramValue>
     <paramValue name="genomeFastaFile">$$genomeFastaFile$$</paramValue>
-    <depends name="makeDataDir"/>
   </subgraph>
 
   <!-- moved from organismSpecificMiscNoAlias -->
@@ -71,6 +69,5 @@
     <paramValue name="genomeExtDbRlsSpec">$$genomeExtDbRlsSpec$$</paramValue>
     <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
     <paramValue name="genomeFastaFile">$$genomeFastaFile$$</paramValue>    
-    <depends name="makeDataDir"/>
   </subgraph>
   </workflowGraph>

Main/lib/xml/workflow/processChipSeqExperiment.xml

@@ -53,7 +53,6 @@
     </step>
 
     <step name="ngsSamplesNextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeNgsSamplesNextflowConfig">
-        <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
         <paramValue name="analysisDirectory">$$analysisDirectory$$</paramValue>
         <paramValue name="finalDirectory">$$finalSymLink$$</paramValue>
         <paramValue name="resultsDirectory">$$analysisDirectory$$/ngs-samples-results</paramValue>
@@ -62,8 +61,9 @@
         <paramValue name="sampleSheetName">samplesheet.csv</paramValue>
         <paramValue name="fromSRA">$$fromSRA$$</paramValue>
         <paramValue name="assayType">ChipSeq</paramValue>
-        <paramValue name="organismAbbrev">$$organismAbbrev$$</paramValue>
+        <paramValue name="genomeFile">$$genomeSymLink$$</paramValue>
         <depends name="makeResultDir"/>
+        <depends name="symLinkGenome"/>
     </step>
 
     <step name="nextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeChipSeqNextflowConfig">
@@ -83,7 +83,7 @@
         <depends name="makeResultDir"/>
     </step>
 
-    <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml">
+    <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml" stepLoadTypes="runClusterTask:chipseqNextflow">
         <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
         <paramValue name="projectName">$$projectName$$</paramValue>
         <paramValue name="parentDataDir">$$dataDir$$</paramValue>

Main/lib/xml/workflow/processOriginsExperiment.xml

@@ -53,7 +53,6 @@
     </step>
 
     <step name="ngsSamplesNextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeNgsSamplesNextflowConfig">
-        <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
         <paramValue name="analysisDirectory">$$analysisDirectory$$</paramValue>
         <paramValue name="finalDirectory">$$finalSymLink$$</paramValue>
         <paramValue name="resultsDirectory">$$analysisDirectory$$/ngs-samples-results</paramValue>
@@ -62,8 +61,9 @@
         <paramValue name="sampleSheetName">samplesheet.csv</paramValue>
         <paramValue name="fromSRA">$$fromSRA$$</paramValue>
         <paramValue name="assayType">DNASeq</paramValue>
-        <paramValue name="organismAbbrev">$$organismAbbrev$$</paramValue>
+        <paramValue name="genomeFile">$$genomeSymLink$$</paramValue>
         <depends name="makeResultDir"/>
+        <depends name="symLinkGenome"/>
     </step>
 
     <step name="nextflowConfig" stepClass="ApiCommonWorkflow::Main::WorkflowSteps::MakeOriginsOfReplicationNextflowConfig">
@@ -81,7 +81,7 @@
         <depends name="makeResultDir"/>
     </step>
 
-    <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml">
+    <subgraph name="runNextflowOnCluster" xmlFile="runNextflowOnCluster.xml" stepLoadTypes="runClusterTask:originsNextflow">
         <paramValue name="gusConfigFile">$$gusConfigFile$$</paramValue>
         <paramValue name="projectName">$$projectName$$</paramValue>
         <paramValue name="parentDataDir">$$dataDir$$</paramValue>