Add support for GCP disk resource

Merge #1244

docs/google.rst

@@ -300,6 +300,8 @@ Process definition
 ------------------
 Processes can be defined as usual and by default the ``cpus`` and ``memory`` directives are used to instantiate a custom
 machine type with the specified compute resources.  If ``memory`` is not specified, 1GB of memory is allocated per cpu.
+A persistent disk will be created with size corresponding to the ``disk`` directive.  If ``disk`` is not specified, the
+instance default is chosen to ensure reasonable I/O performance.
 
 The process ``machineType`` directive may optionally be used to specify a predifined Google Compute Platform `machine type <https://cloud.google.com/compute/docs/machine-types>`_
 If specified, this value overrides the ``cpus`` and ``memory`` directives.
@@ -311,6 +313,7 @@ Examples::
     process custom_resources_task {
         cpus 8
         memory '40 GB'
+        disk '200 GB'
 
         """
         <Your script here>
@@ -379,7 +382,7 @@ specify the local storage for the jobs computed locally::
 Limitation
 ----------
 
-* Currently it's not possible to specify a disk type and size different from the default ones assigned
+* Currently it's not possible to specify a disk type different from the default one assigned
   by the service depending the chosen instance type.
 
 

modules/nf-google/src/main/nextflow/cloud/google/pipelines/GooglePipelinesHelper.groovy

@@ -132,6 +132,7 @@ class GooglePipelinesHelper {
                 req.zone,
                 req.region,
                 req.diskName,
+                req.diskSizeGb,
                 [GooglePipelinesHelper.SCOPE_CLOUD_PLATFORM], req.preemptible)
     }
 
@@ -162,10 +163,11 @@ class GooglePipelinesHelper {
                 [ActionFlags.ALWAYS_RUN, ActionFlags.IGNORE_EXIT_STATUS])
     }
 
-    Resources configureResources(String machineType, String projectId, List<String> zone, List<String> region, String diskName, List<String> scopes = null, boolean preEmptible = false) {
+    Resources configureResources(String machineType, String projectId, List<String> zone, List<String> region, String diskName, Integer diskSizeGb=null, List<String> scopes = null, boolean preEmptible = false) {
 
         def disk = new Disk()
         disk.setName(diskName)
+        disk.setSizeGb(diskSizeGb)
 
         def serviceAccount = new ServiceAccount()
         if (scopes)

modules/nf-google/src/main/nextflow/cloud/google/pipelines/GooglePipelinesSubmitRequest.groovy

@@ -39,6 +39,8 @@ class GooglePipelinesSubmitRequest {
 
     String diskName
 
+    Integer diskSizeGb
+
     boolean preemptible
 
     String taskName

modules/nf-google/src/main/nextflow/cloud/google/pipelines/GooglePipelinesTaskHandler.groovy

@@ -325,6 +325,7 @@ class GooglePipelinesTaskHandler extends TaskHandler {
         req.zone = pipelineConfiguration.zone
         req.region = pipelineConfiguration.region
         req.diskName = diskName
+        req.diskSizeGb = task.config.disk?.giga
         req.preemptible = pipelineConfiguration.preemptible
         req.taskName = "nf-$task.hash"
         req.containerImage = task.container

modules/nf-google/src/test/nextflow/cloud/google/pipelines/GooglePipelinesHelperTest.groovy

@@ -133,8 +133,8 @@ class GooglePipelinesHelperTest extends Specification {
         def helper = new GooglePipelinesHelper()
 
         when:
-        def resources1 = helper.configureResources(type,projectId,zone,null,diskName,scopes,preEmptible)
-        def resources2 = helper.configureResources(type,projectId,null,region,diskName,scopes,preEmptible)
+        def resources1 = helper.configureResources(type,projectId,zone,null,diskName,100, scopes,preEmptible)
+        def resources2 = helper.configureResources(type,projectId,null,region,diskName,200,scopes,preEmptible)
         def resources3 = helper.configureResources(type,projectId,zone,null,diskName)
 
         then:
@@ -144,6 +144,7 @@ class GooglePipelinesHelperTest extends Specification {
             getZones() == zone
             getRegions() == null
             getVirtualMachine().getDisks().get(0).getName() == diskName
+            getVirtualMachine().getDisks().get(0).getSizeGb() == 100
             getVirtualMachine().getServiceAccount().getScopes() == scopes
             getVirtualMachine().getPreemptible() == preEmptible
         }
@@ -154,6 +155,7 @@ class GooglePipelinesHelperTest extends Specification {
             getZones() == null
             getRegions() == region
             getVirtualMachine().getDisks().get(0).getName() == diskName
+            getVirtualMachine().getDisks().get(0).getSizeGb() == 200
             getVirtualMachine().getServiceAccount().getScopes() == scopes
             getVirtualMachine().getPreemptible() == preEmptible
         }
@@ -163,6 +165,7 @@ class GooglePipelinesHelperTest extends Specification {
             getProjectId() == projectId
             getZones() == zone
             getVirtualMachine().getDisks().get(0).getName() == diskName
+            !getVirtualMachine().getDisks().get(0).getSizeGb()
             !getVirtualMachine().getServiceAccount().getScopes()
             !getVirtualMachine().getPreemptible()
         }
@@ -252,6 +255,7 @@ class GooglePipelinesHelperTest extends Specification {
         req.zone >> ['my-zone']
         req.region >> ['my-region']
         req.diskName >> 'my-disk'
+        req.diskSizeGb >> 500
         req.preemptible >> true
 
         1 * helper.configureResources(
@@ -260,6 +264,7 @@ class GooglePipelinesHelperTest extends Specification {
                 ['my-zone'],
                 ['my-region'],
                 'my-disk',
+                500,
                 [GooglePipelinesHelper.SCOPE_CLOUD_PLATFORM],
                 true )
 
@@ -269,6 +274,7 @@ class GooglePipelinesHelperTest extends Specification {
         res.getVirtualMachine().getPreemptible()
         res.getVirtualMachine().getMachineType() == 'n1-abc'
         res.getVirtualMachine().getDisks()[0].getName() == 'my-disk'
+        res.getVirtualMachine().getDisks()[0].getSizeGb() == 500
     }
 
     def 'should create main action' () {

modules/nf-google/src/test/nextflow/cloud/google/pipelines/GooglePipelinesTaskHandlerTest.groovy

@@ -25,6 +25,7 @@ import com.google.api.services.genomics.v2alpha1.model.Operation
 import nextflow.Session
 import nextflow.cloud.google.GoogleSpecification
 import nextflow.exception.ProcessUnrecoverableException
+import nextflow.processor.TaskConfig
 import nextflow.processor.TaskId
 import nextflow.processor.TaskRun
 import nextflow.processor.TaskStatus
@@ -139,6 +140,7 @@ class GooglePipelinesTaskHandlerTest extends GoogleSpecification {
         task.getWorkDir() >> workDir
         task.getHash() >> { CacheHelper.hasher('dummy').hash() }
         task.getContainer() >> 'my/image'
+        task.getConfig() >> new TaskConfig(disk: '250 GB')
 
         def handler = new GooglePipelinesTaskHandler(
                 pipelineConfiguration: config,
@@ -161,6 +163,7 @@ class GooglePipelinesTaskHandlerTest extends GoogleSpecification {
         req.zone == ['my-zone']
         req.region == ['my-region']
         req.diskName == GooglePipelinesTaskHandler.diskName
+        req.diskSizeGb == 250
         req.preemptible == true
         req.taskName == "nf-bad893071e9130b866d43a4fcabb95b6"
         req.containerImage == 'my/image'
@@ -193,6 +196,7 @@ class GooglePipelinesTaskHandlerTest extends GoogleSpecification {
         task.getWorkDir() >> workDir
         task.getHash() >> { CacheHelper.hasher('dummy').hash() }
         task.getContainer() >> 'my/image'
+        task.getConfig() >> new TaskConfig()
 
         def handler = new GooglePipelinesTaskHandler(
                 pipelineConfiguration: config,
@@ -213,6 +217,7 @@ class GooglePipelinesTaskHandlerTest extends GoogleSpecification {
         config.getPreemptible() >> true
         // check staging script
         req.stagingScript == 'mkdir -p /work/dir; (alpha; beta; delta) 2>&1 > /work/dir/.command.log'
+        req.diskSizeGb == null 
         // check main script
         req.mainScript == 'cd /work/dir; bash .command.run 2>&1 | tee -a .command.log'
         // check unstaging script