sp: add hopping window to Stream Processor

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

Author: koleini

include/fluent-bit/stream_processor/flb_sp.h

@@ -61,14 +61,33 @@ struct flb_sp_window_data {
     struct mk_list _head;
 };
 
+struct flb_sp_hopping_slot {
+    struct rb_tree aggr_tree;
+    struct mk_list aggr_list;
+    int records;
+    struct mk_list _head;
+};
+
 struct flb_sp_task_window {
     int type;
 
     int fd;
     struct mk_event event;
+    struct mk_event event_hop;
 
     struct rb_tree aggr_tree;
     struct mk_list aggr_list;
+
+    /* Hopping window parameters */
+    /*
+     * first hopping window. Timer event is set to window size for the first,
+     * and will change to the advance_by time thereafter
+     */
+    bool first_hop;
+    int fd_hop;
+    int advance_by;
+    struct mk_list hopping_slot;
+
     int records;
 
     struct mk_list data;
@@ -114,7 +133,8 @@ int flb_sp_test_do(struct flb_sp *sp, struct flb_sp_task *task,
                    const char *tag, int tag_len,
                    const char *buf_data, size_t buf_size,
                    char **out_data, size_t *out_size);
-int flb_sp_test_fd_event(struct flb_sp_task *task, char **out_data, size_t *out_size);
+int flb_sp_test_fd_event(int fd, struct flb_sp_task *task, char **out_data,
+                         size_t *out_size);
 
 struct flb_sp_task *flb_sp_task_create(struct flb_sp *sp, const char *name,
                                        const char *query);

include/fluent-bit/stream_processor/flb_sp_parser.h

@@ -117,6 +117,7 @@ struct flb_sp_cmd_key {
 struct flb_sp_window {
     int type;
     time_t size;
+    time_t advance_by;
 };
 
 struct flb_sp_cmd {
@@ -222,8 +223,9 @@ void flb_sp_cmd_key_del(struct flb_sp_cmd_key *key);
 int flb_sp_cmd_source(struct flb_sp_cmd *cmd, int type, const char *source);
 void flb_sp_cmd_dump(struct flb_sp_cmd *cmd);
 
-void flb_sp_cmd_window(struct flb_sp_cmd *cmd,
-                       int window_type, int size, int time_unit);
+int flb_sp_cmd_window(struct flb_sp_cmd *cmd, int window_type,
+                      int size, int time_unit,
+                      int advance_by_size, int advance_by_time_unit);
 
 void flb_sp_cmd_condition_add(struct flb_sp_cmd *cmd, struct flb_exp *e);
 struct flb_exp *flb_sp_cmd_operation(struct flb_sp_cmd *cmd,

include/fluent-bit/stream_processor/flb_sp_window.h

@@ -20,6 +20,7 @@
 
 #define FLB_SP_WINDOW_DEFAULT   0
 #define FLB_SP_WINDOW_TUMBLING  1
+#define FLB_SP_WINDOW_HOPPING   2
 
 void flb_sp_window_prune(struct flb_sp_task *task);
 int flb_sp_window_populate(struct flb_sp_task *task, const char *buf_data,