WIP: UTF8 encoding support for in_tail and in_syslog

include/fluent-bit/flb_encoder.h

@@ -0,0 +1,55 @@
+/* -*- 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_ENCODER_H
+#define FLB_ENCODER_H
+
+#include <msgpack.h>
+#include "flb_parser.h"
+
+typedef void *flb_encoder;
+
+#ifdef FLB_HAVE_UTF8_ENCODER
+flb_encoder flb_get_encoder(const char *encoding);
+void flb_msgpack_encode_utf8(flb_encoder encoder, const char *module, msgpack_packer *pk, const void *b, size_t l);
+int flb_parser_do_encode_utf8(flb_encoder encoder, const char *module,
+   struct flb_parser *parser, const char *buf, size_t length,
+   void **out_buf, size_t *out_size, struct flb_time *out_time);
+#else
+static inline flb_encoder flb_get_encoder(const char *encoding)
+{
+    return NULL;
+}
+
+static inline void flb_msgpack_encode_utf8(flb_encoder encoder, const char *module, msgpack_packer *pk, const void *b, size_t l)
+{
+    msgpack_pack_str(pk, l);
+    msgpack_pack_str_body(pk, b, l);
+}
+
+static inline void flb_parser_do_encode_utf8(flb_encoder encoder, const char *module,
+   struct flb_parser *parser, const char *buf, size_t length,
+   void **out_buf, size_t *out_size, struct flb_time *out_time)
+{
+	return flb_parser_do(parser, buf, length, out_buf, out_size, out_time);
+}
+#endif
+
+#endif /* FLB_ENCODER_H */

plugins/in_syslog/syslog.h

@@ -23,6 +23,7 @@
 
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_encoder.h>
 
 /* Syslog modes */
 #define FLB_SYSLOG_UNIX_TCP  1
@@ -56,6 +57,9 @@ struct flb_syslog {
     size_t buffer_max_size;
     size_t buffer_chunk_size;
 
+    /* text encoding, NULL for UTF8 */
+    flb_encoder encoding;
+
     /* Configuration */
     struct flb_parser *parser;
 

plugins/in_syslog/syslog_conf.c

@@ -114,6 +114,15 @@ struct flb_syslog *syslog_conf_create(struct flb_input_instance *ins,
         ctx->buffer_max_size  = flb_utils_size_to_bytes(tmp);
     }
 
+    /* Config: Text encoding other than UTF-8 */
+    tmp = flb_input_get_property("encoding", ins);
+    if (tmp) {
+        ctx->encoding = flb_get_encoder(tmp);
+        if (!ctx->encoding) {
+            flb_error("[in_syslog] encoding '%s' is not supported", tmp);
+        }
+    }
+
     /* Parser */
     tmp = flb_input_get_property("parser", ins);
     if (tmp) {

plugins/in_syslog/syslog_prot.c

@@ -21,6 +21,7 @@
 #include <fluent-bit/flb_input_plugin.h>
 #include <fluent-bit/flb_parser.h>
 #include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_encoder.h>
 
 #include "syslog.h"
 #include "syslog_conn.h"
@@ -97,7 +98,7 @@ int syslog_prot_process(struct syslog_conn *conn)
         }
 
         /* Process the string */
-        ret = flb_parser_do(ctx->parser, p, len,
+        ret = flb_parser_do_encode_utf8(ctx->encoding, "in_syslog", ctx->parser, p, len,
                             &out_buf, &out_size, &out_time);
         if (ret >= 0) {
             pack_line(ctx, &out_time, out_buf, out_size);
@@ -131,7 +132,7 @@ int syslog_prot_process_udp(char *buf, size_t size, struct flb_syslog *ctx)
     size_t out_size;
     struct flb_time out_time = {0};
 
-    ret = flb_parser_do(ctx->parser, buf, size,
+    ret = flb_parser_do_encode_utf8(ctx->encoding, "in_syslog", ctx->parser, buf, size,
                         &out_buf, &out_size, &out_time);
     if (ret >= 0) {
         if (flb_time_to_double(&out_time) == 0) {

plugins/in_tail/tail.c

@@ -503,6 +503,13 @@ static struct flb_config_map config_map[] = {
 
 #endif
 
+    /* Encoding Options */
+    {
+     FLB_CONFIG_MAP_STR, "encoding", NULL,
+     0, FLB_FALSE, 0,
+     "Encoding"
+    },
+
     /* EOF */
     {0}
 };

plugins/in_tail/tail_config.c

@@ -159,6 +159,15 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
         }
     }
 
+    /* Config: Text encoding other than UTF-8 */
+    tmp = flb_input_get_property("encoding", ins);
+    if (tmp) {
+        ctx->encoding = flb_get_encoder(tmp);
+        if (!ctx->encoding) {
+            flb_error("[in_tail] encoding '%s' is not supported", tmp);
+        }
+    }
+
     /* Validate buffer limit */
     if (ctx->buf_chunk_size > ctx->buf_max_size) {
         flb_plg_error(ctx->ins, "buffer_max_size must be >= buffer_chunk");

plugins/in_tail/tail_config.h

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_input.h>
 #include <fluent-bit/flb_parser.h>
 #include <fluent-bit/flb_macros.h>
+#include <fluent-bit/flb_encoder.h>
 #ifdef FLB_HAVE_REGEX
 #include <fluent-bit/flb_regex.h>
 #endif
@@ -80,6 +81,8 @@ struct flb_tail_config {
     int   skip_long_lines;     /* skip long lines              */
     int   exit_on_eof;         /* exit fluent-bit on EOF, test */
 
+    flb_encoder encoding;      /* text encoding, NULL for UTF8 */
+
     /* Database */
 #ifdef FLB_HAVE_SQLDB
     struct flb_sqldb *db;

plugins/in_tail/tail_dockermode.c

@@ -298,7 +298,8 @@ void flb_tail_dmode_flush(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
     }
 #endif
     flb_time_get(&out_time);
-    flb_tail_file_pack_line(mp_sbuf, mp_pck, &out_time,
+    /* Assumed to be UTF-8, no need to encode? */
+    flb_tail_file_pack_line(NULL, "in_tail", mp_sbuf, mp_pck, &out_time,
                             repl_line, repl_line_len, file);
 
  dmode_flush_end:

plugins/in_tail/tail_file.c

@@ -182,7 +182,8 @@ int flb_tail_pack_line_map(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
     return 0;
 }
 
-int flb_tail_file_pack_line(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
+int flb_tail_file_pack_line(flb_encoder encoder, const char *module,
+                            msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                             struct flb_time *time, char *data, size_t data_size,
                             struct flb_tail_file *file)
 {
@@ -207,8 +208,7 @@ int flb_tail_file_pack_line(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
 
     msgpack_pack_str(mp_pck, flb_sds_len(ctx->key));
     msgpack_pack_str_body(mp_pck, ctx->key, flb_sds_len(ctx->key));
-    msgpack_pack_str(mp_pck, data_size);
-    msgpack_pack_str_body(mp_pck, data, data_size);
+    flb_msgpack_encode_utf8(encoder, "in_tail", mp_pck, data, data_size);
 
     return 0;
 }
@@ -302,7 +302,7 @@ static int process_content(struct flb_tail_file *file, off_t *bytes)
 #ifdef FLB_HAVE_PARSER
         if (ctx->parser) {
             /* Common parser (non-multiline) */
-            ret = flb_parser_do(ctx->parser, line, line_len,
+            ret = flb_parser_do_encode_utf8(ctx->encoding, "in_tail", ctx->parser, line, line_len,
                                 &out_buf, &out_size, &out_time);
             if (ret >= 0) {
                 if (flb_time_to_double(&out_time) == 0) {
@@ -328,7 +328,8 @@ static int process_content(struct flb_tail_file *file, off_t *bytes)
             else {
                 /* Parser failed, pack raw text */
                 flb_time_get(&out_time);
-                flb_tail_file_pack_line(out_sbuf, out_pck, &out_time,
+                flb_tail_file_pack_line(ctx->encoding, "in_tail",
+                                        out_sbuf, out_pck, &out_time,
                                         data, len, file);
             }
         }
@@ -342,7 +343,8 @@ static int process_content(struct flb_tail_file *file, off_t *bytes)
                 flb_tail_mult_flush(out_sbuf, out_pck, file, ctx);
 
                 flb_time_get(&out_time);
-                flb_tail_file_pack_line(out_sbuf, out_pck, &out_time,
+                flb_tail_file_pack_line(ctx->encoding, "in_tail",
+                                        out_sbuf, out_pck, &out_time,
                                         line, line_len, file);
             }
             else if (ret == FLB_TAIL_MULT_MORE) {
@@ -355,12 +357,14 @@ static int process_content(struct flb_tail_file *file, off_t *bytes)
         }
         else {
             flb_time_get(&out_time);
-            flb_tail_file_pack_line(out_sbuf, out_pck, &out_time,
+            flb_tail_file_pack_line(ctx->encoding, "in_tail",
+                                    out_sbuf, out_pck, &out_time,
                                     line, line_len, file);
         }
 #else
         flb_time_get(&out_time);
-        flb_tail_file_pack_line(out_sbuf, out_pck, &out_time,
+        flb_tail_file_pack_line(ctx->encoding, "in_tail",
+                                out_sbuf, out_pck, &out_time,
                                 line, line_len, file);
 #endif
 

plugins/in_tail/tail_file.h

@@ -26,6 +26,7 @@
 
 #include <fluent-bit/flb_compat.h>
 #include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_encoder.h>
 
 #include "tail.h"
 #include "tail_fs.h"
@@ -75,8 +76,8 @@ int flb_tail_file_purge(struct flb_input_instance *ins,
 int flb_tail_pack_line_map(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                            struct flb_time *time, char **data,
                            size_t *data_size, struct flb_tail_file *file);
-int flb_tail_file_pack_line(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
+int flb_tail_file_pack_line(flb_encoder encoder, const char *module,
+                            msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
                             struct flb_time *time, char *data, size_t data_size,
                             struct flb_tail_file *file);
-
 #endif

plugins/in_tail/tail_multiline.c

@@ -21,6 +21,7 @@
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_input_plugin.h>
 #include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_encoder.h>
 #include <fluent-bit/flb_kv.h>
 
 #include "tail_config.h"
@@ -130,7 +131,7 @@ static int pack_line(char *data, size_t data_size, struct flb_tail_file *file,
     msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
     flb_time_get(&out_time);
 
-    flb_tail_file_pack_line(&mp_sbuf, &mp_pck, &out_time, data, data_size, file);
+    flb_tail_file_pack_line(NULL, "in_tail", &mp_sbuf, &mp_pck, &out_time, data, data_size, file);
     flb_input_chunk_append_raw(ctx->ins,
                                file->tag_buf,
                                file->tag_len,
@@ -278,7 +279,7 @@ int flb_tail_mult_process_content(time_t now,
     msgpack_unpacked result;
 
     /* Always check if this line is the beginning of a new multiline message */
-    ret = flb_parser_do(ctx->mult_parser_firstline,
+    ret = flb_parser_do_encode_utf8(ctx->encoding, "in_tail", ctx->mult_parser_firstline,
                         buf, len,
                         &out_buf, &out_size, &out_time);
     if (ret >= 0) {

src/CMakeLists.txt

@@ -111,6 +111,13 @@ if(FLB_HTTP_CLIENT_DEBUG)
     )
 endif()
 
+if(FLB_UTF8_ENCODER)
+  set(src
+    ${src}
+    "flb_encoder.c"
+    )
+endif()
+
 if(FLB_LUAJIT)
   set(src
     ${src}