From c52210aedc0c48b2aea295e89fa1776ded1b4c79 Mon Sep 17 00:00:00 2001 From: mrudrego Date: Wed, 18 Oct 2023 11:51:55 +0530 Subject: [PATCH 01/39] Fix for rotate_age where Fluentd passes as Symbol (#4311) Fix for rotate_age where Fluentd passes as Symbol while Ruby Logger expects String --------- Signed-off-by: mrudrego Signed-off-by: mrudrego --- lib/fluent/system_config.rb | 2 +- test/config/test_system_config.rb | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/fluent/system_config.rb b/lib/fluent/system_config.rb index 5fb781d72b..4919294989 100644 --- a/lib/fluent/system_config.rb +++ b/lib/fluent/system_config.rb @@ -62,7 +62,7 @@ class SystemConfig config_param :time_format, :string, default: '%Y-%m-%d %H:%M:%S %z' config_param :rotate_age, default: nil do |v| if Fluent::Log::LOG_ROTATE_AGE.include?(v) - v.to_sym + v else begin Integer(v) diff --git a/test/config/test_system_config.rb b/test/config/test_system_config.rb index 502fb7d75a..a45fde5f15 100644 --- a/test/config/test_system_config.rb +++ b/test/config/test_system_config.rb @@ -151,7 +151,7 @@ def parse_text(text) data('daily' => "daily", 'weekly' => 'weekly', 'monthly' => 'monthly') - test "symbols for rotate_age" do |age| + test "strings for rotate_age" do |age| conf = parse_text(<<-EOS) @@ -160,7 +160,7 @@ def parse_text(text) EOS sc = Fluent::SystemConfig.new(conf) - assert_equal(age.to_sym, sc.log.rotate_age) + assert_equal(age, sc.log.rotate_age) end test "numeric number for rotate age" do From e37785bb7b4a96123c024dcee807c72a3da264b4 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Fri, 27 Oct 2023 19:17:52 +0900 Subject: [PATCH 02/39] in_tail: Fix a stall bug on !follow_inode case (#4327) Fix #3614 Although known stall issues of in_tail on `follow_inode` case are fixed in v1.16.2, it has still a similar problem on `!follow_inode` case. In this case, a tail watcher is possible to mark the position entry as `unwatched` if it's tansitioned to `rotate_wait` state by `refresh_watcher` even if another newer tail watcher is managing it. It's hard to occur in usual because `stat_watcher` will be called immediately after the file is changed while `refresh_wather` is called every 60 seconds by default. However, there is a rare possibility that this order might be swapped especillay if in_tail is busy on processing large amount of logs. Because in_tail is single threadied, event queues such as timers or inotify will be stucked in this case. There is no such problem on `follow_inode` case because position entries are always marked as `unwatched` before entering `rotate_wait` state. --------- Signed-off-by: Takuro Ashie Co-authored-by: Daijiro Fukuda --- lib/fluent/plugin/in_tail.rb | 2 +- test/plugin/test_in_tail.rb | 105 +++++++++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 1 deletion(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 54fd900039..7ef6377aab 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -564,7 +564,7 @@ def detach_watcher(tw, ino, close_io = true) tw.close if close_io - if tw.unwatched && @pf + if @pf && tw.unwatched && (@follow_inode || !@tails[tw.path]) target_info = TargetInfo.new(tw.path, ino) @pf.unwatch(target_info) end diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 1546bf9d4d..8919866683 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -3017,4 +3017,109 @@ def test_path_resurrection ) end end + + sub_test_case "Update watchers for rotation without follow_inodes" do + # The scenario where in_tail wrongly unwatches the PositionEntry. + # This is reported in https://github.com/fluent/fluentd/issues/3614. + def test_refreshTW_during_rotation + config = config_element( + "ROOT", + "", + { + "path" => "#{@tmp_dir}/tail.txt0", + "pos_file" => "#{@tmp_dir}/tail.pos", + "tag" => "t1", + "format" => "none", + "read_from_head" => "true", + # In order to detach the old watcher quickly. + "rotate_wait" => "3s", + # In order to reproduce the same condition stably, ensure that `refresh_watchers` is not + # called by a timer. + "refresh_interval" => "1h", + # stat_watcher often calls `TailWatcher::on_notify` faster than creating a new log file, + # so disable it in order to reproduce the same condition stably. + "enable_stat_watcher" => "false", + } + ) + d = create_driver(config, false) + + tail_watchers = [] + stub.proxy(d.instance).setup_watcher do |tw| + tail_watchers.append(tw) + tw + end + + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file1 log1"} + + d.run(expect_records: 6, timeout: 15) do + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file1 log2"} + FileUtils.move("#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt" + "1") + + # This reproduces the following situation: + # `refresh_watchers` is called during the rotation process and it detects the current file being lost. + # Then it stops and unwatches the TailWatcher. + d.instance.refresh_watchers + + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file2 log1"} + + # `watch_timer` calls `TailWatcher::on_notify`, and then `update_watcher` trys to add the new TailWatcher. + # After `rotate_wait` interval, the PositionEntry is unwatched. + # HOWEVER, the new TailWatcher is still using that PositionEntry, so this breaks the PositionFile!! + # That PositionEntry is removed from `PositionFile::map`, but it is still working and remaining in the real pos file. + sleep 5 + + # Append to the new current log file. + # The PositionEntry is updated although it does not exist in `PositionFile::map`. + # `PositionFile::map`: empty + # Real pos file: `.../tail.txt 0000000000000016 (inode)` + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file2 log2"} + + # Rotate again + [1, 0].each do |i| + FileUtils.move("#{@tmp_dir}/tail.txt#{i}", "#{@tmp_dir}/tail.txt#{i + 1}") + end + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file3 log1"} + + # `watch_timer` calls `TailWatcher::on_notify`, and then `update_watcher` trys to update the TailWatcher. + sleep 3 + + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"} + + # Wait `rotate_wait` for file2 to make sure to close all IO handlers + sleep 3 + end + + inode_0 = tail_watchers[0]&.ino + inode_1 = tail_watchers[1]&.ino + inode_2 = tail_watchers[2]&.ino + record_values = d.events.collect { |event| event[2]["message"] }.sort + position_entries = [] + Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f| + f.readlines(chomp: true).each do |line| + values = line.split("\t") + position_entries.append([values[0], values[1], values[2].to_i(16)]) + end + end + + assert_equal( + { + record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2", "file3 log1", "file3 log2"], + tail_watcher_paths: ["#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0"], + tail_watcher_inodes: [inode_0, inode_1, inode_2], + tail_watcher_io_handler_opened_statuses: [false, false, false], + position_entries: [ + # The recorded path is old, but it is no problem. The path is not used when using follow_inodes. + ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_2], + ], + }, + { + record_values: record_values, + tail_watcher_paths: tail_watchers.collect { |tw| tw.path }, + tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino }, + tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false }, + position_entries: position_entries + }, + ) + end + end end From 38f5650e624a4ab3c9a67df6ffdab43cc792b0b1 Mon Sep 17 00:00:00 2001 From: amdoolittle Date: Mon, 30 Oct 2023 21:40:42 -0700 Subject: [PATCH 03/39] Buffer: Fix NoMethodError with empty unstaged chunk arrays (#4303) Signed-off-by: Alex Doolittle --- lib/fluent/plugin/buffer.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index d04ae08296..3871b4f25b 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -417,7 +417,7 @@ def write(metadata_and_data, format: nil, size: nil, enqueue: false) if c.staged? && (enqueue || chunk_size_full?(c)) m = c.metadata enqueue_chunk(m) - if unstaged_chunks[m] + if unstaged_chunks[m] && !unstaged_chunks[m].empty? u = unstaged_chunks[m].pop u.synchronize do if u.unstaged? && !chunk_size_full?(u) From f8732a9e929380e47690b76f24e976cbd935713e Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Fri, 10 Nov 2023 15:41:03 +0900 Subject: [PATCH 04/39] in_tail: add warning for silent stop on !follow_inodes case (#4339) For `refresh_watcher`, if an exisiting TailWatcher already follows a target path with the different inode, it means that the TailWatcher following the rotated file still exists. In this case, `refresh_watcher` can't start the new TailWatcher for the new current file. So, we should output a warning log in order to prevent silent collection stops, such as #4327. The similar warning may work for follow_inodes too. Just limiting the case to suppress the impact to existing logic. Signed-off-by: Daijiro Fukuda --- lib/fluent/plugin/in_tail.rb | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 7ef6377aab..44761b4061 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -385,7 +385,7 @@ def refresh_watchers # So that inode can't be contained in `removed_hash`, and can't be unwatched by `stop_watchers`. # # This logic may work for `@follow_inodes false` too. - # Just limiting the case to supress the impact to existing logics. + # Just limiting the case to suppress the impact to existing logics. @pf&.unwatch_removed_targets(target_paths_hash) need_unwatch_in_stop_watchers = false end @@ -393,6 +393,28 @@ def refresh_watchers removed_hash = existence_paths_hash.reject {|key, value| target_paths_hash.key?(key)} added_hash = target_paths_hash.reject {|key, value| existence_paths_hash.key?(key)} + # If an exisiting TailWatcher already follows a target path with the different inode, + # it means that the TailWatcher following the rotated file still exists. In this case, + # `refresh_watcher` can't start the new TailWatcher for the new current file. So, we + # should output a warning log in order to prevent silent collection stops. + # (Such as https://github.com/fluent/fluentd/pull/4327) + # (Usually, such a TailWatcher should be removed from `@tails` in `update_watcher`.) + # (The similar warning may work for `@follow_inodes true` too. Just limiting the case + # to suppress the impact to existing logics.) + unless @follow_inodes + target_paths_hash.each do |path, target| + next unless @tails.key?(path) + # We can't use `existence_paths_hash[path].ino` because it is from `TailWatcher.ino`, + # which is very unstable parameter. (It can be `nil` or old). + # So, we need to use `TailWatcher.pe.read_inode`. + existing_watcher_inode = @tails[path].pe.read_inode + if existing_watcher_inode != target.ino + log.warn "Could not follow a file (inode: #{target.ino}) because an existing watcher for that filepath follows a different inode: #{existing_watcher_inode} (e.g. keeps watching a already rotated file). If you keep getting this message, please restart Fluentd.", + filepath: target.path + end + end + end + stop_watchers(removed_hash, unwatched: need_unwatch_in_stop_watchers) unless removed_hash.empty? start_watchers(added_hash) unless added_hash.empty? @startup = false if @startup From d3cf2e0f95a0ad88b9897197db6c5152310f114f Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Tue, 14 Nov 2023 11:29:45 +0900 Subject: [PATCH 05/39] v1.16.3 Signed-off-by: Daijiro Fukuda --- .github/workflows/linux-test.yaml | 4 ++-- .github/workflows/macos-test.yaml | 4 ++-- .github/workflows/windows-test.yaml | 4 ++-- CHANGELOG.md | 13 +++++++++++++ lib/fluent/version.rb | 2 +- 5 files changed, 20 insertions(+), 7 deletions(-) diff --git a/.github/workflows/linux-test.yaml b/.github/workflows/linux-test.yaml index de2757a508..553490db00 100644 --- a/.github/workflows/linux-test.yaml +++ b/.github/workflows/linux-test.yaml @@ -2,9 +2,9 @@ name: Testing on Ubuntu on: push: - branches: [master] + branches: [master, v1.16] pull_request: - branches: [master] + branches: [master, v1.16] jobs: test: diff --git a/.github/workflows/macos-test.yaml b/.github/workflows/macos-test.yaml index de54469ddd..80cdccecd1 100644 --- a/.github/workflows/macos-test.yaml +++ b/.github/workflows/macos-test.yaml @@ -2,9 +2,9 @@ name: Testing on macOS on: push: - branches: [master] + branches: [master, v1.16] pull_request: - branches: [master] + branches: [master, v1.16] jobs: test: diff --git a/.github/workflows/windows-test.yaml b/.github/workflows/windows-test.yaml index 10792577b5..4274f607b4 100644 --- a/.github/workflows/windows-test.yaml +++ b/.github/workflows/windows-test.yaml @@ -2,9 +2,9 @@ name: Testing on Windows on: push: - branches: [master] + branches: [master, v1.16] pull_request: - branches: [master] + branches: [master, v1.16] jobs: test: diff --git a/CHANGELOG.md b/CHANGELOG.md index ca435b2fd6..9e44d3b798 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # v1.16 +## Release v1.16.3 - 2023/11/14 + +### Bug Fix + +* in_tail: Fix a stall bug on !follow_inode case + https://github.com/fluent/fluentd/pull/4327 +* in_tail: add warning for silent stop on !follow_inodes case + https://github.com/fluent/fluentd/pull/4339 +* Buffer: Fix NoMethodError with empty unstaged chunk arrays + https://github.com/fluent/fluentd/pull/4303 +* Fix for rotate_age where Fluentd passes as Symbol + https://github.com/fluent/fluentd/pull/4311 + ## Release v1.16.2 - 2023/07/14 ### Bug Fix diff --git a/lib/fluent/version.rb b/lib/fluent/version.rb index 87bcc578d9..a84ee72e22 100644 --- a/lib/fluent/version.rb +++ b/lib/fluent/version.rb @@ -16,6 +16,6 @@ module Fluent - VERSION = '1.16.2' + VERSION = '1.16.3' end From e2d7885796a4b6f0e299be11c409d49ee2dd3b8e Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Thu, 14 Dec 2023 13:11:37 +0900 Subject: [PATCH 06/39] buffer: Avoid to process discarded chunks in write_step_by_step (for v1.16) (#4363) It fixes following error when many `chunk bytes limit exceeds` errors are occurred: ``` 2020-07-28 14:59:26 +0000 [warn]: #0 emit transaction failed: error_class=IOError error="closed stream" location="/fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer/file_chunk.rb:82:in `pos'" tag="cafiscode-eks-cluster.default" 2020-07-28 14:59:26 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer/file_chunk.rb:82:in `pos' 2020-07-28 14:59:26 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer/file_chunk.rb:82:in `rollback' 2020-07-28 14:59:26 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer.rb:339:in `rescue in block in write' 2020-07-28 14:59:26 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer.rb:332:in `block in write' 2020-07-28 14:59:26 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.11.1/lib/fluent/plugin/buffer.rb:331:in `each' ... ``` Fix #3089 Signed-off-by: Takuro Ashie --- lib/fluent/plugin/buffer.rb | 36 +++++++++++++------------- test/plugin/test_buffer.rb | 51 +++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 17 deletions(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index 3871b4f25b..cda7926fab 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -728,7 +728,6 @@ def write_once(metadata, data, format: nil, size: nil, &block) def write_step_by_step(metadata, data, format, splits_count, &block) splits = [] - errors = [] if splits_count > data.size splits_count = data.size end @@ -749,16 +748,14 @@ def write_step_by_step(metadata, data, format, splits_count, &block) modified_chunks = [] modified_metadata = metadata get_next_chunk = ->(){ - c = if staged_chunk_used - # Staging new chunk here is bad idea: - # Recovering whole state including newly staged chunks is much harder than current implementation. - modified_metadata = modified_metadata.dup_next - generate_chunk(modified_metadata) - else - synchronize { @stage[modified_metadata] ||= generate_chunk(modified_metadata).staged! } - end - modified_chunks << c - c + if staged_chunk_used + # Staging new chunk here is bad idea: + # Recovering whole state including newly staged chunks is much harder than current implementation. + modified_metadata = modified_metadata.dup_next + generate_chunk(modified_metadata) + else + synchronize { @stage[modified_metadata] ||= generate_chunk(modified_metadata).staged! } + end } writing_splits_index = 0 @@ -766,6 +763,8 @@ def write_step_by_step(metadata, data, format, splits_count, &block) while writing_splits_index < splits.size chunk = get_next_chunk.call + errors = [] + modified_chunks << {chunk: chunk, adding_bytesize: 0, errors: errors} chunk.synchronize do raise ShouldRetry unless chunk.writable? staged_chunk_used = true if chunk.staged? @@ -851,15 +850,18 @@ def write_step_by_step(metadata, data, format, splits_count, &block) raise end - block.call(chunk, chunk.bytesize - original_bytesize, errors) - errors = [] + modified_chunks.last[:adding_bytesize] = chunk.bytesize - original_bytesize end end + modified_chunks.each do |data| + block.call(data[:chunk], data[:adding_bytesize], data[:errors]) + end rescue ShouldRetry - modified_chunks.each do |mc| - mc.rollback rescue nil - if mc.unstaged? - mc.purge rescue nil + modified_chunks.each do |data| + chunk = data[:chunk] + chunk.rollback rescue nil + if chunk.unstaged? + chunk.purge rescue nil end end enqueue_chunk(metadata) if enqueue_chunk_before_retry diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index d35d2d6ce7..451358a5b1 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -850,6 +850,57 @@ def create_chunk_es(metadata, es) test '#compress returns :text' do assert_equal :text, @p.compress end + + # https://github.com/fluent/fluentd/issues/3089 + test "closed chunk should not be committed" do + assert_equal 8 * 1024 * 1024, @p.chunk_limit_size + assert_equal 0.95, @p.chunk_full_threshold + + purge_count = 0 + + stub.proxy(@p).generate_chunk(anything) do |chunk| + stub.proxy(chunk).purge do |result| + purge_count += 1 + result + end + stub.proxy(chunk).commit do |result| + assert_false(chunk.closed?) + result + end + stub.proxy(chunk).rollback do |result| + assert_false(chunk.closed?) + result + end + chunk + end + + m = @p.metadata(timekey: Time.parse('2016-04-11 16:40:00 +0000').to_i) + small_row = "x" * 1024 * 400 + big_row = "x" * 1024 * 1024 * 8 # just `chunk_size_limit`, it does't cause BufferOverFlowError. + + # Write 42 events in 1 event stream, last one is for triggering `ShouldRetry` + @p.write({m => [small_row] * 40 + [big_row] + ["x"]}) + + # Above event strem will be splitted twice by `Buffer#write_step_by_step` + # + # 1. `write_once`: 42 [events] * 1 [stream] + # 2. `write_step_by_step`: 4 [events]* 10 [streams] + 2 [events] * 1 [stream] + # 3. `write_step_by_step` (by `ShouldRetry`): 1 [event] * 42 [streams] + # + # The problematic data is built in the 2nd stage. + # In the 2nd stage, 5 streams are packed in a chunk. + # ((1024 * 400) [bytes] * 4 [events] * 5 [streams] = 8192000 [bytes] < `chunk_limit_size` (8MB)). + # So 3 chunks are used to store all data. + # The 1st chunk is already staged by `write_once`. + # The 2nd & 3rd chunks are newly created as unstaged. + # The 3rd chunk is purged before `ShouldRetry`, it's no problem: + # https://github.com/fluent/fluentd/blob/7e9eba736ff40ad985341be800ddc46558be75f2/lib/fluent/plugin/buffer.rb#L850 + # The 2nd chunk is purged in `rescue ShouldRetry`: + # https://github.com/fluent/fluentd/blob/7e9eba736ff40ad985341be800ddc46558be75f2/lib/fluent/plugin/buffer.rb#L862 + # It causes the issue described in https://github.com/fluent/fluentd/issues/3089#issuecomment-1811839198 + + assert_equal 2, purge_count + end end sub_test_case 'standard format with configuration for test with lower chunk limit size' do From b47181b629b6cfea2484ccdceb8f7abe1c8b6743 Mon Sep 17 00:00:00 2001 From: Christian Norbert Menges Date: Sun, 22 Oct 2023 15:10:39 +0200 Subject: [PATCH 07/39] buffer: Avoid calling dump_unique_id_hex if log level is not trace Signed-off-by: Christian Norbert Menges --- lib/fluent/plugin/buffer.rb | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index 3871b4f25b..fd2c889140 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -580,7 +580,7 @@ def takeback_chunk(chunk_id) chunk = @dequeued.delete(chunk_id) return false unless chunk # already purged by other thread @queue.unshift(chunk) - log.trace "chunk taken back", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: chunk.metadata + log.on_trace { log.trace "chunk taken back", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: chunk.metadata } @queued_num[chunk.metadata] += 1 # BUG if nil @dequeued_num[chunk.metadata] -= 1 end @@ -610,7 +610,7 @@ def purge_chunk(chunk_id) @queued_num.delete(metadata) @dequeued_num.delete(metadata) end - log.trace "chunk purged", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: metadata + log.on_trace { log.trace "chunk purged", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: metadata } end nil From 4372698b2e1c870a528156984d73e3c71bafc414 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Fri, 8 Mar 2024 18:53:35 +0900 Subject: [PATCH 08/39] in_tail: Manage tail watchers that are `rorate_wait` state too (#4334) After a tail watcher transitions to `rotate_wait` state, the `rotate_wait` timer is no longer managed by in_tail, it might cause unexpected behaviour. e.g.) * It's never unwatched when shutdown occurs before `rotate_wait` passed. * Needless `rotate_wait` timers are executed when it detects more rotations. This patch fixes such unexpected behaviour. Note: The comment about `detach_watcher` was added in 76f246ae6a5a543c2b302b1a1f61a4223be177eb. At that time, closing was done by event-loop. Now, the situation is completely different, so it should be removed. --------- Signed-off-by: Takuro Ashie Co-authored-by: Daijiro Fukuda --- lib/fluent/plugin/in_tail.rb | 24 +++-- test/plugin/test_in_tail.rb | 169 ++++++++++++++++++++++++++++++++++- 2 files changed, 184 insertions(+), 9 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 44761b4061..7e0289f65b 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -52,6 +52,7 @@ def initialize super @paths = [] @tails = {} + @tails_rotate_wait = {} @pf_file = nil @pf = nil @ignore_list = [] @@ -267,6 +268,9 @@ def shutdown @shutdown_start_time = Fluent::Clock.now # during shutdown phase, don't close io. It should be done in close after all threads are stopped. See close. stop_watchers(existence_path, immediate: true, remove_watcher: false) + @tails_rotate_wait.keys.each do |tw| + detach_watcher(tw, @tails_rotate_wait[tw][:ino], false) + end @pf_file.close if @pf_file super @@ -275,6 +279,7 @@ def shutdown def close super # close file handles after all threads stopped (in #close of thread plugin helper) + # It may be because we need to wait IOHanlder.ready_to_shutdown() close_watcher_handles end @@ -516,6 +521,9 @@ def close_watcher_handles tw.close end end + @tails_rotate_wait.keys.each do |tw| + tw.close + end end # refresh_watchers calls @tails.keys so we don't use stop_watcher -> start_watcher sequence for safety. @@ -570,10 +578,6 @@ def update_watcher(tail_watcher, pe, new_inode) detach_watcher_after_rotate_wait(tail_watcher, pe.read_inode) end - # TailWatcher#close is called by another thread at shutdown phase. - # It causes 'can't modify string; temporarily locked' error in IOHandler - # so adding close_io argument to avoid this problem. - # At shutdown, IOHandler's io will be released automatically after detached the event loop def detach_watcher(tw, ino, close_io = true) if @follow_inodes && tw.ino != ino log.warn("detach_watcher could be detaching an unexpected tail_watcher with a different ino.", @@ -604,7 +608,11 @@ def detach_watcher_after_rotate_wait(tw, ino) if @open_on_every_update # Detach now because it's already closed, waiting it doesn't make sense. detach_watcher(tw, ino) - elsif throttling_is_enabled?(tw) + end + + return if @tails_rotate_wait[tw] + + if throttling_is_enabled?(tw) # When the throttling feature is enabled, it might not reach EOF yet. # Should ensure to read all contents before closing it, with keeping throttling. start_time_to_wait = Fluent::Clock.now @@ -612,14 +620,18 @@ def detach_watcher_after_rotate_wait(tw, ino) elapsed = Fluent::Clock.now - start_time_to_wait if tw.eof? && elapsed >= @rotate_wait timer.detach + @tails_rotate_wait.delete(tw) detach_watcher(tw, ino) end end + @tails_rotate_wait[tw] = { ino: ino, timer: timer } else # when the throttling feature isn't enabled, just wait @rotate_wait - timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do + timer = timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do + @tails_rotate_wait.delete(tw) detach_watcher(tw, ino) end + @tails_rotate_wait[tw] = { ino: ino, timer: timer } end end diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 8919866683..58006c0b98 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -3016,6 +3016,92 @@ def test_path_resurrection }, ) end + + def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait + config = config_element( + "ROOT", + "", + { + "path" => "#{@tmp_dir}/tail.txt*", + "pos_file" => "#{@tmp_dir}/tail.pos", + "tag" => "t1", + "format" => "none", + "read_from_head" => "true", + "follow_inodes" => "true", + "rotate_wait" => "3s", + "refresh_interval" => "1h", + # stat_watcher often calls `TailWatcher::on_notify` faster than creating a new log file, + # so disable it in order to reproduce the same condition stably. + "enable_stat_watcher" => "false", + } + ) + d = create_driver(config, false) + + tail_watchers = [] + stub.proxy(d.instance).setup_watcher do |tw| + tail_watchers.append(tw) + mock.proxy(tw).close.once # Note: Currently, there is no harm in duplicate calls. + tw + end + + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file1 log1"} + + d.run(expect_records: 6, timeout: 15) do + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file1 log2"} + + sleep 1.5 # Need to be larger than 1s (the interval of watch_timer) + + FileUtils.move("#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt" + "1") + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file2 log1"} + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file2 log2"} + + sleep 1.5 # Need to be larger than 1s (the interval of watch_timer) + + # Rotate again (Old TailWatcher waiting rotate_wait also calls update_watcher) + [1, 0].each do |i| + FileUtils.move("#{@tmp_dir}/tail.txt#{i}", "#{@tmp_dir}/tail.txt#{i + 1}") + end + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file3 log1"} + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"} + + # Wait rotate_wait to confirm that TailWatcher.close is not called in duplicate. + # (Note: Currently, there is no harm in duplicate calls) + sleep 4 + end + + inode_0 = tail_watchers[0]&.ino + inode_1 = tail_watchers[1]&.ino + inode_2 = tail_watchers[2]&.ino + record_values = d.events.collect { |event| event[2]["message"] }.sort + position_entries = [] + Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f| + f.readlines(chomp: true).each do |line| + values = line.split("\t") + position_entries.append([values[0], values[1], values[2].to_i(16)]) + end + end + + assert_equal( + { + record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2", "file3 log1", "file3 log2"], + tail_watcher_paths: ["#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0"], + tail_watcher_inodes: [inode_0, inode_1, inode_2], + tail_watcher_io_handler_opened_statuses: [false, false, false], + position_entries: [ + ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_0], + ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_1], + ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_2], + ], + }, + { + record_values: record_values, + tail_watcher_paths: tail_watchers.collect { |tw| tw.path }, + tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino }, + tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false }, + position_entries: position_entries + }, + ) + end end sub_test_case "Update watchers for rotation without follow_inodes" do @@ -3084,9 +3170,6 @@ def test_refreshTW_during_rotation sleep 3 Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"} - - # Wait `rotate_wait` for file2 to make sure to close all IO handlers - sleep 3 end inode_0 = tail_watchers[0]&.ino @@ -3121,5 +3204,85 @@ def test_refreshTW_during_rotation }, ) end + + def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait + config = config_element( + "ROOT", + "", + { + "path" => "#{@tmp_dir}/tail.txt0", + "pos_file" => "#{@tmp_dir}/tail.pos", + "tag" => "t1", + "format" => "none", + "read_from_head" => "true", + "rotate_wait" => "3s", + "refresh_interval" => "1h", + } + ) + d = create_driver(config, false) + + tail_watchers = [] + stub.proxy(d.instance).setup_watcher do |tw| + tail_watchers.append(tw) + mock.proxy(tw).close.once # Note: Currently, there is no harm in duplicate calls. + tw + end + + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file1 log1"} + + d.run(expect_records: 6, timeout: 15) do + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file1 log2"} + + sleep 1.5 # Need to be larger than 1s (the interval of watch_timer) + + FileUtils.move("#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt" + "1") + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file2 log1"} + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file2 log2"} + + sleep 1.5 # Need to be larger than 1s (the interval of watch_timer) + + # Rotate again (Old TailWatcher waiting rotate_wait also calls update_watcher) + [1, 0].each do |i| + FileUtils.move("#{@tmp_dir}/tail.txt#{i}", "#{@tmp_dir}/tail.txt#{i + 1}") + end + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file3 log1"} + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"} + + # Wait rotate_wait to confirm that TailWatcher.close is not called in duplicate. + # (Note: Currently, there is no harm in duplicate calls) + sleep 4 + end + + inode_0 = tail_watchers[0]&.ino + inode_1 = tail_watchers[1]&.ino + inode_2 = tail_watchers[2]&.ino + record_values = d.events.collect { |event| event[2]["message"] }.sort + position_entries = [] + Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f| + f.readlines(chomp: true).each do |line| + values = line.split("\t") + position_entries.append([values[0], values[1], values[2].to_i(16)]) + end + end + + assert_equal( + { + record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2", "file3 log1", "file3 log2"], + tail_watcher_paths: ["#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0"], + tail_watcher_inodes: [inode_0, inode_1, inode_2], + tail_watcher_io_handler_opened_statuses: [false, false, false], + position_entries: [ + ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_2], + ], + }, + { + record_values: record_values, + tail_watcher_paths: tail_watchers.collect { |tw| tw.path }, + tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino }, + tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false }, + position_entries: position_entries + }, + ) + end end end From 7b7116cf463f5a07f5d403e0c807fdf651cc8bb4 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Mon, 15 Jan 2024 17:29:09 +0900 Subject: [PATCH 09/39] Refine CI (#4380) Refine CI settings: * Unify CI settings for each platforms to one file * Don't run CI for Ruby head on each push/pull_request event * Because Ruby head is continually changed and not released yet, running CI one shot isn't so meaningful. * Instead run it regularly on every Sunday. * Add TESTOPT=-v to all platfroms * It's useful to investigate when a test is stalled. * Remove a hack for Ruby 3.0 on Windows * It seems no longer needed. Signed-off-by: Takuro Ashie --- .github/workflows/macos-test.yaml | 34 ------------- .github/workflows/test-ruby-head.yaml | 31 ++++++++++++ .../workflows/{linux-test.yaml => test.yaml} | 22 ++++----- .github/workflows/windows-test.yaml | 49 ------------------- 4 files changed, 40 insertions(+), 96 deletions(-) delete mode 100644 .github/workflows/macos-test.yaml create mode 100644 .github/workflows/test-ruby-head.yaml rename .github/workflows/{linux-test.yaml => test.yaml} (50%) delete mode 100644 .github/workflows/windows-test.yaml diff --git a/.github/workflows/macos-test.yaml b/.github/workflows/macos-test.yaml deleted file mode 100644 index 80cdccecd1..0000000000 --- a/.github/workflows/macos-test.yaml +++ /dev/null @@ -1,34 +0,0 @@ -name: Testing on macOS - -on: - push: - branches: [master, v1.16] - pull_request: - branches: [master, v1.16] - -jobs: - test: - runs-on: ${{ matrix.os }} - continue-on-error: ${{ matrix.experimental }} - strategy: - fail-fast: false - matrix: - ruby-version: ['3.2', '3.1', '3.0', '2.7'] - os: [macos-latest] - experimental: [true] - include: - - ruby-version: head - os: macos-latest - experimental: true - - name: Unit testing with Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }} - steps: - - uses: actions/checkout@v3 - - name: Set up Ruby - uses: ruby/setup-ruby@v1 - with: - ruby-version: ${{ matrix.ruby-version }} - - name: Install dependencies - run: bundle install - - name: Run tests - run: bundle exec rake test diff --git a/.github/workflows/test-ruby-head.yaml b/.github/workflows/test-ruby-head.yaml new file mode 100644 index 0000000000..6aaec99e93 --- /dev/null +++ b/.github/workflows/test-ruby-head.yaml @@ -0,0 +1,31 @@ +name: Test + +on: + schedule: + - cron: '11 14 * * 0' + workflow_dispatch: + +jobs: + test: + runs-on: ${{ matrix.os }} + continue-on-error: true + strategy: + fail-fast: false + matrix: + os: ['ubuntu-latest', 'macos-latest', 'windows-latest'] + ruby-version: ['head'] + + name: Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }} + steps: + - uses: actions/checkout@v3 + - name: Set up Ruby + uses: ruby/setup-ruby@v1 + with: + ruby-version: ${{ matrix.ruby-version }} + - name: Install addons + if: ${{ matrix.os == 'ubuntu-latest' }} + run: sudo apt-get install libgmp3-dev libcap-ng-dev + - name: Install dependencies + run: bundle install + - name: Run tests + run: bundle exec rake test TESTOPTS=-v diff --git a/.github/workflows/linux-test.yaml b/.github/workflows/test.yaml similarity index 50% rename from .github/workflows/linux-test.yaml rename to .github/workflows/test.yaml index 553490db00..9332491372 100644 --- a/.github/workflows/linux-test.yaml +++ b/.github/workflows/test.yaml @@ -1,27 +1,22 @@ -name: Testing on Ubuntu +name: Test on: push: - branches: [master, v1.16] + branches: [v1.16] pull_request: - branches: [master, v1.16] + branches: [v1.16] jobs: test: runs-on: ${{ matrix.os }} - continue-on-error: ${{ matrix.experimental }} + continue-on-error: false strategy: fail-fast: false matrix: - ruby-version: ['3.2', '3.1', '3.0', '2.7'] - os: [ubuntu-latest] - experimental: [false] - include: - - ruby-version: head - os: ubuntu-latest - experimental: true + os: ['ubuntu-latest', 'macos-latest', 'windows-latest'] + ruby-version: ['3.3', '3.2', '3.1', '3.0', '2.7'] - name: Unit testing with Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }} + name: Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }} steps: - uses: actions/checkout@v3 - name: Set up Ruby @@ -29,8 +24,9 @@ jobs: with: ruby-version: ${{ matrix.ruby-version }} - name: Install addons + if: ${{ matrix.os == 'ubuntu-latest' }} run: sudo apt-get install libgmp3-dev libcap-ng-dev - name: Install dependencies run: bundle install - name: Run tests - run: bundle exec rake test + run: bundle exec rake test TESTOPTS=-v diff --git a/.github/workflows/windows-test.yaml b/.github/workflows/windows-test.yaml deleted file mode 100644 index 4274f607b4..0000000000 --- a/.github/workflows/windows-test.yaml +++ /dev/null @@ -1,49 +0,0 @@ -name: Testing on Windows - -on: - push: - branches: [master, v1.16] - pull_request: - branches: [master, v1.16] - -jobs: - test: - runs-on: ${{ matrix.os }} - continue-on-error: ${{ matrix.experimental }} - strategy: - fail-fast: false - matrix: - ruby-version: ['3.2', '3.1', '2.7'] - os: - - windows-latest - experimental: [false] - include: - - ruby-version: head - os: windows-latest - experimental: true - - ruby-version: '3.0.3' - os: windows-latest - experimental: false - # On Ruby 3.0, we need to use fiddle 1.0.8 or later to retrieve correct - # error code. In addition, we have to specify the path of fiddle by RUBYLIB - # because RubyInstaller loads Ruby's bundled fiddle before initializing gem. - # See also: - # * https://github.com/ruby/fiddle/issues/72 - # * https://bugs.ruby-lang.org/issues/17813 - # * https://github.com/oneclick/rubyinstaller2/blob/8225034c22152d8195bc0aabc42a956c79d6c712/lib/ruby_installer/build/dll_directory.rb - ruby-lib-opt: RUBYLIB=%RUNNER_TOOL_CACHE%/Ruby/3.0.3/x64/lib/ruby/gems/3.0.0/gems/fiddle-1.1.0/lib - - name: Unit testing with Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }} - steps: - - uses: actions/checkout@v3 - - name: Set up Ruby - uses: ruby/setup-ruby@v1 - with: - ruby-version: ${{ matrix.ruby-version }} - - name: Add Fiddle 1.1.0 - if: ${{ matrix.ruby-version == '3.0.3' }} - run: gem install fiddle --version 1.1.0 - - name: Install dependencies - run: ridk exec bundle install - - name: Run tests - run: bundle exec rake test TESTOPTS=-v ${{ matrix.ruby-lib-opt }} From a834192c63571db40f80c6d67439d59c3ba63dbd Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Mon, 11 Mar 2024 11:59:56 +0900 Subject: [PATCH 10/39] CI: Remove scheduled tests for Ruby head because Ruby head is never supported on v1.16, and it seems wrong to use schedule trigger outside of the default branch (master). Signed-off-by: Daijiro Fukuda Co-authored-by: Takuro Ashie --- .github/workflows/test-ruby-head.yaml | 31 --------------------------- 1 file changed, 31 deletions(-) delete mode 100644 .github/workflows/test-ruby-head.yaml diff --git a/.github/workflows/test-ruby-head.yaml b/.github/workflows/test-ruby-head.yaml deleted file mode 100644 index 6aaec99e93..0000000000 --- a/.github/workflows/test-ruby-head.yaml +++ /dev/null @@ -1,31 +0,0 @@ -name: Test - -on: - schedule: - - cron: '11 14 * * 0' - workflow_dispatch: - -jobs: - test: - runs-on: ${{ matrix.os }} - continue-on-error: true - strategy: - fail-fast: false - matrix: - os: ['ubuntu-latest', 'macos-latest', 'windows-latest'] - ruby-version: ['head'] - - name: Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }} - steps: - - uses: actions/checkout@v3 - - name: Set up Ruby - uses: ruby/setup-ruby@v1 - with: - ruby-version: ${{ matrix.ruby-version }} - - name: Install addons - if: ${{ matrix.os == 'ubuntu-latest' }} - run: sudo apt-get install libgmp3-dev libcap-ng-dev - - name: Install dependencies - run: bundle install - - name: Run tests - run: bundle exec rake test TESTOPTS=-v From 7f84eae9d9458822e52e1dd90453ae8198c2beeb Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Fri, 5 Jan 2024 18:34:13 +0900 Subject: [PATCH 11/39] test: Check Socket::ResolutionError instead of SocketError on Ruby 3.3 Signed-off-by: Takuro Ashie --- test/plugin/test_out_forward.rb | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb index 80557c5989..3a7f9a2e10 100644 --- a/test/plugin/test_out_forward.rb +++ b/test/plugin/test_out_forward.rb @@ -156,7 +156,14 @@ def try_write(chunk) normal_conf = config_element('match', '**', {}, [ config_element('server', '', {'name' => 'test', 'host' => 'unexisting.yaaaaaaaaaaaaaay.host.example.com'}) ]) - assert_raise SocketError do + + if Socket.const_defined?(:ResolutionError) # as of Ruby 3.3 + error_class = Socket::ResolutionError + else + error_class = SocketError + end + + assert_raise error_class do create_driver(normal_conf) end @@ -165,7 +172,7 @@ def try_write(chunk) ]) @d = d = create_driver(conf) expected_log = "failed to resolve node name when configured" - expected_detail = 'server="test" error_class=SocketError' + expected_detail = "server=\"test\" error_class=#{error_class.name}" logs = d.logs assert{ logs.any?{|log| log.include?(expected_log) && log.include?(expected_detail) } } end From 1a7ad12e156d5fc385ac43d13924d0cb289c113d Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Tue, 9 Jan 2024 21:09:00 +0900 Subject: [PATCH 12/39] test_child_process: Replace some UTF-8 byte sequences with named variables Make easy to understand what these tests do. Signed-off-by: Takuro Ashie --- test/plugin_helper/test_child_process.rb | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/test/plugin_helper/test_child_process.rb b/test/plugin_helper/test_child_process.rb index 5d84a0fd06..e720cef8c7 100644 --- a/test/plugin_helper/test_child_process.rb +++ b/test/plugin_helper/test_child_process.rb @@ -529,7 +529,9 @@ def configure(conf) sleep TEST_WAIT_INTERVAL_FOR_BLOCK_RUNNING until m.locked? || ran m.lock assert_equal Encoding.find('utf-8'), str.encoding - expected = "\xEF\xBF\xBD\xEF\xBF\xBD\x00\xEF\xBF\xBD\xEF\xBF\xBD".force_encoding("utf-8") + replacement = "\uFFFD" # U+FFFD (REPLACEMENT CHARACTER) + nul = "\x00" # U+0000 (NUL) + expected = replacement * 2 + nul + replacement * 2 assert_equal expected, str @d.stop; @d.shutdown; @d.close; @d.terminate end @@ -538,10 +540,11 @@ def configure(conf) test 'can scrub characters without exceptions and replace specified chars' do m = Mutex.new str = nil + replacement = "?" Timeout.timeout(TEST_DEADLOCK_TIMEOUT) do ran = false args = ['-e', 'STDOUT.set_encoding("ascii-8bit"); STDOUT.write "\xFF\xFF\x00\xF0\xF0"'] - @d.child_process_execute(:t13b, "ruby", arguments: args, mode: [:read], scrub: true, replace_string: '?') do |io| + @d.child_process_execute(:t13b, "ruby", arguments: args, mode: [:read], scrub: true, replace_string: replacement) do |io| m.lock ran = true str = io.read @@ -550,7 +553,8 @@ def configure(conf) sleep TEST_WAIT_INTERVAL_FOR_BLOCK_RUNNING until m.locked? || ran m.lock assert_equal Encoding.find('utf-8'), str.encoding - expected = "??\x00??".force_encoding("utf-8") + nul = "\x00" # U+0000 (NUL) + expected = replacement * 2 + nul + replacement * 2 assert_equal expected, str @d.stop; @d.shutdown; @d.close; @d.terminate end From 9796fd35718e7127e0ab2e1098ac74239fcd2aae Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Thu, 11 Jan 2024 09:42:02 +0900 Subject: [PATCH 13/39] test_child_process: Mark scrubbing tests as pending Behaviour of `IO#set_encoding` has been changed as of Ruby 3.3. We don't yet determine how to solve this issue, it might be better to address in Ruby. Signed-off-by: Takuro Ashie --- test/plugin_helper/test_child_process.rb | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/test/plugin_helper/test_child_process.rb b/test/plugin_helper/test_child_process.rb index e720cef8c7..b498b7a0eb 100644 --- a/test/plugin_helper/test_child_process.rb +++ b/test/plugin_helper/test_child_process.rb @@ -515,6 +515,9 @@ def configure(conf) end test 'can scrub characters without exceptions' do + if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.3.0') + pend "Behaviour of IO#set_encoding is changed as of Ruby 3.3 (#4058)" + end m = Mutex.new str = nil Timeout.timeout(TEST_DEADLOCK_TIMEOUT) do @@ -538,6 +541,9 @@ def configure(conf) end test 'can scrub characters without exceptions and replace specified chars' do + if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.3.0') + pend "Behaviour of IO#set_encoding is changed as of Ruby 3.3 (#4058)" + end m = Mutex.new str = nil replacement = "?" From 48c2b056cd19b2ddbe2fecc26b481a001a340814 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Thu, 11 Jan 2024 10:40:13 +0900 Subject: [PATCH 14/39] test_fluentd: Fix failed tests on Ruby 3.3 Signed-off-by: Takuro Ashie --- test/command/test_fluentd.rb | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/test/command/test_fluentd.rb b/test/command/test_fluentd.rb index 87c9d28d96..c6bd88d857 100644 --- a/test/command/test_fluentd.rb +++ b/test/command/test_fluentd.rb @@ -941,7 +941,7 @@ def multi_workers_ready? '-external-encoding' => '--external-encoding=utf-8', '-internal-encoding' => '--internal-encoding=utf-8', ) - test "-E option is set to RUBYOPT" do |opt| + test "-E option is set to RUBYOPT" do |base_opt| conf = < @type dummy @@ -952,6 +952,7 @@ def multi_workers_ready? CONF conf_path = create_conf_file('rubyopt_test.conf', conf) + opt = base_opt.dup opt << " #{ENV['RUBYOPT']}" if ENV['RUBYOPT'] assert_log_matches( create_cmdline(conf_path), @@ -991,9 +992,14 @@ def multi_workers_ready? CONF conf_path = create_conf_file('rubyopt_invalid_test.conf', conf) + if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.3.0') + expected_phrase = 'ruby: invalid switch in RUBYOPT' + else + expected_phrase = 'Invalid option is passed to RUBYOPT' + end assert_log_matches( create_cmdline(conf_path), - 'Invalid option is passed to RUBYOPT', + expected_phrase, env: { 'RUBYOPT' => 'a' }, ) end From 3b93711be8497500417838cc08f314f474cc6198 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Mon, 11 Mar 2024 14:06:46 +0900 Subject: [PATCH 15/39] github: unify YAML file extension to .yml (#4431) Backported form 2ea28c9d556dda1304292a0b48c553911566ca52 --- It is not harmful because GitHub allows both of file extensions (.yaml and .yml), but it may be better to unify it for consistency. .yml is used in https://docs.github.com/en/actions/learn-github-actions/understanding-github-actions Signed-off-by: Kentaro Hayashi --- .github/ISSUE_TEMPLATE/{bug_report.yaml => bug_report.yml} | 0 .../ISSUE_TEMPLATE/{feature_request.yaml => feature_request.yml} | 0 .github/workflows/{test.yaml => test.yml} | 0 3 files changed, 0 insertions(+), 0 deletions(-) rename .github/ISSUE_TEMPLATE/{bug_report.yaml => bug_report.yml} (100%) rename .github/ISSUE_TEMPLATE/{feature_request.yaml => feature_request.yml} (100%) rename .github/workflows/{test.yaml => test.yml} (100%) diff --git a/.github/ISSUE_TEMPLATE/bug_report.yaml b/.github/ISSUE_TEMPLATE/bug_report.yml similarity index 100% rename from .github/ISSUE_TEMPLATE/bug_report.yaml rename to .github/ISSUE_TEMPLATE/bug_report.yml diff --git a/.github/ISSUE_TEMPLATE/feature_request.yaml b/.github/ISSUE_TEMPLATE/feature_request.yml similarity index 100% rename from .github/ISSUE_TEMPLATE/feature_request.yaml rename to .github/ISSUE_TEMPLATE/feature_request.yml diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yml similarity index 100% rename from .github/workflows/test.yaml rename to .github/workflows/test.yml From e66892069228937326305d10b394aa08549ec8af Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Thu, 14 Mar 2024 14:17:19 +0900 Subject: [PATCH 16/39] CI: Fix unstable tests of out_forward (#4436) Backported from 5d18f3551387c4f12b399c7e5ed33f2ac17a27c4 * Avoid calling `instance_start` in duplicate. * Avoid not calling `instance_shutdown`. To fix the following error, which sometimes occur. I'm not sure this actually fixes it, but, at least, we should fix the points above. Error: test: Create new connection per send_data(ForwardOutputTest): ArgumentError: expected loop to be an instance of Coolio::Loop, not nil C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/io.rb:35:in `attach' C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/io.rb:35:in `attach' C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/socket.rb:39:in `attach' (eval):7:in `attach' C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/server.rb:40:in `on_connection' C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/listener.rb:65:in `on_readable' C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/loop.rb:88:in `run_once' C:/hostedtoolcache/windows/Ruby/3.2.3/x64/lib/ruby/gems/3.2.0/gems/cool.io-1.8.0/lib/cool.io/loop.rb:88:in `run' D:/a/fluentd/fluentd/lib/fluent/plugin_helper/event_loop.rb:93:in `block in start' D:/a/fluentd/fluentd/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create' Signed-off-by: Daijiro Fukuda Co-authored-by: Takuro Ashie --- test/plugin/test_out_forward.rb | 62 +++++++++++++-------------------- 1 file changed, 25 insertions(+), 37 deletions(-) diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb index 3a7f9a2e10..438caf8fa2 100644 --- a/test/plugin/test_out_forward.rb +++ b/test/plugin/test_out_forward.rb @@ -1248,27 +1248,22 @@ def plugin_id_for_test? target_input_driver = create_target_input_driver(conf: target_config) output_conf = config d = create_driver(output_conf) - d.instance_start - begin - chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil)) - mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port, - linger_timeout: anything, - send_timeout: anything, - recv_timeout: anything, - connect_timeout: anything - ) { |sock| mock(sock).close.once; sock }.twice + chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil)) + mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port, + linger_timeout: anything, + send_timeout: anything, + recv_timeout: anything, + connect_timeout: anything + ) { |sock| mock(sock).close.once; sock }.twice - target_input_driver.run(timeout: 15) do - d.run(shutdown: false) do - node = d.instance.nodes.first - 2.times do - node.send_data('test', chunk) rescue nil - end + target_input_driver.run(timeout: 15) do + d.run do + node = d.instance.nodes.first + 2.times do + node.send_data('test', chunk) rescue nil end end - ensure - d.instance_shutdown end end @@ -1282,7 +1277,6 @@ def plugin_id_for_test? port #{@target_port} ]) - d.instance_start assert_nothing_raised { d.run } end @@ -1294,33 +1288,28 @@ def plugin_id_for_test? keepalive_timeout 2 ] d = create_driver(output_conf) - d.instance_start - begin - chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil)) - mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port, - linger_timeout: anything, - send_timeout: anything, - recv_timeout: anything, - connect_timeout: anything - ) { |sock| mock(sock).close.once; sock }.once + chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil)) + mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port, + linger_timeout: anything, + send_timeout: anything, + recv_timeout: anything, + connect_timeout: anything + ) { |sock| mock(sock).close.once; sock }.once - target_input_driver.run(timeout: 15) do - d.run(shutdown: false) do - node = d.instance.nodes.first - 2.times do - node.send_data('test', chunk) rescue nil - end + target_input_driver.run(timeout: 15) do + d.run do + node = d.instance.nodes.first + 2.times do + node.send_data('test', chunk) rescue nil end end - ensure - d.instance_shutdown end end test 'create timer of purging obsolete sockets' do output_conf = config + %[keepalive true] - d = create_driver(output_conf) + @d = d = create_driver(output_conf) mock(d.instance).timer_execute(:out_forward_heartbeat_request, 1).once mock(d.instance).timer_execute(:out_forward_keep_alived_socket_watcher, 5).once @@ -1336,7 +1325,6 @@ def plugin_id_for_test? keepalive_timeout 2 ] d = create_driver(output_conf) - d.instance_start chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil)) mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port, From 1ffa7140143184de10d26a39f98b55d3dba9e362 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Thu, 14 Mar 2024 16:54:47 +0900 Subject: [PATCH 17/39] v1.16.4 Signed-off-by: Kentaro Hayashi Signed-off-by: Daijiro Fukuda Co-authored-by: Kentaro Hayashi --- CHANGELOG.md | 16 ++++++++++++++++ lib/fluent/version.rb | 2 +- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e44d3b798..fc6be3a818 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # v1.16 +## Release v1.16.4 - 2024/03/14 + +### Bug Fix + +* Fix to avoid processing discarded chunks in write_step_by_step. + It fixes not to raise pile of IOError when many `chunk + bytes limit exceeds` errors are occurred. + https://github.com/fluent/fluentd/pull/4342 +* in_tail: Fix tail watchers in `rotate_wait` state not being managed. + https://github.com/fluent/fluentd/pull/4334 + +### Misc + +* buffer: Avoid unnecessary log processing. It will improve performance. + https://github.com/fluent/fluentd/pull/4331 + ## Release v1.16.3 - 2023/11/14 ### Bug Fix diff --git a/lib/fluent/version.rb b/lib/fluent/version.rb index a84ee72e22..584e7ded31 100644 --- a/lib/fluent/version.rb +++ b/lib/fluent/version.rb @@ -16,6 +16,6 @@ module Fluent - VERSION = '1.16.3' + VERSION = '1.16.4' end From 403a28ff8d74adfeefb9842c8342292397ba84b7 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Wed, 27 Mar 2024 14:40:36 +0900 Subject: [PATCH 18/39] buffer: fix emit error of race condition (#4450) Backported from 13a5199522ee1fc3f55c40a73d30c6036d34570f. After 95438b2eeb4bb5586a4031c1fbc35756f3c12565 (#4342), there is a section where chunks do not have a lock in `write_step_by_step()`. `write_step_by_step()` must ensure their locks until passing them to the block. Otherwise, race condition can occur and it can cause emit error by IOError. Example of warning messages of emit error: [warn]: #0 emit transaction failed: error_class=IOError error="closed stream" location=... [warn]: #0 send an error event stream to @ERROR: error_class=IOError error="closed stream" location=... Signed-off-by: Daijiro Fukuda --- lib/fluent/plugin/buffer.rb | 143 +++++++++++++++++++----------------- test/plugin/test_buffer.rb | 59 +++++++++++++++ 2 files changed, 134 insertions(+), 68 deletions(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index 0251de3409..80709c12bb 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -764,94 +764,95 @@ def write_step_by_step(metadata, data, format, splits_count, &block) while writing_splits_index < splits.size chunk = get_next_chunk.call errors = [] + # The chunk must be locked until being passed to &block. + chunk.mon_enter modified_chunks << {chunk: chunk, adding_bytesize: 0, errors: errors} - chunk.synchronize do - raise ShouldRetry unless chunk.writable? - staged_chunk_used = true if chunk.staged? - - original_bytesize = committed_bytesize = chunk.bytesize - begin - while writing_splits_index < splits.size - split = splits[writing_splits_index] - formatted_split = format ? format.call(split) : nil - if split.size == 1 # Check BufferChunkOverflowError - determined_bytesize = nil - if @compress != :text - determined_bytesize = nil - elsif formatted_split - determined_bytesize = formatted_split.bytesize - elsif split.first.respond_to?(:bytesize) - determined_bytesize = split.first.bytesize - end + raise ShouldRetry unless chunk.writable? + staged_chunk_used = true if chunk.staged? - if determined_bytesize && determined_bytesize > @chunk_limit_size - # It is a obvious case that BufferChunkOverflowError should be raised here. - # But if it raises here, already processed 'split' or - # the proceeding 'split' will be lost completely. - # So it is a last resort to delay raising such a exception - errors << "a #{determined_bytesize} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})" - writing_splits_index += 1 - next - end + original_bytesize = committed_bytesize = chunk.bytesize + begin + while writing_splits_index < splits.size + split = splits[writing_splits_index] + formatted_split = format ? format.call(split) : nil - if determined_bytesize.nil? || chunk.bytesize + determined_bytesize > @chunk_limit_size - # The split will (might) cause size over so keep already processed - # 'split' content here (allow performance regression a bit). - chunk.commit - committed_bytesize = chunk.bytesize - end + if split.size == 1 # Check BufferChunkOverflowError + determined_bytesize = nil + if @compress != :text + determined_bytesize = nil + elsif formatted_split + determined_bytesize = formatted_split.bytesize + elsif split.first.respond_to?(:bytesize) + determined_bytesize = split.first.bytesize end - if format - chunk.concat(formatted_split, split.size) - else - chunk.append(split, compress: @compress) + if determined_bytesize && determined_bytesize > @chunk_limit_size + # It is a obvious case that BufferChunkOverflowError should be raised here. + # But if it raises here, already processed 'split' or + # the proceeding 'split' will be lost completely. + # So it is a last resort to delay raising such a exception + errors << "a #{determined_bytesize} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})" + writing_splits_index += 1 + next end - adding_bytes = chunk.bytesize - committed_bytesize - if chunk_size_over?(chunk) # split size is larger than difference between size_full? and size_over? - chunk.rollback + if determined_bytesize.nil? || chunk.bytesize + determined_bytesize > @chunk_limit_size + # The split will (might) cause size over so keep already processed + # 'split' content here (allow performance regression a bit). + chunk.commit committed_bytesize = chunk.bytesize + end + end - if split.size == 1 # Check BufferChunkOverflowError again - if adding_bytes > @chunk_limit_size - errors << "concatenated/appended a #{adding_bytes} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})" - writing_splits_index += 1 - next - else - # As already processed content is kept after rollback, then unstaged chunk should be queued. - # After that, re-process current split again. - # New chunk should be allocated, to do it, modify @stage and so on. - synchronize { @stage.delete(modified_metadata) } - staged_chunk_used = false - chunk.unstaged! - break - end - end + if format + chunk.concat(formatted_split, split.size) + else + chunk.append(split, compress: @compress) + end + adding_bytes = chunk.bytesize - committed_bytesize - if chunk_size_full?(chunk) || split.size == 1 - enqueue_chunk_before_retry = true + if chunk_size_over?(chunk) # split size is larger than difference between size_full? and size_over? + chunk.rollback + committed_bytesize = chunk.bytesize + + if split.size == 1 # Check BufferChunkOverflowError again + if adding_bytes > @chunk_limit_size + errors << "concatenated/appended a #{adding_bytes} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})" + writing_splits_index += 1 + next else - splits_count *= 10 + # As already processed content is kept after rollback, then unstaged chunk should be queued. + # After that, re-process current split again. + # New chunk should be allocated, to do it, modify @stage and so on. + synchronize { @stage.delete(modified_metadata) } + staged_chunk_used = false + chunk.unstaged! + break end + end - raise ShouldRetry + if chunk_size_full?(chunk) || split.size == 1 + enqueue_chunk_before_retry = true + else + splits_count *= 10 end - writing_splits_index += 1 + raise ShouldRetry + end - if chunk_size_full?(chunk) - break - end + writing_splits_index += 1 + + if chunk_size_full?(chunk) + break end - rescue - chunk.purge if chunk.unstaged? # unstaged chunk will leak unless purge it - raise end - - modified_chunks.last[:adding_bytesize] = chunk.bytesize - original_bytesize + rescue + chunk.purge if chunk.unstaged? # unstaged chunk will leak unless purge it + raise end + + modified_chunks.last[:adding_bytesize] = chunk.bytesize - original_bytesize end modified_chunks.each do |data| block.call(data[:chunk], data[:adding_bytesize], data[:errors]) @@ -863,9 +864,15 @@ def write_step_by_step(metadata, data, format, splits_count, &block) if chunk.unstaged? chunk.purge rescue nil end + chunk.mon_exit rescue nil end enqueue_chunk(metadata) if enqueue_chunk_before_retry retry + ensure + modified_chunks.each do |data| + chunk = data[:chunk] + chunk.mon_exit + end end STATS_KEYS = [ diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 451358a5b1..14d33af9c7 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -901,6 +901,65 @@ def create_chunk_es(metadata, es) assert_equal 2, purge_count end + + # https://github.com/fluent/fluentd/issues/4446 + test "#write_step_by_step keeps chunks kept in locked in entire #write process" do + assert_equal 8 * 1024 * 1024, @p.chunk_limit_size + assert_equal 0.95, @p.chunk_full_threshold + + mon_enter_counts_by_chunk = {} + mon_exit_counts_by_chunk = {} + + stub.proxy(@p).generate_chunk(anything) do |chunk| + stub(chunk).mon_enter do + enter_count = 1 + mon_enter_counts_by_chunk.fetch(chunk, 0) + exit_count = mon_exit_counts_by_chunk.fetch(chunk, 0) + mon_enter_counts_by_chunk[chunk] = enter_count + + # Assert that chunk is passed to &block of write_step_by_step before exiting the lock. + # (i.e. The lock count must be 2 greater than the exit count). + # Since ShouldRetry occurs once, the staged chunk takes the lock 3 times when calling the block. + if chunk.staged? + lock_in_block = enter_count == 3 + assert_equal(enter_count - 2, exit_count) if lock_in_block + else + lock_in_block = enter_count == 2 + assert_equal(enter_count - 2, exit_count) if lock_in_block + end + end + stub(chunk).mon_exit do + exit_count = 1 + mon_exit_counts_by_chunk.fetch(chunk, 0) + mon_exit_counts_by_chunk[chunk] = exit_count + end + chunk + end + + m = @p.metadata(timekey: Time.parse('2016-04-11 16:40:00 +0000').to_i) + small_row = "x" * 1024 * 400 + big_row = "x" * 1024 * 1024 * 8 # just `chunk_size_limit`, it does't cause BufferOverFlowError. + + # Write 42 events in 1 event stream, last one is for triggering `ShouldRetry` + @p.write({m => [small_row] * 40 + [big_row] + ["x"]}) + + # Above event strem will be splitted twice by `Buffer#write_step_by_step` + # + # 1. `write_once`: 42 [events] * 1 [stream] + # 2. `write_step_by_step`: 4 [events]* 10 [streams] + 2 [events] * 1 [stream] + # 3. `write_step_by_step` (by `ShouldRetry`): 1 [event] * 42 [streams] + # + # Example of staged chunk lock behavior: + # + # 1. mon_enter in write_step_by_step + # 2. ShouldRetry occurs + # 3. mon_exit in write_step_by_step + # 4. mon_enter again in write_step_by_step (retry) + # 5. passed to &block of write_step_by_step + # 6. mon_enter in the block (write) + # 7. mon_exit in write_step_by_step + # 8. mon_exit in write + + assert_equal(mon_enter_counts_by_chunk.values, mon_exit_counts_by_chunk.values) + end end sub_test_case 'standard format with configuration for test with lower chunk limit size' do From 95d130aaa44fc09b14a4e9686ee8018253185fc5 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Wed, 27 Mar 2024 16:56:07 +0900 Subject: [PATCH 19/39] v1.16.5 Signed-off-by: Daijiro Fukuda --- CHANGELOG.md | 8 ++++++++ lib/fluent/version.rb | 2 +- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fc6be3a818..848a5cb5c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # v1.16 +## Release v1.16.5 - 2024/03/27 + +### Bug Fix + +* Buffer: Fix emit error of v1.16.4 sometimes failing to process large data + exceeding chunk size limit + https://github.com/fluent/fluentd/pull/4447 + ## Release v1.16.4 - 2024/03/14 ### Bug Fix diff --git a/lib/fluent/version.rb b/lib/fluent/version.rb index 584e7ded31..140fc43c43 100644 --- a/lib/fluent/version.rb +++ b/lib/fluent/version.rb @@ -16,6 +16,6 @@ module Fluent - VERSION = '1.16.4' + VERSION = '1.16.5' end From 44d132c2e9fe866a1254a9b8676e80125106ce43 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Wed, 27 Mar 2024 18:30:38 +0900 Subject: [PATCH 20/39] ci: do not show test detail immediately (#4454) TESTOPTS=-v shows running test and the result. If something weird happen, it show error immediately. As rake task executes many tests, so it may be better to delay showing test details later. NOTE: --progress-style=fault-only helps you to focus on failure, but it is inconvenient when test case has stalled. Signed-off-by: Kentaro Hayashi Co-authored-by: Kentaro Hayashi --- .github/workflows/test.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 9332491372..666aa97b86 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -29,4 +29,4 @@ jobs: - name: Install dependencies run: bundle install - name: Run tests - run: bundle exec rake test TESTOPTS=-v + run: bundle exec rake test TESTOPTS="-v --no-show-detail-immediately" From aeed909a596bb5a4b72d8a6cc6fc5160e8f46512 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Wed, 29 May 2024 10:24:20 +0900 Subject: [PATCH 21/39] v1.16: keep console v1.23 (#4510) console gem v1.24 and v1.25 has some specification changes that influence Fluentd. Since they have nothing to do with vulnerability, we should keep the version on v1.16 stable branch. Signed-off-by: Daijiro Fukuda --- fluentd.gemspec | 1 + 1 file changed, 1 insertion(+) diff --git a/fluentd.gemspec b/fluentd.gemspec index a5b5c805e3..f519ba3531 100644 --- a/fluentd.gemspec +++ b/fluentd.gemspec @@ -29,6 +29,7 @@ Gem::Specification.new do |gem| gem.add_runtime_dependency("tzinfo-data", ["~> 1.0"]) gem.add_runtime_dependency("strptime", [">= 0.2.4", "< 1.0.0"]) gem.add_runtime_dependency("webrick", ["~> 1.4"]) + gem.add_runtime_dependency("console", ["< 1.24"]) # build gem for a certain platform. see also Rakefile fake_platform = ENV['GEM_BUILD_FAKE_PLATFORM'].to_s From 056f2e43932f6d4e349211e7d2ac1e5ee28bd758 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Wed, 29 May 2024 13:40:36 +0900 Subject: [PATCH 22/39] out_file: add warn message for symlink_path setting (#4512) Backported from 74b2e3d7f86f656d65951c4ea095d68c20b0b858. (#4502) Signed-off-by: Shingo Nakayama Signed-off-by: Daijiro Fukuda Co-authored-by: Shingo Nakayama <152375941+Shingo-Nakayama@users.noreply.github.com> --- lib/fluent/plugin/out_file.rb | 8 ++++++++ test/plugin/test_out_file.rb | 22 +++++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/lib/fluent/plugin/out_file.rb b/lib/fluent/plugin/out_file.rb index 7de4b90979..19c0afa8ef 100644 --- a/lib/fluent/plugin/out_file.rb +++ b/lib/fluent/plugin/out_file.rb @@ -172,6 +172,14 @@ def configure(conf) log.warn "symlink_path is unavailable on Windows platform. disabled." @symlink_path = nil else + placeholder_validators(:symlink_path, @symlink_path).reject{ |v| v.type == :time }.each do |v| + begin + v.validate! + rescue Fluent::ConfigError => e + log.warn "#{e}. This means multiple chunks are competing for a single symlink_path, so some logs may not be taken from the symlink." + end + end + @buffer.extend SymlinkBufferMixin @buffer.symlink_path = @symlink_path @buffer.output_plugin_for_symlink = self diff --git a/test/plugin/test_out_file.rb b/test/plugin/test_out_file.rb index 5727bc1e69..8b6ab161d0 100644 --- a/test/plugin/test_out_file.rb +++ b/test/plugin/test_out_file.rb @@ -130,7 +130,7 @@ def create_driver(conf = CONFIG, opts = {}) 'path' => "#{TMP_DIR}/${tag}/${type}/conf_test.%Y%m%d.%H%M.log", 'add_path_suffix' => 'false', 'append' => "true", - 'symlink_path' => "#{TMP_DIR}/conf_test.current.log", + 'symlink_path' => "#{TMP_DIR}/${tag}/conf_test.current.log", 'compress' => 'gzip', 'recompress' => 'true', }, [ @@ -183,6 +183,26 @@ def create_driver(conf = CONFIG, opts = {}) Fluent::Test::Driver::Output.new(Fluent::Plugin::NullOutput).configure(conf) end end + + test 'warning for symlink_path not including correct placeholders corresponding to chunk keys' do + omit "Windows doesn't support symlink" if Fluent.windows? + conf = config_element('match', '**', { + 'path' => "#{TMP_DIR}/${tag}/${key1}/${key2}/conf_test.%Y%m%d.%H%M.log", + 'symlink_path' => "#{TMP_DIR}/conf_test.current.log", + }, [ + config_element('buffer', 'time,tag,key1,key2', { + '@type' => 'file', + 'timekey' => '1d', + 'path' => "#{TMP_DIR}/buf_conf_test", + }), + ]) + assert_nothing_raised do + d = create_driver(conf) + assert do + d.logs.count { |log| log.include?("multiple chunks are competing for a single symlink_path") } == 2 + end + end + end end sub_test_case 'fully configured output' do From 17c66e1feb0da1533a4d084023df296fef5ba2ad Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Fri, 7 Jun 2024 18:04:45 +0900 Subject: [PATCH 23/39] test: fix timecop version to keep clock specs in tests (#4524) Backported from 57f821b194516fabd8b8d396842932126c6cbaf5. --- timecop 0.9.9 supports `Process.clock_gettime`. This breaks specifications of `process_extenstion` of Fluentd and `Fluent::Clock`. `Fluent::Clock` uses `CLOCK_MONOTONIC_RAW` if possible and it does not be affected. However, `CLOCK_MONOTONIC_RAW` is not available on Windows, so the impact on tests on Windows is very significant. For now, we should avoid this effect by fixing the version. Signed-off-by: Daijiro Fukuda --- fluentd.gemspec | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/fluentd.gemspec b/fluentd.gemspec index f519ba3531..879c26f6ff 100644 --- a/fluentd.gemspec +++ b/fluentd.gemspec @@ -46,7 +46,9 @@ Gem::Specification.new do |gem| gem.add_development_dependency("parallel_tests", ["~> 0.15.3"]) gem.add_development_dependency("simplecov", ["~> 0.7"]) gem.add_development_dependency("rr", ["~> 3.0"]) - gem.add_development_dependency("timecop", ["~> 0.9"]) + # timecop v0.9.9 supports `Process.clock_gettime`. It breaks some tests. + # (https://github.com/fluent/fluentd/pull/4521) + gem.add_development_dependency("timecop", ["< 0.9.9"]) gem.add_development_dependency("test-unit", ["~> 3.3"]) gem.add_development_dependency("test-unit-rr", ["~> 1.0"]) gem.add_development_dependency("oj", [">= 2.14", "< 4"]) From 2020531d98032332cae07f1f8db76efe39413958 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Thu, 2 May 2024 16:03:51 +0900 Subject: [PATCH 24/39] config: yaml_parser: Handle $log_level element for special case Backported from 236d87d07fea8b3396f33985247c96ad59911f3e. --- This is because `@log_level` is invalid for YAML. Instead, we should interpret $log_level as `@log_level` on YAML parser. Signed-off-by: Hiroshi Hatake --- lib/fluent/config/yaml_parser/parser.rb | 4 ++++ test/test_config.rb | 6 ++++++ 2 files changed, 10 insertions(+) diff --git a/lib/fluent/config/yaml_parser/parser.rb b/lib/fluent/config/yaml_parser/parser.rb index dc1caaf46d..862f774802 100644 --- a/lib/fluent/config/yaml_parser/parser.rb +++ b/lib/fluent/config/yaml_parser/parser.rb @@ -138,6 +138,10 @@ def section_build(name, config, indent: 0, arg: nil) sb.add_line('@id', v) end + if (v = config.delete('$log_level')) + sb.add_line('@log_level', v) + end + config.each do |key, val| if val.is_a?(Array) val.each do |v| diff --git a/test/test_config.rb b/test/test_config.rb index 24f63830d4..7533e6e16b 100644 --- a/test/test_config.rb +++ b/test/test_config.rb @@ -167,6 +167,7 @@ def test_included tag: tag.dummy - source: $type: tcp + $log_level: info tag: tag.tcp parse: $arg: @@ -176,6 +177,7 @@ def test_included - match: $tag: tag.* $type: stdout + $log_level: debug buffer: $type: memory flush_interval: 1s @@ -208,10 +210,12 @@ def test_included 'tag.dummy', 'tcp', 'tag.tcp', + 'info', 'none', 'why.parse.section.doesnot.have.arg,huh', 'stdout', 'tag.*', + 'debug', 'null', '**', '@FLUENT_LOG', @@ -224,10 +228,12 @@ def test_included dummy_source_conf['tag'], tcp_source_conf['@type'], tcp_source_conf['tag'], + tcp_source_conf['@log_level'], parse_tcp_conf['@type'], parse_tcp_conf.arg, match_conf['@type'], match_conf.arg, + match_conf['@log_level'], fluent_log_conf['@type'], fluent_log_conf.arg, label_conf.arg, From 260a7f5777a7aaa4556459e755e1fc866c3792b8 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Fri, 16 Aug 2024 10:05:04 +0900 Subject: [PATCH 25/39] test: fix IOError about Tempfile closed when GC (#4591) Backported from 3b2798479769c1c24020c65d0caf170423b3591a. --- `Tempfile#binmode` returns `File` object, not own `Tempfile` object. So, GC will cause its finalizer and the file can be closed during the test. This is the cause why these tests sometimes fail by `IOError: closed stream`. Signed-off-by: Daijiro Fukuda --- test/plugin/in_tail/test_io_handler.rb | 27 +++++++++++------------ test/plugin/in_tail/test_position_file.rb | 13 +++++------ 2 files changed, 19 insertions(+), 21 deletions(-) diff --git a/test/plugin/in_tail/test_io_handler.rb b/test/plugin/in_tail/test_io_handler.rb index 647c4a55a1..0def36b31c 100644 --- a/test/plugin/in_tail/test_io_handler.rb +++ b/test/plugin/in_tail/test_io_handler.rb @@ -5,20 +5,19 @@ require 'tempfile' class IntailIOHandlerTest < Test::Unit::TestCase - setup do - @file = Tempfile.new('intail_io_handler').binmode - opened_file_metrics = Fluent::Plugin::LocalMetrics.new - opened_file_metrics.configure(config_element('metrics', '', {})) - closed_file_metrics = Fluent::Plugin::LocalMetrics.new - closed_file_metrics.configure(config_element('metrics', '', {})) - rotated_file_metrics = Fluent::Plugin::LocalMetrics.new - rotated_file_metrics.configure(config_element('metrics', '', {})) - @metrics = Fluent::Plugin::TailInput::MetricsInfo.new(opened_file_metrics, closed_file_metrics, rotated_file_metrics) - end - - teardown do - @file.close rescue nil - @file.unlink rescue nil + def setup + Tempfile.create('intail_io_handler') do |file| + file.binmode + @file = file + opened_file_metrics = Fluent::Plugin::LocalMetrics.new + opened_file_metrics.configure(config_element('metrics', '', {})) + closed_file_metrics = Fluent::Plugin::LocalMetrics.new + closed_file_metrics.configure(config_element('metrics', '', {})) + rotated_file_metrics = Fluent::Plugin::LocalMetrics.new + rotated_file_metrics.configure(config_element('metrics', '', {})) + @metrics = Fluent::Plugin::TailInput::MetricsInfo.new(opened_file_metrics, closed_file_metrics, rotated_file_metrics) + yield + end end def create_target_info diff --git a/test/plugin/in_tail/test_position_file.rb b/test/plugin/in_tail/test_position_file.rb index af692fdee5..b1957f7353 100644 --- a/test/plugin/in_tail/test_position_file.rb +++ b/test/plugin/in_tail/test_position_file.rb @@ -6,13 +6,12 @@ require 'tempfile' class IntailPositionFileTest < Test::Unit::TestCase - setup do - @file = Tempfile.new('intail_position_file_test').binmode - end - - teardown do - @file.close rescue nil - @file.unlink rescue nil + def setup + Tempfile.create('intail_position_file_test') do |file| + file.binmode + @file = file + yield + end end UNWATCHED_STR = '%016x' % Fluent::Plugin::TailInput::PositionFile::UNWATCHED_POSITION From 2824884b2654303d4102bf5410da163eedc79b26 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Fri, 16 Aug 2024 14:57:55 +0900 Subject: [PATCH 26/39] parser_json: fix wrong LoadError warning (#4592) Backported from 891ce71120198fa168232a796d4267a3cdb2c91f. --- If Oj is not installed, LoadError with the empty message is raised. So, the current condition `/\boj\z/.match?(ex.message)` does not work and the following meaningless warning is displayed. {datetime} [warn]: #x {id} LoadError After this fix, the log message will be: {datetime} [info]: #x {id} Oj is not installed, and failing back to Yajl for json parser Refactor "rescue" logic because this falling back feature is currently only for "oj" (LoadError can not occur for "json" and "yajl"). Signed-off-by: Daijiro Fukuda Co-authored-by: Takuro Ashie --- lib/fluent/plugin/parser_json.rb | 16 ++++------------ test/plugin/test_parser_json.rb | 31 +++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 12 deletions(-) diff --git a/lib/fluent/plugin/parser_json.rb b/lib/fluent/plugin/parser_json.rb index 829aa4b72a..79fea5abe9 100644 --- a/lib/fluent/plugin/parser_json.rb +++ b/lib/fluent/plugin/parser_json.rb @@ -50,23 +50,15 @@ def configure(conf) def configure_json_parser(name) case name when :oj - raise LoadError unless Fluent::OjOptions.available? - [Oj.method(:load), Oj::ParseError] + return [Oj.method(:load), Oj::ParseError] if Fluent::OjOptions.available? + + log&.info "Oj is not installed, and failing back to Yajl for json parser" + configure_json_parser(:yajl) when :json then [JSON.method(:load), JSON::ParserError] when :yajl then [Yajl.method(:load), Yajl::ParseError] else raise "BUG: unknown json parser specified: #{name}" end - rescue LoadError => ex - name = :yajl - if log - if /\boj\z/.match?(ex.message) - log.info "Oj is not installed, and failing back to Yajl for json parser" - else - log.warn ex.message - end - end - retry end def parse(text) diff --git a/test/plugin/test_parser_json.rb b/test/plugin/test_parser_json.rb index 19c45402d1..e60784628e 100644 --- a/test/plugin/test_parser_json.rb +++ b/test/plugin/test_parser_json.rb @@ -8,6 +8,37 @@ def setup @parser = Fluent::Test::Driver::Parser.new(Fluent::Plugin::JSONParser) end + sub_test_case "configure_json_parser" do + data("oj", [:oj, [Oj.method(:load), Oj::ParseError]]) + data("json", [:json, [JSON.method(:load), JSON::ParserError]]) + data("yajl", [:yajl, [Yajl.method(:load), Yajl::ParseError]]) + def test_return_each_loader((input, expected_return)) + result = @parser.instance.configure_json_parser(input) + assert_equal expected_return, result + end + + def test_raise_exception_for_unknown_input + assert_raise RuntimeError do + @parser.instance.configure_json_parser(:unknown) + end + end + + def test_fall_back_oj_to_yajl_if_oj_not_available + stub(Fluent::OjOptions).available? { false } + + result = @parser.instance.configure_json_parser(:oj) + + assert_equal [Yajl.method(:load), Yajl::ParseError], result + logs = @parser.logs.collect do |log| + log.gsub(/\A\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} [-+]\d{4} /, "") + end + assert_equal( + ["[info]: Oj is not installed, and failing back to Yajl for json parser\n"], + logs + ) + end + end + data('oj' => 'oj', 'yajl' => 'yajl') def test_parse(data) @parser.configure('json_parser' => data) From a9b9a2e1457f6e5c6cc36c02339cc14735ec3af3 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Tue, 20 Aug 2024 12:39:20 +0900 Subject: [PATCH 27/39] fluentd command: fix plugin_dirs not to overwrite default value (#4606) Backported from fe5843f8faf1add2bce1a34008bf08d0aaf75d0b --- This option is explained as "add plugin directory". However, since v1.16.0, the behavior has changed to overwrite the default value unintentionally. (PR: #4064, commit: 41678bf0e5f355979fd3b584cea5828690ccbac8). We should revert it to the original behavior. Signed-off-by: Daijiro Fukuda --- lib/fluent/command/fluentd.rb | 2 +- test/command/test_fluentd.rb | 65 ++++++++++++++++++++++++++++++----- 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/lib/fluent/command/fluentd.rb b/lib/fluent/command/fluentd.rb index afa26553e5..da25b74cca 100644 --- a/lib/fluent/command/fluentd.rb +++ b/lib/fluent/command/fluentd.rb @@ -46,7 +46,7 @@ } op.on('-p', '--plugin DIR', "add plugin directory") {|s| - (cmd_opts[:plugin_dirs] ||= []) << s + (cmd_opts[:plugin_dirs] ||= default_opts[:plugin_dirs]) << s } op.on('-I PATH', "add library path") {|s| diff --git a/test/command/test_fluentd.rb b/test/command/test_fluentd.rb index c6bd88d857..d73f0dd1af 100644 --- a/test/command/test_fluentd.rb +++ b/test/command/test_fluentd.rb @@ -128,11 +128,14 @@ def eager_read(io) # ATTENTION: This stops taking logs when all `pattern_list` match or timeout, # so `patterns_not_match` can test only logs up to that point. + # You can pass a block to assert something after log matching. def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout: 20, env: {}) matched = false matched_wrongly = false - assert_error_msg = "" + error_msg_match = "" stdio_buf = "" + succeeded_block = true + error_msg_block = "" begin execute_command(cmdline, @tmp_dir, env) do |pid, stdout| begin @@ -163,6 +166,13 @@ def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout: end end end + + begin + yield if block_given? + rescue => e + succeeded_block = false + error_msg_block = "failed block execution after matching: #{e}" + end ensure if SUPERVISOR_PID_PATTERN =~ stdio_buf @supervisor_pid = $1.to_i @@ -173,19 +183,19 @@ def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout: end end rescue Timeout::Error - assert_error_msg = "execution timeout" + error_msg_match = "execution timeout" # https://github.com/fluent/fluentd/issues/4095 # On Windows, timeout without `@supervisor_pid` means that the test is invalid, # since the supervisor process will survive without being killed correctly. flunk("Invalid test: The pid of supervisor could not be taken, which is necessary on Windows.") if Fluent.windows? && @supervisor_pid.nil? rescue => e - assert_error_msg = "unexpected error in launching fluentd: #{e.inspect}" + error_msg_match = "unexpected error in launching fluentd: #{e.inspect}" else - assert_error_msg = "log doesn't match" unless matched + error_msg_match = "log doesn't match" unless matched end if patterns_not_match.empty? - assert_error_msg = build_message(assert_error_msg, + error_msg_match = build_message(error_msg_match, "\nwas expected to include:\n", stdio_buf, pattern_list) else @@ -197,16 +207,17 @@ def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout: lines.any?{|line| line.include?(ptn) } end if matched_wrongly - assert_error_msg << "\n" unless assert_error_msg.empty? - assert_error_msg << "pattern exists in logs wrongly: #{ptn}" + error_msg_match << "\n" unless error_msg_match.empty? + error_msg_match << "pattern exists in logs wrongly: #{ptn}" end end - assert_error_msg = build_message(assert_error_msg, + error_msg_match = build_message(error_msg_match, "\nwas expected to include:\n\nand not include:\n", stdio_buf, pattern_list, patterns_not_match) end - assert matched && !matched_wrongly, assert_error_msg + assert matched && !matched_wrongly, error_msg_match + assert succeeded_block, error_msg_block if block_given? end def assert_fluentd_fails_to_start(cmdline, *pattern_list, timeout: 20) @@ -1288,4 +1299,40 @@ def multi_workers_ready?; true; end "[debug]") end end + + sub_test_case "plugin option" do + test "should be the default value when not specifying" do + conf_path = create_conf_file('test.conf', <<~CONF) + + @type monitor_agent + + CONF + assert File.exist?(conf_path) + cmdline = create_cmdline(conf_path) + + assert_log_matches(cmdline, "fluentd worker is now running") do + response = Net::HTTP.get(URI.parse("http://localhost:24220/api/config.json")) + actual_conf = JSON.parse(response) + assert_equal Fluent::Supervisor.default_options[:plugin_dirs], actual_conf["plugin_dirs"] + end + end + + data(short: "-p") + data(long: "--plugin") + test "can be added by specifying the option" do |option_name| + conf_path = create_conf_file('test.conf', <<~CONF) + + @type monitor_agent + + CONF + assert File.exist?(conf_path) + cmdline = create_cmdline(conf_path, option_name, @tmp_dir, option_name, @tmp_dir) + + assert_log_matches(cmdline, "fluentd worker is now running") do + response = Net::HTTP.get(URI.parse("http://localhost:24220/api/config.json")) + actual_conf = JSON.parse(response) + assert_equal Fluent::Supervisor.default_options[:plugin_dirs] + [@tmp_dir, @tmp_dir], actual_conf["plugin_dirs"] + end + end + end end From b5db2c6e5cd2dac006e92d6dd8993d59cab90c51 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Fri, 16 Aug 2024 17:20:16 +0900 Subject: [PATCH 28/39] v1.16.6 Signed-off-by: Kentaro Hayashi --- CHANGELOG.md | 16 ++++++++++++++++ lib/fluent/version.rb | 2 +- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 848a5cb5c9..47c2e628a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # v1.16 +## Release v1.16.6 - 2024/08/16 + +### Enhancement + +* yaml_parser: Support $log_level element + https://github.com/fluent/fluentd/pull/4486 + +### Bug Fix + +* Fix LoadError with console gem v1.25 + https://github.com/fluent/fluentd/pull/4510 +* out_file: Add warn message for symlink_path setting + https://github.com/fluent/fluentd/pull/4512 +* parser_json: Fix wrong LoadError warning + https://github.com/fluent/fluentd/pull/4592 + ## Release v1.16.5 - 2024/03/27 ### Bug Fix diff --git a/lib/fluent/version.rb b/lib/fluent/version.rb index 140fc43c43..6c6fdd7b24 100644 --- a/lib/fluent/version.rb +++ b/lib/fluent/version.rb @@ -16,6 +16,6 @@ module Fluent - VERSION = '1.16.5' + VERSION = '1.16.6' end From 63adf0d2bbedf493ab1fb23467aea37fe0f0d161 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Tue, 20 Aug 2024 13:21:39 +0900 Subject: [PATCH 29/39] v1.16.6: add #4605 Signed-off-by: Daijiro Fukuda --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 47c2e628a1..4ec88005a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,8 @@ https://github.com/fluent/fluentd/pull/4512 * parser_json: Fix wrong LoadError warning https://github.com/fluent/fluentd/pull/4592 +* `fluentd` command: Fix `--plugin` (`-p`) option not to overwrite default value + https://github.com/fluent/fluentd/pull/4605 ## Release v1.16.5 - 2024/03/27 From cb5e5156372ba2e67ab52e5995b1263de771e224 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Tue, 20 Aug 2024 14:29:13 +0900 Subject: [PATCH 30/39] v1.16.6: Move #4510 to Misc Signed-off-by: Daijiro Fukuda --- CHANGELOG.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ec88005a1..87a4cdb87a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,6 @@ ### Bug Fix -* Fix LoadError with console gem v1.25 - https://github.com/fluent/fluentd/pull/4510 * out_file: Add warn message for symlink_path setting https://github.com/fluent/fluentd/pull/4512 * parser_json: Fix wrong LoadError warning @@ -18,6 +16,11 @@ * `fluentd` command: Fix `--plugin` (`-p`) option not to overwrite default value https://github.com/fluent/fluentd/pull/4605 +### Misc + +* Keep console gem v1.23 to avoid LoadError + https://github.com/fluent/fluentd/pull/4510 + ## Release v1.16.5 - 2024/03/27 ### Bug Fix From f12fc954541fa771fcecdaaa5449ebb6bf5e6683 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Tue, 20 Aug 2024 14:29:47 +0900 Subject: [PATCH 31/39] v1.16.6: Move #4486 to Bug Fix Signed-off-by: Daijiro Fukuda --- CHANGELOG.md | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 87a4cdb87a..4570fb609b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,13 +2,10 @@ ## Release v1.16.6 - 2024/08/16 -### Enhancement - -* yaml_parser: Support $log_level element - https://github.com/fluent/fluentd/pull/4486 - ### Bug Fix +* YAML config syntax: Fix issue where `$log_level` element was not supported correctly + https://github.com/fluent/fluentd/pull/4486 * out_file: Add warn message for symlink_path setting https://github.com/fluent/fluentd/pull/4512 * parser_json: Fix wrong LoadError warning From f0cc1f0875c79f51166e1118795792f59b62f273 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Tue, 20 Aug 2024 16:19:42 +0900 Subject: [PATCH 32/39] v1.16.6: Move #4512 to Misc Because it does not fix the bug itself. Signed-off-by: Daijiro Fukuda --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4570fb609b..0f27ba6a60 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,6 @@ * YAML config syntax: Fix issue where `$log_level` element was not supported correctly https://github.com/fluent/fluentd/pull/4486 -* out_file: Add warn message for symlink_path setting - https://github.com/fluent/fluentd/pull/4512 * parser_json: Fix wrong LoadError warning https://github.com/fluent/fluentd/pull/4592 * `fluentd` command: Fix `--plugin` (`-p`) option not to overwrite default value @@ -15,6 +13,8 @@ ### Misc +* out_file: Add warn message for symlink_path setting + https://github.com/fluent/fluentd/pull/4512 * Keep console gem v1.23 to avoid LoadError https://github.com/fluent/fluentd/pull/4510 From a0ede80f9452a86946b840c971d2f981ee68d0db Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Tue, 20 Aug 2024 16:37:01 +0900 Subject: [PATCH 33/39] github: backport suppress running CI only for documentation (#4608) For updating documentation or release task, running GitHub Actions is waste of resource. https://docs.github.com/en/actions/writing-workflows/workflow-syntax-for-github-actions#onpushpull_requestpull_request_targetpathspaths-ignore Signed-off-by: Kentaro Hayashi --- .github/workflows/test.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 666aa97b86..2869d32c6d 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -3,8 +3,14 @@ name: Test on: push: branches: [v1.16] + paths-ignore: + - '*.md' + - 'lib/fluent/version.rb' pull_request: branches: [v1.16] + paths-ignore: + - '*.md' + - 'lib/fluent/version.rb' jobs: test: From c1dbd99243d8229f72c344ff42062eab24c60a84 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Thu, 26 Dec 2024 13:54:47 +0900 Subject: [PATCH 34/39] Backport(v1.16): tests: use never instead of dont_allow (#4671) (#4723) Backported from a2b935ae2bc4b4d43e5adddbec01092ea4228b9e (#4671). Signed-off-by: Shizuo Fujita Co-authored-by: Watson --- test/plugin/out_forward/test_socket_cache.rb | 6 +++--- test/test_event_router.rb | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/test/plugin/out_forward/test_socket_cache.rb b/test/plugin/out_forward/test_socket_cache.rb index f7dec5a0f1..1d584d6b4e 100644 --- a/test/plugin/out_forward/test_socket_cache.rb +++ b/test/plugin/out_forward/test_socket_cache.rb @@ -17,7 +17,7 @@ class SocketCacheTest < Test::Unit::TestCase assert_equal(socket, c.checkout_or('key') { socket }) c.checkin(socket) - sock = dont_allow(mock!).open + sock = mock!.open.never.subject assert_equal(socket, c.checkout_or('key') { sock.open }) end @@ -130,7 +130,7 @@ def teardown c = Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log) sock = mock!.close { 'closed' }.subject - sock2 = dont_allow(mock!).close + sock2 = mock!.close.never.subject stub(sock).inspect stub(sock2).inspect @@ -154,7 +154,7 @@ def teardown Timecop.freeze(Time.parse('2016-04-13 14:00:00 +0900')) c = Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log) - sock = dont_allow(mock!).close + sock = mock!.close.never.subject stub(sock).inspect c.checkout_or('key') { sock } diff --git a/test/test_event_router.rb b/test/test_event_router.rb index 3601ddf10f..fda89db230 100644 --- a/test/test_event_router.rb +++ b/test/test_event_router.rb @@ -175,7 +175,7 @@ def event_router test "don't call default collector when tag matched" do event_router.add_rule('test', output) assert_rr do - dont_allow(default_collector).emit_events('test', is_a(OneEventStream)) + mock(default_collector).emit_events('test', is_a(OneEventStream)).never event_router.emit('test', Engine.now, 'k' => 'v') end # check emit handler doesn't catch rr error @@ -201,7 +201,7 @@ def filter_stream(_tag, es); end event_router.add_rule('test', filter) assert_rr do - dont_allow(filter).filter_stream('test', is_a(OneEventStream)) { events } + mock(filter).filter_stream('test', is_a(OneEventStream)).never event_router.emit('foo', Engine.now, 'k' => 'v') end end From 0c5b764e76a219156f3bb182ef416999904fbabf Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 29 Jan 2025 11:05:12 +0900 Subject: [PATCH 35/39] Backport(v1.16) tests: fix unused_port (PR#4675) (#4788) **Which issue(s) this PR fixes**: Fixes https://github.com/fluent/fluentd/issues/4674 **What this PR does / why we need it**: It obtains unused port number for TCP by unused_port method and the number has been used in UDP. And that number may be already used by UDP sockets. This patch will obtain and use unused ports appropriately for each protocol. Backported from https://github.com/fluent/fluentd/pull/4675 **Docs Changes**: **Release Note**: Signed-off-by: Watson Signed-off-by: Kentaro Hayashi Co-authored-by: Watson --- test/command/test_cat.rb | 2 +- test/command/test_fluentd.rb | 2 +- test/helper.rb | 34 ++++-- test/plugin/test_in_forward.rb | 3 +- test/plugin/test_in_http.rb | 2 +- test/plugin/test_in_monitor_agent.rb | 12 +- test/plugin/test_in_syslog.rb | 43 ++++--- test/plugin/test_in_tcp.rb | 2 +- test/plugin/test_in_udp.rb | 2 +- test/plugin/test_out_forward.rb | 3 +- test/plugin/test_out_stream.rb | 2 +- test/plugin_helper/test_http_server_helper.rb | 2 +- test/plugin_helper/test_server.rb | 105 +++++++++++------- test/plugin_helper/test_socket.rb | 2 +- 14 files changed, 134 insertions(+), 82 deletions(-) diff --git a/test/command/test_cat.rb b/test/command/test_cat.rb index 51c1a20bbb..f1e1067823 100644 --- a/test/command/test_cat.rb +++ b/test/command/test_cat.rb @@ -18,7 +18,7 @@ def setup @primary = create_primary metadata = @primary.buffer.new_metadata @chunk = create_chunk(@primary, metadata, @es) - @port = unused_port + @port = unused_port(protocol: :tcp) end def teardown diff --git a/test/command/test_fluentd.rb b/test/command/test_fluentd.rb index d73f0dd1af..c51f7f9ad4 100644 --- a/test/command/test_fluentd.rb +++ b/test/command/test_fluentd.rb @@ -1175,7 +1175,7 @@ def multi_workers_ready?; true; end end end - sub_test_case 'sahred socket options' do + sub_test_case 'shared socket options' do test 'enable shared socket by default' do conf = "" conf_path = create_conf_file('empty.conf', conf) diff --git a/test/helper.rb b/test/helper.rb index 9dc8633423..4e9f028a9c 100644 --- a/test/helper.rb +++ b/test/helper.rb @@ -71,17 +71,31 @@ class Test::Unit::AssertionFailedError < StandardError include Fluent::Test::Helpers -def unused_port(num = 1, protocol: :tcp, bind: "0.0.0.0") +def unused_port(num = 1, protocol:, bind: "0.0.0.0") case protocol - when :tcp + when :tcp, :tls unused_port_tcp(num) when :udp unused_port_udp(num, bind: bind) + when :all + unused_port_tcp_udp(num) else raise ArgumentError, "unknown protocol: #{protocol}" end end +def unused_port_tcp_udp(num = 1) + raise "not support num > 1" if num > 1 + + # The default maximum number of file descriptors in macOS is 256. + # It might need to set num to a smaller value than that. + tcp_ports = unused_port_tcp(200) + port = unused_port_udp(1, port_list: tcp_ports) + raise "can't find unused port" unless port + + port +end + def unused_port_tcp(num = 1) ports = [] sockets = [] @@ -90,7 +104,7 @@ def unused_port_tcp(num = 1) sockets << s ports << s.addr[1] end - sockets.each{|s| s.close } + sockets.each(&:close) if num == 1 return ports.first else @@ -100,12 +114,15 @@ def unused_port_tcp(num = 1) PORT_RANGE_AVAILABLE = (1024...65535) -def unused_port_udp(num = 1, bind: "0.0.0.0") +def unused_port_udp(num = 1, port_list: [], bind: "0.0.0.0") family = IPAddr.new(IPSocket.getaddress(bind)).ipv4? ? ::Socket::AF_INET : ::Socket::AF_INET6 ports = [] sockets = [] - while ports.size < num - port = rand(PORT_RANGE_AVAILABLE) + + use_random_port = port_list.empty? + i = 0 + loop do + port = use_random_port ? rand(PORT_RANGE_AVAILABLE) : port_list[i] u = UDPSocket.new(family) if (u.bind(bind, port) rescue nil) ports << port @@ -113,8 +130,11 @@ def unused_port_udp(num = 1, bind: "0.0.0.0") else u.close end + i += 1 + break if ports.size >= num + break if !use_random_port && i >= port_list.size end - sockets.each{|s| s.close } + sockets.each(&:close) if num == 1 return ports.first else diff --git a/test/plugin/test_in_forward.rb b/test/plugin/test_in_forward.rb index bce0cdc857..63ac8d112c 100644 --- a/test/plugin/test_in_forward.rb +++ b/test/plugin/test_in_forward.rb @@ -18,7 +18,8 @@ def setup Fluent::Test.setup @responses = [] # for testing responses after sending data @d = nil - @port = unused_port + # forward plugin uses TCP and UDP sockets on the same port number + @port = unused_port(protocol: :all) end def teardown diff --git a/test/plugin/test_in_http.rb b/test/plugin/test_in_http.rb index e66845520a..13cb196241 100644 --- a/test/plugin/test_in_http.rb +++ b/test/plugin/test_in_http.rb @@ -18,7 +18,7 @@ def shutdown def setup Fluent::Test.setup - @port = unused_port + @port = unused_port(protocol: :tcp) end def teardown diff --git a/test/plugin/test_in_monitor_agent.rb b/test/plugin/test_in_monitor_agent.rb index 76e0a26a18..eb8d2dea7c 100644 --- a/test/plugin/test_in_monitor_agent.rb +++ b/test/plugin/test_in_monitor_agent.rb @@ -392,7 +392,7 @@ def test_enable_input_metrics(with_config) end test "emit" do - port = unused_port + port = unused_port(protocol: :tcp) d = create_driver(" @type monitor_agent bind '127.0.0.1' @@ -451,7 +451,7 @@ def get(uri, header = {}) sub_test_case "servlets" do setup do - @port = unused_port + @port = unused_port(protocol: :tcp) # check @type and type in one configuration conf = <<-EOC @@ -759,7 +759,7 @@ def write(chunk) end setup do - @port = unused_port + @port = unused_port(protocol: :tcp) # check @type and type in one configuration conf = <<-EOC @@ -840,7 +840,7 @@ def write(chunk) sub_test_case "check the port number of http server" do test "on single worker environment" do - port = unused_port + port = unused_port(protocol: :tcp) d = create_driver(" @type monitor_agent bind '127.0.0.1' @@ -851,7 +851,7 @@ def write(chunk) end test "worker_id = 2 on multi worker environment" do - port = unused_port + port = unused_port(protocol: :tcp) Fluent::SystemConfig.overwrite_system_config('workers' => 4) do d = Fluent::Test::Driver::Input.new(Fluent::Plugin::MonitorAgentInput) d.instance.instance_eval{ @_fluentd_worker_id = 2 } @@ -905,7 +905,7 @@ def filter(tag, time, record) end test "plugins have a variable named buffer does not throws NoMethodError" do - port = unused_port + port = unused_port(protocol: :tcp) d = create_driver(" @type monitor_agent bind '127.0.0.1' diff --git a/test/plugin/test_in_syslog.rb b/test/plugin/test_in_syslog.rb index f715e1fef9..868e77bb04 100755 --- a/test/plugin/test_in_syslog.rb +++ b/test/plugin/test_in_syslog.rb @@ -5,24 +5,24 @@ class SyslogInputTest < Test::Unit::TestCase def setup Fluent::Test.setup - @port = unused_port + @port = unused_port(protocol: :udp) end def teardown @port = nil end - def ipv4_config + def ipv4_config(port = @port) %[ - port #{@port} + port #{port} bind 127.0.0.1 tag syslog ] end - def ipv6_config + def ipv6_config(port = @port) %[ - port #{@port} + port #{port} bind ::1 tag syslog ] @@ -69,7 +69,8 @@ def test_configure_resolve_hostname(param) 'Use transport and protocol' => ["protocol_type udp\n\n ", :udp, :tcp]) def test_configure_protocol(param) conf, proto_type, transport_proto_type = *param - d = create_driver([ipv4_config, conf].join("\n")) + port = unused_port(protocol: proto_type ? proto_type : transport_proto_type) + d = create_driver([ipv4_config(port), conf].join("\n")) assert_equal(d.instance.protocol_type, proto_type) assert_equal(d.instance.transport_config.protocol, transport_proto_type) @@ -158,12 +159,13 @@ def test_msg_size_udp_for_large_msg end def test_msg_size_with_tcp - d = create_driver([ipv4_config, " \n"].join("\n")) + port = unused_port(protocol: :tcp) + d = create_driver([ipv4_config(port), " \n"].join("\n")) tests = create_test_case d.run(expect_emits: 2) do tests.each {|test| - TCPSocket.open('127.0.0.1', @port) do |s| + TCPSocket.open('127.0.0.1', port) do |s| s.send(test['msg'], 0) end } @@ -189,11 +191,12 @@ def test_emit_rfc5452 end def test_msg_size_with_same_tcp_connection - d = create_driver([ipv4_config, " \n"].join("\n")) + port = unused_port(protocol: :tcp) + d = create_driver([ipv4_config(port), " \n"].join("\n")) tests = create_test_case d.run(expect_emits: 2) do - TCPSocket.open('127.0.0.1', @port) do |s| + TCPSocket.open('127.0.0.1', port) do |s| tests.each {|test| s.send(test['msg'], 0) } @@ -347,12 +350,13 @@ def compare_test_result(events, tests, options = {}) sub_test_case 'octet counting frame' do def test_msg_size_with_tcp - d = create_driver([ipv4_config, " \n", 'frame_type octet_count'].join("\n")) + port = unused_port(protocol: :tcp) + d = create_driver([ipv4_config(port), " \n", 'frame_type octet_count'].join("\n")) tests = create_test_case d.run(expect_emits: 2) do tests.each {|test| - TCPSocket.open('127.0.0.1', @port) do |s| + TCPSocket.open('127.0.0.1', port) do |s| s.send(test['msg'], 0) end } @@ -363,11 +367,12 @@ def test_msg_size_with_tcp end def test_msg_size_with_same_tcp_connection - d = create_driver([ipv4_config, " \n", 'frame_type octet_count'].join("\n")) + port = unused_port(protocol: :tcp) + d = create_driver([ipv4_config(port), " \n", 'frame_type octet_count'].join("\n")) tests = create_test_case d.run(expect_emits: 2) do - TCPSocket.open('127.0.0.1', @port) do |s| + TCPSocket.open('127.0.0.1', port) do |s| tests.each {|test| s.send(test['msg'], 0) } @@ -469,7 +474,8 @@ def test_emit_unmatched_lines_with_address end def test_send_keepalive_packet_is_disabled_by_default - d = create_driver(ipv4_config + %[ + port = unused_port(protocol: :tcp) + d = create_driver(ipv4_config(port) + %[ protocol tcp @@ -479,19 +485,20 @@ def test_send_keepalive_packet_is_disabled_by_default def test_send_keepalive_packet_can_be_enabled addr = "127.0.0.1" - d = create_driver(ipv4_config + %[ + port = unused_port(protocol: :tcp) + d = create_driver(ipv4_config(port) + %[ send_keepalive_packet true ]) assert_true d.instance.send_keepalive_packet mock.proxy(d.instance).server_create_connection( - :in_syslog_tcp_server, @port, + :in_syslog_tcp_server, port, bind: addr, resolve_name: nil, send_keepalive_packet: true) d.run do - TCPSocket.open(addr, @port) + TCPSocket.open(addr, port) end end diff --git a/test/plugin/test_in_tcp.rb b/test/plugin/test_in_tcp.rb index c1d917332e..a27e00bba3 100755 --- a/test/plugin/test_in_tcp.rb +++ b/test/plugin/test_in_tcp.rb @@ -5,7 +5,7 @@ class TcpInputTest < Test::Unit::TestCase def setup Fluent::Test.setup - @port = unused_port + @port = unused_port(protocol: :tcp) end def teardown diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb index dcc63e1f8f..1f1bb457cb 100755 --- a/test/plugin/test_in_udp.rb +++ b/test/plugin/test_in_udp.rb @@ -5,7 +5,7 @@ class UdpInputTest < Test::Unit::TestCase def setup Fluent::Test.setup - @port = unused_port + @port = unused_port(protocol: :udp) end def teardown diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb index 438caf8fa2..306d9d8234 100644 --- a/test/plugin/test_out_forward.rb +++ b/test/plugin/test_out_forward.rb @@ -12,7 +12,8 @@ def setup FileUtils.rm_rf(TMP_DIR) FileUtils.mkdir_p(TMP_DIR) @d = nil - @target_port = unused_port + # forward plugin uses TCP and UDP sockets on the same port number + @target_port = unused_port(protocol: :all) end def teardown diff --git a/test/plugin/test_out_stream.rb b/test/plugin/test_out_stream.rb index 05e9011aab..f19328f5be 100644 --- a/test/plugin/test_out_stream.rb +++ b/test/plugin/test_out_stream.rb @@ -54,7 +54,7 @@ class TcpOutputTest < Test::Unit::TestCase def setup super - @port = unused_port + @port = unused_port(protocol: :tcp) end def teardown diff --git a/test/plugin_helper/test_http_server_helper.rb b/test/plugin_helper/test_http_server_helper.rb index 4fd044bb0e..815f3b18fd 100644 --- a/test/plugin_helper/test_http_server_helper.rb +++ b/test/plugin_helper/test_http_server_helper.rb @@ -14,7 +14,7 @@ class HttpHelperTest < Test::Unit::TestCase CERT_CA_DIR = File.expand_path(File.dirname(__FILE__) + '/data/cert/with_ca') def setup - @port = unused_port + @port = unused_port(protocol: :tcp) end def teardown diff --git a/test/plugin_helper/test_server.rb b/test/plugin_helper/test_server.rb index 347938eca3..63beb0bd70 100644 --- a/test/plugin_helper/test_server.rb +++ b/test/plugin_helper/test_server.rb @@ -15,7 +15,7 @@ class Dummy < Fluent::Plugin::TestBase TMP_DIR = File.expand_path(File.dirname(__FILE__) + "/../tmp/plugin_helper_server") setup do - @port = unused_port + @port = unused_port(protocol: :tcp) if Fluent.windows? @socket_manager_server = ServerEngine::SocketManager::Server.open @socket_manager_path = @socket_manager_server.path @@ -233,11 +233,12 @@ class Dummy < Fluent::Plugin::TestBase # 'server_create_connection tcp' => [:server_create_connection, :unix], ) test 'raise error if udp options specified for tcp/tls/unix' do |(m, proto)| + port = unused_port(protocol: proto) assert_raise ArgumentError do - @d.__send__(m, :myserver, @port, proto: proto, max_bytes: 128){|x| x } + @d.__send__(m, :myserver, port, proto: proto, max_bytes: 128){|x| x } end assert_raise ArgumentError do - @d.__send__(m, :myserver, @port, proto: proto, flags: 1){|x| x } + @d.__send__(m, :myserver, port, proto: proto, flags: 1){|x| x } end end @@ -245,8 +246,9 @@ class Dummy < Fluent::Plugin::TestBase 'server_create udp' => [:server_create, :udp], ) test 'raise error if tcp/tls options specified for udp' do |(m, proto)| + port = unused_port(protocol: proto) assert_raise(ArgumentError.new("BUG: linger_timeout is available for tcp/tls")) do - @d.__send__(m, :myserver, @port, proto: proto, linger_timeout: 1, max_bytes: 128){|x| x } + @d.__send__(m, :myserver, port, proto: proto, linger_timeout: 1, max_bytes: 128){|x| x } end end @@ -254,8 +256,9 @@ class Dummy < Fluent::Plugin::TestBase 'server_create udp' => [:server_create, :udp], ) test 'raise error if tcp/tls/unix backlog options specified for udp' do |(m, proto)| + port = unused_port(protocol: proto) assert_raise(ArgumentError.new("BUG: backlog is available for tcp/tls")) do - @d.__send__(m, :myserver, @port, proto: proto, backlog: 500){|x| x } + @d.__send__(m, :myserver, port, proto: proto, backlog: 500){|x| x } end end @@ -263,8 +266,9 @@ class Dummy < Fluent::Plugin::TestBase 'server_create udp' => [:server_create, :udp], ) test 'raise error if tcp/tls send_keepalive_packet option is specified for udp' do |(m, proto)| + port = unused_port(protocol: proto) assert_raise(ArgumentError.new("BUG: send_keepalive_packet is available for tcp/tls")) do - @d.__send__(m, :myserver, @port, proto: proto, send_keepalive_packet: true){|x| x } + @d.__send__(m, :myserver, port, proto: proto, send_keepalive_packet: true){|x| x } end end @@ -276,8 +280,9 @@ class Dummy < Fluent::Plugin::TestBase # 'server_create_connection unix' => [:server_create_connection, :unix, {}], ) test 'raise error if tls options specified for tcp/udp/unix' do |(m, proto, kwargs)| + port = unused_port(protocol: proto) assert_raise(ArgumentError.new("BUG: tls_options is available only for tls")) do - @d.__send__(m, :myserver, @port, proto: proto, tls_options: {}, **kwargs){|x| x } + @d.__send__(m, :myserver, port, proto: proto, tls_options: {}, **kwargs){|x| x } end end @@ -289,7 +294,8 @@ class Dummy < Fluent::Plugin::TestBase 'server_create_connection tls' => [:server_create_connection, :tls, {tls_options: {insecure: true}}], ) test 'can bind specified IPv4 address' do |(m, proto, kwargs)| - @d.__send__(m, :myserver, @port, proto: proto, bind: "127.0.0.1", **kwargs){|x| x } + port = unused_port(protocol: proto) + @d.__send__(m, :myserver, port, proto: proto, bind: "127.0.0.1", **kwargs){|x| x } assert_equal "127.0.0.1", @d._servers.first.bind assert_equal "127.0.0.1", @d._servers.first.server.instance_eval{ instance_variable_defined?(:@listen_socket) ? @listen_socket : @_io }.addr[3] end @@ -303,7 +309,8 @@ class Dummy < Fluent::Plugin::TestBase ) test 'can bind specified IPv6 address' do |(m, proto, kwargs)| # if available omit "IPv6 unavailable here" unless ipv6_enabled? - @d.__send__(m, :myserver, @port, proto: proto, bind: "::1", **kwargs){|x| x } + port = unused_port(protocol: proto) + @d.__send__(m, :myserver, port, proto: proto, bind: "::1", **kwargs){|x| x } assert_equal "::1", @d._servers.first.bind assert_equal "::1", @d._servers.first.server.instance_eval{ instance_variable_defined?(:@listen_socket) ? @listen_socket : @_io }.addr[3] end @@ -320,10 +327,11 @@ class Dummy < Fluent::Plugin::TestBase test 'can create 2 or more servers which share same bind address and port if shared option is true' do |(m, proto, kwargs)| begin d2 = Dummy.new; d2.start; d2.after_start + port = unused_port(protocol: proto) assert_nothing_raised do - @d.__send__(m, :myserver, @port, proto: proto, **kwargs){|x| x } - d2.__send__(m, :myserver, @port, proto: proto, **kwargs){|x| x } + @d.__send__(m, :myserver, port, proto: proto, **kwargs){|x| x } + d2.__send__(m, :myserver, port, proto: proto, **kwargs){|x| x } end ensure d2.stop; d2.before_shutdown; d2.shutdown; d2.after_shutdown; d2.close; d2.terminate @@ -344,12 +352,13 @@ class Dummy < Fluent::Plugin::TestBase test 'cannot create 2 or more servers using same bind address and port if shared option is false' do |(m, proto, kwargs)| begin d2 = Dummy.new; d2.start; d2.after_start + port = unused_port(protocol: proto) assert_nothing_raised do - @d.__send__(m, :myserver, @port, proto: proto, shared: false, **kwargs){|x| x } + @d.__send__(m, :myserver, port, proto: proto, shared: false, **kwargs){|x| x } end assert_raise(Errno::EADDRINUSE, Errno::EACCES) do - d2.__send__(m, :myserver, @port, proto: proto, **kwargs){|x| x } + d2.__send__(m, :myserver, port, proto: proto, **kwargs){|x| x } end ensure d2.stop; d2.before_shutdown; d2.shutdown; d2.after_shutdown; d2.close; d2.terminate @@ -365,16 +374,18 @@ class Dummy < Fluent::Plugin::TestBase # 'unix' => [:unix, {}], ) test 'raise error if block argument is not specified or too many' do |(proto, kwargs)| + port = unused_port(protocol: proto) assert_raise(ArgumentError.new("BUG: block must have 1 or 2 arguments")) do - @d.server_create(:myserver, @port, proto: proto, **kwargs){ 1 } + @d.server_create(:myserver, port, proto: proto, **kwargs){ 1 } end assert_raise(ArgumentError.new("BUG: block must have 1 or 2 arguments")) do - @d.server_create(:myserver, @port, proto: proto, **kwargs){|sock, conn, what_is_this| 1 } + @d.server_create(:myserver, port, proto: proto, **kwargs){|sock, conn, what_is_this| 1 } end end test 'creates udp server if specified in proto' do - @d.server_create(:myserver, @port, proto: :udp, max_bytes: 512){|x| x } + port = unused_port(protocol: :udp) + @d.server_create(:myserver, port, proto: :udp, max_bytes: 512){|x| x } created_server_info = @d._servers.first assert_equal :udp, created_server_info.proto @@ -587,7 +598,8 @@ class Dummy < Fluent::Plugin::TestBase sub_test_case '#server_create_udp' do test 'can accept all keyword arguments valid for udp server' do assert_nothing_raised do - @d.server_create_udp(:s, @port, bind: '127.0.0.1', shared: false, resolve_name: true, max_bytes: 100, flags: 1) do |data, conn| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, bind: '127.0.0.1', shared: false, resolve_name: true, max_bytes: 100, flags: 1) do |data, conn| # ... end end @@ -595,14 +607,15 @@ class Dummy < Fluent::Plugin::TestBase test 'creates a udp server just to read data' do received = "" - @d.server_create_udp(:s, @port, max_bytes: 128) do |data| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, max_bytes: 128) do |data| received << data end bind_port = unused_port(protocol: :udp, bind: "127.0.0.1") 3.times do sock = UDPSocket.new(Socket::AF_INET) sock.bind("127.0.0.1", bind_port) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) sock.puts "yay" sock.puts "foo" sock.close @@ -614,16 +627,17 @@ class Dummy < Fluent::Plugin::TestBase test 'creates a udp server to read and write data' do received = "" responses = [] - @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock| received << data sock.write "ack\n" end - bind_port = unused_port + bind_port = unused_port(protocol: :udp) 3.times do begin sock = UDPSocket.new(Socket::AF_INET) sock.bind("127.0.0.1", bind_port) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) th = Thread.new do while true begin @@ -654,11 +668,12 @@ class Dummy < Fluent::Plugin::TestBase received = "" responses = [] - @d.server_create_udp(:s, @port, bind: "::1", max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, bind: "::1", max_bytes: 128) do |data, sock| received << data sock.write "ack\n" end - bind_port = unused_port + bind_port = unused_port(protocol: :udp) 3.times do begin sock = UDPSocket.new(Socket::AF_INET6) @@ -667,7 +682,7 @@ class Dummy < Fluent::Plugin::TestBase responses << sock.recv(16) true end - sock.connect("::1", @port) + sock.connect("::1", port) sock.write "yay\nfoo\n" th.join(5) ensure @@ -682,13 +697,14 @@ class Dummy < Fluent::Plugin::TestBase test 'does not resolve name of client address in default' do received = "" sources = [] - @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock| received << data sources << sock.remote_host end 3.times do sock = UDPSocket.new(Socket::AF_INET) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) sock.puts "yay" sock.close end @@ -702,13 +718,14 @@ class Dummy < Fluent::Plugin::TestBase received = "" sources = [] - @d.server_create_udp(:s, @port, resolve_name: true, max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, resolve_name: true, max_bytes: 128) do |data, sock| received << data sources << sock.remote_host end 3.times do sock = UDPSocket.new(Socket::AF_INET) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) sock.puts "yay" sock.close end @@ -720,7 +737,8 @@ class Dummy < Fluent::Plugin::TestBase test 'raises error if plugin registers data callback for connection object from #server_create' do received = "" errors = [] - @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock| received << data begin sock.data{|d| received << d.upcase } @@ -729,7 +747,7 @@ class Dummy < Fluent::Plugin::TestBase end end sock = UDPSocket.new(Socket::AF_INET) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) sock.write "foo\n" sock.close @@ -742,7 +760,8 @@ class Dummy < Fluent::Plugin::TestBase test 'raise error if plugin registers write_complete callback for udp' do received = "" errors = [] - @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock| received << data begin sock.on(:write_complete){|conn| "" } @@ -751,7 +770,7 @@ class Dummy < Fluent::Plugin::TestBase end end sock = UDPSocket.new(Socket::AF_INET) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) sock.write "foo\n" sock.close @@ -764,7 +783,8 @@ class Dummy < Fluent::Plugin::TestBase test 'raises error if plugin registers close callback for udp' do received = "" errors = [] - @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock| received << data begin sock.on(:close){|d| "" } @@ -773,7 +793,7 @@ class Dummy < Fluent::Plugin::TestBase end end sock = UDPSocket.new(Socket::AF_INET) - sock.connect("127.0.0.1", @port) + sock.connect("127.0.0.1", port) sock.write "foo\n" sock.close @@ -786,11 +806,12 @@ class Dummy < Fluent::Plugin::TestBase test 'can bind IPv4 / IPv6 together' do omit "IPv6 unavailable here" unless ipv6_enabled? + port = unused_port(protocol: :udp) assert_nothing_raised do - @d.server_create_udp(:s_ipv4_udp, @port, bind: '0.0.0.0', shared: false, max_bytes: 128) do |data, sock| + @d.server_create_udp(:s_ipv4_udp, port, bind: '0.0.0.0', shared: false, max_bytes: 128) do |data, sock| # ... end - @d.server_create_udp(:s_ipv6_udp, @port, bind: '::', shared: false, max_bytes: 128) do |data, sock| + @d.server_create_udp(:s_ipv6_udp, port, bind: '::', shared: false, max_bytes: 128) do |data, sock| # ... end end @@ -803,11 +824,12 @@ class Dummy < Fluent::Plugin::TestBase max_bytes, records, expected = data.values actual_records = [] - @d.server_create_udp(:myserver, @port, max_bytes: max_bytes) do |data, sock| + port = unused_port(protocol: :udp) + @d.server_create_udp(:myserver, port, max_bytes: max_bytes) do |data, sock| actual_records << data end - open_client(:udp, "127.0.0.1", @port) do |sock| + open_client(:udp, "127.0.0.1", port) do |sock| records.each do |record| sock.send(record, 0) end @@ -823,11 +845,12 @@ class Dummy < Fluent::Plugin::TestBase max_bytes, records, expected = data.values actual_records = [] - @d.server_create_udp(:myserver, @port, max_bytes: max_bytes) do |data| + port = unused_port(protocol: :udp) + @d.server_create_udp(:myserver, port, max_bytes: max_bytes) do |data| actual_records << data end - open_client(:udp, "127.0.0.1", @port) do |sock| + open_client(:udp, "127.0.0.1", port) do |sock| records.each do |record| sock.send(record, 0) end diff --git a/test/plugin_helper/test_socket.rb b/test/plugin_helper/test_socket.rb index cb2ff32de0..624169f08b 100644 --- a/test/plugin_helper/test_socket.rb +++ b/test/plugin_helper/test_socket.rb @@ -11,7 +11,7 @@ class SocketHelperTest < Test::Unit::TestCase CERT_CHAINS_DIR = File.expand_path(File.dirname(__FILE__) + '/data/cert/cert_chains') def setup - @port = unused_port + @port = unused_port(protocol: :tcp) end def teardown From 5c08ab6b72579c9493a215eb84a7fbf991ad6e97 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 29 Jan 2025 11:08:46 +0900 Subject: [PATCH 36/39] Backport(v1.16) test_in_udp: add timeout for message_length_limit test (#4676) (#4789) **Which issue(s) this PR fixes**: Fixes # **What this PR does / why we need it**: On Windows, `message_length_limit` test always take 300s for execution. ``` UdpInputTest: test: configure w/o parse section: .: (0.002351) test: configure[ipv4]: .: (0.002723) test: configure[ipv6]: .: (0.002602) test: message size with format[none]: .: (1.029781) test: message size with format[json]: .: (1.113799) test: message size with format[regexp]: .: (1.110006) test: message_length_limit: .: (300.538596) ``` The 300 sec comes from https://github.com/fluent/fluentd/blob/a2b935ae2bc4b4d43e5adddbec01092ea4228b9e/lib/fluent/test/driver/base.rb#L36, and it always times out in Windows. This patch set a short timeout to reduce test execution time on Windows. Backported from https://github.com/fluent/fluentd/pull/4676 **Docs Changes**: **Release Note**: Signed-off-by: Watson Signed-off-by: Kentaro Hayashi Co-authored-by: Watson --- test/plugin/test_in_udp.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb index 1f1bb457cb..847287e87e 100755 --- a/test/plugin/test_in_udp.rb +++ b/test/plugin/test_in_udp.rb @@ -272,7 +272,7 @@ def create_udp_socket(host, port) format none message_length_limit #{message_length_limit} !) - d.run(expect_records: 3) do + d.run(expect_records: 3, timeout: 5) do create_udp_socket('127.0.0.1', @port) do |u| 3.times do |i| u.send("#{i}" * 40 + "\n", 0) From 60c969a0c7f652782acd3b717deef41689f3fba0 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 29 Jan 2025 11:14:43 +0900 Subject: [PATCH 37/39] Backport(v1.16) test_in_udp: Reduce execution time of message_length_limit test (#4682) (#4790) **Which issue(s) this PR fixes**: Backport #4682 **What this PR does / why we need it**: Although it's improved in #4676, it still takes 5 seconds on windows. It can be reduced more by sending data that doesn't exceed the limit. ``` test: message_length_limit: .: (1.104855) ``` **Docs Changes**: **Release Note**: Signed-off-by: Takuro Ashie Co-authored-by: Takuro Ashie --- test/plugin/test_in_udp.rb | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) mode change 100755 => 100644 test/plugin/test_in_udp.rb diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb old mode 100755 new mode 100644 index 847287e87e..34c9c3ac26 --- a/test/plugin/test_in_udp.rb +++ b/test/plugin/test_in_udp.rb @@ -268,25 +268,31 @@ def create_udp_socket(host, port) test 'message_length_limit' do message_length_limit = 32 + + if Fluent.windows? + expected_records = ["0" * 30, "4" * 30] + else + expected_records = 1.upto(3).collect do |i| + "#{i}" * message_length_limit + end + expected_records.prepend("0" * 30) + expected_records.append("4" * 30) + end + d = create_driver(base_config + %! format none message_length_limit #{message_length_limit} !) - d.run(expect_records: 3, timeout: 5) do + d.run(expect_records: expected_records.size, timeout: 5) do create_udp_socket('127.0.0.1', @port) do |u| - 3.times do |i| + u.send("0" * 30 + "\n", 0) + 1.upto(3) do |i| u.send("#{i}" * 40 + "\n", 0) end + u.send("4" * 30 + "\n", 0) end end - if Fluent.windows? - expected_records = [] - else - expected_records = 3.times.collect do |i| - "#{i}" * message_length_limit - end - end actual_records = d.events.collect do |event| event[2]["message"] end From 0b2126a1954457ea838099903a0a6d7387a7f52d Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 29 Jan 2025 11:21:32 +0900 Subject: [PATCH 38/39] Backport(v1.16) test_out_forward: remove unnecessary ack_response_timeout setting (#4685) (#4791) **Which issue(s) this PR fixes**: Backport #4685 **What this PR does / why we need it**: Seems that timeout setting is short in ack_response_timeout. Seems that It may take some time to receive a ACK response so the process in ack handler has expired and the node is disabled. This PR will remove unnecessary ack_response_timeout settings for the test **Docs Changes**: **Release Note**: Signed-off-by: Watson Signed-off-by: Shizuo Fujita Co-authored-by: Watson --- test/plugin/test_out_forward.rb | 2 -- 1 file changed, 2 deletions(-) diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb index 306d9d8234..c2442b631a 100644 --- a/test/plugin/test_out_forward.rb +++ b/test/plugin/test_out_forward.rb @@ -611,7 +611,6 @@ def try_write(chunk) @d = d = create_driver(config + %[ require_ack_response true - ack_response_timeout 1s flush_mode immediate retry_type periodic @@ -659,7 +658,6 @@ def try_write(chunk) @d = d = create_driver(config + %[ require_ack_response true - ack_response_timeout 10s flush_mode immediate retry_type periodic From d13b9df79c0db9dada97018ace8daeb7ab503632 Mon Sep 17 00:00:00 2001 From: Watson Date: Mon, 28 Oct 2024 18:37:00 +0900 Subject: [PATCH 39/39] test_cat: use proper protocol in unused_port method (#4686) Signed-off-by: Watson Co-authored-by: Daijiro Fukuda --- test/command/test_cat.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/command/test_cat.rb b/test/command/test_cat.rb index f1e1067823..40605815bb 100644 --- a/test/command/test_cat.rb +++ b/test/command/test_cat.rb @@ -18,7 +18,7 @@ def setup @primary = create_primary metadata = @primary.buffer.new_metadata @chunk = create_chunk(@primary, metadata, @es) - @port = unused_port(protocol: :tcp) + @port = unused_port(protocol: :all) end def teardown