Fix #2969: Add timeout to establish_connection to prevent infinite loop

- Add timeout check in establish_connection method using send_timeout
- Prevents infinite loop when connection drops during handshake protocol
- Disables problematic nodes and logs timeout warnings
- Add test case to verify timeout functionality works correctly

Fixes issue where logs stop being flushed when handshake gets stuck
in unstable network environments with proxy components.

Signed-off-by: Ian Driver <[email protected]>

Author: Ian Driver

lib/fluent/plugin/out_forward.rb

@@ -620,7 +620,17 @@ def verify_connection
       end
 
       def establish_connection(sock, ri)
+        start_time = Fluent::Clock.now
+        timeout = @sender.send_timeout
+        
         while ri.state != :established
+          # Check for timeout to prevent infinite loop
+          if Fluent::Clock.now - start_time > timeout
+            @log.warn "handshake timeout after #{timeout}s", host: @host, port: @port
+            disable!
+            break
+          end
+          
           begin
             # TODO: On Ruby 2.2 or earlier, read_nonblock doesn't work expectedly.
             # We need rewrite around here using new socket/server plugin helper.

test/plugin/test_out_forward.rb

@@ -1406,4 +1406,27 @@ def plugin_id_for_test?
     assert_equal 0, @d.instance.healthy_nodes_count
     assert_equal 0, @d.instance.registered_nodes_count
   end
+
+  test 'establish_connection_timeout' do
+    @d = d = create_driver(%[
+      send_timeout 1
+      <server>
+        host #{TARGET_HOST}
+        port #{@target_port}
+      </server>
+    ])
+
+    node = d.instance.nodes.first
+    mock_sock = flexmock('socket')
+    mock_sock.should_receive(:read_nonblock).with(anything).and_return('').at_least.once
+
+    ri = Fluent::Plugin::ForwardOutput::ConnectionManager::RequestInfo.new(:helo)
+
+    assert_true node.available?
+    node.establish_connection(mock_sock, ri)
+    assert_false node.available?
+
+    logs = d.logs
+    assert{ logs.any?{|log| log.include?('handshake timeout after 1s') } }
+  end
 end