From d0fc620b1a11da8caec7608e4af9c83c44844c0a Mon Sep 17 00:00:00 2001 From: Sami Korhonen Date: Fri, 2 Sep 2016 16:20:33 +0300 Subject: [PATCH] Fixed issue #84: Added support for TCP-Keepalive --- src/main/java/org/graylog2/GelfTCPSender.java | 26 ++++++++++++++----- .../java/org/graylog2/log/GelfAppender.java | 15 ++++++++--- .../org/graylog2/logging/GelfHandler.java | 4 ++- .../org/graylog2/log/GelfAppenderTest.java | 6 ++--- 4 files changed, 38 insertions(+), 13 deletions(-) diff --git a/src/main/java/org/graylog2/GelfTCPSender.java b/src/main/java/org/graylog2/GelfTCPSender.java index d064f80..39afd18 100644 --- a/src/main/java/org/graylog2/GelfTCPSender.java +++ b/src/main/java/org/graylog2/GelfTCPSender.java @@ -8,19 +8,31 @@ public class GelfTCPSender implements GelfSender { private boolean shutdown = false; private InetAddress host; private int port; + private boolean keepalive; private Socket socket; private OutputStream os; public GelfTCPSender() { } - public GelfTCPSender(String host, int port) throws IOException { + public GelfTCPSender(String host, int port, boolean keepalive) throws IOException { this.host = InetAddress.getByName(host); this.port = port; - this.socket = new Socket(host, port); + this.keepalive = keepalive; + initConnection(); + } + + private void initConnection() throws IOException { + this.socket = createSocket(); this.os = socket.getOutputStream(); } + private Socket createSocket() throws UnknownHostException, IOException { + Socket socket = new Socket(host, port); + socket.setKeepAlive(keepalive); + return socket; + } + public GelfSenderResult sendMessage(GelfMessage message) { if (shutdown || !message.isValid()) { return GelfSenderResult.MESSAGE_NOT_VALID_OR_SHUTTING_DOWN; @@ -28,11 +40,9 @@ public GelfSenderResult sendMessage(GelfMessage message) { try { // reconnect if necessary - if (socket == null || os == null) { - socket = new Socket(host, port); - os = socket.getOutputStream(); + if (!isConnectionInitialized()) { + initConnection(); } - os.write(message.toTCPBuffer().array()); return GelfSenderResult.OK; @@ -43,6 +53,10 @@ public GelfSenderResult sendMessage(GelfMessage message) { } } + private boolean isConnectionInitialized() { + return socket != null && os != null; + } + public void close() { shutdown = true; try { diff --git a/src/main/java/org/graylog2/log/GelfAppender.java b/src/main/java/org/graylog2/log/GelfAppender.java index 0fcd8b9..fa5036b 100644 --- a/src/main/java/org/graylog2/log/GelfAppender.java +++ b/src/main/java/org/graylog2/log/GelfAppender.java @@ -36,6 +36,7 @@ public class GelfAppender extends AppenderSkeleton implements GelfMessageProvide private int amqpMaxRetries = 0; private static String originHost; private int graylogPort = 12201; + private boolean tcpKeepalive; private String facility; private GelfSender gelfSender; private boolean extractStacktrace; @@ -59,6 +60,14 @@ public int getGraylogPort() { public void setGraylogPort(int graylogPort) { this.graylogPort = graylogPort; } + + public boolean isTcpKeepalive() { + return tcpKeepalive; + } + + public void setTcpKeepalive(boolean tcpKeepalive) { + this.tcpKeepalive = tcpKeepalive; + } public String getGraylogHost() { return graylogHost; @@ -177,7 +186,7 @@ public void activateOptions() { try { if (graylogHost != null && graylogHost.startsWith("tcp:")) { String tcpGraylogHost = graylogHost.substring(4); - gelfSender = getGelfTCPSender(tcpGraylogHost, graylogPort); + gelfSender = getGelfTCPSender(tcpGraylogHost, graylogPort, tcpKeepalive); } else if (graylogHost != null && graylogHost.startsWith("udp:")) { String udpGraylogHost = graylogHost.substring(4); gelfSender = getGelfUDPSender(udpGraylogHost, graylogPort); @@ -206,8 +215,8 @@ protected GelfUDPSender getGelfUDPSender(String udpGraylogHost, int graylogPort) return new GelfUDPSender(udpGraylogHost, graylogPort); } - protected GelfTCPSender getGelfTCPSender(String tcpGraylogHost, int graylogPort) throws IOException { - return new GelfTCPSender(tcpGraylogHost, graylogPort); + protected GelfTCPSender getGelfTCPSender(String tcpGraylogHost, int graylogPort, boolean tcpKeepalive) throws IOException { + return new GelfTCPSender(tcpGraylogHost, graylogPort, tcpKeepalive); } protected GelfAMQPSender getGelfAMQPSender(String amqpURI, String amqpExchangeName, String amqpRoutingKey, int amqpMaxRetries) throws IOException, URISyntaxException, NoSuchAlgorithmException, KeyManagementException { diff --git a/src/main/java/org/graylog2/logging/GelfHandler.java b/src/main/java/org/graylog2/logging/GelfHandler.java index a28dc07..4793d49 100644 --- a/src/main/java/org/graylog2/logging/GelfHandler.java +++ b/src/main/java/org/graylog2/logging/GelfHandler.java @@ -27,6 +27,7 @@ public class GelfHandler private String amqpRoutingKey; private int amqpMaxRetries; private String originHost; + private boolean tcpKeepalive; private int graylogPort; private String facility; private GelfSender gelfSender; @@ -40,6 +41,7 @@ public GelfHandler() { graylogHost = manager.getProperty(prefix + ".graylogHost"); final String port = manager.getProperty(prefix + ".graylogPort"); graylogPort = null == port ? 12201 : Integer.parseInt(port); + tcpKeepalive = "true".equalsIgnoreCase(manager.getProperty(prefix + ".tcpKeepalive")); originHost = manager.getProperty(prefix + ".originHost"); extractStacktrace = "true".equalsIgnoreCase(manager.getProperty(prefix + ".extractStacktrace")); int fieldNumber = 0; @@ -128,7 +130,7 @@ public synchronized void publish(final LogRecord record) { try { if (graylogHost.startsWith("tcp:")) { String tcpGraylogHost = graylogHost.substring(4, graylogHost.length()); - gelfSender = new GelfTCPSender(tcpGraylogHost, graylogPort); + gelfSender = new GelfTCPSender(tcpGraylogHost, graylogPort, tcpKeepalive); } else if (graylogHost.startsWith("udp:")) { String udpGraylogHost = graylogHost.substring(4, graylogHost.length()); gelfSender = new GelfUDPSender(udpGraylogHost, graylogPort); diff --git a/src/test/java/org/graylog2/log/GelfAppenderTest.java b/src/test/java/org/graylog2/log/GelfAppenderTest.java index 527a681..62c30f8 100644 --- a/src/test/java/org/graylog2/log/GelfAppenderTest.java +++ b/src/test/java/org/graylog2/log/GelfAppenderTest.java @@ -175,8 +175,8 @@ protected GelfUDPSender getGelfUDPSender(String udpGraylogHost, int port) throws } @Override - protected GelfTCPSender getGelfTCPSender(String tcpGraylogHost, int port) throws IOException { - return new MockGelfTCPSender(tcpGraylogHost, port); + protected GelfTCPSender getGelfTCPSender(String tcpGraylogHost, int port, boolean tcpKeepalive) throws IOException { + return new MockGelfTCPSender(tcpGraylogHost, port, tcpKeepalive); } }; @@ -264,7 +264,7 @@ private MockGelfUDPSender(String host, int port) throws IOException { private class MockGelfTCPSender extends GelfTCPSender { - private MockGelfTCPSender(String host, int port) throws IOException { + private MockGelfTCPSender(String host, int port, boolean keepalive) throws IOException { if (host.contains("tcp:")) { throw new UnknownHostException("tcp: found in host"); }