lib: add option/property checks for input/output/filters. (#3317)

This patch implements helper methods in the library to validate
if a given k, v option is valid for a input/output/plugin.

Signed-off-by: Jorge Niedbalski <j@calyptia.com>

examples/out_lib/out_lib.c

@@ -20,13 +20,13 @@
 #include <fluent-bit.h>
 #include <msgpack.h>
 
-int my_stdout_json(void* data, size_t size)
+int my_stdout_json(void *record, size_t size, void *data)
 {
     printf("[%s]",__FUNCTION__);
-    printf("%s",(char*)data);
+    printf("%s",(char*)record);
     printf("\n");
 
-    flb_lib_free(data);
+    flb_lib_free(record);
     return 0;
 }
 
@@ -46,6 +46,7 @@ int main()
     int n;
     char tmp[256];
     flb_ctx_t *ctx;
+    struct flb_lib_out_cb callback;
     int in_ffd;
     int out_ffd;
 
@@ -61,7 +62,10 @@ int main()
     /* Register my callback function */
 
     /* JSON format */
-    out_ffd = flb_output(ctx, "lib", my_stdout_json);
+    callback.cb = my_stdout_json;
+    callback.data = NULL;
+
+    out_ffd = flb_output(ctx, "lib", &callback);
     flb_output_set(ctx, out_ffd, "match", "test", "format", "json", NULL);
 
     /* Msgpack format */

include/fluent-bit/flb_lib.h

@@ -28,6 +28,7 @@
 #define FLB_LIB_ERROR     -1
 #define FLB_LIB_NONE       0
 #define FLB_LIB_OK         1
+#define FLB_LIB_NO_CONFIG_MAP 2
 
 /* Library mode context data */
 struct flb_lib_ctx {
@@ -50,9 +51,12 @@ FLB_EXPORT void flb_init_env();
 FLB_EXPORT flb_ctx_t *flb_create();
 FLB_EXPORT void flb_destroy(flb_ctx_t *ctx);
 FLB_EXPORT int flb_input(flb_ctx_t *ctx, const char *input, void *data);
-FLB_EXPORT int flb_output(flb_ctx_t *ctx, const char *output, void *data);
+FLB_EXPORT int flb_output(flb_ctx_t *ctx, const char *output, struct flb_lib_out_cb *cb);
 FLB_EXPORT int flb_filter(flb_ctx_t *ctx, const char *filter, void *data);
 FLB_EXPORT int flb_input_set(flb_ctx_t *ctx, int ffd, ...);
+FLB_EXPORT int flb_input_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);
+FLB_EXPORT int flb_output_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);
+FLB_EXPORT int flb_filter_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);
 FLB_EXPORT int flb_output_set(flb_ctx_t *ctx, int ffd, ...);
 FLB_EXPORT int flb_output_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
                                    void (*out_callback) (void *, int, int,

src/flb_engine.c

@@ -735,8 +735,6 @@ int flb_engine_shutdown(struct flb_config *config)
     }
 #endif
 
-    flb_config_exit(config);
-
     return 0;
 }
 

src/flb_input.c

@@ -421,7 +421,7 @@ int flb_input_instance_init(struct flb_input_instance *ins,
     struct mk_list *config_map;
     struct flb_input_plugin *p = ins->p;
 
-    if (ins->log_level == -1) {
+    if (ins->log_level == -1 && config->log != NULL) {
         ins->log_level = config->log->level;
     }
 

src/flb_lib.c

@@ -30,6 +30,7 @@
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_coro.h>
 #include <fluent-bit/flb_callback.h>
+#include <fluent-bit/flb_kv.h>
 #include <fluent-bit/tls/flb_tls.h>
 
 #include <signal.h>
@@ -205,6 +206,13 @@ void flb_destroy(flb_ctx_t *ctx)
 
     /* Remove resources from the event loop */
     mk_event_loop_destroy(ctx->event_loop);
+
+    /* cfg->is_running is set to false when flb_engine_shutdown has been invoked (event loop) */
+    if(ctx->config && ctx->config->is_running == FLB_TRUE) {
+        flb_engine_shutdown(ctx->config);
+        flb_config_exit(ctx->config);
+    }
+
     flb_free(ctx);
     ctx = NULL;
 
@@ -228,11 +236,11 @@ int flb_input(flb_ctx_t *ctx, const char *input, void *data)
 }
 
 /* Defines a new output instance */
-int flb_output(flb_ctx_t *ctx, const char *output, void *data)
+int flb_output(flb_ctx_t *ctx, const char *output, struct flb_lib_out_cb *cb)
 {
     struct flb_output_instance *o_ins;
 
-    o_ins = flb_output_new(ctx->config, output, data);
+    o_ins = flb_output_new(ctx->config, output, cb);
     if (!o_ins) {
         return -1;
     }
@@ -286,6 +294,108 @@ int flb_input_set(flb_ctx_t *ctx, int ffd, ...)
     return 0;
 }
 
+static inline int flb_config_map_property_check(char *plugin_name, struct mk_list *config_map, char *key, char *val)
+{
+    struct flb_kv *kv;
+    struct mk_list properties;
+    int r;
+
+    mk_list_init(&properties);
+
+    kv = flb_kv_item_create(&properties, (char *) key, (char *) val);
+    if (!kv) {
+        return FLB_LIB_ERROR;
+    }
+
+    r = flb_config_map_properties_check(plugin_name, &properties, config_map);
+    flb_kv_item_destroy(kv);
+    return r;
+}
+
+/* Check if a given k, v is a valid config directive for the given output plugin */
+int flb_output_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val)
+{
+    struct flb_output_instance *o_ins;
+    struct mk_list *config_map;
+    struct flb_output_plugin *p;
+    int r;
+
+    o_ins = out_instance_get(ctx, ffd);
+    if (!o_ins) {
+      return FLB_LIB_ERROR;
+    }
+
+    p = o_ins->p;
+    if (!p->config_map) {
+        return FLB_LIB_NO_CONFIG_MAP;
+    }
+
+    config_map = flb_config_map_create(ctx->config, p->config_map);
+    if (!config_map) {
+        return FLB_LIB_ERROR;
+    }
+
+    r = flb_config_map_property_check(p->name, config_map, key, val);
+    flb_config_map_destroy(config_map);
+    return r;
+}
+
+/* Check if a given k, v is a valid config directive for the given input plugin */
+int flb_input_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val)
+{
+    struct flb_input_instance *i_ins;
+    struct flb_input_plugin *p;
+    struct mk_list *config_map;
+    int r;
+
+    i_ins = in_instance_get(ctx, ffd);
+    if (!i_ins) {
+      return FLB_LIB_ERROR;
+    }
+
+    p = i_ins->p;
+    if (!p->config_map) {
+        return FLB_LIB_NO_CONFIG_MAP;
+    }
+
+    config_map = flb_config_map_create(ctx->config, p->config_map);
+    if (!config_map) {
+        return FLB_LIB_ERROR;
+    }
+
+    r = flb_config_map_property_check(p->name, config_map, key, val);
+    flb_config_map_destroy(config_map);
+    return r;
+}
+
+/* Check if a given k, v is a valid config directive for the given filter plugin */
+int flb_filter_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val)
+{
+    struct flb_filter_instance *f_ins;
+    struct flb_filter_plugin *p;
+    struct mk_list *config_map;
+    int r;
+
+    f_ins = filter_instance_get(ctx, ffd);
+    if (!f_ins) {
+      return FLB_LIB_ERROR;
+    }
+
+    p = f_ins->p;
+    if (!p->config_map) {
+        return FLB_LIB_NO_CONFIG_MAP;
+    }
+
+    config_map = flb_config_map_create(ctx->config, p->config_map);
+    if (!config_map) {
+        return FLB_LIB_ERROR;
+    }
+
+    r = flb_config_map_property_check(p->name, config_map, key, val);
+    flb_config_map_destroy(config_map);
+    return r;
+}
+
 /* Set an output interface property */
 int flb_output_set(flb_ctx_t *ctx, int ffd, ...)
 {
@@ -584,6 +694,6 @@ int flb_stop(flb_ctx_t *ctx)
     flb_engine_exit(ctx->config);
     ret = pthread_join(tid, NULL);
     flb_debug("[lib] Fluent Bit engine stopped");
-
+    ctx->config = NULL;
     return ret;
 }

tests/runtime/CMakeLists.txt

@@ -47,6 +47,7 @@ endif()
 # Output Plugins
 if(FLB_IN_LIB)
   FLB_RT_TEST(FLB_OUT_LIB              "core_engine.c")
+  FLB_RT_TEST(FLB_OUT_LIB              "config_map_opts.c")
   FLB_RT_TEST(FLB_OUT_COUNTER          "out_counter.c")
   FLB_RT_TEST(FLB_OUT_DATADOG          "out_datadog.c")
   FLB_RT_TEST(FLB_OUT_ES               "out_elasticsearch.c")

tests/runtime/config_map_opts.c

@@ -0,0 +1,32 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+#include <fluent-bit.h>
+#include "flb_tests_runtime.h"
+
+/* Test functions */
+void flb_test_config_map_opts(void);
+
+/* Test list */
+TEST_LIST = {
+    {"config_map_opts",    flb_test_config_map_opts },
+    {NULL, NULL}
+};
+
+void flb_test_config_map_opts(void)
+{
+    flb_ctx_t    *ctx    = NULL;
+    int in_ffd, r;
+    ctx = flb_create();
+    in_ffd = flb_input(ctx, (char *) "tail", NULL);
+    r = flb_input_property_check(ctx, in_ffd, "invalid_option", "invalid value");
+    TEST_CHECK(r != 0);
+
+    in_ffd = flb_filter(ctx, (char *) "kubernetes", NULL);
+    r = flb_filter_property_check(ctx, in_ffd, "invalid_option", "invalid value");
+    TEST_CHECK(r != 0);
+
+    in_ffd = flb_output(ctx, (char *) "stdout", NULL);
+    r = flb_output_property_check(ctx, in_ffd, "invalid_option", "invalid value");
+    TEST_CHECK(r != 0);
+
+    flb_destroy(ctx);
+}