opentelemetry: preserve otlp log resource grouping - #11781
Merged
Conversation
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
📝 WalkthroughWalkthroughAdds 64-bit hash-based resource and scope identity to OTLP logs rendering; computes msgpack object hashes, stores them on internal resource/scope state, and propagates hashes through lookup/append paths. Tests and slow-flush integration scenarios added to validate resource preservation across requests. ChangesHash-Based OTLP Logs Resource/Scope Deduplication
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py (1)
581-599:⚠️ Potential issue | 🟠 Major | ⚡ Quick winAssert the two requests were actually coalesced into one OTLP render.
This test collects resources across all Kafka messages, so it also passes when each request is flushed separately. In that case the resource-id collision path never runs, and the regression stays untested. Please assert the single rendered payload/message contains both resources instead of merging results across messages.
🤖 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/out_kafka/tests/test_out_kafka_001.py` around lines 581 - 599, The test currently aggregates resources across all Kafka messages which hides whether two requests were coalesced; update the assertions to verify a single OTLP message was produced by asserting messages (from service.wait_for_messages) has length 1 and then call _collect_resources on that single message instead of on all messages, then build body_to_user from the resulting resources and assert both "event-a"/"event-b" are present and that len(resources) == 2; refer to symbols messages, service.wait_for_messages, _collect_resources, resources, and body_to_user when making the change.
🤖 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 `@src/opentelemetry/flb_opentelemetry_otlp_proto.c`:
- Around line 1629-1643: resource grouping ignores resource-level schema_url
because resource_hash is computed only from resource_object; change the identity
to include schema_url so resources with same attributes but different
resource-level schema URLs don't collapse. Extract the schema_url from
group_body (the same value passed into append_logs_resource_state via
group_body["schema_url"]), and incorporate it into the hash used for
resource_hash (e.g., combine msgpack_object_hash(resource_object) with a hash of
the schema_url string or compute a single hash over both values) before calling
find_logs_resource_state and append_logs_resource_state; ensure any place that
compares resource_hash (find_logs_resource_state, append_logs_resource_state)
uses the new combined hash so grouping is correct.
---
Outside diff comments:
In `@tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py`:
- Around line 581-599: The test currently aggregates resources across all Kafka
messages which hides whether two requests were coalesced; update the assertions
to verify a single OTLP message was produced by asserting messages (from
service.wait_for_messages) has length 1 and then call _collect_resources on that
single message instead of on all messages, then build body_to_user from the
resulting resources and assert both "event-a"/"event-b" are present and that
len(resources) == 2; refer to symbols messages, service.wait_for_messages,
_collect_resources, resources, and body_to_user when making the change.
🪄 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: 640e32cd-859e-4737-b768-51a901313f75
📒 Files selected for processing (11)
src/opentelemetry/flb_opentelemetry_otlp_json.csrc/opentelemetry/flb_opentelemetry_otlp_proto.ctests/integration/scenarios/in_opentelemetry/config/stdout-otlp-json-slow-flush.yamltests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.pytests/integration/scenarios/out_kafka/config/out_kafka_otlp_json_slow_flush.yamltests/integration/scenarios/out_kafka/config/out_kafka_otlp_proto_slow_flush.yamltests/integration/scenarios/out_kafka/tests/test_out_kafka_001.pytests/integration/scenarios/out_opentelemetry/config/out_otel_grpc_logs_otlp_input_slow_flush.yamltests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_otlp_input_slow_flush.yamltests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.pytests/internal/opentelemetry.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>
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/opentelemetry/flb_opentelemetry_otlp_proto.c (1)
33-33: ⚡ Quick winRecommended refactor: extract the new hash/schema-url helpers into a shared internal header.
msgpack_object_hash,msgpack_object_pair_hash,resource_schema_url_object, andresource_identity_hashare duplicated byte-for-byte insrc/opentelemetry/flb_opentelemetry_otlp_json.c(lines 76-167). Both files also add the same#include <cfl/cfl_hash.h>. As-is, future fixes (e.g., changes to schema-url precedence or hash composition) must be kept in lockstep across both renderers, which is exactly the kind of drift this PR is fixing in the data path.Consider moving them into an internal header (e.g.,
src/opentelemetry/flb_opentelemetry_otlp_hash.{h,c}orflb_opentelemetry_otlp_internal.h) shared between the proto and JSON renderers.Also applies to: 77-168
🤖 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/opentelemetry/flb_opentelemetry_otlp_proto.c` at line 33, Extract the duplicated helpers msgpack_object_hash, msgpack_object_pair_hash, resource_schema_url_object, and resource_identity_hash (and the shared `#include` <cfl/cfl_hash.h>) into a new internal header+source pair (e.g., flb_opentelemetry_otlp_hash.h/.c or flb_opentelemetry_otlp_internal.h/.c), implement the functions once in the .c and expose prototypes in the .h, replace the duplicate definitions in flb_opentelemetry_otlp_proto.c and flb_opentelemetry_otlp_json.c with an `#include` of the new header, ensure symbol visibility (static vs exported) is set appropriately, and update the build files so the new .c is compiled and linked.
🤖 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/internal/opentelemetry.c`:
- Around line 2250-2252: The test asserts decoded->n_resource_logs == 2 using
TEST_CHECK (which is non-fatal) but then unconditionally dereferences
decoded->resource_logs[0] and [1]; wrap the indexed accesses in a guard that
verifies decoded->n_resource_logs >= 2 before accessing resource_logs (e.g., an
if-check around the strcmp TEST_CHECKs) so the test records the failure without
risking invalid memory access when n_resource_logs is wrong; reference decoded,
n_resource_logs, resource_logs and the TEST_CHECK lines to locate and update the
code.
---
Nitpick comments:
In `@src/opentelemetry/flb_opentelemetry_otlp_proto.c`:
- Line 33: Extract the duplicated helpers msgpack_object_hash,
msgpack_object_pair_hash, resource_schema_url_object, and resource_identity_hash
(and the shared `#include` <cfl/cfl_hash.h>) into a new internal header+source
pair (e.g., flb_opentelemetry_otlp_hash.h/.c or
flb_opentelemetry_otlp_internal.h/.c), implement the functions once in the .c
and expose prototypes in the .h, replace the duplicate definitions in
flb_opentelemetry_otlp_proto.c and flb_opentelemetry_otlp_json.c with an
`#include` of the new header, ensure symbol visibility (static vs exported) is set
appropriately, and update the build files so the new .c is compiled and linked.
🪄 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: 7ce64d65-c7ea-4d1f-9747-38f4f1921e3b
📒 Files selected for processing (4)
src/opentelemetry/flb_opentelemetry_otlp_json.csrc/opentelemetry/flb_opentelemetry_otlp_proto.ctests/integration/scenarios/out_kafka/tests/test_out_kafka_001.pytests/internal/opentelemetry.c
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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_opentelemetry/tests/test_in_opentelemetry_001.py`:
- Around line 738-749: The test currently reads OTLP JSON once via
read_stdout_otlp_json then stops the service; instead, wrap the read/assertion
in a polling loop that repeatedly calls read_stdout_otlp_json (or re-reads the
stdout) with a cumulative timeout and only proceeds to service.stop and to
building records when the parsed output contains both expected entries (i.e.,
len(output["resourceLogs"]) >= 2). Update the block around read_stdout_otlp_json
/ iter_log_records / service.stop to retry until the condition is met (or the
timeout elapses), then perform the existing body_to_user and assert checks;
apply the same polling change to the analogous block at the other location
(lines 780-792).
In `@tests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py`:
- Around line 586-607: The test currently calls wait_for_messages(1) and stops
the service before the merged OTLP payload with both resources may be published;
instead, wait for the decoded OTLP payload to contain both expected
resources/bodies by repeatedly calling wait_for_messages (or increasing the
count) and using _collect_resources(format_name, "logs") to check that both
"event-a" and "event-b" are present before calling service.stop; update the test
logic around wait_for_messages, messages, _collect_resources and body_to_user so
the test only stops the service and asserts the resource contents after the
merged payload includes both resources (use a timeout to fail the test if not
observed).
🪄 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: edd62091-8e21-469c-a2fa-130b4259d81f
📒 Files selected for processing (2)
tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.pytests/integration/scenarios/out_kafka/tests/test_out_kafka_001.py
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.
Fluent Bit can preserve OTLP log resources correctly, but we found a bug in the OTLP log re-rendering path used by Kafka format:
otlp_jsonandotlp_proto. When multiple OTLP log requests are flushed in one chunk, per-request internal resource indexes could collide and cause distinct resources to be merged.changes:
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
Bug Fixes
Tests