in_tail: Encoding parameter for input plugin for conversion to UTF8

Signed-off-by: Nigel Stewart <[email protected]>

Author: nigels-com

include/fluent-bit/flb_encode.h

@@ -23,6 +23,22 @@
 
 #include <msgpack.h>
 
-void flb_msgpack_iso_8859_2_as_utf8(msgpack_packer* pk, const void* b, size_t l);
+typedef void *flb_encoder;
+
+#ifdef FLB_HAVE_ENCODE
+flb_encoder flb_get_encoder(const char *encoding);
+void flb_msgpack_encode_utf8(flb_encoder enc, msgpack_packer* pk, const void* b, size_t l);
+#else
+static inline flb_encoder flb_get_encoder(const char *encoding)
+{
+    return NULL;
+}
+
+static inline void flb_msgpack_encode_utf8(flb_encoder enc, msgpack_packer* pk, const void* b, size_t l)
+{
+    msgpack_pack_str(pk, l);
+    msgpack_pack_str_body(pk, b, l);
+}
+#endif
 
 #endif /* FLB_ENCODE_H */

plugins/in_tail/tail_config.c

@@ -243,6 +243,15 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *i_ins,
         ctx->exit_on_eof = flb_utils_bool(tmp);
     }
 
+    /* Config: Text encoding other than UTF-8 */
+    tmp = flb_input_get_property("encoding", i_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_error("[in_tail] 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_encode.h>
 #ifdef FLB_HAVE_REGEX
 #include <fluent-bit/flb_regex.h>
 #endif
@@ -81,6 +82,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_file.c

@@ -208,7 +208,7 @@ int flb_tail_file_pack_line(msgpack_sbuffer *mp_sbuf, msgpack_packer *mp_pck,
 
     msgpack_pack_str(mp_pck, ctx->key_len);
     msgpack_pack_str_body(mp_pck, ctx->key, ctx->key_len);
-    flb_msgpack_iso_8859_2_as_utf8(mp_pck, data, data_size);
+    flb_msgpack_encode_utf8(ctx->encoding, mp_pck, data, data_size);
 
     return 0;
 }

plugins/in_tail/tail_multiline.c

@@ -239,7 +239,7 @@ static inline void flb_tail_mult_append_raw(char *buf, int size,
                                             struct flb_tail_config *config)
 {
     /* Append the raw string */
-    flb_msgpack_iso_8859_2_as_utf8(&file->mult_pck, buf, size);
+    flb_msgpack_encode_utf8(config->encoding, &file->mult_pck, buf, size);
 }
 
 /* Check if the last key value type of a map is string or not */

src/flb_encode.c

@@ -23,45 +23,48 @@
 
 #ifdef FLB_HAVE_ENCODE
 #include <tutf8e.h>
-#endif
 
-const size_t TUTF8_DEFAULT_BUFFER = 256;
+#define TUTF8_BUFFER_SIZE 256
 
-void flb_msgpack_iso_8859_2_as_utf8(msgpack_packer* pk, const void* b, size_t l)
+flb_encoder flb_get_encoder(const char *encoding)
 {
-#ifdef FLB_HAVE_ENCODE
-    size_t size = 0;
-    if (!tutf8e_buffer_length_iso_8859_2(b, l, &size) && size)
-    {
-        /* Already UTF8 encoded? */
-        if (size == l) {
-        }
-        /* Small enough for encoding to stack? */
-        else if (size<=TUTF8_DEFAULT_BUFFER)
-        {
-            size = TUTF8_DEFAULT_BUFFER;
-            char buffer[TUTF8_DEFAULT_BUFFER];
-            if (!tutf8e_buffer_encode_iso_8859_2(buffer, &size, b, l) && size) {
-                msgpack_pack_str(pk, size);
-                msgpack_pack_str_body(pk, buffer, size);
-                return;
+    return tutf8e_encoder(encoding);
+}
+
+void flb_msgpack_encode_utf8(flb_encoder enc, msgpack_packer* pk, const void* b, size_t l)
+{
+    if (enc) {
+        size_t size = 0;
+        if (!tutf8e_encoder_buffer_length(enc, b, l, &size) && size) {
+            /* Already UTF8 encoded? */
+            if (size == l) {
             }
-        }
-        /* malloc/free the encoded copy */
-        else {
-            char *buffer = (char *) flb_malloc(size);
-            if (buffer && !tutf8e_buffer_encode_iso_8859_2(buffer, &size, b, l) && size) {
-                msgpack_pack_str(pk, size);
-                msgpack_pack_str_body(pk, buffer, size);
+            /* Small enough for encoding to stack? */
+            else if (size<=TUTF8_BUFFER_SIZE) {
+                char buffer[TUTF8_BUFFER_SIZE];
+                if (!tutf8e_encoder_buffer_encode(enc, b, l, buffer, &size) && size) {
+                    msgpack_pack_str(pk, size);
+                    msgpack_pack_str_body(pk, buffer, size);
+                    return;
+                }
+            }
+            /* malloc/free the encoded copy */
+            else {
+                char *buffer = (char *) flb_malloc(size);
+                if (buffer && !tutf8e_encoder_buffer_encode(enc, b, l, buffer, &size) && size) {
+                    msgpack_pack_str(pk, size);
+                    msgpack_pack_str_body(pk, buffer, size);
+                    free(buffer);
+                    return;
+                }
                 free(buffer);
-                return;
-            }                        
-            free(buffer);
+            }
         }
     }
-#endif
 
     /* Could not or need not encode to UTF8 */
     msgpack_pack_str(pk, l);
     msgpack_pack_str_body(pk, b, l);
 }
+#endif
+