Records passed to @ERROR label do not contain modifications made in the pipeline/config

[brian-groux-hs]

Describe the bug

In a pipeline where the log records are mutated, when errors occur the log records passed to the @error label do not contain modifications made further up in the pipeline before the error ocurred.

To Reproduce

Steps

1. Run the config below, which adds a field
2. Use fluent cat to send a simple message echo '{"foo":"bar"}' | fluent-cat test.log
3. Wait for the tiny buffer to overflow
4. Inspect the file output from @error and note the log does not contain the injected field

Results
output file contains
2021-07-22T18:03:32+00:00 test.log {"foo":"bar"}

Config*
Provided below

Expected behavior

Ouput should contain modified record, not original

2021-07-22T18:08:19+00:00	test.log	{"foo":"bar","newfield":"new value"}

Your Environment

- Fluentd version: fluentd 1.3.2

- TD Agent version: N/A

- Operating system: Alpine container (as tested) Debian container also exhibits issues. Containers were running locally and on Ubuntu hosts in K8S. Same behaviours. 

- Kernel version: 4.19.76-linuxkit (Alpine container)

Your Configuration

# Receive events from 24224/tcp
# This is used by log forwarding and the fluent-cat command
<source>
  @type forward
  port 24224
</source>


<filter test.log>
  @type record_transformer
  <record>
    newfield "new value"
  </record>
</filter>


<match test.log>
  @type forward
  send_timeout 10s
  recover_wait 5s
  hard_timeout 20s
  retry_timeout 30s

  # Force buffer overflow errors
  <buffer>
    @type memory
    total_limit_size 128
  </buffer>

  <server>
    name deadend-server
    host 192.168.1.3
    port 24224
  </server>
</match>

<label @ERROR>
  <match test.log>
    @type file
    path /fluentd/etc/backup
  </match>
</label>

Your Error Log

N/A. see outputs and expected outputs.

Additional context

No response

[ashie]

Hmm, it's not hard to fix but I'm not sure it should be treated as a bug or a specification.

fluentd/lib/fluent/event_router.rb

Lines 114 to 121 in 28eda36

	def emit_stream(tag, es)
	match(tag).emit_events(tag, es)
	if callback = find_callback
	callback.call(es)
	end
	rescue => e
	@emit_error_handler.handle_emits_error(tag, es, e)
	end

fluentd/lib/fluent/event_router.rb

Lines 179 to 182 in 28eda36

	def emit_events(tag, es)
	processed = @optimizer.filter_stream(tag, es)
	@output.emit_events(tag, processed)
	end

fluentd/lib/fluent/root_agent.rb

Lines 356 to 370 in 28eda36

	def handle_emits_error(tag, es, error)
	error_info = {error: error, location: (error.backtrace ? error.backtrace.first : nil), tag: tag}
	if @error_collector
	log.warn "send an error event stream to @ERROR:", error_info
	@error_collector.emit_stream(tag, es)
	else
	now = Time.now.to_i
	if @suppress_emit_error_log_interval.zero? || now > @next_emit_error_log_time
	log.warn "emit transaction failed:", error_info
	log.warn_backtrace
	@next_emit_error_log_time = now + @suppress_emit_error_log_interval
	end
	raise error
	end
	end

[github-actions]

This issue has been automatically marked as stale because it has been open 90 days with no activity. Remove stale label or comment or this issue will be closed in 30 days