Fix bug with FluxExecutor jobs output (#3424)

currently, for some reason the jobs listing is passing tests, but in testing with a workflow there is an error warning printed to the terminal. This tweak suggested by Paolo seems to fix the issue. Signed-off-by: vsoch <[email protected]> Co-authored-by: vsoch <[email protected]>
nextflow-io · Nov 23, 2022 · 3711cef · 3711cef
1 parent aa974d4
commit 3711cef
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 6 deletions.
diff --git a/docs/executor.rst b/docs/executor.rst
@@ -103,6 +103,8 @@ Resource requests and other job characteristics can be controlled via the follow
 * :ref:`process-queue`
 * :ref:`process-time`
 
+Additionally, to have Flux print all output to stderr and stdout, add `flux.terminalOutput` to be true.
+
 .. note:: Flux does not support specifying memory. 
 
 

diff --git a/modules/nextflow/src/main/groovy/nextflow/executor/FluxExecutor.groovy b/modules/nextflow/src/main/groovy/nextflow/executor/FluxExecutor.groovy
@@ -61,7 +61,12 @@ class FluxExecutor extends AbstractGridExecutor {
         List<String> result = ['flux', 'mini', 'submit']
         result << '--setattr=cwd=' + quote(task.workDir)
         result << '--job-name="' + getJobNameFor(task) + '"'
-        result << '--output=' + quote(task.workDir.resolve(TaskRun.CMD_LOG))  // -o OUTFILE
+
+        // Only write output to file if user doesn't want written entirely to terminal
+        Boolean terminalOutput = task.config.navigate('flux.terminalOutput') as Boolean
+        if ( !terminalOutput ) {
+            result << '--output=' + quote(task.workDir.resolve(TaskRun.CMD_LOG))  // -o OUTFILE
+        }
 
         if( task.config.cpus > 1 ) {
             result << '--cores-per-task=' + task.config.cpus.toString()
@@ -129,17 +134,18 @@ class FluxExecutor extends AbstractGridExecutor {
     protected List<String> queueStatusCommand(Object queue) {
 
         // Look at jobs from last 15 minutes
-        final result = ['flux', 'jobs', '--suppress-header', '--format="{id.f58} {status_abbrev}"', '--since="-15m"']
+        String command = 'flux jobs --suppress-header --format="{id.f58} {status_abbrev}" --since="-15m"'
 
         if( queue )
-            result << '--queue' << queue.toString()
+            command += ' --queue=' + queue.toString()
 
         final user = System.getProperty('user.name')
         if( user )
-            result << '--user' << user
+            command += ' --user=' + user
         else
             log.debug "Cannot retrieve current user"
 
+        final result = ['sh', '-c', command]
         return result
     }
 

diff --git a/modules/nextflow/src/test/groovy/nextflow/executor/FluxExecutorTest.groovy b/modules/nextflow/src/test/groovy/nextflow/executor/FluxExecutorTest.groovy
@@ -89,6 +89,12 @@ class FluxExecutorTest extends Specification {
         task.config.clusterOptions = '--tasks-per-node=4 --cpus-per-node=4'
         then:
         executor.getSubmitCommandLine(task, Paths.get('/some/path/job.sh')) == ['flux', 'mini', 'submit', '--setattr=cwd=/work/path', '--job-name="nf-my_task"', '--output=/work/path/.command.log', '--time-limit=60', '--tasks-per-node=4', '--cpus-per-node=4', '/bin/bash', 'job.sh']
+
+        when:
+        task.config = new TaskConfig()
+        task.config.flux = [terminalOutput: true]
+        then:
+        executor.getSubmitCommandLine(task, Paths.get('/some/path/job.sh')) == ['flux', 'mini', 'submit', '--setattr=cwd=/work/path', '--job-name="nf-my_task"', '/bin/bash', 'job.sh']
     }
 
     def testWorkDirWithBlanks() {
@@ -144,7 +150,7 @@ class FluxExecutorTest extends Specification {
         def executor = [:] as FluxExecutor
         then:
         usr
-        executor.queueStatusCommand(null) == ['flux', 'jobs', '--suppress-header', '--format="{id.f58} {status_abbrev}"', '--since="-15m"', '--user', usr]
-        executor.queueStatusCommand('xxx') == ['flux', 'jobs', '--suppress-header', '--format="{id.f58} {status_abbrev}"', '--since="-15m"', '--queue', 'xxx', '--user', usr]
+        executor.queueStatusCommand(null) == ['sh', '-c', "flux jobs --suppress-header --format=\"{id.f58} {status_abbrev}\" --since=\"-15m\" --user=" + usr]
+        executor.queueStatusCommand('xxx') == ['sh', '-c', "flux jobs --suppress-header --format=\"{id.f58} {status_abbrev}\" --since=\"-15m\" --queue=xxx --user=" + usr]
     }
 }