ml: group: stream: Add multiline_truncated: true metadata for truncated records

Signed-off-by: Hiroshi Hatake <[email protected]>

Author: cosmo0920

include/fluent-bit/multiline/flb_ml.h

@@ -113,6 +113,7 @@ struct flb_ml_stream_group {
     msgpack_sbuffer mp_sbuf;    /* temporary msgpack buffer              */
     msgpack_packer mp_pck;      /* temporary msgpack packer              */
     struct flb_time mp_time;    /* multiline time parsed from first line */
+    int truncated;              /* was the buffer truncated?         */
 
     /* parent stream reference */
     struct flb_ml_stream *stream;

src/multiline/flb_ml.c

@@ -1419,6 +1419,14 @@ int flb_ml_flush_stream_group(struct flb_ml_parser *ml_parser,
                                                FLB_TRUE);
         }
 
+        /* If the buffer was truncated, append the marker to the metadata */
+        if (ret == FLB_EVENT_ENCODER_SUCCESS && group->truncated) {
+            ret = flb_log_event_encoder_append_metadata_values(
+                    &mst->ml->log_event_encoder,
+                    FLB_LOG_EVENT_CSTRING_VALUE("multiline_truncated"),
+                    FLB_LOG_EVENT_BOOLEAN_VALUE(FLB_TRUE));
+        }
+
         if (ret == FLB_EVENT_ENCODER_SUCCESS) {
             ret = flb_log_event_encoder_set_body_from_raw_msgpack(
                     &mst->ml->log_event_encoder,
@@ -1457,6 +1465,7 @@ int flb_ml_flush_stream_group(struct flb_ml_parser *ml_parser,
 
     msgpack_sbuffer_destroy(&mp_sbuf);
     flb_sds_len_set(group->buf, 0);
+    group->truncated = FLB_FALSE;
 
     /* Update last flush time */
     group->last_flush = time_ms_now();

src/multiline/flb_ml_group.c

@@ -97,12 +97,14 @@ int flb_ml_group_cat(struct flb_ml_stream_group *group,
     limit = group->stream->ml->buffer_limit;
     if (limit > 0) {
         if (flb_sds_len(group->buf) >= limit) {
+            group->truncated = FLB_TRUE;
             return FLB_MULTILINE_TRUNCATED;
         }
 
         avail = limit - flb_sds_len(group->buf);
         if (len > avail) {
             len = avail;
+            group->truncated = FLB_TRUE;
             status = FLB_MULTILINE_TRUNCATED;
         }
     }

src/multiline/flb_ml_stream.c

@@ -79,6 +79,7 @@ static struct flb_ml_stream_group *stream_group_create(struct flb_ml_stream *mst
     msgpack_sbuffer_init(&group->mp_sbuf);
     msgpack_packer_init(&group->mp_pck, &group->mp_sbuf, msgpack_sbuffer_write);
 
+    group->truncated = FLB_FALSE;
     /* parent stream reference */
     group->stream = mst;