sp: prevent converting from string to num if config is set

Signed-off-by: Takahiro Yamashita <[email protected]>
fluent · Sep 23, 2023 · 3325727 · 3325727
1 parent 1b2306a
commit 3325727
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 9 deletions.
diff --git a/include/fluent-bit/stream_processor/flb_sp.h b/include/fluent-bit/stream_processor/flb_sp.h
@@ -171,7 +171,7 @@ int sp_process_data(const char *tag, int tag_len,
 int sp_process_data_aggr(const char *buf_data, size_t buf_size,
                          const char *tag, int tag_len,
                          struct flb_sp_task *task,
-                         struct flb_sp *sp);
+                         struct flb_sp *sp, int convert_str_to_num);
 void package_results(const char *tag, int tag_len,
                      char **out_buf, size_t *out_size,
                      struct flb_sp_task *task);

diff --git a/src/stream_processor/flb_sp.c b/src/stream_processor/flb_sp.c
@@ -349,7 +349,8 @@ static int string_to_number(const char *str, int len, int64_t *i, double *d)
  *
  * This function aims to take care of strings representing a value too.
  */
-static int object_to_number(msgpack_object obj, int64_t *i, double *d)
+static int object_to_number(msgpack_object obj, int64_t *i, double *d,
+                            int convert_str_to_num)
 {
     int ret;
     int64_t i_out;
@@ -366,7 +367,7 @@ static int object_to_number(msgpack_object obj, int64_t *i, double *d)
         *d = obj.via.f64;
         return FLB_STR_FLOAT;
     }
-    else if (obj.type == MSGPACK_OBJECT_STR) {
+    else if (obj.type == MSGPACK_OBJECT_STR && convert_str_to_num == FLB_TRUE) {
         /* A numeric representation of a string should not exceed 19 chars */
         if (obj.via.str.size > 19) {
             return -1;
@@ -1230,7 +1231,8 @@ void package_results(const char *tag, int tag_len,
 }
 
 static struct aggregate_node * sp_process_aggregate_data(struct flb_sp_task *task,
-                                                         msgpack_object map)
+                                                         msgpack_object map,
+                                                         int convert_str_to_num)
 {
     int i;
     int ret;
@@ -1289,7 +1291,7 @@ static struct aggregate_node * sp_process_aggregate_data(struct flb_sp_task *tas
                 values_found++;
 
                 /* Convert string to number if that is possible */
-                ret = object_to_number(sval->o, &ival, &dval);
+                ret = object_to_number(sval->o, &ival, &dval, convert_str_to_num);
                 if (ret == -1) {
                     if (sval->o.type == MSGPACK_OBJECT_STR) {
                         gb_nums[key_id].type = FLB_SP_STRING;
@@ -1386,7 +1388,8 @@ static struct aggregate_node * sp_process_aggregate_data(struct flb_sp_task *tas
 int sp_process_data_aggr(const char *buf_data, size_t buf_size,
                          const char *tag, int tag_len,
                          struct flb_sp_task *task,
-                         struct flb_sp *sp)
+                         struct flb_sp *sp,
+                         int convert_str_to_num)
 {
     int i;
     int ok;
@@ -1444,7 +1447,7 @@ int sp_process_data_aggr(const char *buf_data, size_t buf_size,
             }
         }
 
-        aggr_node = sp_process_aggregate_data(task, map);
+        aggr_node = sp_process_aggregate_data(task, map, convert_str_to_num);
         if (!aggr_node)
         {
             continue;
@@ -1497,7 +1500,7 @@ int sp_process_data_aggr(const char *buf_data, size_t buf_size,
                 ival = 0;
                 dval = 0.0;
                 if (ckey->aggr_func != FLB_SP_NOP) {
-                    ret = object_to_number(sval->o, &ival, &dval);
+                    ret = object_to_number(sval->o, &ival, &dval, convert_str_to_num);
                     if (ret == -1) {
                         /* Value cannot be represented as a number */
                         key_id++;
@@ -1987,7 +1990,7 @@ int flb_sp_do(struct flb_sp *sp, struct flb_input_instance *in,
         if (task->aggregate_keys == FLB_TRUE) {
             ret = sp_process_data_aggr(buf_data, buf_size,
                                        tag, tag_len,
-                                       task, sp);
+                                       task, sp, in->config->stream_processor_str_conv);
 
             if (ret == -1) {
                 flb_error("[sp] error processing records for '%s'",