add exponential backoff (#687)

Author: jackshirazi

opamp/src/main/java/co/elastic/opamp/client/CentralConfigurationManagerImpl.java

@@ -122,13 +122,28 @@ public void onConnect(OpampClient client) {
   }
 
   @Override
-  public void onConnectFailed(OpampClient client, Throwable throwable) {
-    logger.log(Level.INFO, "onConnect({0}, {1})", new Object[] {client, throwable});
+  public void onConnectFailed(OpampClient client, Throwable throwable, Duration nextTry) {
+    if (nextTry == null) {
+      logger.log(Level.INFO, "onConnect({0}, {1})", new Object[] {client, throwable});
+    } else {
+      logger.log(
+          Level.INFO,
+          "onConnect({0}, {1}, next attempt to connect in {2})",
+          new Object[] {client, throwable, nextTry});
+    }
   }
 
   @Override
-  public void onErrorResponse(OpampClient client, Opamp.ServerErrorResponse errorResponse) {
-    logger.log(Level.INFO, "onErrorResponse({0}, {1})", new Object[] {client, errorResponse});
+  public void onErrorResponse(
+      OpampClient client, Opamp.ServerErrorResponse errorResponse, Duration nextTry) {
+    if (nextTry == null) {
+      logger.log(Level.INFO, "onErrorResponse({0}, {1})", new Object[] {client, errorResponse});
+    } else {
+      logger.log(
+          Level.INFO,
+          "onErrorResponse({0}, {1}, next attempt to send in {2})",
+          new Object[] {client, errorResponse, nextTry});
+    }
   }
 
   public static class Builder {
@@ -175,6 +190,7 @@ public CentralConfigurationManagerImpl build() {
       OpampClientBuilder builder = OpampClient.builder();
       OkHttpSender httpSender = OkHttpSender.create("http://localhost:4320/v1/opamp");
       PeriodicDelay pollingDelay = HttpRequestService.DEFAULT_DELAY_BETWEEN_REQUESTS;
+      PeriodicDelay retryDelay = PeriodicDelay.ofVariableDuration(pollingDelay.getNextDelay());
       if (serviceName != null) {
         builder.setServiceName(serviceName);
       }
@@ -192,8 +208,9 @@ public CentralConfigurationManagerImpl build() {
       }
       if (pollingInterval != null) {
         pollingDelay = PeriodicDelay.ofFixedDuration(pollingInterval);
+        retryDelay = PeriodicDelay.ofVariableDuration(pollingInterval);
       }
-      builder.setRequestService(HttpRequestService.create(httpSender, pollingDelay, pollingDelay));
+      builder.setRequestService(HttpRequestService.create(httpSender, pollingDelay, retryDelay));
       return new CentralConfigurationManagerImpl(builder.build());
     }
   }

opamp/src/main/java/co/elastic/opamp/client/OpampClient.java

@@ -19,6 +19,7 @@
 package co.elastic.opamp.client;
 
 import co.elastic.opamp.client.response.MessageData;
+import java.time.Duration;
 import opamp.proto.Opamp;
 
 public interface OpampClient {
@@ -31,7 +32,7 @@ static OpampClientBuilder builder() {
    * Starts the client and begin attempts to connect to the Server. Once connection is established
    * the client will attempt to maintain it by reconnecting if the connection is lost. All failed
    * connection attempts will be reported via {@link Callback#onConnectFailed(OpampClient,
-   * Throwable)} callback.
+   * Throwable, Duration)} callback.
    *
    * <p>This method does not wait until the connection to the Server is established and will likely
    * return before the connection attempts are even made.
@@ -78,7 +79,7 @@ interface Callback {
      * @param client The relevant {@link co.elastic.opamp.client.OpampClient} instance.
      * @param throwable The exception.
      */
-    void onConnectFailed(OpampClient client, Throwable throwable);
+    void onConnectFailed(OpampClient client, Throwable throwable, Duration nextTry);
 
     /**
      * Called when the Server reports an error in response to some previously sent request. Useful
@@ -89,7 +90,8 @@ interface Callback {
      * @param client The relevant {@link co.elastic.opamp.client.OpampClient} instance.
      * @param errorResponse The error returned by the Server.
      */
-    void onErrorResponse(OpampClient client, Opamp.ServerErrorResponse errorResponse);
+    void onErrorResponse(
+        OpampClient client, Opamp.ServerErrorResponse errorResponse, Duration nextTry);
 
     /**
      * Called when the Agent receives a message that needs processing. See {@link

opamp/src/main/java/co/elastic/opamp/client/internal/OpampClientImpl.java

@@ -40,6 +40,7 @@
 import co.elastic.opamp.client.response.Response;
 import co.elastic.opamp.client.state.observer.Observable;
 import com.google.protobuf.ByteString;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.locks.Lock;
@@ -146,8 +147,8 @@ public void onConnectionSuccess() {
   }
 
   @Override
-  public void onConnectionFailed(Throwable throwable) {
-    callback.onConnectFailed(this, throwable);
+  public void onConnectionFailed(Throwable throwable, Duration nextTry) {
+    callback.onConnectFailed(this, throwable, nextTry);
     preserveFailedRequestRecipe();
   }
 
@@ -160,7 +161,7 @@ public void onRequestSuccess(Response response) {
   }
 
   @Override
-  public void onRequestFailed(Throwable throwable) {
+  public void onRequestFailed(Throwable throwable, Duration nextTry) {
     final Opamp.ServerErrorResponse error;
     if (throwable == null) {
       error =
@@ -175,14 +176,14 @@ public void onRequestFailed(Throwable throwable) {
                       throwable.getClass().getName() + ": " + throwable.getMessage()))
               .build();
     }
-    callback.onErrorResponse(this, error);
+    callback.onErrorResponse(this, error, nextTry);
     preserveFailedRequestRecipe();
   }
 
   private void handleResponsePayload(Opamp.ServerToAgent response) {
     if (response.hasErrorResponse()) {
       Opamp.ServerErrorResponse errorResponse = response.getErrorResponse();
-      callback.onErrorResponse(this, errorResponse);
+      callback.onErrorResponse(this, errorResponse, null);
     }
     long reportFullState = Opamp.ServerToAgentFlags.ServerToAgentFlags_ReportFullState_VALUE;
     if ((response.getFlags() & reportFullState) == reportFullState) {

opamp/src/main/java/co/elastic/opamp/client/request/delay/PeriodicDelay.java

@@ -25,6 +25,10 @@ static PeriodicDelay ofFixedDuration(Duration duration) {
     return new FixedPeriodicDelay(duration);
   }
 
+  static PeriodicDelay ofVariableDuration(Duration duration) {
+    return new VariablePeriodicDelay(duration);
+  }
+
   Duration getNextDelay();
 
   void reset();

opamp/src/main/java/co/elastic/opamp/client/request/delay/VariablePeriodicDelay.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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 co.elastic.opamp.client.request.delay;
+
+import java.time.Duration;
+
+final class VariablePeriodicDelay implements PeriodicDelay, AcceptsDelaySuggestion {
+  private volatile Duration duration;
+
+  public VariablePeriodicDelay(Duration duration) {
+    this.duration = duration;
+  }
+
+  @Override
+  public Duration getNextDelay() {
+    return duration;
+  }
+
+  @Override
+  public void reset() {}
+
+  @Override
+  public void suggestDelay(Duration delay) {
+    duration = delay;
+  }
+}

opamp/src/main/java/co/elastic/opamp/client/request/service/HttpRequestService.java

@@ -45,6 +45,7 @@ public final class HttpRequestService implements RequestService, Runnable {
   private Supplier<Request> requestSupplier;
   private boolean isRunning = false;
   private boolean isStopped = false;
+  private int exponentialBackoffSkips;
   public static final PeriodicDelay DEFAULT_DELAY_BETWEEN_REQUESTS =
       PeriodicDelay.ofFixedDuration(Duration.ofSeconds(30));
 
@@ -114,15 +115,26 @@ public void stop() {
     }
   }
 
-  private void enableRetryMode(Duration suggestedDelay) {
-    if (retryModeEnabled.compareAndSet(false, true)) {
+  private void setRetryMode(Duration suggestedDelay) {
+    retryModeEnabled.set(true);
+    setRetryModeIfEnabled(suggestedDelay);
+  }
+
+  private void setRetryModeIfEnabled(Duration suggestedDelay) {
+    if (retryModeEnabled.get()) {
       if (suggestedDelay != null && periodicRetryDelay instanceof AcceptsDelaySuggestion) {
         ((AcceptsDelaySuggestion) periodicRetryDelay).suggestDelay(suggestedDelay);
       }
       executor.setPeriodicDelay(periodicRetryDelay);
     }
   }
 
+  private void enableRetryMode(Duration suggestedDelay) {
+    if (retryModeEnabled.compareAndSet(false, true)) {
+      setRetryModeIfEnabled(suggestedDelay);
+    }
+  }
+
   private void disableRetryMode() {
     if (retryModeEnabled.compareAndSet(true, false)) {
       executor.setPeriodicDelay(periodicRequestDelay);
@@ -152,25 +164,27 @@ private void doSendRequest() {
                   agentToServer.getSerializedSize())
               .get()) {
         if (isSuccessful(response)) {
+          resetExponentialBackoffSkips();
           handleResponse(
               Response.create(Opamp.ServerToAgent.parseFrom(response.bodyInputStream())));
         } else {
           handleHttpError(response);
         }
-      } catch (IOException e) {
-        callback.onRequestFailed(e);
       }
-
-    } catch (InterruptedException e) {
-      callback.onRequestFailed(e);
+    } catch (IOException | InterruptedException e) {
+      incrementExponentialBackoff();
+      callback.onConnectionFailed(e, periodicRetryDelay.getNextDelay());
     } catch (ExecutionException e) {
-      callback.onRequestFailed(e.getCause());
+      incrementExponentialBackoff();
+      callback.onConnectionFailed(e.getCause(), periodicRetryDelay.getNextDelay());
     }
   }
 
   private void handleHttpError(HttpSender.Response response) {
     int errorCode = response.statusCode();
-    callback.onRequestFailed(new HttpErrorException(errorCode, response.statusMessage()));
+    callback.onRequestFailed(
+        new HttpErrorException(errorCode, response.statusMessage()),
+        periodicRetryDelay.getNextDelay());
 
     if (errorCode == 503 || errorCode == 429) {
       String retryAfterHeader = response.getHeader("Retry-After");
@@ -208,9 +222,30 @@ private void handleErrorResponse(Opamp.ServerErrorResponse errorResponse) {
         retryAfter = Duration.ofNanos(errorResponse.getRetryInfo().getRetryAfterNanoseconds());
       }
       enableRetryMode(retryAfter);
+    } else {
+      incrementExponentialBackoff();
+      Duration retryAfter = Duration.ofSeconds(30 * exponentialBackoffSkips);
+      enableRetryMode(retryAfter);
     }
   }
 
+  private void incrementExponentialBackoff() {
+    if (exponentialBackoffSkips == 0) {
+      exponentialBackoffSkips = 1;
+    } else {
+      exponentialBackoffSkips *= 2;
+    }
+    if (exponentialBackoffSkips >= 32) {
+      exponentialBackoffSkips = 32;
+    }
+    setRetryMode(Duration.ofSeconds(30 * exponentialBackoffSkips));
+  }
+
+  private void resetExponentialBackoffSkips() {
+    exponentialBackoffSkips = 0;
+    disableRetryMode();
+  }
+
   private static class ByteArrayWriter implements Consumer<OutputStream> {
     private final byte[] data;
 

opamp/src/main/java/co/elastic/opamp/client/request/service/RequestService.java

@@ -21,6 +21,7 @@
 import co.elastic.opamp.client.OpampClient;
 import co.elastic.opamp.client.request.Request;
 import co.elastic.opamp.client.response.Response;
+import java.time.Duration;
 import java.util.function.Supplier;
 
 /**
@@ -66,7 +67,7 @@ interface Callback {
      *
      * @param throwable The detailed error.
      */
-    void onConnectionFailed(Throwable throwable);
+    void onConnectionFailed(Throwable throwable, Duration nextTry);
 
     /**
      * For WebSocket implementations, this is called every time there's a new message from the
@@ -83,6 +84,6 @@ interface Callback {
      *
      * @param throwable The detailed error.
      */
-    void onRequestFailed(Throwable throwable);
+    void onRequestFailed(Throwable throwable, Duration nextTry);
   }
 }

opamp/src/main/java/co/elastic/opamp/client/request/service/WebSocketRequestService.java

@@ -97,7 +97,7 @@ private void doSendRequest() {
       codedOutput.flush();
       webSocket.send(outputStream.toByteArray());
     } catch (IOException e) {
-      callback.onRequestFailed(e);
+      callback.onRequestFailed(e, null);
     }
   }
 
@@ -128,7 +128,7 @@ public void onMessage(WebSocket webSocket, byte[] data) {
 
       callback.onRequestSuccess(Response.create(serverToAgent));
     } catch (IOException e) {
-      callback.onRequestFailed(e);
+      callback.onRequestFailed(e, null);
     }
   }
 
@@ -185,7 +185,7 @@ public void onClosed(WebSocket webSocket) {
 
   @Override
   public void onFailure(WebSocket webSocket, Throwable t) {
-    callback.onConnectionFailed(t);
+    callback.onConnectionFailed(t, null);
     enableRetryMode(null);
   }