in_tail: add new 'multiline.parser' option and feature

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

Author: edsiper

plugins/in_tail/tail.c

@@ -649,6 +649,12 @@ static struct flb_config_map config_map[] = {
      "Parser_2 ab2, Parser_N abN."
     },
 
+    /* Multiline Core Engine based API */
+    {
+     FLB_CONFIG_MAP_CLIST, "multiline.parser", NULL,
+     FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct flb_tail_config, multiline_parsers),
+     "specify one or multiple multiline parsers: docker, cri, go, java, etc."
+    },
 #endif
 
     /* EOF */

plugins/in_tail/tail_config.c

@@ -20,6 +20,8 @@
 
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/multiline/flb_ml.h>
+#include <fluent-bit/multiline/flb_ml_parser.h>
 
 #include <stdlib.h>
 #include <fcntl.h>
@@ -35,6 +37,43 @@
 #include "tail_multiline.h"
 #endif
 
+static int multiline_load_parsers(struct flb_tail_config *ctx)
+{
+    struct mk_list *head;
+    struct mk_list *head_p;
+    struct flb_config_map_val *mv;
+    struct flb_slist_entry *val = NULL;
+    struct flb_ml_parser_ins *parser_i;
+
+    if (!ctx->multiline_parsers) {
+        return 0;
+    }
+
+    /* Create Multiline context using the plugin instance name */
+    ctx->ml_ctx = flb_ml_create(ctx->config, ctx->ins->name);
+    if (!ctx->ml_ctx) {
+        return -1;
+    }
+
+    /*
+     * Iterate all 'multiline.parser' entries. Every entry is considered
+     * a group which can have multiple multiline parser instances.
+     */
+    flb_config_map_foreach(head, mv, ctx->multiline_parsers) {
+        mk_list_foreach(head_p, mv->val.list) {
+            val = mk_list_entry(head_p, struct flb_slist_entry, _head);
+
+            /* Create an instance of the defined parser */
+            parser_i = flb_ml_parser_instance_create(ctx->ml_ctx, val->str);
+            if (!parser_i) {
+                return -1;
+            }
+        }
+    }
+
+    return 0;
+}
+
 struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
                                                struct flb_config *config)
 {
@@ -50,6 +89,7 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
         flb_errno();
         return NULL;
     }
+    ctx->config = config;
     ctx->ins = ins;
     ctx->ignore_older = 0;
     ctx->skip_long_lines = FLB_FALSE;
@@ -319,6 +359,27 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
     }
 #endif
 
+#ifdef FLB_HAVE_PARSER
+    /* Multiline core API */
+    if (ctx->multiline_parsers && mk_list_size(ctx->multiline_parsers) > 0) {
+        ret = multiline_load_parsers(ctx);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins, "could not load multiline parsers");
+            flb_tail_config_destroy(ctx);
+            return NULL;
+        }
+
+        /* Enable auto-flush routine */
+        ret = flb_ml_auto_flush_init(ctx->ml_ctx);
+        if (ret == -1) {
+            flb_plg_error(ctx->ins, "could not start multiline auto-flush");
+            flb_tail_config_destroy(ctx);
+            return NULL;
+        }
+        flb_plg_info(ctx->ins, "multiline core started");
+    }
+#endif
+
 #ifdef FLB_HAVE_METRICS
     flb_metrics_add(FLB_TAIL_METRIC_F_OPENED,
                     "files_opened", ctx->ins->metrics);
@@ -336,6 +397,10 @@ int flb_tail_config_destroy(struct flb_tail_config *config)
 
 #ifdef FLB_HAVE_PARSER
     flb_tail_mult_destroy(config);
+
+    if (config->ml_ctx) {
+        flb_ml_destroy(config->ml_ctx);
+    }
 #endif
 
     /* Close pipe ends */

plugins/in_tail/tail_config.h

@@ -29,6 +29,9 @@
 #ifdef FLB_HAVE_REGEX
 #include <fluent-bit/flb_regex.h>
 #endif
+#ifdef FLB_HAVE_PARSER
+#include <fluent-bit/multiline/flb_ml.h>
+#endif
 
 /* Metrics */
 #ifdef FLB_HAVE_METRICS
@@ -111,6 +114,10 @@ struct flb_tail_config {
     int docker_mode_flush;     /* Docker mode flush/wait */
     struct flb_parser *docker_mode_parser; /* Parser for separate multiline logs */
 
+    /* Multiline core engine */
+    struct flb_ml *ml_ctx;
+    struct mk_list *multiline_parsers;
+
     /* Lists head for files consumed statically (read) and by events (inotify) */
     struct mk_list files_static;
     struct mk_list files_event;
@@ -123,6 +130,8 @@ struct flb_tail_config {
 
     /* Plugin input instance */
     struct flb_input_instance *ins;
+
+    struct flb_config *config;
 };
 
 struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,

plugins/in_tail/tail_file.c

@@ -248,19 +248,24 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
             continue;
         }
 
-        /* Empty line (just \n) */
-        if (len == 0) {
+        /*
+         * Empty line (just \n): we skip empty lines only if we are NOT using
+         * multiline core mode.
+         */
+        if (len == 0 && !ctx->ml_ctx) {
             data++;
             processed_bytes++;
             continue;
         }
 
         /* Process '\r\n' */
-        crlf = (data[len-1] == '\r');
-        if (len == 1 && crlf) {
-            data += 2;
-            processed_bytes += 2;
-            continue;
+        if (len >= 2) {
+            crlf = (data[len-1] == '\r');
+            if (len == 1 && crlf) {
+                data += 2;
+                processed_bytes += 2;
+                continue;
+            }
         }
 
         /* Reset time for each line */
@@ -269,7 +274,14 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
         line = data;
         line_len = len - crlf;
         repl_line = NULL;
-        if (ctx->docker_mode) {
+
+        if (ctx->ml_ctx) {
+            ret = flb_ml_append(ctx->ml_ctx, file->ml_stream_id,
+                                FLB_ML_TYPE_TEXT,
+                                &out_time, line, line_len);
+            goto go_next;
+        }
+        else if (ctx->docker_mode) {
             ret = flb_tail_dmode_process_content(now, line, line_len,
                                                  &repl_line, &repl_line_len,
                                                  file, ctx, out_sbuf, out_pck);
@@ -627,11 +639,26 @@ static int set_file_position(struct flb_tail_config *ctx,
     return 0;
 }
 
+static int ml_flush_callback(struct flb_ml_parser *parser,
+                             struct flb_ml_stream *mst,
+                             void *data, char *buf_data, size_t buf_size)
+{
+    struct flb_tail_file *file = data;
+
+    flb_input_chunk_append_raw(file->config->ins,
+                               file->tag_buf,
+                               file->tag_len,
+                               buf_data,
+                               buf_size);
+    return 0;
+}
+
 int flb_tail_file_append(char *path, struct stat *st, int mode,
                          struct flb_tail_config *ctx)
 {
     int fd;
     int ret;
+    uint64_t stream_id;
     size_t len;
     char *tag;
     size_t tag_len;
@@ -716,6 +743,22 @@ int flb_tail_file_append(char *path, struct stat *st, int mode,
     file->skip_next = FLB_FALSE;
     file->skip_warn = FLB_FALSE;
 
+    /* Multiline core mode */
+    if (ctx->ml_ctx) {
+        /* Create a stream for this file */
+        ret = flb_ml_stream_create(ctx->ml_ctx,
+                                   file->name, file->name_len,
+                                   ml_flush_callback, file,
+                                   &stream_id);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins,
+                          "could not create multiline stream for file: %s",
+                          file->name);
+            goto error;
+        }
+        file->ml_stream_id = stream_id;
+    }
+
     /* Local buffer */
     file->buf_size = ctx->buf_chunk_size;
     file->buf_data = flb_malloc(file->buf_size);
@@ -816,6 +859,11 @@ void flb_tail_file_remove(struct flb_tail_file *file)
     flb_plg_debug(ctx->ins, "inode=%"PRIu64" removing file name %s",
                   file->inode, file->name);
 
+    /* remove the multiline.core stream */
+    if (ctx->ml_ctx && file->ml_stream_id > 0) {
+        flb_ml_stream_id_destroy_all(ctx->ml_ctx, file->ml_stream_id);
+    }
+
     if (file->rotated > 0) {
 #ifdef FLB_HAVE_SQLDB
         /*
@@ -833,6 +881,7 @@ void flb_tail_file_remove(struct flb_tail_file *file)
     flb_sds_destroy(file->dmode_lastline);
     mk_list_del(&file->_head);
     flb_tail_fs_remove(ctx, file);
+
     /* avoid deleting file with -1 fd */
     if (file->fd != -1) {
         close(file->fd);

plugins/in_tail/tail_file_internal.h

@@ -26,6 +26,10 @@
 #include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_time.h>
 
+#ifdef FLB_HAVE_PARSER
+#include <fluent-bit/multiline/flb_ml.h>
+#endif
+
 #include "tail.h"
 #include "tail_config.h"
 
@@ -67,6 +71,9 @@ struct flb_tail_file {
     bool dmode_complete;        /* buffer contains completed log         */
     bool dmode_firstline;       /* dmode mult firstline found ?          */
 
+    /* multiline engine: file stream_id */
+    uint64_t ml_stream_id;
+
     /* buffering */
     size_t parsed;
     size_t buf_len;