diff options
Diffstat (limited to 'fluent-bit/src/multiline')
| -rw-r--r-- | fluent-bit/src/multiline/CMakeLists.txt | 15 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml.c | 1562 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_group.c | 86 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_mode.c | 111 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser.c | 347 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser_cri.c | 81 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser_docker.c | 110 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser_go.c | 140 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser_java.c | 143 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser_python.c | 98 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_parser_ruby.c | 87 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_rule.c | 421 | ||||
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml_stream.c | 338 |
13 files changed, 0 insertions, 3539 deletions
diff --git a/fluent-bit/src/multiline/CMakeLists.txt b/fluent-bit/src/multiline/CMakeLists.txt deleted file mode 100644 index 294ef3e8f..000000000 --- a/fluent-bit/src/multiline/CMakeLists.txt +++ /dev/null @@ -1,15 +0,0 @@ -set(src_multiline - # built-in parsers - multiline/flb_ml_parser_cri.c - multiline/flb_ml_parser_docker.c - multiline/flb_ml_parser_python.c - multiline/flb_ml_parser_java.c - multiline/flb_ml_parser_go.c - multiline/flb_ml_parser_ruby.c - # core - multiline/flb_ml_stream.c - multiline/flb_ml_parser.c - multiline/flb_ml_group.c - multiline/flb_ml_rule.c - multiline/flb_ml.c PARENT_SCOPE - ) diff --git a/fluent-bit/src/multiline/flb_ml.c b/fluent-bit/src/multiline/flb_ml.c deleted file mode 100644 index 28e123cee..000000000 --- a/fluent-bit/src/multiline/flb_ml.c +++ /dev/null @@ -1,1562 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_mem.h> -#include <fluent-bit/flb_log.h> -#include <fluent-bit/flb_regex.h> -#include <fluent-bit/flb_time.h> -#include <fluent-bit/flb_pack.h> -#include <fluent-bit/flb_scheduler.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <fluent-bit/multiline/flb_ml_group.h> - -#include <stdarg.h> -#include <math.h> - -static inline int match_negate(struct flb_ml_parser *ml_parser, int matched) -{ - int rule_match = matched; - - /* Validate pattern matching against expected 'negate' condition */ - if (matched == FLB_TRUE) { - if (ml_parser->negate == FLB_FALSE) { - rule_match = FLB_TRUE; - } - else { - rule_match = FLB_FALSE; - } - } - else { - if (ml_parser->negate == FLB_TRUE) { - rule_match = FLB_TRUE; - } - } - - return rule_match; -} - -static uint64_t time_ms_now() -{ - uint64_t ms; - struct flb_time tm; - - flb_time_get(&tm); - ms = (tm.tm.tv_sec * 1000) + lround(tm.tm.tv_nsec/1.0e6); - return ms; -} - - -int flb_ml_flush_stdout(struct flb_ml_parser *parser, - struct flb_ml_stream *mst, - void *data, char *buf_data, size_t buf_size) -{ - fprintf(stdout, "\n%s----- MULTILINE FLUSH (stream_id=%" PRIu64 ") -----%s\n", - ANSI_GREEN, mst->id, ANSI_RESET); - - /* Print incoming flush buffer */ - flb_pack_print(buf_data, buf_size); - - fprintf(stdout, "%s----------- EOF -----------%s\n", - ANSI_GREEN, ANSI_RESET); - return 0; -} - -int flb_ml_type_lookup(char *str) -{ - int type = -1; - - if (strcasecmp(str, "regex") == 0) { - type = FLB_ML_REGEX; - } - else if (strcasecmp(str, "endswith") == 0) { - type = FLB_ML_ENDSWITH; - } - else if (strcasecmp(str, "equal") == 0 || strcasecmp(str, "eq") == 0) { - type = FLB_ML_EQ; - } - - return type; -} - -void flb_ml_flush_parser_instance(struct flb_ml *ml, - struct flb_ml_parser_ins *parser_i, - uint64_t stream_id, int forced_flush) -{ - struct mk_list *head; - struct mk_list *head_group; - struct flb_ml_stream *mst; - struct flb_ml_stream_group *group; - - mk_list_foreach(head, &parser_i->streams) { - mst = mk_list_entry(head, struct flb_ml_stream, _head); - if (stream_id != 0 && mst->id != stream_id) { - continue; - } - - /* Iterate stream groups */ - mk_list_foreach(head_group, &mst->groups) { - group = mk_list_entry(head_group, struct flb_ml_stream_group, _head); - flb_ml_flush_stream_group(parser_i->ml_parser, mst, group, forced_flush); - } - } -} - -void flb_ml_flush_pending(struct flb_ml *ml, uint64_t now, int forced_flush) -{ - struct mk_list *head; - struct flb_ml_parser_ins *parser_i; - struct flb_ml_group *group; - - /* set the last flush time */ - ml->last_flush = now; - - /* flush only the first group of the context */ - group = mk_list_entry_first(&ml->groups, struct flb_ml_group, _head); - - /* iterate group parser instances */ - mk_list_foreach(head, &group->parsers) { - parser_i = mk_list_entry(head, struct flb_ml_parser_ins, _head); - flb_ml_flush_parser_instance(ml, parser_i, 0, forced_flush); - } -} - -void flb_ml_flush_pending_now(struct flb_ml *ml) -{ - uint64_t now; - - now = time_ms_now(); - flb_ml_flush_pending(ml, now, FLB_TRUE); -} - -static void cb_ml_flush_timer(struct flb_config *ctx, void *data) -{ - uint64_t now; - struct flb_ml *ml = data; - - now = time_ms_now(); - if (ml->last_flush + ml->flush_ms > now) { - return; - } - - /* - * Iterate over all streams and groups and for a flush for expired groups - * which has not flushed in the last N milliseconds. - */ - flb_ml_flush_pending(ml, now, FLB_TRUE); -} - -int flb_ml_register_context(struct flb_ml_stream_group *group, - struct flb_time *tm, msgpack_object *map) -{ - if (tm) { - flb_time_copy(&group->mp_time, tm); - } - - if (map) { - msgpack_pack_object(&group->mp_pck, *map); - } - - return 0; -} - -static inline void breakline_prepare(struct flb_ml_parser_ins *parser_i, - struct flb_ml_stream_group *stream_group) -{ - int len; - - if (parser_i->key_content) { - return; - } - - len = flb_sds_len(stream_group->buf); - if (len <= 0) { - return; - } - - if (stream_group->buf[len - 1] != '\n') { - flb_sds_cat_safe(&stream_group->buf, "\n", 1); - } -} - -/* - * package content into a multiline stream: - * - * full_map: if the original content to process comes in msgpack map, this variable - * reference the map. It's only used in case we will package a first line so we - * store a copy of the other key values in the map for flush time. - */ -static int package_content(struct flb_ml_stream *mst, - msgpack_object *metadata, - msgpack_object *full_map, - void *buf, size_t size, struct flb_time *tm, - msgpack_object *val_content, - msgpack_object *val_pattern, - msgpack_object *val_group) -{ - int len; - int ret; - int rule_match = FLB_FALSE; - int processed = FLB_FALSE; - int type; - size_t offset = 0; - size_t buf_size; - char *buf_data; - msgpack_object *val = val_content; - struct flb_ml_parser *parser; - struct flb_ml_parser_ins *parser_i; - struct flb_ml_stream_group *stream_group; - - parser_i = mst->parser; - parser = parser_i->ml_parser; - - /* Get stream group */ - stream_group = flb_ml_stream_group_get(mst->parser, mst, val_group); - if (!mst->last_stream_group) { - mst->last_stream_group = stream_group; - } - else { - if (mst->last_stream_group != stream_group) { - mst->last_stream_group = stream_group; - } - } - - /* Set the parser type */ - type = parser->type; - - if (val_pattern) { - val = val_pattern; - } - - if (val) { - buf_data = (char *) val->via.str.ptr; - buf_size = val->via.str.size; - } - else { - buf_data = buf; - buf_size = size; - - } - if (type == FLB_ML_REGEX) { - ret = flb_ml_rule_process(parser, mst, - stream_group, full_map, buf, size, tm, - val_content, val_pattern); - if (ret == -1) { - processed = FLB_FALSE; - } - else { - processed = FLB_TRUE; - } - } - else if (type == FLB_ML_ENDSWITH) { - len = flb_sds_len(parser->match_str); - if (buf_data && len <= buf_size) { - /* Validate if content ends with expected string */ - offset = buf_size - len; - ret = memcmp(buf_data + offset, parser->match_str, len); - if (ret == 0) { - rule_match = match_negate(parser, FLB_TRUE); - } - else { - rule_match = match_negate(parser, FLB_FALSE); - } - - if (stream_group->mp_sbuf.size == 0) { - flb_ml_register_context(stream_group, tm, full_map); - } - - /* Prepare concatenation */ - breakline_prepare(parser_i, stream_group); - - /* Concatenate value */ - if (val_content) { - flb_sds_cat_safe(&stream_group->buf, - val_content->via.str.ptr, - val_content->via.str.size); - } - else { - flb_sds_cat_safe(&stream_group->buf, buf_data, buf_size); - } - - /* on ENDSWITH mode, a rule match means flush the content */ - if (rule_match) { - flb_ml_flush_stream_group(parser, mst, stream_group, FLB_FALSE); - } - processed = FLB_TRUE; - } - } - else if (type == FLB_ML_EQ) { - if (buf_size == flb_sds_len(parser->match_str) && - memcmp(buf_data, parser->match_str, buf_size) == 0) { - /* EQ match */ - rule_match = match_negate(parser, FLB_TRUE); - } - else { - rule_match = match_negate(parser, FLB_FALSE); - } - - if (stream_group->mp_sbuf.size == 0) { - flb_ml_register_context(stream_group, tm, full_map); - } - - /* Prepare concatenation */ - breakline_prepare(parser_i, stream_group); - - /* Concatenate value */ - if (val_content) { - flb_sds_cat_safe(&stream_group->buf, - val_content->via.str.ptr, - val_content->via.str.size); - } - else { - flb_sds_cat_safe(&stream_group->buf, buf_data, buf_size); - } - - /* on ENDSWITH mode, a rule match means flush the content */ - if (rule_match) { - flb_ml_flush_stream_group(parser, mst, stream_group, FLB_FALSE); - } - processed = FLB_TRUE; - } - - if (processed && metadata != NULL) { - msgpack_pack_object(&stream_group->mp_md_pck, *metadata); - } - - return processed; -} - -/* - * Retrieve the ID of a specific key name in a map. This function might be - * extended later to use record accessor, since all current cases are solved - * now quering the first level of keys in the map, we avoid record accessor - * to avoid extra memory allocations. - */ -static int get_key_id(msgpack_object *map, flb_sds_t key_name) -{ - int i; - int len; - int found = FLB_FALSE; - msgpack_object key; - msgpack_object val; - - if (!key_name) { - return -1; - } - - len = flb_sds_len(key_name); - for (i = 0; i < map->via.map.size; i++) { - key = map->via.map.ptr[i].key; - val = map->via.map.ptr[i].val; - - if (key.type != MSGPACK_OBJECT_STR || val.type != MSGPACK_OBJECT_STR) { - continue; - } - - if (key.via.str.size != len) { - continue; - } - - if (strncmp(key.via.str.ptr, key_name, len) == 0) { - found = FLB_TRUE; - break; - } - } - - if (found) { - return i; - } - - return -1; -} - -static int process_append(struct flb_ml_parser_ins *parser_i, - struct flb_ml_stream *mst, - int type, - struct flb_time *tm, - msgpack_object *metadata, - msgpack_object *obj, - void *buf, size_t size) -{ - int ret; - int id_content = -1; - int id_pattern = -1; - int id_group = -1; - int unpacked = FLB_FALSE; - size_t off = 0; - msgpack_object *full_map = NULL; - msgpack_object *val_content = NULL; - msgpack_object *val_pattern = NULL; - msgpack_object *val_group = NULL; - msgpack_unpacked result; - - /* Lookup the key */ - if (type == FLB_ML_TYPE_TEXT) { - ret = package_content(mst, NULL, NULL, buf, size, tm, NULL, NULL, NULL); - if (ret == FLB_FALSE) { - return -1; - } - return 0; - } - else if (type == FLB_ML_TYPE_MAP) { - full_map = obj; - /* - * When full_map and buf is not NULL, - * we use 'buf' since buf is already processed from full_map at - * ml_append_try_parser_type_map. - */ - if (!full_map || (buf != NULL && full_map != NULL)) { - off = 0; - msgpack_unpacked_init(&result); - ret = msgpack_unpack_next(&result, buf, size, &off); - if (ret != MSGPACK_UNPACK_SUCCESS) { - return -1; - } - full_map = &result.data; - unpacked = FLB_TRUE; - } - else if (full_map->type != MSGPACK_OBJECT_MAP) { - msgpack_unpacked_destroy(&result); - return -1; - } - } - - /* Lookup for key_content entry */ - id_content = get_key_id(full_map, parser_i->key_content); - if (id_content == -1) { - if (unpacked) { - msgpack_unpacked_destroy(&result); - } - return -1; - } - - val_content = &full_map->via.map.ptr[id_content].val; - if (val_content->type != MSGPACK_OBJECT_STR) { - val_content = NULL; - } - - /* Optional: Lookup for key_pattern entry */ - if (parser_i->key_pattern) { - id_pattern = get_key_id(full_map, parser_i->key_pattern); - if (id_pattern >= 0) { - val_pattern = &full_map->via.map.ptr[id_pattern].val; - if (val_pattern->type != MSGPACK_OBJECT_STR) { - val_pattern = NULL; - } - } - } - - /* Optional: lookup for key_group entry */ - if (parser_i->key_group) { - id_group = get_key_id(full_map, parser_i->key_group); - if (id_group >= 0) { - val_group = &full_map->via.map.ptr[id_group].val; - if (val_group->type != MSGPACK_OBJECT_STR) { - val_group = NULL; - } - } - } - - /* Package the content */ - ret = package_content(mst, metadata, full_map, buf, size, tm, - val_content, val_pattern, val_group); - if (unpacked) { - msgpack_unpacked_destroy(&result); - } - if (ret == FLB_FALSE) { - return -1; - } - return 0; -} - -static int ml_append_try_parser_type_text(struct flb_ml_parser_ins *parser, - uint64_t stream_id, - int *type, - struct flb_time *tm, void *buf, size_t size, - msgpack_object *map, - void **out_buf, size_t *out_size, int *out_release, - struct flb_time *out_time) -{ - int ret; - - if (parser->ml_parser->parser) { - /* Parse incoming content */ - ret = flb_parser_do(parser->ml_parser->parser, (char *) buf, size, - out_buf, out_size, out_time); - if (flb_time_to_nanosec(out_time) == 0L) { - flb_time_copy(out_time, tm); - } - if (ret >= 0) { - *out_release = FLB_TRUE; - *type = FLB_ML_TYPE_MAP; - } - else { - *out_buf = buf; - *out_size = size; - return -1; - } - } - else { - *out_buf = buf; - *out_size = size; - } - return 0; -} - -static int ml_append_try_parser_type_map(struct flb_ml_parser_ins *parser, - uint64_t stream_id, - int *type, - struct flb_time *tm, void *buf, size_t size, - msgpack_object *map, - void **out_buf, size_t *out_size, int *out_release, - struct flb_time *out_time) -{ - int map_size; - int i; - int len; - msgpack_object key; - msgpack_object val; - - if (map == NULL || map->type != MSGPACK_OBJECT_MAP) { - flb_error("%s:invalid map", __FUNCTION__); - return -1; - } - - if (parser->ml_parser->parser) { - /* lookup key_content */ - - len = flb_sds_len(parser->key_content); - map_size = map->via.map.size; - for(i = 0; i < map_size; i++) { - key = map->via.map.ptr[i].key; - val = map->via.map.ptr[i].val; - if (key.type == MSGPACK_OBJECT_STR && - parser->key_content && - key.via.str.size == len && - strncmp(key.via.str.ptr, parser->key_content, len) == 0) { - /* key_content found */ - if (val.type == MSGPACK_OBJECT_STR) { - /* try parse the value of key_content e*/ - return ml_append_try_parser_type_text(parser, stream_id, type, - tm, (void*) val.via.str.ptr, - val.via.str.size, - map, - out_buf, out_size, out_release, - out_time); - } else { - flb_error("%s: not string", __FUNCTION__); - return -1; - } - } - } - } - else { - *out_buf = buf; - *out_size = size; - } - return 0; -} - -static int ml_append_try_parser(struct flb_ml_parser_ins *parser, - uint64_t stream_id, - int type, - struct flb_time *tm, void *buf, size_t size, - msgpack_object *metadata, - msgpack_object *map) -{ - int ret; - int release = FLB_FALSE; - void *out_buf = NULL; - size_t out_size = 0; - struct flb_time out_time; - struct flb_ml_stream *mst; - - flb_time_zero(&out_time); - - switch (type) { - case FLB_ML_TYPE_TEXT: - ret = ml_append_try_parser_type_text(parser, stream_id, &type, - tm, buf, size, map, - &out_buf, &out_size, &release, - &out_time); - if (ret < 0) { - return -1; - } - break; - case FLB_ML_TYPE_MAP: - ret = ml_append_try_parser_type_map(parser, stream_id, &type, - tm, buf, size, map, - &out_buf, &out_size, &release, - &out_time); - if (ret < 0) { - return -1; - } - break; - - default: - flb_error("[multiline] unknown type=%d", type); - return -1; - } - - if (flb_time_to_nanosec(&out_time) == 0L) { - if (tm && flb_time_to_nanosec(tm) != 0L) { - flb_time_copy(&out_time, tm); - } - else { - flb_time_get(&out_time); - } - } - - /* Get the stream */ - mst = flb_ml_stream_get(parser, stream_id); - if (!mst) { - flb_error("[multiline] invalid stream_id %" PRIu64 ", could not " - "append content to multiline context", stream_id); - goto exit; - } - - /* Process the binary record */ - ret = process_append(parser, mst, type, &out_time, metadata, map, out_buf, out_size); - if (ret == -1) { - if (release == FLB_TRUE) { - flb_free(out_buf); - } - return -1; - } - - exit: - if (release == FLB_TRUE) { - flb_free(out_buf); - } - - return 0; -} - -int flb_ml_append_text(struct flb_ml *ml, uint64_t stream_id, - struct flb_time *tm, void *buf, size_t size) -{ - int ret; - int processed = FLB_FALSE; - struct mk_list *head; - struct mk_list *head_group; - struct flb_ml_group *group; - struct flb_ml_stream *mst; - struct flb_ml_parser_ins *lru_parser = NULL; - struct flb_ml_parser_ins *parser_i; - struct flb_time out_time; - struct flb_ml_stream_group *st_group; - int type; - - type = FLB_ML_TYPE_TEXT; - - flb_time_zero(&out_time); - - mk_list_foreach(head, &ml->groups) { - group = mk_list_entry(head, struct flb_ml_group, _head); - - /* Check if the incoming data matches the last recently used parser */ - lru_parser = group->lru_parser; - - if (lru_parser && lru_parser->last_stream_id == stream_id) { - ret = ml_append_try_parser(lru_parser, lru_parser->last_stream_id, type, - tm, buf, size, NULL, NULL); - if (ret == 0) { - processed = FLB_TRUE; - break; - } - else { - flb_ml_flush_parser_instance(ml, - lru_parser, - lru_parser->last_stream_id, - FLB_FALSE); - } - } - else if (lru_parser && lru_parser->last_stream_id > 0) { - /* - * Clear last recently used parser to match new parser. - * Do not flush last_stream_id since it should continue to parsing. - */ - lru_parser = NULL; - } - } - - mk_list_foreach(head_group, &group->parsers) { - parser_i = mk_list_entry(head_group, struct flb_ml_parser_ins, _head); - if (lru_parser && lru_parser == parser_i && - lru_parser->last_stream_id == stream_id) { - continue; - } - - ret = ml_append_try_parser(parser_i, stream_id, type, - tm, buf, size, NULL, NULL); - if (ret == 0) { - group->lru_parser = parser_i; - group->lru_parser->last_stream_id = stream_id; - lru_parser = parser_i; - processed = FLB_TRUE; - break; - } - else { - parser_i = NULL; - } - } - - if (!processed) { - if (lru_parser) { - flb_ml_flush_parser_instance(ml, lru_parser, stream_id, FLB_FALSE); - parser_i = lru_parser; - } - else { - /* get the first parser (just to make use of it buffers) */ - parser_i = mk_list_entry_first(&group->parsers, - struct flb_ml_parser_ins, - _head); - } - - flb_ml_flush_parser_instance(ml, parser_i, stream_id, FLB_FALSE); - mst = flb_ml_stream_get(parser_i, stream_id); - if (!mst) { - flb_error("[multiline] invalid stream_id %" PRIu64 ", could not " - "append content to multiline context", stream_id); - return -1; - } - - /* Get stream group */ - st_group = flb_ml_stream_group_get(mst->parser, mst, NULL); - flb_sds_cat_safe(&st_group->buf, buf, size); - flb_ml_flush_stream_group(parser_i->ml_parser, mst, st_group, FLB_FALSE); - } - - return 0; -} - - - -int flb_ml_append_object(struct flb_ml *ml, - uint64_t stream_id, - struct flb_time *tm, - msgpack_object *metadata, - msgpack_object *obj) -{ - int ret; - int type; - int processed = FLB_FALSE; - struct mk_list *head; - struct mk_list *head_group; - struct flb_ml_group *group; - struct flb_ml_parser_ins *lru_parser = NULL; - struct flb_ml_parser_ins *parser_i; - struct flb_ml_stream *mst; - struct flb_ml_stream_group *st_group; - struct flb_log_event event; - - if (metadata == NULL) { - metadata = ml->log_event_decoder.empty_map; - } - - /* - * As incoming objects, we accept packed events - * and msgpack Maps containing key/value pairs. - */ - if (obj->type == MSGPACK_OBJECT_ARRAY) { - flb_error("[multiline] appending object with invalid type, expected " - "map, received type=%i", obj->type); - return -1; - - - flb_log_event_decoder_reset(&ml->log_event_decoder, NULL, 0); - - ret = flb_event_decoder_decode_object(&ml->log_event_decoder, - &event, - obj); - - if (ret != FLB_EVENT_DECODER_SUCCESS) { - flb_error("[multiline] invalid event object"); - - return -1; - } - - tm = &event.timestamp; - obj = event.body; - metadata = event.metadata; - - type = FLB_ML_TYPE_MAP; - } - else if (obj->type == MSGPACK_OBJECT_MAP) { - type = FLB_ML_TYPE_MAP; - } - else { - flb_error("[multiline] appending object with invalid type, expected " - "array or map, received type=%i", obj->type); - return -1; - } - - mk_list_foreach(head, &ml->groups) { - group = mk_list_entry(head, struct flb_ml_group, _head); - - /* Check if the incoming data matches the last recently used parser */ - lru_parser = group->lru_parser; - - if (lru_parser && lru_parser->last_stream_id == stream_id) { - ret = ml_append_try_parser(lru_parser, lru_parser->last_stream_id, type, - tm, NULL, 0, metadata, obj); - if (ret == 0) { - processed = FLB_TRUE; - break; - } - else { - flb_ml_flush_parser_instance(ml, - lru_parser, - lru_parser->last_stream_id, - FLB_FALSE); - } - } - else if (lru_parser && lru_parser->last_stream_id > 0) { - /* - * Clear last recently used parser to match new parser. - * Do not flush last_stream_id since it should continue to parsing. - */ - lru_parser = NULL; - } - } - - mk_list_foreach(head_group, &group->parsers) { - parser_i = mk_list_entry(head_group, struct flb_ml_parser_ins, _head); - if (lru_parser && parser_i == lru_parser) { - continue; - } - - ret = ml_append_try_parser(parser_i, stream_id, type, - tm, NULL, 0, metadata, obj); - if (ret == 0) { - group->lru_parser = parser_i; - group->lru_parser->last_stream_id = stream_id; - lru_parser = parser_i; - processed = FLB_TRUE; - break; - } - else { - parser_i = NULL; - } - } - - if (!processed) { - if (lru_parser) { - flb_ml_flush_parser_instance(ml, lru_parser, stream_id, FLB_FALSE); - parser_i = lru_parser; - } - else { - /* get the first parser (just to make use of it buffers) */ - parser_i = mk_list_entry_first(&group->parsers, - struct flb_ml_parser_ins, - _head); - } - - flb_ml_flush_parser_instance(ml, parser_i, stream_id, FLB_FALSE); - mst = flb_ml_stream_get(parser_i, stream_id); - if (!mst) { - flb_error("[multiline] invalid stream_id %" PRIu64 ", could not " - "append content to multiline context", stream_id); - - return -1; - } - - /* Get stream group */ - st_group = flb_ml_stream_group_get(mst->parser, mst, NULL); - - ret = flb_log_event_encoder_begin_record(&ml->log_event_encoder); - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_set_timestamp( - &ml->log_event_encoder, tm); - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - if (metadata != ml->log_event_decoder.empty_map) { - ret = flb_log_event_encoder_set_metadata_from_msgpack_object( - &ml->log_event_encoder, metadata); - } - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_set_body_from_msgpack_object( - &ml->log_event_encoder, obj); - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_commit_record(&ml->log_event_encoder); - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - mst->cb_flush(parser_i->ml_parser, - mst, - mst->cb_data, - ml->log_event_encoder.output_buffer, - ml->log_event_encoder.output_length); - } - else { - flb_error("[multiline] log event encoder error : %d", ret); - } - - flb_log_event_encoder_reset(&ml->log_event_encoder); - - /* reset group buffer counters */ - st_group->mp_sbuf.size = 0; - flb_sds_len_set(st_group->buf, 0); - - /* Update last flush time */ - st_group->last_flush = time_ms_now(); - } - - return 0; -} - -int flb_ml_append_event(struct flb_ml *ml, uint64_t stream_id, - struct flb_log_event *event) -{ - return flb_ml_append_object(ml, - stream_id, - &event->timestamp, - event->metadata, - event->body); -} - - -struct flb_ml *flb_ml_create(struct flb_config *ctx, char *name) -{ - int result; - struct flb_ml *ml; - - ml = flb_calloc(1, sizeof(struct flb_ml)); - if (!ml) { - flb_errno(); - return NULL; - } - ml->name = flb_sds_create(name); - if (!ml) { - flb_free(ml); - return NULL; - } - - ml->config = ctx; - ml->last_flush = time_ms_now(); - mk_list_init(&ml->groups); - - result = flb_log_event_decoder_init(&ml->log_event_decoder, - NULL, - 0); - - if (result != FLB_EVENT_DECODER_SUCCESS) { - flb_error("cannot initialize log event decoder"); - - flb_ml_destroy(ml); - - return NULL; - } - - result = flb_log_event_encoder_init(&ml->log_event_encoder, - FLB_LOG_EVENT_FORMAT_DEFAULT); - - if (result != FLB_EVENT_ENCODER_SUCCESS) { - flb_error("cannot initialize log event encoder"); - - flb_ml_destroy(ml); - - return NULL; - } - - return ml; -} - -/* - * Some multiline contexts might define a parser name but not a parser context, - * for missing contexts, just lookup the parser and perform the assignment. - * - * The common use case is when reading config files with [PARSER] and [MULTILINE_PARSER] - * definitions, so we need to delay the parser loading. - */ -int flb_ml_parsers_init(struct flb_config *ctx) -{ - struct mk_list *head; - struct flb_parser *p; - struct flb_ml_parser *ml_parser; - - mk_list_foreach(head, &ctx->multiline_parsers) { - ml_parser = mk_list_entry(head, struct flb_ml_parser, _head); - if (ml_parser->parser_name && !ml_parser->parser) { - p = flb_parser_get(ml_parser->parser_name, ctx); - if (!p) { - flb_error("multiline parser '%s' points to an undefined parser '%s'", - ml_parser->name, ml_parser->parser_name); - return -1; - } - ml_parser->parser = p; - } - } - - return 0; -} - -int flb_ml_auto_flush_init(struct flb_ml *ml) -{ - struct flb_sched *scheduler; - int ret; - - if (ml == NULL) { - return -1; - } - - scheduler = flb_sched_ctx_get(); - - if (scheduler == NULL) { - flb_error("[multiline] scheduler context has not been created"); - return -1; - } - - if (ml->flush_ms < 500) { - flb_error("[multiline] flush timeout '%i' is too low", ml->flush_ms); - return -1; - } - - /* Create flush timer */ - ret = flb_sched_timer_cb_create(scheduler, - FLB_SCHED_TIMER_CB_PERM, - ml->flush_ms, - cb_ml_flush_timer, - ml, NULL); - return ret; -} - -int flb_ml_destroy(struct flb_ml *ml) -{ - struct mk_list *head; - struct mk_list *tmp; - struct flb_ml_group *group; - - if (!ml) { - return 0; - } - - flb_log_event_decoder_destroy(&ml->log_event_decoder); - flb_log_event_encoder_destroy(&ml->log_event_encoder); - - if (ml->name) { - flb_sds_destroy(ml->name); - } - - /* destroy groups */ - mk_list_foreach_safe(head, tmp, &ml->groups) { - group = mk_list_entry(head, struct flb_ml_group, _head); - flb_ml_group_destroy(group); - } - - flb_free(ml); - return 0; -} - -static int flb_msgpack_object_hash_internal(cfl_hash_state_t *state, - msgpack_object *object) -{ - void *dummy_pointer; - int result; - int index; - - if (object == NULL) { - return 0; - } - - dummy_pointer = NULL; - result = 0; - - if (object->type == MSGPACK_OBJECT_NIL) { - cfl_hash_64bits_update(state, - &dummy_pointer, - sizeof(dummy_pointer)); - } - else if (object->type == MSGPACK_OBJECT_BOOLEAN) { - cfl_hash_64bits_update(state, - &object->via.boolean, - sizeof(object->via.boolean)); - } - else if (object->type == MSGPACK_OBJECT_POSITIVE_INTEGER || - object->type == MSGPACK_OBJECT_NEGATIVE_INTEGER) { - cfl_hash_64bits_update(state, - &object->via.u64, - sizeof(object->via.u64)); - } - else if (object->type == MSGPACK_OBJECT_FLOAT32 || - object->type == MSGPACK_OBJECT_FLOAT64 || - object->type == MSGPACK_OBJECT_FLOAT) { - cfl_hash_64bits_update(state, - &object->via.f64, - sizeof(object->via.f64)); - } - else if (object->type == MSGPACK_OBJECT_STR) { - cfl_hash_64bits_update(state, - object->via.str.ptr, - object->via.str.size); - } - else if (object->type == MSGPACK_OBJECT_ARRAY) { - for (index = 0 ; - index < object->via.array.size && - result == 0; - index++) { - result = flb_msgpack_object_hash_internal( - state, - &object->via.array.ptr[index]); - } - } - else if (object->type == MSGPACK_OBJECT_MAP) { - for (index = 0 ; - index < object->via.map.size && - result == 0; - index++) { - result = flb_msgpack_object_hash_internal( - state, - &object->via.map.ptr[index].key); - - if (result == 0) { - result = flb_msgpack_object_hash_internal( - state, - &object->via.map.ptr[index].val); - } - } - } - else if (object->type == MSGPACK_OBJECT_BIN) { - cfl_hash_64bits_update(state, - object->via.bin.ptr, - object->via.bin.size); - } - else if (object->type == MSGPACK_OBJECT_EXT) { - cfl_hash_64bits_update(state, - &object->via.ext.type, - sizeof(object->via.ext.type)); - - cfl_hash_64bits_update(state, - object->via.ext.ptr, - object->via.ext.size); - } - - return result; -} - -static int flb_hash_msgpack_object_list(cfl_hash_64bits_t *hash, - size_t entry_count, - ...) -{ - cfl_hash_state_t hash_state; - va_list arguments; - msgpack_object *object; - int result; - size_t index; - - cfl_hash_64bits_reset(&hash_state); - - va_start(arguments, entry_count); - - result = 0; - - for (index = 0 ; - index < entry_count && - result == 0 ; - index++) { - object = va_arg(arguments, msgpack_object *); - - if (object == NULL) { - break; - } - - result = flb_msgpack_object_hash_internal(&hash_state, object); - } - - va_end(arguments); - - if (result == 0) { - *hash = cfl_hash_64bits_digest(&hash_state); - } - - return result; -} - -struct flb_deduplication_list_entry { - cfl_hash_64bits_t hash; - struct cfl_list _head; -}; - -void flb_deduplication_list_init(struct cfl_list *deduplication_list) -{ - cfl_list_init(deduplication_list); -} - -int flb_deduplication_list_validate(struct cfl_list *deduplication_list, - cfl_hash_64bits_t hash) -{ - struct cfl_list *iterator; - struct flb_deduplication_list_entry *entry; - - cfl_list_foreach(iterator, deduplication_list) { - entry = cfl_list_entry(iterator, - struct flb_deduplication_list_entry, - _head); - - if (entry->hash == hash) { - return FLB_TRUE; - } - } - - return FLB_FALSE; -} - -int flb_deduplication_list_add(struct cfl_list *deduplication_list, - cfl_hash_64bits_t hash) -{ - struct flb_deduplication_list_entry *entry; - - entry = (struct flb_deduplication_list_entry *) - flb_calloc(1, - sizeof(struct flb_deduplication_list_entry)); - - if (entry == NULL) { - return -1; - } - - cfl_list_entry_init(&entry->_head); - entry->hash = hash; - - cfl_list_append(&entry->_head, deduplication_list); - - return 0; -} - -void flb_deduplication_list_purge(struct cfl_list *deduplication_list) -{ - struct cfl_list *iterator; - struct cfl_list *backup; - struct flb_deduplication_list_entry *entry; - - cfl_list_foreach_safe(iterator, backup, deduplication_list) { - entry = cfl_list_entry(iterator, - struct flb_deduplication_list_entry, - _head); - - cfl_list_del(&entry->_head); - - free(entry); - } -} - -int flb_ml_flush_metadata_buffer(struct flb_ml_stream *mst, - struct flb_ml_stream_group *group, - int deduplicate_metadata) -{ - int append_metadata_entry; - cfl_hash_64bits_t metadata_entry_hash; - struct cfl_list deduplication_list; - msgpack_unpacked metadata_map; - size_t offset; - size_t index; - msgpack_object value; - msgpack_object key; - int ret; - - ret = FLB_EVENT_ENCODER_SUCCESS; - - if (deduplicate_metadata) { - flb_deduplication_list_init(&deduplication_list); - } - - msgpack_unpacked_init(&metadata_map); - - offset = 0; - while (ret == FLB_EVENT_ENCODER_SUCCESS && - msgpack_unpack_next(&metadata_map, - group->mp_md_sbuf.data, - group->mp_md_sbuf.size, - &offset) == MSGPACK_UNPACK_SUCCESS) { - - for (index = 0; - index < metadata_map.data.via.map.size && - ret == FLB_EVENT_ENCODER_SUCCESS; - index++) { - key = metadata_map.data.via.map.ptr[index].key; - value = metadata_map.data.via.map.ptr[index].val; - - append_metadata_entry = FLB_TRUE; - - if (deduplicate_metadata) { - ret = flb_hash_msgpack_object_list(&metadata_entry_hash, - 2, - &key, - &value); - if (ret != 0) { - ret = FLB_EVENT_ENCODER_ERROR_INVALID_ARGUMENT; - } - else { - ret = flb_deduplication_list_validate( - &deduplication_list, - metadata_entry_hash); - - if (ret) { - append_metadata_entry = FLB_FALSE; - - ret = FLB_EVENT_ENCODER_SUCCESS; - } - else { - ret = flb_deduplication_list_add( - &deduplication_list, - metadata_entry_hash); - - if (ret == 0) { - ret = FLB_EVENT_ENCODER_SUCCESS; - } - else { - ret = FLB_EVENT_ENCODER_ERROR_ALLOCATION_ERROR; - } - } - } - } - - if (append_metadata_entry) { - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_append_metadata_values( - &mst->ml->log_event_encoder, - FLB_LOG_EVENT_MSGPACK_OBJECT_VALUE(&key), - FLB_LOG_EVENT_MSGPACK_OBJECT_VALUE(&value)); - } - } - } - } - - msgpack_unpacked_destroy(&metadata_map); - - if (deduplicate_metadata) { - flb_deduplication_list_purge(&deduplication_list); - } - - return ret; -} - -int flb_ml_flush_stream_group(struct flb_ml_parser *ml_parser, - struct flb_ml_stream *mst, - struct flb_ml_stream_group *group, - int forced_flush) -{ - int i; - int ret; - int size; - int len; - size_t off = 0; - msgpack_object map; - msgpack_object k; - msgpack_object v; - msgpack_sbuffer mp_sbuf; - msgpack_packer mp_pck; - msgpack_unpacked result; - struct flb_ml_parser_ins *parser_i = mst->parser; - struct flb_time *group_time; - struct flb_time now; - - breakline_prepare(parser_i, group); - len = flb_sds_len(group->buf); - - /* init msgpack buffer */ - msgpack_sbuffer_init(&mp_sbuf); - msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); - - /* if the group don't have a time set, use current time */ - if (flb_time_to_nanosec(&group->mp_time) == 0L) { - flb_time_get(&now); - group_time = &now; - } else { - group_time = &group->mp_time; - } - - /* compose final record if we have a first line context */ - if (group->mp_sbuf.size > 0) { - msgpack_unpacked_init(&result); - ret = msgpack_unpack_next(&result, - group->mp_sbuf.data, group->mp_sbuf.size, - &off); - if (ret != MSGPACK_UNPACK_SUCCESS) { - flb_error("[multiline] could not unpack first line state buffer"); - msgpack_unpacked_destroy(&result); - group->mp_sbuf.size = 0; - return -1; - } - map = result.data; - - if (map.type != MSGPACK_OBJECT_MAP) { - flb_error("[multiline] expected MAP type in first line state buffer"); - msgpack_unpacked_destroy(&result); - group->mp_sbuf.size = 0; - return -1; - } - - /* Take the first line keys and repack */ - len = flb_sds_len(parser_i->key_content); - size = map.via.map.size; - msgpack_pack_map(&mp_pck, size); - - for (i = 0; i < size; i++) { - k = map.via.map.ptr[i].key; - v = map.via.map.ptr[i].val; - - /* - * Check if the current key is the key that will contain the - * concatenated multiline buffer - */ - if (k.type == MSGPACK_OBJECT_STR && - parser_i->key_content && - k.via.str.size == len && - strncmp(k.via.str.ptr, parser_i->key_content, len) == 0) { - - /* key */ - msgpack_pack_object(&mp_pck, k); - - /* value */ - len = flb_sds_len(group->buf); - msgpack_pack_str(&mp_pck, len); - msgpack_pack_str_body(&mp_pck, group->buf, len); - } - else { - /* key / val */ - msgpack_pack_object(&mp_pck, k); - msgpack_pack_object(&mp_pck, v); - } - } - msgpack_unpacked_destroy(&result); - group->mp_sbuf.size = 0; - } - else if (len > 0) { - /* Pack raw content as Fluent Bit record */ - msgpack_pack_map(&mp_pck, 1); - - /* key */ - if (parser_i->key_content) { - len = flb_sds_len(parser_i->key_content); - msgpack_pack_str(&mp_pck, len); - msgpack_pack_str_body(&mp_pck, parser_i->key_content, len); - } - else { - msgpack_pack_str(&mp_pck, 3); - msgpack_pack_str_body(&mp_pck, "log", 3); - } - - /* val */ - len = flb_sds_len(group->buf); - msgpack_pack_str(&mp_pck, len); - msgpack_pack_str_body(&mp_pck, group->buf, len); - } - - if (mp_sbuf.size > 0) { - /* - * a 'forced_flush' means to alert the caller that the data 'must be flushed to it destination'. This flag is - * only enabled when the flush process has been triggered by the multiline timer, e.g: - * - * - the message is complete or incomplete and its time to dispatch it. - */ - if (forced_flush) { - mst->forced_flush = FLB_TRUE; - } - - /* encode and invoke the user callback */ - - ret = flb_log_event_encoder_begin_record( - &mst->ml->log_event_encoder); - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_set_timestamp( - &mst->ml->log_event_encoder, - group_time); - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_ml_flush_metadata_buffer(mst, - group, - FLB_TRUE); - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_set_body_from_raw_msgpack( - &mst->ml->log_event_encoder, - mp_sbuf.data, - mp_sbuf.size); - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_commit_record( - &mst->ml->log_event_encoder); - } - - if (ret != FLB_EVENT_ENCODER_SUCCESS) { - flb_error("[multiline] error packing event"); - - return -1; - } - - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - mst->cb_flush(ml_parser, - mst, - mst->cb_data, - mst->ml->log_event_encoder.output_buffer, - mst->ml->log_event_encoder.output_length); - } - else { - flb_error("[multiline] log event encoder error : %d", ret); - } - - flb_log_event_encoder_reset(&mst->ml->log_event_encoder); - - if (forced_flush) { - mst->forced_flush = FLB_FALSE; - } - } - - msgpack_sbuffer_destroy(&mp_sbuf); - flb_sds_len_set(group->buf, 0); - - /* Update last flush time */ - group->last_flush = time_ms_now(); - - return 0; -} - -/* - * Initialize multiline global environment. - * - * note: function must be invoked before any flb_ml_create() call. - */ -int flb_ml_init(struct flb_config *config) -{ - int ret; - - ret = flb_ml_parser_builtin_create(config); - if (ret == -1) { - return -1; - } - - return 0; -} - -int flb_ml_exit(struct flb_config *config) -{ - flb_ml_parser_destroy_all(&config->multiline_parsers); - return 0; -} diff --git a/fluent-bit/src/multiline/flb_ml_group.c b/fluent-bit/src/multiline/flb_ml_group.c deleted file mode 100644 index 895a71056..000000000 --- a/fluent-bit/src/multiline/flb_ml_group.c +++ /dev/null @@ -1,86 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_mem.h> -#include <fluent-bit/flb_log.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -struct flb_ml_group *flb_ml_group_create(struct flb_ml *ml) -{ - struct flb_ml_group *group; - - group = flb_calloc(1, sizeof(struct flb_ml_group)); - if (!group) { - flb_errno(); - return NULL; - } - group->id = mk_list_size(&ml->groups); - group->ml = ml; - group->lru_parser = NULL; - mk_list_init(&group->parsers); - - mk_list_add(&group->_head, &ml->groups); - - return group; -} - -/* - * Link a parser instance into the active group, if no group exists, a default - * one is created. - */ -int flb_ml_group_add_parser(struct flb_ml *ctx, struct flb_ml_parser_ins *p) -{ - struct flb_ml_group *group = NULL; - - if (mk_list_size(&ctx->groups) == 0) { - group = flb_ml_group_create(ctx); - if (!group) { - return -1; - } - } - else { - /* retrieve the latest active group */ - group = mk_list_entry_last(&ctx->groups, struct flb_ml_group, _head); - } - - if (!group) { - return -1; - } - - mk_list_add(&p->_head, &group->parsers); - return 0; -} - -void flb_ml_group_destroy(struct flb_ml_group *group) -{ - struct mk_list *head; - struct mk_list *tmp; - struct flb_ml_parser_ins *parser_i; - - /* destroy parser instances */ - mk_list_foreach_safe(head, tmp, &group->parsers) { - parser_i = mk_list_entry(head, struct flb_ml_parser_ins, _head); - flb_ml_parser_instance_destroy(parser_i); - } - - mk_list_del(&group->_head); - flb_free(group); -} diff --git a/fluent-bit/src/multiline/flb_ml_mode.c b/fluent-bit/src/multiline/flb_ml_mode.c deleted file mode 100644 index 964672d82..000000000 --- a/fluent-bit/src/multiline/flb_ml_mode.c +++ /dev/null @@ -1,111 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_log.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_mode.h> - -struct flb_ml *flb_ml_mode_create(struct flb_config *config, char *mode, int flush_ms, - char *key) -{ - if (strcmp(mode, "docker") == 0) { - return flb_ml_mode_docker(config, flush_ms); - } - else if (strcmp(mode, "cri") == 0) { - return flb_ml_mode_cri(config, flush_ms); - } - else if (strcmp(mode, "python") == 0) { - return flb_ml_mode_python(config, flush_ms, key); - } - else if (strcmp(mode, "java") == 0) { - return flb_ml_mode_java(config, flush_ms, key); - } - else if (strcmp(mode, "go") == 0) { - return flb_ml_mode_go(config, flush_ms, key); - } - - flb_error("[multiline] built-in mode '%s' not found", mode); - return NULL; -} - - -struct flb_ml_mode *flb_ml_parser_create(struct flb_config *ctx, - char *name, - int type, char *match_str, int negate, - int flush_ms, - char *key_content, - char *key_group, - char *key_pattern, - struct flb_parser *parser_ctx, - char *parser_name) -{ - struct flb_ml_mode *ml; - - ml = flb_calloc(1, sizeof(struct flb_ml)); - if (!ml) { - flb_errno(); - return NULL; - } - ml->name = flb_sds_create(name); - ml->type = type; - - if (match_str) { - ml->match_str = flb_sds_create(match_str); - if (!ml->match_str) { - flb_free(ml); - return NULL; - } - } - - ml->parser = parser_ctx; - if (parser_name) { - ml->parser_name = flb_sds_create(parser_name); - } - - ml->negate = negate; - mk_list_init(&ml->streams); - mk_list_init(&ml->regex_rules); - mk_list_add(&ml->_head, &ctx->multilines); - - if (key_content) { - ml->key_content = flb_sds_create(key_content); - if (!ml->key_content) { - flb_ml_destroy(ml); - return NULL; - } - } - - if (key_group) { - ml->key_group = flb_sds_create(key_group); - if (!ml->key_group) { - flb_ml_destroy(ml); - return NULL; - } - } - - if (key_pattern) { - ml->key_pattern = flb_sds_create(key_pattern); - if (!ml->key_pattern) { - flb_ml_destroy(ml); - return NULL; - } - } - return ml; -} diff --git a/fluent-bit/src/multiline/flb_ml_parser.c b/fluent-bit/src/multiline/flb_ml_parser.c deleted file mode 100644 index 7aa33789d..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser.c +++ /dev/null @@ -1,347 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_log.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <fluent-bit/multiline/flb_ml_mode.h> -#include <fluent-bit/multiline/flb_ml_group.h> - -int flb_ml_parser_init(struct flb_ml_parser *ml_parser) -{ - int ret; - - ret = flb_ml_rule_init(ml_parser); - if (ret == -1) { - return -1; - } - - return 0; -} - -/* Create built-in multiline parsers */ -int flb_ml_parser_builtin_create(struct flb_config *config) -{ - struct flb_ml_parser *mlp; - - /* Docker */ - mlp = flb_ml_parser_docker(config); - if (!mlp) { - flb_error("[multiline] could not init 'docker' built-in parser"); - return -1; - } - - /* CRI */ - mlp = flb_ml_parser_cri(config); - if (!mlp) { - flb_error("[multiline] could not init 'cri' built-in parser"); - return -1; - } - - /* Java */ - mlp = flb_ml_parser_java(config, NULL); - if (!mlp) { - flb_error("[multiline] could not init 'java' built-in parser"); - return -1; - } - - /* Go */ - mlp = flb_ml_parser_go(config, NULL); - if (!mlp) { - flb_error("[multiline] could not init 'go' built-in parser"); - return -1; - } - - /* Ruby */ - mlp = flb_ml_parser_ruby(config, NULL); - if (!mlp) { - flb_error("[multiline] could not init 'ruby' built-in parser"); - return -1; - } - - /* Python */ - mlp = flb_ml_parser_python(config, NULL); - if (!mlp) { - flb_error("[multiline] could not init 'python' built-in parser"); - return -1; - } - - return 0; -} - -struct flb_ml_parser *flb_ml_parser_create(struct flb_config *ctx, - char *name, - int type, char *match_str, int negate, - int flush_ms, - char *key_content, - char *key_group, - char *key_pattern, - struct flb_parser *parser_ctx, - char *parser_name) -{ - struct flb_ml_parser *ml_parser; - - ml_parser = flb_calloc(1, sizeof(struct flb_ml_parser)); - if (!ml_parser) { - flb_errno(); - return NULL; - } - ml_parser->name = flb_sds_create(name); - ml_parser->type = type; - - if (match_str) { - ml_parser->match_str = flb_sds_create(match_str); - if (!ml_parser->match_str) { - if (ml_parser->name) { - flb_sds_destroy(ml_parser->name); - } - flb_free(ml_parser); - return NULL; - } - } - - ml_parser->parser = parser_ctx; - - if (parser_name) { - ml_parser->parser_name = flb_sds_create(parser_name); - } - ml_parser->negate = negate; - ml_parser->flush_ms = flush_ms; - mk_list_init(&ml_parser->regex_rules); - mk_list_add(&ml_parser->_head, &ctx->multiline_parsers); - - if (key_content) { - ml_parser->key_content = flb_sds_create(key_content); - if (!ml_parser->key_content) { - flb_ml_parser_destroy(ml_parser); - return NULL; - } - } - - if (key_group) { - ml_parser->key_group = flb_sds_create(key_group); - if (!ml_parser->key_group) { - flb_ml_parser_destroy(ml_parser); - return NULL; - } - } - - if (key_pattern) { - ml_parser->key_pattern = flb_sds_create(key_pattern); - if (!ml_parser->key_pattern) { - flb_ml_parser_destroy(ml_parser); - return NULL; - } - } - - return ml_parser; -} - -struct flb_ml_parser *flb_ml_parser_get(struct flb_config *ctx, char *name) -{ - struct mk_list *head; - struct flb_ml_parser *ml_parser; - - mk_list_foreach(head, &ctx->multiline_parsers) { - ml_parser = mk_list_entry(head, struct flb_ml_parser, _head); - if (strcasecmp(ml_parser->name, name) == 0) { - return ml_parser; - } - } - - return NULL; -} - -int flb_ml_parser_instance_has_data(struct flb_ml_parser_ins *ins) -{ - struct mk_list *head; - struct mk_list *head_group; - struct flb_ml_stream *st; - struct flb_ml_stream_group *st_group; - - mk_list_foreach(head, &ins->streams) { - st = mk_list_entry(head, struct flb_ml_stream, _head); - mk_list_foreach(head_group, &st->groups) { - st_group = mk_list_entry(head_group, struct flb_ml_stream_group, _head); - if (st_group->mp_sbuf.size > 0) { - return FLB_TRUE; - } - } - } - - return FLB_FALSE; -} - -struct flb_ml_parser_ins *flb_ml_parser_instance_create(struct flb_ml *ml, - char *name) -{ - int ret; - struct flb_ml_parser_ins *ins; - struct flb_ml_parser *parser; - - parser = flb_ml_parser_get(ml->config, name); - if (!parser) { - flb_error("[multiline] parser '%s' not registered", name); - return NULL; - } - - ins = flb_calloc(1, sizeof(struct flb_ml_parser_ins)); - if (!ins) { - flb_errno(); - return NULL; - } - ins->last_stream_id = 0; - ins->ml_parser = parser; - mk_list_init(&ins->streams); - - /* Copy parent configuration */ - if (parser->key_content) { - ins->key_content = flb_sds_create(parser->key_content); - } - if (parser->key_pattern) { - ins->key_pattern = flb_sds_create(parser->key_pattern); - } - if (parser->key_group) { - ins->key_group = flb_sds_create(parser->key_group); - } - - /* Append this multiline parser instance to the active multiline group */ - ret = flb_ml_group_add_parser(ml, ins); - if (ret != 0) { - flb_error("[multiline] could not register parser '%s' on " - "multiline '%s 'group", name, ml->name); - flb_free(ins); - return NULL; - } - - /* - * Update flush_interval for pending records on multiline context. We always - * use the greater value found. - */ - if (parser->flush_ms > ml->flush_ms) { - ml->flush_ms = parser->flush_ms; - } - - return ins; -} - -/* Override a fixed parser property for the instance only*/ -int flb_ml_parser_instance_set(struct flb_ml_parser_ins *p, char *prop, char *val) -{ - if (strcasecmp(prop, "key_content") == 0) { - if (p->key_content) { - flb_sds_destroy(p->key_content); - } - p->key_content = flb_sds_create(val); - } - else if (strcasecmp(prop, "key_pattern") == 0) { - if (p->key_pattern) { - flb_sds_destroy(p->key_pattern); - } - p->key_pattern = flb_sds_create(val); - } - else if (strcasecmp(prop, "key_group") == 0) { - if (p->key_group) { - flb_sds_destroy(p->key_group); - } - p->key_group = flb_sds_create(val); - } - else { - return -1; - } - - return 0; -} - -int flb_ml_parser_destroy(struct flb_ml_parser *ml_parser) -{ - if (!ml_parser) { - return 0; - } - - if (ml_parser->name) { - flb_sds_destroy(ml_parser->name); - } - - if (ml_parser->parser_name) { - flb_sds_destroy(ml_parser->parser_name); - } - - if (ml_parser->match_str) { - flb_sds_destroy(ml_parser->match_str); - } - if (ml_parser->key_content) { - flb_sds_destroy(ml_parser->key_content); - } - if (ml_parser->key_group) { - flb_sds_destroy(ml_parser->key_group); - } - if (ml_parser->key_pattern) { - flb_sds_destroy(ml_parser->key_pattern); - } - - /* Regex rules */ - flb_ml_rule_destroy_all(ml_parser); - - /* Unlink from struct flb_config->multiline_parsers */ - mk_list_del(&ml_parser->_head); - - flb_free(ml_parser); - return 0; -} - -int flb_ml_parser_instance_destroy(struct flb_ml_parser_ins *ins) -{ - struct mk_list *tmp; - struct mk_list *head; - struct flb_ml_stream *stream; - - /* Destroy streams */ - mk_list_foreach_safe(head, tmp, &ins->streams) { - stream = mk_list_entry(head, struct flb_ml_stream, _head); - flb_ml_stream_destroy(stream); - } - - if (ins->key_content) { - flb_sds_destroy(ins->key_content); - } - if (ins->key_pattern) { - flb_sds_destroy(ins->key_pattern); - } - if (ins->key_group) { - flb_sds_destroy(ins->key_group); - } - - flb_free(ins); - - return 0; -} - -void flb_ml_parser_destroy_all(struct mk_list *list) -{ - struct mk_list *tmp; - struct mk_list *head; - struct flb_ml_parser *parser; - - mk_list_foreach_safe(head, tmp, list) { - parser = mk_list_entry(head, struct flb_ml_parser, _head); - flb_ml_parser_destroy(parser); - } -} diff --git a/fluent-bit/src/multiline/flb_ml_parser_cri.c b/fluent-bit/src/multiline/flb_ml_parser_cri.c deleted file mode 100644 index 669fa39a2..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser_cri.c +++ /dev/null @@ -1,81 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -#define FLB_ML_CRI_REGEX \ - "^(?<time>.+?) (?<stream>stdout|stderr) (?<_p>F|P) (?<log>.*)$" -#define FLB_ML_CRI_TIME \ - "%Y-%m-%dT%H:%M:%S.%L%z" - -/* Creates a parser for Docker */ -static struct flb_parser *cri_parser_create(struct flb_config *config) -{ - struct flb_parser *p; - - p = flb_parser_create("_ml_cri", /* parser name */ - "regex", /* backend type */ - FLB_ML_CRI_REGEX, /* regex */ - FLB_FALSE, /* skip_empty */ - FLB_ML_CRI_TIME, /* time format */ - "time", /* time key */ - NULL, /* time offset */ - FLB_TRUE, /* time keep */ - FLB_FALSE, /* time strict */ - FLB_FALSE, /* no bare keys */ - NULL, /* parser types */ - 0, /* types len */ - NULL, /* decoders */ - config); /* Fluent Bit context */ - return p; -} - -/* Our first multiline mode: 'docker' */ -struct flb_ml_parser *flb_ml_parser_cri(struct flb_config *config) -{ - struct flb_parser *parser; - struct flb_ml_parser *mlp; - - /* Create a Docker parser */ - parser = cri_parser_create(config); - if (!parser) { - return NULL; - } - - mlp = flb_ml_parser_create(config, - "cri", /* name */ - FLB_ML_EQ, /* type */ - "F", /* match_str */ - FLB_FALSE, /* negate */ - FLB_ML_FLUSH_TIMEOUT, /* flush_ms */ - "log", /* key_content */ - "stream", /* key_group */ - "_p", /* key_pattern */ - parser, /* parser ctx */ - NULL); /* parser name */ - - if (!mlp) { - flb_error("[multiline] could not create 'cri mode'"); - return NULL; - } - - return mlp; -} diff --git a/fluent-bit/src/multiline/flb_ml_parser_docker.c b/fluent-bit/src/multiline/flb_ml_parser_docker.c deleted file mode 100644 index 5b622d32c..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser_docker.c +++ /dev/null @@ -1,110 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -/* Creates a parser for Docker */ -static struct flb_parser *docker_parser_create(struct flb_config *config) -{ - struct flb_parser *p; - - p = flb_parser_create("_ml_json_docker", /* parser name */ - "json", /* backend type */ - NULL, /* regex */ - FLB_TRUE, /* skip_empty */ - "%Y-%m-%dT%H:%M:%S.%L", /* time format */ - "time", /* time key */ - NULL, /* time offset */ - FLB_TRUE, /* time keep */ - FLB_FALSE, /* time strict */ - FLB_FALSE, /* no bare keys */ - NULL, /* parser types */ - 0, /* types len */ - NULL, /* decoders */ - config); /* Fluent Bit context */ - return p; -} - -/* Our first multiline mode: 'docker' */ -struct flb_ml_parser *flb_ml_parser_docker(struct flb_config *config) -{ - struct flb_parser *parser; - struct flb_ml_parser *mlp; - - /* Create a Docker parser */ - parser = docker_parser_create(config); - if (!parser) { - return NULL; - } - - /* - * Let's explain this multiline mode, then you (the reader) might want - * to submit a PR with new built-in modes :) - * - * Containerized apps under Docker writes logs to stdout/stderr. These streams - * (stdout/stderr) are handled by Docker, in most of cases the content is - * stored in a .json file in your file system. A message like "hey!" gets into - * a JSON map like this: - * - * {"log": "hey!\n", "stream": "stdout", "time": "2021-02-01T01:40:03.53412Z"} - * - * By Docker log spec, any 'log' key that "ends with a \n" it's a complete - * log record, but Docker also limits the log record size to 16KB, so a long - * message that does not fit into 16KB can be split in multiple JSON lines, - * the following example use short words to describe the context: - * - * - original message: 'one, two, three\n' - * - * Docker log interpretation: - * - * - {"log": "one, ", "stream": "stdout", "time": "2021-02-01T01:40:03.53413Z"} - * - {"log": "two, ", "stream": "stdout", "time": "2021-02-01T01:40:03.53414Z"} - * - {"log": "three\n", "stream": "stdout", "time": "2021-02-01T01:40:03.53415Z"} - * - * So every 'log' key that does not ends with '\n', it's a partial log record - * and for logging purposes it needs to be concatenated with further messages - * until a final '\n' is found. - * - * We setup the Multiline mode as follows: - * - * - Use the type 'FLB_ML_ENDSWITH' to specify that we expect the 'log' - * key must ends with a '\n' for complete messages, otherwise it means is - * a continuation message. In case a message is not complete just wait until - * 500 milliseconds (0.5 second) and flush the buffer. - */ - mlp = flb_ml_parser_create(config, /* Fluent Bit context */ - "docker", /* name */ - FLB_ML_ENDSWITH, /* type */ - "\n", /* match_str */ - FLB_FALSE, /* negate */ - FLB_ML_FLUSH_TIMEOUT, /* flush_ms */ - "log", /* key_content */ - "stream", /* key_group */ - NULL, /* key_pattern */ - parser, /* parser ctx */ - NULL); /* parser name */ - if (!mlp) { - flb_error("[multiline] could not create 'docker mode'"); - return NULL; - } - - return mlp; -} diff --git a/fluent-bit/src/multiline/flb_ml_parser_go.c b/fluent-bit/src/multiline/flb_ml_parser_go.c deleted file mode 100644 index f1cd5407f..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser_go.c +++ /dev/null @@ -1,140 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -#define rule flb_ml_rule_create - -static void rule_error(struct flb_ml_parser *mlp) -{ - int id; - - id = mk_list_size(&mlp->regex_rules); - flb_error("[multiline: go] rule #%i could not be created", id); - flb_ml_parser_destroy(mlp); -} - -/* Go mode */ -struct flb_ml_parser *flb_ml_parser_go(struct flb_config *config, char *key) -{ - int ret; - struct flb_ml_parser *mlp; - - mlp = flb_ml_parser_create(config, /* Fluent Bit context */ - "go", /* name */ - FLB_ML_REGEX, /* type */ - NULL, /* match_str */ - FLB_FALSE, /* negate */ - FLB_ML_FLUSH_TIMEOUT, /* flush_ms */ - key, /* key_content */ - NULL, /* key_group */ - NULL, /* key_pattern */ - NULL, /* parser ctx */ - NULL); /* parser name */ - - if (!mlp) { - flb_error("[multiline] could not create 'go mode'"); - return NULL; - } - - ret = rule(mlp, - "start_state", - "/\\bpanic: /", - "go_after_panic", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "start_state", - "/http: panic serving/", - "go_goroutine", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "go_after_panic", - "/^$/", - "go_goroutine", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "go_after_panic, go_after_signal, go_frame_1", - "/^$/", - "go_goroutine", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "go_after_panic", - "/^\\[signal /", - "go_after_signal", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "go_goroutine", - "/^goroutine \\d+ \\[[^\\]]+\\]:$/", - "go_frame_1", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "go_frame_1", - "/^(?:[^\\s.:]+\\.)*[^\\s.():]+\\(|^created by /", - "go_frame_2", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "go_frame_2", - "/^\\s/", - "go_frame_1", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - /* Map the rules (mandatory for regex rules) */ - ret = flb_ml_parser_init(mlp); - if (ret != 0) { - flb_error("[multiline: go] error on mapping rules"); - flb_ml_parser_destroy(mlp); - return NULL; - } - - return mlp; -} diff --git a/fluent-bit/src/multiline/flb_ml_parser_java.c b/fluent-bit/src/multiline/flb_ml_parser_java.c deleted file mode 100644 index 4df5a00f7..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser_java.c +++ /dev/null @@ -1,143 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -#define rule flb_ml_rule_create - -static void rule_error(struct flb_ml_parser *ml_parser) -{ - int id; - - id = mk_list_size(&ml_parser->regex_rules); - flb_error("[multiline: java] rule #%i could not be created", id); - flb_ml_parser_destroy(ml_parser); -} - -/* Java mode */ -struct flb_ml_parser *flb_ml_parser_java(struct flb_config *config, char *key) -{ - int ret; - struct flb_ml_parser *mlp; - - mlp = flb_ml_parser_create(config, /* Fluent Bit context */ - "java", /* name */ - FLB_ML_REGEX, /* type */ - NULL, /* match_str */ - FLB_FALSE, /* negate */ - FLB_ML_FLUSH_TIMEOUT, /* flush_ms */ - key, /* key_content */ - NULL, /* key_group */ - NULL, /* key_pattern */ - NULL, /* parser ctx */ - NULL); /* parser name */ - - if (!mlp) { - flb_error("[multiline] could not create 'java mode'"); - return NULL; - } - - ret = rule(mlp, - "start_state, java_start_exception", - "/(.)(?:Exception|Error|Throwable|V8 errors stack trace)[:\\r\\n]/", - "java_after_exception", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception", - "/^[\\t ]*nested exception is:[\\t ]*/", - "java_start_exception", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception", - "/^[\\r\\n]*$/", - "java_after_exception", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception, java", - "/^[\\t ]+(?:eval )?at /", - "java", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception, java", - /* C# nested exception */ - "/^[\\t ]+--- End of inner exception stack trace ---$/", - "java", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception, java", - /* C# exception from async code */ - "/^--- End of stack trace from previous (?x:" - ")location where exception was thrown ---$/", - "java", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception, java", - "/^[\\t ]*(?:Caused by|Suppressed):/", - "java_after_exception", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "java_after_exception, java", - "/^[\\t ]*... \\d+ (?:more|common frames omitted)/", - "java", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - /* Map the rules (mandatory for regex rules) */ - ret = flb_ml_parser_init(mlp); - if (ret != 0) { - flb_error("[multiline: java] error on mapping rules"); - flb_ml_parser_destroy(mlp); - return NULL; - } - - return mlp; -} diff --git a/fluent-bit/src/multiline/flb_ml_parser_python.c b/fluent-bit/src/multiline/flb_ml_parser_python.c deleted file mode 100644 index a92088397..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser_python.c +++ /dev/null @@ -1,98 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -#define rule flb_ml_rule_create - -static void rule_error(struct flb_ml_parser *mlp) -{ - int id; - - id = mk_list_size(&mlp->regex_rules); - flb_error("[multiline: python] rule #%i could not be created", id); - flb_ml_parser_destroy(mlp); -} - -/* Python */ -struct flb_ml_parser *flb_ml_parser_python(struct flb_config *config, char *key) -{ - int ret; - struct flb_ml_parser *mlp; - - mlp = flb_ml_parser_create(config, /* Fluent Bit context */ - "python", /* name */ - FLB_ML_REGEX, /* type */ - NULL, /* match_str */ - FLB_FALSE, /* negate */ - FLB_ML_FLUSH_TIMEOUT, /* flush_ms */ - key, /* key_content */ - NULL, /* key_group */ - NULL, /* key_pattern */ - NULL, /* parser ctx */ - NULL); /* parser name */ - - if (!mlp) { - flb_error("[multiline] could not create 'python mode'"); - return NULL; - } - - /* rule(:start_state, /^Traceback \(most recent call last\):$/, :python) */ - ret = rule(mlp, - "start_state", "/^Traceback \\(most recent call last\\):$/", - "python", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - /* rule(:python, /^[\t ]+File /, :python_code) */ - ret = rule(mlp, "python", "/^[\\t ]+File /", "python_code", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - /* rule(:python_code, /[^\t ]/, :python) */ - ret = rule(mlp, "python_code", "/[^\\t ]/", "python", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - /* rule(:python, /^(?:[^\s.():]+\.)*[^\s.():]+:/, :start_state) */ - ret = rule(mlp, "python", "/^(?:[^\\s.():]+\\.)*[^\\s.():]+:/", "start_state", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - /* Map the rules (mandatory for regex rules) */ - ret = flb_ml_parser_init(mlp); - if (ret != 0) { - flb_error("[multiline: python] error on mapping rules"); - flb_ml_parser_destroy(mlp); - return NULL; - } - - return mlp; -} diff --git a/fluent-bit/src/multiline/flb_ml_parser_ruby.c b/fluent-bit/src/multiline/flb_ml_parser_ruby.c deleted file mode 100644 index 780f829d0..000000000 --- a/fluent-bit/src/multiline/flb_ml_parser_ruby.c +++ /dev/null @@ -1,87 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <fluent-bit/multiline/flb_ml_parser.h> - -#define rule flb_ml_rule_create - -static void rule_error(struct flb_ml_parser *mlp) -{ - int id; - - id = mk_list_size(&mlp->regex_rules); - flb_error("[multiline: ruby] rule #%i could not be created", id); - flb_ml_parser_destroy(mlp); -} - -/* Ruby mode */ -struct flb_ml_parser *flb_ml_parser_ruby(struct flb_config *config, char *key) -{ - int ret; - struct flb_ml_parser *mlp; - - mlp = flb_ml_parser_create(config, /* Fluent Bit context */ - "ruby", /* name */ - FLB_ML_REGEX, /* type */ - NULL, /* match_str */ - FLB_FALSE, /* negate */ - FLB_ML_FLUSH_TIMEOUT, /* flush_ms */ - key, /* key_content */ - NULL, /* key_group */ - NULL, /* key_pattern */ - NULL, /* parser ctx */ - NULL); /* parser name */ - - if (!mlp) { - flb_error("[multiline] could not create 'ruby mode'"); - return NULL; - } - - ret = rule(mlp, - "start_state, ruby_start_exception", - "/^.+:\\d+:in\\s+.*/", - "ruby_after_exception", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - ret = rule(mlp, - "ruby_after_exception, ruby", - "/^\\s+from\\s+.*:\\d+:in\\s+.*/", - "ruby", NULL); - if (ret != 0) { - rule_error(mlp); - return NULL; - } - - - /* Map the rules (mandatory for regex rules) */ - ret = flb_ml_parser_init(mlp); - if (ret != 0) { - flb_error("[multiline: ruby] error on mapping rules"); - flb_ml_parser_destroy(mlp); - return NULL; - } - - return mlp; -} diff --git a/fluent-bit/src/multiline/flb_ml_rule.c b/fluent-bit/src/multiline/flb_ml_rule.c deleted file mode 100644 index 26520dfed..000000000 --- a/fluent-bit/src/multiline/flb_ml_rule.c +++ /dev/null @@ -1,421 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_regex.h> -#include <fluent-bit/flb_slist.h> - -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> - -struct to_state { - struct flb_ml_rule *rule; - struct mk_list _head; -}; - -struct flb_slist_entry *get_start_state(struct mk_list *list) -{ - struct mk_list *head; - struct flb_slist_entry *e; - - mk_list_foreach(head, list) { - e = mk_list_entry(head, struct flb_slist_entry, _head); - if (strcmp(e->str, "start_state") == 0) { - return e; - } - } - - return NULL; -} - -int flb_ml_rule_create(struct flb_ml_parser *ml_parser, - flb_sds_t from_states, - char *regex_pattern, - flb_sds_t to_state, - char *end_pattern) -{ - int ret; - int first_rule = FLB_FALSE; - struct flb_ml_rule *rule; - - rule = flb_calloc(1, sizeof(struct flb_ml_rule)); - if (!rule) { - flb_errno(); - return -1; - } - flb_slist_create(&rule->from_states); - mk_list_init(&rule->to_state_map); - - if (mk_list_size(&ml_parser->regex_rules) == 0) { - first_rule = FLB_TRUE; - } - mk_list_add(&rule->_head, &ml_parser->regex_rules); - - /* from_states */ - ret = flb_slist_split_string(&rule->from_states, from_states, ',', -1); - if (ret <= 0) { - flb_error("[multiline] rule is empty or has invalid 'from_states' tokens"); - flb_ml_rule_destroy(rule); - return -1; - } - - /* Check if the rule contains a 'start_state' */ - if (get_start_state(&rule->from_states)) { - rule->start_state = FLB_TRUE; - } - else if (first_rule) { - flb_error("[multiline] rule don't contain a 'start_state'"); - flb_ml_rule_destroy(rule); - return -1; - } - - /* regex content pattern */ - rule->regex = flb_regex_create(regex_pattern); - if (!rule->regex) { - flb_ml_rule_destroy(rule); - return -1; - } - - /* to_state */ - if (to_state) { - rule->to_state = flb_sds_create(to_state); - if (!rule->to_state) { - flb_ml_rule_destroy(rule); - return -1; - } - } - - /* regex end pattern */ - if (end_pattern) { - rule->regex_end = flb_regex_create(end_pattern); - if (!rule->regex_end) { - flb_ml_rule_destroy(rule); - return -1; - } - } - - return 0; -} - -void flb_ml_rule_destroy(struct flb_ml_rule *rule) -{ - struct mk_list *tmp; - struct mk_list *head; - struct to_state *st; - - flb_slist_destroy(&rule->from_states); - - if (rule->regex) { - flb_regex_destroy(rule->regex); - } - - - if (rule->to_state) { - flb_sds_destroy(rule->to_state); - } - - mk_list_foreach_safe(head, tmp, &rule->to_state_map) { - st = mk_list_entry(head, struct to_state, _head); - mk_list_del(&st->_head); - flb_free(st); - } - - if (rule->regex_end) { - flb_regex_destroy(rule->regex_end); - } - - mk_list_del(&rule->_head); - flb_free(rule); -} - -void flb_ml_rule_destroy_all(struct flb_ml_parser *ml_parser) -{ - struct mk_list *tmp; - struct mk_list *head; - struct flb_ml_rule *rule; - - mk_list_foreach_safe(head, tmp, &ml_parser->regex_rules) { - rule = mk_list_entry(head, struct flb_ml_rule, _head); - flb_ml_rule_destroy(rule); - } -} - -static inline int to_states_exists(struct flb_ml_parser *ml_parser, - flb_sds_t state) -{ - struct mk_list *head; - struct mk_list *r_head; - struct flb_ml_rule *rule; - struct flb_slist_entry *e; - - mk_list_foreach(head, &ml_parser->regex_rules) { - rule = mk_list_entry(head, struct flb_ml_rule, _head); - - mk_list_foreach(r_head, &rule->from_states) { - e = mk_list_entry(r_head, struct flb_slist_entry, _head); - if (strcmp(e->str, state) == 0) { - return FLB_TRUE; - } - } - } - - return FLB_FALSE; -} - -static inline int to_states_matches_rule(struct flb_ml_rule *rule, - flb_sds_t state) -{ - struct mk_list *head; - struct flb_slist_entry *e; - - mk_list_foreach(head, &rule->from_states) { - e = mk_list_entry(head, struct flb_slist_entry, _head); - if (strcmp(e->str, state) == 0) { - return FLB_TRUE; - } - } - - return FLB_FALSE; -} - -static int set_to_state_map(struct flb_ml_parser *ml_parser, - struct flb_ml_rule *rule) -{ - int ret; - struct to_state *s; - struct mk_list *head; - struct flb_ml_rule *r; - - if (!rule->to_state) { - /* no to_state */ - return 0; - } - - /* Iterate all rules that matches the to_state */ - mk_list_foreach(head, &ml_parser->regex_rules) { - r = mk_list_entry(head, struct flb_ml_rule, _head); - - /* Check if rule->to_state, matches an existing (registered) from_state */ - ret = to_states_exists(ml_parser, rule->to_state); - if (!ret) { - flb_error("[multiline parser: %s] to_state='%s' is not registered", - ml_parser->name, rule->to_state); - return -1; - } - - /* - * A rule can have many 'from_states', check if the current 'rule->to_state' - * matches any 'r->from_states' - */ - ret = to_states_matches_rule(r, rule->to_state); - if (!ret) { - continue; - } - - /* We have a match. Create a 'to_state' entry into the 'to_state_map' list */ - s = flb_malloc(sizeof(struct to_state)); - if (!s) { - flb_errno(); - return -1; - } - s->rule = r; - mk_list_add(&s->_head, &rule->to_state_map); - } - - return 0; -} - -static int try_flushing_buffer(struct flb_ml_parser *ml_parser, - struct flb_ml_stream *mst, - struct flb_ml_stream_group *group) -{ - int next_start = FLB_FALSE; - struct mk_list *head; - struct to_state *st; - struct flb_ml_rule *rule; - - rule = group->rule_to_state; - if (!rule) { - if (flb_sds_len(group->buf) > 0) { - flb_ml_flush_stream_group(ml_parser, mst, group, FLB_FALSE); - group->first_line = FLB_TRUE; - } - return 0; - } - - /* Check if any 'to_state_map' referenced rules is a possible start */ - mk_list_foreach(head, &rule->to_state_map) { - st = mk_list_entry(head, struct to_state, _head); - if (st->rule->start_state) { - next_start = FLB_TRUE; - break; - } - } - - if (next_start && flb_sds_len(group->buf) > 0) { - flb_ml_flush_stream_group(ml_parser, mst, group, FLB_FALSE); - group->first_line = FLB_TRUE; - } - - return 0; -} - -/* Initialize all rules */ -int flb_ml_rule_init(struct flb_ml_parser *ml_parser) -{ - int ret; - struct mk_list *head; - struct flb_ml_rule *rule; - - /* FIXME: sort rules by start_state first (let's trust in the caller) */ - - /* For each rule, compose it to_state_map list */ - mk_list_foreach(head, &ml_parser->regex_rules) { - rule = mk_list_entry(head, struct flb_ml_rule, _head); - /* Populate 'rule->to_state_map' list */ - ret = set_to_state_map(ml_parser, rule); - if (ret == -1) { - return -1; - } - } - - return 0; -} - -/* Search any 'start_state' matching the incoming 'buf_data' */ -static struct flb_ml_rule *try_start_state(struct flb_ml_parser *ml_parser, - char *buf_data, size_t buf_size) -{ - int ret = -1; - struct mk_list *head; - struct flb_ml_rule *rule = NULL; - - mk_list_foreach(head, &ml_parser->regex_rules) { - rule = mk_list_entry(head, struct flb_ml_rule, _head); - - /* Is this rule matching a start_state ? */ - if (!rule->start_state) { - rule = NULL; - continue; - } - - /* Matched a start_state. Check if we have a regex match */ - ret = flb_regex_match(rule->regex, (unsigned char *) buf_data, buf_size); - if (ret) { - return rule; - } - } - - return NULL; -} - -int flb_ml_rule_process(struct flb_ml_parser *ml_parser, - struct flb_ml_stream *mst, - struct flb_ml_stream_group *group, - msgpack_object *full_map, - void *buf, size_t size, struct flb_time *tm, - msgpack_object *val_content, - msgpack_object *val_pattern) -{ - int ret; - int len; - char *buf_data = NULL; - size_t buf_size = 0; - struct mk_list *head; - struct to_state *st = NULL; - struct flb_ml_rule *rule = NULL; - struct flb_ml_rule *tmp_rule = NULL; - - if (val_content) { - buf_data = (char *) val_content->via.str.ptr; - buf_size = val_content->via.str.size; - } - else { - buf_data = buf; - buf_size = size; - } - - if (group->rule_to_state) { - /* are we in a continuation ? */ - tmp_rule = group->rule_to_state; - - /* Lookup all possible next rules by state reference */ - rule = NULL; - mk_list_foreach(head, &tmp_rule->to_state_map) { - st = mk_list_entry(head, struct to_state, _head); - - /* skip start states */ - if (st->rule->start_state) { - continue; - } - - /* Try regex match */ - ret = flb_regex_match(st->rule->regex, - (unsigned char *) buf_data, buf_size); - if (ret) { - /* Regex matched */ - len = flb_sds_len(group->buf); - if (len >= 1 && group->buf[len - 1] != '\n') { - flb_sds_cat_safe(&group->buf, "\n", 1); - } - - if (buf_size == 0) { - flb_sds_cat_safe(&group->buf, "\n", 1); - } - else { - flb_sds_cat_safe(&group->buf, buf_data, buf_size); - } - rule = st->rule; - break; - } - rule = NULL; - } - - } - - if (!rule) { - /* Check if we are in a 'start_state' */ - rule = try_start_state(ml_parser, buf_data, buf_size); - if (rule) { - /* if the group buffer has any previous data just flush it */ - if (flb_sds_len(group->buf) > 0) { - flb_ml_flush_stream_group(ml_parser, mst, group, FLB_FALSE); - } - - /* set the rule state */ - group->rule_to_state = rule; - - /* concatenate the data */ - flb_sds_cat_safe(&group->buf, buf_data, buf_size); - - /* Copy full map content in stream buffer */ - flb_ml_register_context(group, tm, full_map); - - return 0; - } - } - - if (rule) { - group->rule_to_state = rule; - try_flushing_buffer(ml_parser, mst, group); - return 0; - } - - return -1; -} diff --git a/fluent-bit/src/multiline/flb_ml_stream.c b/fluent-bit/src/multiline/flb_ml_stream.c deleted file mode 100644 index c92ba3220..000000000 --- a/fluent-bit/src/multiline/flb_ml_stream.c +++ /dev/null @@ -1,338 +0,0 @@ -/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ - -/* Fluent Bit - * ========== - * Copyright (C) 2015-2022 The Fluent Bit Authors - * - * 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. - */ - -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_mem.h> -#include <fluent-bit/flb_log.h> -#include <fluent-bit/flb_pack.h> -#include <fluent-bit/multiline/flb_ml.h> -#include <fluent-bit/multiline/flb_ml_rule.h> -#include <cfl/cfl.h> - -static int ml_flush_stdout(struct flb_ml_parser *parser, - struct flb_ml_stream *mst, - void *data, char *buf_data, size_t buf_size) -{ - fprintf(stdout, "\n%s----- MULTILINE FLUSH -----%s\n", - ANSI_GREEN, ANSI_RESET); - - /* Print incoming flush buffer */ - flb_pack_print(buf_data, buf_size); - - fprintf(stdout, "%s----------- EOF -----------%s\n", - ANSI_GREEN, ANSI_RESET); - return 0; -} - -static struct flb_ml_stream_group *stream_group_create(struct flb_ml_stream *mst, - char *name, int len) -{ - struct flb_ml_stream_group *group; - - if (!name) { - name = "_default"; - } - - group = flb_calloc(1, sizeof(struct flb_ml_stream_group)); - if (!group) { - flb_errno(); - return NULL; - } - group->name = flb_sds_create_len(name, len); - if (!group->name) { - flb_free(group); - return NULL; - } - - /* status */ - group->first_line = FLB_TRUE; - - /* multiline buffer */ - group->buf = flb_sds_create_size(FLB_ML_BUF_SIZE); - if (!group->buf) { - flb_error("cannot allocate multiline stream buffer in group %s", name); - flb_sds_destroy(group->name); - flb_free(group); - return NULL; - } - - /* msgpack buffer */ - msgpack_sbuffer_init(&group->mp_md_sbuf); - msgpack_packer_init(&group->mp_md_pck, &group->mp_md_sbuf, msgpack_sbuffer_write); - - msgpack_sbuffer_init(&group->mp_sbuf); - msgpack_packer_init(&group->mp_pck, &group->mp_sbuf, msgpack_sbuffer_write); - - mk_list_add(&group->_head, &mst->groups); - - return group; -} - -struct flb_ml_stream_group *flb_ml_stream_group_get(struct flb_ml_parser_ins *parser_i, - struct flb_ml_stream *mst, - msgpack_object *group_name) -{ - int len; - char *name; - struct flb_ml_parser *mlp; - struct mk_list *head; - struct flb_ml_stream_group *group = NULL; - - mlp = parser_i->ml_parser; - - /* If key_group was not defined, we already have a default group */ - if (!mlp->key_group || !group_name) { - group = mk_list_entry_first(&mst->groups, - struct flb_ml_stream_group, - _head); - return group; - } - - /* Lookup for a candidate group */ - len = group_name->via.str.size; - name = (char *)group_name->via.str.ptr; - - mk_list_foreach(head, &mst->groups) { - group = mk_list_entry(head, struct flb_ml_stream_group, _head); - if (flb_sds_cmp(group->name, name, len) == 0) { - return group; - } - else { - group = NULL; - continue; - } - } - - /* No group has been found, create a new one */ - if (mk_list_size(&mst->groups) >= FLB_ML_MAX_GROUPS) { - flb_error("[multiline] stream %s exceeded number of allowed groups (%i)", - mst->name, FLB_ML_MAX_GROUPS); - return NULL; - } - - group = stream_group_create(mst, name, len); - return group; -} - -static void stream_group_destroy(struct flb_ml_stream_group *group) -{ - if (group->name) { - flb_sds_destroy(group->name); - } - if (group->buf) { - flb_sds_destroy(group->buf); - } - - msgpack_sbuffer_destroy(&group->mp_md_sbuf); - msgpack_sbuffer_destroy(&group->mp_sbuf); - - mk_list_del(&group->_head); - flb_free(group); -} - -static void stream_group_destroy_all(struct flb_ml_stream *mst) -{ - struct mk_list *tmp; - struct mk_list *head; - struct flb_ml_stream_group *group; - - mk_list_foreach_safe(head, tmp, &mst->groups) { - group = mk_list_entry(head, struct flb_ml_stream_group, _head); - stream_group_destroy(group); - } -} - -static int stream_group_init(struct flb_ml_stream *stream) -{ - struct flb_ml_stream_group *group = NULL; - - mk_list_init(&stream->groups); - - /* create a default group */ - group = stream_group_create(stream, NULL, 0); - if (!group) { - flb_error("[multiline] error initializing default group for " - "stream '%s'", stream->name); - return -1; - } - - return 0; -} - -static struct flb_ml_stream *stream_create(struct flb_ml *ml, - uint64_t id, - struct flb_ml_parser_ins *parser, - int (*cb_flush) (struct flb_ml_parser *, - struct flb_ml_stream *, - void *cb_data, - char *buf_data, - size_t buf_size), - void *cb_data) -{ - int ret; - struct flb_ml_stream *stream; - - stream = flb_calloc(1, sizeof(struct flb_ml_stream)); - if (!stream) { - flb_errno(); - return NULL; - } - stream->ml = ml; - stream->id = id; - stream->parser = parser; - - /* Flush Callback and opaque data type */ - if (cb_flush) { - stream->cb_flush = cb_flush; - } - else { - stream->cb_flush = ml_flush_stdout; - } - stream->cb_data = cb_data; - - ret = stream_group_init(stream); - if (ret != 0) { - flb_free(stream); - return NULL; - } - - mk_list_add(&stream->_head, &parser->streams); - return stream; -} - -int flb_ml_stream_create(struct flb_ml *ml, - char *name, - int name_len, - int (*cb_flush) (struct flb_ml_parser *, - struct flb_ml_stream *, - void *cb_data, - char *buf_data, - size_t buf_size), - void *cb_data, - uint64_t *stream_id) -{ - uint64_t id; - struct mk_list *head; - struct mk_list *head_group; - struct flb_ml_stream *mst; - struct flb_ml_group *group; - struct flb_ml_parser_ins *parser; - - if (!name) { - return -1; - } - - if (name_len <= 0) { - name_len = strlen(name); - } - - /* Set the stream id by creating a hash using the name */ - id = cfl_hash_64bits(name, name_len); - - /* For every group and parser, create a stream for this stream_id/hash */ - mk_list_foreach(head, &ml->groups) { - group = mk_list_entry(head, struct flb_ml_group, _head); - mk_list_foreach(head_group, &group->parsers) { - parser = mk_list_entry(head_group, struct flb_ml_parser_ins, _head); - - /* Check if the stream already exists on the parser */ - if (flb_ml_stream_get(parser, id) != NULL) { - continue; - } - - /* Create the stream */ - mst = stream_create(ml, id, parser, cb_flush, cb_data); - if (!mst) { - flb_error("[multiline] could not create stream_id=%" PRIu64 - "for stream '%s' on parser '%s'", - *stream_id, name, parser->ml_parser->name); - return -1; - } - } - } - - *stream_id = id; - return 0; -} - -struct flb_ml_stream *flb_ml_stream_get(struct flb_ml_parser_ins *parser, - uint64_t stream_id) -{ - struct mk_list *head; - struct flb_ml_stream *mst = NULL; - - mk_list_foreach(head, &parser->streams) { - mst = mk_list_entry(head, struct flb_ml_stream, _head); - if (mst->id == stream_id) { - return mst; - } - } - - return NULL; -} - -void flb_ml_stream_id_destroy_all(struct flb_ml *ml, uint64_t stream_id) -{ - struct mk_list *tmp; - struct mk_list *head; - struct mk_list *head_group; - struct mk_list *head_stream; - struct flb_ml_group *group; - struct flb_ml_stream *mst; - struct flb_ml_parser_ins *parser_i; - - /* groups */ - mk_list_foreach(head, &ml->groups) { - group = mk_list_entry(head, struct flb_ml_group, _head); - - /* parser instances */ - mk_list_foreach(head_group, &group->parsers) { - parser_i = mk_list_entry(head_group, struct flb_ml_parser_ins, _head); - - /* streams */ - mk_list_foreach_safe(head_stream, tmp, &parser_i->streams) { - mst = mk_list_entry(head_stream, struct flb_ml_stream, _head); - if (mst->id != stream_id) { - continue; - } - - /* flush any pending data */ - flb_ml_flush_parser_instance(ml, parser_i, stream_id, FLB_TRUE); - - /* destroy internal groups of the stream */ - flb_ml_stream_destroy(mst); - } - } - } -} - -int flb_ml_stream_destroy(struct flb_ml_stream *mst) -{ - mk_list_del(&mst->_head); - if (mst->name) { - flb_sds_destroy(mst->name); - } - - /* destroy groups */ - stream_group_destroy_all(mst); - - flb_free(mst); - - return 0; -} |