Merge pull request #3644 from daipom/enable-server-plugins-to-specify…

…-so_linger

Enable server plugins to specify socket-option `SO_LINGER`

lib/fluent/plugin/in_forward.rb

@@ -40,7 +40,7 @@ class ForwardInput < Input
     config_param :backlog, :integer, default: nil
     # SO_LINGER 0 to send RST rather than FIN to avoid lots of connections sitting in TIME_WAIT at src
     desc 'The timeout time used to set linger option.'
-    config_param :linger_timeout, :integer, default: 0
+    config_param :linger_timeout, :integer, default: nil, deprecated: "use transport directive"
     # This option is for Cool.io's loop wait timeout to avoid loop stuck at shutdown. Almost users don't need to change this value.
     config_param :blocking_timeout, :time, default: 0.5
     desc 'Try to resolve hostname from IP addresses or not.'

lib/fluent/plugin_helper/server.rb

@@ -80,8 +80,8 @@ def server_create_connection(title, port, proto: nil, bind: '0.0.0.0', shared: t
         raise ArgumentError, "BUG: block not specified which handles connection" unless block_given?
         raise ArgumentError, "BUG: block must have just one argument" unless block.arity == 1
 
-        if proto == :tcp || proto == :tls # default linger_timeout only for server
-          socket_options[:linger_timeout] ||= 0
+        if proto == :tcp || proto == :tls
+          socket_options[:linger_timeout] ||= @transport_config&.linger_timeout || 0
         end
 
         socket_option_validate!(proto, **socket_options)
@@ -132,8 +132,8 @@ def server_create(title, port, proto: nil, bind: '0.0.0.0', shared: true, socket
         raise ArgumentError, "BUG: block not specified which handles received data" unless block_given?
         raise ArgumentError, "BUG: block must have 1 or 2 arguments" unless callback.arity == 1 || callback.arity == 2
 
-        if proto == :tcp || proto == :tls # default linger_timeout only for server
-          socket_options[:linger_timeout] ||= 0
+        if proto == :tcp || proto == :tls
+          socket_options[:linger_timeout] ||= @transport_config&.linger_timeout || 0
         end
 
         unless socket
@@ -263,6 +263,23 @@ module ServerTransportParams
         include Fluent::Configurable
         config_section :transport, required: false, multi: false, init: true, param_name: :transport_config do
           config_argument :protocol, :enum, list: [:tcp, :tls], default: :tcp
+
+          ### Socket Params ###
+
+          # 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.
+          # NOTE:
+            # Socket-options can be specified from each plugin as needed, so most of them is not defined here for now.
+            # This is because there is no positive reason to do so.
+            # `linger_timeout` option in particular needs to be defined here
+            # although it can be specified from each plugin as well.
+            # This is because this helper fixes the default value to `0` for its own reason
+            # and it has a critical effect on the behavior.
+          desc 'The timeout time used to set linger option.'
+          config_param :linger_timeout, :integer, default: 0
+
+          ### TLS Params ###
+
           config_param :version, :enum, list: Fluent::TLS::SUPPORTED_VERSIONS, default: Fluent::TLS::DEFAULT_VERSION
           config_param :min_version, :enum, list: Fluent::TLS::SUPPORTED_VERSIONS, default: nil
           config_param :max_version, :enum, list: Fluent::TLS::SUPPORTED_VERSIONS, default: nil

test/plugin/test_in_forward.rb

@@ -68,7 +68,6 @@ def create_driver(conf=base_config)
       @d = d = create_driver
       assert_equal @port, d.instance.port
       assert_equal '127.0.0.1', d.instance.bind
-      assert_equal 0, d.instance.linger_timeout
       assert_equal 0.5, d.instance.blocking_timeout
       assert !d.instance.backlog
     end
@@ -77,7 +76,6 @@ def create_driver(conf=base_config)
       @d = d = create_driver(config_auth)
       assert_equal @port, d.instance.port
       assert_equal '127.0.0.1', d.instance.bind
-      assert_equal 0, d.instance.linger_timeout
       assert !d.instance.backlog
 
       assert d.instance.security

test/plugin_helper/test_server.rb

@@ -64,6 +64,24 @@ class Dummy < Fluent::Plugin::TestBase
       end
       assert d.plugin_id
       assert d.log
+      assert_equal 0, d.transport_config.linger_timeout
+    end
+
+    test 'can change linger_timeout option' do
+      d = Dummy.new
+
+      transport_opts = {
+        'linger_timeout' => 1,
+      }
+      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 1, d.transport_config.linger_timeout
     end
   end