engine: http_server: Add on-demand flush mechanism

[RaJiska]

Adds a /api/v2/flush route to the HTTP server to handle an on-demand flush request. This would be required for specific cases such as event-driven environments.

The route supports POST and PUT to initiate a flush, as well as GET to retrieve the number of flushes that have been issued so far. Currently an on-demand flush forces chunks in a backoff to be retried regardless of their timer. This may not be appropriate in every cases and could be addressed with a URL parameter if needed.

This PR adds a feature in itself, but is a building block to make fluent-bit working with AWS Lambda extensions, as outlined in #12191.

Following the configuration + valgrind run showcasing the feature:

[SERVICE]
    Flush         30
    Log_Level     info
    HTTP_Server   On
    HTTP_Listen   0.0.0.0
    HTTP_PORT     2020

[INPUT]
    Name    dummy
    Tag     test.dummy
    Dummy   {"message": "on-demand flush test"}
    Rate    1

[OUTPUT]
    Name    stdout
    Match   *
    Format  json_lines

$ docker build --target builder  -t flb-test-build .
[...]
$ docker run --rm -it -p 2020:2020 -v ./fb.conf:/fb.conf flb-test-build bash -c "(apt-get update && apt-get install -y valgrind)>/dev/null && valgrind --leak-check=full --show-leak-kinds=all --track-origins=yes /fluent-bit/bin/fluent-bit -c /fb.conf"
==1== Memcheck, a memory error detector
==1== Copyright (C) 2002-2024, and GNU GPL'd, by Julian Seward et al.
==1== Using Valgrind-3.24.0 and LibVEX; rerun with -h for copyright info
==1== Command: /fluent-bit/bin/fluent-bit -c /fb.conf
==1== 
Fluent Bit v5.1.0
* Copyright (C) 2015-2026 The Fluent Bit Authors
* Fluent Bit is a CNCF graduated project under the Fluent organization
* https://fluentbit.io

______ _                  _    ______ _ _           _____  _____ 
|  ___| |                | |   | ___ (_) |         |  ___||  _  |
| |_  | |_   _  ___ _ __ | |_  | |_/ /_| |_  __   _|___ \ | |/' |
|  _| | | | | |/ _ \ '_ \| __| | ___ \ | __| \ \ / /   \ \|  /| |
| |   | | |_| |  __/ | | | |_  | |_/ / | |_   \ V //\__/ /\ |_/ /
\_|   |_|\__,_|\___|_| |_|\__| \____/|_|\__|   \_/ \____(_)\___/


[2026/08/01 08:29:38.570] [ info] [fluent bit] version=5.1.0, commit=, pid=1
[2026/08/01 08:29:38.613] [ info] [storage] ver=1.5.4, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2026/08/01 08:29:38.614] [ info] [simd    ] SSE2
[2026/08/01 08:29:38.614] [ info] [cmetrics] version=2.2.1
[2026/08/01 08:29:38.615] [ info] [ctraces ] version=0.7.1
[2026/08/01 08:29:38.631] [ info] [input:dummy:dummy.0] initializing
[2026/08/01 08:29:38.631] [ info] [input:dummy:dummy.0] storage_strategy='memory' (memory only)
[2026/08/01 08:29:38.779] [ info] [output:stdout:stdout.0] worker #0 started
[2026/08/01 08:29:38.781] [ info] [http_server] listen iface=0.0.0.0 tcp_port=2020
[2026/08/01 08:29:38.782] [ info] [sp] stream processor started
[2026/08/01 08:29:38.784] [ info] [engine] Shutdown Grace Period=5, Shutdown Input Grace Period=2
==1== Warning: client switching stacks?  SP change: 0x87fd558 --> 0x9038fc0
==1==          to suppress, use: --max-stackframe=8632936 or greater
==1== Warning: client switching stacks?  SP change: 0x9038f58 --> 0x87fd558
==1==          to suppress, use: --max-stackframe=8632832 or greater
==1== Warning: client switching stacks?  SP change: 0x87fd558 --> 0x9038f58
==1==          to suppress, use: --max-stackframe=8632832 or greater
==1==          further instances of this message will not be shown.
# First flush-now triggered
{"date":1785572979.422088,"message":"on-demand flush test"}
{"date":1785572980.409847,"message":"on-demand flush test"}
{"date":1785572981.409755,"message":"on-demand flush test"}
{"date":1785572982.409762,"message":"on-demand flush test"}
{"date":1785572983.421976,"message":"on-demand flush test"}
{"date":1785572984.409824,"message":"on-demand flush test"}
{"date":1785572985.409862,"message":"on-demand flush test"}
# Second flush-now triggered
{"date":1785572986.409839,"message":"on-demand flush test"}
{"date":1785572987.421188,"message":"on-demand flush test"}
{"date":1785572988.425044,"message":"on-demand flush test"}
{"date":1785572989.425176,"message":"on-demand flush test"}
{"date":1785572990.409848,"message":"on-demand flush test"}
{"date":1785572991.410025,"message":"on-demand flush test"}
{"date":1785572992.409828,"message":"on-demand flush test"}
{"date":1785572993.41025,"message":"on-demand flush test"}
# Regular flush from clock
{"date":1785572994.410283,"message":"on-demand flush test"}
{"date":1785572995.409827,"message":"on-demand flush test"}
{"date":1785572996.409802,"message":"on-demand flush test"}
{"date":1785572997.409989,"message":"on-demand flush test"}
{"date":1785572998.411321,"message":"on-demand flush test"}
{"date":1785572999.409794,"message":"on-demand flush test"}
{"date":1785573000.410043,"message":"on-demand flush test"}
{"date":1785573001.409791,"message":"on-demand flush test"}
{"date":1785573002.409835,"message":"on-demand flush test"}
{"date":1785573003.41133,"message":"on-demand flush test"}
{"date":1785573004.409793,"message":"on-demand flush test"}
{"date":1785573005.423899,"message":"on-demand flush test"}
{"date":1785573006.430092,"message":"on-demand flush test"}
{"date":1785573007.429388,"message":"on-demand flush test"}
^C[2026/08/01 08:30:11] [engine] caught signal (SIGINT)
{"date":1785573008.423449,"message":"on-demand flush test"}
{"date":1785573009.423997,"message":"on-demand flush test"}
{"date":1785573010.423483,"message":"on-demand flush test"}
[2026/08/01 08:30:11.151] [ warn] [engine] service will shutdown in max 5 seconds
[2026/08/01 08:30:11.153] [ info] [engine] pausing all inputs..
[2026/08/01 08:30:11.155] [ info] [input] pausing dummy.0
[2026/08/01 08:30:11.423] [ info] [engine] service has stopped (0 pending tasks)
[2026/08/01 08:30:11.424] [ info] [input] pausing dummy.0
[2026/08/01 08:30:11.426] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2026/08/01 08:30:11.431] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==1== 
==1== HEAP SUMMARY:
==1==     in use at exit: 0 bytes in 0 blocks
==1==   total heap usage: 82,045 allocs, 82,045 frees, 11,884,906 bytes allocated
==1== 
==1== All heap blocks were freed -- no leaks are possible
==1== 
==1== For lists of detected and suppressed errors, rerun with: -s
==1== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

In another terminal issuing flush commands:

$ date -u && curl localhost:2020/api/v2/flush -d ''
Sat 01 Aug 2026 08:29:45 AM UTC
{"flush":"done","flush_now_count":1}
$ date -u && curl localhost:2020/api/v2/flush -d ''
Sat 01 Aug 2026 08:29:53 AM UTC
{"flush":"done","flush_now_count":2}

---

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

• Documentation required for this feature

fluent/fluent-bit-docs#2642

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.

Summary by CodeRabbit

New Features

• Added an HTTP API for triggering immediate flushes and checking flush status.
• Added a public engine API for requesting immediate flushes.
• POST and PUT responses include completion status and flush counts; GET returns the current count.

Bug Fixes

• Buffered records can now be dispatched without waiting for the scheduled interval.
• Added clear responses for timeouts, unsupported methods, and processing errors.

Tests

• Added runtime coverage for immediate delivery after a flush request.

[coderabbitai]

📝 Walkthrough

Walkthrough

The change adds an engine API for on-demand flushing, tracks completed flush requests, and exposes /api/v2/flush for execution and status queries. A runtime test verifies immediate delivery without waiting for the periodic flush timer.

Changes

On-demand flush

Layer / File(s)	Summary
Engine flush request flow include/fluent-bit/flb_config.h, include/fluent-bit/flb_engine.h, include/fluent-bit/flb_engine_macros.h, src/flb_engine.c	Defines the flush counter, engine event, return value, and flb_engine_flush_request() API. The engine processes the event and increments the counter.
API v2 flush endpoint src/http_server/api/v2/flush.h, src/http_server/api/v2/flush.c, src/http_server/api/v2/register.c, src/http_server/api/v2/CMakeLists.txt	Adds /api/v2/flush. POST and PUT request a flush and report completion or timeout. GET returns the current counter. Unsupported methods return HTTP 405.
Runtime flush validation tests/runtime/core_engine_flush_now.c, tests/runtime/CMakeLists.txt	Adds a runtime test that verifies immediate buffered-record delivery after an explicit flush request.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Suggested reviewers: edsiper

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant APIv2Flush
  participant EngineManager
  participant flb_config

  Client->>APIv2Flush: POST or PUT /api/v2/flush
  APIv2Flush->>EngineManager: flb_engine_flush_request()
  EngineManager->>flb_config: flush pending input tasks
  EngineManager->>flb_config: increment flush_now_count
  APIv2Flush-->>Client: JSON status and counter

Loading

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 20.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely describes the on-demand flush mechanism added through the engine and HTTP server.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

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

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 51c9b4e4ad

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@include/fluent-bit/flb_config.h`:
- Around line 73-77: Make flush_now_count in struct flb_config an atomic counter
and use matching atomic operations for cross-thread access: replace the
increment in src/flb_engine.c lines 708-714, and use atomic loads for baseline,
the wait_for_flush_ack loop, and the final response value in
src/http_server/api/v2/flush.c lines 33-112; update the declaration in
include/fluent-bit/flb_config.h lines 73-77 accordingly.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 319c340c-05f0-47d0-b9c9-ef0fcdffbcaf

📥 Commits

Reviewing files that changed from the base of the PR and between f724311 and 51c9b4e.

📒 Files selected for processing (10)

• include/fluent-bit/flb_config.h
• include/fluent-bit/flb_engine.h
• include/fluent-bit/flb_engine_macros.h
• src/flb_engine.c
• src/http_server/api/v2/CMakeLists.txt
• src/http_server/api/v2/flush.c
• src/http_server/api/v2/flush.h
• src/http_server/api/v2/register.c
• tests/runtime/CMakeLists.txt
• tests/runtime/core_engine_flush_now.c

[cosmo0920]

Thanks for working on this. The endpoint is a useful building block for event-driven environments, including Lambda, but it does not yet provide a deterministic drain guarantee.

A successful POST /api/v2/flush currently means that the engine accepted the request and dispatched available chunks. It does not mean that output operations completed or that the destination received the records. In addition, one dispatch pass can leave chunks waiting because of task limits, non-multiplexed outputs, retries, or data still held in pre-chunk input buffers. Therefore, the proposed Lambda flow of calling flush once and then only polling storage metrics can stall or report an incomplete result.

I suggest:

• Keep the feature documented as an on-demand dispatch operation, not a completed flush.
• For Lambda, repeatedly trigger dispatch while checking drain state, or introduce a request-scoped drain operation that continues until chunks and tasks reach a terminal state.
• Ensure the Lambda adapter establishes an ingestion barrier before flushing, so the final Telemetry API batch has reached Fluent Bit chunks.
• Consider making forced retry rescheduling optional. Forcing every retry on every invocation defeats backoff and can consume retry attempts quickly.
• Avoid using the global counter as a request-specific acknowledgement; concurrent requests can acknowledge one another.
• Add HTTP-level runtime coverage, including multiple chunks, retries/slow outputs, concurrent requests, and timeout responses. The current test covers only the direct engine API with one immediately successful record.

For the Lambda lifecycle specifically, Fluent Bit should remain alive between invocations because Lambda normally freezes and later reuses the environment. On an actual SHUTDOWN event, the extension must stop Fluent Bit gracefully within Lambda’s two-second external-extension shutdown window; this endpoint does not perform that cleanup by itself.

Finally, the PR is not currently merge-ready because Commit Prefix Lint rejects commit e4ed85c using engine: http_server:; the checker expects either engine: or http_server:.

Overall, I support this as a generic on-demand flush/dispatch primitive, but it should not be considered sufficient to close the Lambda-support issue without the lifecycle adapter and a reliable completion mechanism.

Plus, can we add integration tests for confirming this behavior? It would be nice to have to test this feature.