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   */
@@ -409,6 +410,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

include/fluent-bit/multiline/flb_ml_parser.h

@@ -23,6 +23,40 @@
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_parser.h>
 
+/* fwd decl */
+struct flb_config;
+
+/*
+ * Size based parameter bag for creating multiline parsers.
+ * - Call flb_ml_parser_params_default(name) to obtain defaults,
+ *   then tweak only the fields you need before passing to v2.
+ */
+struct flb_ml_parser_params {
+    uint16_t size;      /* sizeof(struct flb_ml_parser_params) */
+
+    /* creation parameters (mirror of old positional args) */
+    char *name;
+    int   type;         /* FLB_ML_REGEX / FLB_ML_ENDSWITH / FLB_ML_EQ */
+    char *match_str;    /* used for ENDSWITH/EQ; NULL for REGEX */
+    int   negate;       /* 0/1 */
+    int   flush_ms;     /* default: FLB_ML_FLUSH_TIMEOUT */
+    char *key_content;
+    char *key_group;
+    char *key_pattern;
+    struct flb_parser *parser_ctx; /* immediate */
+    char *parser_name;             /* delayed init */
+
+    /* for future toggles */
+    uint32_t flags;
+};
+
+/* Fill sane defaults */
+struct flb_ml_parser_params flb_ml_parser_params_default(const char *name);
+
+/* New initializer with params */
+struct flb_ml_parser *flb_ml_parser_create_params(struct flb_config *ctx,
+                                                  const struct flb_ml_parser_params *p);
+
 int flb_ml_parser_init(struct flb_ml_parser *ml_parser);
 
 int flb_ml_parser_builtin_create(struct flb_config *config);

plugins/filter_multiline/ml.c

@@ -369,12 +369,36 @@ static int cb_ml_init(struct flb_filter_instance *ins,
                                             "fluentbit", "filter", "emit_records_total",
                                             "Total number of emitted records",
                                             1, (char *[]) {"name"});
+        if (!ctx->cmt_emitted) {
+            flb_errno();
+            flb_free(ctx);
+
+            return -1;
+        }
 
         /* OLD api */
         flb_metrics_add(FLB_MULTILINE_METRIC_EMITTED,
                         "emit_records", ctx->ins->metrics);
 #endif
     }
+        /* Register a metric to count the number of emitted records */
+        /* 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"});
+        if (!ctx->cmt_truncated) {
+            flb_errno();
+            flb_free(ctx);
+
+            return -1;
+        }
+
+        /* 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 +804,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,10 +848,34 @@ 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);
             }
+            else if (ret == FLB_MULTILINE_OK) {
+#ifdef FLB_HAVE_METRICS
+                name = (char *) flb_filter_name(ctx->ins);
+                ts = cfl_time_now();
+                if (ctx->cmt_emitted) {
+                    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);
@@ -871,10 +923,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

@@ -481,6 +481,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;
 
@@ -596,6 +601,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,
@@ -362,6 +366,12 @@ 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_strdup(FLB_ML_BUFFER_LIMIT_DEFAULT_STR);
+    if (config->multiline_buffer_limit == NULL) {
+        flb_errno();
+        flb_config_exit(config);
+        return NULL;
+    }
 
     /* Task map */
     ret = flb_config_task_map_resize(config, FLB_CONFIG_DEFAULT_TASK_MAP_SIZE);
@@ -468,6 +478,12 @@ void flb_config_exit(struct flb_config *config)
         }
     }
 
+    /* free heap-owned multiline_buffer_limit if set */
+    if (config->multiline_buffer_limit) {
+        flb_free(config->multiline_buffer_limit);
+        config->multiline_buffer_limit = NULL;
+    }
+
     if (config->env) {
         flb_env_destroy(config->env);
     }

src/flb_config_map.c

@@ -97,6 +97,7 @@ static struct mk_list *parse_string_map_to_list(struct flb_config_map *map, char
 
     if (ret == -1) {
         flb_error("[config map] error reading list of options");
+        flb_slist_destroy(list);
         flb_free(list);
         return NULL;
     }

src/flb_slist.c

@@ -188,7 +188,10 @@ int flb_slist_split_tokens(struct mk_list *list, const char *str, int max_split)
 
     buf = (char *) str;
     while ((tmp = token_retrieve(&buf))) {
-        flb_slist_add_sds(list, tmp);
+        if (flb_slist_add_sds(list, tmp) == -1) {
+            flb_sds_destroy(tmp);
+            return -1;
+        }
         if (!buf) {
             break;
         }
@@ -202,7 +205,9 @@ int flb_slist_split_tokens(struct mk_list *list, const char *str, int max_split)
             }
 
             if (*p) {
-                flb_slist_add(list, p);
+                if (flb_slist_add(list, p) == -1) {
+                    return -1;
+                }
             }
             break;
         }