diff --git a/CMakeLists.txt b/CMakeLists.txt index 2cc355b7b6c..167ab0c620c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -216,6 +216,7 @@ if(FLB_ALL) # Global set(FLB_DEBUG 1) set(FLB_TLS 1) + set(FLB_UTF8_ENCODER 1) # Input plugins set(FLB_IN_CPU 1) @@ -402,7 +403,8 @@ endif() # tutf8e if(FLB_UTF8_ENCODER) - add_subdirectory(${FLB_PATH_LIB_TUTF8E} EXCLUDE_FROM_ALL) + add_subdirectory(${FLB_PATH_LIB_TUTF8E}) + FLB_DEFINITION(FLB_HAVE_UTF8_ENCODER) endif() # xxHash diff --git a/include/fluent-bit/flb_encoding.h b/include/fluent-bit/flb_encoding.h new file mode 100644 index 00000000000..07d13a6605a --- /dev/null +++ b/include/fluent-bit/flb_encoding.h @@ -0,0 +1,44 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2019 The Fluent Bit Authors + * Copyright (C) 2015-2018 Treasure Data Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FLB_ENCODING_H +#define FLB_ENCODING_H + +#include + + +#define FLB_ENCODING_SUCCESS 0 +#define FLB_ENCODING_FAILURE -1 + +struct flb_encoding { + TUTF8encoder encoder; + const char *invalid; +}; + + +struct flb_encoding *flb_encoding_open(const char *encoding, const char *replacement); + +int flb_encoding_decode(struct flb_encoding *ec, + char *str, size_t slen, + char **result, size_t *result_len); + +void flb_encoding_close(struct flb_encoding *ic); + +#endif diff --git a/plugins/in_syslog/syslog.h b/plugins/in_syslog/syslog.h index 728675678fa..9a810953ca9 100644 --- a/plugins/in_syslog/syslog.h +++ b/plugins/in_syslog/syslog.h @@ -24,6 +24,10 @@ #include #include +#ifdef FLB_HAVE_UTF8_ENCODER +#include +#endif + /* Syslog modes */ #define FLB_SYSLOG_UNIX_TCP 1 #define FLB_SYSLOG_UNIX_UDP 2 @@ -63,6 +67,11 @@ struct flb_syslog { struct mk_list connections; struct mk_event_loop *evl; struct flb_input_instance *ins; + +#ifdef FLB_HAVE_UTF8_ENCODER + struct flb_encoding *encoding; +#endif + }; #endif diff --git a/plugins/in_syslog/syslog_conf.c b/plugins/in_syslog/syslog_conf.c index 46391675bab..5fd4b6edb62 100644 --- a/plugins/in_syslog/syslog_conf.c +++ b/plugins/in_syslog/syslog_conf.c @@ -34,6 +34,9 @@ struct flb_syslog *syslog_conf_create(struct flb_input_instance *ins, struct flb_config *config) { const char *tmp; +#ifdef FLB_HAVE_UTF8_ENCODER + const char *tmp2; +#endif char port[16]; struct flb_syslog *ctx; @@ -134,6 +137,20 @@ struct flb_syslog *syslog_conf_create(struct flb_input_instance *ins, return NULL; } +#ifdef FLB_HAVE_UTF8_ENCODER + /* utf8 encoder */ + tmp = flb_input_get_property("encoding", ins); + if (tmp) { + tmp2 = flb_input_get_property("encoding_replacement", ins); + ctx->encoding = flb_encoding_open(tmp,tmp2); + if (!ctx->encoding) { + flb_error("[in_syslog] illegal encoding: %s", tmp); + syslog_conf_destroy(ctx); + return NULL; + } + } +#endif + return ctx; } @@ -143,6 +160,13 @@ int syslog_conf_destroy(struct flb_syslog *ctx) flb_free(ctx->buffer_data); ctx->buffer_data = NULL; } + +#ifdef FLB_HAVE_UTF8_ENCODER + if(ctx->encoding) { + flb_encoding_close(ctx->encoding); + } +#endif + syslog_server_destroy(ctx); flb_free(ctx); diff --git a/plugins/in_syslog/syslog_prot.c b/plugins/in_syslog/syslog_prot.c index 0fee0f46345..61dd94e2951 100644 --- a/plugins/in_syslog/syslog_prot.c +++ b/plugins/in_syslog/syslog_prot.c @@ -52,17 +52,64 @@ static inline int pack_line(struct flb_syslog *ctx, return 0; } + +int syslog_prot_process_msg(struct flb_syslog *ctx, char *p, size_t len) +{ + int ret; + void *out_buf; + size_t out_size; + struct flb_time out_time; +#ifdef FLB_HAVE_UTF8_ENCODER + char *decoded = NULL; + size_t decoded_size; +#endif + +#ifdef FLB_HAVE_UTF8_ENCODER + if (ctx->encoding) { + ret = flb_encoding_decode(ctx->encoding, p, len, &decoded, &decoded_size); + if (ret != FLB_ENCODING_SUCCESS) { + flb_plg_error(ctx->ins, "decoding failed '%.*s'", p, len); + goto finish; + } + p = decoded; + len = decoded_size; + } +#endif + + /* Process the string */ + ret = flb_parser_do(ctx->parser, p, len, + &out_buf, &out_size, &out_time); + if (ret < 0) { + flb_plg_warn(ctx->ins, "error parsing log message with parser '%s'", + ctx->parser->name); + flb_plg_debug(ctx->ins, "unparsed log message: %.*s", len, p); + goto finish; + } + + if (flb_time_to_double(&out_time) == 0.0) { + flb_time_get(&out_time); + } + + pack_line(ctx, &out_time, out_buf, out_size); + ret = 0; + + finish: + +#ifdef FLB_HAVE_UTF8_ENCODER + if (decoded) { + flb_free(decoded); + } +#endif + return ret; + +} + int syslog_prot_process(struct syslog_conn *conn) { int len; - int ret; char *p; char *eof; char *end; - void *out_buf; - size_t out_size; - struct flb_time out_time; - struct flb_syslog *ctx = conn->ctx; eof = conn->buf_data; end = conn->buf_data + conn->buf_len; @@ -96,21 +143,7 @@ int syslog_prot_process(struct syslog_conn *conn) continue; } - /* Process the string */ - ret = flb_parser_do(ctx->parser, p, len, - &out_buf, &out_size, &out_time); - if (ret >= 0) { - if (flb_time_to_double(&out_time) == 0.0) { - flb_time_get(&out_time); - } - pack_line(ctx, &out_time, out_buf, out_size); - flb_free(out_buf); - } - else { - flb_plg_warn(ctx->ins, "error parsing log message with parser '%s'", - ctx->parser->name); - flb_plg_debug(ctx->ins, "unparsed log message: %.*s", len, p); - } + syslog_prot_process_msg(conn->ctx, p, len); conn->buf_parsed += len + 1; end = conn->buf_data + conn->buf_len; @@ -129,27 +162,5 @@ int syslog_prot_process(struct syslog_conn *conn) int syslog_prot_process_udp(char *buf, size_t size, struct flb_syslog *ctx) { - int ret; - void *out_buf; - size_t out_size; - struct flb_time out_time = {0}; - - ret = flb_parser_do(ctx->parser, buf, size, - &out_buf, &out_size, &out_time); - if (ret >= 0) { - if (flb_time_to_double(&out_time) == 0) { - flb_time_get(&out_time); - } - pack_line(ctx, &out_time, out_buf, out_size); - flb_free(out_buf); - } - else { - flb_plg_warn(ctx->ins, "error parsing log message with parser '%s'", - ctx->parser->name); - flb_plg_debug(ctx->ins, "unparsed log message: %.*s", - (int) size, buf); - return -1; - } - - return 0; + return syslog_prot_process_msg(ctx, buf, size); } diff --git a/plugins/in_tail/tail.c b/plugins/in_tail/tail.c index 65588a0327b..8d227537741 100644 --- a/plugins/in_tail/tail.c +++ b/plugins/in_tail/tail.c @@ -628,6 +628,19 @@ static struct flb_config_map config_map[] = { }, #endif +#ifdef FLB_HAVE_UTF8_ENCODER + { + FLB_CONFIG_MAP_STR, "encoding", NULL, + 0, FLB_FALSE, 0, + "specify the input encoding for converting to UTF-8", + }, + { + FLB_CONFIG_MAP_STR, "encoding_replacement", NULL, + 0, FLB_FALSE, 0, + "Replacement in case of decoding error (default: unicode replacement char)", + }, +#endif + /* Multiline Options */ #ifdef FLB_HAVE_PARSER { diff --git a/plugins/in_tail/tail_config.c b/plugins/in_tail/tail_config.c index 2ac3f7468dc..b0c4e2a2679 100644 --- a/plugins/in_tail/tail_config.c +++ b/plugins/in_tail/tail_config.c @@ -82,8 +82,10 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins, int i; long nsec; const char *tmp; +#ifdef FLB_HAVE_UTF8_ENCODER + const char *tmp2; +#endif struct flb_tail_config *ctx; - ctx = flb_calloc(1, sizeof(struct flb_tail_config)); if (!ctx) { flb_errno(); @@ -96,6 +98,9 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins, #ifdef FLB_HAVE_SQLDB ctx->db_sync = 1; /* sqlite sync 'normal' */ #endif +#ifdef FLB_HAVE_UTF8_ENCODER + ctx->encoding = NULL; +#endif /* Load the config map */ ret = flb_input_config_map_set(ins, (void *) ctx); @@ -190,6 +195,19 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins, } #endif +#ifdef FLB_HAVE_UTF8_ENCODER + tmp = flb_input_get_property("encoding", ins); + if (tmp) { + tmp2 = flb_input_get_property("encoding_relacement", ins); + ctx->encoding = flb_encoding_open(tmp,tmp2); + if (!ctx->encoding) { + flb_plg_error(ctx->ins,"illegal encoding: %s", tmp); + flb_tail_config_destroy(ctx); + return NULL; + } + } +#endif + /* Config: Docker mode */ if(ctx->docker_mode == FLB_TRUE) { ret = flb_tail_dmode_create(ctx, ins, config); @@ -445,6 +463,12 @@ int flb_tail_config_destroy(struct flb_tail_config *config) } #endif +#ifdef FLB_HAVE_UTF8_ENCODER + if(config->encoding) { + flb_encoding_close(config->encoding); + } +#endif + flb_free(config); return 0; } diff --git a/plugins/in_tail/tail_config.h b/plugins/in_tail/tail_config.h index 8e8e13a92d0..6febc0e1e9e 100644 --- a/plugins/in_tail/tail_config.h +++ b/plugins/in_tail/tail_config.h @@ -33,6 +33,9 @@ #ifdef FLB_HAVE_PARSER #include #endif +#ifdef FLB_HAVE_UTF8_ENCODER +#include +#endif /* Metrics */ @@ -104,6 +107,10 @@ struct flb_tail_config { sqlite3_stmt *stmt_offset; #endif +#ifdef FLB_HAVE_UTF8_ENCODER + struct flb_encoding *encoding; +#endif + /* Parser / Format */ struct flb_parser *parser; diff --git a/plugins/in_tail/tail_file.c b/plugins/in_tail/tail_file.c index 134282d7180..fc71484a958 100644 --- a/plugins/in_tail/tail_file.c +++ b/plugins/in_tail/tail_file.c @@ -307,6 +307,10 @@ static int process_content(struct flb_tail_file *file, size_t *bytes) msgpack_sbuffer *out_sbuf; msgpack_packer *out_pck; struct flb_tail_config *ctx = file->config; +#ifdef FLB_HAVE_UTF8_ENCODER + char *decoded; + size_t decoded_len; +#endif /* Create a temporary msgpack buffer */ msgpack_sbuffer_init(&mp_sbuf); @@ -371,6 +375,18 @@ static int process_content(struct flb_tail_file *file, size_t *bytes) line_len = len - crlf; repl_line = NULL; +#ifdef FLB_HAVE_UTF8_ENCODER + decoded = NULL; + if(ctx->encoding) { + ret = flb_encoding_decode(ctx->encoding, line, line_len, &decoded, &decoded_len); + if (ret != FLB_ENCODING_SUCCESS) { + flb_plg_error(ctx->ins, "encoding failed '%.*s'", line_len, line); + goto go_next; + } + line = decoded; + line_len = decoded_len; + } +#endif if (ctx->ml_ctx) { ret = flb_ml_append(ctx->ml_ctx, file->ml_stream_id, FLB_ML_TYPE_TEXT, @@ -421,7 +437,7 @@ static int process_content(struct flb_tail_file *file, size_t *bytes) /* Parser failed, pack raw text */ flb_time_get(&out_time); flb_tail_file_pack_line(out_sbuf, out_pck, &out_time, - data, len, file, processed_bytes); + line, len, file, processed_bytes); } } else if (ctx->multiline == FLB_TRUE) { @@ -465,6 +481,12 @@ static int process_content(struct flb_tail_file *file, size_t *bytes) processed_bytes += len + 1; file->parsed = 0; lines++; +#ifdef FLB_HAVE_UTF8_ENCODER + if(decoded) { + flb_free(decoded); + decoded = NULL; + } +#endif } file->parsed = file->buf_len; @@ -519,6 +541,7 @@ static int process_content(struct flb_tail_file *file, size_t *bytes) } msgpack_sbuffer_destroy(out_sbuf); + return lines; } diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 18a5336ecf1..d589e19ac30 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -335,6 +335,10 @@ set(FLB_DEPS ${FLB_DEPS} tutf8e ) +set(src + ${src} + "flb_encoding.c" + ) endif() # Record Accessor diff --git a/src/flb_encoding.c b/src/flb_encoding.c new file mode 100644 index 00000000000..7bdb624b1fa --- /dev/null +++ b/src/flb_encoding.c @@ -0,0 +1,161 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2019 The Fluent Bit Authors + * Copyright (C) 2015-2018 Treasure Data Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include + + +#include + +/* + * + * flb_encoding_open(encoding,replacement): + * + * encoding: + * iso-8859-1,... + * windows-1251 windows-1252, .. + * + * replacement: + * \R use replacement character 0xFFD (default) + * \I ignore/skip bad chars + * \E fail if bad chars + * ... use that as replacement char + * + */ + + +static unsigned char replacement_utf8[4] = { 0xEF, 0xBF, 0xBD , 0 }; + + +static const char *parse_replacement(const char *replacement) { + + if (!replacement) { + return (const char*)replacement_utf8; + } + if (!strcmp(replacement,"\\R")) { + return (const char*)replacement_utf8; + } + if (!strcmp(replacement,"\\I")) { + return ""; + } + if (!strcmp(replacement,"\\?")) { + return "?"; + } + if (!strcmp(replacement,"\\E")) { + return NULL; + } + + return replacement; +} + +struct flb_encoding *flb_encoding_open(const char *encoding, const char *replacement) { + struct flb_encoding *ec; + TUTF8encoder encoder; + const char *invalid; + + if ((encoder = tutf8e_encoder(encoding)) == NULL) { + flb_error("[flb_encoding] unknown encoding: %s", encoding); + return NULL; + } + + invalid = parse_replacement(replacement); + + ec = flb_calloc(sizeof(struct flb_encoding),1); + + if (!ec) { + flb_errno(); + return NULL; + } + + if (invalid) { + invalid = flb_strdup(invalid); + if (!invalid) { + flb_errno(); + flb_free(ec); + return NULL; + } + } + + ec->encoder = encoder; + ec->invalid = invalid; + return ec; +} + + +int flb_encoding_decode(struct flb_encoding *ec, + char *str, size_t slen, + char **result, size_t *result_len) +{ + size_t outlen = 0; + char *outbuf; + int ret; + + *result = NULL; + *result_len = 0; + + if (slen == 0) { + *result = flb_strdup(""); + *result_len = 0; + return FLB_ENCODING_SUCCESS; + } + + ret = tutf8e_encoder_buffer_length(ec->encoder, str, ec->invalid, slen, &outlen); + + if (ret != TUTF8E_OK) { + return FLB_ENCODING_FAILURE; + } + + outbuf = flb_malloc(outlen + 1); + if(outbuf == NULL) { + flb_error("[flb_encoding] out of memory (%zu)", (int) outlen + 1); + return FLB_ENCODING_FAILURE; + } + + ret = tutf8e_encoder_buffer_encode(ec->encoder, str, slen, ec->invalid, outbuf, &outlen); + + if (ret != TUTF8E_OK) { + flb_free(outbuf); + return FLB_ENCODING_FAILURE; + } + + outbuf[outlen] = 0; + *result = outbuf; + *result_len = outlen; + + return FLB_ENCODING_SUCCESS; +} + +void flb_encoding_close(struct flb_encoding *ec) { + if (ec) { + if (ec->invalid) { + flb_free((char*)ec->invalid); + } + } +}