Add send_keepalive_packet option to out_forward plugin

This adds the send_keepalive_packet configuration parameter to enable
TCP keepalive (SO_KEEPALIVE) on forward output connections, similar to
the implementation in in_forward (PR #2352).

The existing 'keepalive' parameter only enables connection pooling,
not TCP-level keepalive probes.

Signed-off-by: Ian Driver <idriver@williamhill.us>

Author: Ian Driver

lib/fluent/plugin/out_forward.rb

@@ -119,6 +119,8 @@ class ForwardOutput < Output
     config_param :keepalive, :bool, default: false
     desc "Expired time of keepalive. Default value is nil, which means to keep connection as long as possible"
     config_param :keepalive_timeout, :time, default: nil
+    desc 'Check the remote connection is still available by sending a keepalive packet if this value is true.'
+    config_param :send_keepalive_packet, :bool, default: false
 
     config_section :security, required: false, multi: false do
       desc 'The hostname'
@@ -270,6 +272,10 @@ def configure(conf)
         log.warn('The value of keepalive_timeout is ignored. if you want to use keepalive, please add `keepalive true` to your conf.')
       end
 
+      if @send_keepalive_packet && !@keepalive
+        raise Fluent::ConfigError, "'send_keepalive_packet' is enabled but 'keepalive' is not. Enable 'keepalive' to use TCP keepalive."
+      end
+
       raise Fluent::ConfigError, "ack_response_timeout must be a positive integer" if @ack_response_timeout < 1
 
       if @compress == :zstd
@@ -414,6 +420,7 @@ def create_transfer_socket(host, port, hostname, &block)
           send_timeout: @send_timeout,
           recv_timeout: @ack_response_timeout,
           connect_timeout: @connect_timeout,
+          send_keepalive_packet: @send_keepalive_packet,
           &block
         )
       when :tcp
@@ -423,6 +430,7 @@ def create_transfer_socket(host, port, hostname, &block)
           send_timeout: @send_timeout,
           recv_timeout: @ack_response_timeout,
           connect_timeout: @connect_timeout,
+          send_keepalive_packet: @send_keepalive_packet,
           &block
         )
       else

test/plugin/test_out_forward.rb

@@ -153,6 +153,27 @@ def try_write(chunk)
     end
   end
 
+  test 'send_keepalive_packet is disabled by default' do
+    @d = d = create_driver(config)
+    assert_false d.instance.send_keepalive_packet
+  end
+
+  test 'send_keepalive_packet can be enabled' do
+    @d = d = create_driver(config + %[
+      keepalive true
+      send_keepalive_packet true
+    ])
+    assert_true d.instance.send_keepalive_packet
+  end
+
+  test 'send_keepalive_packet without keepalive raises error' do
+    assert_raise(Fluent::ConfigError.new("'send_keepalive_packet' is enabled but 'keepalive' is not. Enable 'keepalive' to use TCP keepalive.")) do
+      create_driver(config + %[
+        send_keepalive_packet true
+      ])
+    end
+  end
+
   test 'configure with ignore_network_errors_at_startup' do
     normal_conf = config_element('match', '**', {}, [
         config_element('server', '', {'name' => 'test', 'host' => 'unexisting.yaaaaaaaaaaaaaay.host.example.com'})
@@ -1303,7 +1324,8 @@ def plugin_id_for_test?
                                              linger_timeout: anything,
                                              send_timeout: anything,
                                              recv_timeout: anything,
-                                             connect_timeout: anything
+                                             connect_timeout: anything,
+                                             send_keepalive_packet: anything
                                             ) { |sock| mock(sock).close.once; sock }.twice
 
     target_input_driver.run(timeout: 15) do
@@ -1343,7 +1365,8 @@ def plugin_id_for_test?
                                                linger_timeout: anything,
                                                send_timeout: anything,
                                                recv_timeout: anything,
-                                               connect_timeout: anything
+                                               connect_timeout: anything,
+                                               send_keepalive_packet: anything
                                               ) { |sock| mock(sock).close.once; sock }.once
 
       target_input_driver.run(timeout: 15) do
@@ -1380,7 +1403,8 @@ def plugin_id_for_test?
                                                  linger_timeout: anything,
                                                  send_timeout: anything,
                                                  recv_timeout: anything,
-                                                 connect_timeout: anything) { |sock|
+                                                 connect_timeout: anything,
+                                                 send_keepalive_packet: anything) { |sock|
           mock(sock).close.once; sock
         }.twice