pack: add metadata and group info into log JSON record (fix #10258)

Before this patch, the JSON log record encoder only supported the log
record body as an output content, missing support for log record metadata
and group metadata and group attributes (if defined).

This patch makes the JSON encoder add support for the missing data, packaging into
the record a new parent key called '__internal__' that holds the following inside it:

  - group_attributes: group header attributes
  - log_metadata: log record metadata

Note that for cases where the origin of the data was OTLP and as part of the groups
we add 'group metadata' (in the origin), this is not being used since it only contains
internal references that are not used for this JSON export: we do not aim to reassmble
this JSON into a Fluent Bit record.

Consider the following original OTLP JSON record ingested into the pipeline:

{
  "resourceLogs": [
    {
      "resource": {
        "attributes": [
          {
            "key": "service.name",
            "value": {
              "stringValue": "my.service"
            }
          }
        ]
      },
      "scopeLogs": [
        {
          "scope": {
            "name": "my.library",
            "version": "1.0.0",
            "attributes": [
              {
                "key": "my.scope.attribute",
                "value": {
                  "stringValue": "some scope attribute"
                }
              }
            ]
          },
          "logRecords": [
            {
              "timeUnixNano": "1544712660300000000",
              "observedTimeUnixNano": "1544712660300000000",
              "severityNumber": 10,
              "severityText": "Information",
              "traceId": "5B8EFFF798038103D269B633813FC60C",
              "spanId": "EEE19B7EC3C1B174",
              "body": {
                "stringValue": "Example log record"
              },
              "attributes": [
                {
                  "key": "string.attribute",
                  "value": {
                    "stringValue": "some string"
                  }
                },
                {
                  "key": "boolean.attribute",
                  "value": {
                    "boolValue": true
                  }
                },
                {
                  "key": "int.attribute",
                  "value": {
                    "intValue": "10"
                  }
                },
                {
                  "key": "double.attribute",
                  "value": {
                    "doubleValue": 637.704
                  }
                },
                {
                  "key": "array.attribute",
                  "value": {
                    "arrayValue": {
                      "values": [
                        {
                          "stringValue": "many"
                        },
                        {
                          "stringValue": "values"
                        }
                      ]
                    }
                  }
                },
                {
                  "key": "map.attribute",
                  "value": {
                    "kvlistValue": {
                      "values": [
                        {
                          "key": "some.map.key",
                          "value": {
                            "stringValue": "some value"
                          }
                        }
                      ]
                    }
                  }
                }
              ]
            }
          ],
          "schemaUrl": "https://example.com/schema"
        }
      ],
      "schemaUrl": "https://example.com/schema"
    }
  ]
}

now when encoded in msgpack and then JSON the output will be:

{
  "date": 1544712658.274599,
  "__internal__": {
    "group_attributes": {
      "resource": {
        "attributes": {
          "service.name": "my.service"
        }
      },
      "scope": {
        "name": "my.library",
        "version": "1.0.0",
        "attributes": {
          "my.scope.attribute": "some scope attribute"
        }
      }
    },
    "log_metadata": {
      "otlp": {
        "observed_timestamp": 1544712660300000000,
        "severity_number": 10,
        "severity_text": "Information",
        "attributes": {
          "string.attribute": "some string",
          "boolean.attribute": true,
          "int.attribute": 10,
          "double.attribute": 637.704,
          "array.attribute": [
            "many",
            "values"
          ],
          "map.attribute": {
            "some.map.key": "some value"
          }
        },
        "trace_id": "5B8EFFF798038103D269B633813FC60C",
        "span_id": "EEE19B7EC3C1B174"
      }
    }
  },
  "log": "Example log record"
}

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

Author: edsiper

src/flb_pack.c

@@ -30,6 +30,9 @@
 #include <fluent-bit/flb_pack.h>
 #include <fluent-bit/flb_unescape.h>
 
+#include <fluent-bit/flb_log_event_encoder.h>
+#include <fluent-bit/flb_log_event_decoder.h>
+
 /* cmetrics */
 #include <cmetrics/cmetrics.h>
 #include <cmetrics/cmt_decode_msgpack.h>
@@ -917,23 +920,21 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
                                           flb_sds_t date_key)
 {
     int i;
-    int ok = MSGPACK_UNPACK_SUCCESS;
-    int records = 0;
-    int map_size;
-    size_t off = 0;
+    int ret;
     char time_formatted[38];
     flb_sds_t out_tmp;
     flb_sds_t out_js;
     flb_sds_t out_buf = NULL;
-    msgpack_unpacked result;
-    msgpack_object root;
-    msgpack_object map;
     msgpack_sbuffer tmp_sbuf;
     msgpack_packer tmp_pck;
-    msgpack_object *obj;
     msgpack_object *k;
     msgpack_object *v;
     struct flb_time tms;
+    struct flb_log_event_decoder log_decoder;
+    struct flb_log_event log_event;
+    struct flb_mp_map_header mh_array;
+    struct flb_mp_map_header mh_map;
+    struct flb_mp_map_header mh_internal;
 
     /* For json lines and streams mode we need a pre-allocated buffer */
     if (json_format == FLB_PACK_JSON_FORMAT_LINES ||
@@ -949,6 +950,15 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
     msgpack_sbuffer_init(&tmp_sbuf);
     msgpack_packer_init(&tmp_pck, &tmp_sbuf, msgpack_sbuffer_write);
 
+    ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes);
+    if (ret != FLB_EVENT_DECODER_SUCCESS) {
+        flb_error("Log event decoder initialization error : %d", ret);
+        if (out_buf) {
+            flb_sds_destroy(out_buf);
+        }
+        return NULL;
+    }
+
     /*
      * If the format is the original msgpack style of one big array,
      * registrate the array, otherwise is not necessary. FYI, original format:
@@ -960,43 +970,25 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
      * ]
      */
     if (json_format == FLB_PACK_JSON_FORMAT_JSON) {
-        records = flb_mp_count(data, bytes);
-        if (records <= 0) {
-            msgpack_sbuffer_destroy(&tmp_sbuf);
-            return NULL;
-        }
-        msgpack_pack_array(&tmp_pck, records);
+        /* register the array. Note must be finalized with flb_mp_map_header_end() */
+        flb_mp_array_header_init(&mh_array, &tmp_pck);
     }
 
-    msgpack_unpacked_init(&result);
-    while (msgpack_unpack_next(&result, data, bytes, &off) == ok) {
-        /* Each array must have two entries: time and record */
-        root = result.data;
-        if (root.type != MSGPACK_OBJECT_ARRAY) {
-            continue;
-        }
-        if (root.via.array.size != 2) {
-            continue;
+    /* Iterate log records */
+    while ((ret = flb_log_event_decoder_next(&log_decoder, &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
+        if (json_format == FLB_PACK_JSON_FORMAT_JSON) {
+            /* register a new entry for the array entry */
+            flb_mp_array_header_append(&mh_array);
         }
+        tms = log_event.timestamp;
 
-        /* Unpack time */
-        flb_time_pop_from_msgpack(&tms, &result, &obj);
-
-        /* Get the record/map */
-        map = root.via.array.ptr[1];
-        if (map.type != MSGPACK_OBJECT_MAP) {
-            continue;
-        }
-        map_size = map.via.map.size;
+        /* initialize the map for the record key/values */
+        flb_mp_map_header_init(&mh_map, &tmp_pck);
 
+        /* date key */
         if (date_key != NULL) {
-            msgpack_pack_map(&tmp_pck, map_size + 1);
-        }
-        else {
-            msgpack_pack_map(&tmp_pck, map_size);
-        }
+            flb_mp_array_header_append(&mh_map);
 
-        if (date_key != NULL) {
             /* Append date key */
             msgpack_pack_str(&tmp_pck, flb_sds_len(date_key));
             msgpack_pack_str_body(&tmp_pck, date_key, flb_sds_len(date_key));
@@ -1011,7 +1003,7 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
                                                     FLB_PACK_JSON_DATE_JAVA_SQL_TIMESTAMP_FMT, ".%06" PRIu64)) {
                     flb_sds_destroy(out_buf);
                     msgpack_sbuffer_destroy(&tmp_sbuf);
-                    msgpack_unpacked_destroy(&result);
+                    flb_log_event_decoder_destroy(&log_decoder);
                     return NULL;
                 }
                 break;
@@ -1020,7 +1012,7 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
                                                     FLB_PACK_JSON_DATE_ISO8601_FMT, ".%06" PRIu64 "Z")) {
                     flb_sds_destroy(out_buf);
                     msgpack_sbuffer_destroy(&tmp_sbuf);
-                    msgpack_unpacked_destroy(&result);
+                    flb_log_event_decoder_destroy(&log_decoder);
                     return NULL;
                 }
                 break;
@@ -1033,12 +1025,77 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
             }
         }
 
-        /* Append remaining keys/values */
-        for (i = 0; i < map_size; i++) {
-            k = &map.via.map.ptr[i].key;
-            v = &map.via.map.ptr[i].val;
-            msgpack_pack_object(&tmp_pck, *k);
-            msgpack_pack_object(&tmp_pck, *v);
+        /* register __internal__ data that comes from the group information */
+        if ((log_event.group_attributes && log_event.group_attributes->type == MSGPACK_OBJECT_MAP && log_event.group_attributes->via.map.size > 0) ||
+            (log_event.metadata && log_event.metadata->type == MSGPACK_OBJECT_MAP && log_event.metadata->via.map.size > 0)) {
+
+            flb_mp_map_header_append(&mh_map);
+            msgpack_pack_str(&tmp_pck, 12);
+            msgpack_pack_str_body(&tmp_pck, "__internal__", 12);
+
+            flb_mp_map_header_init(&mh_internal, &tmp_pck);
+
+            /*
+             * group metadata: the JSON export of this record do not aim to be re-assembled into a Fluent pipeline,
+             * actually it is a generic JSON representation of the log record. For this reason, we need to add the group
+             * metadata to the JSON output.
+             *
+             * Just leaving this code commented as a reference...
+             */
+
+            /*
+             * if (log_event.group_metadata != NULL) {
+             *    flb_mp_map_header_append(&mh_internal);
+             *    msgpack_pack_str(&tmp_pck, 14);
+             *    msgpack_pack_str_body(&tmp_pck, "group_metadata", 14);
+             *    msgpack_pack_object(&tmp_pck, *log_event.group_metadata);
+             * }
+             */
+
+            /* Append group attributes */
+            if (log_event.group_attributes != NULL) {
+                flb_mp_map_header_append(&mh_internal);
+                msgpack_pack_str(&tmp_pck, 16);
+                msgpack_pack_str_body(&tmp_pck, "group_attributes", 16);
+                msgpack_pack_object(&tmp_pck, *log_event.group_attributes);
+            }
+
+            /* log/record metadata if exists */
+            if (log_event.metadata != NULL) {
+                flb_mp_map_header_append(&mh_internal);
+                msgpack_pack_str(&tmp_pck, 12);
+                msgpack_pack_str_body(&tmp_pck, "log_metadata", 12);
+                msgpack_pack_object(&tmp_pck, *log_event.metadata);
+            }
+
+            /* finalize the internal map */
+            flb_mp_map_header_end(&mh_internal);
+        }
+
+        /* Append keys/values from the log body */
+        if (log_event.body != NULL) {
+            if (log_event.body->type == MSGPACK_OBJECT_MAP) {
+                for (i = 0; i < log_event.body->via.map.size; i++) {
+                    flb_mp_map_header_append(&mh_map);
+                    k = &log_event.body->via.map.ptr[i].key;
+                    v = &log_event.body->via.map.ptr[i].val;
+
+                    /* Append key/value */
+                    msgpack_pack_object(&tmp_pck, *k);
+                    msgpack_pack_object(&tmp_pck, *v);
+                }
+
+                flb_mp_map_header_end(&mh_map);
+            }
+            else {
+                /* for any other data type, nest the content inside log */
+                flb_mp_map_header_append(&mh_map);
+                msgpack_pack_str(&tmp_pck, 4);
+                msgpack_pack_str_body(&tmp_pck, "log", 3);
+                msgpack_pack_object(&tmp_pck, *log_event.body);
+
+                flb_mp_map_header_end(&mh_map);
+            }
         }
 
         /*
@@ -1071,7 +1128,7 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
             if (!out_js) {
                 flb_sds_destroy(out_buf);
                 msgpack_sbuffer_destroy(&tmp_sbuf);
-                msgpack_unpacked_destroy(&result);
+                flb_log_event_decoder_destroy(&log_decoder);
                 return NULL;
             }
 
@@ -1084,7 +1141,7 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
                 flb_sds_destroy(out_js);
                 flb_sds_destroy(out_buf);
                 msgpack_sbuffer_destroy(&tmp_sbuf);
-                msgpack_unpacked_destroy(&result);
+                flb_log_event_decoder_destroy(&log_decoder);
                 return NULL;
             }
 
@@ -1102,7 +1159,7 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
                 if (!out_tmp) {
                     flb_sds_destroy(out_buf);
                     msgpack_sbuffer_destroy(&tmp_sbuf);
-                    msgpack_unpacked_destroy(&result);
+                    flb_log_event_decoder_destroy(&log_decoder);
                     return NULL;
                 }
                 if (out_tmp != out_buf) {
@@ -1113,14 +1170,18 @@ flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
         }
     }
 
-    /* Release the unpacker */
-    msgpack_unpacked_destroy(&result);
+    /* destroy the decoder */
+    flb_log_event_decoder_destroy(&log_decoder);
+
+    /* finalize the main array */
+    if (json_format == FLB_PACK_JSON_FORMAT_JSON) {
+        flb_mp_array_header_end(&mh_array);
+    }
 
     /* Format to JSON */
     if (json_format == FLB_PACK_JSON_FORMAT_JSON) {
         out_buf = flb_msgpack_raw_to_json_sds(tmp_sbuf.data, tmp_sbuf.size);
         msgpack_sbuffer_destroy(&tmp_sbuf);
-
         if (!out_buf) {
             return NULL;
         }