Merge branch 'master' into k8sOutOf

VERSION

@@ -1 +1 @@
-22.09.5-edge
+22.09.6-edge

changelog.txt

@@ -1,5 +1,16 @@
 NEXTFLOW CHANGE-LOG
 ===================
+22.09.6-edge - 26 Sep 2022
+- Add SocketTimeoutException to k8s client request retry [527e0d5d]
+- Add MaxErrorRetry to K8s config [58be2128]
+- Add tags propagation to AWS Batch [d64eeffc]
+- Fix task resume when updating fusion layer [f38fd2db]
+- Fix Channel merge still deprecated for DSL2 (#3220) [d27384d2] <Kibubu>
+- Apply GCP resourceLabels to the VirtualMachine (#3234) [2275c03c] <Doug Daniels>
+- Update Google Batch mount point with the requirements [5aec28ac]
+- Improve wave error reporting [73842215]
+- Bump fusion 0.4.x [26f1f896]
+
 22.09.5-edge - 21 Sep 2022
 - Use default wave strategy [abbfa7f4]
 - Handle errors reported by tower report writer [0e814647]

docs/conf.py

@@ -54,7 +54,7 @@
 # The short X.Y version.
 version = '22.09'
 # The full version, including alpha/beta/rc tags.
-release = '22.09.5-edge'
+release = '22.09.6-edge'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

modules/nextflow/src/main/resources/META-INF/plugins-info.txt

@@ -1,7 +1,7 @@
-[email protected].5
-[email protected].3
+[email protected].6
+[email protected].4
 [email protected]
-[email protected].3
+[email protected].4
 [email protected]
 [email protected]
 [email protected]

modules/nf-commons/src/main/nextflow/Const.groovy

@@ -53,17 +53,17 @@ class Const {
     /**
      * The application version
      */
-    static public final String APP_VER = "22.09.5-edge"
+    static public final String APP_VER = "22.09.6-edge"
 
     /**
      * The app build time as linux/unix timestamp
      */
-    static public final long APP_TIMESTAMP = 1663747945997
+    static public final long APP_TIMESTAMP = 1664205183116
 
     /**
      * The app build number
      */
-    static public final int APP_BUILDNUM = 5799
+    static public final int APP_BUILDNUM = 5802
 
     /**
      * The app build time string relative to UTC timezone

nextflow

@@ -16,7 +16,7 @@
 #  limitations under the License.
 
 [[ "$NXF_DEBUG" == 'x' ]] && set -x
-NXF_VER=${NXF_VER:-'22.09.5-edge'}
+NXF_VER=${NXF_VER:-'22.09.6-edge'}
 NXF_ORG=${NXF_ORG:-'nextflow-io'}
 NXF_HOME=${NXF_HOME:-$HOME/.nextflow}
 NXF_PROT=${NXF_PROT:-'https'}

nextflow.md5

@@ -1 +1 @@
-3ba1581357bbc9874fceb1f7a995bc27
+70fd600b06967e1b03c76ede2545eea9

nextflow.sha1

@@ -1 +1 @@
-35aff3b925492727d8ce1b4f410fb30f3aaada17
+1decc861772398b1bb2414ef0d594c6d266d856e

nextflow.sha256

@@ -1 +1 @@
-511fe4e911c96d6d379739f86bde066d02b18acfaa55edb3f7306d78a31fa1ba
+ce1612e5ee1b7713c30c63d98dac44761b915587ff25f82d0c6228cdece11183

plugins/nf-amazon/changelog.txt

@@ -1,5 +1,8 @@
 nf-amazon changelog
 ===================
+1.10.6 - 26 Sep 2022
+- Add tags propagation to AWS Batch [d64eeffc]
+
 1.10.5 - 20 Sep 2022
 - Fix AWS S3 copy object [b3b90d23]
 

plugins/nf-amazon/src/main/nextflow/cloud/aws/batch/AwsBatchTaskHandler.groovy

@@ -368,10 +368,6 @@ class AwsBatchTaskHandler extends TaskHandler implements BatchHandler<String,Job
         resolveJobDefinition(container)
     }
 
-    protected Map<String, String> getResourceLabels(TaskRun task){
-        task.config.getResourceLabels() ?: Collections.<String,String>emptyMap()
-    }
-
     /**
      * Maps a docker container image to a Batch job definition name
      *
@@ -652,11 +648,15 @@ class AwsBatchTaskHandler extends TaskHandler implements BatchHandler<String,Job
         /*
          * create the request object
          */
+        final labels = task.config.getResourceLabels()
         final result = new SubmitJobRequest()
         result.setJobName(normalizeJobName(task.name))
         result.setJobQueue(getJobQueue(task))
         result.setJobDefinition(getJobDefinition(task))
-        result.setTags(getResourceLabels(task))
+        if( labels ) {
+            result.setTags(labels)
+            result.setPropagateTags(true)
+        }
         // set the share identifier
         if( this.getAwsOptions().shareIdentifier )
             result.setShareIdentifier(this.getAwsOptions().shareIdentifier)

plugins/nf-amazon/src/resources/META-INF/MANIFEST.MF

@@ -1,6 +1,6 @@
 Manifest-Version: 1.0
 Plugin-Class: nextflow.cloud.aws.AmazonPlugin
 Plugin-Id: nf-amazon
-Plugin-Version: 1.10.5
+Plugin-Version: 1.10.6
 Plugin-Provider: Seqera Labs
 Plugin-Requires: >=22.09.2-edge

plugins/nf-amazon/src/test/nextflow/cloud/aws/batch/AwsBatchTaskHandlerTest.groovy

@@ -897,6 +897,7 @@ class AwsBatchTaskHandlerTest extends Specification {
                 .withAttempts(5)
                 .withEvaluateOnExit( new EvaluateOnExit().withAction('RETRY').withOnStatusReason('Host EC2*'), new EvaluateOnExit().withOnReason('*').withAction('EXIT') )
         req.getTags() == [a:'b']
+        req.getPropagateTags() == true
     }
     def 'get fusion submit command' () {
         given:

plugins/nf-google/changelog.txt

@@ -1,5 +1,9 @@
 nf-google changelog
 ===================
+1.4.3 - 26 Sep 2022
+- Update Google Batch mount point with the requirements [5aec28ac]
+- Apply GCP resourceLabels to the VirtualMachine (#3234) [2275c03c] <Doug Daniels>
+
 1.4.2 - 1 Sep 2022
 - Add support for resource labels for google batch (#3168) (#3170) [2d24917b] (#2853) [5d0b7c35]
 - Add support for project resources [c2ad6566]

plugins/nf-google/src/main/nextflow/cloud/google/batch/GoogleBatchScriptLauncher.groovy

@@ -40,7 +40,7 @@ import nextflow.util.PathTrie
 @CompileStatic
 class GoogleBatchScriptLauncher extends BashWrapperBuilder {
 
-    private static final String MOUNT_ROOT = '/mnt'
+    private static final String MOUNT_ROOT = '/mnt/disks'
 
     private CloudStoragePath remoteWorkDir
     private Path remoteBinDir

plugins/nf-google/src/main/nextflow/cloud/google/lifesciences/GoogleLifeSciencesHelper.groovy

@@ -197,6 +197,7 @@ class GoogleLifeSciencesHelper {
                 .setDisks([disk])
                 .setServiceAccount(serviceAccount)
                 .setPreemptible(req.preemptible)
+                .setLabels(req.resourceLabels)
 
         def network = new Network()
 

plugins/nf-google/src/resources/META-INF/MANIFEST.MF

@@ -1,6 +1,6 @@
 Manifest-Version: 1.0
 Plugin-Class: nextflow.cloud.google.GoogleCloudPlugin
 Plugin-Id: nf-google
-Plugin-Version: 1.4.3
+Plugin-Version: 1.4.4
 Plugin-Provider: Seqera Labs
 Plugin-Requires: >=22.09.0-edge

plugins/nf-google/src/test/nextflow/cloud/google/batch/GoogleBatchScriptLauncherTest.groovy

@@ -40,10 +40,10 @@ class GoogleBatchScriptLauncherTest extends Specification{
         and:
         launcher.getContainerMounts() == [MOUNTS] 
         where:
-        BUCKET  | PATH          | PARENT    | EXPECTED                          | MOUNTS
-        'foo'   | '/'           | false     | Paths.get('/mnt/foo')             | '/mnt/foo:/mnt/foo:rw'
-        'foo'   | '/some/dir'   | false     | Paths.get('/mnt/foo/some/dir')    | '/mnt/foo/some/dir:/mnt/foo/some/dir:rw'
-        'foo'   | '/some/dir'   | true      | Paths.get('/mnt/foo/some/dir')    | '/mnt/foo/some:/mnt/foo/some:rw'
+        BUCKET  | PATH          | PARENT    | EXPECTED                              | MOUNTS
+        'foo'   | '/'           | false     | Paths.get('/mnt/disks/foo')           | '/mnt/disks/foo:/mnt/disks/foo:rw'
+        'foo'   | '/some/dir'   | false     | Paths.get('/mnt/disks/foo/some/dir')  | '/mnt/disks/foo/some/dir:/mnt/disks/foo/some/dir:rw'
+        'foo'   | '/some/dir'   | true      | Paths.get('/mnt/disks/foo/some/dir')  | '/mnt/disks/foo/some:/mnt/disks/foo/some:rw'
     }
 
     def 'should compute volume mounts' () {
@@ -54,26 +54,26 @@ class GoogleBatchScriptLauncherTest extends Specification{
         def PATH3 = CloudStorageFileSystem.forBucket('omega').getPath('/data/sample3.bam')
 
         expect:
-        launcher.toContainerMount(PATH1) == Paths.get('/mnt/alpha/data/sample1.bam')
-        launcher.toContainerMount(PATH2) == Paths.get('/mnt/alpha/data/sample2.bam')
-        launcher.toContainerMount(PATH3) == Paths.get('/mnt/omega/data/sample3.bam')
+        launcher.toContainerMount(PATH1) == Paths.get('/mnt/disks/alpha/data/sample1.bam')
+        launcher.toContainerMount(PATH2) == Paths.get('/mnt/disks/alpha/data/sample2.bam')
+        launcher.toContainerMount(PATH3) == Paths.get('/mnt/disks/omega/data/sample3.bam')
 
         and:
         def containerMounts = launcher.getContainerMounts()
         and:
         containerMounts.size() == 2
-        containerMounts[0] == '/mnt/alpha/data:/mnt/alpha/data:rw'
-        containerMounts[1] == '/mnt/omega/data/sample3.bam:/mnt/omega/data/sample3.bam:rw'
+        containerMounts[0] == '/mnt/disks/alpha/data:/mnt/disks/alpha/data:rw'
+        containerMounts[1] == '/mnt/disks/omega/data/sample3.bam:/mnt/disks/omega/data/sample3.bam:rw'
 
         and:
         def volumes = launcher.getVolumes()
         and:
         volumes.size() == 2
         volumes[0].getGcs().getRemotePath() == 'alpha'
-        volumes[0].getMountPath() == '/mnt/alpha'
+        volumes[0].getMountPath() == '/mnt/disks/alpha'
         volumes[0].getMountOptionsList() == ['-o rw,allow_other', '-implicit-dirs']
         volumes[1].getGcs().getRemotePath() == 'omega'
-        volumes[1].getMountPath() == '/mnt/omega'
+        volumes[1].getMountPath() == '/mnt/disks/omega'
         volumes[1].getMountOptionsList() == ['-o rw,allow_other', '-implicit-dirs']
     }
 

plugins/nf-google/src/test/nextflow/cloud/google/batch/GoogleBatchTaskHandlerTest.groovy

@@ -71,10 +71,10 @@ class GoogleBatchTaskHandlerTest extends Specification {
         taskGroup.getTaskSpec().getComputeResource().getMemoryMib() == 0
         taskGroup.getTaskSpec().getMaxRunDuration().getSeconds() == 0
         and:
-        runnable.getContainer().getCommandsList().join(' ') == '/bin/bash -o pipefail -c trap "{ cp .command.log /mnt/foo/scratch/.command.log; }" ERR; /bin/bash /mnt/foo/scratch/.command.run 2>&1 | tee .command.log'
+        runnable.getContainer().getCommandsList().join(' ') == '/bin/bash -o pipefail -c trap "{ cp .command.log /mnt/disks/foo/scratch/.command.log; }" ERR; /bin/bash /mnt/disks/foo/scratch/.command.run 2>&1 | tee .command.log'
         runnable.getContainer().getImageUri() == CONTAINER_IMAGE
         runnable.getContainer().getOptions() == ''
-        runnable.getContainer().getVolumesList() == ['/mnt/foo/scratch:/mnt/foo/scratch:rw']
+        runnable.getContainer().getVolumesList() == ['/mnt/disks/foo/scratch:/mnt/disks/foo/scratch:rw']
         and:
         instancePolicy.getAcceleratorsCount() == 0
         instancePolicy.getMachineType() == ''
@@ -146,10 +146,10 @@ class GoogleBatchTaskHandlerTest extends Specification {
         taskGroup.getTaskSpec().getComputeResource().getMemoryMib() == MEM.toMega()
         taskGroup.getTaskSpec().getMaxRunDuration().getSeconds() == TIMEOUT.seconds
         and:
-        runnable.getContainer().getCommandsList().join(' ') == '/bin/bash -o pipefail -c trap "{ cp .command.log /mnt/foo/scratch/.command.log; }" ERR; /bin/bash /mnt/foo/scratch/.command.run 2>&1 | tee .command.log'
+        runnable.getContainer().getCommandsList().join(' ') == '/bin/bash -o pipefail -c trap "{ cp .command.log /mnt/disks/foo/scratch/.command.log; }" ERR; /bin/bash /mnt/disks/foo/scratch/.command.run 2>&1 | tee .command.log'
         runnable.getContainer().getImageUri() == CONTAINER_IMAGE
         runnable.getContainer().getOptions() == CONTAINER_OPTS
-        runnable.getContainer().getVolumesList() == ['/mnt/foo/scratch:/mnt/foo/scratch:rw']
+        runnable.getContainer().getVolumesList() == ['/mnt/disks/foo/scratch:/mnt/disks/foo/scratch:rw']
         and:
         instancePolicy.getAccelerators(0).getCount() == 1
         instancePolicy.getAccelerators(0).getType() == ACCELERATOR.type

plugins/nf-google/src/test/nextflow/cloud/google/lifesciences/GoogleLifeSciencesHelperTest.groovy

@@ -194,7 +194,8 @@ class GoogleLifeSciencesHelperTest extends GoogleSpecification {
                     network: 'net/123',
                     subnetwork: 'sub/192',
                     serviceAccountEmail: '[email protected]',
-                    usePrivateAddress: true ))
+                    usePrivateAddress: true,
+                    resourceLabels: [foo:'bar'] ))
         then:
         with(resources3) {
             getVirtualMachine().getMachineType() == type
@@ -208,6 +209,7 @@ class GoogleLifeSciencesHelperTest extends GoogleSpecification {
             getVirtualMachine().getAccelerators()[0].getType()=='nvidia-tesla-k80'
             getVirtualMachine().getBootDiskSizeGb() == 75
             getVirtualMachine().getCpuPlatform() == 'Intel Skylake'
+            getVirtualMachine().getLabels() == [foo: 'bar']
             getVirtualMachine().getNetwork().getUsePrivateAddress()
             getVirtualMachine().getNetwork().getNetwork() == 'net/123'
             getVirtualMachine().getNetwork().getSubnetwork() == 'sub/192'

plugins/nf-wave/src/main/io/seqera/wave/plugin/ContainerConfig.groovy

@@ -20,6 +20,8 @@ package io.seqera.wave.plugin
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
 import groovy.transform.ToString
+import nextflow.util.CacheHelper
+
 /**
  * Model a container configuration
  *
@@ -91,4 +93,19 @@ class ContainerConfig {
         if( right ) result.addAll(right)
         return result
     }
+
+    String fingerprint() {
+        final allMeta = new ArrayList()
+        allMeta.add( entrypoint ?: 'no-entry' )
+        allMeta.add( cmd ?: 'no-cmd' )
+        allMeta.add( env ?: 'no-env' )
+        allMeta.add( workingDir ?: 'no-workdir')
+        final layers0 = layers ?: Collections.<ContainerLayer>emptyList()
+
+        for( ContainerLayer it : layers0 ) {
+            if( !it.skipHashing )
+                allMeta.add(it.fingerprint())
+        }
+        return CacheHelper.hasher(allMeta).hash().toString()
+    }
 }

plugins/nf-wave/src/main/io/seqera/wave/plugin/ContainerLayer.groovy

@@ -20,6 +20,7 @@ package io.seqera.wave.plugin
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
 import groovy.transform.ToString
+import nextflow.util.CacheHelper
 
 /**
  * Model a container layer meta-info
@@ -30,15 +31,45 @@ import groovy.transform.ToString
 @ToString(includeNames = true, includePackage = false)
 @CompileStatic
 class ContainerLayer {
+    /**
+     * the layer location, it can be either `http:` or `https:` prefixed URI
+     * or a `data:` pseudo-protocol followed by a base64 encoded tar gzipped layer payload
+     */
     String location
+
+    /**
+     * The layer gzip sha256 checksum
+     */
     String gzipDigest
+
+    /**
+     * The layer gzip size in bytes
+     */
     Integer gzipSize
+
+    /**
+     * The layer tar sha256 checksum
+     */
     String tarDigest
 
+    /**
+     * When {@code this layer is not added in the final config fingerprint}
+     */
+    Boolean skipHashing
+
     void validate() {
         if( !location ) throw new IllegalArgumentException("Missing layer location")
         if( !gzipDigest ) throw new IllegalArgumentException("Missing layer gzip digest")
         if( !gzipSize ) throw new IllegalArgumentException("Missing layer gzip size")
         if( !tarDigest ) throw new IllegalArgumentException("Missing layer tar digest")
     }
+
+    String fingerprint() {
+        final allMeta = new ArrayList()
+        allMeta.add( location ?: 'no-location' )
+        allMeta.add( gzipDigest ?: 'no-gzipDigest' )
+        allMeta.add( gzipSize ?: 0 )
+        allMeta.add( tarDigest ?: 'no-tarDigest')
+        return CacheHelper.hasher(allMeta).hash().toString()
+    }
 }

plugins/nf-wave/src/main/io/seqera/wave/plugin/WaveAssets.groovy

@@ -64,7 +64,7 @@ class WaveAssets {
         final allMeta = new ArrayList(10)
         allMeta.add( this.containerImage )
         allMeta.add( this.moduleResources?.fingerprint() )
-        allMeta.add( this.containerConfig?.hashCode() )
+        allMeta.add( this.containerConfig?.fingerprint() )
         allMeta.add( this.dockerFileContent )
         allMeta.add( this.condaFile )
         allMeta.add( this.projectResources?.fingerprint() )

plugins/nf-wave/src/main/io/seqera/wave/plugin/WaveClient.groovy

@@ -159,11 +159,13 @@ class WaveClient {
 
         try {
             final resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString())
-            if( resp.statusCode()==200 ) {
-                log.debug "Wave response: ${resp.body()}"
+            log.debug "Wave response: statusCode=${resp.statusCode()}; body=${resp.body()}"
+            if( resp.statusCode()==200 )
                 return jsonToSubmitResponse(resp.body())
-            }
-            throw new BadResponseException("Wave invalid response: [${resp.statusCode()}] ${resp.body()}")
+            if( resp.statusCode()==401 )
+                throw new BadResponseException("Unauthorised [401] - Verify you have provided a valid access token")
+            else
+                throw new BadResponseException("Wave invalid response: [${resp.statusCode()}] ${resp.body()}")
         }
         catch (ConnectException e) {
             throw new IllegalStateException("Unable to connect Wave service: $endpoint")

plugins/nf-wave/src/main/io/seqera/wave/plugin/config/FusionConfig.groovy

@@ -27,7 +27,7 @@ import groovy.transform.CompileStatic
 @CompileStatic
 class FusionConfig {
 
-    final static public String DEFAULT_FUSION_URL = 'https://fusionfs-release.s3.eu-west-1.amazonaws.com/fusionfs-latest.json'
+    final static public String DEFAULT_FUSION_URL = 'https://s3.eu-west-1.amazonaws.com/fusionfs.io/releases/v0.4.json'
 
     final private enabled
     final private String containerConfigUrl

plugins/nf-wave/src/main/io/seqera/wave/plugin/config/WaveConfig.groovy

@@ -28,7 +28,7 @@ import nextflow.util.Duration
 @Slf4j
 @CompileStatic
 class WaveConfig {
-    final private static String DEF_ENDPOINT = 'http://localhost:9090'
+    final private static String DEF_ENDPOINT = 'https://wave.seqera.io'
     final private Boolean enabled
     final private String endpoint
     final private List<URL> containerConfigUrl
@@ -111,7 +111,7 @@ class WaveConfig {
     }
 
     List<URL> containerConfigUrl() {
-        return containerConfigUrl
+        return containerConfigUrl ?: Collections.<URL>emptyList()
     }
 
     Duration tokensCacheMaxDuration() {