New configuration option to allow socket to use TCP keep-alive

.gitignore

@@ -8,3 +8,6 @@ build
 *.iml
 *.iws
 *.ipr
+out
+.rakeTasks
+

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 5.0.9
+  - New configuration option to set TCP keep-alive [#16](https://github.com/logstash-plugins/logstash-input-tcp/pull/116)
+
 ## 5.0.8
   - Reorder shut down of the two event loops to prevent RejectedExecutionException
 

docs/index.asciidoc

@@ -88,6 +88,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-ssl_key>> |a valid filesystem path|No
 | <<plugins-{type}s-{plugin}-ssl_key_passphrase>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-ssl_verify>> |<<boolean,boolean>>|No
+| <<plugins-{type}s-{plugin}-tcp_keep_alive>> |<<boolean,boolean>>|No
 |=======================================================================
 
 Also see <<plugins-{type}s-{plugin}-common-options>> for a list of options supported by all
@@ -182,6 +183,13 @@ SSL key passphrase
 Verify the identity of the other end of the SSL connection against the CA.
 For input, sets the field `sslsubject` to that of the client certificate.
 
+[id="plugins-{type}s-{plugin}-tcp_keep_alive"]
+===== `tcp_keep_alive`
+
+  * Value type is <<boolean,boolean>>
+  * Default value is `false`
+
+Instruct the socket to use TCP keep alives. Uses OS defaults for keep alive settings.
 
 
 [id="plugins-{type}s-{plugin}-common-options"]

lib/logstash/inputs/tcp.rb

@@ -105,6 +105,9 @@ class LogStash::Inputs::Tcp < LogStash::Inputs::Base
   # Useful when the CA chain is not necessary in the system store.
   config :ssl_extra_chain_certs, :validate => :array, :default => []
 
+  # Instruct the socket to use TCP keep alives. Uses OS defaults for keep alive settings.
+  config :tcp_keep_alive, :validate => :boolean, :default => false
+
   HOST_FIELD = "host".freeze
   HOST_IP_FIELD = "[@metadata][ip_address]".freeze
   PORT_FIELD = "port".freeze
@@ -140,7 +143,7 @@ def register
       if @ssl_enable
         self.server_socket = new_server_socket
       else
-        @loop = InputLoop.new(@host, @port, DecoderImpl.new(@codec, self))
+        @loop = InputLoop.new(@host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive)
       end
     end
   end
@@ -331,19 +334,26 @@ def load_cert_store
   def new_server_socket
     begin
       socket = TCPServer.new(@host, @port)
+      socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, @tcp_keep_alive)
     rescue Errno::EADDRINUSE
       @logger.error("Could not start TCP server: Address in use", :host => @host, :port => @port)
       raise
     end
 
-    @ssl_enable ? OpenSSL::SSL::SSLServer.new(socket, ssl_context) : socket
+    if @ssl_enable
+      socket = OpenSSL::SSL::SSLServer.new(socket, ssl_context)
+      socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, @tcp_keep_alive)
+    end
+    socket
   end
 
   def new_client_socket
     socket = TCPSocket.new(@host, @port)
+    socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, @tcp_keep_alive)
 
     if @ssl_enable
       socket = OpenSSL::SSL::SSLSocket.new(socket, ssl_context)
+      socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, @tcp_keep_alive)
       socket.connect
     end
 

src/main/java/org/logstash/tcp/InputLoop.java

@@ -41,13 +41,15 @@ public final class InputLoop implements Runnable, Closeable {
      * @param host Host to bind the listen to
      * @param port Port to listen on
      * @param decoder {@link Decoder} provided by Jruby
+     * @param keepAlive set to true to instruct the socket to issue TCP keep alive
      */
-    public InputLoop(final String host, final int port, final Decoder decoder) {
+    public InputLoop(final String host, final int port, final Decoder decoder, final boolean keepAlive) {
         worker = new NioEventLoopGroup();
         boss = new NioEventLoopGroup(1);
         future = new ServerBootstrap().group(boss, worker)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
+            .childOption(ChannelOption.SO_KEEPALIVE, keepAlive)
             .childHandler(new InputLoop.InputHandler(decoder)).bind(host, port);
     }
 

version

@@ -1 +1 @@
-5.0.8
+5.0.9