multiline: new multiline core interface

include/fluent-bit/multiline/flb_ml.h

@@ -0,0 +1,215 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_ML_H
+#define FLB_ML_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_log.h>
+#include <fluent-bit/flb_regex.h>
+#include <fluent-bit/flb_mp.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_parser.h>
+
+/* Types available */
+#define FLB_ML_COUNT     0    /* concatenate N consecutive records  */
+#define FLB_ML_REGEX     1    /* pattern is a regular expression    */
+#define FLB_ML_ENDSWITH  2    /* record key/content ends with 'abc' */
+#define FLB_ML_EQ        3    /* record key/content equaks 'abc'    */
+
+/* Rule types */
+#define FLB_ML_RULE_STATE_START 0
+#define FLB_ML_RULE_STATE_PART  1
+#define FLB_ML_RULE_STATE_END   2
+
+/* Payload types */
+#define FLB_ML_TYPE_TEXT        0   /* raw text */
+#define FLB_ML_TYPE_RECORD      1   /* Fluent Bit msgpack record */
+#define FLB_ML_TYPE_MAP         2   /* msgpack object/map (k/v pairs) */
+
+/* Default multiline buffer size: 4Kb */
+#define FLB_ML_BUF_SIZE  1024*4
+
+struct flb_ml;
+struct flb_ml_stream;
+
+struct flb_ml_rule {
+    /* If the rule contains a 'start_state' this flag is turned on */
+    int start_state;
+
+    /* Name of the rule (state name) */
+    struct mk_list from_states;
+
+    /*
+     * Defined 'to_state' in the rule. Note that this work only at definition
+     * since we do a mapping to multiple possible rules having the same
+     * name.
+     */
+    flb_sds_t to_state;
+
+    struct mk_list to_state_map;
+
+    /* regex content pattern */
+    struct flb_regex *regex;
+
+    /* regex end pattern */
+    struct flb_regex *regex_end;
+
+    struct mk_list _head;
+};
+
+struct flb_ml_stream {
+    flb_sds_t name;           /* Name of the stream, mostly for debugging purposes */
+
+    int counter_lines;        /* counter for the number of lines */
+
+    /* Multiline content buffer */
+    flb_sds_t buf;
+
+    /* internal state */
+    int first_line;           /* first line of multiline message ? */
+
+    struct flb_ml_rule *rule_to_state;
+
+    /* packaging buffers */
+    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 */
+
+    /* Flush callback and opaque data type */
+    int (*cb_flush) (struct flb_ml *,
+                     struct flb_ml_stream *,
+                     void *cb_data,
+                     void *buf_data,
+                     size_t buf_size);
+    void *cb_data;
+
+    struct mk_list _head;
+};
+
+struct flb_ml {
+    int type;                 /* type: COUNT, REGEX, ENDSWITH or EQ */
+    int negate;               /* negate start pattern ? */
+
+    /*
+     * If 'multiline type' is ENDSWITH or EQ, a 'match_str' string is passed
+     * so we can compare against it. We don't use a regex pattern for efficiency.
+     */
+    flb_sds_t match_str;
+
+    /*
+     * The 'key' name that contains the multiline message. For REGEX, ENDSWITH and
+     * EQ types, the conditions are applied to 'key_content' unless 'key_pattern'
+     * is set, on that case 'key_content' is used as a raw buffer and appended
+     * as part of the multiline message.
+     */
+    flb_sds_t key_content;
+
+    /*
+     * Optional: define a 'key' name that matches the pattern to decide if the
+     * line is complete or a continuation.
+     *
+     * This is not mandatory, most of the rules works directly on 'key_content'
+     * but other use-cases like 'CRI' uses a different 'key' to define if the
+     * line is complete or not.
+     */
+    flb_sds_t key_pattern;
+
+    /* Flush interval */
+    int flush_ms;
+
+    /* flush callback */
+    int (*cb_flush)(struct flb_ml *,           /* multiline context */
+                    struct flb_ml_stream *,    /* stream context */
+                    void *,                    /* opaque data */
+                    char *,                    /* text buffer */
+                    size_t);                   /* buffer length */
+
+    void *cb_data;                             /* opaque data */
+
+    /* internal */
+    struct flb_parser *parser;
+
+    /*
+     * Every multiline context has N streams, a stream represent a source
+     * of data. To avoid having 'multiple' context of 'multiline' we use
+     * streams.
+     */
+    struct mk_list streams;
+
+    /*
+     * If multiline type is REGEX, it needs a set of pre-defined rules to deal
+     * with messages.
+     */
+    struct mk_list regex_rules;
+
+    /* Fluent Bit parent context */
+    struct flb_config *config;
+};
+
+struct flb_ml *flb_ml_create(struct flb_config *ctx,
+                             int type, char *match_str, int negate,
+                             int flush_ms,
+                             char *key_content,
+                             char *key_pattern,
+                             struct flb_parser *parser);
+
+int flb_ml_destroy(struct flb_ml *ml);
+
+int flb_ml_register_context(struct flb_ml *ml, struct flb_ml_stream *mst,
+                            struct flb_time *tm, msgpack_object *map);
+
+int flb_ml_append(struct flb_ml *ml, struct flb_ml_stream *mst,
+                  int type,
+                  struct flb_time *tm, void *buf, size_t size);
+int flb_ml_append_object(struct flb_ml *ml,
+                         struct flb_ml_stream *mst,
+                         struct flb_time *tm, msgpack_object *obj);
+
+int flb_ml_flush(struct flb_ml *ml, struct flb_ml_stream *mst);
+
+/* Multiline streams */
+struct flb_ml_stream *flb_ml_stream_create(struct flb_ml *ml,
+                                           char *name,
+                                           int (*cb_flush) (struct flb_ml *,
+                                                            struct flb_ml_stream *,
+                                                            void *cb_data,
+                                                            void *buf_data,
+                                                            size_t buf_size),
+                                           void *cb_data);
+
+int flb_ml_stream_destroy(struct flb_ml_stream *mst);
+
+/* Regex Rules */
+int flb_ml_rule_create(struct flb_ml *ml,
+                       flb_sds_t from_states,
+                       char *regex_pattern,
+                       flb_sds_t to_state,
+                       char *end_pattern);
+void flb_ml_rule_destroy(struct flb_ml_rule *rule);
+void flb_ml_rule_destroy_all(struct flb_ml *ml);
+
+int flb_ml_init(struct flb_ml *ml);
+
+#include "flb_ml_mode.h"
+
+#endif

include/fluent-bit/multiline/flb_ml_mode.h

@@ -0,0 +1,42 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_ML_MODE_H
+#define FLB_ML_MODE_H
+
+/* Docker container mode */
+struct flb_ml *flb_ml_mode_docker(struct flb_config *config, int flush_ms);
+
+/* CRI container mode */
+struct flb_ml *flb_ml_mode_cri(struct flb_config *config, int flush_ms);
+
+/* Python language mode */
+struct flb_ml *flb_ml_mode_python(struct flb_config *config,
+                                  int flush_ms, char *key);
+
+/* Java language mode */
+struct flb_ml *flb_ml_mode_java(struct flb_config *config,
+                                int flush_ms, char *key);
+
+/* Go language mode */
+struct flb_ml *flb_ml_mode_go(struct flb_config *config,
+                              int flush_ms, char *key);
+
+#endif

include/fluent-bit/multiline/flb_ml_rule.h

@@ -0,0 +1,43 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_ML_RULE_H
+#define FLB_ML_RULE_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/multiline/flb_ml.h>
+
+int flb_ml_rule_create(struct flb_ml *ml,
+                       flb_sds_t from_states,
+                       char *regex_pattern,
+                       flb_sds_t to_state,
+                       char *end_pattern);
+void flb_ml_rule_destroy(struct flb_ml_rule *rule);
+void flb_ml_rule_destroy_all(struct flb_ml *ml);
+int flb_ml_rule_process(struct flb_ml *ml,
+                        struct flb_ml_stream *mst,
+                        msgpack_object *full_map,
+                        void *buf, size_t size, struct flb_time *tm,
+                        msgpack_object *val_content,
+                        msgpack_object *val_pattern);
+int flb_ml_rule_init(struct flb_ml *ml);
+
+#endif

src/CMakeLists.txt

@@ -56,6 +56,13 @@ set(src
   flb_routes_mask.c
   )
 
+# Multiline subsystem
+add_subdirectory(multiline)
+set(src
+  ${src}
+  ${src_multiline}
+  )
+
 if(FLB_SYSTEM_WINDOWS)
   set(src
     ${src}

src/flb_config.c

@@ -345,6 +345,11 @@ void flb_config_exit(struct flb_config *config)
     }
 #endif
 
+#ifdef FLB_HAVE_PARSER
+    /* parsers */
+    flb_parser_exit(config);
+#endif
+
     if (config->storage_path) {
         flb_free(config->storage_path);
     }

src/flb_engine.c

@@ -713,11 +713,6 @@ int flb_engine_shutdown(struct flb_config *config)
     /* router */
     flb_router_exit(config);
 
-#ifdef FLB_HAVE_PARSER
-    /* parsers */
-    flb_parser_exit(config);
-#endif
-
     /* cleanup plugins */
     flb_filter_exit(config);
     flb_input_exit_all(config);

src/multiline/CMakeLists.txt

@@ -0,0 +1,9 @@
+set(src_multiline
+  multiline/flb_ml_mode_cri.c
+  multiline/flb_ml_mode_docker.c
+  multiline/flb_ml_mode_python.c
+  multiline/flb_ml_mode_java.c
+  multiline/flb_ml_mode_go.c
+  multiline/flb_ml_rule.c
+  multiline/flb_ml.c PARENT_SCOPE
+  )