http_server: add settings for ingress queue - #11704
Merged
Merged
Conversation
Member
Author
|
cc: @lecaros |
|
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:
📝 WalkthroughWalkthroughAdds configurable ingress-queue event/byte limits and per-input CMetrics for HTTP-server inputs; updates enqueue/drain/backpressure paths to publish those metrics and honor HTTP-server config limits; moves ingress teardown earlier in instance destroy; adds integration tests for metrics, tiny-queue behavior, and updates AGENTS.md Valgrind guidance. Changes
Sequence Diagram(s)sequenceDiagram
actor Client
participant HTTP_Server as HTTP Server
participant Ingress_Queue as Ingress Queue
participant Input_Plugin as Input Plugin
participant CMetrics as CMetrics
rect rgba(76, 175, 80, 0.5)
Note over Client,Input_Plugin: Nominal flow
Client->>HTTP_Server: POST OTLP payload
HTTP_Server->>Ingress_Queue: Enqueue event
Ingress_Queue->>CMetrics: set pending_events / pending_bytes
Input_Plugin->>Ingress_Queue: Drain events
Ingress_Queue->>CMetrics: reset pending_events / pending_bytes
end
rect rgba(244, 67, 54, 0.5)
Note over Client,Input_Plugin: Backpressure (queue full / timeout)
Client->>HTTP_Server: POST OTLP payload
HTTP_Server->>Ingress_Queue: Enqueue attempt (times out)
Ingress_Queue->>CMetrics: increment busy counter
HTTP_Server->>Client: HTTP 503 "deferred ingress queue is full"
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 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.
🧹 Nitpick comments (1)
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py (1)
1003-1048: Add an explicit drop-path forwarding assertion.Alongside Line 1032–Line 1048 metric checks, assert the rejected request did not add a forwarded log payload.
Suggested patch
def test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits(): service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG) service.start() service.wait_for_log_message("with 4 workers", timeout=10) + logs_before = len(data_storage["logs"]) response = service.send_raw_request( "/v1/logs", service.build_otel_payload("test_logs_001.in.json", "logs"), ) @@ assert response.status_code == 503 assert "deferred ingress queue is full" in response.text + assert len(data_storage["logs"]) == logs_before assert read_prometheus_metric_value( metrics_text, "fluentbit_input_http_server_ingress_queue_busy_total", "opentelemetry.0", ) >= 1Based on learnings: Verify metrics/counters for success, retry, and drop paths when reviewing changes.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py` around lines 1003 - 1048, In test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits add an assertion after the existing metric/assert checks to ensure the rejected request did not produce a forwarded payload: use read_prometheus_metric_value to assert the forwarder metric (e.g. "fluentbit_output_forward_total" with label "opentelemetry.0") is 0, or if the test harness exposes forwarded payloads on the Service object (e.g. service.forwarded_payloads or service.get_forwarded_payloads()), assert that collection is empty; place this new assertion after service.stop() and alongside the other read_prometheus_metric_value checks.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 1003-1048: In
test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits add an
assertion after the existing metric/assert checks to ensure the rejected request
did not produce a forwarded payload: use read_prometheus_metric_value to assert
the forwarder metric (e.g. "fluentbit_output_forward_total" with label
"opentelemetry.0") is 0, or if the test harness exposes forwarded payloads on
the Service object (e.g. service.forwarded_payloads or
service.get_forwarded_payloads()), assert that collection is empty; place this
new assertion after service.stop() and alongside the other
read_prometheus_metric_value checks.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 28609743-4211-45ca-8354-f8c7e1f3a15f
📒 Files selected for processing (9)
AGENTS.mdinclude/fluent-bit/flb_input.hinclude/fluent-bit/http_server/flb_http_server.hsrc/flb_input.csrc/flb_input_ingest.csrc/http_server/flb_http_server.csrc/http_server/flb_http_server_config_map.ctests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
edsiper
force-pushed
the
deferred-ingress-queue-is-full
branch
from
April 13, 2026 18:06
a710831 to
83aa6c7
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py (1)
960-1000: Please add one non-log ingress-queue case.These new regressions only exercise logs, but the deferred-ingress/accounting path is shared and may branch by signal type. Adding at least one metrics or traces variant here would make the coverage much safer.
Based on learnings: Shared paths must branch correctly by
event_type(logs vs non-logs); add regression tests for mixed signals.Also applies to: 1003-1050
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py` around lines 960 - 1000, Update test_in_opentelemetry_http_workers_export_ingress_queue_metrics to also exercise a non-log signal path (metrics or traces) so the deferred-ingress/accounting branch that uses event_type is covered; after the existing logs request, send an additional request using service.send_raw_request with service.build_otel_payload("test_metrics_001.in.json", "metrics") (or "traces") and assert response.status_code == 201, wait for the corresponding signal count (e.g., service.wait_for_signal_count("metrics", 1, ...)), then re-scrape prometheus metrics and assert the same ingress queue metrics exist and their values for the new signal label (e.g., "opentelemetry.0") are as expected; ensure you reference the same test function name test_in_opentelemetry_http_workers_export_ingress_queue_metrics and reuse service.send_raw_request, service.build_otel_payload, service.wait_for_signal_count, and read_prometheus_metric_value to perform the checks.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 1003-1050: The test
test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits currently
only exercises the event-count limit
(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG) so it won't catch regressions in
the byte-limit enforcement; add a separate test that mirrors this one but uses a
config where http_server.ingress_queue_event_limit is permissive (large or 0
meaning unlimited) and http_server.ingress_queue_byte_limit is small (to force
byte-only rejection), e.g. create IN_OPENTELEMETRY_TINY_BYTE_LIMIT_ONLY_CONFIG,
start Service with that config, send the same payload via
service.send_raw_request/service.build_otel_payload, wait for the same
prometheus metric fluentbit_input_http_server_ingress_queue_busy_total, assert
response.status_code == 503 and the "deferred ingress queue is full" text, and
assert the byte/ pending metrics
(fluentbit_input_http_server_ingress_queue_pending_bytes == 0 and busy_total
>=1) to validate byte-limit path separately from event-count enforcement.
---
Nitpick comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 960-1000: Update
test_in_opentelemetry_http_workers_export_ingress_queue_metrics to also exercise
a non-log signal path (metrics or traces) so the deferred-ingress/accounting
branch that uses event_type is covered; after the existing logs request, send an
additional request using service.send_raw_request with
service.build_otel_payload("test_metrics_001.in.json", "metrics") (or "traces")
and assert response.status_code == 201, wait for the corresponding signal count
(e.g., service.wait_for_signal_count("metrics", 1, ...)), then re-scrape
prometheus metrics and assert the same ingress queue metrics exist and their
values for the new signal label (e.g., "opentelemetry.0") are as expected;
ensure you reference the same test function name
test_in_opentelemetry_http_workers_export_ingress_queue_metrics and reuse
service.send_raw_request, service.build_otel_payload,
service.wait_for_signal_count, and read_prometheus_metric_value to perform the
checks.
🪄 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
Run ID: 824415cd-f0bf-4800-b468-e6400d849a1d
📒 Files selected for processing (9)
AGENTS.mdinclude/fluent-bit/flb_input.hinclude/fluent-bit/http_server/flb_http_server.hsrc/flb_input.csrc/flb_input_ingest.csrc/http_server/flb_http_server.csrc/http_server/flb_http_server_config_map.ctests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
✅ Files skipped from review due to trivial changes (3)
- include/fluent-bit/flb_input.h
- tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml
- AGENTS.md
🚧 Files skipped from review as they are similar to previous changes (4)
- src/http_server/flb_http_server_config_map.c
- src/http_server/flb_http_server.c
- src/flb_input_ingest.c
- include/fluent-bit/http_server/flb_http_server.h
edsiper
force-pushed
the
deferred-ingress-queue-is-full
branch
from
April 13, 2026 18:27
83aa6c7 to
8c7ba0b
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/flb_input_ingest.c`:
- Around line 449-454: The current copy copies
http_server_config->ingress_queue_byte_limit (which can be 0 to mean "disabled")
directly into ins->ingress_queue_byte_limit, causing the later clamp (which only
runs when mem_buf_limit < ingress_queue_byte_limit) to be skipped and removing
the per-input memory cap; change the logic so that when
http_server_config->ingress_queue_byte_limit is zero you leave a non-zero
fallback (use mem_buf_limit) — e.g. only assign the config value into
ins->ingress_queue_byte_limit if it is > 0, otherwise set
ins->ingress_queue_byte_limit = mem_buf_limit (or apply the existing clamp
afterwards with a guard that treats 0 as "use mem_buf_limit"); update the block
that references ins->http_server_config, mem_buf_limit, and the clamp around
mem_buf_limit < ingress_queue_byte_limit accordingly.
- Around line 243-248: The timeout path currently releases
ins->ingress_queue_lock and then dereferences ins (reading
ins->cmt_ingress_queue_busy and calling flb_input_name(ins)), which can race
with flb_input_ingress_destroy()/flb_input_instance_destroy() and cause
use-after-free; fix by capturing needed pointers/values while holding the lock
(e.g., store ins->cmt_ingress_queue_busy into a local cmt pointer and copy the
input name string or index into a local buffer) or by re-acquiring/validating
the instance before using it, then release the lock and use only those local
copies (also null-check the saved cmt pointer before calling cmt_counter_add);
this change should be applied around the code that references ins, the
ingress_queue_lock, cmt_ingress_queue_busy, and flb_input_name to avoid touching
ins after the lock is dropped.
In `@src/flb_input.c`:
- Around line 1555-1580: The metrics for HTTP-server ingress queue
(ins->cmt_ingress_queue_busy, ins->cmt_ingress_queue_pending_events,
ins->cmt_ingress_queue_pending_bytes and their corresponding cmt_*_set calls)
are being registered for every input instance; guard this entire block so it
only runs when the input is an HTTP-server by checking the input flags (use
p->flags & FLB_INPUT_HTTP_SERVER) before creating or setting these metrics; move
the three cmt_*_create and cmt_*_set calls inside that conditional so non-HTTP
inputs do not export zero-valued HTTP-server metrics.
In `@src/http_server/flb_http_server_config_map.c`:
- Around line 50-59: The two new config map entries
"http_server.ingress_queue_event_limit" and
"http_server.ingress_queue_byte_limit" need bare compatibility aliases so
plugin-local keys "ingress_queue_event_limit" and "ingress_queue_byte_limit" are
accepted; update the config map (the array that registers FLB_CONFIG_MAP_SIZE
entries) to add duplicate entries with the unprefixed names and same properties
(pointing to offsetof(struct flb_http_server_config, ingress_queue_event_limit)
and offsetof(..., ingress_queue_byte_limit)), and also add those bare names to
the existing whitelist/alias list that currently contains "http2",
"buffer_max_size", "buffer_chunk_size", "max_connections", "workers" so the
parser accepts both prefixed and unprefixed keys.
🪄 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
Run ID: b5b8720c-6961-4765-b89a-89a4ce390c38
📒 Files selected for processing (9)
AGENTS.mdinclude/fluent-bit/flb_input.hinclude/fluent-bit/http_server/flb_http_server.hsrc/flb_input.csrc/flb_input_ingest.csrc/http_server/flb_http_server.csrc/http_server/flb_http_server_config_map.ctests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
✅ Files skipped from review due to trivial changes (4)
- src/http_server/flb_http_server.c
- AGENTS.md
- include/fluent-bit/flb_input.h
- tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml
🚧 Files skipped from review as they are similar to previous changes (1)
- include/fluent-bit/http_server/flb_http_server.h
Comment on lines
+50
to
+59
| { | ||
| FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_event_limit", "8192", | ||
| 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_event_limit), | ||
| "Set the maximum number of deferred ingress queue events. Applies only when http_server.workers > 1." | ||
| }, | ||
| { | ||
| FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_byte_limit", "256M", | ||
| 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_byte_limit), | ||
| "Set the maximum number of deferred ingress queue bytes. Applies only when http_server.workers > 1." | ||
| }, |
There was a problem hiding this comment.
Add the same compatibility aliases the older HTTP-server settings already have.
These two options are only registered in http_server.* form here, while Line 111 still only whitelists the older bare aliases (http2, buffer_max_size, buffer_chunk_size, max_connections, workers). That means plugin-local ingress_queue_event_limit / ingress_queue_byte_limit keys will be rejected even though the rest of this config surface supports bare aliases.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/http_server/flb_http_server_config_map.c` around lines 50 - 59, The two
new config map entries "http_server.ingress_queue_event_limit" and
"http_server.ingress_queue_byte_limit" need bare compatibility aliases so
plugin-local keys "ingress_queue_event_limit" and "ingress_queue_byte_limit" are
accepted; update the config map (the array that registers FLB_CONFIG_MAP_SIZE
entries) to add duplicate entries with the unprefixed names and same properties
(pointing to offsetof(struct flb_http_server_config, ingress_queue_event_limit)
and offsetof(..., ingress_queue_byte_limit)), and also add those bare names to
the existing whitelist/alias list that currently contains "http2",
"buffer_max_size", "buffer_chunk_size", "max_connections", "workers" so the
parser accepts both prefixed and unprefixed keys.
edsiper
force-pushed
the
deferred-ingress-queue-is-full
branch
from
April 13, 2026 18:41
8c7ba0b to
c433fe1
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py (1)
1003-1050:⚠️ Potential issue | 🟡 MinorThis still doesn't isolate byte-limit enforcement.
With the tiny-queue config keeping
http_server.ingress_queue_event_limitat0, this request can be rejected on event count alone. A regression inhttp_server.ingress_queue_byte_limitwould still pass this test, so please add a second scenario with a permissive event limit and only the byte limit set tiny. Based on learnings: Validate both success and failure paths in tests (invalid payloads, boundary sizes, null/missing fields).🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py` around lines 1003 - 1050, The current test test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits only exercises event-count rejection (IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG keeps http_server.ingress_queue_event_limit=0) and thus doesn't verify byte-limit enforcement; add a second scenario that uses a config with a permissive event limit and a tiny http_server.ingress_queue_byte_limit (e.g., set ingress_queue_event_limit high and ingress_queue_byte_limit to a small value), then send an OTLP payload built via service.build_otel_payload whose event count is within the limit but whose serialized size exceeds the byte limit, assert a 503 and the "deferred ingress queue is full" message, verify the appropriate prometheus metrics via maybe_read_prometheus_metric_value/read_prometheus_metric_value, and also add a complementary case that sends a slightly smaller payload that stays under the byte limit and asserts success and logs ingestion.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 973-978: The current wait_for_condition call exits on the first
non-empty scrape; change the predicate passed to
service.service.wait_for_condition (the lambda around
service.scrape_prometheus_metrics) to only return a value when the scraped
metrics include the specific "ingress-queue" series (e.g., check for the metric
name or unique series label substring) so the wait retries until that series is
present before proceeding with assertions.
---
Duplicate comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 1003-1050: The current test
test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits only
exercises event-count rejection (IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG
keeps http_server.ingress_queue_event_limit=0) and thus doesn't verify
byte-limit enforcement; add a second scenario that uses a config with a
permissive event limit and a tiny http_server.ingress_queue_byte_limit (e.g.,
set ingress_queue_event_limit high and ingress_queue_byte_limit to a small
value), then send an OTLP payload built via service.build_otel_payload whose
event count is within the limit but whose serialized size exceeds the byte
limit, assert a 503 and the "deferred ingress queue is full" message, verify the
appropriate prometheus metrics via
maybe_read_prometheus_metric_value/read_prometheus_metric_value, and also add a
complementary case that sends a slightly smaller payload that stays under the
byte limit and asserts success and logs ingestion.
🪄 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
Run ID: 89f9eceb-4c15-461d-bbec-b49c19e43621
📒 Files selected for processing (9)
AGENTS.mdinclude/fluent-bit/flb_input.hinclude/fluent-bit/http_server/flb_http_server.hsrc/flb_input.csrc/flb_input_ingest.csrc/http_server/flb_http_server.csrc/http_server/flb_http_server_config_map.ctests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
✅ Files skipped from review due to trivial changes (5)
- AGENTS.md
- src/http_server/flb_http_server_config_map.c
- include/fluent-bit/flb_input.h
- tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml
- src/flb_input_ingest.c
🚧 Files skipped from review as they are similar to previous changes (3)
- src/http_server/flb_http_server.c
- include/fluent-bit/http_server/flb_http_server.h
- src/flb_input.c
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>
edsiper
force-pushed
the
deferred-ingress-queue-is-full
branch
from
April 13, 2026 19:28
c433fe1 to
0582f31
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
src/http_server/flb_http_server_config_map.c (1)
50-59:⚠️ Potential issue | 🟡 MinorRegister the bare compatibility aliases too.
These two settings are only added in
http_server.*form here, whileflb_http_server_property_is_allowed()still only accepts the older unprefixed aliases. Plugin-localingress_queue_event_limit/ingress_queue_byte_limitkeys will still be rejected even though the rest of this config surface accepts both forms.Possible fix
{ FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_event_limit", "8192", 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_event_limit), "Set the maximum number of deferred ingress queue events. Applies only when http_server.workers > 1." }, { FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_byte_limit", "256M", 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_byte_limit), "Set the maximum number of deferred ingress queue bytes. Applies only when http_server.workers > 1." }, + { + FLB_CONFIG_MAP_SIZE, "ingress_queue_event_limit", "8192", + 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_event_limit), + "Compatibility alias for http_server.ingress_queue_event_limit" + }, + { + FLB_CONFIG_MAP_SIZE, "ingress_queue_byte_limit", "256M", + 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_byte_limit), + "Compatibility alias for http_server.ingress_queue_byte_limit" + }, ... - if (strcasecmp("http2", property_name) == 0 || + if (strcasecmp("http2", property_name) == 0 || strcasecmp("buffer_max_size", property_name) == 0 || strcasecmp("buffer_chunk_size", property_name) == 0 || strcasecmp("max_connections", property_name) == 0 || - strcasecmp("workers", property_name) == 0) { + strcasecmp("workers", property_name) == 0 || + strcasecmp("ingress_queue_event_limit", property_name) == 0 || + strcasecmp("ingress_queue_byte_limit", property_name) == 0) { return FLB_TRUE; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/http_server/flb_http_server_config_map.c` around lines 50 - 59, The two new config entries are only registered as "http_server.ingress_queue_event_limit" and "http_server.ingress_queue_byte_limit" but flb_http_server_property_is_allowed() still accepts the old unprefixed aliases, so plugin-local keys like "ingress_queue_event_limit" / "ingress_queue_byte_limit" will be rejected; add duplicate config map entries for the bare aliases (e.g. an entry with FLB_CONFIG_MAP_SIZE, "ingress_queue_event_limit", "8192", ... offsetof(struct flb_http_server_config, ingress_queue_event_limit) and likewise for "ingress_queue_byte_limit") so both forms map to the same struct fields and behavior, keeping the same flags and descriptions.
🧹 Nitpick comments (1)
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py (1)
1019-1132: Add one non-log backpressure case.These regressions only exercise
/v1/logs, but the deferred ingress queue is shared across OTLP logs, metrics, and traces. Mirroring one busy-path case for/v1/metricsor/v1/traceswould catch event-type-specific accounting regressions off the log path too.As per coding guidelines, "
src/**/*.c: Shared paths must branch correctly by event_type (logs vs non-logs)".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py` around lines 1019 - 1132, Add a non-log backpressure test by duplicating one of the existing tests (e.g., test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits or test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit) but target the non-log OTLP endpoint (/v1/metrics or /v1/traces) and use service.build_otel_payload(..., "metrics") or ("traces") instead of "logs"; start the Service with the same tiny ingress queue config, send the payload(s) to /v1/metrics or /v1/traces, wait for the same prometheus busy metric fluentbit_input_http_server_ingress_queue_busy_total for "opentelemetry.0", assert the 503 and "deferred ingress queue is full" on overflow and that data_storage did not increase, and mirror all other prometheus assertions from the mirrored test so the shared deferred ingress queue behavior is exercised for non-log event types.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 429-443: The poller currently lets exceptions from
scrape_prometheus_metrics bubble up and break wait_for_condition; update
wait_for_prometheus_metric to call self.scrape_prometheus_metrics inside a
try/except that catches requests.RequestException (and any relevant requests
timeouts) and returns None on failure so the predicate keeps retrying; reference
scrape_prometheus_metrics, wait_for_prometheus_metric,
self.scrape_prometheus_metrics, requests.RequestException and ensure the lambda
passed to service.wait_for_condition returns None on scrape errors instead of
raising.
---
Duplicate comments:
In `@src/http_server/flb_http_server_config_map.c`:
- Around line 50-59: The two new config entries are only registered as
"http_server.ingress_queue_event_limit" and
"http_server.ingress_queue_byte_limit" but flb_http_server_property_is_allowed()
still accepts the old unprefixed aliases, so plugin-local keys like
"ingress_queue_event_limit" / "ingress_queue_byte_limit" will be rejected; add
duplicate config map entries for the bare aliases (e.g. an entry with
FLB_CONFIG_MAP_SIZE, "ingress_queue_event_limit", "8192", ... offsetof(struct
flb_http_server_config, ingress_queue_event_limit) and likewise for
"ingress_queue_byte_limit") so both forms map to the same struct fields and
behavior, keeping the same flags and descriptions.
---
Nitpick comments:
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 1019-1132: Add a non-log backpressure test by duplicating one of
the existing tests (e.g.,
test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits or
test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit) but target
the non-log OTLP endpoint (/v1/metrics or /v1/traces) and use
service.build_otel_payload(..., "metrics") or ("traces") instead of "logs";
start the Service with the same tiny ingress queue config, send the payload(s)
to /v1/metrics or /v1/traces, wait for the same prometheus busy metric
fluentbit_input_http_server_ingress_queue_busy_total for "opentelemetry.0",
assert the 503 and "deferred ingress queue is full" on overflow and that
data_storage did not increase, and mirror all other prometheus assertions from
the mirrored test so the shared deferred ingress queue behavior is exercised for
non-log event types.
🪄 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
Run ID: 771f0c7e-5298-4e35-b8ea-df37c9f36fdd
📒 Files selected for processing (10)
AGENTS.mdinclude/fluent-bit/flb_input.hinclude/fluent-bit/http_server/flb_http_server.hsrc/flb_input.csrc/flb_input_ingest.csrc/http_server/flb_http_server.csrc/http_server/flb_http_server_config_map.ctests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yamltests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py
✅ Files skipped from review due to trivial changes (6)
- src/http_server/flb_http_server.c
- AGENTS.md
- tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yaml
- include/fluent-bit/flb_input.h
- tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml
- src/flb_input_ingest.c
🚧 Files skipped from review as they are similar to previous changes (1)
- include/fluent-bit/http_server/flb_http_server.h
Comment on lines
+429
to
+443
| def scrape_prometheus_metrics(self): | ||
| url = f"http://127.0.0.1:{self.flb.http_monitoring_port}/api/v2/metrics/prometheus" | ||
| response = requests.get(url, timeout=5) | ||
| response.raise_for_status() | ||
| return response.text | ||
|
|
||
| def wait_for_prometheus_metric(self, metric_name, timeout=10, interval=0.25): | ||
| return self.service.wait_for_condition( | ||
| lambda: ( | ||
| metrics if metric_name in metrics else None | ||
| ) if (metrics := self.scrape_prometheus_metrics()) else None, | ||
| timeout=timeout, | ||
| interval=interval, | ||
| description=f"prometheus metric {metric_name}", | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
fd -a 'test_service.py' . | while read -r file; do
printf '\n== %s ==\n' "$file"
sed -n '/def wait_for_condition/,/^[^[:space:]]/p' "$file"
done
printf '\n== scrape_prometheus_metrics call sites ==\n'
rg -n -C2 'scrape_prometheus_metrics|wait_for_condition\(' tests/integrationRepository: fluent/fluent-bit
Length of output: 28758
Handle transient Prometheus scrape failures in the poller.
The wait_for_condition() implementation does not catch exceptions from the predicate callback. If requests.get(...).raise_for_status() fails during startup (connection refused, timeout, HTTP 5xx), the exception propagates and terminates the wait instead of retrying, making the metrics tests flaky. Wrap the scrape in a try-except and return None on requests.RequestException to preserve the polling contract.
Suggested fix
def scrape_prometheus_metrics(self):
url = f"http://127.0.0.1:{self.flb.http_monitoring_port}/api/v2/metrics/prometheus"
- response = requests.get(url, timeout=5)
- response.raise_for_status()
- return response.text
+ try:
+ response = requests.get(url, timeout=5)
+ response.raise_for_status()
+ except requests.RequestException:
+ return None
+ return response.text🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py`
around lines 429 - 443, The poller currently lets exceptions from
scrape_prometheus_metrics bubble up and break wait_for_condition; update
wait_for_prometheus_metric to call self.scrape_prometheus_metrics inside a
try/except that catches requests.RequestException (and any relevant requests
timeouts) and returns None on failure so the predicate keeps retrying; reference
scrape_prometheus_metrics, wait_for_prometheus_metric,
self.scrape_prometheus_metrics, requests.RequestException and ensure the lambda
passed to service.wait_for_condition returns None on scrape errors instead of
raising.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
When http_server.workers > 1, Fluent Bit uses an internal deferred ingress queue before data enters the normal pipeline. These new options let users control that queue, and the new v2 metrics let them see if it is filling up.
New options:
http_server.ingress_queue_event_limit: maximum number of queued ingress eventshttp_server.ingress_queue_byte_limit: maximum total bytes queued in ingressNew
/api/v2/metrics/prometheusmetrics:fluentbit_input_http_server_ingress_queue_pending_events: current queued event countfluentbit_input_http_server_ingress_queue_pending_bytes: current queued byte countfluentbit_input_http_server_ingress_queue_busy_total: number of times the queue was full and Fluent Bit applied backpressureFluent 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
Tests
Documentation