in_syslog: add receive_buffer_size parameter in transport

Signed-off-by: Shizuo Fujita <[email protected]>

lib/fluent/plugin/in_syslog.rb

@@ -114,6 +114,8 @@ class SyslogInput < Input
     # overwrite server plugin to change default to :udp
     config_section :transport, required: false, multi: false, init: true, param_name: :transport_config do
       config_argument :protocol, :enum, list: [:tcp, :udp, :tls], default: :udp
+      desc "The max size of socket receive buffer. SO_RCVBUF"
+      config_param :receive_buffer_size, :size, default: nil
     end
 
     def configure(conf)
@@ -170,7 +172,7 @@ def start
     end
 
     def start_udp_server
-      server_create_udp(:in_syslog_udp_server, @port, bind: @bind, max_bytes: @message_length_limit, resolve_name: @resolve_hostname) do |data, sock|
+      server_create_udp(:in_syslog_udp_server, @port, bind: @bind, max_bytes: @message_length_limit, resolve_name: @resolve_hostname, receive_buffer_size: @transport_config.receive_buffer_size) do |data, sock|
         message_handler(data.chomp, sock)
       end
     end

test/plugin/test_in_syslog.rb

@@ -92,6 +92,37 @@ def test_respect_protocol_type_than_transport
     compare_test_result(d.events, tests)
   end
 
+  def test_transport_receive_buffer_size
+    # check default SO_RCVBUF size
+    d = create_driver([ipv4_config, "<transport udp>\n </transport>"].join("\n"))
+    d.run do
+      sock = d.instance.instance_variable_get(:@_servers)[0].server.instance_variable_get(:@sock)
+      begin
+        default_sock = UDPSocket.new
+        assert_equal(default_sock.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).int,
+                     sock.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).int)
+      ensure
+        default_sock.close
+      end
+    end
+
+    # check receive_buffer_size
+    d = create_driver([ipv4_config, "<transport udp>\n receive_buffer_size 2048\n </transport>"].join("\n"))
+    d.run do
+      sock = d.instance.instance_variable_get(:@_servers)[0].server.instance_variable_get(:@sock)
+      begin
+        default_sock = UDPSocket.new
+        assert_not_equal(default_sock.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).int,
+                         sock.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).int)
+
+        default_sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 2048)
+        assert_equal(default_sock.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).int,
+                     sock.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).int)
+      ensure
+        default_sock.close
+      end
+    end
+  end
 
   data(
     ipv4: ['127.0.0.1', :ipv4, ::Socket::AF_INET],