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

@@ -639,7 +639,7 @@ static struct flb_config_map config_map[] = {
      "specify the charset encoder to decode message",
     },
 #endif
-    
+
     /* Multiline Options */
 #ifdef FLB_HAVE_PARSER
     {

plugins/in_tail/tail_config.c

@@ -98,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);
@@ -441,6 +444,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

@@ -32,6 +32,9 @@
 #ifdef FLB_HAVE_PARSER
 #include <fluent-bit/multiline/flb_ml.h>
 #endif
+#ifdef FLB_HAVE_UTF8_ENCODER
+#include <fluent-bit/flb_encoding.h>
+#endif
 
 /* Metrics */
 #ifdef FLB_HAVE_METRICS
@@ -102,6 +105,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

@@ -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);
@@ -370,6 +374,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,
@@ -420,7 +436,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) {
@@ -464,6 +480,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;
 
@@ -518,6 +540,7 @@ static int process_content(struct flb_tail_file *file, size_t *bytes)
     }
 
     msgpack_sbuffer_destroy(out_sbuf);
+
     return lines;
 }