utils: plugins: Provide a choice for encoding raw UTF-8 strings

include/fluent-bit/flb_config.h

@@ -318,6 +318,8 @@ struct flb_config {
     struct flb_task_map *task_map;
     size_t task_map_size;
 
+    int json_escape_unicode;
+
     int dry_run;
 };
 
@@ -412,4 +414,7 @@ enum conf_type {
 #define FLB_CONF_STR_SCHED_CAP        "scheduler.cap"
 #define FLB_CONF_STR_SCHED_BASE       "scheduler.base"
 
+/* json escape */
+#define FLB_CONF_UNICODE_STR_JSON_ESCAPE "json.escape_unicode"
+
 #endif

include/fluent-bit/flb_pack.h

@@ -91,15 +91,17 @@ int flb_pack_json_valid(const char *json, size_t len);
 
 flb_sds_t flb_pack_msgpack_to_json_format(const char *data, uint64_t bytes,
                                           int json_format, int date_format,
-                                          flb_sds_t date_key);
+                                          flb_sds_t date_key, int escape_unicode);
 int flb_pack_to_json_format_type(const char *str);
 int flb_pack_to_json_date_type(const char *str);
 
 void flb_pack_print(const char *data, size_t bytes);
 int flb_msgpack_to_json(char *json_str, size_t str_len,
-                        const msgpack_object *obj);
-char* flb_msgpack_to_json_str(size_t size, const msgpack_object *obj);
-flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size);
+                        const msgpack_object *obj,
+                        int escape_unicode);
+char* flb_msgpack_to_json_str(size_t size, const msgpack_object *obj,
+                              int escape_unicode);
+flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size, int escape_unicode);
 
 int flb_pack_time_now(msgpack_packer *pck);
 int flb_msgpack_expand_map(char *map_data, size_t map_size,

include/fluent-bit/flb_utils.h

@@ -57,9 +57,9 @@ void flb_utils_bytes_to_human_readable_size(size_t bytes,
                                             char *out_buf, size_t size);
 int flb_utils_time_split(const char *time, int *sec, long *nsec);
 int flb_utils_write_str(char *buf, int *off, size_t size,
-                        const char *str, size_t str_len);
+                        const char *str, size_t str_len, int escape_unicode);
 int flb_utils_write_str_buf(const char *str, size_t str_len,
-                            char **out, size_t *out_size);
+                            char **out, size_t *out_size, int escape_unicode);
 
 int flb_utils_url_split(const char *in_url, char **out_protocol,
                         char **out_host, char **out_port, char **out_uri);

plugins/filter_expect/expect.c

@@ -272,7 +272,7 @@ static char *ra_value_type_to_str(struct flb_ra_value *val)
     return "UNKNOWN";
 }
 
-static int rule_apply(struct flb_expect *ctx, msgpack_object map)
+static int rule_apply(struct flb_expect *ctx, msgpack_object map, struct flb_config *config)
 {
     int n = 0;
     char *json;
@@ -292,7 +292,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
                 continue;
             }
 
-            json = flb_msgpack_to_json_str(size, &map);
+            json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
             flb_plg_error(ctx->ins,
                           "exception on rule #%i 'key_exists', key '%s' "
                           "not found. Record content:\n%s",
@@ -305,7 +305,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
                 n++;
                 continue;
             }
-            json = flb_msgpack_to_json_str(size, &map);
+            json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
             flb_plg_error(ctx->ins,
                           "exception on rule #%i 'key_not_exists', key '%s' "
                           "exists. Record content:\n%s",
@@ -316,7 +316,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
         }
         else if (rule->type == FLB_EXP_KEY_VAL_NULL) {
             if (!val) {
-                json = flb_msgpack_to_json_str(size, &map);
+                json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
                 flb_plg_error(ctx->ins,
                               "exception on rule #%i 'key_val_is_null', "
                               "key '%s' not found. Record content:\n%s",
@@ -325,7 +325,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
                 return FLB_FALSE;
             }
             if (val->type != FLB_RA_NULL) {
-                json = flb_msgpack_to_json_str(size, &map);
+                json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
                 flb_plg_error(ctx->ins,
                               "exception on rule #%i 'key_val_is_null', "
                               "key '%s' contains a value type '%s'. "
@@ -340,7 +340,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
         }
         else if (rule->type == FLB_EXP_KEY_VAL_NOT_NULL) {
             if (!val) {
-                json = flb_msgpack_to_json_str(size, &map);
+                json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
                 flb_plg_error(ctx->ins,
                               "exception on rule #%i 'key_val_is_not_null', "
                               "key '%s' not found. Record content:\n%s",
@@ -349,7 +349,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
                 return FLB_FALSE;
             }
             if (val->type == FLB_RA_NULL) {
-                json = flb_msgpack_to_json_str(size, &map);
+                json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
                 flb_plg_error(ctx->ins,
                               "exception on rule #%i 'key_val_is_not_null', "
                               "key '%s' contains a value type '%s'. "
@@ -364,7 +364,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
         }
         else if (rule->type == FLB_EXP_KEY_VAL_EQ) {
             if (!val) {
-                json = flb_msgpack_to_json_str(size, &map);
+                json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
                 flb_plg_error(ctx->ins,
                               "exception on rule #%i 'key_val_is_null', "
                               "key '%s' not found. Record content:\n%s",
@@ -376,7 +376,7 @@ static int rule_apply(struct flb_expect *ctx, msgpack_object map)
             if (val->type == FLB_RA_STRING) {
                 if (flb_sds_cmp(val->val.string, rule->expect,
                                 flb_sds_len(rule->expect)) != 0) {
-                    json = flb_msgpack_to_json_str(size, &map);
+                    json = flb_msgpack_to_json_str(size, &map, config->json_escape_unicode);
                     flb_plg_error(ctx->ins,
                                   "exception on rule #%i 'key_val_eq', "
                                   "key value '%s' is different than "
@@ -430,7 +430,7 @@ static int cb_expect_filter(const void *data, size_t bytes,
     while ((ret = flb_log_event_decoder_next(
                     &log_decoder,
                     &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
-        ret = rule_apply(ctx, *log_event.body);
+        ret = rule_apply(ctx, *log_event.body, config);
         if (ret == FLB_TRUE) {
             /* rule matches, we are good */
             continue;

plugins/filter_nightfall/nightfall_api.c

@@ -195,8 +195,8 @@ static flb_sds_t build_request_body(struct flb_filter_nightfall *ctx,
     msgpack_pack_str_with_body(&req_pk, "policyUUIDs", 11);
     msgpack_pack_array(&req_pk, 1);
     msgpack_pack_str_with_body(&req_pk, ctx->policy_id, 36);
-    
-    request_body = flb_msgpack_raw_to_json_sds(req_sbuf.data, req_sbuf.size);
+
+    request_body = flb_msgpack_raw_to_json_sds(req_sbuf.data, req_sbuf.size, FLB_TRUE);
 
     msgpack_sbuffer_destroy(&req_sbuf);
     flb_sds_destroy(num_str);

plugins/filter_wasm/filter_wasm.c

@@ -104,7 +104,7 @@ static int cb_wasm_filter(const void *data, size_t bytes,
         switch(ctx->event_format) {
         case FLB_FILTER_WASM_FMT_JSON:
             /* Encode as JSON from msgpack */
-            buf = flb_msgpack_to_json_str(alloc_size, log_event.body);
+            buf = flb_msgpack_to_json_str(alloc_size, log_event.body, config->json_escape_unicode);
 
             if (buf) {
                 /* Execute WASM program */

plugins/out_azure/azure.c

@@ -51,7 +51,8 @@ static int cb_azure_init(struct flb_output_instance *ins,
 static int azure_format(const void *in_buf, size_t in_bytes,
                         flb_sds_t tag, flb_sds_t *tag_val_out,
                         char **out_buf, size_t *out_size,
-                        struct flb_azure *ctx)
+                        struct flb_azure *ctx,
+                        struct flb_config *config)
 {
     int i;
     int array_size = 0;
@@ -160,7 +161,8 @@ static int azure_format(const void *in_buf, size_t in_bytes,
         msgpack_sbuffer_destroy(&tmp_sbuf);
     }
 
-    record = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
+    record = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size,
+                                         config->json_escape_unicode);
     if (!record) {
         flb_errno();
 
@@ -317,7 +319,8 @@ static void cb_azure_flush(struct flb_event_chunk *event_chunk,
 
     /* Convert binary logs into a JSON payload */
     ret = azure_format(event_chunk->data, event_chunk->size,
-                      event_chunk->tag, &final_log_type, &buf_data, &buf_size, ctx);
+                       event_chunk->tag, &final_log_type, &buf_data, &buf_size, ctx,
+                       config);
     /* If cannot get matching record using log_type_prefix, use log_type directly */
     if (!final_log_type) {
         final_log_type = ctx->log_type;

plugins/out_azure_blob/azure_blob.c

@@ -68,7 +68,8 @@ static int azure_blob_format(struct flb_config *config,
     out_buf = flb_pack_msgpack_to_json_format(data, bytes,
                                               FLB_PACK_JSON_FORMAT_LINES,
                                               FLB_PACK_JSON_DATE_ISO8601,
-                                              ctx->date_key);
+                                              ctx->date_key,
+                                              config->json_escape_unicode);
     if (!out_buf) {
         return -1;
     }

plugins/out_azure_kusto/azure_kusto.c

@@ -957,7 +957,8 @@ static int cb_azure_kusto_init(struct flb_output_instance *ins, struct flb_confi
      */
 static int azure_kusto_format(struct flb_azure_kusto *ctx, const char *tag, int tag_len,
                               const void *data, size_t bytes, void **out_data,
-                              size_t *out_size)
+                              size_t *out_size,
+                              struct flb_config *config)
 {
     int index;
     int records = 0;
@@ -1064,7 +1065,8 @@ static int azure_kusto_format(struct flb_azure_kusto *ctx, const char *tag, int
             msgpack_pack_str_body(&mp_pck, "log_attribute_missing", 20);
         }
 
-        flb_sds_t json_record = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
+        flb_sds_t json_record = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size,
+                                                            config->json_escape_unicode);
         if (!json_record) {
             flb_plg_error(ctx->ins, "error converting msgpack to JSON");
             flb_sds_destroy(out_buf);
@@ -1234,7 +1236,8 @@ static void cb_azure_kusto_flush(struct flb_event_chunk *event_chunk,
 
         /* Reformat msgpack to JSON payload */
         ret = azure_kusto_format(ctx, tag_name, tag_name_len, event_chunk->data,
-                                 event_chunk->size, (void **)&json, &json_size);
+                                 event_chunk->size, (void **)&json, &json_size,
+                                 config);
         if (ret != 0) {
             flb_plg_error(ctx->ins, "cannot reformat data into json");
             ret = FLB_RETRY;
@@ -1345,7 +1348,8 @@ static void cb_azure_kusto_flush(struct flb_event_chunk *event_chunk,
 
         /* Reformat msgpack data to JSON payload */
         ret = azure_kusto_format(ctx, event_chunk->tag, tag_len, event_chunk->data,
-                                 event_chunk->size, (void **)&json, &json_size);
+                                 event_chunk->size, (void **)&json, &json_size,
+                                 config);
         if (ret != 0) {
             flb_plg_error(ctx->ins, "cannot reformat data into json");
             ret = FLB_RETRY;

plugins/out_azure_logs_ingestion/azure_logs_ingestion.c

@@ -54,7 +54,8 @@ static int cb_azure_logs_ingestion_init(struct flb_output_instance *ins,
     allocates sds string */
 static int az_li_format(const void *in_buf, size_t in_bytes,
                         char **out_buf, size_t *out_size,
-                        struct flb_az_li *ctx)
+                        struct flb_az_li *ctx,
+                        struct flb_config *config)
 {
     int i;
     int array_size = 0;
@@ -141,7 +142,8 @@ static int az_li_format(const void *in_buf, size_t in_bytes,
         msgpack_sbuffer_destroy(&tmp_sbuf);
     }
 
-    record = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
+    record = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size,
+                                         config->json_escape_unicode);
     if (!record) {
         flb_errno();
         msgpack_sbuffer_destroy(&mp_sbuf);
@@ -266,7 +268,8 @@ static void cb_azure_logs_ingestion_flush(struct flb_event_chunk *event_chunk,
 
     /* Convert binary logs into a JSON payload */
     ret = az_li_format(event_chunk->data, event_chunk->size,
-                       &json_payload, &json_payload_size, ctx);
+                       &json_payload, &json_payload_size, ctx,
+                       config);
     if (ret == -1) {
         flb_upstream_conn_release(u_conn);
         FLB_OUTPUT_RETURN(FLB_ERROR);

plugins/out_bigquery/bigquery.c

@@ -847,7 +847,8 @@ static int cb_bigquery_init(struct flb_output_instance *ins,
 static int bigquery_format(const void *data, size_t bytes,
                            const char *tag, size_t tag_len,
                            char **out_data, size_t *out_size,
-                           struct flb_bigquery *ctx)
+                           struct flb_bigquery *ctx,
+                           struct flb_config *config)
 {
     int array_size = 0;
     flb_sds_t out_buf;
@@ -937,7 +938,8 @@ static int bigquery_format(const void *data, size_t bytes,
     }
 
     /* Convert from msgpack to JSON */
-    out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
+    out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size,
+                                          config->json_escape_unicode);
 
     flb_log_event_decoder_destroy(&log_decoder);
     msgpack_sbuffer_destroy(&mp_sbuf);
@@ -996,7 +998,7 @@ static void cb_bigquery_flush(struct flb_event_chunk *event_chunk,
     /* Reformat msgpack to bigquery JSON payload */
     ret = bigquery_format(event_chunk->data, event_chunk->size,
                           event_chunk->tag, flb_sds_len(event_chunk->tag),
-                          &payload_buf, &payload_size, ctx);
+                          &payload_buf, &payload_size, ctx, config);
     if (ret != 0) {
         flb_upstream_conn_release(u_conn);
         flb_sds_destroy(token);

plugins/out_calyptia/calyptia.c

@@ -311,7 +311,7 @@ static flb_sds_t get_agent_metadata(struct flb_calyptia *ctx)
     flb_mp_map_header_end(&mh);
 
     /* convert to json */
-    meta = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
+    meta = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size, FLB_TRUE); /* could be ASCII */
     msgpack_sbuffer_destroy(&mp_sbuf);
 
     return meta;
@@ -532,7 +532,7 @@ static int store_session_get(struct flb_calyptia *ctx,
     }
 
     /* decode */
-    json = flb_msgpack_raw_to_json_sds(buf, size);
+    json = flb_msgpack_raw_to_json_sds(buf, size, FLB_TRUE); /* TODO: could be ASCII? */
     flb_free(buf);
     if (!json) {
         return -1;
@@ -1029,7 +1029,8 @@ static void cb_calyptia_flush(struct flb_event_chunk *event_chunk,
                                             event_chunk->size,
                                             FLB_PACK_JSON_FORMAT_STREAM,
                                             FLB_PACK_JSON_DATE_DOUBLE,
-                                            NULL);
+                                            NULL,
+                                            FLB_TRUE); /* Trace is ASCII */
         if (json == NULL) {
             flb_upstream_conn_release(u_conn);
             FLB_OUTPUT_RETURN(FLB_RETRY);