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

@@ -640,7 +640,7 @@ static struct flb_config_map config_map[] = {
      "Replacement in case of decoding error (default: unicode replacement char)",
     },
 #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);
@@ -460,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;
 }

plugins/in_tail/tail_config.h

@@ -33,6 +33,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 */
@@ -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;
 

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;
 }