Add Fusion support for custom S3 endpoints

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>
nextflow-io · Nov 18, 2022 · fba9b64 · fba9b64
1 parent 41021cb
commit fba9b64
Show file tree

Hide file tree

Showing 5 changed files with 109 additions and 4 deletions.
diff --git a/docs/fusion.rst b/docs/fusion.rst
@@ -75,11 +75,11 @@ The following configuration should be added in your Nextflow configuration file:
 
     docker {
       enabled = true
-      envWhitelist = 'AWS_ACCESS_KEY_ID,AWS_SECRET_ACCESS_KEY'
     }
 
     fusion {
       enabled = true
+      exportAwsAccessKeys = true
     }
 
     wave {

diff --git a/modules/nextflow/src/main/groovy/nextflow/executor/fusion/FusionScriptLauncher.groovy b/modules/nextflow/src/main/groovy/nextflow/executor/fusion/FusionScriptLauncher.groovy
@@ -22,7 +22,9 @@ import static nextflow.executor.fusion.FusionHelper.*
 import java.nio.file.Path
 
 import groovy.transform.CompileStatic
+import groovy.transform.Memoized
 import groovy.util.logging.Slf4j
+import nextflow.Global
 import nextflow.executor.BashWrapperBuilder
 import nextflow.processor.TaskBean
 import nextflow.processor.TaskRun
@@ -97,6 +99,14 @@ class FusionScriptLauncher extends BashWrapperBuilder {
             final result = new LinkedHashMap(10)
             result.NXF_FUSION_WORK = work
             result.NXF_FUSION_BUCKETS = buckets
+            final endpoint = Global.getAwsS3Endpoint()
+            final creds = exportAwsAccessKeys() ? Global.getAwsCredentials() : Collections.<String>emptyList()
+            if( creds ) {
+                result.AWS_ACCESS_KEY_ID = creds[0]
+                result.AWS_SECRET_ACCESS_KEY = creds[1]
+            }
+            if( endpoint )
+                result.AWS_S3_ENDPOINT = endpoint
             env = result
         }
         return env
@@ -117,4 +127,12 @@ class FusionScriptLauncher extends BashWrapperBuilder {
         return remoteWorkDir.resolve(TaskRun.CMD_INFILE)
     }
 
+    boolean exportAwsAccessKeys() {
+        exportAwsAccessKeys0()
+    }
+
+    @Memoized
+    protected boolean exportAwsAccessKeys0() {
+        return Global.config?.navigate('fusion.exportAwsAccessKeys', false)
+    }
 }
diff --git a/modules/nextflow/src/test/groovy/nextflow/executor/fusion/FusionScriptLauncherTest.groovy b/modules/nextflow/src/test/groovy/nextflow/executor/fusion/FusionScriptLauncherTest.groovy
@@ -19,6 +19,8 @@ package nextflow.executor.fusion
 
 import java.nio.file.Path
 
+import nextflow.Global
+import nextflow.SysEnv
 import nextflow.file.http.XPath
 import nextflow.processor.TaskBean
 import spock.lang.Specification
@@ -64,6 +66,69 @@ class FusionScriptLauncherTest extends Specification {
                                NXF_FUSION_WORK: '/fusion/http/foo/work']
     }
 
+    def 'should get fusion env with s3 endpoint' () {
+        given:
+        SysEnv.push([AWS_S3_ENDPOINT: 'http://foo.com'])
+        and:
+        def fusion = new FusionScriptLauncher(
+                scheme: 'http',
+                buckets: ['foo'] as Set,
+                remoteWorkDir: XPath.get('http://foo/work'))
+
+        expect:
+        fusion.fusionEnv() == [AWS_S3_ENDPOINT: 'http://foo.com',
+                               NXF_FUSION_BUCKETS: 'http://foo',
+                               NXF_FUSION_WORK: '/fusion/http/foo/work']
+
+        cleanup:
+        SysEnv.pop()
+    }
+
+    def 'should get fusion env with aws credentials' () {
+        given:
+        SysEnv.push([AWS_ACCESS_KEY_ID: 'xxx', AWS_SECRET_ACCESS_KEY: 'zzz'])
+        Global.config = [fusion: [exportAwsAccessKeys: true]]
+        and:
+        def fusion = new FusionScriptLauncher(
+                scheme: 'http',
+                buckets: ['foo'] as Set,
+                remoteWorkDir: XPath.get('http://foo/work'))
+
+        expect:
+        fusion.fusionEnv() == [AWS_ACCESS_KEY_ID: 'xxx',
+                               AWS_SECRET_ACCESS_KEY: 'zzz',
+                               NXF_FUSION_BUCKETS: 'http://foo',
+                               NXF_FUSION_WORK: '/fusion/http/foo/work']
+
+        cleanup:
+        Global.config = null
+        SysEnv.pop()
+    }
+
+    def 'should get fusion env with aws credentials in nextflow config' () {
+        given:
+        SysEnv.push([:])
+        and:
+        def CONFIG = [fusion: [exportAwsAccessKeys: true], aws: [accessKey: 'k1', secretKey: 's1', client: [endpoint: 'http://minio.com']]]
+        Global.config = CONFIG
+        and:
+        def fusion = new FusionScriptLauncher(
+                scheme: 'http',
+                buckets: ['foo'] as Set,
+                remoteWorkDir: XPath.get('http://foo/work'))
+
+        expect:
+        fusion.fusionEnv() == [AWS_ACCESS_KEY_ID: 'k1',
+                               AWS_SECRET_ACCESS_KEY: 's1',
+                               AWS_S3_ENDPOINT: 'http://minio.com',
+                               NXF_FUSION_BUCKETS: 'http://foo',
+                               NXF_FUSION_WORK: '/fusion/http/foo/work']
+
+        cleanup:
+        Global.config = null
+        SysEnv.pop()
+    }
+
     def 'should get header script' () {
         given:
         def fusion = new FusionScriptLauncher(scheme: 's3')

diff --git a/modules/nf-commons/src/main/nextflow/Global.groovy b/modules/nf-commons/src/main/nextflow/Global.groovy
@@ -28,7 +28,6 @@ import nextflow.util.IniFile
 import nextflow.util.MemoryUnit
 import org.apache.commons.lang.StringUtils
 import org.apache.commons.lang.exception.ExceptionUtils
-
 /**
  * Hold global variables
  *
@@ -162,7 +161,7 @@ class Global {
     }
 
     static String getAwsRegion(Map env=null, Map config=null) {
-        if( env==null ) env = System.getenv()
+        if( env==null ) env = SysEnv.get()
         if( config==null ) config = this.config
 
         // check nxf config file
@@ -191,7 +190,7 @@ class Global {
     }
 
     static List<String> getAwsCredentials() {
-        getAwsCredentials(System.getenv(), config)
+        getAwsCredentials(SysEnv.get(), config)
     }
 
     static Map<String,?> getAwsClientConfig() {
@@ -202,6 +201,14 @@ class Global {
         return null
     }
 
+    static String getAwsS3Endpoint() {
+        getAwsS3Endpoint0(SysEnv.get(), config ?: Collections.emptyMap())
+    }
+
+    static protected String getAwsS3Endpoint0(Map<String,String> env, Map<String,Object> config) {
+        config.navigate('aws.client.endpoint', env.get('AWS_S3_ENDPOINT'))
+    }
+
     /**
      * Convert configuration keys from camel-case notation (nextflow) to underscore
      * separated notation expected by the AWS client

diff --git a/modules/nf-commons/src/test/nextflow/GlobalTest.groovy b/modules/nf-commons/src/test/nextflow/GlobalTest.groovy
@@ -22,6 +22,8 @@ import java.nio.file.Files
 import nextflow.util.Duration
 import nextflow.util.MemoryUnit
 import spock.lang.Specification
+import spock.lang.Unroll
+
 /**
  *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
@@ -128,5 +130,18 @@ class GlobalTest extends Specification {
         Global.normalizeAwsClientConfig(config).upload_retry_sleep == '5000'
     }
 
+    @Unroll
+    def 'should get aws s3 endpoint' () {
+
+        expect:
+        Global.getAwsS3Endpoint0(ENV, CONFIG) == EXPECTED
+
+        where:
+        ENV                             | CONFIG                                    | EXPECTED
+        [:]                             | [:]                                       | null
+        [AWS_S3_ENDPOINT: 'http://foo'] | [:]                                       | 'http://foo'
+        [:]                             | [aws:[client:[endpoint: 'http://bar']]]   | 'http://bar'
+        [AWS_S3_ENDPOINT: 'http://foo'] | [aws:[client:[endpoint: 'http://bar']]]   | 'http://bar'  // <-- config should have priority
+    }
 
 }