Apply modifications in pipeline to the records being passed to @ERROR label

[daipom]

Which issue(s) this PR fixes:
Fixes #3465

What this PR does / why we need it:
When an error occurs in the output process in a pipeline,
the modifications made by the filter process should be applied to the records
that are passed to @ERROR label.

Docs Changes:
None.

Release Note:
Same with the title.

I have confirmed all success of bundle exec rake test in my environment, Ubunty 20.04.3, Ruby 2.7.0.

I have confirmed @Error label outputs the records including the modifications made by record_transformer as follows.

Config

<source>
  @type forward
  @id forward_input
</source>

<filter debug.**>
  @type record_transformer
  <record>
    newfield "new value"
  </record>
</filter>

<match debug.**>
  @type stdout
  @id stdout_output
  <buffer>
    @type memory
    total_limit_size 128
  </buffer>
</match>

<label @ERROR>
  <match debug.**>
    @type file
    path "{output_filepath}"
  </match>
</label>

Operation

1. Run echo '{"hoge":"fuga"}' | fluent-cat debug.hoge about 3 or 4 times and generate BufferOverflowError as follows.

2022-02-14 17:40:05 +0900 [warn]: #0 [stdout_output] failed to write data into buffer by buffer overflow action=:throw_exception
2022-02-14 17:40:05 +0900 [warn]: #0 send an error event stream to @ERROR: error_class=Fluent::Plugin::Buffer::BufferOverflowError error="buffer space has too many data" location="/home/daipom/work/fluentd/fluentd/lib/fluent/plugin/buffer.rb:327:in `write'" tag="debug.hoge"

2. Confirm records being outputted by @ERROR label.

2022-02-14T17:40:05+09:00	debug.hoge	{"hoge":"fuga","newfield":"new value"}
2022-02-14T17:40:06+09:00	debug.hoge	{"hoge":"fuga","newfield":"new value"}
2022-02-14T17:40:21+09:00	debug.hoge	{"hoge":"fuga","newfield":"new value"}
2022-02-14T17:40:21+09:00	debug.hoge	{"hoge":"fuga","newfield":"new value"}

[ashie]

The patch seems good.
Could you add tests for it?

[daipom]

Thank you!
Sure, I will.

[daipom]

I have added a test code.

[daipom]

Following tests failed in Ruby 3.1 on macOS and Ruby 3.0.3 on Windows.

• Testing on macOS / Unit testing with Ruby 3.1 on macos-latest

Failure: test: can execute external command many times, which finishes immediately(ChildProcessTest)
/Users/runner/work/fluentd/fluentd/test/plugin_helper/test_child_process.rb:332:in `block (2 levels) in <class:ChildProcessTest>'
     329:       @d.stop
     330:       assert_equal [], @d.log.out.logs
     331:       @d.shutdown; @d.close; @d.terminate
  => 332:       assert_equal 2, ary.size
     333:     end
     334:   end
     335: 
/Users/runner/hostedtoolcache/Ruby/3.1.0/x64/lib/ruby/3.1.0/timeout.rb:107:in `block in timeout'
/Users/runner/hostedtoolcache/Ruby/3.1.0/x64/lib/ruby/3.1.0/timeout.rb:36:in `block in catch'
/Users/runner/hostedtoolcache/Ruby/3.1.0/x64/lib/ruby/3.1.0/timeout.rb:36:in `catch'
/Users/runner/hostedtoolcache/Ruby/3.1.0/x64/lib/ruby/3.1.0/timeout.rb:36:in `catch'
/Users/runner/hostedtoolcache/Ruby/3.1.0/x64/lib/ruby/3.1.0/timeout.rb:123:in `timeout'
/Users/runner/work/fluentd/fluentd/test/plugin_helper/test_child_process.rb:323:in `block in <class:ChildProcessTest>'
<2> expected but was
<3>

• Testing on Windows / Unit testing with Ruby 3.0.3 on windows-latest

test: using json format[with sections]:		E, [2022-02-15T04:44:31.625844 #3584] ERROR -- : unexpected error while shutdown, Errno::EPERM:Operation not permitted
test: using json format[traditional]:			E, [2022-02-15T04:44:33.160880 #3584] ERROR -- : unexpected error while shutdown, Errno::EPERM:Operation not permitted
test: using json format with custom time format[with sections]:	E, [2022-02-15T04:44:34.691648 #3584] ERROR -- : unexpected error while shutdown, Errno::EPERM:Operation not permitted
test: using json format with custom time format[traditional]:	E, [2022-02-15T04:44:36.221443 #3584] ERROR -- : unexpected error while shutdown, Errno::EPERM:Operation not permitted
test: using json format with float time[with sections]:	E, [2022-02-15T04:44:37.753885 #3584] ERROR -- : unexpected error while shutdown, Errno::EPERM:Operation not permitted
test: using json format with float time[traditional]:	E, [2022-02-15T04:44:39.282508 #3584] ERROR -- : unexpected error while shutdown, Errno::EPERM:Operation not permitted

[ashie]

They are known unstable tests, please ignore for now.
(I'll investigate them when I have spare time...)

[ashie]

Memo: When I run the test without the fix, handle_emits_error is called with unmodified record (expected):

Failure: test: can pass records modified by filters to handle_emits_error(EventRouterTest::Fluent::EventRouter::emit_error_handler):
  On subject #<FluentTest::TestEmitErrorHandler:0x00005619e534a698>,
  unexpected method invocation:
    handle_emits_error("test", #<Fluent::OneEventStream:0x00005619e5362e78 @time=2022-02-16 10:26:42.773182005 +0900, @record={"pre_label"=>"pre_value"}>, #<RuntimeError: emit error!>)
  expected invocations:
  - handle_emits_error("test", #<Fluent::OneEventStream:0x00005619e535bc90 @time=2022-02-16 10:26:42.773182005 +0900, @record={"modified_label"=>"modified_value"}>, is_a(RuntimeError))
/home/aho/Projects/Fluentd/fluentd/lib/fluent/event_router.rb:122:in `rescue in emit_stream'
/home/aho/Projects/Fluentd/fluentd/lib/fluent/event_router.rb:114:in `emit_stream'
/home/aho/Projects/Fluentd/fluentd/lib/fluent/event_router.rb:106:in `emit'
/home/aho/Projects/Fluentd/fluentd/test/test_event_router.rb:343:in `block (4 levels) in <class:EventRouterTest>'
     340:         assert_rr do
     341:           stub(filter).filter_stream { modified_es }
     342:           mock(emit_handler).handle_emits_error('test', modified_es, is_a(RuntimeError))
  => 343:           event_router.emit('test', time, 'pre_label' => 'pre_value')
     344:         end
     345:       end
     346:     end
/home/aho/Projects/Fluentd/fluentd/vendor/bundle/ruby/2.7.0/gems/test-unit-rr-1.0.5/lib/test/unit/rr.rb:98:in `assert_rr'
/home/aho/Projects/Fluentd/fluentd/test/test_event_router.rb:340:in `block (3 levels) in <class:EventRouterTest>'

[ashie]

LGTM

[ashie]

Thanks!

[daipom]

Thank you for reviewing and merging!