engine: support of tests which require generation of context passed to the test callback based on configuration of Fluent Bit and based on configuration of plugin

Author: mabrarov

include/fluent-bit/flb_lib.h

@@ -22,6 +22,7 @@
 
 #include <fluent-bit/flb_macros.h>
 #include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_input.h>
 
 /* Lib engine status */
 #define FLB_LIB_ERROR     -1
@@ -62,6 +63,16 @@ FLB_EXPORT int flb_output_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
                                                          void *, size_t, void *),
                                    void *out_callback_data,
                                    void *test_ctx);
+FLB_EXPORT int flb_output_set_test_with_ctx_callback(
+                                   flb_ctx_t *ctx, int ffd, char *test_name,
+                                   void (*out_callback) (void *, int, int,
+                                                         void *, size_t, void *),
+                                   void *out_callback_data,
+                                   void *test_ctx,
+                                   void *(*test_ctx_callback) (
+                                           struct flb_config *,
+                                           struct flb_input_instance *,
+                                           void *, void *));
 FLB_EXPORT int flb_output_set_callback(flb_ctx_t *ctx, int ffd, char *name,
                                        void (*cb)(char *, void *, void *));
 

include/fluent-bit/flb_output.h

@@ -136,9 +136,24 @@ struct flb_test_out_formatter {
      */
     void *rt_data;
 
-    /* optional context for flush callback */
+    /* optional context for "flush context callback" */
     void *flush_ctx;
 
+    /*
+     * Callback
+     * =========
+     * Optional "flush context callback": it references the function that extracts
+     * optional flush context for "formatter callback".
+     */
+    void *(*flush_ctx_callback) (/* Fluent Bit context */
+                                 struct flb_config *,
+                                 /* plugin that ingested the records */
+                                 struct flb_input_instance *,
+                                 /* plugin instance context */
+                                 void *plugin_context,
+                                 /* context for "flush context callback" */
+                                 void *flush_ctx);
+
     /*
      * Callback
      * =========

src/flb_engine_dispatch.c

@@ -100,18 +100,23 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry,
 static void test_run_formatter(struct flb_config *config,
                                struct flb_input_instance *i_ins,
                                struct flb_output_instance *o_ins,
-                               struct flb_task *task,
-                               void *flush_ctx)
+                               struct flb_task *task)
 {
     int ret;
     void *out_buf = NULL;
     size_t out_size = 0;
     struct flb_test_out_formatter *otf;
     struct flb_event_chunk *evc;
+    void *flush_ctx;
 
     otf = &o_ins->test_formatter;
     evc = task->event_chunk;
 
+    flush_ctx = otf->flush_ctx;
+    if (otf->flush_ctx_callback) {
+        flush_ctx = otf->flush_ctx_callback(config, i_ins, o_ins->context, flush_ctx);
+    }
+
     /* Invoke the output plugin formatter test callback */
     ret = otf->callback(config,
                         i_ins,
@@ -175,9 +180,7 @@ static int tasks_start(struct flb_input_instance *in,
                 out->test_formatter.callback != NULL) {
 
                 /* Run the formatter test */
-                test_run_formatter(config, in, out,
-                                   task,
-                                   out->test_formatter.flush_ctx);
+                test_run_formatter(config, in, out, task);
 
                 /* Remove the route */
                 mk_list_del(&route->_head);

src/flb_lib.c

@@ -482,6 +482,18 @@ int flb_output_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
                         void (*out_callback) (void *, int, int, void *, size_t, void *),
                         void *out_callback_data,
                         void *test_ctx)
+{
+    return flb_output_set_test_with_ctx_callback(ctx, ffd, test_name, out_callback,
+                                                 out_callback_data, test_ctx, NULL);
+}
+
+int flb_output_set_test_with_ctx_callback(flb_ctx_t *ctx, int ffd, char *test_name,
+                        void (*out_callback) (void *, int, int, void *, size_t, void *),
+                        void *out_callback_data,
+                        void *test_ctx,
+                        void *(*test_ctx_callback) (struct flb_config *,
+                                                    struct flb_input_instance *,
+                                                    void *, void *))
 {
     struct flb_output_instance *o_ins;
 
@@ -503,6 +515,7 @@ int flb_output_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
         o_ins->test_formatter.rt_out_callback = out_callback;
         o_ins->test_formatter.rt_data = out_callback_data;
         o_ins->test_formatter.flush_ctx = test_ctx;
+        o_ins->test_formatter.flush_ctx_callback = test_ctx_callback;
     }
     else {
         return -1;