diff --git a/lib/fluent/plugin/in_udp.rb b/lib/fluent/plugin/in_udp.rb index 172c102b9f..295870285f 100644 --- a/lib/fluent/plugin/in_udp.rb +++ b/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| diff --git a/lib/fluent/plugin_helper/socket_option.rb b/lib/fluent/plugin_helper/socket_option.rb index 77fd9e3b66..7eb82b6a9c 100644 --- a/lib/fluent/plugin_helper/socket_option.rb +++ b/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 diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb index b596d7fa0e..b2e65d741b 100755 --- a/test/plugin/test_in_udp.rb +++ b/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