Fork HttpEventCollectorResendMiddleware to properly implement retrying

runtime/src/main/java/io/quarkiverse/logging/splunk/SplunkLogHandler.java

@@ -10,11 +10,12 @@
 import java.util.logging.Filter;
 import java.util.logging.Formatter;
 
+import io.quarkiverse.logging.splunk.middleware.HttpEventCollectorResendMiddleware;
 import org.jboss.logmanager.ExtHandler;
 import org.jboss.logmanager.ExtLogRecord;
 import org.jboss.logmanager.filters.AllFilter;
 
-import com.splunk.logging.HttpEventCollectorResendMiddleware;
+
 import com.splunk.logging.HttpEventCollectorSender;
 
 public class SplunkLogHandler extends ExtHandler {
@@ -45,7 +46,7 @@ public SplunkLogHandler(HttpEventCollectorSender sender, boolean includeExceptio
     @Override
     public void doPublish(ExtLogRecord record) {
         String formatted = formatMessage(record);
-        if (formatted.length() == 0) {
+        if (formatted.isEmpty()) {
             // nothing to write; don't bother
             return;
         }
@@ -54,7 +55,7 @@ public void doPublish(ExtLogRecord record) {
                 record.getLevel().toString(),
                 formatted,
                 includeLoggerName ? record.getLoggerName() : null,
-                includeThreadName ? String.format(Locale.US, "%d", record.getThreadID()) : null,
+                includeThreadName ? String.format(Locale.US, "%d", record.getLongThreadID()) : null,
                 record.getMdcCopy(),
                 (!includeException || record.getThrown() == null) ? null : record.getThrown().getMessage(),
                 null);

runtime/src/main/java/io/quarkiverse/logging/splunk/middleware/HttpEventCollectorResendMiddleware.java

@@ -0,0 +1,125 @@
+package io.quarkiverse.logging.splunk.middleware;
+
+import com.splunk.logging.HttpEventCollectorEventInfo;
+import com.splunk.logging.HttpEventCollectorMiddleware;
+
+import java.util.Arrays;
+import java.util.HashSet;
+
+/**
+ * @copyright
+ *
+ * Copyright 2013-2015 Splunk, Inc.
+ * Derived from https://github.com/splunk/splunk-library-javalogging/pull/287
+ * by Simon Hege.
+ *
+ * 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.
+ */
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Splunk http event collector resend middleware.
+ *
+ *
+ * HTTP event collector middleware plug in that implements a simple resend policy.
+ * When HTTP post reply isn't an application error it tries to resend the data.
+ * An exponentially growing delay is used to prevent server overflow.
+ */
+public class HttpEventCollectorResendMiddleware
+        extends HttpEventCollectorMiddleware.HttpSenderMiddleware {
+
+    /**
+     * List of HTTP event collector server application error statuses. These statuses
+     * indicate non-transient problems that cannot be fixed by resending the
+     * data.
+     */
+    private static final Set<Integer> HttpEventCollectorApplicationErrors  = new HashSet<>(Arrays.asList(
+            // Forbidden
+            403,
+            // Method Not Allowed
+            405,
+            // Bad Request
+            400
+    ));
+    private long retriesOnError = 0;
+
+    /**
+     * Create a resend middleware component.
+     * @param retriesOnError is the max retry count.
+     */
+    public HttpEventCollectorResendMiddleware(long retriesOnError) {
+        this.retriesOnError = retriesOnError;
+    }
+
+    public void postEvents(
+            final List<HttpEventCollectorEventInfo> events,
+            HttpEventCollectorMiddleware.IHttpSender sender,
+            HttpEventCollectorMiddleware.IHttpSenderCallback callback) {
+        callNext(events, sender, new Callback(events, sender, callback));
+    }
+
+    private boolean shouldRetry(int statusCode) {
+        return statusCode != 200 && !HttpEventCollectorApplicationErrors.contains(statusCode);
+    }
+
+    private class Callback implements HttpEventCollectorMiddleware.IHttpSenderCallback {
+        private long retries = 0;
+        private final List<HttpEventCollectorEventInfo> events;
+        private HttpEventCollectorMiddleware.IHttpSenderCallback prevCallback;
+        private HttpEventCollectorMiddleware.IHttpSender sender;
+        private final long RetryDelayCeiling = 60 * 1000; // 1 minute
+        private long retryDelay = 1000; // start with 1 second
+
+        public Callback(
+                final List<HttpEventCollectorEventInfo> events,
+                HttpEventCollectorMiddleware.IHttpSender sender,
+                HttpEventCollectorMiddleware.IHttpSenderCallback prevCallback) {
+            this.events = events;
+            this.prevCallback = prevCallback;
+            this.sender = sender;
+        }
+
+        @Override
+        public void completed(int statusCode, final String reply) {
+            if (shouldRetry(statusCode) && retries < retriesOnError) {
+                retry();
+            } else {
+                // if non-retryable, resend wouldn't help, delegate to previous callback
+                prevCallback.completed(statusCode, reply);
+            }
+        }
+
+        @Override
+        public void failed(final Exception ex) {
+            if (retries < retriesOnError) {
+                retry();
+            } else {
+                prevCallback.failed(ex);
+            }
+        }
+
+        private void retry() {
+            retries++;
+            try {
+                Thread.sleep(retryDelay);
+                callNext(events, sender, this);
+            } catch (InterruptedException ie) {
+                prevCallback.failed(ie);
+            }
+            // increase delay exponentially
+            retryDelay = Math.min(RetryDelayCeiling, retryDelay * 2);
+        }
+    }
+}

runtime/src/test/java/io/quarkiverse/logging/splunk/SplunkLogHandlerTest.java

@@ -23,9 +23,10 @@
 import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import com.splunk.logging.HttpEventCollectorResendMiddleware;
 import com.splunk.logging.HttpEventCollectorSender;
 
+import io.quarkiverse.logging.splunk.middleware.HttpEventCollectorResendMiddleware;
+
 @ExtendWith(MockitoExtension.class)
 class SplunkLogHandlerTest {
 

runtime/src/test/java/io/quarkiverse/logging/splunk/middleware/HttpEventCollectorResendMiddlewareTest.java

@@ -0,0 +1,131 @@
+package io.quarkiverse.logging.splunk.middleware;
+/**
+ * @copyright
+ *
+ * Copyright 2013-2015 Splunk, Inc.
+ * Derived from https://github.com/splunk/splunk-library-javalogging/pull/287
+ * by Simon Hege. Modifications include updating the unit testing framework to JUnit 5.
+ *
+ * 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.
+ */
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.splunk.logging.HttpEventCollectorMiddleware;
+import org.junit.jupiter.api.Test;
+
+
+public class HttpEventCollectorResendMiddlewareTest {
+
+    @Test
+    public void testPostEvents_whenSuccesShouldNotRetry() {
+        // Arrange
+        HttpEventCollectorResendMiddleware middleware = new HttpEventCollectorResendMiddleware(3);
+        final AtomicInteger callCount = new AtomicInteger(0);
+        HttpEventCollectorMiddleware.IHttpSender sender = getSender(callCount, 0);
+        final List<Integer> recordedStatusCodes = new ArrayList<>();
+        final List<String> recordedReplies = new ArrayList<>();
+        final List<Exception> recordedExceptions = new ArrayList<>();
+        HttpEventCollectorMiddleware.IHttpSenderCallback callback = getCallback(recordedStatusCodes, recordedReplies,
+                recordedExceptions);
+
+        // Act
+        middleware.postEvents(null, sender, callback);
+
+        // Assert
+        assertEquals(1, callCount.get());
+        assertEquals(1, recordedStatusCodes.size());
+        assertEquals(1, recordedReplies.size());
+        assertEquals(0, recordedExceptions.size());
+        assertEquals(200, recordedStatusCodes.get(0).intValue());
+        assertEquals("Success", recordedReplies.get(0));
+    }
+
+    @Test
+    public void testPostEvents_whenUnavailableThenSuccessShouldRetry() {
+        // Arrange
+        HttpEventCollectorResendMiddleware middleware = new HttpEventCollectorResendMiddleware(3);
+        final AtomicInteger callCount = new AtomicInteger(0);
+        HttpEventCollectorMiddleware.IHttpSender sender = getSender(callCount, 2);
+        final List<Integer> recordedStatusCodes = new ArrayList<>();
+        final List<String> recordedReplies = new ArrayList<>();
+        final List<Exception> recordedExceptions = new ArrayList<>();
+        HttpEventCollectorMiddleware.IHttpSenderCallback callback = getCallback(recordedStatusCodes, recordedReplies,
+                recordedExceptions);
+
+        // Act
+        middleware.postEvents(null, sender, callback);
+
+        // Assert
+        assertEquals(3, callCount.get());
+        assertEquals(1, recordedStatusCodes.size());
+        assertEquals(1, recordedReplies.size());
+        assertEquals(0, recordedExceptions.size());
+        assertEquals(200, recordedStatusCodes.get(0).intValue());
+    }
+
+    @Test
+    public void testPostEvents_whenUnavailableShouldRetryThenStop() {
+        // Arrange
+        HttpEventCollectorResendMiddleware middleware = new HttpEventCollectorResendMiddleware(3);
+        final AtomicInteger callCount = new AtomicInteger(0);
+        HttpEventCollectorMiddleware.IHttpSender sender = getSender(callCount, 10);
+        final List<Integer> recordedStatusCodes = new ArrayList<>();
+        final List<String> recordedReplies = new ArrayList<>();
+        final List<Exception> recordedExceptions = new ArrayList<>();
+        HttpEventCollectorMiddleware.IHttpSenderCallback callback = getCallback(recordedStatusCodes, recordedReplies,
+                recordedExceptions);
+
+        // Act
+        middleware.postEvents(null, sender, callback);
+
+        // Assert
+        assertEquals(4, callCount.get());
+        assertEquals(1, recordedStatusCodes.size());
+        assertEquals(1, recordedReplies.size());
+        assertEquals(0, recordedExceptions.size());
+        assertEquals(503, recordedStatusCodes.get(0).intValue());
+    }
+
+    private static HttpEventCollectorMiddleware.IHttpSender getSender(AtomicInteger callCount, int errorCount) {
+        return (events, callback) -> {
+            callCount.incrementAndGet();
+            if (callCount.get() > errorCount) {
+                callback.completed(200, "Success");
+            } else {
+                callback.completed(503, "Service Unavailable");
+            }
+        };
+    }
+
+    private static HttpEventCollectorMiddleware.IHttpSenderCallback getCallback(List<Integer> recordedStatusCodes,
+            List<String> recordedReplies, List<Exception> recordedExceptions) {
+        return new HttpEventCollectorMiddleware.IHttpSenderCallback() {
+
+            @Override
+            public void completed(int statusCode, String reply) {
+                recordedStatusCodes.add(statusCode);
+                recordedReplies.add(reply);
+            }
+
+            @Override
+            public void failed(Exception ex) {
+                recordedExceptions.add(ex);
+            }
+        };
+    }
+}