plugin_proxy: Implement a capability to process custom Go plugins

Signed-off-by: Rob Skillington <[email protected]>
Signed-off-by: Hiroshi Hatake <[email protected]>

Co-authored-by: Rob Skillington <[email protected]>

Author: 2 people

include/fluent-bit/flb_plugin_proxy.h

@@ -28,6 +28,7 @@
 /* Plugin Types */
 #define FLB_PROXY_INPUT_PLUGIN     1
 #define FLB_PROXY_OUTPUT_PLUGIN    2
+#define FLB_PROXY_CUSTOM_PLUGIN    3
 
 /* Proxies available */
 #define FLB_PROXY_GOLANG          11

src/flb_plugin_proxy.c

@@ -33,6 +33,7 @@
 #include <fluent-bit/flb_utils.h>
 #include <fluent-bit/flb_plugin_proxy.h>
 #include <fluent-bit/flb_input_log.h>
+#include <fluent-bit/flb_custom.h>
 
 /* Proxies */
 #include "proxy/go/go.h"
@@ -422,6 +423,44 @@ static int flb_proxy_register_input(struct flb_plugin_proxy *proxy,
     return 0;
 }
 
+int flb_proxy_custom_cb_init(struct flb_custom_instance *c_ins,
+                             struct flb_config *config, void *data);
+
+static int flb_proxy_custom_cb_exit(void *custom_context,
+                                    struct flb_config *config);
+static void flb_proxy_custom_cb_destroy(struct flb_custom_plugin *plugin);
+
+static int flb_proxy_register_custom(struct flb_plugin_proxy *proxy,
+                                     struct flb_plugin_proxy_def *def,
+                                     struct flb_config *config)
+{
+    struct flb_custom_plugin *custom;
+
+    custom = flb_calloc(1, sizeof(struct flb_custom_plugin));
+    if (!custom) {
+        flb_errno();
+        return -1;
+    }
+
+    /* Plugin registration */
+    custom->type  = FLB_CUSTOM_PLUGIN_PROXY;
+    custom->proxy = proxy;
+    custom->flags = def->flags;
+    custom->name  = flb_strdup(def->name);
+    custom->description = def->description;
+    mk_list_add(&custom->_head, &config->custom_plugins);
+
+    /*
+     * Set proxy callbacks: external plugins which are not following
+     * the core plugins specs, have a different callback approach, so
+     * we put our proxy-middle callbacks to do the translation properly.
+     */
+    custom->cb_init = flb_proxy_custom_cb_init;
+    custom->cb_exit = flb_proxy_custom_cb_exit;
+    custom->cb_destroy = flb_proxy_custom_cb_destroy;
+    return 0;
+}
+
 void *flb_plugin_proxy_symbol(struct flb_plugin_proxy *proxy,
                               const char *symbol)
 {
@@ -487,6 +526,9 @@ int flb_plugin_proxy_register(struct flb_plugin_proxy *proxy,
         }
         else if (def->type == FLB_PROXY_INPUT_PLUGIN) {
             ret = proxy_go_input_register(proxy, def);
+        } 
+        else if (def->type == FLB_PROXY_CUSTOM_PLUGIN) {
+            ret = proxy_go_custom_register(proxy, def);
         }
 #endif
     }
@@ -501,6 +543,9 @@ int flb_plugin_proxy_register(struct flb_plugin_proxy *proxy,
         else if (def->type == FLB_PROXY_INPUT_PLUGIN) {
             flb_proxy_register_input(proxy, def, config);
         }
+        else if (def->type == FLB_PROXY_CUSTOM_PLUGIN) {
+            flb_proxy_register_custom(proxy, def, config);
+        }
     }
 
     return 0;
@@ -616,3 +661,70 @@ int flb_plugin_proxy_set(struct flb_plugin_proxy_def *def, int type,
 
     return 0;
 }
+
+int flb_proxy_custom_cb_init(struct flb_custom_instance *c_ins,
+                             struct flb_config *config, void *data)
+{
+    int ret = -1;
+    struct flb_plugin_proxy_context *pc;
+    struct flb_plugin_proxy *proxy;
+
+    pc = (struct flb_plugin_proxy_context *)(c_ins->context);
+    proxy = pc->proxy;
+
+    /* Before to initialize, set the instance reference */
+    pc->proxy->instance = c_ins;
+
+    if (proxy->def->proxy == FLB_PROXY_GOLANG) {
+#ifdef FLB_HAVE_PROXY_GO
+        ret = proxy_go_custom_init(proxy);
+#endif
+    }
+
+    if (ret == -1) {
+        flb_error("[custom] could not initialize '%s' plugin",
+                  c_ins->p->name);
+        return -1;
+    }
+
+    return 0;
+}
+
+int flb_proxy_custom_cb_exit(void *custom_context, 
+                             struct flb_config *config)
+{
+    int ret = -1;
+    struct flb_plugin_proxy_context *ctx = custom_context;
+    struct flb_plugin_proxy *proxy = (ctx->proxy);
+    if (!custom_context) {
+        return ret;
+    }
+
+    if (proxy->def->proxy == FLB_PROXY_GOLANG) {
+#ifdef FLB_HAVE_PROXY_GO
+        ret = proxy_go_custom_destroy(ctx);
+#endif
+    }
+
+    flb_free(ctx);
+    return ret;
+}
+
+static void flb_proxy_custom_cb_destroy(struct flb_custom_plugin *plugin)
+{
+    struct flb_plugin_proxy *proxy = (struct flb_plugin_proxy *) plugin->proxy;
+
+    if (plugin->name != NULL) {
+        flb_free(plugin->name);
+
+        plugin->name = NULL;
+    }
+
+    if (proxy->def->proxy == FLB_PROXY_GOLANG) {
+#ifdef FLB_HAVE_PROXY_GO
+        proxy_go_custom_unregister(proxy->data);
+#endif
+    }
+
+    flb_plugin_proxy_destroy(proxy);
+}

src/proxy/go/go.c

@@ -21,6 +21,7 @@
 #include <fluent-bit/flb_config.h>
 #include <fluent-bit/flb_plugin_proxy.h>
 #include <fluent-bit/flb_output.h>
+#include <fluent-bit/flb_custom.h>
 #include "./go.h"
 
 /*
@@ -41,7 +42,7 @@
  *
  *      - name: shortname of the plugin.
  *      - description: plugin description.
- *      - type: input, output, filter, whatever.
+ *      - type: input, output, filter, custom, whatever.
  *      - proxy: type of proxy e.g. GOLANG
  *      - flags: optional flags, not used by Go plugins at the moment.
  *
@@ -285,3 +286,87 @@ void proxy_go_input_unregister(void *data) {
     flb_free(plugin->name);
     flb_free(plugin);
 }
+
+int proxy_go_custom_register(struct flb_plugin_proxy *proxy,
+                            struct flb_plugin_proxy_def *def)
+{
+    struct flbgo_custom_plugin *plugin;
+
+    plugin = flb_malloc(sizeof(struct flbgo_custom_plugin));
+    if (!plugin) {
+        flb_errno();
+        return -1;
+    }
+
+    /*
+     * Lookup the entry point function:
+     *
+     * - FLBPluginInit
+     * - FLBPluginExit
+     *
+     * note: registration callback FLBPluginRegister() is resolved by the
+     * parent proxy interface.
+     */
+
+    plugin->cb_init  = flb_plugin_proxy_symbol(proxy, "FLBPluginInit");
+    if (!plugin->cb_init) {
+        flb_error("[go proxy]: could not load FLBPluginInit symbol");
+        flb_free(plugin);
+        return -1;
+    }
+
+    plugin->cb_exit  = flb_plugin_proxy_symbol(proxy, "FLBPluginExit");
+
+    plugin->name = flb_strdup(def->name);
+
+    /* This Go plugin context is an opaque data for the parent proxy */
+    proxy->data = plugin;
+
+    return 0;
+}
+
+int proxy_go_custom_init(struct flb_plugin_proxy *proxy)
+{
+    int ret = 0;
+    struct flbgo_custom_plugin *plugin = proxy->data;
+
+    /* set the API */
+    plugin->api   = proxy->api;
+    plugin->i_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_custom_instance *)proxy->instance)->context;
+
+    ret = plugin->cb_init(plugin);
+    if (ret <= 0) {
+        flb_error("[go proxy]: plugin '%s' failed to initialize",
+                  plugin->name);
+        flb_free(plugin);
+        return -1;
+    }
+
+    return ret;
+}
+
+int proxy_go_custom_destroy(struct flb_plugin_proxy_context *ctx)
+{
+    int ret = 0;
+    struct flbgo_custom_plugin *plugin;
+
+    plugin = (struct flbgo_custom_plugin *) ctx->proxy->data;
+
+    if (plugin->cb_exit) {
+        ret = plugin->cb_exit();
+    }
+
+    return ret;
+}
+
+void proxy_go_custom_unregister(void *data) {
+    struct flbgo_custom_plugin *plugin;
+
+    plugin = (struct flbgo_custom_plugin *) data;
+    flb_free(plugin->name);
+    flb_free(plugin);
+}
+

src/proxy/go/go.h

@@ -48,6 +48,16 @@ struct flbgo_input_plugin {
     int (*cb_exit)();
 };
 
+struct flbgo_custom_plugin {
+    char *name;
+    void *api;
+    void *i_ins;
+    struct flb_plugin_proxy_context *context;
+
+    int (*cb_init)();
+    int (*cb_exit)();
+};
+
 int proxy_go_output_register(struct flb_plugin_proxy *proxy,
                              struct flb_plugin_proxy_def *def);
 
@@ -69,4 +79,12 @@ int proxy_go_input_cleanup(struct flb_plugin_proxy *ctx,
                            void *allocated_data);
 int proxy_go_input_destroy(struct flb_plugin_input_proxy_context *ctx);
 void proxy_go_input_unregister(void *data);
+
+int proxy_go_custom_register(struct flb_plugin_proxy *proxy,
+                             struct flb_plugin_proxy_def *def);
+
+int proxy_go_custom_init(struct flb_plugin_proxy *proxy);
+
+int proxy_go_custom_destroy(struct flb_plugin_proxy_context *ctx);
+void proxy_go_custom_unregister(void *data);
 #endif