in_opentelemetry: ensure log record bodies are always wrapped by a map

Signed-off-by: Aditya Prajapati <[email protected]>

plugins/in_opentelemetry/opentelemetry_prot.c

@@ -270,7 +270,10 @@ static int otel_pack_array(msgpack_packer *mp_pck,
     int ret;
     int array_index;
 
-    ret = 0;
+    ret = msgpack_pack_array(mp_pck, array->n_values);
+    if (ret != 0) {
+        return ret;
+    }
 
     for (array_index = 0; array_index < array->n_values && ret == 0; array_index++) {
         ret = otlp_pack_any_value(mp_pck, array->values[array_index]);
@@ -332,6 +335,32 @@ static int otlp_pack_any_value(msgpack_packer *mp_pck,
     return result;
 }
 
+static int otel_pack_body(msgpack_packer *mp_pck,
+                          Opentelemetry__Proto__Common__V1__AnyValue *body)
+{
+    int result;
+
+    if (body->value_case != OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_KVLIST_VALUE) {
+        result = msgpack_pack_map(mp_pck, 1);
+        if (result != 0) {
+            return result;
+        }
+
+        result = otel_pack_string(mp_pck, "message");
+        if (result != 0) {
+            return result;
+        }
+    }
+
+    result = otlp_pack_any_value(mp_pck, body);
+    if (result != 0) {
+        flb_error("[otel] Failed to convert log record body");
+        return -1;
+    }
+
+    return result;
+}
+
 static int binary_payload_to_msgpack(msgpack_packer *mp_pck,
                                      uint8_t *in_buf,
                                      size_t in_size)
@@ -385,15 +414,18 @@ static int binary_payload_to_msgpack(msgpack_packer *mp_pck,
 
                 log_record = log_records[log_record_index];
 
-                ret = otlp_pack_any_value(mp_pck, log_record->body);
-
+                ret = otel_pack_body(mp_pck, log_record->body);
                 if (ret != 0) {
-                    flb_error("[otel] Failed to convert log record body");
+                    flb_error("[otel] Failed to pack body field");
                     return -1;
                 }
+
             }
         }
     }
+
+    opentelemetry__proto__collector__logs__v1__export_logs_service_request__free_unpacked(input_logs, NULL);
+
     return 0;
 }