input: add pause/resume callbacks - #10616
Conversation
edsiper
changed the title
|
note: this PR needs to be based on top of #10790 (new downstream pause/resume functionality) |
2d4f20b to
c123f0e
Compare
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughChangesThe HTTP server now supports pause/resume lifecycle control through downstream connection management. Supported HTTP input plugins register lifecycle callbacks, and integration tests cover protocols, worker modes, connection teardown, metrics, shutdown behavior, and Valgrind execution. ChangesHTTP server pause and resume
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant InputPlugin
participant HTTPServer
participant Downstream
participant MetricsAPI
Client->>InputPlugin: Send pause-triggering request
InputPlugin->>HTTPServer: Pause HTTP server
HTTPServer->>Downstream: Pause and release connections
MetricsAPI-->>Client: Report paused input state
Client->>InputPlugin: Send request while paused
InputPlugin-->>Client: Close or reject connection
Client->>InputPlugin: Send resume request
InputPlugin->>HTTPServer: Resume HTTP server
HTTPServer->>Downstream: Reap stale sessions and resume
MetricsAPI-->>Client: Report resumed input state
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
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 `@tests/integration/src/utils/input_pause_resume.py`:
- Around line 61-84: Update the pause/resume test around
wait_for_input_pause_state to establish and retain a client socket or session
before triggering the pause, then verify that the already-active connection is
closed after pause becomes active. Preserve the existing paused-state request
assertion so both active-session teardown and rejection of new requests remain
covered.
🪄 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: 6ecd7cd5-20e2-471b-98c0-f14e44767f70
📒 Files selected for processing (14)
include/fluent-bit/http_server/flb_http_server.hplugins/in_elasticsearch/in_elasticsearch.cplugins/in_http/http.cplugins/in_opentelemetry/opentelemetry.cplugins/in_prometheus_remote_write/prom_rw.cplugins/in_splunk/splunk.csrc/http_server/flb_http_server.ctests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yamltests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.pytests/integration/scenarios/in_http/config/in_http_pause_resume.yamltests/integration/scenarios/in_http/tests/test_in_http_001.pytests/integration/scenarios/in_splunk/config/splunk_pause_resume.yamltests/integration/scenarios/in_splunk/tests/test_in_splunk_001.pytests/integration/src/utils/input_pause_resume.py
c123f0e to
82f3e6f
Compare
There was a problem hiding this comment.
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 `@tests/integration/scenarios/in_http/tests/test_in_http_001.py`:
- Around line 160-170: Update the pause/resume test around
assert_pause_resume_cycles to pass partial_connection into the helper, then
assert connection closure immediately whenever each paused-state observation
occurs rather than only after all cycles complete. Preserve validation of both
successful cycles and failed/paused teardown behavior, and remove the delayed
standalone assertion if the helper now performs it.
🪄 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: 9b8ed12b-ba84-4afd-81f1-6304b412a76a
📒 Files selected for processing (14)
include/fluent-bit/http_server/flb_http_server.hplugins/in_elasticsearch/in_elasticsearch.cplugins/in_http/http.cplugins/in_opentelemetry/opentelemetry.cplugins/in_prometheus_remote_write/prom_rw.cplugins/in_splunk/splunk.csrc/http_server/flb_http_server.ctests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yamltests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.pytests/integration/scenarios/in_http/config/in_http_pause_resume.yamltests/integration/scenarios/in_http/tests/test_in_http_001.pytests/integration/scenarios/in_splunk/config/splunk_pause_resume.yamltests/integration/scenarios/in_splunk/tests/test_in_splunk_001.pytests/integration/src/utils/input_pause_resume.py
| assert_pause_resume_cycles( | ||
| service.flb, | ||
| f"http://localhost:{service.flb_listener_port}/", | ||
| large_json_payload(size=6144), | ||
| ["Content-Type: application/json"], | ||
| input_name="http.0", | ||
| success_status=201, | ||
| cycles=2, | ||
| pause_trigger_requests=2, | ||
| ) | ||
| assert_connection_closed(partial_connection) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Assert active-connection teardown during the paused state.
Line 170 runs after both pause/resume cycles complete, so it can pass if the socket closes later than pause. Pass the connection into assert_pause_resume_cycles and assert it immediately after each paused-state observation. As per coding guidelines, “Validate both success and failure paths.”
Proposed fix
def assert_pause_resume_cycles(
flb,
url,
payload,
headers,
*,
input_name,
success_status,
+ active_connection=None,
cycles=2,
http_mode="http1.1",
pause_trigger_requests=1,
resume_payload=None,
):
@@
wait_for_input_pause_state(
flb,
input_name,
True,
timeout=20 if is_valgrind() else 10,
)
+ if active_connection is not None:
+ assert_connection_closed(active_connection) cycles=2,
pause_trigger_requests=2,
+ active_connection=partial_connection,
)
- assert_connection_closed(partial_connection)🤖 Prompt for 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.
In `@tests/integration/scenarios/in_http/tests/test_in_http_001.py` around lines
160 - 170, Update the pause/resume test around assert_pause_resume_cycles to
pass partial_connection into the helper, then assert connection closure
immediately whenever each paused-state observation occurs rather than only after
all cycles complete. Preserve validation of both successful cycles and
failed/paused teardown behavior, and remove the delayed standalone assertion if
the helper now performs it.
Source: Coding guidelines
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 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 `@tests/integration/scenarios/in_http/tests/test_in_http_001.py`:
- Line 230: Update the direct wait_for_input_pause_state call in the HTTP
integration test to pass the Valgrind-aware timeout, using 20 seconds when
is_valgrind() is true and 10 seconds otherwise, instead of relying on the
default timeout.
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 1568-1571: Move each service startup call into its corresponding
protected try/finally block so failures during partial startup still execute
cleanup: update
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
lines 1568-1571 for service.start(),
tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py
lines 150-152 for service.start(start_sender=False), and
tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py lines 270-273
for service.start(). Validate the failure paths to ensure stop() runs when
startup raises.
- Around line 1573-1584: Extend the pause/resume coverage at
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py:1573-1584
by retaining an active HTTP/2 client stream during pause and asserting it
receives EOF or reset before invoking resume validation; update the
assert_pause_resume_cycles scenario setup as needed without changing
fresh-request checks. Also update
tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py:276-286 to
retain an active HTTP/1.1 partial or keep-alive request during pause and assert
its closure before testing resume.
In
`@tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py`:
- Around line 168-185: Reorder the teardown flow around
open_stalled_tcp_connection and assert_connection_closed: create and validate
paused_connection while the sender remains running and the input pause trigger
is active, then call service.sender.stop(), clear service.sender, and finally
wait_for_input_pause_state with False.
- Around line 180-191: Extend the resumed phase in the test around
wait_for_input_pause_state so it sends a valid remote-write payload after
confirming paused=False, then assert successful ingestion and downstream
delivery before triggering the second pause. Replace the log-string-only
verification with checks covering the expected success behavior, while
preserving the existing pause/resume sequence.
In `@tests/integration/src/utils/input_pause_resume.py`:
- Around line 81-93: Update input_pause_state to use the existing
is_valgrind()-based monitoring timeout instead of the fixed 2-second value, and
catch requests request/HTTP exceptions from requests.get and
response.raise_for_status. Return None on transient polling failures so
wait_for_input_pause_state can continue retrying until its configured deadline.
🪄 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: 825e74a1-5ef5-42ce-9935-1d4f1d92589b
📒 Files selected for processing (14)
tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yamltests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yamltests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.pytests/integration/scenarios/in_http/config/in_http_pause_resume.yamltests/integration/scenarios/in_http/config/in_http_pause_resume_http2_tls.yamltests/integration/scenarios/in_http/tests/test_in_http_001.pytests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.pytests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume.yamltests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.pytests/integration/scenarios/in_splunk/config/splunk_pause_resume.yamltests/integration/scenarios/in_splunk/config/splunk_pause_resume_workers.yamltests/integration/scenarios/in_splunk/tests/test_in_splunk_001.pytests/integration/src/utils/input_pause_resume.py
| ) | ||
| assert result["status_code"] == 201, result | ||
|
|
||
| wait_for_input_pause_state(service.flb, "http.0", True) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Missing Valgrind-aware timeout for this pause-state wait.
Every other "become paused" wait in this PR's shared helper doubles the timeout under Valgrind (timeout=20 if is_valgrind() else 10), but this direct call uses the default timeout=15. Under strict Valgrind this could time out and fail the test even though pause would eventually occur.
🐛 Proposed fix
- wait_for_input_pause_state(service.flb, "http.0", True)
+ wait_for_input_pause_state(
+ service.flb,
+ "http.0",
+ True,
+ timeout=20 if is_valgrind() else 10,
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| wait_for_input_pause_state(service.flb, "http.0", True) | |
| wait_for_input_pause_state( | |
| service.flb, | |
| "http.0", | |
| True, | |
| timeout=20 if is_valgrind() else 10, | |
| ) |
🤖 Prompt for 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.
In `@tests/integration/scenarios/in_http/tests/test_in_http_001.py` at line 230,
Update the direct wait_for_input_pause_state call in the HTTP integration test
to pass the Valgrind-aware timeout, using 20 seconds when is_valgrind() is true
and 10 seconds otherwise, instead of relying on the default timeout.
| service = Service("otlp_pause_resume_workers.yaml") | ||
| service.start() | ||
|
|
||
| try: |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Ensure partial startup is cleaned up.
Each service starts before entering try/finally; a readiness/startup failure bypasses stop() and can leak a process or environment state.
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py#L1568-L1571: moveservice.start()inside the protected block.tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py#L150-L152: moveservice.start(start_sender=False)inside the protected block.tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py#L270-L273: moveservice.start()inside the protected block.
As per coding guidelines, validate failure paths.
📍 Affects 3 files
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py#L1568-L1571(this comment)tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py#L150-L152tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py#L270-L273
🤖 Prompt for 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.
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`
around lines 1568 - 1571, Move each service startup call into its corresponding
protected try/finally block so failures during partial startup still execute
cleanup: update
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
lines 1568-1571 for service.start(),
tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py
lines 150-152 for service.start(start_sender=False), and
tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py lines 270-273
for service.start(). Validate the failure paths to ensure stop() runs when
startup raises.
Source: Coding guidelines
| wait_for_input_pause_state( | ||
| service.receiver, | ||
| "prometheus_remote_write.0", | ||
| False, | ||
| timeout=30, | ||
| ) | ||
| service.wait_for_log( | ||
| service.receiver.log_file, | ||
| "fluentbit_input_metrics_scrapes_total", | ||
| timeout=30, | ||
| interval=0.5, | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Verify successful remote-write ingestion after resume.
These lines only observe paused=0 and a log string while the sender is stopped. Submit a valid remote-write payload after resume and assert its success and downstream delivery before triggering the second pause. As per coding guidelines, shared lifecycle changes must verify success/retry/drop behavior.
🤖 Prompt for 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.
In
`@tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py`
around lines 180 - 191, Extend the resumed phase in the test around
wait_for_input_pause_state so it sends a valid remote-write payload after
confirming paused=False, then assert successful ingestion and downstream
delivery before triggering the second pause. Replace the log-string-only
verification with checks covering the expected success behavior, while
preserving the existing pause/resume sequence.
Source: Coding guidelines
| def input_pause_state(flb, input_name): | ||
| response = requests.get( | ||
| f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus", | ||
| timeout=2, | ||
| ) | ||
| response.raise_for_status() | ||
|
|
||
| for line in response.text.splitlines(): | ||
| match = INGESTION_PAUSED_RE.match(line) | ||
| if match and match.group(1) == input_name: | ||
| return int(float(match.group(2))) | ||
|
|
||
| return None |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
input_pause_state can raise uncaught request errors during polling, breaking Valgrind-mode retries.
The 2s request timeout here doesn't scale with is_valgrind() like every other wait in this file (timeout=20 if is_valgrind() else 10, etc.), and there's no exception handling around requests.get/raise_for_status. Under strict Valgrind — which coding guidelines require running for this scenario — a slow monitoring response can raise requests.exceptions.Timeout/HTTPError, propagating out of wait_for_input_pause_state's polling loop and failing the test immediately instead of retrying until the (valgrind-extended) deadline.
As per coding guidelines, "For components covered by integration scenarios, run the focused scenario normally and with strict Valgrind, and report any blocker instead of silently skipping it."
🛡️ Proposed fix
def input_pause_state(flb, input_name):
- response = requests.get(
- f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus",
- timeout=2,
- )
- response.raise_for_status()
+ try:
+ response = requests.get(
+ f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus",
+ timeout=5 if is_valgrind() else 2,
+ )
+ response.raise_for_status()
+ except requests.exceptions.RequestException:
+ return None
for line in response.text.splitlines():📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def input_pause_state(flb, input_name): | |
| response = requests.get( | |
| f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus", | |
| timeout=2, | |
| ) | |
| response.raise_for_status() | |
| for line in response.text.splitlines(): | |
| match = INGESTION_PAUSED_RE.match(line) | |
| if match and match.group(1) == input_name: | |
| return int(float(match.group(2))) | |
| return None | |
| def input_pause_state(flb, input_name): | |
| try: | |
| response = requests.get( | |
| f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus", | |
| timeout=5 if is_valgrind() else 2, | |
| ) | |
| response.raise_for_status() | |
| except requests.exceptions.RequestException: | |
| return None | |
| for line in response.text.splitlines(): | |
| match = INGESTION_PAUSED_RE.match(line) | |
| if match and match.group(1) == input_name: | |
| return int(float(match.group(2))) | |
| return None |
🧰 Tools
🪛 ast-grep (0.45.0)
[warning] 81-84: Request-controlled URL passed to requests; validate against an allowlist to prevent SSRF.
Context: requests.get(
f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus",
timeout=2,
)
Note: [CWE-918] Server-Side Request Forgery (SSRF).
(ssrf-requests)
🤖 Prompt for 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.
In `@tests/integration/src/utils/input_pause_resume.py` around lines 81 - 93,
Update input_pause_state to use the existing is_valgrind()-based monitoring
timeout instead of the fixed 2-second value, and catch requests request/HTTP
exceptions from requests.get and response.raise_for_status. Return None on
transient polling failures so wait_for_input_pause_state can continue retrying
until its configured deadline.
Source: Coding guidelines
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
addc809 to
6481fbf
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
include/fluent-bit/flb_downstream.h (1)
91-100: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSingle root cause:
flb_downstream_conn_event_accept()has no distinguishable result for "no pending connection". Because EAGAIN, paused-drain, and fatal failures all collapse into-1/0, the accept batch loop cannot terminate meaningfully or report real errors.
include/fluent-bit/flb_downstream.h#L91-L100: document the return values and introduce a distinct result (or out-param) for "no connection pending" versus a fatal failure.src/http_server/flb_http_server.c#L568-L586: branch on that distinct result — exit quietly on EAGAIN, log on fatal failure — and renameaccepted_connectionsto reflect that it counts attempts.🤖 Prompt for 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. In `@include/fluent-bit/flb_downstream.h` around lines 91 - 100, The downstream accept API must distinguish “no connection pending” from fatal failure. Update flb_downstream_conn_event_accept and its documented return contract in include/fluent-bit/flb_downstream.h, then update the accept loop in src/http_server/flb_http_server.c to exit quietly for EAGAIN, log fatal failures, and rename accepted_connections to reflect that it counts attempts.src/http_server/flb_http_server.c (1)
976-977: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueEnsure the synchronous listener path is intentionally retained
flb_http_server_start()creates the downstream withflb_http_server_client_connection_event_handler, enables async immediately before registering the listener event, and no other path for this server downstream disables async mode. That makes theflb_downstream_is_async(...) == FLB_FALSEbranch at L542 unreachable unless runtime start or another external path can take it; document that intent or remove the dead sync path.🤖 Prompt for 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. In `@src/http_server/flb_http_server.c` around lines 976 - 977, Update flb_http_server_start() and the downstream event handling around flb_stream_enable_async_mode() to make the intended async-only behavior explicit: either document why the synchronous branch guarded by flb_downstream_is_async(...) is retained, or remove that unreachable synchronous path and its related handling. Preserve listener registration and async startup behavior.
🤖 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 @.github/workflows/input-http-pause-tests.yaml:
- Line 63: Update the actions/checkout step in the input HTTP pause test
workflow to disable credential persistence by configuring persist-credentials as
false. Keep the existing pinned checkout action version unchanged.
In
`@tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py`:
- Around line 286-289: Move service.start() inside the existing try/finally
block surrounding the integration scenario, while keeping service.stop() in
finally so cleanup runs even when startup raises. Update the setup flow around
the Service instance without changing the scenario’s normal execution behavior.
---
Nitpick comments:
In `@include/fluent-bit/flb_downstream.h`:
- Around line 91-100: The downstream accept API must distinguish “no connection
pending” from fatal failure. Update flb_downstream_conn_event_accept and its
documented return contract in include/fluent-bit/flb_downstream.h, then update
the accept loop in src/http_server/flb_http_server.c to exit quietly for EAGAIN,
log fatal failures, and rename accepted_connections to reflect that it counts
attempts.
In `@src/http_server/flb_http_server.c`:
- Around line 976-977: Update flb_http_server_start() and the downstream event
handling around flb_stream_enable_async_mode() to make the intended async-only
behavior explicit: either document why the synchronous branch guarded by
flb_downstream_is_async(...) is retained, or remove that unreachable synchronous
path and its related handling. Preserve listener registration and async startup
behavior.
🪄 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: 8d506891-cc7d-44e2-9c97-b93d73fe291d
📒 Files selected for processing (29)
.github/workflows/input-http-pause-tests.yamlinclude/fluent-bit/flb_connection.hinclude/fluent-bit/flb_downstream.hinclude/fluent-bit/http_server/flb_http_server.hplugins/in_elasticsearch/in_elasticsearch.cplugins/in_http/http.cplugins/in_opentelemetry/opentelemetry.cplugins/in_prometheus_remote_write/prom_rw.cplugins/in_splunk/splunk.csrc/flb_downstream.csrc/flb_network.csrc/http_server/flb_http_server.ctests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yamltests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yamltests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.pytests/integration/scenarios/in_http/config/in_http_accept_timeout_tls.yamltests/integration/scenarios/in_http/config/in_http_pause_resume.yamltests/integration/scenarios/in_http/config/in_http_pause_resume_http2_tls.yamltests/integration/scenarios/in_http/tests/test_in_http_001.pytests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.pytests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume.yamltests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume_workers.yamltests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.pytests/integration/scenarios/in_splunk/config/splunk_pause_resume.yamltests/integration/scenarios/in_splunk/config/splunk_pause_resume_workers.yamltests/integration/scenarios/in_splunk/tests/test_in_splunk_001.pytests/integration/src/utils/input_pause_resume.pytests/internal/http_server.c
🚧 Files skipped from review as they are similar to previous changes (2)
- tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yaml
- tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yaml
| timeout-minutes: 30 | ||
|
|
||
| steps: | ||
| - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Checkout doesn't disable credential persistence.
This job only builds and runs tests (no push), so persisting GITHUB_TOKEN in the local git config is unnecessary exposure surface for any later step/dependency.
🔒 Proposed fix
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 | |
| - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 | |
| with: | |
| persist-credentials: false |
🧰 Tools
🪛 zizmor (1.28.0)
[warning] 63-63: credential persistence through GitHub Actions artifacts (artipacked): does not set persist-credentials: false
(artipacked)
🤖 Prompt for 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.
In @.github/workflows/input-http-pause-tests.yaml at line 63, Update the
actions/checkout step in the input HTTP pause test workflow to disable
credential persistence by configuring persist-credentials as false. Keep the
existing pinned checkout action version unchanged.
Source: Linters/SAST tools
| service = Service(config_file) | ||
| service.start() | ||
|
|
||
| try: |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
service.start() runs outside the try/finally.
If start() raises partway through startup, service.stop() never runs, potentially leaking the Fluent Bit process.
🛡️ Proposed fix
service = Service(config_file)
- service.start()
try:
+ service.start()
large_document = '{"index":{}}\n{"message":"' + ("x" * 6144) + '"}\n'📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| service = Service(config_file) | |
| service.start() | |
| try: | |
| service = Service(config_file) | |
| try: | |
| service.start() |
🤖 Prompt for 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.
In
`@tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py`
around lines 286 - 289, Move service.start() inside the existing try/finally
block surrounding the integration scenario, while keeping service.stop() in
finally so cleanup runs even when startup raises. Update the setup flow around
the Service instance without changing the scenario’s normal execution behavior.
1 participant
This patch implements the missing pause and resume callbacks for in_tcp, in_http and in_opentelemetry plugins. When the plugin is paused, all active client connections are forcefully closed, and new incoming connections are rejected. The
plugin tracks its paused state to ensure correct behavior until it is resumed.
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