Allow 1 restart per task (#82)

* Allow 1 restart per task

This should fix a concurrency issue with the CSI driver
ceph/ceph-csi#3511
hashicorp/nomad#15197

* expose the reschedule and restart config vars

* remove unused import

---------

Co-authored-by: Jorge <[email protected]>
Co-authored-by: Abhinav Sharma <[email protected]>

Author: 3 people

plugins/nf-nomad/src/main/nextflow/nomad/config/NomadJobOpts.groovy

@@ -44,9 +44,11 @@ class NomadJobOpts{
     JobVolume[] volumeSpec
     JobAffinity affinitySpec
     JobConstraint constraintSpec
-
     JobConstraints constraintsSpec
 
+    Integer rescheduleAttempts
+    Integer restartAttempts
+
     NomadSecretOpts secretOpts
 
     NomadJobOpts(Map nomadJobOpts, Map<String,String> env=null){
@@ -69,6 +71,10 @@ class NomadJobOpts{
         region = nomadJobOpts.region ?: sysEnv.get('NOMAD_REGION')
         namespace = nomadJobOpts.namespace ?: sysEnv.get('NOMAD_NAMESPACE')
 
+        //NOTE: Default to a single attempt per nomad job definition
+        rescheduleAttempts = nomadJobOpts.rescheduleAttempts as Integer ?: 1
+        restartAttempts = nomadJobOpts.restartAttempts as Integer ?: 1
+
         dockerVolume = nomadJobOpts.dockerVolume ?: null
         if( dockerVolume ){
             log.info "dockerVolume config will be deprecated, use volume type:'docker' name:'name' instead"

plugins/nf-nomad/src/main/nextflow/nomad/executor/NomadService.groovy

@@ -125,12 +125,8 @@ class NomadService implements Closeable{
     }
 
     TaskGroup createTaskGroup(TaskRun taskRun, List<String> args, Map<String, String>env){
-        //NOTE: Force a single-allocation with no-retries per nomad job definition
-        final TASK_RESCHEDULE_ATTEMPTS = 0
-        final TASK_RESTART_ATTEMPTS = 0
-
-        final ReschedulePolicy taskReschedulePolicy  = new ReschedulePolicy().attempts(TASK_RESCHEDULE_ATTEMPTS)
-        final RestartPolicy taskRestartPolicy  = new RestartPolicy().attempts(TASK_RESTART_ATTEMPTS)
+        final ReschedulePolicy taskReschedulePolicy  = new ReschedulePolicy().attempts(this.config.jobOpts().rescheduleAttempts)
+        final RestartPolicy taskRestartPolicy  = new RestartPolicy().attempts(this.config.jobOpts().restartAttempts)
 
         def task = createTask(taskRun, args, env)
         def taskGroup = new TaskGroup(