Add recv_buffer parameter for in_udp

lib/fluent/plugin/in_udp.rb

@@ -40,6 +40,8 @@ class UdpInput < Input
     config_param :message_length_limit, :size, default: 4096
     desc "Remove newline from the end of incoming payload"
     config_param :remove_newline, :bool, default: true
+    desc "The max size of socket receive buffer. SO_RCVBUF"
+    config_param :receive_buffer_size, :size, default: nil
 
     config_param :blocking_timeout, :time, default: 0.5
 
@@ -61,7 +63,7 @@ def start
       super
 
       log.info "listening udp socket", bind: @bind, port: @port
-      server_create(:in_udp_server, @port, proto: :udp, bind: @bind, max_bytes: @message_length_limit) do |data, sock|
+      server_create(:in_udp_server, @port, proto: :udp, bind: @bind, max_bytes: @message_length_limit, receive_buffer_size: @receive_buffer_size) do |data, sock|
         data.chomp! if @remove_newline
         begin
           @parser.parse(data) do |time, record|

lib/fluent/plugin_helper/socket_option.rb

@@ -24,7 +24,7 @@ module SocketOption
       FORMAT_STRUCT_LINGER  = 'I!I!' # { int l_onoff; int l_linger; }
       FORMAT_STRUCT_TIMEVAL = 'L!L!' # { time_t tv_sec; suseconds_t tv_usec; }
 
-      def socket_option_validate!(protocol, resolve_name: nil, linger_timeout: nil, recv_timeout: nil, send_timeout: nil)
+      def socket_option_validate!(protocol, resolve_name: nil, linger_timeout: nil, recv_timeout: nil, send_timeout: nil, receive_buffer_size: nil)
         unless resolve_name.nil?
           if protocol != :tcp && protocol != :udp && protocol != :tls
             raise ArgumentError, "BUG: resolve_name in available for tcp/udp/tls"
@@ -37,7 +37,7 @@ def socket_option_validate!(protocol, resolve_name: nil, linger_timeout: nil, re
         end
       end
 
-      def socket_option_set(sock, resolve_name: nil, nonblock: false, linger_timeout: nil, recv_timeout: nil, send_timeout: nil)
+      def socket_option_set(sock, resolve_name: nil, nonblock: false, linger_timeout: nil, recv_timeout: nil, send_timeout: nil, receive_buffer_size: nil)
         unless resolve_name.nil?
           sock.do_not_reverse_lookup = !resolve_name
         end
@@ -56,6 +56,9 @@ def socket_option_set(sock, resolve_name: nil, nonblock: false, linger_timeout:
           optval = [send_timeout.to_i, 0].pack(FORMAT_STRUCT_TIMEVAL)
           socket_option_set_one(sock, :SO_SNDTIMEO, optval)
         end
+        if receive_buffer_size
+          socket_option_set_one(sock, :SO_RCVBUF, receive_buffer_size.to_i)
+        end
         sock
       end
 

test/plugin/test_in_udp.rb

@@ -55,6 +55,7 @@ def create_udp_socket(host, port)
     assert_equal PORT, d.instance.port
     assert_equal bind, d.instance.bind
     assert_equal 4096, d.instance.message_length_limit
+    assert_equal nil, d.instance.receive_buffer_size
   end
 
   data(
@@ -172,4 +173,37 @@ def create_udp_socket(host, port)
       assert_equal expected_record, d.events[i][2]
     end
   end
+
+  test 'receive_buffer_size' do
+    # doesn't check exact value because it depends on platform and condition
+
+    # check if default socket and in_udp's one without receive_buffer_size have same size buffer
+    d0 = create_driver(BASE_CONFIG + %!
+      format none
+    !)
+    d0.run do
+      sock = d0.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 if default socket and in_udp's one with receive_buffer_size have different size buffer
+    d1 = create_driver(BASE_CONFIG + %!
+      format none
+      receive_buffer_size 1001
+    !)
+    d1.run do
+      sock = d1.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)
+      ensure
+        default_sock.close
+      end
+    end
+  end
 end