sp: allow RECORD_TIME() as parameter in timeseries functions (#1533)

Signed-off-by: Masoud Koleini <[email protected]>

Author: koleini

include/fluent-bit/stream_processor/flb_sp_parser.h

@@ -205,6 +205,7 @@ struct flb_exp_key {
     struct mk_list _head;
     flb_sds_t name;
     struct mk_list *subkeys;
+    int func;
 };
 
 struct flb_exp_func {
@@ -295,7 +296,7 @@ int flb_sp_cmd_gb_key_add(struct flb_sp_cmd *cmd, const char *key);
 void flb_sp_cmd_gb_key_del(struct flb_sp_cmd_gb_key *key);
 
 /* Timeseries */
-int flb_sp_cmd_param_add(struct flb_sp_cmd *cmd, struct flb_exp *param);
+int flb_sp_cmd_param_add(struct flb_sp_cmd *cmd, int func, struct flb_exp *param);
 int flb_sp_cmd_timeseries(struct flb_sp_cmd *cmd, char *func, const char *key_alias);
 void flb_cmd_params_del(struct mk_list *params);
 

src/flb_config.c

@@ -439,8 +439,7 @@ int flb_config_set_property(struct flb_config *config,
                         flb_free(*s_val); /* release before overwriting */
                     }
 
-                    *s_val = malloc(flb_sds_len(tmp) * sizeof(char));
-                    strncpy(*s_val, tmp, flb_sds_len(tmp));
+                    *s_val = flb_strdup(tmp);
                     flb_sds_destroy(tmp);
                     break;
                 default:

src/stream_processor/flb_sp.c

@@ -1402,6 +1402,23 @@ static int sp_process_timeseries_data(struct flb_sp_cmd *cmd,
 
         nums_ts = f->nums;
 
+        /* Populate pre-defined function parameters */
+        key_id_ts = 0;
+        mk_list_foreach(head_ts, &ckey->timeseries->params) {
+            param = mk_list_entry(head_ts, struct flb_exp_param, _head);
+
+            switch (param->param->type) {
+            case FLB_EXP_KEY:
+                exp_key = (struct flb_exp_key *) param->param;
+
+                if (exp_key->func == FLB_SP_RECORD_TIME) {
+                    nums_ts[key_id_ts].type = FLB_SP_NUM_F64;
+                    nums_ts[key_id_ts].f64 = flb_time_to_double(tms);
+                }
+            }
+            key_id_ts++;
+        }
+
         for (i = 0; i < map_size; i++) {
             key = map.via.map.ptr[i].key;
 

src/stream_processor/parser/flb_sp_parser.c

@@ -792,9 +792,19 @@ void flb_sp_cmd_condition_del(struct flb_sp_cmd *cmd)
 }
 
 /* Timeseries functions */
-int flb_sp_cmd_param_add(struct flb_sp_cmd *cmd, struct flb_exp *param)
+int flb_sp_cmd_param_add(struct flb_sp_cmd *cmd, int func, struct flb_exp *param)
 {
     struct flb_exp_param *p;
+    struct flb_exp_key *key;
+
+    if (func > 0) { /* Parameter is a pre-defined function */
+        key = (struct flb_exp_key *) flb_sp_cmd_condition_key(cmd, NULL);
+        if (!key) {
+            return -1;
+        }
+        key->func = func;
+        param = (struct flb_exp *) key;
+    }
 
     p = flb_calloc(1, sizeof(struct flb_exp_param));
 

src/stream_processor/parser/sql.y

@@ -512,19 +512,24 @@ select: SELECT keys FROM source window where groupby ';'
              subkey subkey
         param: IDENTIFIER
                {
-                   flb_sp_cmd_param_add(cmd, flb_sp_cmd_condition_key(cmd, $1));
+                   flb_sp_cmd_param_add(cmd, -1, flb_sp_cmd_condition_key(cmd, $1));
                    flb_free($1);
                }
                |
                IDENTIFIER record_subkey
                {
-                   flb_sp_cmd_param_add(cmd, flb_sp_cmd_condition_key(cmd, $1));
+                   flb_sp_cmd_param_add(cmd, -1, flb_sp_cmd_condition_key(cmd, $1));
                    flb_free($1);
                }
                |
+               RECORD_TIME '(' ')'
+               {
+                 flb_sp_cmd_param_add(cmd, FLB_SP_RECORD_TIME, NULL);
+               }
+               |
                value
                {
-                 flb_sp_cmd_param_add(cmd, $1);
+                 flb_sp_cmd_param_add(cmd, -1, $1);
                }
         params: param | param ',' params
         value: INTEGER