Merge pull request #3474 from fluent/in_syslog_keepalive

in_syslog: Add send_keepalive_packet option as same as in_forward

lib/fluent/plugin/in_syslog.rb

@@ -89,6 +89,8 @@ class SyslogInput < Input
     config_param :source_hostname_key, :string, default: nil
     desc 'Try to resolve hostname from IP addresses or not.'
     config_param :resolve_hostname, :bool, default: nil
+    desc 'Check the remote connection is still available by sending a keepalive packet if this value is true.'
+    config_param :send_keepalive_packet, :bool, default: false
     desc 'The field name of source address of sender.'
     config_param :source_address_key, :string, default: nil
     desc 'The field name of the severity.'
@@ -143,6 +145,11 @@ def configure(conf)
       end
 
       @_event_loop_run_timeout = @blocking_timeout
+
+      protocol = @protocol_type || @transport_config.protocol
+      if @send_keepalive_packet && protocol == :udp
+        raise Fluent::ConfigError, "send_keepalive_packet is available for tcp/tls"
+      end
     end
 
     def multi_workers_ready?
@@ -173,7 +180,12 @@ def start_tcp_server(tls: false)
 
       delimiter = octet_count_frame ? " " : @delimiter
       delimiter_size = delimiter.size
-      server_create_connection(tls ? :in_syslog_tls_server : :in_syslog_tcp_server, @port, bind: @bind, resolve_name: @resolve_hostname) do |conn|
+      server_create_connection(
+        tls ? :in_syslog_tls_server : :in_syslog_tcp_server, @port,
+        bind: @bind,
+        resolve_name: @resolve_hostname,
+        send_keepalive_packet: @send_keepalive_packet
+      ) do |conn|
         conn.data do |data|
           buffer = conn.buffer
           buffer << data

test/plugin/test_in_syslog.rb

@@ -467,4 +467,39 @@ def test_emit_unmatched_lines_with_address
     assert_equal tests.size, d.events.size
     compare_unmatched_lines_test_result(d.events, tests, {address: address})
   end
+
+  def test_send_keepalive_packet_is_disabled_by_default
+    d = create_driver(ipv4_config + %[
+      <transport tcp>
+      </transport>
+      protocol tcp
+    ])
+    assert_false d.instance.send_keepalive_packet
+  end
+
+  def test_send_keepalive_packet_can_be_enabled
+    addr = "127.0.0.1"
+    d = create_driver(ipv4_config + %[
+      <transport tcp>
+      </transport>
+      send_keepalive_packet true
+    ])
+    assert_true d.instance.send_keepalive_packet
+    mock.proxy(d.instance).server_create_connection(
+      :in_syslog_tcp_server, @port,
+      bind: addr,
+      resolve_name: nil,
+      send_keepalive_packet: true)
+    d.run do
+      TCPSocket.open(addr, @port)
+    end
+  end
+
+  def test_send_keepalive_packet_can_not_be_enabled_for_udp
+    assert_raise(Fluent::ConfigError) do
+      d = create_driver(ipv4_config + %[
+        send_keepalive_packet true
+      ])
+    end
+  end
 end