Skip to content

opentelemetry: preserve otlp log resource grouping #11781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 7 commits into from
May 7, 2026
Merged

opentelemetry: preserve otlp log resource grouping #11781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d9c7dec
opentelemetry: preserve otlp log resource grouping
edsiper May 6, 2026
ed82eeb
tests: internal: cover otlp log resource collisions
edsiper May 6, 2026
c04b319
tests: integration: cover otlp log resource collisions
edsiper May 6, 2026
b8e3037
opentelemetry: include schema url in log grouping
edsiper May 6, 2026
c0d8a79
tests: internal: cover otlp resource schema url grouping
edsiper May 6, 2026
703065f
tests: integration: require kafka otlp coalescing coverage
edsiper May 6, 2026
8d49043
tests: integration: cover otlp resource schema url grouping
edsiper May 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 140 additions & 18 deletions src/opentelemetry/flb_opentelemetry_otlp_json.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@
#include <ctraces/ctraces.h>
#include <ctraces/ctr_decode_msgpack.h>

#include <cfl/cfl_hash.h>

#include <msgpack.h>
#include <fluent-bit/flb_json.h>

Expand All @@ -50,12 +52,14 @@

struct otlp_logs_scope_state {
int64_t scope_id;
uint64_t scope_hash;
struct flb_json_mut_val *scope_log;
struct flb_json_mut_val *log_records;
};

struct otlp_logs_resource_state {
int64_t resource_id;
uint64_t resource_hash;
struct flb_json_mut_val *resource_log;
struct flb_json_mut_val *scope_logs;
struct otlp_logs_scope_state *scopes;
Expand All @@ -69,6 +73,99 @@ struct otlp_metrics_scope_state {
static msgpack_object *msgpack_map_get_object(msgpack_object_map *map,
const char *key);

static uint64_t msgpack_object_hash(msgpack_object *object)
{
uint64_t hash;
msgpack_sbuffer buffer;
msgpack_packer packer;

if (object == NULL) {
return cfl_hash_64bits("null", 4);
}

msgpack_sbuffer_init(&buffer);
msgpack_packer_init(&packer, &buffer, msgpack_sbuffer_write);

if (msgpack_pack_object(&packer, *object) != 0) {
msgpack_sbuffer_destroy(&buffer);
return 0;
}

hash = cfl_hash_64bits(buffer.data, buffer.size);
msgpack_sbuffer_destroy(&buffer);

return hash;
}

static uint64_t msgpack_object_pair_hash(msgpack_object *left,
msgpack_object *right)
{
uint64_t hash;
msgpack_sbuffer buffer;
msgpack_packer packer;

msgpack_sbuffer_init(&buffer);
msgpack_packer_init(&packer, &buffer, msgpack_sbuffer_write);

if (msgpack_pack_array(&packer, 2) != 0) {
msgpack_sbuffer_destroy(&buffer);
return 0;
}

if (left == NULL) {
msgpack_pack_nil(&packer);
}
else if (msgpack_pack_object(&packer, *left) != 0) {
msgpack_sbuffer_destroy(&buffer);
return 0;
}

if (right == NULL) {
msgpack_pack_nil(&packer);
}
else if (msgpack_pack_object(&packer, *right) != 0) {
msgpack_sbuffer_destroy(&buffer);
return 0;
}

hash = cfl_hash_64bits(buffer.data, buffer.size);
msgpack_sbuffer_destroy(&buffer);

return hash;
}

static msgpack_object *resource_schema_url_object(msgpack_object *resource_object,
msgpack_object *resource_body)
{
msgpack_object *schema_url;

if (resource_body != NULL && resource_body->type == MSGPACK_OBJECT_MAP) {
schema_url = msgpack_map_get_object(&resource_body->via.map, "schema_url");
if (schema_url != NULL) {
return schema_url;
}
}

if (resource_object != NULL && resource_object->type == MSGPACK_OBJECT_MAP) {
schema_url = msgpack_map_get_object(&resource_object->via.map, "schema_url");
if (schema_url != NULL) {
return schema_url;
}
}

return NULL;
}

static uint64_t resource_identity_hash(msgpack_object *resource_object,
msgpack_object *resource_body)
{
msgpack_object *schema_url;

schema_url = resource_schema_url_object(resource_object, resource_body);

return msgpack_object_pair_hash(resource_object, schema_url);
}

static void set_result(int *result, int value)
{
if (result != NULL) {
Expand Down Expand Up @@ -890,12 +987,14 @@ static msgpack_object *msgpack_map_get_object(msgpack_object_map *map,
static struct otlp_logs_resource_state *find_logs_resource_state(
struct otlp_logs_resource_state *states,
size_t state_count,
int64_t resource_id)
int64_t resource_id,
uint64_t resource_hash)
{
size_t index;

for (index = 0; index < state_count; index++) {
if (states[index].resource_id == resource_id) {
if (states[index].resource_id == resource_id &&
states[index].resource_hash == resource_hash) {
return &states[index];
}
}
Expand All @@ -905,12 +1004,14 @@ static struct otlp_logs_resource_state *find_logs_resource_state(

static struct otlp_logs_scope_state *find_logs_scope_state(
struct otlp_logs_resource_state *resource,
int64_t scope_id)
int64_t scope_id,
uint64_t scope_hash)
{
size_t index;

for (index = 0; index < resource->scope_count; index++) {
if (resource->scopes[index].scope_id == scope_id) {
if (resource->scopes[index].scope_id == scope_id &&
resource->scopes[index].scope_hash == scope_hash) {
return &resource->scopes[index];
}
}
Expand Down Expand Up @@ -1036,6 +1137,7 @@ static struct otlp_logs_resource_state *append_logs_resource_state(
struct otlp_logs_resource_state **states,
size_t *state_count,
int64_t resource_id,
uint64_t resource_hash,
msgpack_object *resource_object,
msgpack_object *resource_body)
{
Expand All @@ -1060,16 +1162,14 @@ static struct otlp_logs_resource_state *append_logs_resource_state(
return NULL;
}

if (resource_body != NULL && resource_body->type == MSGPACK_OBJECT_MAP) {
schema_url = msgpack_map_get_object(&resource_body->via.map, "schema_url");
if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) {
if (!flb_json_mut_obj_add_strncpy(doc,
resource_log,
"schemaUrl",
schema_url->via.str.ptr,
schema_url->via.str.size)) {
return NULL;
}
schema_url = resource_schema_url_object(resource_object, resource_body);
if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) {
if (!flb_json_mut_obj_add_strncpy(doc,
resource_log,
"schemaUrl",
schema_url->via.str.ptr,
schema_url->via.str.size)) {
return NULL;
}
}

Expand All @@ -1086,6 +1186,7 @@ static struct otlp_logs_resource_state *append_logs_resource_state(
memset(state, 0, sizeof(struct otlp_logs_resource_state));

state->resource_id = resource_id;
state->resource_hash = resource_hash;
state->resource_log = resource_log;
state->scope_logs = scope_logs;

Expand All @@ -1098,6 +1199,7 @@ static struct otlp_logs_scope_state *append_logs_scope_state(
struct flb_json_mut_doc *doc,
struct otlp_logs_resource_state *resource_state,
int64_t scope_id,
uint64_t scope_hash,
msgpack_object *scope_object)
{
struct otlp_logs_scope_state *new_scopes;
Expand Down Expand Up @@ -1147,6 +1249,7 @@ static struct otlp_logs_scope_state *append_logs_scope_state(
memset(state, 0, sizeof(struct otlp_logs_scope_state));

state->scope_id = scope_id;
state->scope_hash = scope_hash;
state->scope_log = scope_log;
state->log_records = log_records;

Expand All @@ -1163,27 +1266,36 @@ static int ensure_default_logs_scope_state(
struct otlp_logs_resource_state **current_resource,
struct otlp_logs_scope_state **current_scope)
{
uint64_t resource_hash;
uint64_t scope_hash;

resource_hash = resource_identity_hash(NULL, NULL);
scope_hash = msgpack_object_hash(NULL);

*current_resource = find_logs_resource_state(*resource_states,
*resource_state_count,
0);
0,
resource_hash);
if (*current_resource == NULL) {
*current_resource = append_logs_resource_state(doc,
resource_logs,
resource_states,
resource_state_count,
0,
resource_hash,
NULL,
NULL);
if (*current_resource == NULL) {
return -1;
}
}

*current_scope = find_logs_scope_state(*current_resource, 0);
*current_scope = find_logs_scope_state(*current_resource, 0, scope_hash);
if (*current_scope == NULL) {
*current_scope = append_logs_scope_state(doc,
*current_resource,
0,
scope_hash,
NULL);
if (*current_scope == NULL) {
return -1;
Expand Down Expand Up @@ -1483,6 +1595,8 @@ static flb_sds_t flb_opentelemetry_logs_to_otlp_json_render(
msgpack_object *group_metadata;
msgpack_object *resource_object;
msgpack_object *scope_object;
uint64_t resource_hash;
uint64_t scope_hash;
flb_sds_t json;
struct flb_log_event event;
struct flb_log_event_decoder decoder;
Expand Down Expand Up @@ -1600,15 +1714,20 @@ static flb_sds_t flb_opentelemetry_logs_to_otlp_json_render(
scope_object = msgpack_map_get_object(&group_body->via.map, "scope");
}

resource_hash = resource_identity_hash(resource_object, group_body);
scope_hash = msgpack_object_hash(scope_object);

current_resource = find_logs_resource_state(resource_states,
resource_state_count,
resource_id);
resource_id,
resource_hash);
if (current_resource == NULL) {
current_resource = append_logs_resource_state(doc,
resource_logs,
&resource_states,
&resource_state_count,
resource_id,
resource_hash,
resource_object,
group_body);
if (current_resource == NULL) {
Expand All @@ -1620,11 +1739,14 @@ static flb_sds_t flb_opentelemetry_logs_to_otlp_json_render(
}
}

current_scope = find_logs_scope_state(current_resource, scope_id);
current_scope = find_logs_scope_state(current_resource,
scope_id,
scope_hash);
if (current_scope == NULL) {
current_scope = append_logs_scope_state(doc,
current_resource,
scope_id,
scope_hash,
scope_object);
if (current_scope == NULL) {
flb_log_event_decoder_destroy(&decoder);
Expand Down
Loading
Loading