in_emitter: Add backpressure limitation for filesystem storage

[cosmo0920]

Aligned emitter initialization with storage-type semantics by enabling pause on chunk overlimit only for filesystem storage and resetting it when another storage type is in use, keeping behavior consistent with other inputs.

Closes #11279.

---

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.

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

Documentation

• Documentation required for this feature

Backporting

• 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.

Summary by CodeRabbit

• Bug Fixes
  • Improved automatic storage backpressure: pausing is now automatically enabled for filesystem storage when needed, ensuring configured maximum chunk limits are respected and preserving prior behavior for other storage types. Added clearer debug logging for these state transitions.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

Walkthrough

Automatically enable emitter pausing for filesystem storage when storage_pause_on_chunks_overlimit is unset, ensuring storage.max_chunks_up is honored; disable pausing for non-filesystem storage if previously active. Adds storage header and debug logs reflecting these state changes. (≤50 words)

Changes

Cohort / File(s)	Change Summary
Emitter backpressure initialization plugins/in_emitter/emitter.c	Add storage header; in cb_emitter_init read storage.pause_on_chunks_overlimit and, if storage.type is filesystem and the pause property is unset, enable pausing so storage.max_chunks_up is enforced; if storage is not filesystem and pausing was active, disable it; add debug logs.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

• Single-file change but touches emitter-storage interaction and conditional behavior.
• Review attention points:
  • Correct detection of "unset" vs explicit property values.
  • Proper inclusion and use of storage header.
  • Correctness of pausing enable/disable logic and associated debug messages.

Suggested reviewers

• edsiper
• koleini
• fujimotos

Suggested labels

backport to v4.1.x

Poem

🐰 I nibble code where chunks may creep,
When filesystem heaps begin toreep,
I pause the stream, keep memory deep,
A careful hop, a gentle sweep,
Logs whisper calm — now systems sleep. 🥕

Pre-merge checks and finishing touches

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main change: adding backpressure limitation (pause_on_chunks_overlimit) specifically for filesystem storage in the emitter plugin.
Linked Issues check	✅ Passed	The code change directly addresses the core requirement from issue #11279 by automatically enabling pause_on_chunks_overlimit for filesystem storage to prevent unbounded memory growth under backpressure conditions.
Out of Scope Changes check	✅ Passed	All changes are focused and in-scope: adding a storage support header and conditional logic to enable pausing for filesystem storage only, which directly addresses the backpressure limitation objective.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch cosmo0920-add-fs-limit-for-emitter

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

plugins/in_emitter/emitter.c (1)

321-342: Consider logging effective storage.type + prior value for easier field debugging.
The debug logs don’t tell you what storage_type was or whether the flag was previously set by user vs default. Adding the numeric/label value (and old/new) would make issue triage easier, especially for “why did my pause behavior change?” reports.

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 3a501b2 and 523141e.

📒 Files selected for processing (1)

• plugins/in_emitter/emitter.c (1 hunks)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (31)

• GitHub Check: pr-windows-build / call-build-windows-package (Windows 64bit (Arm64), amd64_arm64, -DCMAKE_SYSTEM_NAME=Windows -DCMA...
• GitHub Check: pr-windows-build / call-build-windows-package (Windows 32bit, x86, x86-windows-static, 3.31.6)
• GitHub Check: pr-windows-build / call-build-windows-package (Windows 64bit, x64, x64-windows-static, 3.31.6)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=Off, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_ARROW=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_ADDRESS=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=Off, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_MEMORY=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_UNDEFINED=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_COMPILER_STRICT_POINTER_TYPES=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_THREAD=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_THREAD=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SMALL=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_MEMORY=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_COVERAGE=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_UNDEFINED=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_ADDRESS=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=Off, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SMALL=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=Off, 3.31.6, clang, clang++)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, gcc, g++, ubuntu-22.04, clang-12)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, gcc, g++, ubuntu-24.04, clang-14)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, clang, clang++, ubuntu-22.04, clang-12)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, clang, clang++, ubuntu-24.04, clang-14)
• GitHub Check: pr-compile-centos-7
• GitHub Check: pr-compile-without-cxx (3.31.6)
• GitHub Check: PR - fuzzing test

[coderabbitai]

Actionable comments posted: 0

🧹 Nitpick comments (1)

plugins/in_emitter/emitter.c (1)

301-301: Use const char * for const-correctness.

The variable should be declared as const char * since flb_input_get_property returns a const char * (as shown in src/flb_input.c:775).

Apply this diff:

-    char *pause_prop = NULL;
+    const char *pause_prop = NULL;

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 2ce33db and f979501.

📒 Files selected for processing (1)

• plugins/in_emitter/emitter.c (3 hunks)

🧰 Additional context used

🧠 Learnings (2)

📚 Learning: 2025-08-29T06:25:02.561Z

Learnt from: shadowshot-x
Repo: fluent/fluent-bit PR: 10794
File: tests/internal/aws_compress.c:7-7
Timestamp: 2025-08-29T06:25:02.561Z
Learning: In Fluent Bit, ZSTD (zstandard) compression library is bundled directly in the source tree at `lib/zstd-1.5.7` and is built unconditionally as a static library. Unlike optional external dependencies, ZSTD does not use conditional compilation guards like `FLB_HAVE_ZSTD` and is always available. Headers like `<fluent-bit/flb_zstd.h>` can be included directly without guards.

Applied to files:

• plugins/in_emitter/emitter.c

📚 Learning: 2025-08-31T12:46:11.940Z

Learnt from: ThomasDevoogdt
Repo: fluent/fluent-bit PR: 9277
File: .github/workflows/pr-compile-check.yaml:147-151
Timestamp: 2025-08-31T12:46:11.940Z
Learning: In fluent-bit CMakeLists.txt, the system library preference flags are defined as FLB_PREFER_SYSTEM_LIB_ZSTD and FLB_PREFER_SYSTEM_LIB_KAFKA with the FLB_ prefix.

Applied to files:

• plugins/in_emitter/emitter.c

🧬 Code graph analysis (1)

plugins/in_emitter/emitter.c (1)

src/flb_input.c (1)

• flb_input_get_property (776-780)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (31)

• GitHub Check: pr-windows-build / call-build-windows-package (Windows 64bit, x64, x64-windows-static, 3.31.6)
• GitHub Check: pr-windows-build / call-build-windows-package (Windows 64bit (Arm64), amd64_arm64, -DCMAKE_SYSTEM_NAME=Windows -DCMA...
• GitHub Check: pr-windows-build / call-build-windows-package (Windows 32bit, x86, x86-windows-static, 3.31.6)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_MEMORY=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_ARROW=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_UNDEFINED=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_THREAD=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=Off, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SIMD=Off, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_COMPILER_STRICT_POINTER_TYPES=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_COVERAGE=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_THREAD=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_UNDEFINED=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SMALL=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=Off, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SMALL=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_ADDRESS=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DSANITIZE_ADDRESS=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_SANITIZE_MEMORY=On, 3.31.6, clang, clang++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=On, 3.31.6, gcc, g++)
• GitHub Check: run-ubuntu-unit-tests (-DFLB_JEMALLOC=Off, 3.31.6, clang, clang++)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, clang, clang++, ubuntu-22.04, clang-12)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, gcc, g++, ubuntu-22.04, clang-12)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, gcc, g++, ubuntu-24.04, clang-14)
• GitHub Check: pr-compile-system-libs (-DFLB_PREFER_SYSTEM_LIBS=On, 3.31.6, clang, clang++, ubuntu-24.04, clang-14)
• GitHub Check: pr-compile-without-cxx (3.31.6)
• GitHub Check: pr-compile-centos-7
• GitHub Check: PR - fuzzing test

🔇 Additional comments (2)

plugins/in_emitter/emitter.c (2)

28-28: LGTM: Storage header required for FLB_STORAGE_FS.

The include is necessary to support the storage type check in the initialization logic.

---

323-341: LGTM: Logic correctly respects user configuration while providing sensible defaults.

The implementation properly addresses the backpressure issue by:

• Checking if the user explicitly configured storage.pause_on_chunks_overlimit using flb_input_get_property
• Only applying the default (enabling pause) when the property is not set AND filesystem storage is in use
• Honoring user intent when they explicitly configure the property for any storage type

This ensures storage.max_chunks_up is respected for filesystem storage while avoiding unbounded memory growth under backpressure, directly addressing issue #11279.

Based on learnings from the extensive past review discussion, this approach correctly balances automatic defaults with user configuration precedence.

[khalillilahk]

Hello @cosmo0920,
I’m experiencing a similar issue related to in-memory storage, memory keeps on increasing even though I added mem_buf_limit. Also, it seems that memory isn’t being released even after all files have been shipped and are no longer monitored (I’ve set ignore_older to 10 minutes).
Can you please check it out?
#11282