Skip to content

Commit 22214c5

Browse files
pditommasopditommaso
authored
Align Seqera executor maxSpotAttempts and Fusion handling (#7412) [ci fast]
* Align Seqera executor maxSpotAttempts and Fusion handling [ci fast] Make the Seqera executor follow the same pattern used by AWS Batch and Google Batch: - The spot attempts policy now lives in a `maxSpotAttempts` method in `SeqeraTaskHandler`, matching `AwsBatchTaskHandler` and `GoogleBatchTaskHandler`: the config value wins, otherwise fall back to `FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS` when Fusion snapshots are enabled. `SchemaMapperUtil` just maps the resolved value. - The Fusion snapshot flag is guarded by `fusionEnabled()` at the call site, as done by the other executors. - `SeqeraExecutor.isFusionEnabled` now simply delegates to `FusionHelper.isFusionEnabled` instead of throwing from a boolean getter; the "requires Fusion" check is done in `register`. Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> Assisted-by: Claude Code (Opus 5) Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> * Drop redundant fusionEnabled() guards in SeqeraTaskHandler [ci fast] Fusion is mandatory for the Seqera executor -- `register` aborts when it is not enabled -- so `fusionEnabled()` is always true by the time a task is submitted. Test the snapshot flag directly instead, both at the `toMachineRequirement` call site and in `maxSpotAttempts`. Also move the misplaced `createHandlerWithError` javadoc back onto its method. Assisted-by: Claude Code (Opus 5) Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> * Reject negative maxSpotAttempts in SeqeraTaskHandler [ci fast] A negative `machineRequirement.maxSpotAttempts` was silently dropped: `maxSpotAttempts` returned it as-is and `SchemaMapperUtil` then mapped any non-positive value to null. Fail early instead, using the same `IllegalArgumentException` style as the neighbouring machine requirement validation in `SchemaMapperUtil`. Assisted-by: Claude Code (Opus 5) Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com> --------- Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>
1 parent 1dc8cf6 commit 22214c5

5 files changed

Lines changed: 88 additions & 24 deletions

File tree

‎plugins/nf-seqera/src/main/io/seqera/executor/SeqeraExecutor.groovy‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,8 @@ class SeqeraExecutor extends Executor implements ExtensionPoint {
7373

7474
@Override
7575
protected void register() {
76+
if( !isFusionEnabled() )
77+
throw new AbortOperationException("Seqera executor requires the use of Fusion file system")
7678
applyFusionDefaults()
7779
createClient()
7880
}
@@ -199,10 +201,7 @@ class SeqeraExecutor extends Executor implements ExtensionPoint {
199201

200202
@Override
201203
boolean isFusionEnabled() {
202-
final enabled = FusionHelper.isFusionEnabled(session)
203-
if (!enabled)
204-
throw new AbortOperationException("Seqera executor requires the use of Fusion file system")
205-
return true
204+
return FusionHelper.isFusionEnabled(session)
206205
}
207206

208207
/**

‎plugins/nf-seqera/src/main/io/seqera/executor/SeqeraTaskHandler.groovy‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import java.nio.file.Path
2121
import groovy.transform.CompileStatic
2222
import groovy.transform.PackageScope
2323
import groovy.util.logging.Slf4j
24+
import io.seqera.config.MachineRequirementOpts
2425
import io.seqera.executor.Labels
2526
import io.seqera.sched.api.schema.v1a1.AcceleratorType
2627
import io.seqera.sched.api.schema.v1a1.GetTaskLogsResponse
@@ -40,6 +41,7 @@ import nextflow.exception.ProcessUnrecoverableException
4041
import nextflow.util.Duration
4142
import nextflow.util.MemoryUnit
4243
import nextflow.fusion.FusionAwareTask
44+
import nextflow.fusion.FusionConfig
4345
import nextflow.processor.TaskHandler
4446
import nextflow.processor.TaskRun
4547
import nextflow.processor.TaskStatus
@@ -125,7 +127,8 @@ class SeqeraTaskHandler extends TaskHandler implements FusionAwareTask {
125127
baseMachineOpts,
126128
task.getContainerPlatform(),
127129
task.config.getDisk(),
128-
fusionConfig().snapshotsEnabled()
130+
fusionConfig().snapshotsEnabled(),
131+
maxSpotAttempts(baseMachineOpts)
129132
)
130133
// resolve optional per-task prediction model override from the seqera/predictionModel hint;
131134
// when unset the task inherits the run-level model
@@ -165,6 +168,17 @@ class SeqeraTaskHandler extends TaskHandler implements FusionAwareTask {
165168
executor.getBatchSubmitter().submit(this, schedTask)
166169
}
167170

171+
protected int maxSpotAttempts(MachineRequirementOpts opts) {
172+
final result = opts?.maxSpotAttempts
173+
if( result != null && result < 0 )
174+
throw new IllegalArgumentException("Invalid maxSpotAttempts value: ${result} -- the value must be zero or a positive number")
175+
if( result )
176+
return result
177+
// when fusion snapshot is enabled max attempt should be > 0
178+
// to enable to allow snapshot retry the job execution in a new compute instance
179+
return fusionConfig().snapshotsEnabled() ? FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS : 0
180+
}
181+
168182
/**
169183
* Build the map of container environment variable name to the pipeline secret store
170184
* reference for each {@code secret} process directive.

‎plugins/nf-seqera/src/main/io/seqera/util/SchemaMapperUtil.groovy‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ import io.seqera.sched.api.schema.v1a1.PriceModel as SchedPriceModel
2727
import io.seqera.sched.api.schema.v1a1.ProvisioningModel
2828
import io.seqera.sched.api.schema.v1a1.SchedulingRequirement
2929
import nextflow.cloud.types.PriceModel
30-
import nextflow.fusion.FusionConfig
3130
import nextflow.util.MemoryUnit
3231

3332
/**
@@ -96,7 +95,7 @@ class SchemaMapperUtil {
9695
* @return the MachineRequirement API object, or null if no settings
9796
*/
9897
static MachineRequirement toMachineRequirement(MachineRequirementOpts opts, String taskArch) {
99-
return toMachineRequirement(opts, taskArch, null, false)
98+
return toMachineRequirement(opts, taskArch, null, false, opts?.maxSpotAttempts ?: 0)
10099
}
101100

102101
/**
@@ -106,13 +105,13 @@ class SchemaMapperUtil {
106105
* @param taskArch the task container platform/arch (can be null)
107106
* @param diskSize the disk size from task config (can be null)
108107
* @param snapshotEnabled whether Fusion snapshots are enabled
108+
* @param spotAttempts the number of spot retry attempts, {@code 0} when not applicable
109109
* @return the MachineRequirement API object, or null if no settings
110110
*/
111-
static MachineRequirement toMachineRequirement(MachineRequirementOpts opts, String taskArch, MemoryUnit diskSize, boolean snapshotEnabled) {
111+
static MachineRequirement toMachineRequirement(MachineRequirementOpts opts, String taskArch, MemoryUnit diskSize, boolean snapshotEnabled, int spotAttempts) {
112112
final arch = taskArch
113113
final provisioning = opts?.provisioning
114-
final maxSpotAttempts = opts?.maxSpotAttempts
115-
?: (snapshotEnabled ? FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS : null)
114+
final maxSpotAttempts = spotAttempts > 0 ? spotAttempts : null
116115
final machineTypes = opts?.machineTypes
117116
// task disk overrides config disk
118117
final effectiveDiskSize = diskSize ?: opts?.diskSize

‎plugins/nf-seqera/src/test/io/seqera/executor/SeqeraTaskHandlerTest.groovy‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,10 @@ import nextflow.processor.TaskId
4343
import nextflow.processor.TaskProcessor
4444
import nextflow.processor.TaskRun
4545
import nextflow.processor.TaskStatus
46+
import nextflow.Global
47+
import nextflow.fusion.FusionConfig
4648
import spock.lang.Specification
49+
import spock.lang.Unroll
4750

4851
import java.nio.file.Paths
4952

@@ -1300,6 +1303,52 @@ class SeqeraTaskHandlerTest extends Specification {
13001303
return new SeqeraTaskHandler(taskRun, executor)
13011304
}
13021305

1306+
@Unroll
1307+
def 'should validate max spot attempts' () {
1308+
given:
1309+
Global.config = [fusion: [enabled: true, snapshots: SNAPSHOTS]]
1310+
def executor = Mock(SeqeraExecutor) { getClient() >> Mock(SchedClient); isFusionEnabled() >> true }
1311+
def proc = Mock(TaskProcessor) { getExecutor() >> executor }
1312+
def task = Mock(TaskRun) { getWorkDir() >> Paths.get('/work/ab/cd1234'); getProcessor() >> proc }
1313+
def handler = new SeqeraTaskHandler(task, executor)
1314+
1315+
expect:
1316+
handler.maxSpotAttempts(ATTEMPTS != null ? new MachineRequirementOpts([maxSpotAttempts: ATTEMPTS]) : null) == EXPECTED
1317+
1318+
cleanup:
1319+
Global.config = null
1320+
1321+
where:
1322+
ATTEMPTS | SNAPSHOTS | EXPECTED
1323+
null | false | 0
1324+
0 | false | 0
1325+
2 | false | 2
1326+
and:
1327+
null | true | FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS
1328+
0 | true | FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS
1329+
2 | true | 2
1330+
}
1331+
1332+
def 'should reject negative max spot attempts' () {
1333+
given:
1334+
Global.config = [fusion: [enabled: true, snapshots: true]]
1335+
def executor = Mock(SeqeraExecutor) { getClient() >> Mock(SchedClient); isFusionEnabled() >> true }
1336+
def proc = Mock(TaskProcessor) { getExecutor() >> executor }
1337+
def task = Mock(TaskRun) { getWorkDir() >> Paths.get('/work/ab/cd1234'); getProcessor() >> proc }
1338+
def handler = new SeqeraTaskHandler(task, executor)
1339+
1340+
when:
1341+
handler.maxSpotAttempts(new MachineRequirementOpts([maxSpotAttempts: -1]))
1342+
1343+
then:
1344+
def e = thrown(IllegalArgumentException)
1345+
e.message.contains('maxSpotAttempts')
1346+
e.message.contains('-1')
1347+
1348+
cleanup:
1349+
Global.config = null
1350+
}
1351+
13031352
/**
13041353
* Creates a test handler with an error set on the task
13051354
*/

‎plugins/nf-seqera/src/test/io/seqera/util/MapperUtilTest.groovy‎

Lines changed: 17 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -191,7 +191,8 @@ class MapperUtilTest extends Specification {
191191
new MachineRequirementOpts([:]),
192192
'x86_64',
193193
MemoryUnit.of('200 GB'),
194-
false
194+
false,
195+
0
195196
)
196197

197198
then:
@@ -202,7 +203,7 @@ class MapperUtilTest extends Specification {
202203

203204
def 'should return machine requirement with only disk' () {
204205
when:
205-
def result = SchemaMapperUtil.toMachineRequirement(null, null, MemoryUnit.of('100 GB'), false)
206+
def result = SchemaMapperUtil.toMachineRequirement(null, null, MemoryUnit.of('100 GB'), false, 0)
206207

207208
then:
208209
result != null
@@ -213,7 +214,7 @@ class MapperUtilTest extends Specification {
213214

214215
def 'should return null when no taskArch, no opts, and no disk' () {
215216
expect:
216-
SchemaMapperUtil.toMachineRequirement(null, null, null, false) == null
217+
SchemaMapperUtil.toMachineRequirement(null, null, null, false, 0) == null
217218
}
218219

219220
// tests for custom disk configuration options
@@ -303,7 +304,7 @@ class MapperUtilTest extends Specification {
303304
])
304305

305306
when:
306-
def result = SchemaMapperUtil.toMachineRequirement(opts, 'arm64', MemoryUnit.of('500 GB'), false)
307+
def result = SchemaMapperUtil.toMachineRequirement(opts, 'arm64', MemoryUnit.of('500 GB'), false, 0)
307308

308309
then:
309310
result.arch == 'arm64'
@@ -375,7 +376,7 @@ class MapperUtilTest extends Specification {
375376
])
376377

377378
when:
378-
def result = SchemaMapperUtil.toMachineRequirement(opts, 'x86_64', MemoryUnit.of('200 GB'), false)
379+
def result = SchemaMapperUtil.toMachineRequirement(opts, 'x86_64', MemoryUnit.of('200 GB'), false, 0)
379380

380381
then:
381382
result.arch == 'x86_64'
@@ -465,30 +466,30 @@ class MapperUtilTest extends Specification {
465466
e.message.contains('diskEncrypted')
466467
}
467468

468-
// tests for snapshot maxSpotAttempts defaulting
469+
// tests for snapshot and spot attempts mapping
469470

470471
def 'should return machine requirement with only snapshot enabled' () {
471472
when:
472-
def result = SchemaMapperUtil.toMachineRequirement(null, null, null, true)
473+
def result = SchemaMapperUtil.toMachineRequirement(null, null, null, true, FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS)
473474

474475
then:
475476
result != null
476477
result.snapshotEnabled == true
477478
result.maxSpotAttempts == FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS
478479
}
479480

480-
def 'should use explicit maxSpotAttempts when snapshot enabled' () {
481+
def 'should use given spot attempts' () {
481482
when:
482-
def result = SchemaMapperUtil.toMachineRequirement(new MachineRequirementOpts([maxSpotAttempts: 2]), null, null, true)
483+
def result = SchemaMapperUtil.toMachineRequirement(new MachineRequirementOpts([maxSpotAttempts: 2]), null, null, true, 2)
483484

484485
then:
485486
result.snapshotEnabled == true
486487
result.maxSpotAttempts == 2
487488
}
488489

489-
def 'should not default maxSpotAttempts when snapshot disabled' () {
490+
def 'should not set maxSpotAttempts when spot attempts is zero' () {
490491
when:
491-
def result = SchemaMapperUtil.toMachineRequirement(new MachineRequirementOpts([:]), 'x86_64', null, false)
492+
def result = SchemaMapperUtil.toMachineRequirement(new MachineRequirementOpts([:]), 'x86_64', null, false, 0)
492493

493494
then:
494495
result.snapshotEnabled == null
@@ -527,7 +528,8 @@ class MapperUtilTest extends Specification {
527528
new MachineRequirementOpts([capacityMode: 'managed']),
528529
'arm64',
529530
null,
530-
false
531+
false,
532+
0
531533
)
532534

533535
then:
@@ -541,7 +543,8 @@ class MapperUtilTest extends Specification {
541543
new MachineRequirementOpts([provisioning: 'spot']),
542544
'arm64',
543545
MemoryUnit.of('100 GB'),
544-
true
546+
true,
547+
FusionConfig.DEFAULT_SNAPSHOT_MAX_SPOT_ATTEMPTS
545548
)
546549

547550
then:
@@ -589,7 +592,7 @@ class MapperUtilTest extends Specification {
589592
def opts = new MachineRequirementOpts([diskAllocation: 'nvme', capacityMode: 'asg'])
590593
591594
when:
592-
def result = SchemaMapperUtil.toMachineRequirement(opts, 'x86_64', null, false)
595+
def result = SchemaMapperUtil.toMachineRequirement(opts, 'x86_64', null, false, 0)
593596
594597
then:
595598
result != null

0 commit comments

Comments
 (0)