Add retry policy to Wave http client

Signed-off-by: Paolo Di Tommaso <[email protected]>

Author: pditommaso

docs/wave.md

@@ -201,3 +201,15 @@ The following configuration options are available:
 
 `wave.report.file` (preview)
 : The name of the containers report file (default: `containers-<timestamp>.config` requires version `23.06.0-edge` or later).
+
+`wave.retry.delay` 
+: The initial delay when a failing HTTP request is retried (default: `150ms`). 
+
+`wave.retry.maxDelay`
+: The max delay when a failing HTTP request is retried (default: `90 seconds`).
+
+`wave.retry.maxAttempts`
+: The max number of attempts a failing HTTP request is retried (default: `5`).
+
+`wave.retry.jitter`
+: Sets the jitterFactor to randomly vary retry delays by (default: `0.25`).

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

@@ -25,8 +25,10 @@ import java.nio.file.Path
 import java.time.Duration
 import java.time.Instant
 import java.time.OffsetDateTime
+import java.time.temporal.ChronoUnit
 import java.util.concurrent.Callable
 import java.util.concurrent.TimeUnit
+import java.util.function.Predicate
 import java.util.regex.Pattern
 
 import com.google.common.cache.Cache
@@ -35,6 +37,11 @@ import com.google.common.util.concurrent.UncheckedExecutionException
 import com.google.gson.Gson
 import com.google.gson.GsonBuilder
 import com.google.gson.reflect.TypeToken
+import dev.failsafe.Failsafe
+import dev.failsafe.RetryPolicy
+import dev.failsafe.event.EventListener
+import dev.failsafe.event.ExecutionAttemptedEvent
+import dev.failsafe.function.CheckedSupplier
 import groovy.json.JsonOutput
 import groovy.transform.CompileStatic
 import groovy.transform.Memoized
@@ -214,7 +221,7 @@ class WaveClient {
                 .build()
 
         try {
-            final resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString())
+            final resp = httpSend(req)
             log.debug "Wave response: statusCode=${resp.statusCode()}; body=${resp.body()}"
             if( resp.statusCode()==200 )
                 return jsonToSubmitResponse(resp.body())
@@ -232,7 +239,7 @@ class WaveClient {
             else
                 throw new BadResponseException("Wave invalid response: [${resp.statusCode()}] ${resp.body()}")
         }
-        catch (ConnectException e) {
+        catch (IOException e) {
             throw new IllegalStateException("Unable to connect Wave service: $endpoint")
         }
     }
@@ -279,7 +286,7 @@ class WaveClient {
                 .GET()
                 .build()
 
-        final resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString())
+        final resp = httpSend(req)
         final code = resp.statusCode()
         if( code>=200 && code<400 ) {
             log.debug "Wave container config response: [$code] ${resp.body()}"
@@ -580,7 +587,7 @@ class WaveClient {
                 .POST(HttpRequest.BodyPublishers.ofString("grant_type=refresh_token&refresh_token=${URLEncoder.encode(refresh, 'UTF-8')}"))
                 .build()
 
-        final resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString())
+        final resp = httpSend(req)
         log.debug "Refresh cookie response: [${resp.statusCode()}] ${resp.body()}"
         if( resp.statusCode() != 200 )
             return false
@@ -674,4 +681,31 @@ class WaveClient {
         final type = new TypeToken<DescribeContainerResponse>(){}.getType()
         return gson.fromJson(json, type)
     }
+
+    protected <T> RetryPolicy<T> retryPolicy(Predicate<? extends Throwable> cond) {
+        final cfg = config.retryOpts()
+        final listener = new EventListener<ExecutionAttemptedEvent<T>>() {
+            @Override
+            void accept(ExecutionAttemptedEvent<T> event) throws Throwable {
+                log.debug("Azure TooManyRequests reponse error - attempt: ${event.attemptCount}", event.lastFailure)
+            }
+        }
+        return RetryPolicy.<T>builder()
+                .handleIf(cond)
+                .withBackoff(cfg.delay.toMillis(), cfg.maxDelay.toMillis(), ChronoUnit.MILLIS)
+                .withMaxAttempts(cfg.maxAttempts)
+                .withJitter(cfg.jitter)
+                .onRetry(listener)
+                .build()
+    }
+
+    protected <T> T safeApply(CheckedSupplier<T> action) {
+        final cond = (e -> e instanceof IOException)  as Predicate<? extends Throwable>
+        final policy = retryPolicy(cond)
+        return Failsafe.with(policy).get(action)
+    }
+
+    protected HttpResponse<String> httpSend(HttpRequest req)  {
+        return safeApply(() -> httpClient.send(req, HttpResponse.BodyHandlers.ofString()))
+    }
 }

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2020-2022, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.seqera.wave.plugin.config
+
+import nextflow.util.Duration
+
+/**
+ * Model retry options for Wave http requests
+ */
+class RetryOpts {
+    Duration delay = Duration.of('150ms')
+    Duration maxDelay = Duration.of('90s')
+    int maxAttempts = 5
+    double jitter = 0.25
+
+    RetryOpts() {
+        this(Collections.emptyMap())
+    }
+
+    RetryOpts(Map config) {
+        if( config.delay )
+            delay = config.delay as Duration
+        if( config.maxDelay )
+            maxDelay = config.maxDelay as Duration
+        if( config.maxAttempts )
+            maxAttempts = config.maxAttempts as int
+        if( config.jitter )
+            jitter = config.jitter as double
+    }
+}

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

@@ -41,6 +41,7 @@ class WaveConfig {
     final private String buildRepository
     final private String cacheRepository
     final private ReportOpts reportOpts
+    final private RetryOpts retryOpts
 
     WaveConfig(Map opts, Map<String,String> env=System.getenv()) {
         this.enabled = opts.enabled
@@ -54,6 +55,7 @@ class WaveConfig {
         this.strategy = parseStrategy(opts.strategy)
         this.bundleProjectResources = opts.bundleProjectResources
         this.reportOpts = new ReportOpts(opts.report as Map ?: Map.of())
+        this.retryOpts = new RetryOpts(opts.retry as Map ?: Map.of())
         if( !endpoint.startsWith('http://') && !endpoint.startsWith('https://') )
             throw new IllegalArgumentException("Endpoint URL should start with 'http:' or 'https:' protocol prefix - offending value: $endpoint")
     }
@@ -66,6 +68,8 @@ class WaveConfig {
 
     SpackOpts spackOpts() { this.spackOpts }
 
+    RetryOpts retryOpts() { this.retryOpts }
+
     List<String> strategy() { this.strategy }
 
     boolean bundleProjectResources() { bundleProjectResources }

plugins/nf-wave/src/test/io/seqera/wave/plugin/config/RetryOptsTest.groovy

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2020-2022, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.seqera.wave.plugin.config
+
+import nextflow.util.Duration
+import spock.lang.Specification
+
+/**
+ *
+ * @author Paolo Di Tommaso <[email protected]>
+ */
+class RetryOptsTest extends Specification {
+
+    def 'should create retry config' () {
+
+        expect:
+        new RetryOpts().delay == Duration.of('150ms')
+        new RetryOpts().maxDelay == Duration.of('90s')
+        new RetryOpts().maxAttempts == 5
+        new RetryOpts().jitter == 0.25d
+
+        and:
+        new RetryOpts([maxAttempts: 20]).maxAttempts == 20
+        new RetryOpts([delay: '1s']).delay == Duration.of('1s')
+        new RetryOpts([maxDelay: '1m']).maxDelay == Duration.of('1m')
+        new RetryOpts([jitter: '0.5']).jitter == 0.5d
+
+    }
+
+}