Fix #64: Rebind RELP connections after configurable amount of message…

…s have been sent (#66) * Open a new RELP connection after a certain amount of messages have been sent with it. Two new configuration parameters for rebinding. * Refactor counting sent records to IRelpConnection * Add more rebind tests for RelpConnectionPool * Refactor rebinding to a decorator, use generic Pool with managedRelpConnection * Remove unnecessary function already defined in extending interface, apply spotless * Add test for rebind * Remove unnecessary functions from IManagedRelpConnection interface, create new rebind tests * Change ManagedRelpConnection to only follow has been a connection made atleast once and not the connection itself. Change counter in CountingFrameDelegate to AtomicInteger. * Fix hasConnected in ManagedRelpConnection * Add tests for invalid rebindRequestAmount system property
teragrep · May 24, 2024 · 3cce9b7 · 3cce9b7
1 parent 0126474
commit 3cce9b7
Show file tree

Hide file tree

Showing 22 changed files with 592 additions and 252 deletions.
diff --git a/README.adoc b/README.adoc
@@ -27,6 +27,8 @@ server.maxContentLength,262144,How big requests are allowed in bytes
 relp.target,127.0.0.1,RELP server address
 relp.port,601,RELP server port
 relp.reconnectInterval,10000,How long to wait before reconnecting in milliseconds
+relp.rebindRequestAmount, 1000000, How many requests from a RELP connection until rebinding it
+relp.rebindEnabled, false, Sets whether rebinding RELP connections is enabled
 healthcheck.enabled,true,Sets if an internal healthcheck endpoint is enabled.
 healthcheck.url,/healthcheck,An internal healthcheck endpoint that will always reply 200 ok regardless of security settings. Accessing this url won't generate any events.
 security.authRequired,true,Sets whether Basic HTTP Authorization headers are required. Username for lookups will be empty string '' if set to false.

diff --git a/etc/config.properties b/etc/config.properties
@@ -10,6 +10,8 @@ healthcheck.url=/healthcheck
 relp.target=127.0.0.1
 relp.port=601
 relp.reconnectInterval=10000
+relp.rebindRequestAmount=1000000
+relp.rebindEnabled=false
 
 security.authRequired=true
 

diff --git a/src/main/java/com/teragrep/lsh_01/Main.java b/src/main/java/com/teragrep/lsh_01/Main.java
@@ -22,8 +22,7 @@
 import com.teragrep.lsh_01.authentication.BasicAuthentication;
 import com.teragrep.lsh_01.authentication.BasicAuthenticationFactory;
 import com.teragrep.lsh_01.config.*;
-import com.teragrep.lsh_01.pool.RelpConnectionFactory;
-import com.teragrep.lsh_01.pool.RelpConnectionPool;
+import com.teragrep.lsh_01.pool.*;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -59,10 +58,10 @@ public static void main(String[] args) {
         LOGGER.info("Authentication required: <[{}]>", securityConfig.authRequired);
 
         RelpConnectionFactory relpConnectionFactory = new RelpConnectionFactory(relpConfig);
-        RelpConnectionPool relpConnectionPool = new RelpConnectionPool(relpConnectionFactory);
+        Pool<IManagedRelpConnection> pool = new Pool<>(relpConnectionFactory, new ManagedRelpConnectionStub());
 
         RelpConversion relpConversion = new RelpConversion(
-                relpConnectionPool,
+                pool,
                 securityConfig,
                 basicAuthentication,
                 lookupConfig,
@@ -81,7 +80,7 @@ public static void main(String[] args) {
             server.run();
         }
         finally {
-            relpConnectionPool.close();
+            pool.close();
         }
     }
 }
diff --git a/src/main/java/com/teragrep/lsh_01/RelpConversion.java b/src/main/java/com/teragrep/lsh_01/RelpConversion.java
@@ -26,9 +26,7 @@
 import com.teragrep.lsh_01.config.PayloadConfig;
 import com.teragrep.lsh_01.config.SecurityConfig;
 import com.teragrep.lsh_01.lookup.LookupTableFactory;
-import com.teragrep.lsh_01.pool.IRelpConnection;
-import com.teragrep.lsh_01.pool.ManagedRelpConnection;
-import com.teragrep.lsh_01.pool.RelpConnectionPool;
+import com.teragrep.lsh_01.pool.*;
 import com.teragrep.rlo_14.*;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -42,7 +40,7 @@
 public class RelpConversion implements IMessageHandler {
 
     private final static Logger LOGGER = LogManager.getLogger(RelpConversion.class);
-    private final RelpConnectionPool relpConnectionPool;
+    private final Pool<IManagedRelpConnection> relpConnectionPool;
     private final SecurityConfig securityConfig;
     private final BasicAuthentication basicAuthentication;
     private final LookupConfig lookupConfig;
@@ -51,7 +49,7 @@ public class RelpConversion implements IMessageHandler {
     private final StringLookupTable appnameLookup;
 
     public RelpConversion(
-            RelpConnectionPool relpConnectionPool,
+            Pool<IManagedRelpConnection> relpConnectionPool,
             SecurityConfig securityConfig,
             BasicAuthentication basicAuthentication,
             LookupConfig lookupConfig,
@@ -139,9 +137,8 @@ private void sendMessage(
                 .withSDElement(headerSDElement)
                 .withSDElement(sdElement);
 
-        IRelpConnection relpConnection = relpConnectionPool.take();
-        ManagedRelpConnection managedRelpConnection = new ManagedRelpConnection(relpConnection);
-        managedRelpConnection.ensureSent(syslogMessage.toRfc5424SyslogMessage().getBytes(StandardCharsets.UTF_8));
+        IManagedRelpConnection relpConnection = relpConnectionPool.get();
+        relpConnection.ensureSent(syslogMessage.toRfc5424SyslogMessage().getBytes(StandardCharsets.UTF_8));
         relpConnectionPool.offer(relpConnection);
     }
 }
diff --git a/src/main/java/com/teragrep/lsh_01/config/RelpConfig.java b/src/main/java/com/teragrep/lsh_01/config/RelpConfig.java
@@ -24,6 +24,8 @@ public class RelpConfig implements Validateable {
     public final String relpTarget;
     public final int relpPort;
     public final int relpReconnectInterval;
+    public final int rebindRequestAmount;
+    public final boolean rebindEnabled;
 
     public RelpConfig() {
         PropertiesReaderUtilityClass propertiesReader = new PropertiesReaderUtilityClass(
@@ -32,16 +34,21 @@ public RelpConfig() {
         relpTarget = propertiesReader.getStringProperty("relp.target");
         relpPort = propertiesReader.getIntProperty("relp.port");
         relpReconnectInterval = propertiesReader.getIntProperty("relp.reconnectInterval");
+        rebindRequestAmount = propertiesReader.getIntProperty("relp.rebindRequestAmount");
+        rebindEnabled = propertiesReader.getBooleanProperty("relp.rebindEnabled");
     }
 
     @Override
     public void validate() {
-
+        if (rebindEnabled && rebindRequestAmount < 1) {
+            throw new IllegalArgumentException("relp.rebindRequestAmount has to be a positive number");
+        }
     }
 
     @Override
     public String toString() {
         return "RelpConfig{" + "relpTarget='" + relpTarget + '\'' + ", relpPort=" + relpPort
-                + ", relpReconnectInterval=" + relpReconnectInterval + '}';
+                + ", relpReconnectInterval=" + relpReconnectInterval + ", rebindRequestAmount=" + rebindRequestAmount
+                + ", rebindEnabled=" + rebindEnabled + '}';
     }
 }
diff --git a/src/main/java/com/teragrep/lsh_01/pool/IManagedRelpConnection.java b/src/main/java/com/teragrep/lsh_01/pool/IManagedRelpConnection.java
@@ -0,0 +1,25 @@
+/*
+  logstash-http-input to syslog bridge
+  Copyright 2024 Suomen Kanuuna Oy
+
+  Derivative Work of Elasticsearch
+  Copyright 2012-2015 Elasticsearch
+
+  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 com.teragrep.lsh_01.pool;
+
+public interface IManagedRelpConnection extends Poolable {
+
+    void ensureSent(byte[] bytes);
+}
diff --git a/src/main/java/com/teragrep/lsh_01/pool/IRelpConnection.java b/src/main/java/com/teragrep/lsh_01/pool/IRelpConnection.java
@@ -57,7 +57,5 @@ public interface IRelpConnection {
 
     void commit(RelpBatch relpBatch) throws IOException, IllegalStateException, TimeoutException;
 
-    boolean isStub();
-
     RelpConfig relpConfig();
 }
diff --git a/src/main/java/com/teragrep/lsh_01/pool/ManagedRelpConnection.java b/src/main/java/com/teragrep/lsh_01/pool/ManagedRelpConnection.java
@@ -26,20 +26,22 @@
 import java.io.IOException;
 import java.util.concurrent.TimeoutException;
 
-public class ManagedRelpConnection {
+public class ManagedRelpConnection implements IManagedRelpConnection {
 
     private static final Logger LOGGER = LogManager.getLogger(ManagedRelpConnection.class);
     private final IRelpConnection relpConnection;
+    private boolean hasConnected;
 
     public ManagedRelpConnection(IRelpConnection relpConnection) {
         this.relpConnection = relpConnection;
+        this.hasConnected = false;
     }
 
-    void connect() {
-        boolean notConnected = true;
-        while (notConnected) {
-            boolean connected = false;
+    private void connect() {
+        boolean connected = false;
+        while (!connected) {
             try {
+                this.hasConnected = true;
                 connected = relpConnection
                         .connect(relpConnection.relpConfig().relpTarget, relpConnection.relpConfig().relpPort);
             }
@@ -51,37 +53,27 @@ void connect() {
                                 e.getMessage()
                         );
             }
-            if (connected) {
-                notConnected = false;
+
+            try {
+                Thread.sleep(relpConnection.relpConfig().relpReconnectInterval);
             }
-            else {
-                try {
-                    Thread.sleep(relpConnection.relpConfig().relpReconnectInterval);
-                }
-                catch (InterruptedException e) {
-                    LOGGER.error("Reconnect timer interrupted, reconnecting now");
-                }
+            catch (InterruptedException e) {
+                LOGGER.error("Reconnect timer interrupted, reconnecting now");
             }
         }
     }
 
     private void tearDown() {
-        relpConnection.tearDown();
-    }
-
-    private void disconnect() {
-        boolean disconnected = false;
-        try {
-            disconnected = relpConnection.disconnect();
-        }
-        catch (IllegalStateException | IOException | TimeoutException e) {
-            LOGGER.error("Forcefully closing connection due to exception <{}>", e.getMessage());
-        }
-        finally {
-            this.tearDown();
+        /*
+         TODO remove: wouldn't need a check hasConnected but there is a bug in RLP-01 tearDown()
+         see https://github.com/teragrep/rlp_01/issues/63 for further info
+         */
+        if (hasConnected) {
+            relpConnection.tearDown();
         }
     }
 
+    @Override
     public void ensureSent(byte[] bytes) {
         final RelpBatch relpBatch = new RelpBatch();
         relpBatch.insert(bytes);
@@ -103,4 +95,22 @@ public void ensureSent(byte[] bytes) {
             }
         }
     }
+
+    @Override
+    public boolean isStub() {
+        return false;
+    }
+
+    @Override
+    public void close() {
+        try {
+            this.relpConnection.disconnect();
+        }
+        catch (IllegalStateException | IOException | TimeoutException e) {
+            LOGGER.error("Forcefully closing connection due to exception <{}>", e.getMessage());
+        }
+        finally {
+            tearDown();
+        }
+    }
 }
diff --git a/src/main/java/com/teragrep/lsh_01/pool/ManagedRelpConnectionStub.java b/src/main/java/com/teragrep/lsh_01/pool/ManagedRelpConnectionStub.java
@@ -0,0 +1,38 @@
+/*
+  logstash-http-input to syslog bridge
+  Copyright 2024 Suomen Kanuuna Oy
+
+  Derivative Work of Elasticsearch
+  Copyright 2012-2015 Elasticsearch
+
+  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 com.teragrep.lsh_01.pool;
+
+public class ManagedRelpConnectionStub implements IManagedRelpConnection {
+
+    @Override
+    public void ensureSent(byte[] bytes) {
+        throw new IllegalStateException("ManagedRelpConnectionStub does not support this");
+    }
+
+    @Override
+    public boolean isStub() {
+        return true;
+    }
+
+    @Override
+    public void close() {
+        throw new IllegalStateException("ManagedRelpConnectionStub does not support this");
+    }
+}