plugin_helper/server: Add receive_buffer_size parameter in transport …

…section

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

lib/fluent/plugin_helper/server.rb

@@ -84,6 +84,8 @@ def server_create_connection(title, port, proto: nil, bind: '0.0.0.0', shared: t
           socket_options[:linger_timeout] ||= @transport_config&.linger_timeout || 0
         end
 
+        socket_options[:receive_buffer_size] = @transport_config&.receive_buffer_size
+
         socket_option_validate!(proto, **socket_options)
         socket_option_setter = ->(sock){ socket_option_set(sock, **socket_options) }
 
@@ -136,6 +138,8 @@ def server_create(title, port, proto: nil, bind: '0.0.0.0', shared: true, socket
           socket_options[:linger_timeout] ||= @transport_config&.linger_timeout || 0
         end
 
+        socket_options[:receive_buffer_size] = @transport_config&.receive_buffer_size
+
         unless socket
           socket_option_validate!(proto, **socket_options)
           socket_option_setter = ->(sock){ socket_option_set(sock, **socket_options) }
@@ -266,6 +270,9 @@ module ServerTransportParams
 
           ### Socket Params ###
 
+          desc "The max size of socket receive buffer. SO_RCVBUF"
+          config_param :receive_buffer_size, :size, default: nil
+
           # SO_LINGER 0 to send RST rather than FIN to avoid lots of connections sitting in TIME_WAIT at src.
           # Set positive value if needing to send FIN on closing on non-Windows.
           # (On Windows, Fluentd can send FIN with zero `linger_timeout` since Fluentd doesn't set 0 to SO_LINGER on Windows.

test/plugin_helper/test_server.rb

@@ -89,6 +89,23 @@ class Dummy < Fluent::Plugin::TestBase
       assert d.log
       assert_equal 1, d.transport_config.linger_timeout
     end
+
+    test 'can change receive_buffer_size option' do
+      d = Dummy.new
+
+      transport_opts = {
+        'receive_buffer_size' => 1024,
+      }
+      transport_conf = config_element('transport', 'tcp', transport_opts)
+      conf = config_element('source', 'tag.*', {}, [transport_conf])
+
+      assert_nothing_raised do
+        d.configure(conf)
+      end
+      assert d.plugin_id
+      assert d.log
+      assert_equal 1024, d.transport_config.receive_buffer_size
+    end
   end
 
   # run tests for tcp, udp, tls and unix