proxy: send proxy context to go input plugin

This is necessary so that go input plugin can implement methods with
context in the same way as it is done in go output plugin.

Signed-off-by: milkrage <hello@milkrage.ru>

Author: milkrage

include/fluent-bit/flb_plugin_proxy.h

@@ -64,6 +64,8 @@ struct flb_plugin_proxy_context {
 
 struct flb_plugin_input_proxy_context {
     int coll_fd;
+    /* This context is set by the remote init and is passed to remote collect */
+    void *remote_context;
     /* A proxy ptr is needed to store the proxy type/lang (OUTPUT/GOLANG) */
     struct flb_plugin_proxy *proxy;
 };

src/flb_plugin_proxy.c

@@ -81,7 +81,7 @@ static int flb_proxy_input_cb_collect(struct flb_input_instance *ins,
 #ifdef FLB_HAVE_PROXY_GO
     if (ctx->proxy->def->proxy == FLB_PROXY_GOLANG) {
         flb_trace("[GO] entering go_collect()");
-        ret = proxy_go_input_collect(ctx->proxy, &data, &len);
+        ret = proxy_go_input_collect(ctx, &data, &len);
 
         if (len == 0) {
             flb_trace("[GO] No logs are ingested");
@@ -95,7 +95,7 @@ static int flb_proxy_input_cb_collect(struct flb_input_instance *ins,
 
         flb_input_log_append(ins, NULL, 0, data, len);
 
-        ret = proxy_go_input_cleanup(ctx->proxy, data);
+        ret = proxy_go_input_cleanup(ctx, data);
         if (ret == -1) {
             flb_errno();
             return -1;
@@ -110,19 +110,10 @@ static int flb_proxy_input_cb_init(struct flb_input_instance *ins,
                                    struct flb_config *config, void *data)
 {
     int ret = -1;
-    struct flb_plugin_input_proxy_context *ctx;
-    struct flb_plugin_proxy_context *pc;
-
-    /* Allocate space for the configuration context */
-    ctx = flb_malloc(sizeof(struct flb_plugin_input_proxy_context));
-    if (!ctx) {
-        flb_errno();
-        return -1;
-    }
+    struct flb_plugin_input_proxy_context *pc;
 
     /* Before to initialize for proxy, set the proxy instance reference */
-    pc = (struct flb_plugin_proxy_context *)(ins->context);
-    ctx->proxy = pc->proxy;
+    pc = (struct flb_plugin_input_proxy_context *)(ins->context);
 
     /* Before to initialize, set the instance reference */
     pc->proxy->instance = ins;
@@ -147,7 +138,7 @@ static int flb_proxy_input_cb_init(struct flb_input_instance *ins,
     }
 
     /* Set the context */
-    flb_input_set_context(ins, ctx);
+    flb_input_set_context(ins, pc);
 
     /* Collect upon data available on timer */
     ret = flb_input_set_collector_time(ins,
@@ -159,12 +150,12 @@ static int flb_proxy_input_cb_init(struct flb_input_instance *ins,
         flb_error("Could not set collector for threaded proxy input plugin");
         goto init_error;
     }
-    ctx->coll_fd = ret;
+    pc->coll_fd = ret;
 
     return ret;
 
 init_error:
-    flb_free(ctx);
+    flb_free(pc);
 
     return -1;
 }
@@ -311,10 +302,10 @@ static int flb_proxy_input_cb_pre_run(struct flb_input_instance *ins,
                                       struct flb_config *config, void *data)
 {
     int ret = -1;
-    struct flb_plugin_proxy_context *pc;
+    struct flb_plugin_input_proxy_context *pc;
     struct flb_plugin_proxy *proxy;
 
-    pc = (struct flb_plugin_proxy_context *)(ins->context);
+    pc = (struct flb_plugin_input_proxy_context *)(ins->context);
     proxy = pc->proxy;
 
     /* pre_run */