Merge branch 'master' of github.com:fluent/fluent-bit

Author: edsiper

include/fluent-bit/flb_plugin_proxy.h

@@ -57,6 +57,14 @@ struct flb_plugin_proxy {
     struct mk_list _head;     /* link to parent config->proxies              */
 };
 
+/* This is the context for proxy plugins */
+struct flb_plugin_proxy_context {
+    /* This context is set by the remote init and is passed to remote flush */
+    void *remote_context;
+    /* A proxy ptr is needed to detect the proxy type/lang (OUTPUT/GOLANG) */
+    struct flb_plugin_proxy *proxy;
+};
+
 void *flb_plugin_proxy_symbol(struct flb_plugin_proxy *proxy,
                               const char *symbol);
 

src/flb_output.c

@@ -272,7 +272,17 @@ struct flb_output_instance *flb_output_new(struct flb_config *config,
         instance->context = NULL;
     }
     else {
-        instance->context = plugin->proxy;
+        struct flb_plugin_proxy_context *ctx;
+
+        ctx = flb_calloc(1, sizeof(struct flb_plugin_proxy_context));
+        if (!ctx) {
+            perror("calloc");
+            return NULL;
+        }
+
+        ctx->proxy = plugin->proxy;
+
+        instance->context = ctx;
     }
 
     instance->alias       = NULL;

src/flb_plugin_proxy.c

@@ -49,18 +49,18 @@ static void flb_proxy_cb_flush(void *data, size_t bytes,
                                struct flb_config *config)
 {
     int ret = FLB_ERROR;
-    struct flb_plugin_proxy *p = out_context;
+    struct flb_plugin_proxy_context *ctx = out_context;
     (void) tag_len;
     (void) i_ins;
     (void) config;
 
 #ifdef FLB_HAVE_PROXY_GO
-    if (p->proxy == FLB_PROXY_GOLANG) {
+    if (ctx->proxy->proxy == FLB_PROXY_GOLANG) {
         flb_trace("[GO] entering go_flush()");
-        ret = proxy_go_flush(p, data, bytes, tag, tag_len);
+        ret = proxy_go_flush(ctx, data, bytes, tag, tag_len);
     }
 #else
-    (void) p;
+    (void) ctx;
 #endif
 
     if (ret != FLB_OK && ret != FLB_RETRY && ret != FLB_ERROR) {

src/proxy/go/go.c

@@ -42,6 +42,7 @@
  *      - name: shortname of the plugin.
  *      - description: plugin description.
  *      - type: input, output, filter, whatever.
+ *      - proxy: type of proxy e.g. GOLANG
  *      - flags: optional flags, not used by Go plugins at the moment.
  *
  *     this is done through Go Wrapper:
@@ -55,8 +56,11 @@ struct flbgo_output_plugin {
     char *name;
     void *api;
     void *o_ins;
+    struct flb_plugin_proxy_context *context;
+
     int (*cb_init)();
     int (*cb_flush)(void *, size_t, char *);
+    int (*cb_flush_ctx)(void *, void *, size_t, char *);
     int (*cb_exit)(void *);
 };
 /*------------------------EOF------------------------------------------------*/
@@ -76,6 +80,7 @@ int proxy_go_register(struct flb_plugin_proxy *proxy,
      *
      * - FLBPluginInit
      * - FLBPluginFlush
+     * - FLBPluginFlushCtx
      * - FLBPluginExit
      *
      * note: registration callback FLBPluginRegister() is resolved by the
@@ -90,6 +95,7 @@ int proxy_go_register(struct flb_plugin_proxy *proxy,
     }
 
     plugin->cb_flush = flb_plugin_proxy_symbol(proxy, "FLBPluginFlush");
+    plugin->cb_flush_ctx = flb_plugin_proxy_symbol(proxy, "FLBPluginFlushCtx");
     plugin->cb_exit  = flb_plugin_proxy_symbol(proxy, "FLBPluginExit");
     plugin->name     = flb_strdup(def->name);
 
@@ -107,6 +113,9 @@ int proxy_go_init(struct flb_plugin_proxy *proxy)
     /* set the API */
     plugin->api   = proxy->api;
     plugin->o_ins = proxy->instance;
+    // In order to avoid having the whole instance as part of the ABI we
+    // copy the context pointer into the plugin.
+    plugin->context = ((struct flb_output_instance *)proxy->instance)->context;
 
     ret = plugin->cb_init(plugin);
     if (ret <= 0) {
@@ -119,12 +128,12 @@ int proxy_go_init(struct flb_plugin_proxy *proxy)
     return ret;
 }
 
-int proxy_go_flush(struct flb_plugin_proxy *proxy, void *data, size_t size,
+int proxy_go_flush(struct flb_plugin_proxy_context *ctx, void *data, size_t size,
                    char *tag, int tag_len)
 {
     int ret;
     char *buf;
-    struct flbgo_output_plugin *plugin = proxy->data;
+    struct flbgo_output_plugin *plugin = ctx->proxy->data;
 
     /* temporal buffer for the tag */
     buf = flb_malloc(tag_len + 1);
@@ -136,7 +145,12 @@ int proxy_go_flush(struct flb_plugin_proxy *proxy, void *data, size_t size,
     memcpy(buf, tag, tag_len);
     buf[tag_len] = '\0';
 
-    ret = plugin->cb_flush(data, size, buf);
+    if (plugin->cb_flush_ctx) {
+        ret = plugin->cb_flush_ctx(ctx->remote_context, data, size, buf);
+    }
+    else {
+        ret = plugin->cb_flush(data, size, buf);
+    }
     flb_free(buf);
     return ret;
 }

src/proxy/go/go.h

@@ -29,7 +29,7 @@ int proxy_go_register(struct flb_plugin_proxy *proxy,
 
 int proxy_go_init(struct flb_plugin_proxy *proxy);
 
-int proxy_go_flush(struct flb_plugin_proxy *proxy, void *data, size_t size,
+int proxy_go_flush(struct flb_plugin_proxy_context  *ctx, void *data, size_t size,
                    char *tag, int tag_len);
 
 #endif