Skip to content

otlp_metadata, cat: keep metadata and metric semantics across mixed-version pipelines #315

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
Oct 23, 2026
Merged

otlp_metadata, cat: keep metadata and metric semantics across mixed-version pipelines #315

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
37a6705
cat: clear stale OTLP point metadata when the source has none
edsiper Oct 22, 2026
99a8bd1
opentelemetry: keep points whose attributes hold empty nested values
edsiper Oct 22, 2026
b11af08
mpack: report oversized array counts as zero when peeking
edsiper Oct 22, 2026
46c1002
otlp_metadata: bind point metadata by point identity
edsiper Oct 22, 2026
83fcce0
cat: preserve family semantics and validate contributors
edsiper Oct 22, 2026
ad80cf5
tests: check OTLP data points through every pipeline
edsiper Oct 22, 2026
ec49ff8
tests: decode MessagePack written by released Fluent Bit versions
edsiper Oct 22, 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
228 changes: 196 additions & 32 deletions src/cmt_cat.c
View file
Open in desktop

Large diffs are not rendered by default.

41 changes: 33 additions & 8 deletions src/cmt_decode_opentelemetry.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,11 @@
#include <limits.h>
#include <string.h>

static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyValue *source);
/* Distinguish absent optional values from allocation failures. */
#define CLONE_UNSUPPORTED_VALUE -1

static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyValue *source,
int *status);

static int clone_array(struct cfl_array *target,
Opentelemetry__Proto__Common__V1__ArrayValue *source);
Expand All @@ -63,14 +67,18 @@ static int clone_exemplars_to_kvlist(struct cfl_kvlist *target,
Opentelemetry__Proto__Metrics__V1__Exemplar **exemplars,
size_t exemplar_count);

static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyValue *source)
static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyValue *source,
int *status)
{
struct cfl_kvlist *new_child_kvlist;
struct cfl_array *new_child_array;
struct cfl_variant *result_instance = NULL;
int result;

if (source == NULL) {
*status = CMT_DECODE_OPENTELEMETRY_ALLOCATION_ERROR;
if (source == NULL || source->value_case ==
OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE__NOT_SET) {
*status = CLONE_UNSUPPORTED_VALUE;
return NULL;
}
if (source->value_case == OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_STRING_VALUE) {
Expand Down Expand Up @@ -105,6 +113,7 @@ static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyVa
result = clone_kvlist(new_child_kvlist, source->kvlist_value);
if (result) {
cfl_variant_destroy(result_instance);
*status = result;

return NULL;
}
Expand All @@ -126,6 +135,7 @@ static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyVa
result = clone_array(new_child_array, source->array_value);
if (result) {
cfl_variant_destroy(result_instance);
*status = result;

return NULL;
}
Expand All @@ -134,6 +144,13 @@ static struct cfl_variant *clone_variant(Opentelemetry__Proto__Common__V1__AnyVa
result_instance = cfl_variant_create_from_bytes((char *) source->bytes_value.data, source->bytes_value.len,
CFL_FALSE);
}
else {
*status = CLONE_UNSUPPORTED_VALUE;
}

if (result_instance != NULL) {
*status = CMT_DECODE_OPENTELEMETRY_SUCCESS;
}

return result_instance;
}
Expand Down Expand Up @@ -167,9 +184,9 @@ static int clone_array_entry(struct cfl_array *target,
return CMT_DECODE_OPENTELEMETRY_SUCCESS;
}

new_child_instance = clone_variant(source);
new_child_instance = clone_variant(source, &result);
if (new_child_instance == NULL) {
return CMT_DECODE_OPENTELEMETRY_ALLOCATION_ERROR;
return result;
}

result = cfl_array_append(target, new_child_instance);
Expand Down Expand Up @@ -214,10 +231,10 @@ static int clone_kvlist_entry(struct cfl_kvlist *target,
return CMT_DECODE_OPENTELEMETRY_SUCCESS;
}

new_child_instance = clone_variant(source->value);
new_child_instance = clone_variant(source->value, &result);

if (new_child_instance == NULL) {
return CMT_DECODE_OPENTELEMETRY_ALLOCATION_ERROR;
return result;
}

result = cfl_kvlist_insert(target, source->key, new_child_instance);
Expand Down Expand Up @@ -378,6 +395,7 @@ static int clone_point_attributes(struct cfl_kvlist *metadata,
{
struct cfl_kvlist *copy;
size_t index;
int result;

copy = cfl_kvlist_create();
if (copy == NULL) {
Expand All @@ -390,7 +408,14 @@ static int clone_point_attributes(struct cfl_kvlist *metadata,
OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE__NOT_SET)) {
continue;
}
if (attributes[index] == NULL || clone_kvlist_entry(copy, attributes[index]) != 0) {
result = clone_kvlist_entry(copy, attributes[index]);
if (result == CLONE_UNSUPPORTED_VALUE) {
/* A partial typed tree would change the attributes. Retain the
* label-derived representation when any nested value is absent. */
cfl_kvlist_destroy(copy);
return 0;
}
if (attributes[index] == NULL || result != 0) {
cfl_kvlist_destroy(copy);
return -1;
}
Expand Down
20 changes: 14 additions & 6 deletions src/cmt_encode_opentelemetry.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,8 @@ static char *map_type_to_key(int map_type)
}
}

static struct cfl_kvlist *get_metric_otlp_metadata_context(struct cmt *cmt, struct cmt_map *map)
static struct cfl_kvlist *get_legacy_metric_otlp_metadata_context(struct cmt *cmt,
struct cmt_map *map)
{
struct cfl_kvlist *otlp_root;
struct cfl_kvlist *metrics_root;
Expand All @@ -155,10 +156,6 @@ static struct cfl_kvlist *get_metric_otlp_metadata_context(struct cmt *cmt, stru
return NULL;
}

if (map->otlp_metadata) {
return map->otlp_metadata;
}

otlp_root = fetch_metadata_kvlist_key(cmt->external_metadata, "otlp");
if (otlp_root == NULL) {
return NULL;
Expand All @@ -177,6 +174,15 @@ static struct cfl_kvlist *get_metric_otlp_metadata_context(struct cmt *cmt, stru
return fetch_metadata_kvlist_key(type_root, map->opts->fqname);
}

static struct cfl_kvlist *get_metric_otlp_metadata_context(struct cmt *cmt, struct cmt_map *map)
{
if (map != NULL && map->otlp_metadata) {
return map->otlp_metadata;
}

return get_legacy_metric_otlp_metadata_context(cmt, map);
}

static struct cfl_kvlist *get_data_point_otlp_metadata_context(struct cmt *cmt,
struct cmt_map *map,
struct cmt_metric *sample)
Expand All @@ -189,7 +195,9 @@ static struct cfl_kvlist *get_data_point_otlp_metadata_context(struct cmt *cmt,
return sample->otlp_metadata;
}

metric_context = get_metric_otlp_metadata_context(cmt, map);
/* Native family metadata never carries points. A point an intermediary
* left unbound keeps its legacy entry, as released readers do. */
metric_context = get_legacy_metric_otlp_metadata_context(cmt, map);
if (metric_context == NULL) {
return NULL;
}
Expand Down
6 changes: 6 additions & 0 deletions src/cmt_mpack_utils.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,5 +332,11 @@ int cmt_mpack_peek_array_length(mpack_reader_t *reader)
return 0;
}

/* Callers size allocations from this value before the array is read, so
* a declared count that cmt_mpack_unpack_array() rejects reports zero. */
if (CMT_MPACK_MAX_ARRAY_ENTRY_COUNT < mpack_tag_array_count(&tag)) {
return 0;
}

return mpack_tag_array_count(&tag);
}
Loading
Loading