RandomDestination at first & round-robin

src/main/java/net/logstash/logback/appender/AbstractLogstashTcpSocketAppender.java

@@ -27,6 +27,7 @@
 import java.util.Collections;
 import java.util.Formatter;
 import java.util.List;
+import java.util.Random;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
@@ -113,7 +114,7 @@ public abstract class AbstractLogstashTcpSocketAppender<Event extends DeferredPr
      * Index into {@link #destinations} of the "primary" destination.
      */
     private static final int PRIMARY_DESTINATION_INDEX = 0;
-    
+
     /**
      * The host to which to connect and send events
      */
@@ -252,7 +253,32 @@ public abstract class AbstractLogstashTcpSocketAppender<Event extends DeferredPr
      * @see #isGetHostStringPossible()
      */
     private final boolean getHostStringPossible = isGetHostStringPossible();
-
+
+    /**
+     * The first set destination is random.
+     *
+     * Defaults to disable. The first destination is considered the primary destination.
+     */
+    private boolean randomDestination;
+
+    /**
+     * Object to create the first random index.
+     */
+    private Random firstRandomIndexGenerator = new Random();
+
+    /**
+     * If the size is exceeded, round-robin occurs.
+     * Destination cycles by round-robin.
+     *
+     * When null (the default), disabled round-robin.
+     */
+    private Integer roundRobinSize;
+
+    /**
+     * Count of round-robin
+     */
+    private int roundRobinCount;
+
     /**
      * Event handler responsible for performing the TCP transmission.
      */
@@ -284,7 +310,7 @@ private class TcpSendingEventHandler implements EventHandler<LogEvent<Event>>, L
          * This is a buffered wrapper of the socket output stream.
          */
         private volatile OutputStream outputStream;
-        
+
         /**
          * Time at which the last event was sent.
          * Used to calculate if a keep alive message
@@ -333,7 +359,7 @@ private class TcpSendingEventHandler implements EventHandler<LogEvent<Event>>, L
          */
         private class KeepAliveRunnable implements Runnable {
 
-            private int previousDestinationIndex = PRIMARY_DESTINATION_INDEX;
+            private int previousDestinationIndex = connectedDestinationIndex;
 
             @Override
             public void run() {
@@ -342,8 +368,8 @@ public void run() {
                 if (hasKeepAliveDurationElapsed(lastSent, currentTime)) {
                     /*
                      * Publish a keep alive message to the RingBuffer.
-                     * 
-                     * A null event indicates that this is a keep alive message. 
+                     *
+                     * A null event indicates that this is a keep alive message.
                      */
                     getDisruptor().getRingBuffer().publishEvent(getEventTranslator(), null);
                     scheduleKeepAlive(currentTime);
@@ -439,6 +465,7 @@ public void onEvent(LogEvent<Event> logEvent, long sequence, boolean endOfBatch)
                          * Therefore, we need to send the event.
                          */
                         encoder.doEncode(logEvent.event);
+                        roundRobin();
                     } else if (hasKeepAliveDurationElapsed(lastSentTimestamp, currentTime)){
                         /*
                          * This is a keep alive event, and the keepAliveDuration has passed,
@@ -474,6 +501,21 @@ public void onEvent(LogEvent<Event> logEvent, long sequence, boolean endOfBatch)
             }
         }
 
+        /**
+         * If the condition is satisfied, round-robin occurs.
+         * The index is repeated in a circle.
+         */
+        private void roundRobin() {
+            if (isRoundRobin() && (roundRobinCount++ >= roundRobinSize)) {
+				connectedDestinationIndex++;
+				if (connectedDestinationIndex >= destinations.size()) {
+					connectedDestinationIndex = PRIMARY_DESTINATION_INDEX;
+				}
+				roundRobinCount = 0;
+				reopenSocket();
+			}
+        }
+
         private boolean hasKeepAliveDurationElapsed(long lastSent, long currentTime) {
             return isKeepAliveEnabled()
                     && lastSent + keepAliveDuration.getMilliseconds() < currentTime;
@@ -485,6 +527,9 @@ private boolean shouldCloseConnection(long currentTime) {
 
         @Override
         public void onStart() {
+            if (randomDestination) {
+                connectedDestinationIndex = firstRandomIndexGenerator.nextInt(destinations.size());
+            }
             openSocket();
             scheduleKeepAlive(System.currentTimeMillis());
         }
@@ -509,7 +554,8 @@ private synchronized void reopenSocket() {
          * then it should be able to be used to send.
          */
         private synchronized void openSocket() {
-            int destinationIndex = 0;
+            int destinationIndex = randomDestination || isRoundRobin() ? connectedDestinationIndex : PRIMARY_DESTINATION_INDEX;
+            int retryCount = 0;
             int errorCount = 0;
             while (isStarted() && !Thread.currentThread().isInterrupted()) {
                 long startTime = System.currentTimeMillis();
@@ -563,24 +609,25 @@ private synchronized void openSocket() {
                          */
                         updateCurrentThreadName();
                     }
-                    
+
                     this.readerFuture = scheduleReaderRunnable(
                             new ReaderRunnable(tempSocket.getInputStream()));
 
                     return;
 
                 } catch (Exception e) {
-                    
+
                     CloseUtil.closeQuietly(tempOutputStream);
                     CloseUtil.closeQuietly(tempSocket);
 
                     /*
                      * Retry immediately with next available host if any. Otherwise, sleep and retry with primary
                      */
                     destinationIndex++;
-                    if (destinationIndex >= destinations.size()) {
+                    retryCount++;
+                    if (destinationIndex >= destinations.size() && (retryCount >= destinations.size())) {
                         destinationIndex = PRIMARY_DESTINATION_INDEX;
-                        
+
                         /*
                          * If the connection timed out, then take the elapsed time into account
                          * when calculating time to sleep
@@ -604,6 +651,9 @@ private synchronized void openSocket() {
                         }
                     }
                     else {
+                        if (destinationIndex >= destinations.size()) {
+                            destinationIndex = PRIMARY_DESTINATION_INDEX;
+                        }
                         /*
                          * Avoid spamming status messages by checking the MAX_REPEAT_CONNECTION_ERROR_LOG.
                          */
@@ -677,7 +727,11 @@ private synchronized void unscheduleKeepAlive() {
             }
         }
     }
-
+
+    private boolean isRoundRobin() {
+        return roundRobinSize != null;
+    }
+
     /**
      * An extension of logback's {@link ConfigurableSSLSocketFactory}
      * that supports creating unconnected sockets
@@ -1132,4 +1186,34 @@ public void setThreadNameFormat(String threadNameFormat) {
         super.setThreadNameFormat(threadNameFormat);
     }
 
+    public boolean isRandomDestination() {
+        return randomDestination;
+    }
+
+    /**
+     * Sets random Destination. The first set destination is random.
+     *
+     * <p>
+     * @param randomDestination enable random Destination.
+     */
+    public void setRandomDestination(boolean randomDestination) {
+        this.randomDestination = randomDestination;
+    }
+
+    public Integer getRoundRobinSize() {
+        return roundRobinSize;
+    }
+
+    /**
+     * If the size is exceeded, round-robin occurs.
+     * Destination cycles by round-robin.
+     *
+     * <p>
+     * When null (the default), disabled round-robin.
+     *
+     * @param roundRobinSize the number of round-robin incidents.
+     */
+    public void setRoundRobinSize(Integer roundRobinSize) {
+        this.roundRobinSize = roundRobinSize;
+    }
 }

src/test/java/net/logstash/logback/appender/LogstashTcpSocketAppenderTest.java

@@ -90,6 +90,9 @@ public class LogstashTcpSocketAppenderTest {
 
     @Mock
     private Future<?> readableRunnableFuture;
+
+    @Mock
+    private RandomWrapper randomWrapper;
 
     private class TestableLogstashTcpSocketAppender extends LogstashTcpSocketAppender {
         @Override
@@ -251,6 +254,35 @@ public void testReconnectToSecondaryOnOpen() throws Exception {
         // 2) Second attempt on SECONDARY
         inOrder.verify(socket).connect(host("localhost", 10001), anyInt());
     }
+
+    @Test
+    public void testRandomDestinationAndReconnectToSecondaryOnOpen() throws Exception {
+        appender.addDestination("localhost:10000");
+        appender.addDestination("localhost:10001");
+        appender.setRandomDestination(true);
+
+        // Make it failed to connect to second destination
+        doThrow(SocketTimeoutException.class)
+            .when(socket).connect(host("localhost", 10001), anyInt());
+
+        // The first index is second destination.
+        when(randomWrapper.nextInt(appender.getDestinations().size())).thenReturn(1);
+
+        // Start the appender and verify it is actually started.
+        // It should try to connect to second destination by random destination, fail then retry on first destination.
+        appender.start();
+        verify(encoder).start();
+
+        // TWO connection attempts must have been made (without delay)
+        verify(socket, timeout(VERIFICATION_TIMEOUT).times(2)).connect(any(SocketAddress.class), anyInt());
+        InOrder inOrder = inOrder(socket);
+
+        // 1) First attempt on second destination: failure
+        inOrder.verify(socket).connect(host("localhost", 10001), anyInt());
+
+        // 2) Second attempt on first destination
+        inOrder.verify(socket).connect(host("localhost", 10000), anyInt());
+    }
 
 
     /**
@@ -310,12 +342,12 @@ public void testReconnectToPrimaryWhileOnSecondary() throws Exception {
         appender.addDestination("localhost:10000");
         appender.addDestination("localhost:10001");
         appender.setSecondaryConnectionTTL(Duration.buildByMilliseconds(100));
-        
+
         // Primary refuses first connection to force the appender to go on the secondary.
         doThrow(SocketTimeoutException.class)
             .doNothing()
             .when(socket).connect(host("localhost", 10000), anyInt());
-        
+
 
         // Start the appender and verify it is actually started
         // At this point, it should be connected to primary.
@@ -494,6 +526,32 @@ public void testDestination_MixedType() throws Exception {
         appender.start();
         Assert.assertFalse(appender.isStarted());
     }
+
+    @Test
+    public void testRoundRobin() throws Exception {
+        appender.addDestination("localhost:10000");
+        appender.addDestination("localhost:10001");
+        appender.setRoundRobinSize(2);
+
+        appender.start();
+
+        verify(encoder).start();
+
+        appender.append(event1);
+        appender.append(event1);
+        appender.append(event1);
+
+        verify(socket, timeout(VERIFICATION_TIMEOUT).times(2)).connect(any(SocketAddress.class), anyInt());
+        InOrder inOrder = inOrder(socket);
+
+        // 1) connected to primary at startup
+        inOrder.verify(socket).connect(host("localhost", 10000), anyInt());
+
+        // 2) connected to next destination by round-robin
+        inOrder.verify(socket).connect(host("localhost", 10001), anyInt());
+
+
+    }
 
     private SocketAddress host(final String host, final int port) {
     	return argThat(hasHostAndPort(host, port));

src/test/java/net/logstash/logback/appender/RandomWrapper.java

@@ -0,0 +1,46 @@
+/**
+ * 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 net.logstash.logback.appender;
+
+import java.util.Random;
+
+/**
+ * RandomWrapper
+ *
+ * @author [email protected]
+ * @since 2016. 12. 22.
+ */
+public class RandomWrapper extends Random {
+	int seq = 0;
+	public RandomWrapper() {
+		super();
+	}
+
+	@Override
+	public int nextInt() {
+		return seq++;
+	}
+
+	@Override
+	public int nextInt(int bound) {
+		int result = seq;
+		if (result < bound) {
+			seq++;
+		} else {
+			result = 0;
+			seq = 1;
+		}
+		return result;
+	}
+}