[GH-25] Enable s3 work directory.

Allow user to set s3 work directory.  We will create an option
to tell MMC to mount the s3 bucket on the worker nodes.

README.md

@@ -53,7 +53,7 @@ Just make sure you have proper internet access.
 
 ```groovy
 plugins {
-    id 'nf-float@0.1.8'
+    id 'nf-float@0.2.0'
 }
 ```
 
@@ -66,7 +66,7 @@ Go to the folder where you just install the `nextflow` command line.
 Let's call this folder the Nextflow home directory.
 Create the float plugin folder with:
 ```bash
-mkdir -p .nextflow/plugins/nf-float-0.1.8
+mkdir -p .nextflow/plugins/nf-float-0.2.0
 ```
 where `0.1.8` is the version of the float plugin.  This version number should 
 align with the version in of your plugin and the property in your configuration
@@ -76,7 +76,7 @@ Retrieve your plugin zip file and unzip it in this folder.
 If everything goes right, you should be able to see two sub-folders:
 
 ```bash
-$ ll .nextflow/plugins/nf-float-0.1.8/
+$ ll .nextflow/plugins/nf-float-0.2.0/
 total 48
 drwxr-xr-x 4 ec2-user ec2-user    51 Jan  5 07:17 classes
 drwxr-xr-x 2 ec2-user ec2-user    25 Jan  5 07:17 META-INF
@@ -89,7 +89,7 @@ file with the command line option `-c`.  Here is a sample of the configuration.
 
 ```groovy
 plugins {
-    id 'nf-float@0.1.8'
+    id 'nf-float@0.2.0'
 }
 
 workDir = '/mnt/memverge/shared'

conf/float-rt.conf

@@ -1,5 +1,5 @@
 plugins {
-    id 'nf-float@0.1.8'
+    id 'nf-float@0.2.0'
 }
 
 workDir = '/mnt/memverge/shared'

plugins/nf-float/src/main/com/memverge/nextflow/FloatConf.groovy

@@ -16,30 +16,36 @@
 package com.memverge.nextflow
 
 import groovy.transform.CompileStatic
-import groovy.transform.Memoized
 import groovy.util.logging.Slf4j
 import nextflow.exception.AbortOperationException
-import nextflow.util.MemoryUnit
 import org.apache.commons.lang.StringUtils
 
+import java.nio.file.Path
+
 /**
  * @author Cedric Zhuang <cedric.zhuang@memverge.com>
  */
 @Slf4j
 @CompileStatic
 class FloatConf {
-    static String MMC_ADDRESS = "MMC_ADDRESS"
-    static String MMC_USERNAME = "MMC_USERNAME"
-    static String MMC_PASSWORD = "MMC_PASSWORD"
-    static String ADDR_SEP = ","
-    static String NF_JOB_ID = "nf-job-id"
+    static final String MMC_ADDRESS = "MMC_ADDRESS"
+    static final String MMC_USERNAME = "MMC_USERNAME"
+    static final String MMC_PASSWORD = "MMC_PASSWORD"
+    static final String ADDR_SEP = ","
+    static final String NF_JOB_ID = "nf-job-id"
+
+    static final String ACCESS_KEY = "access_key"
+    static final String SECRET_KEY = "secret_key";
 
     /** credentials for op center */
     String username
     String password
     Collection<String> addresses
     String nfs
 
+    String s3accessKey
+    String s3secretKey
+
     /** parameters for submitting the tasks */
     String commonExtra
 
@@ -56,38 +62,80 @@ class FloatConf {
     static FloatConf getConf(Map config) {
         FloatConf ret = new FloatConf()
 
-        if (!config || config.float !instanceof Map)
-            return ret
+        ret.initFloatConf(config.float as Map)
+        ret.initAwsConf(config.aws as Map)
+
+        return ret
+    }
+
+    String getDataVolume(URI workDir)  {
+        def scheme = workDir.getScheme()
+        if (scheme == "s3") {
+            def options = [ "mode=rw" ]
+            if (s3accessKey && s3secretKey) {
+                options.add("accessKey=" + s3accessKey)
+                options.add("secretKey=" + s3secretKey)
+            }
+            def optionsStr = options.join(",")
+            def path = "/${workDir.host}${workDir.path}"
+            return "[$optionsStr]$workDir:$path"
+        }
+        // local directory, need nfs support
+        if (!nfs) {
+            log.warn "local work directory need nfs support"
+            return ""
+        }
+        if (nfs.split(":").size() > 2) {
+            // already have mount point
+            return nfs
+        }
+        return "$nfs:${workDir.path}"
+    }
 
-        Map node = config.float as Map
-        ret.username = node.username ?: System.getenv(MMC_USERNAME)
-        ret.password = node.password ?: System.getenv(MMC_PASSWORD)
-        if (node.address instanceof Collection) {
-            ret.addresses = node.address.collect { it.toString() }
+    private def initFloatConf(Map floatNode) {
+        if (!floatNode) {
+            return
+        }
+        this.username = floatNode.username ?: System.getenv(MMC_USERNAME)
+        this.password = floatNode.password ?: System.getenv(MMC_PASSWORD)
+        if (floatNode.address instanceof Collection) {
+            this.addresses = floatNode.address.collect { it.toString() }
         } else {
-            String address = node.address ?: System.getenv(MMC_ADDRESS) ?: ""
-            ret.addresses = address
+            String address = floatNode.address ?: System.getenv(MMC_ADDRESS) ?: ""
+            this.addresses = address
                     .tokenize(ADDR_SEP)
                     .collect { it.trim() }
                     .findAll { it.size() > 0 }
         }
-        ret.nfs = node.nfs
-        ret.commonExtra = node.commonExtra
-
-        if (node.cpu)
-            log.warn "Config option `float.cpu` is no longer supported, use `process.cpus` instead"
-        if (node.cpus)
-            log.warn "Config option `float.cpus` is no longer supported, use `process.cpus` instead"
-        if (node.mem)
-            log.warn "Config option `float.mem` is no longer supported, use `process.memory` instead"
-        if (node.memory)
-            log.warn "Config option `float.memory` is no longer supported, use `process.memory` instead"
-        if (node.image)
-            log.warn "Config option `float.image` is no longer supported, use `process.container` instead"
-        if (node.container)
-            log.warn "Config option `float.container` is no longer supported, use `process.container` instead"
+        this.nfs = floatNode.nfs
+        this.commonExtra = floatNode.commonExtra
 
-        return ret
+        if (floatNode.cpu)
+            warnDeprecated("float.cpu", "process.cpus")
+        if (floatNode.cpus)
+            warnDeprecated("float.cpus", "process.cpus")
+        if (floatNode.mem)
+            warnDeprecated("float.mem", "process.memory")
+        if (floatNode.memory)
+            warnDeprecated("float.memory", "process.memory")
+        if (floatNode.image)
+            warnDeprecated("float.image", "process.container")
+        if (floatNode.container)
+            warnDeprecated("float.container", "process.container")
+    }
+
+    private static def warnDeprecated(String deprecated, String replacement) {
+        log.warn "[flaot] config option `$deprecated` " +
+                "is no longer supported, " +
+                "use `$replacement` instead"
+    }
+
+    private def initAwsConf(Map awsNode) {
+        if (!awsNode) {
+            return
+        }
+        s3accessKey = awsNode.accessKey ?: System.getenv(ACCESS_KEY)
+        s3secretKey = awsNode.secretKey ?: System.getenv(SECRET_KEY)
     }
 
     void validate() {