Merge pull request #57 from exceptionless/rate-limiting

Implemented rate limiting detection while posting events

Author: pc9795

src/main/java/com/exceptionless/exceptionlessclient/configuration/Configuration.java

@@ -18,6 +18,7 @@ public class Configuration {
   public static class Property {
     public static final String API_KEY = "apiKey";
     public static final String SERVER_URL = "serverUrl";
+    public static final String CONFIG_SERVER_URL = "configServerUrl";
     public static final String HEART_BEAT_SERVER_URL = "heartbeatServerUrl";
     public static final String UPDATE_SETTINGS_WHEN_IDLE_INTERVAL =
         "updateSettingsWhenIdleInterval";
@@ -94,6 +95,13 @@ public void setServerUrl(String serverUrl) {
     propertyChangeSupport.firePropertyChange(Property.SERVER_URL, prevValue, serverUrl);
   }
 
+  public void setConfigServerUrl(String configServerUrl) {
+    String prevValue = this.configServerUrl;
+    this.configServerUrl = configServerUrl;
+    propertyChangeSupport.firePropertyChange(
+        Property.CONFIG_SERVER_URL, prevValue, configServerUrl);
+  }
+
   public void setHeartbeatServerUrl(String heartbeatServerUrl) {
     String prevValue = this.heartbeatServerUrl;
     this.heartbeatServerUrl = heartbeatServerUrl;

src/main/java/com/exceptionless/exceptionlessclient/queue/DefaultEventQueue.java

@@ -159,6 +159,13 @@ private void processSubmissionResponse(
       return;
     }
 
+    if (response.isRateLimited()) {
+      log.error(
+          "Service is rate limited because of either you have exceeded your rate limit or server is under stress.");
+      suspendProcessing();
+      return;
+    }
+
     if (response.isServiceUnavailable()) {
       log.error("Service returns service unavailable");
       suspendProcessing();

src/main/java/com/exceptionless/exceptionlessclient/submission/DefaultSubmissionClient.java

@@ -14,12 +14,12 @@
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 @Slf4j
 public class DefaultSubmissionClient implements SubmissionClientIF {
   private static final String CONFIGURATION_VERSION_HEADER = "x-exceptionless-configversion";
+  private static final String RATE_LIMITING_HEADER = "x-ratelimit-remaining";
 
   private final Configuration configuration;
   private final SettingsManager settingsManager;
@@ -74,29 +74,33 @@ private SubmissionResponse postSubmission(String url, Object data) {
       Response response = httpClient.newCall(request).execute();
 
       if (response.isSuccessful()) {
-        updateSettingsFromHeaders(response.headers());
+        updateSettingsFromHeaders(response);
       }
 
       ResponseBody body = response.body();
       return SubmissionResponse.builder()
           .code(response.code())
           .body(body == null ? "" : body.string())
+          .rateLimitingHeaderFound(isRateLimitingHeaderFound(response))
           .build();
     } catch (Exception e) {
       return SubmissionResponse.builder().exception(e).build();
     }
   }
 
-  private void updateSettingsFromHeaders(Headers headers) {
-    Optional<String> maybeSettingsVersion =
-        Optional.ofNullable(headers.get(CONFIGURATION_VERSION_HEADER));
-    if (maybeSettingsVersion.isPresent()) {
-      settingsManager.checkVersion(Long.parseLong(maybeSettingsVersion.get()));
+  private void updateSettingsFromHeaders(Response response) {
+    String settingsVersion = response.headers().get(CONFIGURATION_VERSION_HEADER);
+    if (settingsVersion != null) {
+      settingsManager.checkVersion(Long.parseLong(settingsVersion));
     } else {
       log.error("No config version header was returned");
     }
   }
 
+  private boolean isRateLimitingHeaderFound(Response response) {
+    return response.headers().get(RATE_LIMITING_HEADER) != null;
+  }
+
   @Override
   public void sendHeartBeat(String sessionIdOrUserId, boolean closeSession) {
     try {

src/main/java/com/exceptionless/exceptionlessclient/submission/SubmissionResponse.java

@@ -8,6 +8,7 @@
 public class SubmissionResponse {
   int code;
   String body;
+  boolean rateLimitingHeaderFound;
   Exception exception;
 
   public boolean isSuccess() {
@@ -38,6 +39,10 @@ public boolean isRequestEntityTooLarge() {
     return code == 413;
   }
 
+  public boolean isRateLimited() {
+    return rateLimitingHeaderFound || code == 429;
+  }
+
   public boolean hasException() {
     return exception != null;
   }

src/test/java/com/exceptionless/exceptionlessclient/configuration/ConfigurationTest.java

@@ -31,6 +31,7 @@ public void itCanNotifyChangesToListeners() {
         Configuration.builder()
             .apiKey("old-api-key")
             .serverUrl("old-server-url")
+            .configServerUrl("old-config-server-url")
             .heartbeatServerUrl("old-heartbeat-server-url")
             .updateSettingsWhenIdleInterval(1L)
             .submissionBatchSize(50)
@@ -41,6 +42,7 @@ public void itCanNotifyChangesToListeners() {
 
     configuration.setApiKey("new-api-key");
     configuration.setServerUrl("new-server-url");
+    configuration.setConfigServerUrl("new-config-server-url");
     configuration.setHeartbeatServerUrl("new-heartbeat-server-url");
     configuration.setUpdateSettingsWhenIdleInterval(2L);
     configuration.setSubmissionBatchSize(100);
@@ -51,15 +53,32 @@ public void itCanNotifyChangesToListeners() {
         List.of(
             "apiKey",
             "serverUrl",
+            "configServerUrl",
             "heartbeatServerUrl",
             "updateSettingsWhenIdleInterval",
             "submissionBatchSize",
             "submissionClientTimeoutInMillis",
             "settingsClientTimeoutInMillis");
     List<Object> oldValues =
-        List.of("old-api-key", "old-server-url", "old-heartbeat-server-url", 1L, 50, 10, 10);
+        List.of(
+            "old-api-key",
+            "old-server-url",
+            "old-config-server-url",
+            "old-heartbeat-server-url",
+            1L,
+            50,
+            10,
+            10);
     List<Object> newValues =
-        List.of("new-api-key", "new-server-url", "new-heartbeat-server-url", 2L, 100, 20, 20);
+        List.of(
+            "new-api-key",
+            "new-server-url",
+            "new-config-server-url",
+            "new-heartbeat-server-url",
+            2L,
+            100,
+            20,
+            20);
 
     properties.forEach(
         property ->

src/test/java/com/exceptionless/exceptionlessclient/queue/DefaultEventQueueTest.java

@@ -138,6 +138,42 @@ public void itShouldSuspendProcessingIfServiceIsUnavailable() {
     verify(testHandler, times(1)).accept(List.of(event), response);
   }
 
+  @Test
+  public void itShouldSuspendProcessingIfServiceIsRateLimitedByHeader() {
+    storage.save(event);
+
+    SubmissionResponse response =
+        SubmissionResponse.builder()
+            .body("test-message")
+            .code(999)
+            .rateLimitingHeaderFound(true)
+            .build();
+    doReturn(response).when(submissionClient).postEvents(List.of(event));
+
+    queue.onEventsPosted(testHandler);
+    queue.process();
+
+    assertThat(queue.isProcessingCurrentlySuspended()).isTrue();
+    assertThat(storage.peek().getValue()).isEqualTo(event);
+    verify(testHandler, times(1)).accept(List.of(event), response);
+  }
+
+  @Test
+  public void itShouldSuspendProcessingIfServiceIsRateLimitedByCode() {
+    storage.save(event);
+
+    SubmissionResponse response =
+        SubmissionResponse.builder().body("test-message").code(429).build();
+    doReturn(response).when(submissionClient).postEvents(List.of(event));
+
+    queue.onEventsPosted(testHandler);
+    queue.process();
+
+    assertThat(queue.isProcessingCurrentlySuspended()).isTrue();
+    assertThat(storage.peek().getValue()).isEqualTo(event);
+    verify(testHandler, times(1)).accept(List.of(event), response);
+  }
+
   @Test
   public void itShouldSuspendAndDiscardProcessingAndClearQueueIfNoPayment() {
     storage.save(event);

src/test/java/com/exceptionless/exceptionlessclient/submission/DefaultSubmissionClientTest.java

@@ -74,6 +74,36 @@ public void itCanPostEventsSuccessfully() throws IOException {
     verify(settingsManager, times(1)).checkVersion(3);
   }
 
+  @Test
+  public void itCanDetectRateLimitingFromHeaders() throws IOException {
+    Response response =
+        responseBuilder
+            .headers(
+                Headers.of(
+                    Map.of("x-exceptionless-configversion", "3", "x-ratelimit-remaining", "0")))
+            .build();
+    doReturn(response).when(call).execute();
+    doReturn(call)
+        .when(httpClient)
+        .newCall(
+            argThat(
+                request ->
+                    request
+                            .url()
+                            .toString()
+                            .equals(
+                                "http://test-server-url/api/v2/events?access_token=test-api-key")
+                        && request.method().equals("POST")));
+
+    SubmissionResponse submissionResponse =
+        submissionClient.postEvents(List.of(Event.builder().build()));
+
+    assertThat(submissionResponse.getBody()).isEqualTo("test-body");
+    assertThat(submissionResponse.getCode()).isEqualTo(200);
+    verify(settingsManager, times(1)).checkVersion(3);
+    assertThat(submissionResponse.isRateLimited()).isTrue();
+  }
+
   @Test
   public void itCanPostEventsSuccessfullyWhenNoSettingHeaderIsReturned() throws IOException {
     doReturn(responseBuilder.build()).when(call).execute();