in_tail: flb_parser_do_encode_utf8 for UTF8 encoding and parsing in one step

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

Author: nigels-com

include/fluent-bit/flb_encoder.h

@@ -22,12 +22,16 @@
 #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)
 {
@@ -39,6 +43,13 @@ static inline void flb_msgpack_encode_utf8(flb_encoder encoder, const char *modu
     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_prot.c

@@ -45,7 +45,7 @@ static inline int pack_line(struct flb_syslog *ctx,
 
     msgpack_pack_array(&mp_pck, 2);
     flb_time_append_to_msgpack(time, &mp_pck, 0);
-    flb_msgpack_encode_utf8(ctx->encoding, "in_syslog", &mp_pck, data, data_size);
+    msgpack_sbuffer_write(&mp_sbuf, data, data_size);
 
     flb_input_chunk_append_raw(ctx->i_ins, NULL, 0, mp_sbuf.data, mp_sbuf.size);
     msgpack_sbuffer_destroy(&mp_sbuf);
@@ -98,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);
@@ -130,7 +130,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_dockermode.c

@@ -295,7 +295,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

@@ -27,7 +27,6 @@
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_input.h>
 #include <fluent-bit/flb_parser.h>
-#include <fluent-bit/flb_encoder.h>
 #ifdef FLB_HAVE_REGEX
 #include <fluent-bit/flb_regex.h>
 #include <fluent-bit/flb_hash.h>
@@ -183,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)
 {
@@ -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_encode_utf8(ctx->encoding, "in_tail", 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"
@@ -63,8 +64,8 @@ int flb_tail_file_rotated_purge(struct flb_input_instance *i_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

@@ -137,7 +137,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->i_ins,
                                file->tag_buf,
                                file->tag_len,
@@ -239,7 +239,8 @@ static inline void flb_tail_mult_append_raw(char *buf, int size,
                                             struct flb_tail_config *config)
 {
     /* Append the raw string */
-    flb_msgpack_encode_utf8(config->encoding, "in_tail", &file->mult_pck, buf, size);
+    msgpack_pack_str(&file->mult_pck, size);
+    msgpack_pack_str_body(&file->mult_pck, buf, size);
 }
 
 /* Check if the last key value type of a map is string or not */
@@ -284,7 +285,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/flb_encoder.c

@@ -57,11 +57,11 @@ void flb_msgpack_encode_utf8(flb_encoder encoder, const char *module, msgpack_pa
                 if (buffer && !tutf8e_encoder_buffer_encode(encoder, b, l, buffer, &size) && size) {
                     msgpack_pack_str(pk, size);
                     msgpack_pack_str_body(pk, buffer, size);
-                    free(buffer);
+                    flb_free(buffer);
                     return;
                 }
                 /* Not expecting to get here ordinarily */
-                free(buffer);
+                flb_free(buffer);
                 flb_warn("[%s] failed to encode to UTF8", module);
             }
         }
@@ -74,5 +74,47 @@ void flb_msgpack_encode_utf8(flb_encoder encoder, const char *module, msgpack_pa
     msgpack_pack_str(pk, l);
     msgpack_pack_str_body(pk, b, l);
 }
+
+int flb_parser_do_encode_utf8(flb_encoder encoder, const char *module,
+   struct flb_parser *parser, const char *b, size_t l,
+   void **out_buf, size_t *out_size, struct flb_time *out_time)
+{
+    if (encoder) {
+        size_t size = 0;
+        if (!tutf8e_encoder_buffer_length(encoder, b, l, &size) && size) {
+            /* Already UTF8 encoded? */
+            if (size == l) {
+            }
+            /* Small enough for encoding to stack? */
+            else if (size<=TUTF8_BUFFER_SIZE) {
+                char buffer[TUTF8_BUFFER_SIZE];
+                if (!tutf8e_encoder_buffer_encode(encoder, b, l, buffer, &size) && size) {
+                    return flb_parser_do(parser, buffer, size, out_buf, out_size, out_time);
+                }
+                /* Not expecting to get here ordinarily */
+                flb_warn("[%s] failed to encode to UTF8", module);
+            }
+            /* malloc/free the encoded copy */
+            else {
+                char *buffer = (char *) flb_malloc(size);
+                if (buffer && !tutf8e_encoder_buffer_encode(encoder, b, l, buffer, &size) && size) {
+                    int ret = flb_parser_do(parser, buffer, size, out_buf, out_size, out_time);
+                    flb_free(buffer);
+                    return ret;
+                }
+                /* Not expecting to get here ordinarily */
+                flb_free(buffer);
+                flb_warn("[%s] failed to encode to UTF8", module);
+            }
+        }
+        else {
+            flb_warn("[%s] failed to encode to UTF8", module);
+        }
+    }
+
+    /* Could not or need not encode to UTF8 */
+    return flb_parser_do(parser, b, l, out_buf, out_size, out_time);
+}
+
 #endif