diff options
Diffstat (limited to 'fluent-bit/src/multiline/flb_ml.c')
| -rw-r--r-- | fluent-bit/src/multiline/flb_ml.c | 1562 |
1 files changed, 0 insertions, 1562 deletions
diff --git a/fluent-bit/src/multiline/flb_ml.c b/fluent-bit/src/multiline/flb_ml.c deleted file mode 100644 index 28e123ce..00000000 --- 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; -} |