s3 download retry

docs/config.rst

@@ -511,6 +511,8 @@ cliPath                     The path where the AWS command line tool is installe
 jobRole                     The AWS Job Role ARN that needs to be used to execute the Batch Job.
 maxParallelTransfers        Max parallel upload/download transfer operations *per job* (default: ``16``).
 volumes                     One or more container mounts. Mounts can be specified as simple e.g. `/some/path` or canonical format e.g. ``/host/path:/mount/path[:ro|rw]``. Multiple mounts can be specifid separating them with a comma or using a list object.
+maxTransferAttempts         The maximum number of downloads attempts from s3 (default: `1`).
+delayBetweenAttempts        Delay between download attempts from s3 (default `10`).
 =========================== ================
 
 .. _config-cloud:

modules/nextflow/src/main/groovy/nextflow/cloud/aws/batch/AwsBatchFileCopyStrategy.groovy

@@ -93,7 +93,10 @@ class AwsBatchFileCopyStrategy extends SimpleFileCopyStrategy {
     @Override
     String stageInputFile( Path path, String targetName ) {
         // third param should not be escaped, because it's used in the grep match rule
-        "downloads+=(\"nxf_s3_download s3:/${Escape.path(path)} ${Escape.path(targetName)}\")"
+        def stage_cmd = opts.maxTransferAttempts > 1
+                ? "downloads+=(\"nxf_s3_retry nxf_s3_download s3:/${Escape.path(path)} ${Escape.path(targetName)}\")"
+                : "downloads+=(\"nxf_s3_download s3:/${Escape.path(path)} ${Escape.path(targetName)}\")"
+        return stage_cmd
     }
 
     /**

modules/nextflow/src/main/groovy/nextflow/cloud/aws/batch/AwsOptions.groovy

@@ -34,6 +34,10 @@ class AwsOptions {
 
     static final public int MAX_TRANSFER = 16
 
+    static final public int MAX_TRANSFER_ATTEMPTS = 1
+
+    static final public int DEFAULT_DELAY_BETWEEN_ATTEMPTS = 10
+
     String cliPath
 
     String storageClass
@@ -46,6 +50,10 @@ class AwsOptions {
 
     int maxParallelTransfers = MAX_TRANSFER
 
+    int maxTransferAttempts = MAX_TRANSFER_ATTEMPTS
+
+    int delayBetweenAttempts = DEFAULT_DELAY_BETWEEN_ATTEMPTS
+
     /**
      * The job role ARN that should be used
      */
@@ -71,6 +79,8 @@ class AwsOptions {
         storageClass = session.config.navigate('aws.client.uploadStorageClass') as String
         storageEncryption = session.config.navigate('aws.client.storageEncryption') as String
         maxParallelTransfers = session.config.navigate('aws.batch.maxParallelTransfers', MAX_TRANSFER) as int
+        maxTransferAttempts = session.config.navigate('aws.batch.maxTransferAttempts', MAX_TRANSFER_ATTEMPTS) as int
+        delayBetweenAttempts = session.config.navigate('aws.batch.delayBetweenAttempts', DEFAULT_DELAY_BETWEEN_ATTEMPTS) as int
         region = session.config.navigate('aws.region') as String
         volumes = makeVols(session.config.navigate('aws.batch.volumes'))
         jobRole = session.config.navigate('aws.batch.jobRole')
@@ -133,4 +143,4 @@ class AwsOptions {
         throw new IllegalArgumentException("Not a valid `aws.batch.volumes` value: $obj [${obj.getClass().getName()}]")
     }
 
-}
+}

modules/nextflow/src/main/groovy/nextflow/cloud/aws/batch/S3Helper.groovy

@@ -26,6 +26,8 @@ class S3Helper {
         def storage = opts.storageClass ?: 'STANDARD'
         def encryption = opts.storageEncryption ? "--sse $opts.storageEncryption " : ''
         def maxConnect = opts.maxParallelTransfers ?: AwsOptions.MAX_TRANSFER
+        def attempts = opts.maxTransferAttempts ?: AwsOptions.MAX_TRANSFER_ATTEMPTS
+        def delayBetweenAttempts = opts.delayBetweenAttempts ?: AwsOptions.DEFAULT_DELAY_BETWEEN_ATTEMPTS
 
         """
         # aws helper
@@ -43,6 +45,29 @@ class S3Helper {
             unset IFS
         }
 
+        nxf_s3_retry() {
+            local max_attempts=$attempts
+            local timeout=$delayBetweenAttempts
+            local attempt=0
+            local exitCode=0
+            while (( \$attempt < \$max_attempts ))
+            do
+              if "\$@"
+                then
+                  return 0
+              else
+                exitCode=\$?
+              fi
+              if [[ \$exitCode == 0 ]]
+              then
+                break
+              fi
+              sleep \$timeout
+              attempt=\$(( attempt + 1 ))
+              timeout=\$(( timeout * 2 ))
+            done
+        }
+
         nxf_s3_download() {
             local source=\$1
             local target=\$2
@@ -56,6 +81,7 @@ class S3Helper {
         }
 
         nxf_parallel() {
+            IFS=\$'\\n'
             local cmd=("\$@")
             local cpus=\$(nproc 2>/dev/null || < /proc/cpuinfo grep '^process' -c)
             local max=\$(if (( cpus>$maxConnect )); then echo $maxConnect; else echo \$cpus; fi)
@@ -80,6 +106,7 @@ class S3Helper {
             done
             ((\${#pid[@]}>0)) && wait \${pid[@]}
             )
+            unset IFS
         }
         """.stripIndent()
     }

modules/nextflow/src/test/groovy/nextflow/cloud/aws/batch/AwsBatchFileCopyStrategyTest.groovy

@@ -123,6 +123,29 @@ class AwsBatchFileCopyStrategyTest extends Specification {
                         done
                         unset IFS
                     }
+
+                    nxf_s3_retry() {
+                        local max_attempts=1
+                        local timeout=10
+                        local attempt=0
+                        local exitCode=0
+                        while (( \$attempt < \$max_attempts ))
+                        do
+                          if "\$@"
+                            then
+                              return 0
+                          else
+                            exitCode=\$?
+                          fi
+                          if [[ \$exitCode == 0 ]]
+                          then
+                            break
+                          fi
+                          sleep \$timeout
+                          attempt=\$(( attempt + 1 ))
+                          timeout=\$(( timeout * 2 ))
+                        done
+                    }
 
                     nxf_s3_download() {
                         local source=$1
@@ -137,6 +160,7 @@ class AwsBatchFileCopyStrategyTest extends Specification {
                     }
 
                     nxf_parallel() {
+                        IFS=$'\\n\'
                         local cmd=("$@")
                         local cpus=$(nproc 2>/dev/null || < /proc/cpuinfo grep '^process' -c)
                         local max=$(if (( cpus>16 )); then echo 16; else echo $cpus; fi)
@@ -161,6 +185,7 @@ class AwsBatchFileCopyStrategyTest extends Specification {
                         done
                         ((${#pid[@]}>0)) && wait ${pid[@]}
                         )
+                        unset IFS
                     }
                     '''.stripIndent()
 
@@ -186,6 +211,29 @@ class AwsBatchFileCopyStrategyTest extends Specification {
                     done
                     unset IFS
                 }
+
+                nxf_s3_retry() {
+                    local max_attempts=1
+                    local timeout=10
+                    local attempt=0
+                    local exitCode=0
+                    while (( \$attempt < \$max_attempts ))
+                    do
+                      if "\$@"
+                        then
+                          return 0
+                      else
+                        exitCode=\$?
+                      fi
+                      if [[ \$exitCode == 0 ]]
+                      then
+                        break
+                      fi
+                      sleep \$timeout
+                      attempt=\$(( attempt + 1 ))
+                      timeout=\$(( timeout * 2 ))
+                    done
+                }
 
                 nxf_s3_download() {
                     local source=$1
@@ -200,6 +248,7 @@ class AwsBatchFileCopyStrategyTest extends Specification {
                 }
 
                 nxf_parallel() {
+                    IFS=$'\\n\'
                     local cmd=("$@")
                     local cpus=$(nproc 2>/dev/null || < /proc/cpuinfo grep '^process' -c)
                     local max=$(if (( cpus>16 )); then echo 16; else echo $cpus; fi)
@@ -224,6 +273,7 @@ class AwsBatchFileCopyStrategyTest extends Specification {
                     done
                     ((${#pid[@]}>0)) && wait ${pid[@]}
                     )
+                    unset IFS
                 }
             '''.stripIndent()
     }