Add DiskAllocation support to Seqera executor [ci fast]

pditommaso · claude · pditommaso · commit de4bab1ba19e · 2026-01-30T12:35:24.000+01:00
- Add diskAllocation config option (task/node) to MachineRequirementOpts
- Update MapperUtil to map allocation and use correct volumeType API
- Add validation: node allocation only supports disk size
- Bump sched-client to 0.16.0-SNAPSHOT

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;
Signed-off-by: Paolo Di Tommaso &lt;paolo.ditommaso@gmail.com&gt;
diff --git a/plugins/nf-seqera/VERSION b/plugins/nf-seqera/VERSION
@@ -1 +1 @@
-0.6.0
+0.7.0
diff --git a/plugins/nf-seqera/build.gradle b/plugins/nf-seqera/build.gradle
@@ -46,7 +46,7 @@ dependencies {
     compileOnly project(':nextflow')
     compileOnly 'org.slf4j:slf4j-api:2.0.17'
     compileOnly 'org.pf4j:pf4j:3.12.0'
-    api 'io.seqera:sched-client:0.14.0-SNAPSHOT'
+    api 'io.seqera:sched-client:0.16.0-SNAPSHOT'
 
     testImplementation(testFixtures(project(":nextflow")))
     testImplementation "org.apache.groovy:groovy:4.0.29"
diff --git a/plugins/nf-seqera/src/main/io/seqera/config/MachineRequirementOpts.groovy b/plugins/nf-seqera/src/main/io/seqera/config/MachineRequirementOpts.groovy
@@ -82,6 +82,14 @@ class MachineRequirementOpts implements ConfigScope {
     """)
     final Boolean diskEncrypted
 
+    @ConfigOption
+    @Description("""
+        The disk allocation strategy: `task` or `node`.
+        - `task`: Per-task EBS volume created at task launch (default)
+        - `node`: Per-node instance storage attached at cluster level
+    """)
+    final String diskAllocation
+
     /* required by config scope -- do not remove */
     MachineRequirementOpts() {}
 
@@ -94,6 +102,7 @@ class MachineRequirementOpts implements ConfigScope {
         this.diskThroughputMiBps = opts.diskThroughputMiBps as Integer
         this.diskIops = opts.diskIops as Integer
         this.diskEncrypted = opts.diskEncrypted as Boolean
+        this.diskAllocation = opts.diskAllocation as String
     }
 
     String getArch() {
@@ -127,4 +136,8 @@ class MachineRequirementOpts implements ConfigScope {
     Boolean getDiskEncrypted() {
         return diskEncrypted
     }
+
+    String getDiskAllocation() {
+        return diskAllocation
+    }
 }
diff --git a/plugins/nf-seqera/src/main/io/seqera/util/MapperUtil.groovy b/plugins/nf-seqera/src/main/io/seqera/util/MapperUtil.groovy
@@ -19,6 +19,7 @@ package io.seqera.util
 
 import groovy.transform.CompileStatic
 import io.seqera.config.MachineRequirementOpts
+import io.seqera.sched.api.schema.v1a1.DiskAllocation
 import io.seqera.sched.api.schema.v1a1.DiskRequirement
 import io.seqera.sched.api.schema.v1a1.MachineRequirement
 import io.seqera.sched.api.schema.v1a1.PriceModel as SchedPriceModel
@@ -111,14 +112,29 @@ class MapperUtil {
      * Uses config options if provided, otherwise defaults to Fusion recommended settings:
      * EBS gp3 volume with 325 MiB/s throughput.
      *
+     * For 'node' allocation, only sizeGiB and mountPath are applicable.
+     * For 'task' allocation (default), all EBS options can be specified.
+     *
      * @param diskSize the disk size (can be null)
      * @param opts the machine requirement options with disk settings (can be null)
      * @return the DiskRequirement API object, or null if diskSize is null or zero
      */
     static DiskRequirement toDiskRequirement(MemoryUnit diskSize, MachineRequirementOpts opts=null) {
         if (!diskSize || diskSize.toGiga() <= 0)
             return null
-        // Use config values or Fusion recommended defaults
+
+        final allocation = toDiskAllocation(opts?.diskAllocation)
+
+        // For 'node' allocation, only size and mountPath are valid
+        if (allocation == DiskAllocation.NODE) {
+            validateNodeAllocationOpts(opts)
+            final DiskRequirement req = new DiskRequirement()
+            req.sizeGiB(diskSize.toGiga() as Integer)
+            req.allocation(allocation)
+            return req
+        }
+
+        // For 'task' allocation (default), apply EBS-specific options
         final type = opts?.diskType ?: DEFAULT_DISK_TYPE
         // Validate disk type is supported
         if (!SUPPORTED_DISK_TYPES.contains(type)) {
@@ -128,10 +144,11 @@ class MapperUtil {
         final iops = opts?.diskIops
         final encrypted = opts?.diskEncrypted ?: false
 
-        def req = new DiskRequirement()
-            .sizeGiB(diskSize.toGiga() as Integer)
-            .type(type)
-            .encrypted(encrypted)
+        final DiskRequirement req = new DiskRequirement()
+        req.sizeGiB(diskSize.toGiga() as Integer)
+        req.volumeType(type)
+        req.encrypted(encrypted)
+        req.allocation(allocation)
         // Only set throughput for gp3 volumes
         if (type == DEFAULT_DISK_TYPE) {
             req.throughputMiBps(throughput)
@@ -143,6 +160,44 @@ class MapperUtil {
         return req
     }
 
+    /**
+     * Validates that no EBS-specific options are set when using 'node' allocation.
+     * Node allocation uses instance storage, not EBS volumes.
+     *
+     * @param opts the machine requirement options
+     * @throws IllegalArgumentException if EBS-specific options are set with node allocation
+     */
+    private static void validateNodeAllocationOpts(MachineRequirementOpts opts) {
+        if (!opts)
+            return
+        final List<String> invalidOpts = []
+        if (opts.diskType)
+            invalidOpts.add('diskType')
+        if (opts.diskThroughputMiBps)
+            invalidOpts.add('diskThroughputMiBps')
+        if (opts.diskIops)
+            invalidOpts.add('diskIops')
+        if (opts.diskEncrypted)
+            invalidOpts.add('diskEncrypted')
+
+        if (invalidOpts) {
+            throw new IllegalArgumentException(
+                "The following options are not valid with 'node' disk allocation: ${invalidOpts.join(', ')}. " +
+                "Node allocation uses instance storage; only disk size is applicable."
+            )
+        }
+    }
+
+    /**
+     * Maps a disk allocation string to DiskAllocation enum.
+     *
+     * @param value the disk allocation string (task, node)
+     * @return the DiskAllocation enum value, or null if value is null
+     */
+    static DiskAllocation toDiskAllocation(String value) {
+        value ? DiskAllocation.fromValue(value) : null
+    }
+
     /**
      * Maps a provisioning string to ProvisioningModel enum.
      *
diff --git a/plugins/nf-seqera/src/test/io/seqera/util/MapperUtilTest.groovy b/plugins/nf-seqera/src/test/io/seqera/util/MapperUtilTest.groovy
@@ -18,6 +18,7 @@
 package io.seqera.util
 
 import io.seqera.config.MachineRequirementOpts
+import io.seqera.sched.api.schema.v1a1.DiskAllocation
 import io.seqera.sched.api.schema.v1a1.PriceModel as SchedPriceModel
 import io.seqera.sched.api.schema.v1a1.ProvisioningModel
 import nextflow.cloud.types.PriceModel
@@ -157,18 +158,23 @@ class MapperUtilTest extends Specification {
         when:
         def result = MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'))
 
-        then:
+        then: 'disk size is set'
         result.sizeGiB == 100
-        result.type == MapperUtil.DEFAULT_DISK_TYPE
+        and: 'allocation defaults to null (task allocation on API side)'
+        result.allocation == null
+        and: 'EBS defaults are applied for task allocation'
+        result.volumeType == MapperUtil.DEFAULT_DISK_TYPE
         result.throughputMiBps == MapperUtil.DEFAULT_DISK_THROUGHPUT_MIBPS
+        result.encrypted == false
+        result.iops == null
     }
 
     def 'should map disk size in different units' () {
         expect:
         MapperUtil.toDiskRequirement(MemoryUnit.of('1 TB')).sizeGiB == 1024
         MapperUtil.toDiskRequirement(MemoryUnit.of('50 GB')).sizeGiB == 50
         and: 'defaults are applied'
-        MapperUtil.toDiskRequirement(MemoryUnit.of('1 TB')).type == MapperUtil.DEFAULT_DISK_TYPE
+        MapperUtil.toDiskRequirement(MemoryUnit.of('1 TB')).volumeType == MapperUtil.DEFAULT_DISK_TYPE
         MapperUtil.toDiskRequirement(MemoryUnit.of('1 TB')).throughputMiBps == MapperUtil.DEFAULT_DISK_THROUGHPUT_MIBPS
     }
 
@@ -226,7 +232,7 @@ class MapperUtilTest extends Specification {
 
         then:
         result.sizeGiB == 100
-        result.type == 'ebs/io1'
+        result.volumeType == 'ebs/io1'
         result.iops == 10000
         result.throughputMiBps == null  // throughput only for gp3
     }
@@ -239,7 +245,7 @@ class MapperUtilTest extends Specification {
         def result = MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
 
         then:
-        result.type == MapperUtil.DEFAULT_DISK_TYPE
+        result.volumeType == MapperUtil.DEFAULT_DISK_TYPE
         result.throughputMiBps == 500
     }
 
@@ -268,7 +274,7 @@ class MapperUtilTest extends Specification {
 
         then:
         result.sizeGiB == 200
-        result.type == 'ebs/gp3'
+        result.volumeType == 'ebs/gp3'
         result.throughputMiBps == 600
         result.iops == 8000
         result.encrypted == true
@@ -289,10 +295,160 @@ class MapperUtilTest extends Specification {
         then:
         result.arch == 'arm64'
         result.disk.sizeGiB == 500
-        result.disk.type == 'ebs/io2'
+        result.disk.volumeType == 'ebs/io2'
         result.disk.iops == 15000
         result.disk.encrypted == true
         result.disk.throughputMiBps == null  // io2 doesn't use throughput
     }
 
+    // tests for disk allocation mapping
+
+    def 'should map disk allocation' () {
+        expect:
+        MapperUtil.toDiskAllocation(null) == null
+        MapperUtil.toDiskAllocation('task') == DiskAllocation.TASK
+        MapperUtil.toDiskAllocation('node') == DiskAllocation.NODE
+    }
+
+    def 'should throw exception for invalid disk allocation' () {
+        when:
+        MapperUtil.toDiskAllocation('invalid')
+
+        then:
+        thrown(IllegalArgumentException)
+    }
+
+    def 'should use task disk allocation from config' () {
+        given:
+        def opts = new MachineRequirementOpts([diskAllocation: 'task'])
+
+        when:
+        def result = MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        result.allocation == DiskAllocation.TASK
+    }
+
+    def 'should use node disk allocation from config' () {
+        given:
+        def opts = new MachineRequirementOpts([diskAllocation: 'node'])
+
+        when:
+        def result = MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        result.allocation == DiskAllocation.NODE
+        result.sizeGiB == 100
+        result.volumeType == null  // node allocation doesn't set EBS options
+        result.throughputMiBps == null
+        result.iops == null
+        result.encrypted == null
+    }
+
+    def 'should default to null disk allocation when not specified' () {
+        when:
+        def result = MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'))
+
+        then:
+        result.allocation == null
+    }
+
+    def 'should include disk allocation in machine requirement' () {
+        given:
+        def opts = new MachineRequirementOpts([
+            arch: 'x86_64',
+            diskAllocation: 'node'
+        ])
+
+        when:
+        def result = MapperUtil.toMachineRequirement(opts, null, MemoryUnit.of('200 GB'))
+
+        then:
+        result.arch == 'x86_64'
+        result.disk.sizeGiB == 200
+        result.disk.allocation == DiskAllocation.NODE
+    }
+
+    // tests for node allocation validation
+
+    def 'should throw error when diskType is set with node allocation' () {
+        given:
+        def opts = new MachineRequirementOpts([
+            diskAllocation: 'node',
+            diskType: 'ebs/gp3'
+        ])
+
+        when:
+        MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains('diskType')
+        e.message.contains('node')
+    }
+
+    def 'should throw error when diskIops is set with node allocation' () {
+        given:
+        def opts = new MachineRequirementOpts([
+            diskAllocation: 'node',
+            diskIops: 10000
+        ])
+
+        when:
+        MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains('diskIops')
+    }
+
+    def 'should throw error when diskThroughputMiBps is set with node allocation' () {
+        given:
+        def opts = new MachineRequirementOpts([
+            diskAllocation: 'node',
+            diskThroughputMiBps: 500
+        ])
+
+        when:
+        MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains('diskThroughputMiBps')
+    }
+
+    def 'should throw error when diskEncrypted is set with node allocation' () {
+        given:
+        def opts = new MachineRequirementOpts([
+            diskAllocation: 'node',
+            diskEncrypted: true
+        ])
+
+        when:
+        MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains('diskEncrypted')
+    }
+
+    def 'should report all invalid options with node allocation' () {
+        given:
+        def opts = new MachineRequirementOpts([
+            diskAllocation: 'node',
+            diskType: 'ebs/io1',
+            diskIops: 10000,
+            diskEncrypted: true
+        ])
+
+        when:
+        MapperUtil.toDiskRequirement(MemoryUnit.of('100 GB'), opts)
+
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains('diskType')
+        e.message.contains('diskIops')
+        e.message.contains('diskEncrypted')
+    }
+
 }
