Repository navigation
Conversation
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
|
Important Review skippedWe couldn't safely recover the incremental review. No full review was started, and the last reviewed checkpoint was preserved. Retry later, or explicitly request a full review by commenting You can disable this status message by setting the Use the checkbox below for a quick retry:
No actionable comments were generated in the recent review.
 |
| Layer / File(s) | Summary |
|---|---|
Shared protobuf log conversion include/fluent-bit/flb_opentelemetry.h, src/opentelemetry/*, src/CMakeLists.txt, plugins/in_opentelemetry/opentelemetry_protobuf.h, plugins/in_opentelemetry/CMakeLists.txt, tests/internal/opentelemetry_protobuf.c |
Adds shared protobuf validation and log conversion APIs. The implementation converts OTLP logs to MessagePack or CFL, tracks encoded size, and exposes CFL result ownership operations. The build uses the shared implementation, and the validator tests call the public API. |
CFL processor and processed-ingress path include/fluent-bit/flb_processor.h, include/fluent-bit/flb_input_log.h, include/fluent-bit/flb_input.h, src/flb_mp.c, src/flb_processor.c, src/flb_input_log.c, src/flb_input_ingest.c, plugins/processor_content_modifier/cm.c, plugins/processor_sql/sql.c, tests/internal/processor.c, tests/integration/scenarios/out_stdout/* |
Adds worker-aware chunk processor APIs and worker-safety flags. Input-log and ingress paths can append chunks or queue already-processed records. Processor handling materializes chunks before callbacks and preserves valid prefixes when MessagePack decoding encounters malformed tails. Tests cover corrupt tails and processor chains with multiple output workers. |
OpenTelemetry worker routing and parity coverage plugins/in_opentelemetry/opentelemetry.c, plugins/in_opentelemetry/opentelemetry.h, plugins/in_opentelemetry/opentelemetry_logs.c, tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py |
Adds logs_processor_in_workers and logs_direct_cfl options. The input can convert protobuf logs directly to CFL and can process eligible logs in HTTP workers before queueing. Integration tests compare conversion and processing paths across transports, processor chains, worker modes, and configured size limits. |
Estimated code review effort: 4 (Complex) | ~60 minutes
Change: Feature
Sequence Diagram(s)
sequenceDiagram
participant OTLPInput
participant LogsProtoDecoder
participant InputLogProcessor
participant IngressQueue
OTLPInput->>LogsProtoDecoder: Convert protobuf payload to CFL records
OTLPInput->>InputLogProcessor: Append CFL chunk
InputLogProcessor->>IngressQueue: Queue processed records
Possibly related PRs
- fluent/fluent-bit#12589: Adds processor window and chunk APIs used by this PR’s processor and input-log paths.
Suggested reviewers: cosmo0920, leonardo-albertovich
Merge Risk: 
0bd51
-
This change adds opt-in worker-side log processing and direct CFL conversion to the OpenTelemetry input. The default behavior is unchanged. No verified defect was found in the reviewed paths, so no merge-blocking risk remains.
Pre-merge checks: 
-
Check name Status Explanation Resolution Docstring Coverage Docstring coverage is 30.00% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff with cyclomatic complexity of at least 2. Documenta… Write docstrings for the functions missing them to satisfy the coverage threshold. -
Check name Status Explanation Description Check Check skipped - CodeRabbit’s high-level summary is enabled. Title check The title clearly and concisely identifies the main change: processing eligible OpenTelemetry logs in HTTP workers. Linked Issues check Check skipped because no linked issues were found for this pull request. Out of Scope Changes check Check skipped because no linked issues were found for this pull request.
Full details: Docstring Coverage
-
Explanation
Docstring coverage is 30.00% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff with cyclomatic complexity of at least 2. Documentation-only edits and deletions within surviving functions are included; unchanged diff context and fully deleted functions are excluded. Analyzed 70 functions across 19 files. (2 skipped: 2 unsupported.)
Resolution
Write docstrings for the functions missing them to satisfy the coverage threshold.
-
Finishing Touches
-
Generate docstrings
-
- Commit to this branch
- Create a new PR
🧪 Generate unit tests (beta)
-
- Commit to this branch
- Create a new PR
-
- Autofix · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 0bd512b27e
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| int flb_processor_logs_chunk_supported(struct flb_processor *proc, | ||
| int worker_thread); | ||
| int flb_processor_run_logs_chunk(struct flb_processor *proc, |
There was a problem hiding this comment.
This commit combines the new generic processor chunk API, input-log/ingress APIs, protobuf-decoder API, and OpenTelemetry plugin options and tests. These are distinct interfaces supporting the same feature, which AGENTS.md explicitly says must not be bundled into one commit. Split the interfaces into focused commits, keeping each interface's corresponding declarations and implementation together.
AGENTS.md reference: AGENTS.md:L176-L181
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Resolved by the cleaned branch history. Core interfaces, processor declarations, plugin changes, and integration tests are separate scoped commits; 0bd512b27 contains only the integration-test interface. The complete PR range also passes the repository commit-prefix checker against master.
| if (ret == 0) { | ||
| ret = otel_cfl_insert_kvlist(builder, body, "scope", 5, scope_kv); | ||
| } | ||
| scope_kv = NULL; |
There was a problem hiding this comment.
With logs_direct_cfl enabled, if inserting the resource into body fails—for example, because its key exceeds the remaining logs_max_size budget—the scope insertion is skipped, but scope_kv is still cleared. The error handler consequently cannot free the already-built scope and its attributes, leaking allocations on each such rejected request. Clear scope_kv only after passing it to the consuming insertion helper, and cover this boundary rejection under strict Valgrind.
AGENTS.md reference: AGENTS.md:L122-L126
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 0d48ee6. scope_kv is now cleared only after otel_cfl_insert_kvlist() is actually called, so the common error path destroys an unconsumed scope when resource insertion fails. The focused OTLP tests pass, and a rejected generated-limit case passes under strict Valgrind.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a21ceff8d4
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| return -1; | ||
| } | ||
| output->arena = json_arena_create(); | ||
| output->document = yyjson_read(body, length, 0); |
There was a problem hiding this comment.
For OTLP JSON containing invalid Unicode, such as a body string with an unpaired \uD800 escape, this strict parse rejects the entire request instead of preserving the previous replacement behavior. The existing flb_pack_json() path in src/flb_pack.c uses YYJSON_READ_ALLOW_INVALID_UNICODE | YYJSON_READ_REPLACE_INVALID_UNICODE; a focused check with the bundled yyjson confirms that the old flags accept this payload while the new flags reject it. Because JSON now takes this path even without processors, upgrading changes previously ingested batches into failed requests. Use the same Unicode flags here, or fall back to the existing conversion when strict parsing fails.
Useful? React with 👍 / 👎.
Feed OTLP/HTTP log payloads to the JSON and protobuf decoders used by the OpenTelemetry input. The CFL decoders run followed by record iteration and encoding, then the msgpack decoders run on the same payload. The first input byte selects the encoding, the body key and the logs_max_size budget. Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
When the input processors run in the HTTP workers, a full ingress queue was reported as a decoding failure and the client got a 400, so it did not resend the request. Return the busy status, which is answered with 503 like the unprocessed path. Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Run the ingress queue byte limit case with a logs processor too, and send the oversized request as JSON and protobuf, both must get a 503. Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
The --storage option buffers records in filesystem chunks on Fluent Bit and in a file_storage sending queue on the Collector. Requests the Collector refuses because its queue is full are resent by the clients, like Fluent Bit 503 answers, the delivered record accounting proves there is no loss. Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
The request length prefix is little-endian, decode it byte by byte so the corpus is also read on big-endian hosts such as s390x. Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Summary
in_opentelemetrydecoded OTLP/HTTP logs in the HTTP workers, but the inputprocessors ran later, serialized in the input collector thread. JSON and
protobuf logs were also converted to MessagePack only for native processors to
decode them back into CFL. On top of that, the HTTP server relied on
SO_REUSEPORTalone to spread connections, so a few long-lived keep-aliveclients could saturate one worker while the others stayed idle.
This PR:
There are no new configuration options. The fast paths apply automatically
when they are eligible; everything else keeps the existing MessagePack path.
How logs are processed
content_modifierandsqldeclare their logs callbacks worker safe and keeparena ownership when they create OTLP containers or insert converted values.
Requests without log records (empty
resourceLogs, scopes withoutlogRecords) are accepted, as the OpenTelemetry Collector does. They used tobe answered with 400 when no processors were configured.
Decoder parity
The CFL decoders produce the same records as the MessagePack decoders, and
accept, reject and size requests the same way:
flb_opentelemetry_logs_json_to_msgpack()check bycheck, with the same error statuses and value conversion rules;
logs_max_sizeis enforced at exactly the same size on both paths, for bothencodings;
or a timestamp beyond what an event can carry) uses the MessagePack
conversion, so it is handled exactly as before.
Internal tests compare both paths over the shared OTLP JSON cases, payloads
that exercise the leniency of the JSON conversion, and generated corpora of
200 JSON and 200 protobuf requests (many invalid on purpose), with a
logs_max_sizesweep around each request's size. The same comparison also ranover 40,000 random JSON and 40,000 random protobuf requests without a
difference.
HTTP worker connection balancing
downstream:flb_downstream_conn_event_adopt()drives an already acceptedsocket through the regular connection setup, including TLS, on the calling
worker's event loop. The socket is always consumed.
http_server: with more than one worker, the accepting worker hands each newconnection to the worker with the fewest active and pending connections,
through a per-worker queue and wake-up pipe. The
max_connectionsslot isreserved before the handoff, so admission limits stay exact. Servers with a
single worker keep the existing accept path.
Compatibility
logs_max_sizebehavior are unchanged; the onlyvisible change is that requests without log records are accepted (201).
flb_opentelemetry_logs_json_to_msgpack()remains exported and handles therequests CFL cannot represent. The new
flb_opentelemetry_logs_json_to_cfl()returns
FLB_OTEL_LOGS_JSON_CFL_UNSUPPORTEDfor them.expects the request to be accepted.
Benchmark
Fluent Bit 5.1, Fluent Bit 5.2 with this PR, and OpenTelemetry Collector 0.163
receive the same 30 million OTLP/HTTP log records and deliver them to a null
output. "Ingest" passes records through unchanged; "modified" upserts one log
attribute,
bench.added = "yes"(Fluent Bitcontent_modifier, CollectorOTTL). Every scenario runs twice: with records buffered in memory, and with
filesystem buffering on both sides.
Records/s is the whole run: 30M records divided by the time from the first
request to the last record delivered. Median of five runs, with the slowest and
fastest run in parentheses. Fluent Bit times are rounded to about one second
because its delivery counter is read once per second; the Collector's counters
are live.
Memory buffering
(0.63-0.94M)
(1.43-1.67M)
(0.78-0.79M)
(1.36-1.58M)
(1.58-1.77M)
(1.15-1.17M)
(0.20-0.21M)
(1.31-1.58M)
(0.75-0.76M)
(0.20-0.20M)
(1.31-1.43M)
(1.11-1.12M)
(protobuf) faster than 5.1, using 30-39% of its CPU per record.
JSON, 1.4x on protobuf ingest and 1.3x on modified protobuf, using 21-54%
less CPU per record.
memory chunks until the next flush, so its memory grows with throughput;
the Collector sends a batch every 8,192 records. The input can be bounded
with
mem_buf_limit.5.1 modified JSON 16,116 and modified protobuf 20,064; 5.2 protobuf ingest
340; every other scenario 0.
Filesystem buffering
Fluent Bit uses
storage.type: filesystemwithstorage.sync: normalandstorage.checksum: off. The Collector uses thefile_storageextension asits persistent sending queue; neither syncs every write.
(0.57-0.75M)
(0.88-0.91M)
(0.20-0.20M)
(0.88-0.94M)
(0.91-0.94M)
(0.21-0.22M)
(0.20-0.21M)
(0.97-0.97M)
(0.19-0.19M)
(0.19-0.20M)
(0.88-0.97M)
(0.20-0.21M)
modification, and 4.3x to 5.1x faster than the Collector.
ingest and modified workloads run at the same speed.
full the Collector refuses requests and the clients resend them: 1.6 to 2.2
million refused requests per scenario over five runs, against 133 to 20,318
503 answers for Fluent Bit. That retry traffic is most of its CPU per
record and latency.
records.
Method
request. 16 persistent keep-alive clients, one process each.
per scenario, rotating the product order between runs.
cores and their SMT siblings (CPUs 4-7, 12-15). The generator used at most
22% of its CPUs.
threaded: true,flush: 0.5, null output,same configuration for both versions.
GOMAXPROCS=4, exporter sending queue with one consumer,batching with
min_size: 8192andflush_timeout: 0.5s, counting nullexporter.
the clients.
Release,
-O3 -DNDEBUG,FLB_DEBUG=Off,FLB_JEMALLOC=On. Collector:standard Go 1.26 release build (
-trimpath,CGO_ENABLED=0).45abcd0724af355edcb1301ee9b408e16fc051762c79df6d604db2f2a79c7783ad5c7b5812ee668c5fe5f712eb453164bba26227a21ceff8d4a2b1709916d6d2598fe4c93c14be5c839b086e2100b8721493ce9560bc0d52668700cae8041d1302242870ee14ffea9ed77ea76209eb47a29a15eefae4669a27490d9c4838098e2263d4340d14bca3936cc2e0f94e4b54e2518690b774e6a512a45349file_storage+ counting null exporter431debc8eeb931bf77de094870d26274c944329d4ec88c784b4b7352afbb35e9The two 5.2 sources differ only in the busy answer for full ingress queues,
which the memory runs never reached. The harness is
benchmarks/compare_fluent_bit_otelcol.py, with--storage memoryor--storage filesystem.Verification
Every commit in the range builds on its own (Release), and the range passes
the repository commit-prefix checker against
master.Focused CTest:
Result: 9/9 passed.
flb-it-opentelemetryunder Valgrind(
--leak-check=full): 0 errors.Integration scenarios for every HTTP-server-based input:
Result: 607 passed.
The same scenarios, without
internal_http_server, under strict Valgrind(
VALGRIND=1 VALGRIND_STRICT=1): 594 passed, no Valgrind errors. Two casesfailed in that parallel run: a gRPC case could not bind its listener port
because another parallel test held it, and
in_http_max_connections_blocks_and_recoverstimed out waiting for a heldconnection under the full-suite load. Both pass 5/5 when run alone under strict
Valgrind, and the
in_http_max_connectionsandin_httpscenarios pass 96/96in each of four parallel strict Valgrind runs.
A request that finds the HTTP worker ingress queue full is answered with 503,
with and without input processors, for JSON and protobuf:
test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit, normaland under strict Valgrind with no errors.
tests/internal/fuzzers/opentelemetry_logs_fuzzer.cfeeds request bodies tothe JSON and protobuf decoders, CFL and MessagePack, with and without a body
key and with a range of
logs_max_sizelimits. A local run under ASan andUBSan (8 workers, about 17 million inputs, seeded with the parity corpora)
found no crash.
Scope
This PR covers OTLP logs and HTTP worker connection distribution. Metrics and
traces keep their current decoding.