multiline: new api to destroy all stream_id in context

Signed-off-by: Eduardo Silva <[email protected]>

Author: edsiper

include/fluent-bit/multiline/flb_ml.h

@@ -268,6 +268,11 @@ int flb_ml_append_object(struct flb_ml *ml, uint64_t stream_id,
                          struct flb_time *tm, msgpack_object *obj);
 
 int flb_ml_parsers_init(struct flb_config *ctx);
+
+void flb_ml_flush_parser_instance(struct flb_ml *ml,
+                                  struct flb_ml_parser_ins *parser_i,
+                                  uint64_t stream_id);
+
 int flb_ml_auto_flush_init(struct flb_ml *ml);
 
 int flb_ml_flush_stream_group(struct flb_ml_parser *ml_parser,
@@ -288,6 +293,8 @@ int flb_ml_stream_create(struct flb_ml *ml,
 
 int flb_ml_stream_destroy(struct flb_ml_stream *mst);
 
+void flb_ml_stream_id_destroy_all(struct flb_ml *ml, uint64_t stream_id);
+
 struct flb_ml_stream *flb_ml_stream_get(struct flb_ml_parser_ins *parser,
                                         uint64_t stream_id);
 

src/multiline/flb_ml.c

@@ -80,9 +80,9 @@ int flb_ml_type_lookup(char *str)
     return type;
 }
 
-static void flb_ml_flush_parser_instance(struct flb_ml *ml,
-                                         struct flb_ml_parser_ins *parser_i,
-                                         uint64_t stream_id)
+void flb_ml_flush_parser_instance(struct flb_ml *ml,
+                                  struct flb_ml_parser_ins *parser_i,
+                                  uint64_t stream_id)
 {
     struct mk_list *head;
     struct mk_list *head_group;

src/multiline/flb_ml_stream.c

@@ -280,6 +280,41 @@ struct flb_ml_stream *flb_ml_stream_get(struct flb_ml_parser_ins *parser,
     return NULL;
 }
 
+void flb_ml_stream_id_destroy_all(struct flb_ml *ml, uint64_t stream_id)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct mk_list *head_group;
+    struct mk_list *head_stream;
+    struct flb_ml_group *group;
+    struct flb_ml_stream *mst;
+    struct flb_ml_parser_ins *parser_i;
+
+    /* groups */
+    mk_list_foreach(head, &ml->groups) {
+        group = mk_list_entry(head, struct flb_ml_group, _head);
+
+        /* parser instances */
+        mk_list_foreach(head_group, &group->parsers) {
+            parser_i = mk_list_entry(head_group, struct flb_ml_parser_ins, _head);
+
+            /* streams */
+            mk_list_foreach_safe(head_stream, tmp, &parser_i->streams) {
+                mst = mk_list_entry(head_stream, struct flb_ml_stream, _head);
+                if (mst->id != stream_id) {
+                    continue;
+                }
+
+                /* flush any pending data */
+                flb_ml_flush_parser_instance(ml, parser_i, stream_id);
+
+                /* destroy internal groups of the stream */
+                flb_ml_stream_destroy(mst);
+            }
+        }
+    }
+}
+
 int flb_ml_stream_destroy(struct flb_ml_stream *mst)
 {
     mk_list_del(&mst->_head);