in_tail: add flb_encoding support to in_tail-plugin

* Adds new option:  "encoding" to in_tail.
* If encoding fails. message is skipped.

Signed-off-by: Jukka Pihl <[email protected]>

Author: bluebike

plugins/in_tail/tail.c

@@ -613,6 +613,14 @@ 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 charset encoder to decode message",
+    },
+#endif
+
     /* Multiline Options */
 #ifdef FLB_HAVE_PARSER
     {

plugins/in_tail/tail_config.c

@@ -56,6 +56,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);
@@ -151,6 +154,18 @@ 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) {
+        ctx->encoding = flb_encoding_open(tmp);
+        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);
@@ -344,6 +359,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;
 }

plugins/in_tail/tail_config.h

@@ -29,6 +29,9 @@
 #ifdef FLB_HAVE_REGEX
 #include <fluent-bit/flb_regex.h>
 #endif
+#ifdef FLB_HAVE_UTF8_ENCODER
+#include <fluent-bit/flb_encoding.h>
+#endif
 
 /* Metrics */
 #ifdef FLB_HAVE_METRICS
@@ -94,6 +97,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;
 

plugins/in_tail/tail_file.c

@@ -221,6 +221,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);
@@ -262,6 +266,20 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
         line = data;
         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->docker_mode) {
             ret = flb_tail_dmode_process_content(now, line, line_len,
                                                  &repl_line, &repl_line_len,
@@ -306,7 +324,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) {
@@ -350,6 +368,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;
     *bytes = processed_bytes;
@@ -362,6 +386,7 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
                                out_sbuf->size);
 
     msgpack_sbuffer_destroy(out_sbuf);
+
     return lines;
 }