From 55023341f63650ca0cc4f85a1c636be5d49149fd Mon Sep 17 00:00:00 2001 From: JSap0914 <116227558+JSap0914@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:02:01 +0900 Subject: [PATCH] out_forward: drop keepalive sockets with failed or mismatched acks (#5436) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Which issue(s) this PR fixes**: Fixes #4057 **What this PR does / why we need it**: In Forwarder-Aggregator mode with `keepalive true` and `flush_thread_count` > 1, when the aggregator restarts/reconnects the forwarder starts logging ``` [warn]: ack in response and chunk id in sent data are different chunk_id="..." ack="...\n" ``` endlessly, and never recovers on its own until the pod is killed. Root cause: when a chunk's ack times out (`FAILED`) or comes back with a chunk id that does not match the one tracked for its socket (`CHUNKID_UNMATCHED`), `ack_check` returned the keepalive socket to the reuse pool via `ConnectionManager#close` (→ `SocketCache#checkin`). Such a connection may still carry an in-flight/stale ack for a previous chunk. When the socket is checked out again for a new chunk, that stale ack is read first and compared against the new chunk's id, so it never matches. From then on every ack on that connection is offset by one chunk, producing an unbounded stream of mismatch warnings. This PR adds `ConnectionManager#discard`, which **revokes** a keepalive socket (or closes a non-keepalive one) instead of checking it back in, and uses it from `ack_check` on the `FAILED` and `CHUNKID_UNMATCHED` paths. The tainted connection is torn down, so the next chunk uses a fresh connection with a clean ack stream and the warning storm stops. Successful acks keep reusing the socket exactly as before. **Docs Changes**: None. **Release Note**: out_forward: stop the endless "ack in response and chunk id in sent data are different" warning storm by discarding (instead of reusing) a keepalive socket whose ack failed or came back with a mismatched chunk id. --- Tests (`ruby 3.4.10`): - `test/plugin/out_forward/test_connection_manager.rb` — new `#discard` cases: closes a non-keepalive socket, revokes a keepalive socket instead of checking it in, and confirms a discarded keepalive socket is not reused (a checked-in one is). - `test/plugin/test_out_forward.rb` — new `ack_check` cases: the socket is discarded on `CHUNKID_UNMATCHED` and `FAILED`, and reused on `SUCCESS`. ``` $ bundle exec ruby -Itest test/plugin/out_forward/test_connection_manager.rb 10 tests, 16 assertions, 0 failures, 0 errors $ bundle exec ruby -Itest test/plugin/test_out_forward.rb 74 tests, 147 assertions, 0 failures, 0 errors, 3 omissions (Windows-only) ``` The new tests fail on `master` (the mismatched/failed ack socket is reused) and pass with this change. Signed-off-by: JSup Signed-off-by: github-actions[bot] --- lib/fluent/plugin/out_forward.rb | 15 ++++- .../plugin/out_forward/connection_manager.rb | 18 +++++ .../out_forward/test_connection_manager.rb | 62 +++++++++++++++++ test/plugin/test_out_forward.rb | 66 +++++++++++++++++++ 4 files changed, 159 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/out_forward.rb b/lib/fluent/plugin/out_forward.rb index fe9c253ff0..81468c3979 100644 --- a/lib/fluent/plugin/out_forward.rb +++ b/lib/fluent/plugin/out_forward.rb @@ -525,17 +525,28 @@ def ack_reader def ack_check(select_interval) @ack_handler.collect_response(select_interval) do |chunk_id, node, sock, result| - @connection_manager.close(sock) - case result when AckHandler::Result::SUCCESS + @connection_manager.close(sock) commit_write(chunk_id) when AckHandler::Result::FAILED + # The ack timed out or the peer closed the connection. A keepalive + # socket may still have an in-flight ack for this chunk, so drop it + # instead of returning it to the reuse pool; otherwise a late ack + # could later be read against a chunk sent on the reused socket (#4057). + @connection_manager.discard(sock) node&.disable! rollback_write(chunk_id, update_retry: false) if chunk_id when AckHandler::Result::CHUNKID_UNMATCHED + # The ack read from this socket belongs to a different chunk than the + # one tracked for it: the keepalive connection's ack stream is out of + # sync (e.g. a stale ack left over from before an aggregator restart). + # Discard the socket so the following acks are not each offset by one, + # which would otherwise flood the log with mismatch warnings (#4057). + @connection_manager.discard(sock) rollback_write(chunk_id, update_retry: false) else + @connection_manager.close(sock) log.warn("BUG: invalid status #{result} #{chunk_id}") if chunk_id diff --git a/lib/fluent/plugin/out_forward/connection_manager.rb b/lib/fluent/plugin/out_forward/connection_manager.rb index 3b8ad32595..75cfe539f1 100644 --- a/lib/fluent/plugin/out_forward/connection_manager.rb +++ b/lib/fluent/plugin/out_forward/connection_manager.rb @@ -78,6 +78,24 @@ def close(sock) end end + # Drop a socket instead of returning it to the keepalive reuse pool. + # + # Used when a chunk's ack timed out or came back with a mismatched chunk + # id: such a connection may still carry an in-flight/stale ack for a + # previous chunk. Reusing it would let that stale ack be read against a + # chunk sent on the reused socket, permanently offsetting every following + # ack and flooding the log with 'ack in response and chunk id in sent data + # are different' warnings (see #4057). Revoking removes it from the pool so + # the next chunk uses a fresh connection with a clean ack stream. + def discard(sock) + if @socket_cache + @socket_cache.revoke(sock) + else + sock.close_write rescue nil + sock.close rescue nil + end + end + private def connect_keepalive(host:, port:, hostname:, ack: nil) diff --git a/test/plugin/out_forward/test_connection_manager.rb b/test/plugin/out_forward/test_connection_manager.rb index 5c528cc768..a4133d8746 100644 --- a/test/plugin/out_forward/test_connection_manager.rb +++ b/test/plugin/out_forward/test_connection_manager.rb @@ -142,4 +142,66 @@ class ConnectionManager < Test::Unit::TestCase end end end + + sub_test_case '#discard' do + test 'closes the socket directly when socket_cache is nil' do + sock = 'sock' + cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new( + log: $log, + secure: false, + connection_factory: -> (_, _, _) { sock }, + socket_cache: nil, + ) + + mock(sock).close_write.once + mock(sock).close.once + cm.discard(sock) + end + + test 'revokes a keepalive socket instead of returning it to the reuse pool' do + cache = Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log) + mock(cache).revoke('sock').once + mock(cache).checkin('sock').never + + cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new( + log: $log, + secure: false, + connection_factory: -> (_, _, _) { 'sock' }, + socket_cache: cache, + ) + + cm.discard('sock') + end + + test 'a discarded keepalive socket is not reused, unlike a checked-in one' do + ack = Object.new + def ack.enqueue(_sock); end + + # Baseline: a checked-in socket is returned to the pool and reused. + reused = [] + reuse_cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new( + log: $log, + secure: false, + connection_factory: -> (_, _, _) { s = "sock#{reused.size}"; reused << s; s }, + socket_cache: Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log), + ) + reuse_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| } + reuse_cm.close('sock0') + reuse_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| } + assert_equal(['sock0'], reused) + + # A discarded socket is dropped, so the next checkout builds a fresh one. + discarded = [] + discard_cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new( + log: $log, + secure: false, + connection_factory: -> (_, _, _) { s = "sock#{discarded.size}"; discarded << s; s }, + socket_cache: Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log), + ) + discard_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| } + discard_cm.discard('sock0') + discard_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| } + assert_equal(['sock0', 'sock1'], discarded) + end + end end diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb index 0c44e1d74e..1f4821df4e 100644 --- a/test/plugin/test_out_forward.rb +++ b/test/plugin/test_out_forward.rb @@ -1429,4 +1429,70 @@ def plugin_id_for_test? logs = d.logs assert{ logs.any?{|log| log.include?('handshake timeout after 1.0s') } } end + + sub_test_case 'ack_check discards tainted keepalive sockets (#4057)' do + def create_ack_driver + create_driver(config + %[ + require_ack_response true + keepalive true + ]) + end + + # A keepalive socket whose ack failed or came back mismatched may still + # carry an in-flight/stale ack for a previous chunk. Returning it to the + # reuse pool lets that stale ack be read against a chunk sent on the reused + # socket, permanently offsetting every following ack and flooding the log + # with 'ack in response and chunk id in sent data are different' warnings. + data( + 'mismatched ack' => Fluent::Plugin::ForwardOutput::AckHandler::Result::CHUNKID_UNMATCHED, + 'failed ack' => Fluent::Plugin::ForwardOutput::AckHandler::Result::FAILED, + ) + test 'discards the socket instead of reusing it' do |result| + @d = d = create_ack_driver + instance = d.instance + node = instance.nodes.first + sock = 'sock' + + # One-shot: only the explicit ack_check below yields; the collect_response + # that after_shutdown triggers during teardown stays silent. + fake_ack_handler = Object.new + fake_ack_handler.instance_variable_set(:@yielded, false) + fake_ack_handler.define_singleton_method(:collect_response) do |_interval, &block| + next if @yielded + @yielded = true + block.call('chunk-id', node, sock, result) + end + instance.instance_variable_set(:@ack_handler, fake_ack_handler) + stub(instance).rollback_write + + cm = instance.instance_variable_get(:@connection_manager) + mock(cm).discard(sock).once + mock(cm).close(sock).never + + instance.__send__(:ack_check, 0) + end + + test 'reuses the socket on a successful ack' do + @d = d = create_ack_driver + instance = d.instance + node = instance.nodes.first + sock = 'sock' + + fake_ack_handler = Object.new + fake_ack_handler.instance_variable_set(:@yielded, false) + fake_ack_handler.define_singleton_method(:collect_response) do |_interval, &block| + next if @yielded + @yielded = true + block.call('chunk-id', node, sock, Fluent::Plugin::ForwardOutput::AckHandler::Result::SUCCESS) + end + instance.instance_variable_set(:@ack_handler, fake_ack_handler) + stub(instance).commit_write + + cm = instance.instance_variable_get(:@connection_manager) + mock(cm).close(sock).once + mock(cm).discard(sock).never + + instance.__send__(:ack_check, 0) + end + end end