in_emitter: Fix single record chunks and respect mem_buf_limit pause

[drbugfinder-work]

This PR covers two issues:

1. This PR corrects the behavior of a reached mem_buf_limit. The plugin used to accept further records even after being paused which leads to Potential log loss during high load at Multiline & Rewrite Tag Filter (in_emitter) #8198

2. The current in_emitter implementation results in only one record per chunk being created, which is suboptimal. This PR fixes the collector handling:

Please refer to line:

fluent-bit/plugins/in_emitter/emitter.c

Line 165 in 9652b0d

return do_in_emitter_add_record(ec, in);

To validate this observation, please use:

• fluent-bit.conf https://gist.github.com/drbugfinder-work/e5b28ed2c2247c741695eca7c3add110
• multiline_parsers.conf https://gist.github.com/drbugfinder-work/a9979a730fef75242d5cb4058dc93757
• in_emitter_debug_output.diff https://gist.github.com/drbugfinder-work/f5eb5bf175e9a164cf27b954ba41b188

Check the output:

[2024/02/09 15:58:22] [trace] [filter:multiline:multiline.0 at /Users/root/Development/fluent-bit-in_emitter/fluent-bit/plugins/filter_multiline/ml.c:177] emitting from dummy.0 to dummy.0
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] did not use ring-buffer
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] Number of existing chunks: 0
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] no candidate chunk found, create a new one for tag: dummy.0
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] new chunk created for tag: dummy.0
[2024/02/09 15:58:22] [trace] [filter:multiline:multiline.0 at /Users/root/Development/fluent-bit-in_emitter/fluent-bit/plugins/filter_multiline/ml.c:740] not processing records from the emitter
[2024/02/09 15:58:22] [debug] [input chunk] update output instances with new chunk size diff=205, records=1, input=emitter_for_multiline.0
[2024/02/09 15:58:22] [trace] [filter:multiline:multiline.0 at /Users/root/Development/fluent-bit-in_emitter/fluent-bit/plugins/filter_multiline/ml.c:177] emitting from dummy.0 to dummy.0
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] did not use ring-buffer
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] Number of existing chunks: 0
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] no candidate chunk found, create a new one for tag: dummy.0
[2024/02/09 15:58:22] [debug] [input:emitter:emitter_for_multiline.0] new chunk created for tag: dummy.0
[2024/02/09 15:58:22] [trace] [filter:multiline:multiline.0 at /Users/root/Development/fluent-bit-in_emitter/fluent-bit/plugins/filter_multiline/ml.c:740] not processing records from the emitter

Full logs here:

• output before PR change: https://gist.github.com/drbugfinder-work/d1e216dd89fee32fe6e8239993b3e390
• output after PR change: https://gist.github.com/drbugfinder-work/0c90824d2a947315d3cef5ce4f1f4360

I'm uncertain if this modification is the correct approach to resolve this issue, as it appears that the ring buffer is unused. This PR is partially reverting some parts to older code version.
However, implementing this does address the behavior of only one record being generated per chunk.

Please also see the valgrind output:
https://gist.github.com/drbugfinder-work/83352c59799659db9741f33a22083eaa

---

Other comments, not directly related to this PR:

Additionally, the entire file appears to require restructuring, as some comments are inaccurate, and the DEFAULT_EMITTER_RING_BUFFER_FLUSH_FREQUENCY, which I think should represent a time duration, is instead utilized as a 'buffer size'.

fluent-bit/plugins/in_emitter/emitter.c

Line 245 in 9652b0d

ctx->ring_buffer_size = DEFAULT_EMITTER_RING_BUFFER_FLUSH_FREQUENCY;

The scheduler seems to be empty here, so the ring buffer will never be started (proved by another debug output):

fluent-bit/plugins/in_emitter/emitter.c

Lines 241 to 259 in 9652b0d

	if (scheduler != config->sched &&
	scheduler != NULL &&
	ctx->ring_buffer_size == 0) {
	ctx->ring_buffer_size = DEFAULT_EMITTER_RING_BUFFER_FLUSH_FREQUENCY;
	flb_plg_debug(in,
	"threaded emitter instances require ring_buffer_size"
	" being set, using default value of %u",
	ctx->ring_buffer_size);
	}
	if (ctx->ring_buffer_size > 0) {
	ret = in_emitter_start_ring_buffer(in, ctx);
	if (ret == -1) {
	flb_free(ctx);
	return -1;
	}
	}

---

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

• Example configuration file for the change
• Debug log output from testing the change

• Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

• [N/A] Run local packaging test showing all targets (including any new ones) build.
• [N/A] Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

• [N/A] Documentation required for this feature

Backporting

• [N/A] Backport to latest stable release.

---

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.

[drbugfinder-work]

@edsiper @nokute78 @leonardo-albertovich @pwhelan
Can you please look at this?

[drbugfinder-work]

@edsiper @nokute78 @agup006

I further noticed, that the in_emitter continues to accept records even after reaching the mem_buf_limit and being paused. I added a check if the plugin is paused.

This PR should now fix:
#8198
#4049

[bdeetz]

While rolling fluentbit out in my environment, we ran into this bug. We have built and run fluentbit off of this PR with no sign of failure. It has been moving 10+TB/day of logs for 12 days across 500-1500 collectors. We would very much like to see a fix, if not this one, introduced in order to avoid log loss.

[edsiper]

@drbugfinder-work thanks for triaging and fixing this. In order to merge it, would you please split the commits per component ? ,e.g:

• filter_multiline: ...
• filter_rewrite_tag: ...
• ..

thanks

[drbugfinder-work]

@drbugfinder-work thanks for triaging and fixing this. In order to merge it, would you please split the commits per component ? ,e.g:

• filter_multiline: ...
• filter_rewrite_tag: ...
• ..

thanks

@edsiper Done! I've split the commits according to components.