out_opentelemetry: fix behavior of record attributes fix the merge process

The log attribute code now merges additional attributes into the existing array
instead of overwriting it by calling otlp_kvarray_append.

A new helper otlp_kvarray_append grows an array of OTLP key/value pairs and copies the
new entries into it.

Added a unit test test_kvarray_append validating that two key/value arrays are
combined correctly.

Signed-off-by: Eduardo Silva <[email protected]>

Author: edsiper

plugins/out_opentelemetry/opentelemetry_logs.c

@@ -392,7 +392,9 @@ static int append_v1_logs_metadata_and_fields(struct opentelemetry_context *ctx,
     int severity_text_set = FLB_FALSE;
     int severity_number_set = FLB_FALSE;
     int trace_flags_set = FLB_FALSE;
+    size_t attr_count = 0;
     struct flb_ra_value *ra_val;
+    Opentelemetry__Proto__Common__V1__KeyValue **attrs = NULL;
 
     if (ctx == NULL || event == NULL || log_record == NULL) {
         return -1;
@@ -519,21 +521,43 @@ static int append_v1_logs_metadata_and_fields(struct opentelemetry_context *ctx,
     ra_val = flb_ra_get_value_object(ctx->ra_log_meta_otlp_attr, *event->metadata);
     if (ra_val != NULL) {
         if (ra_val->o.type == MSGPACK_OBJECT_MAP) {
-            if (log_record->attributes != NULL) {
-                otlp_kvarray_destroy(log_record->attributes, log_record->n_attributes);
+            attr_count = 0;
+            attrs = msgpack_map_to_otlp_kvarray(&ra_val->o, &attr_count);
+            if (attrs) {
+                if (log_record->attributes != NULL) {
+                    if (otlp_kvarray_append(&log_record->attributes,
+                                            &log_record->n_attributes,
+                                            attrs, attr_count) != 0) {
+                        otlp_kvarray_destroy(attrs, attr_count);
+                    }
+                }
+                else {
+                    log_record->attributes = attrs;
+                    log_record->n_attributes = attr_count;
+                }
             }
-            log_record->attributes = msgpack_map_to_otlp_kvarray(&ra_val->o, &log_record->n_attributes);
         }
         flb_ra_key_value_destroy(ra_val);
     }
     else if (ctx->ra_attributes_metadata) {
         ra_val = flb_ra_get_value_object(ctx->ra_attributes_metadata, *event->metadata);
         if (ra_val != NULL) {
             if (ra_val->o.type == MSGPACK_OBJECT_MAP) {
-                if (log_record->attributes != NULL) {
-                    otlp_kvarray_destroy(log_record->attributes, log_record->n_attributes);
+                attr_count = 0;
+                attrs = msgpack_map_to_otlp_kvarray(&ra_val->o, &attr_count);
+                if (attrs) {
+                    if (log_record->attributes != NULL) {
+                        if (otlp_kvarray_append(&log_record->attributes,
+                                                &log_record->n_attributes,
+                                                attrs, attr_count) != 0) {
+                            otlp_kvarray_destroy(attrs, attr_count);
+                        }
+                    }
+                    else {
+                        log_record->attributes = attrs;
+                        log_record->n_attributes = attr_count;
+                    }
                 }
-                log_record->attributes = msgpack_map_to_otlp_kvarray(&ra_val->o, &log_record->n_attributes);
             }
             flb_ra_key_value_destroy(ra_val);
         }

plugins/out_opentelemetry/opentelemetry_utils.c

@@ -167,6 +167,33 @@ void otlp_kvarray_destroy(Opentelemetry__Proto__Common__V1__KeyValue **kvarray,
     }
 }
 
+int otlp_kvarray_append(Opentelemetry__Proto__Common__V1__KeyValue ***base,
+                        size_t *base_count,
+                        Opentelemetry__Proto__Common__V1__KeyValue **extra,
+                        size_t extra_count)
+{
+    size_t new_count;
+    Opentelemetry__Proto__Common__V1__KeyValue **tmp;
+
+    if (extra == NULL || extra_count == 0) {
+        return 0;
+    }
+
+    new_count = *base_count + extra_count;
+    tmp = flb_realloc(*base, new_count * sizeof(Opentelemetry__Proto__Common__V1__KeyValue *));
+    if (!tmp) {
+        return -1;
+    }
+
+    *base = tmp;
+    memcpy(*base + *base_count, extra,
+           extra_count * sizeof(Opentelemetry__Proto__Common__V1__KeyValue *));
+    *base_count = new_count;
+    flb_free(extra);
+
+    return 0;
+}
+
 void otlp_kvpair_destroy(Opentelemetry__Proto__Common__V1__KeyValue *kvpair)
 {
     if (kvpair == NULL) {

plugins/out_opentelemetry/opentelemetry_utils.h

@@ -26,6 +26,10 @@ Opentelemetry__Proto__Common__V1__KeyValueList *otlp_kvlist_value_initialize(siz
 Opentelemetry__Proto__Common__V1__AnyValue *otlp_any_value_initialize(int data_type, size_t entry_count);
 
 void otlp_kvarray_destroy(Opentelemetry__Proto__Common__V1__KeyValue **kvarray, size_t entry_count);
+int otlp_kvarray_append(Opentelemetry__Proto__Common__V1__KeyValue ***base,
+                        size_t *base_count,
+                        Opentelemetry__Proto__Common__V1__KeyValue **extra,
+                        size_t extra_count);
 void otlp_kvpair_destroy(Opentelemetry__Proto__Common__V1__KeyValue *kvpair);
 void otlp_kvlist_destroy(Opentelemetry__Proto__Common__V1__KeyValueList *kvlist);
 void otlp_array_destroy(Opentelemetry__Proto__Common__V1__ArrayValue *array);