Implement access for Google Cloud buckets with Requester Pays set - nextflow-io#1466

Signed-off-by: Emilio Palumbo <[email protected]>

Author: emi80

docs/google.rst

@@ -169,7 +169,7 @@ When completed, shutdown the cluster instances by using the following command::
 
 Replace ``my-cluster`` with the name used in your execution.
 
-Preemptible instances 
+Preemptible instances
 ---------------------
 
 An optional parameter allows you to set the instance to be preemptible. Both master and worker instances can be set to
@@ -230,7 +230,7 @@ autoscaler. For example::
     }
 
 By doing so it is possible to create a cluster with a single node i.e. the master node. The autoscaler will then
-automatically add the missing instances, up to the number defined by the ``minInstances`` attributes. 
+automatically add the missing instances, up to the number defined by the ``minInstances`` attributes.
 
 
 Limitation
@@ -301,7 +301,7 @@ Example::
 .. warning:: Make sure to specify in the above setting the project ID not the project name.
 
 .. Note:: A container image must be specified to deploy the process execution. You can use a different Docker image for
-  each process using one or more :ref:`config-process-selectors`. 
+  each process using one or more :ref:`config-process-selectors`.
 
 Process definition
 ------------------
@@ -384,7 +384,7 @@ specify the local storage for the jobs computed locally::
     nextflow run <script or project name> -bucket-dir gs://my-bucket/some/path
 
 .. warning:: The Google Storage path needs to contain at least sub-directory. Don't use only the
-  bucket name e.g. ``gs://my-bucket``. 
+  bucket name e.g. ``gs://my-bucket``.
 
 Limitation
 ----------
@@ -470,21 +470,22 @@ Example::
 
 The following configuration options are available:
 
-======================================= =================
-Name                                    Description
-======================================= =================
-google.project                          The Google Project Id to use for the pipeline execution.
-google.region                           The Google *region* where the computation is executed in Compute Engine VMs. Multiple regions can be provided separating them by a comma. Do not specify if a zone is provided. See  `available Compute Engine regions and zones <https://cloud.google.com/compute/docs/regions-zones/>`_
-google.zone                             The Google *zone* where the computation is executed in Compute Engine VMs. Multiple zones can be provided separating them by a comma. Do not specify if a region is provided. See  `available Compute Engine regions and zones <https://cloud.google.com/compute/docs/regions-zones/>`_
-google.location                         The Google *location* where the job executions are deployed to Cloud Life Sciences API. See  `available Cloud Life Sciences API locations <https://cloud.google.com/life-sciences/docs/concepts/locations>`_ (default: the same as the region or the zone specified).
-google.lifeSciences.bootDiskSize        Set the size of the virtual machine boot disk e.g `50.GB` (default: none).
-google.lifeSciences.copyImage           The container image run to copy input and output files. It must include the ``gsutil`` tool (default: ``google/cloud-sdk:alpine``).
-google.lifeSciences.debug               When ``true`` copies the `/google` debug directory in that task bucket directory (defualt: ``false``)
-google.lifeSciences.preemptible         When ``true`` enables the usage of *preemptible* virtual machines or ``false`` otherwise (default: ``true``)
-google.lifeSciences.usePrivateAddress   When ``true`` the VM will NOT be provided with a public IP address, and only contain an internal IP. If this option is enabled, the associated job can only load docker images from Google Container Registry, and the job executable cannot use external services other than Google APIs (default: ``false``). Requires version `20.03.0-edge` or later.
-google.lifeSciences.sshDaemon           When ``true`` runs SSH daemon in the VM carrying out the job to which it's possible to connect for debugging purposes (default: ``false``).
-google.lifeSciences.sshImage            The container image used to run the SSH daemon (default: ``gcr.io/cloud-genomics-pipelines/tools``).
-======================================= =================
+============================================== =================
+Name                                           Description
+============================================== =================
+google.project                                 The Google Project Id to use for the pipeline execution.
+google.region                                  The Google *region* where the computation is executed in Compute Engine VMs. Multiple regions can be provided separating them by a comma. Do not specify if a zone is provided. See  `available Compute Engine regions and zones <https://cloud.google.com/compute/docs/regions-zones/>`_
+google.zone                                    The Google *zone* where the computation is executed in Compute Engine VMs. Multiple zones can be provided separating them by a comma. Do not specify if a region is provided. See  `available Compute Engine regions and zones <https://cloud.google.com/compute/docs/regions-zones/>`_
+google.location                                The Google *location* where the job executions are deployed to Cloud Life Sciences API. See  `available Cloud Life Sciences API locations <https://cloud.google.com/life-sciences/docs/concepts/locations>`_ (default: the same as the region or the zone specified).
+google.enableRequesterPaysBuckets              When ``true`` uses the configured Google project id as the billing project for storage access. This is required when accessing data from *reqester pays enabled* buckets. See `Requester Pays on Google Cloud Storage documentation  <https://cloud.google.com/storage/docs/requester-pays>`_ (default: ``false``)
+google.lifeSciences.bootDiskSize               Set the size of the virtual machine boot disk e.g `50.GB` (default: none).
+google.lifeSciences.copyImage                  The container image run to copy input and output files. It must include the ``gsutil`` tool (default: ``google/cloud-sdk:alpine``).
+google.lifeSciences.debug                      When ``true`` copies the `/google` debug directory in that task bucket directory (defualt: ``false``)
+google.lifeSciences.preemptible                When ``true`` enables the usage of *preemptible* virtual machines or ``false`` otherwise (default: ``true``)
+google.lifeSciences.usePrivateAddress          When ``true`` the VM will NOT be provided with a public IP address, and only contain an internal IP. If this option is enabled, the associated job can only load docker images from Google Container Registry, and the job executable cannot use external services other than Google APIs (default: ``false``). Requires version `20.03.0-edge` or later.
+google.lifeSciences.sshDaemon                  When ``true`` runs SSH daemon in the VM carrying out the job to which it's possible to connect for debugging purposes (default: ``false``).
+google.lifeSciences.sshImage                   The container image used to run the SSH daemon (default: ``gcr.io/cloud-genomics-pipelines/tools``).
+============================================== =================
 
 
 Process definition
@@ -617,6 +618,6 @@ Troubleshooting
 
 * Enable the optional SSH daemon in the job VM using the option ``google.lifeSciences.sshDaemon = true``
 
-* Make sure you are choosing a `location` where  `Cloud Life Sciences API is available <https://cloud.google.com/life-sciences/docs/concepts/locations>`_, 
+* Make sure you are choosing a `location` where  `Cloud Life Sciences API is available <https://cloud.google.com/life-sciences/docs/concepts/locations>`_,
   and a `region` or `zone` where `Compute Engine is available <https://cloud.google.com/compute/docs/regions-zones/>`_.
-  
+

modules/nf-google/src/main/nextflow/cloud/google/lifesciences/GoogleLifeSciencesConfig.groovy

@@ -21,11 +21,13 @@ import java.nio.file.Path
 
 import groovy.json.JsonSlurper
 import groovy.transform.CompileStatic
+import groovy.transform.Memoized
 import groovy.transform.ToString
 import groovy.util.logging.Slf4j
 import nextflow.Session
 import nextflow.exception.AbortOperationException
 import nextflow.util.MemoryUnit
+
 /**
  * Helper class wrapping configuration required for Google Pipelines.
  *
@@ -55,6 +57,7 @@ class GoogleLifeSciencesConfig {
     Integer debugMode
     String copyImage
     boolean usePrivateAddress
+    boolean enableRequesterPaysBuckets
 
     @Deprecated
     GoogleLifeSciencesConfig(String project, List<String> zone, List<String> region, Path remoteBinDir = null, boolean preemptible = false) {
@@ -71,6 +74,7 @@ class GoogleLifeSciencesConfig {
 
     GoogleLifeSciencesConfig() {}
 
+    @Memoized
     static GoogleLifeSciencesConfig fromSession(Session session) {
         try {
             fromSession0(session.config)
@@ -110,6 +114,7 @@ class GoogleLifeSciencesConfig {
         final copyImage = config.navigate('google.lifeSciences.copyImage', DEFAULT_COPY_IMAGE) as String
         final debugMode = config.navigate('google.lifeSciences.debug', System.getenv('NXF_DEBUG'))
         final privateAddr  = config.navigate('google.lifeSciences.usePrivateAddress') as boolean
+        final requesterPays = config.navigate('google.enableRequesterPaysBuckets') as boolean
 
         def zones = (config.navigate("google.zone") as String)?.split(",")?.toList() ?: Collections.<String>emptyList()
         def regions = (config.navigate("google.region") as String)?.split(",")?.toList() ?: Collections.<String>emptyList()
@@ -127,7 +132,8 @@ class GoogleLifeSciencesConfig {
                 copyImage: copyImage,
                 sshDaemon: sshDaemon,
                 sshImage: sshImage,
-                usePrivateAddress: privateAddr )
+                usePrivateAddress: privateAddr,
+                enableRequesterPaysBuckets: requesterPays)
     }
 
     static private Integer debugMode0(value) {
@@ -172,4 +178,4 @@ class GoogleLifeSciencesConfig {
             throw new AbortOperationException("Missing Google credentials file: $credsFilePath")
         }
     }
-}
+}

modules/nf-google/src/main/nextflow/cloud/google/lifesciences/GoogleLifeSciencesFileCopyStrategy.groovy

@@ -58,6 +58,13 @@ class GoogleLifeSciencesFileCopyStrategy extends SimpleFileCopyStrategy {
         final createDirectories  = []
         final stagingCommands = []
 
+        final gsutilPrefix = new StringBuilder()
+        gsutilPrefix.append("gsutil -m -q")
+
+        if(config.enableRequesterPaysBuckets) {
+            gsutilPrefix.append(" -u ${config.project}")
+        }
+
         for( String stageName : inputFiles.keySet() ) {
             final storePath = inputFiles.get(stageName)
             final storePathIsDir = storePath.isDirectory()
@@ -72,13 +79,13 @@ class GoogleLifeSciencesFileCopyStrategy extends SimpleFileCopyStrategy {
             }
 
             if(storePathIsDir) {
-                stagingCommands << "gsutil -m -q cp -R $escapedStoreUri/ $localTaskDir".toString()
+                stagingCommands << "$gsutilPrefix cp -R $escapedStoreUri/ $localTaskDir".toString()
                 //check if we need to move the directory (gsutil doesn't support renaming directories on copy)
                 if(parent || !storePath.toString().endsWith(stageName)) {
                     stagingCommands << "mv $localTaskDir/${Escape.path(storePath.name)} $localTaskDir/$escapedStageName".toString()
                 }
             } else {
-                stagingCommands << "gsutil -m -q cp $escapedStoreUri $localTaskDir/$escapedStageName".toString()
+                stagingCommands << "$gsutilPrefix cp $escapedStoreUri $localTaskDir/$escapedStageName".toString()
             }
         }
 
@@ -159,4 +166,4 @@ class GoogleLifeSciencesFileCopyStrategy extends SimpleFileCopyStrategy {
         return result.toString()
     }
 
-}
+}

modules/nf-google/src/main/nextflow/cloud/google/util/GsPathFactory.groovy

@@ -18,8 +18,11 @@ package nextflow.cloud.google.util
 
 import java.nio.file.Path
 
+import com.google.cloud.storage.contrib.nio.CloudStorageConfiguration
 import com.google.cloud.storage.contrib.nio.CloudStorageFileSystem
 import groovy.transform.CompileStatic
+import nextflow.Global
+import nextflow.cloud.google.lifesciences.GoogleLifeSciencesConfig
 import nextflow.file.FileSystemPathFactory
 
 /**
@@ -30,14 +33,32 @@ import nextflow.file.FileSystemPathFactory
 @CompileStatic
 class GsPathFactory extends FileSystemPathFactory {
 
+    @Lazy
+    static private final CloudStorageConfiguration storageConfig = {
+        return getCloudStorageConfig()
+    } ()
+
+    static private CloudStorageConfiguration getCloudStorageConfig() {
+        def session = Global.getSession() as nextflow.Session
+        if (!session)
+            new IllegalStateException("Cannot initialize GsPathFactory: missing session")
+
+        final config = GoogleLifeSciencesConfig.fromSession(session)
+        final builder = CloudStorageConfiguration.builder()
+        if (config.enableRequesterPaysBuckets) {
+            builder.userProject(config.project)
+        }
+        return builder.build()
+    }
+
     @Override
     protected Path parseUri(String uri) {
         if( !uri.startsWith('gs://') )
             return null
         final str = uri.substring(5)
         final p = str.indexOf('/')
         return ( p==-1
-                ? CloudStorageFileSystem.forBucket(str).getPath('')
-                : CloudStorageFileSystem.forBucket(str.substring(0,p)).getPath(str.substring(p)) )
+                ? CloudStorageFileSystem.forBucket(str, storageConfig).getPath('')
+                : CloudStorageFileSystem.forBucket(str.substring(0,p), storageConfig).getPath(str.substring(p)) )
     }
 }

modules/nf-google/src/test/nextflow/cloud/google/lifesciences/GoogleLifeSciencesConfigTest.groovy

@@ -228,4 +228,17 @@ class GoogleLifeSciencesConfigTest extends Specification {
         'foo'           | 'us-central1'
     }
 
+    def 'should set requester pays' () {
+        when:
+        def config = GoogleLifeSciencesConfig.fromSession0([google:[project:'foo', region:'x', lifeSciences: [:]]])
+        then:
+        config.enableRequesterPaysBuckets == false
+
+        when:
+        config = GoogleLifeSciencesConfig.fromSession0([google:[project:'foo', region:'x', enableRequesterPaysBuckets:true]])
+        then:
+        config.enableRequesterPaysBuckets == true
+
+    }
+
 }

modules/nf-google/src/test/nextflow/cloud/google/lifesciences/GoogleLifeSciencesFileCopyStrategyTest.groovy

@@ -101,6 +101,41 @@ class GoogleLifeSciencesFileCopyStrategyTest extends GoogleSpecification {
                 '''.stripIndent()
     }
 
+    def 'create stage files using Requester Pays' () {
+        given:
+        def bean = Mock(TaskBean) {
+            getWorkDir() >> mockGsPath('gs://my-bucket/work/xx/yy')
+        }
+        def handler = Mock(GoogleLifeSciencesTaskHandler) {
+            getExecutor() >> Mock(GoogleLifeSciencesExecutor) {
+                getConfig() >> GoogleLifeSciencesConfig.fromSession0([google:[project:'foo', region:'x', enableRequesterPaysBuckets:true]])
+            }
+        }
+        and:
+        def strategy = new GoogleLifeSciencesFileCopyStrategy(bean, handler)
+
+        // file with the same name
+        when:
+        def inputs = ['foo.txt': mockGsPath('gs://my-bucket/bar/foo.txt')]
+        def result = strategy.getStageInputFilesScript(inputs)
+        then:
+        result == '''\
+                echo start | gsutil -q cp  -c - gs://my-bucket/work/xx/yy/.command.begin
+                gsutil -m -q -u foo cp gs://my-bucket/bar/foo.txt /work/xx/yy/foo.txt
+                '''.stripIndent()
+
+        // file a directory name
+        when:
+        inputs = ['dir1': mockGsPath('gs://my-bucket/foo/dir1', true)]
+        result = strategy.getStageInputFilesScript(inputs)
+        then:
+        result == '''\
+                echo start | gsutil -q cp  -c - gs://my-bucket/work/xx/yy/.command.begin
+                gsutil -m -q -u foo cp -R gs://my-bucket/foo/dir1/ /work/xx/yy
+                '''.stripIndent()
+
+    }
+
     def 'should unstage files' () {
         given:
         def bean = Mock(TaskBean) {

modules/nf-google/src/test/nextflow/cloud/google/util/GsPathPathFactoryTest.groovy

@@ -18,6 +18,14 @@ package nextflow.cloud.google.util
 
 
 import spock.lang.Specification
+
+import com.google.cloud.storage.contrib.nio.CloudStorageConfiguration
+import com.google.cloud.storage.contrib.nio.CloudStorageFileSystem
+import nextflow.Global
+import nextflow.Session
+import nextflow.cloud.google.lifesciences.GoogleLifeSciencesConfig
+import nextflow.config.ConfigBuilder
+
 /**
  *
  * @author Paolo Di Tommaso <[email protected]>
@@ -26,6 +34,9 @@ class GsPathPathFactoryTest extends Specification {
 
     def 'should create gs path' () {
         given:
+        def sess = new Session()
+        sess.config = [google:[project:'foo', region:'x']]
+        Global.session = sess
         def factory = new GsPathFactory()
 
         expect:
@@ -39,4 +50,30 @@ class GsPathPathFactoryTest extends Specification {
         _ | 'gs://f o o/bar'
         _ | 'gs://f_o_o/bar'
     }
+
+    def 'should use requester pays' () {
+        given:
+        def sess = new Session()
+        sess.config = [google:[project:'foo', region:'x', enableRequesterPaysBuckets:true]]
+        Global.session = sess
+
+        when:
+        def storageConfig = GsPathFactory.getCloudStorageConfig()
+
+        then:
+        storageConfig.userProject() == 'foo'
+    }
+
+    def 'should not use requester pays' () {
+        given:
+        def sess = new Session()
+        sess.config = [google:[project:'foo', region:'x', lifeSciences: [:]]]
+        Global.session = sess
+
+        when:
+        def storageConfig = GsPathFactory.getCloudStorageConfig()
+
+        then:
+        storageConfig.userProject() == null
+    }
 }

modules/nf-google/src/test/nextflow/extension/FilesExTest2.groovy

@@ -22,6 +22,8 @@ import spock.lang.Unroll
 import java.nio.file.Path
 
 import com.google.cloud.storage.contrib.nio.CloudStoragePath
+import nextflow.Global
+import nextflow.Session
 
 /**
  *
@@ -33,6 +35,9 @@ class FilesExTest2 extends Specification {
     def 'should return uri string for #PATH' () {
 
         when:
+        def sess = new Session()
+        sess.config = [google:[project:'foo', region:'x']]
+        Global.session = sess
         def path = PATH as Path
         then:
         path instanceof CloudStoragePath

modules/nf-google/src/test/nextflow/file/FileHelperGsTest.groovy

@@ -22,6 +22,9 @@ import java.nio.file.Paths
 import com.google.cloud.storage.contrib.nio.CloudStorageFileSystem
 import spock.lang.Specification
 
+import nextflow.Global
+import nextflow.Session
+
 /**
  *
  * @author Paolo Di Tommaso <[email protected]>
@@ -30,6 +33,11 @@ class FileHelperGsTest extends Specification {
 
     def 'should parse google storage path' () {
 
+        given:
+        def sess = new Session()
+        sess.config = [google:[project:'foo', region:'x']]
+        Global.session = sess
+
         expect:
         FileHelper.asPath('file.txt') ==
                 Paths.get('file.txt')
@@ -44,7 +52,7 @@ class FileHelperGsTest extends Specification {
         and:
         FileHelper.asPath('gs://foo/b a r.txt') ==
                 CloudStorageFileSystem.forBucket('foo').getPath('/b a r.txt')
-        
+
         and:
         FileHelper.asPath('gs://f o o/bar.txt') ==
                 CloudStorageFileSystem.forBucket('f o o').getPath('/bar.txt')
@@ -57,6 +65,9 @@ class FileHelperGsTest extends Specification {
 
     def 'should strip ending slash' () {
         given:
+        def sess = new Session()
+        sess.config = [google:[project:'foo', region:'x']]
+        Global.session = sess
         def nxFolder = Paths.get('/my-bucket/foo')
         def nxNested = Paths.get('/my-bucket/foo/bar/')
         and: