feat: add remote bin support for TES in a workdir

plugins/nf-ga4gh/src/main/nextflow/ga4gh/tes/executor/TesBashBuilder.groovy

@@ -29,12 +29,12 @@ import nextflow.processor.TaskRun
 @CompileStatic
 class TesBashBuilder extends BashWrapperBuilder {
 
-    TesBashBuilder(TaskRun task) {
-        super(new TaskBean(task), new TesFileCopyStrategy())
+    TesBashBuilder(TaskRun task, String remoteBinDir) {
+        super(new TaskBean(task), new TesFileCopyStrategy(remoteBinDir))
     }
 
-    TesBashBuilder(TaskBean task) {
-        super(task, new TesFileCopyStrategy())
+    TesBashBuilder(TaskBean task, String remoteBinDir) {
+        super(task, new TesFileCopyStrategy(remoteBinDir))
     }
 
 }

plugins/nf-ga4gh/src/main/nextflow/ga4gh/tes/executor/TesExecutor.groovy

@@ -17,9 +17,11 @@
 package nextflow.ga4gh.tes.executor
 
 import groovy.transform.CompileStatic
+import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import nextflow.exception.AbortOperationException
 import nextflow.executor.Executor
+import nextflow.extension.FilesEx
 import nextflow.ga4gh.tes.client.ApiClient
 import nextflow.ga4gh.tes.client.api.TaskServiceApi
 import nextflow.processor.TaskHandler
@@ -30,6 +32,8 @@ import nextflow.util.Duration
 import nextflow.util.ServiceName
 import org.pf4j.ExtensionPoint
 
+import java.nio.file.Path
+
 /**
  * Experimental TES executor
  *
@@ -44,14 +48,15 @@ class TesExecutor extends Executor implements ExtensionPoint {
 
     private TaskServiceApi client
 
+    /**
+     * A path accessible to TES where executable scripts need to be uploaded
+     */
+    private Path remoteBinDir = null
+
     @Override
     protected void register() {
-        if( session.binDir && !session.binDir.empty() ) {
-            session.abort()
-            throw new AbortOperationException("ERROR: TES executor does not allow the use of custom scripts in the `bin` folder")
-        }
-
         super.register()
+        uploadBinDir()
 
         client = new TaskServiceApi( new ApiClient(basePath: getEndPoint()) )
     }
@@ -64,6 +69,22 @@ class TesExecutor extends Executor implements ExtensionPoint {
         client
     }
 
+    @PackageScope
+    Path getRemoteBinDir() {
+        remoteBinDir
+    }
+
+    protected void uploadBinDir() {
+        /*
+         * upload local binaries
+         */
+        if( session.binDir && !session.binDir.empty() && !session.disableRemoteBinDir ) {
+            def tempBin = getTempDir()
+            log.info "Uploading local `bin` scripts folder to ${tempBin.toUriString()}/bin"
+            remoteBinDir = FilesEx.copyTo(session.binDir, tempBin)
+        }
+    }
+
     protected String getEndPoint() {
         def result = session.getConfigAttribute('executor.tes.endpoint', 'http://localhost:8000')
         log.debug "[TES] endpoint=$result"

plugins/nf-ga4gh/src/main/nextflow/ga4gh/tes/executor/TesFileCopyStrategy.groovy

@@ -30,6 +30,11 @@ import nextflow.util.Escape
  */
 @CompileStatic
 class TesFileCopyStrategy implements ScriptFileCopyStrategy {
+    private String remoteBinDir
+
+    TesFileCopyStrategy(String remoteBinDir) {
+        this.remoteBinDir = remoteBinDir
+    }
 
     /**
      * {@inheritDoc}
@@ -97,7 +102,25 @@ class TesFileCopyStrategy implements ScriptFileCopyStrategy {
      */
     @Override
     String getEnvScript(Map env, boolean container) {
-        if(container) throw new UnsupportedOperationException()
-        TaskProcessor.bashEnvironmentScript(env,false)
+        if( container )
+            throw new UnsupportedOperationException("Parameter `container` not supported by ${this.class.simpleName}")
+
+        final result = new StringBuilder()
+        final copy = env ? new HashMap<String,String>(env) : Collections.<String,String>emptyMap()
+        final path = copy.containsKey('PATH')
+        // remove any external PATH
+        if( path )
+            copy.remove('PATH')
+        // when a remote bin directory is provide managed it properly
+        if( remoteBinDir ) {
+            result << "cp -r ${remoteBinDir}/* \$PWD/nextflow-bin/\n"
+            result << "chmod +x \$PWD/nextflow-bin/* || true\n"
+            result << "export PATH=\$PWD/nextflow-bin:\$PATH\n"
+        }
+        // finally render the environment
+        final envSnippet = TaskProcessor.bashEnvironmentScript(copy,false)
+        if( envSnippet )
+            result << envSnippet
+        return result.toString()
     }
 }

plugins/nf-ga4gh/src/main/nextflow/ga4gh/tes/executor/TesTaskHandler.groovy

@@ -17,6 +17,7 @@
 package nextflow.ga4gh.tes.executor
 
 import nextflow.ga4gh.tes.client.model.TesFileType
+import nextflow.processor.TaskBean
 
 import static nextflow.processor.TaskStatus.COMPLETED
 import static nextflow.processor.TaskStatus.RUNNING
@@ -53,7 +54,7 @@ class TesTaskHandler extends TaskHandler {
 
     final List<TesState> STARTED_STATUSES = [TesState.INITIALIZING, TesState.RUNNING, TesState.PAUSED] + COMPLETE_STATUSES
 
-    final TesExecutor executor
+    private TesExecutor executor
 
     private final Path exitFile
 
@@ -75,6 +76,9 @@ class TesTaskHandler extends TaskHandler {
 
     private String requestId
 
+    /** only for testing purpose -- do not use */
+    protected TesTaskHandler() {}
+
     TesTaskHandler(TaskRun task, TesExecutor executor) {
         super(task)
         this.executor = executor
@@ -90,6 +94,8 @@ class TesTaskHandler extends TaskHandler {
         this.traceFile = task.workDir.resolve(TaskRun.CMD_TRACE)
     }
 
+    protected String getRequestId() { requestId }
+
     @Override
     boolean checkIfRunning() {
 
@@ -148,7 +154,8 @@ class TesTaskHandler extends TaskHandler {
     void submit() {
 
         // create task wrapper
-        final bash = new TesBashBuilder(task)
+        String remoteBinDir = executor.getRemoteBinDir()
+        final bash = newTesBashBuilder(task, remoteBinDir)
         bash.build()
 
         final body = newTesTask()
@@ -159,7 +166,11 @@ class TesTaskHandler extends TaskHandler {
         status = TaskStatus.SUBMITTED
     }
 
-    protected final TesTask newTesTask() {
+    protected TesBashBuilder newTesBashBuilder(TaskRun task, String remoteBinDir) {
+        return new TesBashBuilder(task, remoteBinDir)
+    }
+
+    protected TesTask newTesTask() {
         // the cmd list to launch it
         def job = new ArrayList(BashWrapperBuilder.BASH) << wrapperFile.getName()
         List cmd = ['/bin/bash','-c', job.join(' ') + " &> $TaskRun.CMD_LOG" ]
@@ -195,6 +206,8 @@ class TesTaskHandler extends TaskHandler {
             body.addOutputsItem(outItem(fileName))
         }
 
+        body.setName(task.getName())
+
         // add the executor
         body.executors = [exec]
 

plugins/nf-ga4gh/src/test/nextflow/ga4gh/tes/executor/TesBashBuilderTest.groovy

@@ -41,7 +41,7 @@ class TesBashBuilderTest extends Specification {
                 name: 'Hello 1',
                 workDir: folder,
                 script: 'echo Hello world!',
-        ] as TaskBean )
+        ] as TaskBean, null)
         bash.build()
 
         then:

plugins/nf-ga4gh/src/test/nextflow/ga4gh/tes/executor/TesTaskHandlerTest.groovy

@@ -16,6 +16,12 @@
 
 package nextflow.ga4gh.tes.executor
 
+import nextflow.ga4gh.tes.client.api.TaskServiceApi
+import nextflow.ga4gh.tes.client.model.TesCreateTaskResponse
+import nextflow.ga4gh.tes.client.model.TesTask
+import nextflow.processor.TaskStatus
+
+import java.nio.file.Path
 import java.nio.file.Paths
 
 import nextflow.processor.TaskConfig
@@ -49,5 +55,33 @@ class TesTaskHandlerTest extends Specification {
 
     }
 
+    def 'should submit job' () {
+
+        given:
+        def executor = Mock(TesExecutor)
+        def task = Mock(TaskRun)
+        def bashBuilder = Mock(TesBashBuilder)
+        def client = Mock(TaskServiceApi)
+        def handler = Spy(TesTaskHandler)
+        handler.@client = client
+        handler.@executor = executor
+        handler.task = task
+
+        def req = Mock(TesTask)
+        def resp = Mock(TesCreateTaskResponse)
+
+        when:
+        handler.submit()
+        then:
+        1 * executor.getRemoteBinDir() >> Path.of("/work/bin")
+        1 * handler.newTesBashBuilder(task, "/work/bin") >> bashBuilder
+        1 * bashBuilder.build() >> null
+        1 * handler.newTesTask() >> req
+        1 * client.createTask(req) >> resp
+        1 * resp.getId() >> '12345'
+
+        handler.status == TaskStatus.SUBMITTED
+        handler.requestId == '12345'
+    }
 
 }