config: multiline: in_tail: filter_multiline: Add configurable buffer limit for multiline interface

include/fluent-bit/flb_config.h

@@ -145,6 +145,7 @@ struct flb_config {
 
     /* Multiline core parser definitions */
     struct mk_list multiline_parsers;
+    char *multiline_buffer_limit; /* limit for multiline concatenated data */
 
     /* Outputs instances */
     struct mk_list outputs;             /* list of output plugins   */
@@ -408,6 +409,9 @@ enum conf_type {
 /* Coroutines */
 #define FLB_CONF_STR_CORO_STACK_SIZE "Coro_Stack_Size"
 
+/* Multiline */
+#define FLB_CONF_STR_MULTILINE_BUFFER_LIMIT "multiline_buffer_limit"
+
 /* Scheduler */
 #define FLB_CONF_STR_SCHED_CAP        "scheduler.cap"
 #define FLB_CONF_STR_SCHED_BASE       "scheduler.base"

include/fluent-bit/flb_utils.h

@@ -49,6 +49,7 @@ void flb_utils_split_free_entry(struct flb_split_entry *entry);
 void flb_utils_split_free(struct mk_list *list);
 int flb_utils_timer_consume(flb_pipefd_t fd);
 int64_t flb_utils_size_to_bytes(const char *size);
+int64_t flb_utils_size_to_binary_bytes(const char *size);
 int64_t flb_utils_hex2int(char *hex, int len);
 int flb_utils_time_to_seconds(const char *time);
 int flb_utils_pipe_byte_consume(flb_pipefd_t fd);

include/fluent-bit/multiline/flb_ml.h

@@ -52,6 +52,15 @@
 /* Default multiline buffer size: 4Kb */
 #define FLB_ML_BUF_SIZE         1024*4
 
+/* Default limit for concatenated multiline messages: 2MB */
+#define FLB_ML_BUFFER_LIMIT_DEFAULT_STR "2MB"
+#define FLB_ML_BUFFER_LIMIT_DEFAULT     (1024 * 1024 * 2)
+
+/* Return codes */
+#define FLB_MULTILINE_OK         0
+#define FLB_MULTILINE_PROCESSED  1 /* Reserved */
+#define FLB_MULTILINE_TRUNCATED  2
+
 /* Maximum number of groups per stream */
 #define FLB_ML_MAX_GROUPS       6
 
@@ -106,6 +115,10 @@ 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;
 
     struct mk_list _head;
 };
@@ -275,6 +288,9 @@ struct flb_ml {
     struct flb_log_event_encoder log_event_encoder;
     struct flb_log_event_decoder log_event_decoder;
     struct flb_config *config;             /* Fluent Bit context */
+
+    /* Limit for concatenated multiline messages */
+    size_t buffer_limit;
 };
 
 struct flb_ml *flb_ml_create(struct flb_config *ctx, char *name);

include/fluent-bit/multiline/flb_ml_group.h

@@ -28,4 +28,12 @@ struct flb_ml_group *flb_ml_group_create(struct flb_ml *ml);
 void flb_ml_group_destroy(struct flb_ml_group *group);
 int flb_ml_group_add_parser(struct flb_ml *ctx, struct flb_ml_parser_ins *p);
 
+/*
+ * Append data to a multiline stream group respecting the configured
+ * buffer limit. The length of the appended data might be reduced if
+ * the limit is reached.
+ */
+int flb_ml_group_cat(struct flb_ml_stream_group *group,
+                     const char *data, size_t len);
+
 #endif

plugins/filter_multiline/ml.c

@@ -375,6 +375,17 @@ static int cb_ml_init(struct flb_filter_instance *ins,
                         "emit_records", ctx->ins->metrics);
 #endif
     }
+        /* Truncated metrics always should be existing. */
+#ifdef FLB_HAVE_METRICS
+        ctx->cmt_truncated = cmt_counter_create(ins->cmt,
+                                                "fluentbit", "filter", "emit_truncated_total",
+                                                "Total number of truncated occurence of multiline",
+                                                1, (char *[]) {"name"});
+
+        /* OLD api */
+        flb_metrics_add(FLB_MULTILINE_METRIC_TRUNCATED,
+                        "emit_truncated", ctx->ins->metrics);
+#endif
 
     mk_list_init(&ctx->ml_streams);
     mk_list_init(&ctx->split_message_packers);
@@ -780,6 +791,10 @@ static int cb_ml_filter(const void *data, size_t bytes,
     struct flb_log_event         event;
     int                          ret;
     struct ml_ctx               *ctx;
+#ifdef FLB_HAVE_METRICS
+    uint64_t ts;
+    char *name;
+#endif
 
     (void) f_ins;
     (void) config;
@@ -820,7 +835,19 @@ static int cb_ml_filter(const void *data, size_t bytes,
                FLB_EVENT_DECODER_SUCCESS) {
             ret = flb_ml_append_event(ctx->m, ctx->stream_id, &event);
 
-            if (ret != 0) {
+            if (ret == FLB_MULTILINE_TRUNCATED) {
+                flb_plg_warn(ctx->ins,
+                             "multiline message truncated due to buffer limit");
+#ifdef FLB_HAVE_METRICS
+                name = (char *) flb_filter_name(ctx->ins);
+                ts = cfl_time_now();
+                cmt_counter_inc(ctx->cmt_truncated, ts, 1, (char *[]) {name});
+
+                /* old api */
+                flb_metrics_sum(FLB_MULTILINE_METRIC_TRUNCATED, 1, ctx->ins->metrics);
+#endif
+            }
+            else if (ret != FLB_MULTILINE_OK) {
                 flb_plg_debug(ctx->ins,
                               "could not append object from tag: %s", tag);
             }
@@ -871,10 +898,32 @@ static int cb_ml_filter(const void *data, size_t bytes,
                FLB_EVENT_DECODER_SUCCESS) {
             ret = flb_ml_append_event(ctx->m, stream->stream_id, &event);
 
-            if (ret != 0) {
+            if (ret == FLB_MULTILINE_TRUNCATED) {
+                flb_plg_warn(ctx->ins,
+                             "multiline message truncated due to buffer limit");
+#ifdef FLB_HAVE_METRICS
+                name = (char *) flb_filter_name(ctx->ins);
+                ts = cfl_time_now();
+                cmt_counter_inc(ctx->cmt_truncated, ts, 1, (char *[]) {name});
+
+                /* old api */
+                flb_metrics_sum(FLB_MULTILINE_METRIC_TRUNCATED, 1, ctx->ins->metrics);
+#endif
+            }
+            else if (ret != FLB_MULTILINE_OK) {
                 flb_plg_debug(ctx->ins,
                               "could not append object from tag: %s", tag);
             }
+            else if (ret == FLB_MULTILINE_OK) {
+#ifdef FLB_HAVE_METRICS
+                name = (char *) flb_filter_name(ctx->ins);
+                ts = cfl_time_now();
+                cmt_counter_inc(ctx->cmt_emitted, ts, 1, (char *[]) {name});
+
+                /* old api */
+                flb_metrics_sum(FLB_MULTILINE_METRIC_EMITTED, 1, ctx->ins->metrics);
+#endif
+            }
         }
 
         flb_log_event_decoder_destroy(&decoder);

plugins/filter_multiline/ml.h

@@ -25,10 +25,11 @@
 
 #define FLB_MULTILINE_MEM_BUF_LIMIT_DEFAULT  "10M"
 #define FLB_MULTILINE_METRIC_EMITTED         200
+#define FLB_MULTILINE_METRIC_TRUNCATED       201
 #define FLB_MULTILINE_MODE_PARTIAL_MESSAGE   "partial_message"
 #define FLB_MULTILINE_MODE_PARSER            "parser"
 
-/* 
+/*
  * input instance + tag is the unique identifier
  * for a multiline stream
  * TODO: implement clean up of streams that haven't been used recently
@@ -77,6 +78,7 @@ struct ml_ctx {
 
 #ifdef FLB_HAVE_METRICS
     struct cmt_counter *cmt_emitted;
+    struct cmt_counter *cmt_truncated;
 #endif
 };
 

plugins/in_tail/tail_config.c

@@ -479,13 +479,21 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
                                                 "Total number of rotated files",
                                                 1, (char *[]) {"name"});
 
+    ctx->cmt_multiline_truncated = \
+            cmt_counter_create(ins->cmt,
+                               "fluentbit", "input",
+                               "multiline_truncated_total",
+                               "Total number of truncated occurences for multilines",
+                               1, (char *[]) {"name"});
     /* OLD metrics */
     flb_metrics_add(FLB_TAIL_METRIC_F_OPENED,
                     "files_opened", ctx->ins->metrics);
     flb_metrics_add(FLB_TAIL_METRIC_F_CLOSED,
                     "files_closed", ctx->ins->metrics);
     flb_metrics_add(FLB_TAIL_METRIC_F_ROTATED,
                     "files_rotated", ctx->ins->metrics);
+    flb_metrics_add(FLB_TAIL_METRIC_M_TRUNCATED,
+                    "multiline_truncated", ctx->ins->metrics);
 #endif
 
     return ctx;

plugins/in_tail/tail_config.h

@@ -41,6 +41,7 @@
 #define FLB_TAIL_METRIC_F_OPENED  100  /* number of opened files  */
 #define FLB_TAIL_METRIC_F_CLOSED  101  /* number of closed files  */
 #define FLB_TAIL_METRIC_F_ROTATED 102  /* number of rotated files */
+#define FLB_TAIL_METRIC_M_TRUNCATED 103  /* number of truncated occurrences of multiline */
 #endif
 
 struct flb_tail_config {
@@ -167,6 +168,7 @@ struct flb_tail_config {
     struct cmt_counter *cmt_files_opened;
     struct cmt_counter *cmt_files_closed;
     struct cmt_counter *cmt_files_rotated;
+    struct cmt_counter *cmt_multiline_truncated;
 
     /* Hash: hash tables for quick acess to registered files */
     struct flb_hash_table *static_hash;

plugins/in_tail/tail_file.c

@@ -447,6 +447,11 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
 #ifdef FLB_HAVE_UNICODE_ENCODER
     size_t decoded_len;
 #endif
+#ifdef FLB_HAVE_METRICS
+    uint64_t ts;
+    char *name;
+#endif
+
 
     ctx = (struct flb_tail_config *) file->config;
 
@@ -563,6 +568,17 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
                                      &out_time,
                                      line,
                                      line_len);
+            if (ret == FLB_MULTILINE_TRUNCATED) {
+                flb_plg_warn(ctx->ins, "multiline message truncated due to buffer limit");
+#ifdef FLB_HAVE_METRICS
+                name = (char *) flb_input_name(ctx->ins);
+                ts = cfl_time_now();
+                cmt_counter_inc(ctx->cmt_multiline_truncated, ts, 1, (char *[]) {name});
+
+                /* Old api */
+                flb_metrics_sum(FLB_TAIL_METRIC_M_TRUNCATED, 1, ctx->ins->metrics);
+#endif
+            }
             goto go_next;
         }
         else if (ctx->docker_mode) {

src/flb_config.c

@@ -169,6 +169,10 @@ struct flb_service_config service_configs[] = {
      FLB_CONF_TYPE_INT,
      offsetof(struct flb_config, coro_stack_size)},
 
+    {FLB_CONF_STR_MULTILINE_BUFFER_LIMIT,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, multiline_buffer_limit)},
+
     /* Scheduler */
     {FLB_CONF_STR_SCHED_CAP,
      FLB_CONF_TYPE_INT,
@@ -361,6 +365,7 @@ struct flb_config *flb_config_init()
      * on we use flb_config_exit to cleanup the config, which requires
      * the config->multiline_parsers list to be initialized. */
     mk_list_init(&config->multiline_parsers);
+    config->multiline_buffer_limit = FLB_ML_BUFFER_LIMIT_DEFAULT_STR;
 
     /* Task map */
     ret = flb_config_task_map_resize(config, FLB_CONFIG_DEFAULT_TASK_MAP_SIZE);

src/flb_utils.c

@@ -607,6 +607,92 @@ int64_t flb_utils_size_to_bytes(const char *size)
     return (int64_t)val;
 }
 
+int64_t flb_utils_size_to_binary_bytes(const char *size)
+{
+    int i;
+    int len;
+    int plen = 0;
+    double val;
+    char tmp[4] = {0};
+    int64_t KiB = 1024;
+    int64_t MiB = 1024 * KiB;
+    int64_t GiB = 1024 * MiB;
+
+    if (!size) {
+        return -1;
+    }
+
+    if (strcasecmp(size, "false") == 0) {
+        return 0;
+    }
+
+    len = strlen(size);
+    val = atof(size);
+
+    if (len == 0) {
+        return -1;
+    }
+
+    for (i = len - 1; i >= 0; i--) {
+        if (isalpha(size[i])) {
+            plen++;
+        }
+        else {
+            break;
+        }
+    }
+
+    if (plen == 0) {
+        return (int64_t)val;
+    }
+    else if (plen > 3) {
+        return -1;
+    }
+
+    for (i = 0; i < plen; i++) {
+        tmp[i] = toupper(size[len - plen + i]);
+    }
+
+    if (plen == 2) {
+        if (tmp[1] != 'B') {
+            return -1;
+        }
+    }
+    if (plen == 3) {
+        if (tmp[1] != 'I' || tmp[2] != 'B') {
+            return -1;
+        }
+    }
+
+    if (tmp[0] == 'K') {
+        /* set upper bound (2**64/KiB)/2 to avoid overflows */
+        if (val >= 9223372036854775.0 || val <= -9223372036854774.0)
+        {
+            return -1;
+        }
+        return (int64_t)(val * KiB);
+    }
+    else if (tmp[0] == 'M') {
+        /* set upper bound (2**64/MiB)/2 to avoid overflows */
+        if (val >= 9223372036854.0 || val <= -9223372036853.0) {
+            return -1;
+        }
+        return (int64_t)(val * MiB);
+    }
+    else if (tmp[0] == 'G') {
+        /* set upper bound (2**64/GiB)/2 to avoid overflows */
+        if (val >= 9223372036.0 || val <= -9223372035.0) {
+            return -1;
+        }
+        return (int64_t)(val * GiB);
+    }
+    else {
+        return -1;
+    }
+
+    return (int64_t)val;
+}
+
 int64_t flb_utils_hex2int(char *hex, int len)
 {
     int i = 0;