Merging upstream version 1.44.3.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 942 insertions, 0 deletions

diff --git a/fluent-bit/src/flb_config.c b/fluent-bit/src/flb_config.c
new file mode 100644
index 00000000..882a93c7
--- /dev/null
+++ b/fluent-bit/src/flb_config.c
@@ -0,0 +1,942 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <signal.h>
+#include <stddef.h>
+
+#include <monkey/mk_core.h>
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_str.h>
+#include <fluent-bit/flb_kv.h>
+#include <fluent-bit/flb_env.h>
+#include <fluent-bit/flb_meta.h>
+#include <fluent-bit/flb_macros.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_parser.h>
+#include <fluent-bit/flb_plugin.h>
+#include <fluent-bit/flb_plugins.h>
+#include <fluent-bit/flb_slist.h>
+#include <fluent-bit/flb_kernel.h>
+#include <fluent-bit/flb_worker.h>
+#include <fluent-bit/flb_scheduler.h>
+#include <fluent-bit/flb_http_server.h>
+#include <fluent-bit/flb_plugin.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_config_format.h>
+#include <fluent-bit/multiline/flb_ml.h>
+#include <fluent-bit/flb_bucket_queue.h>
+
+const char *FLB_CONF_ENV_LOGLEVEL = "FLB_LOG_LEVEL";
+
+int flb_regex_init();
+
+struct flb_service_config service_configs[] = {
+    {FLB_CONF_STR_FLUSH,
+     FLB_CONF_TYPE_DOUBLE,
+     offsetof(struct flb_config, flush)},
+
+    {FLB_CONF_STR_GRACE,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, grace)},
+
+    {FLB_CONF_STR_CONV_NAN,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, convert_nan_to_null)},
+
+    {FLB_CONF_STR_DAEMON,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, daemon)},
+
+    {FLB_CONF_STR_LOGFILE,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, log_file)},
+
+    {FLB_CONF_STR_PARSERS_FILE,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, parsers_file)},
+
+    {FLB_CONF_STR_PLUGINS_FILE,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, plugins_file)},
+
+    {FLB_CONF_STR_LOGLEVEL,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, log)},
+
+#ifdef FLB_HAVE_HTTP_SERVER
+    {FLB_CONF_STR_HTTP_SERVER,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, http_server)},
+
+    {FLB_CONF_STR_HTTP_LISTEN,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, http_listen)},
+
+    {FLB_CONF_STR_HTTP_PORT,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, http_port)},
+
+     {FLB_CONF_STR_HEALTH_CHECK,
+      FLB_CONF_TYPE_BOOL,
+      offsetof(struct flb_config, health_check)},
+
+    {FLB_CONF_STR_HC_ERRORS_COUNT,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, hc_errors_count)},
+
+    {FLB_CONF_STR_HC_RETRIES_FAILURE_COUNT,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, hc_retry_failure_count)},
+
+    {FLB_CONF_STR_HC_PERIOD,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, health_check_period)},
+
+#endif
+    /* DNS*/
+    {FLB_CONF_DNS_MODE,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, dns_mode)},
+
+    {FLB_CONF_DNS_RESOLVER,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, dns_resolver)},
+
+    {FLB_CONF_DNS_PREFER_IPV4,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, dns_prefer_ipv4)},
+
+    /* Storage */
+    {FLB_CONF_STORAGE_PATH,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, storage_path)},
+    {FLB_CONF_STORAGE_SYNC,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, storage_sync)},
+    {FLB_CONF_STORAGE_METRICS,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, storage_metrics)},
+    {FLB_CONF_STORAGE_CHECKSUM,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, storage_checksum)},
+    {FLB_CONF_STORAGE_BL_MEM_LIMIT,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, storage_bl_mem_limit)},
+    {FLB_CONF_STORAGE_MAX_CHUNKS_UP,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, storage_max_chunks_up)},
+    {FLB_CONF_STORAGE_DELETE_IRRECOVERABLE_CHUNKS,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, storage_del_bad_chunks)},
+    {FLB_CONF_STORAGE_TRIM_FILES,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, storage_trim_files)},
+
+    /* Coroutines */
+    {FLB_CONF_STR_CORO_STACK_SIZE,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, coro_stack_size)},
+
+    /* Scheduler */
+    {FLB_CONF_STR_SCHED_CAP,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, sched_cap)},
+    {FLB_CONF_STR_SCHED_BASE,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, sched_base)},
+
+#ifdef FLB_HAVE_STREAM_PROCESSOR
+    {FLB_CONF_STR_STREAMS_FILE,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, stream_processor_file)},
+    {FLB_CONF_STR_STREAMS_STR_CONV,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, stream_processor_str_conv)},
+#endif
+
+#ifdef FLB_HAVE_CHUNK_TRACE
+    {FLB_CONF_STR_ENABLE_CHUNK_TRACE,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, enable_chunk_trace)},
+#endif
+
+    {FLB_CONF_STR_HOT_RELOAD,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, enable_hot_reload)},
+
+    {FLB_CONF_STR_HOT_RELOAD_ENSURE_THREAD_SAFETY,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, ensure_thread_safety_on_hot_reloading)},
+
+    {NULL, FLB_CONF_TYPE_OTHER, 0} /* end of array */
+};
+
+
+struct flb_config *flb_config_init()
+{
+    int ret;
+    struct flb_config *config;
+    struct flb_cf *cf;
+    struct flb_cf_section *section;
+
+    config = flb_calloc(1, sizeof(struct flb_config));
+    if (!config) {
+        flb_errno();
+        return NULL;
+    }
+
+    MK_EVENT_ZERO(&config->ch_event);
+    MK_EVENT_ZERO(&config->event_flush);
+    MK_EVENT_ZERO(&config->event_shutdown);
+
+    /* is data ingestion active ? */
+    config->is_ingestion_active = FLB_TRUE;
+
+    /* Is the engine (event loop) actively running ? */
+    config->is_running = FLB_TRUE;
+
+    /* Initialize config_format context */
+    cf = flb_cf_create();
+    if (!cf) {
+        flb_free(config);
+        return NULL;
+    }
+    config->cf_main = cf;
+
+    section = flb_cf_section_create(cf, "service", 0);
+    if (!section) {
+        flb_cf_destroy(cf);
+        flb_free(config);
+        return NULL;
+    }
+
+    /* Flush */
+    config->flush        = FLB_CONFIG_FLUSH_SECS;
+    config->daemon       = FLB_FALSE;
+    config->init_time    = time(NULL);
+    config->kernel       = flb_kernel_info();
+    config->verbose      = 3;
+    config->grace        = 5;
+    config->grace_count  = 0;
+    config->exit_status_code = 0;
+
+    /* json */
+    config->convert_nan_to_null = FLB_FALSE;
+
+#ifdef FLB_HAVE_HTTP_SERVER
+    config->http_ctx                     = NULL;
+    config->http_server                  = FLB_FALSE;
+    config->http_listen                  = flb_strdup(FLB_CONFIG_HTTP_LISTEN);
+    config->http_port                    = flb_strdup(FLB_CONFIG_HTTP_PORT);
+    config->health_check                 = FLB_FALSE;
+    config->hc_errors_count              = HC_ERRORS_COUNT_DEFAULT;
+    config->hc_retry_failure_count       = HC_RETRY_FAILURE_COUNTS_DEFAULT;
+    config->health_check_period          = HEALTH_CHECK_PERIOD;
+#endif
+
+    config->http_proxy = getenv("HTTP_PROXY");
+    if (flb_str_emptyval(config->http_proxy) == FLB_TRUE) {
+        config->http_proxy = getenv("http_proxy");
+        if (flb_str_emptyval(config->http_proxy) == FLB_TRUE) {
+            /* Proxy should not be set when `HTTP_PROXY` or `http_proxy` are set to "" */
+            config->http_proxy = NULL;
+        }
+    }
+    config->no_proxy = getenv("NO_PROXY");
+    if (flb_str_emptyval(config->no_proxy) == FLB_TRUE || config->http_proxy == NULL) {
+        config->no_proxy = getenv("no_proxy");
+        if (flb_str_emptyval(config->no_proxy) == FLB_TRUE || config->http_proxy == NULL) {
+            /* NoProxy  should not be set when `NO_PROXY` or `no_proxy` are set to "" or there is no Proxy. */
+            config->no_proxy = NULL;
+        }
+    }
+
+    config->cio          = NULL;
+    config->storage_path = NULL;
+    config->storage_input_plugin = NULL;
+    config->storage_metrics = FLB_TRUE;
+
+    config->sched_cap  = FLB_SCHED_CAP;
+    config->sched_base = FLB_SCHED_BASE;
+
+    /* reload */
+    config->ensure_thread_safety_on_hot_reloading = FLB_TRUE;
+    config->hot_reloaded_count = 0;
+
+#ifdef FLB_HAVE_SQLDB
+    mk_list_init(&config->sqldb_list);
+#endif
+
+#ifdef FLB_HAVE_LUAJIT
+    mk_list_init(&config->luajit_list);
+#endif
+
+#ifdef FLB_HAVE_STREAM_PROCESSOR
+    flb_slist_create(&config->stream_processor_tasks);
+    config->stream_processor_str_conv = FLB_TRUE;
+#endif
+
+    flb_slist_create(&config->external_plugins);
+
+    /* Set default coroutines stack size */
+    config->coro_stack_size = FLB_CORO_STACK_SIZE_BYTE;
+    if (config->coro_stack_size < getpagesize()) {
+        flb_info("[config] changing coro_stack_size from %u to %u bytes",
+                 config->coro_stack_size, getpagesize());
+        config->coro_stack_size = (unsigned int)getpagesize();
+    }
+
+    /* collectors */
+    pthread_mutex_init(&config->collectors_mutex, NULL);
+
+    /* Initialize linked lists */
+    mk_list_init(&config->processor_plugins);
+    mk_list_init(&config->custom_plugins);
+    mk_list_init(&config->in_plugins);
+    mk_list_init(&config->parser_plugins);
+    mk_list_init(&config->filter_plugins);
+    mk_list_init(&config->out_plugins);
+    mk_list_init(&config->customs);
+    mk_list_init(&config->inputs);
+    mk_list_init(&config->parsers);
+    mk_list_init(&config->filters);
+    mk_list_init(&config->outputs);
+    mk_list_init(&config->proxies);
+    mk_list_init(&config->workers);
+    mk_list_init(&config->upstreams);
+    mk_list_init(&config->downstreams);
+    mk_list_init(&config->cmetrics);
+    mk_list_init(&config->cf_parsers_list);
+
+    memset(&config->tasks_map, '\0', sizeof(config->tasks_map));
+
+    /* Initialize multiline-parser list. We need this here, because from now
+     * on we use flb_config_exit to cleanup the config, which requires
+     * the config->multiline_parsers list to be initialized. */
+    mk_list_init(&config->multiline_parsers);
+
+    /* Environment */
+    config->env = flb_env_create();
+    if (config->env == NULL) {
+        flb_error("[config] environment creation failed");
+        flb_config_exit(config);
+        return NULL;
+    }
+
+    /* Multiline core */
+    ret = flb_ml_init(config);
+    if (ret == -1) {
+        flb_error("[config] multiline core initialization failed");
+        flb_config_exit(config);
+        return NULL;
+    }
+
+    /* Register static plugins */
+    ret = flb_plugins_register(config);
+    if (ret == -1) {
+        flb_error("[config] plugins registration failed");
+        flb_config_exit(config);
+        return NULL;
+    }
+
+    /* Create environment for dynamic plugins */
+    config->dso_plugins = flb_plugin_create();
+
+    /* Ignoring SIGPIPE on Windows (scary) */
+#ifndef _WIN32
+    /* Ignore SIGPIPE */
+    signal(SIGPIPE, SIG_IGN);
+#endif
+
+    /* Prepare worker interface */
+    flb_worker_init(config);
+
+#ifdef FLB_HAVE_REGEX
+    /* Regex support */
+    flb_regex_init();
+#endif
+
+    return config;
+}
+
+void flb_config_exit(struct flb_config *config)
+{
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct flb_cf *cf;
+
+    if (config->log_file) {
+        flb_free(config->log_file);
+    }
+
+    if (config->log) {
+        flb_log_destroy(config->log, config);
+    }
+
+    if (config->parsers_file) {
+        flb_free(config->parsers_file);
+    }
+
+    if (config->plugins_file) {
+        flb_free(config->plugins_file);
+    }
+
+    if (config->kernel) {
+        flb_kernel_destroy(config->kernel);
+    }
+
+    /* release resources */
+    if (config->ch_event.fd) {
+        mk_event_closesocket(config->ch_event.fd);
+    }
+
+    /* Pipe */
+    if (config->ch_data[0]) {
+        mk_event_closesocket(config->ch_data[0]);
+        mk_event_closesocket(config->ch_data[1]);
+    }
+
+    /* Channel manager */
+    if (config->ch_manager[0] > 0) {
+        mk_event_closesocket(config->ch_manager[0]);
+        if (config->ch_manager[0] != config->ch_manager[1]) {
+            mk_event_closesocket(config->ch_manager[1]);
+        }
+    }
+
+    /* Channel notifications */
+    if (config->ch_notif[0] > 0) {
+        mk_event_closesocket(config->ch_notif[0]);
+        if (config->ch_notif[0] != config->ch_notif[1]) {
+            mk_event_closesocket(config->ch_notif[1]);
+        }
+    }
+
+    if (config->env) {
+        flb_env_destroy(config->env);
+    }
+
+    /* Program name */
+    if (config->program_name) {
+        flb_sds_destroy(config->program_name);
+    }
+
+    /* Conf path */
+    if (config->conf_path) {
+        flb_free(config->conf_path);
+    }
+
+    /* conf path file (file system config path) */
+    if (config->conf_path_file) {
+        flb_sds_destroy(config->conf_path_file);
+    }
+
+    /* Working directory */
+    if (config->workdir) {
+        flb_free(config->workdir);
+    }
+
+    /* Destroy any DSO context */
+    if (config->dso_plugins) {
+        flb_plugin_destroy(config->dso_plugins);
+    }
+
+    /* Workers */
+    flb_worker_exit(config);
+
+    /* Event flush */
+    if (config->evl) {
+        if (config->event_flush.status != MK_EVENT_NONE) {
+            mk_event_timeout_destroy(config->evl, &config->event_flush);
+        }
+    }
+
+    /* Release scheduler */
+    if (config->sched) {
+        flb_sched_destroy(config->sched);
+    }
+
+#ifdef FLB_HAVE_HTTP_SERVER
+    if (config->http_listen) {
+        flb_free(config->http_listen);
+    }
+
+    if (config->http_port) {
+        flb_free(config->http_port);
+    }
+#endif
+
+#ifdef FLB_HAVE_PARSER
+    /* parsers */
+    flb_parser_exit(config);
+#endif
+
+    if (config->dns_mode) {
+        flb_free(config->dns_mode);
+    }
+    if (config->dns_resolver) {
+        flb_free(config->dns_resolver);
+    }
+
+    if (config->storage_path) {
+        flb_free(config->storage_path);
+    }
+    if (config->storage_sync) {
+        flb_free(config->storage_sync);
+    }
+    if (config->storage_bl_mem_limit) {
+        flb_free(config->storage_bl_mem_limit);
+    }
+
+#ifdef FLB_HAVE_STREAM_PROCESSOR
+    if (config->stream_processor_file) {
+        flb_free(config->stream_processor_file);
+    }
+
+    flb_slist_destroy(&config->stream_processor_tasks);
+#endif
+
+    flb_slist_destroy(&config->external_plugins);
+
+    if (config->evl) {
+        mk_event_loop_destroy(config->evl);
+    }
+    if (config->evl_bktq) {
+        flb_bucket_queue_destroy(config->evl_bktq);
+    }
+
+    flb_plugins_unregister(config);
+
+    if (config->cf_main) {
+        flb_cf_destroy(config->cf_main);
+    }
+
+    /* cf_opts' lifetime should differ from config's lifetime.
+     * This member should be storing just for the cf_opts reference.
+     * Don't destroy it here.
+     */
+
+    /* remove parsers */
+    mk_list_foreach_safe(head, tmp, &config->cf_parsers_list) {
+        cf = mk_list_entry(head, struct flb_cf, _head);
+        mk_list_del(&cf->_head);
+        flb_cf_destroy(cf);
+    }
+
+    flb_free(config);
+}
+
+const char *flb_config_prop_get(const char *key, struct mk_list *list)
+{
+    return flb_kv_get_key_value(key, list);
+}
+
+static inline int prop_key_check(const char *key, const char *kv, int k_len)
+{
+    size_t len;
+
+    len = strnlen(key,256);
+    if (strncasecmp(key, kv, k_len) == 0 && len == k_len) {
+        return 0;
+    }
+    return -1;
+}
+
+static int set_log_level(struct flb_config *config, const char *v_str)
+{
+    if (v_str != NULL) {
+        if (strcasecmp(v_str, "error") == 0) {
+            config->verbose = 1;
+        }
+        else if (strcasecmp(v_str, "warn") == 0 ||
+                 strcasecmp(v_str, "warning") == 0) {
+            config->verbose = 2;
+        }
+        else if (strcasecmp(v_str, "info") == 0) {
+            config->verbose = 3;
+        }
+        else if (strcasecmp(v_str, "debug") == 0) {
+            config->verbose = 4;
+        }
+        else if (strcasecmp(v_str, "trace") == 0) {
+            config->verbose = 5;
+        }
+        else if (strcasecmp(v_str, "off") == 0) {
+            config->verbose = FLB_LOG_OFF;
+        }
+        else {
+            return -1;
+        }
+    }
+    else if (config->log) {
+        config->verbose = 3;
+    }
+    return 0;
+}
+
+int set_log_level_from_env(struct flb_config *config)
+{
+    const char *val = NULL;
+    val = flb_env_get(config->env, FLB_CONF_ENV_LOGLEVEL);
+    if (val) {
+        return set_log_level(config, val);
+    }
+    return -1;
+}
+
+int flb_config_set_property(struct flb_config *config,
+                            const char *k, const char *v)
+{
+    int i=0;
+    int ret = -1;
+    int *i_val;
+    double *d_val;
+    char **s_val;
+    size_t len = strnlen(k, 256);
+    char *key = service_configs[0].key;
+    flb_sds_t tmp = NULL;
+
+    while (key != NULL) {
+        if (prop_key_check(key, k,len) == 0) {
+            if (!strncasecmp(key, FLB_CONF_STR_LOGLEVEL, 256)) {
+                #ifndef FLB_HAVE_STATIC_CONF
+                if (set_log_level_from_env(config) < 0) {
+                #endif
+                    tmp = flb_env_var_translate(config->env, v);
+                    if (tmp) {
+                        ret = set_log_level(config, tmp);
+                        flb_sds_destroy(tmp);
+                        tmp = NULL;
+                    }
+                    else {
+                        ret = set_log_level(config, v);
+                    }
+                #ifndef FLB_HAVE_STATIC_CONF
+                }
+                #endif
+            }
+            else if (!strncasecmp(key, FLB_CONF_STR_PARSERS_FILE, 32)) {
+#ifdef FLB_HAVE_PARSER
+                tmp = flb_env_var_translate(config->env, v);
+                ret = flb_parser_conf_file(tmp, config);
+                flb_sds_destroy(tmp);
+                tmp = NULL;
+#endif
+            }
+            else if (!strncasecmp(key, FLB_CONF_STR_PLUGINS_FILE, 32)) {
+                tmp = flb_env_var_translate(config->env, v);
+                ret = flb_plugin_load_config_file(tmp, config);
+                flb_sds_destroy(tmp);
+                tmp = NULL;
+            }
+            else {
+                ret = 0;
+                tmp = flb_env_var_translate(config->env, v);
+                switch(service_configs[i].type) {
+                case FLB_CONF_TYPE_INT:
+                    i_val  = (int*)((char*)config + service_configs[i].offset);
+                    *i_val = atoi(tmp);
+                    flb_sds_destroy(tmp);
+                    break;
+                case FLB_CONF_TYPE_DOUBLE:
+                    d_val  = (double*)((char*)config + service_configs[i].offset);
+                    *d_val = atof(tmp);
+                    flb_sds_destroy(tmp);
+                    break;
+                case FLB_CONF_TYPE_BOOL:
+                    i_val = (int*)((char*)config+service_configs[i].offset);
+                    *i_val = flb_utils_bool(tmp);
+                    flb_sds_destroy(tmp);
+                    break;
+                case FLB_CONF_TYPE_STR:
+                    s_val = (char**)((char*)config+service_configs[i].offset);
+                    if ( *s_val != NULL ) {
+                        flb_free(*s_val); /* release before overwriting */
+                    }
+
+                    *s_val = flb_strdup(tmp);
+                    flb_sds_destroy(tmp);
+                    break;
+                default:
+                    ret = -1;
+                }
+            }
+
+            if (ret < 0) {
+                if (tmp) {
+                    flb_sds_destroy(tmp);
+                }
+                return -1;
+            }
+            return 0;
+        }
+        key = service_configs[++i].key;
+    }
+    return 0;
+}
+
+int flb_config_set_program_name(struct flb_config *config, char *name)
+{
+    config->program_name = flb_sds_create(name);
+
+    if (!config->program_name) {
+        return -1;
+    }
+
+    return 0;
+}
+
+static int configure_plugins_type(struct flb_config *config, struct flb_cf *cf, enum section_type type)
+{
+    int ret;
+    char *tmp;
+    char *name;
+    char *s_type;
+    struct mk_list *list;
+    struct mk_list *head;
+    struct cfl_list *h_prop;
+    struct cfl_kvpair *kv;
+    struct cfl_variant *val;
+    struct flb_cf_section *s;
+    struct flb_cf_group *processors = NULL;
+    int i;
+    void *ins;
+
+    if (type == FLB_CF_CUSTOM) {
+        s_type = "custom";
+        list = &cf->customs;
+    }
+    else if (type == FLB_CF_INPUT) {
+        s_type = "input";
+        list = &cf->inputs;
+    }
+    else if (type == FLB_CF_FILTER) {
+        s_type = "filter";
+        list = &cf->filters;
+    }
+    else if (type == FLB_CF_OUTPUT) {
+        s_type = "output";
+        list = &cf->outputs;
+    }
+    else {
+        return -1;
+    }
+
+    mk_list_foreach(head, list) {
+        s = mk_list_entry(head, struct flb_cf_section, _head_section);
+        name = flb_cf_section_property_get_string(cf, s, "name");
+        if (!name) {
+            flb_error("[config] section '%s' is missing the 'name' property",
+                      s_type);
+            return -1;
+        }
+
+        /* translate the variable */
+        tmp = flb_env_var_translate(config->env, name);
+
+        /* create an instance of the plugin */
+        ins = NULL;
+        if (type == FLB_CF_CUSTOM) {
+            ins = flb_custom_new(config, tmp, NULL);
+        }
+        else if (type == FLB_CF_INPUT) {
+            ins = flb_input_new(config, tmp, NULL, FLB_TRUE);
+        }
+        else if (type == FLB_CF_FILTER) {
+            ins = flb_filter_new(config, tmp, NULL);
+        }
+        else if (type == FLB_CF_OUTPUT) {
+            ins = flb_output_new(config, tmp, NULL, FLB_TRUE);
+        }
+        flb_sds_destroy(tmp);
+
+        /* validate the instance creation */
+        if (!ins) {
+            flb_error("[config] section '%s' tried to instance a plugin name "
+                      "that don't exists", name);
+            flb_sds_destroy(name);
+            return -1;
+        }
+        flb_sds_destroy(name);
+
+        /*
+         * iterate section properties and populate instance by using specific
+         * api function.
+         */
+        cfl_list_foreach(h_prop, &s->properties->list) {
+            kv = cfl_list_entry(h_prop, struct cfl_kvpair, _head);
+            if (strcasecmp(kv->key, "name") == 0) {
+                continue;
+            }
+
+            /* set ret to -1 to ensure that we treat any unhandled plugin or
+             * value types as errors.
+             */
+            ret = -1;
+
+            if (type == FLB_CF_CUSTOM) {
+                if (kv->val->type == CFL_VARIANT_STRING) {
+                    ret = flb_custom_set_property(ins, kv->key, kv->val->data.as_string);
+                } else if (kv->val->type == CFL_VARIANT_ARRAY) {
+                    for (i = 0; i < kv->val->data.as_array->entry_count; i++) {
+                        val = kv->val->data.as_array->entries[i];
+                        ret = flb_custom_set_property(ins, kv->key, val->data.as_string);
+                    }
+                }
+            }
+            else if (type == FLB_CF_INPUT) {
+                 if (kv->val->type == CFL_VARIANT_STRING) {
+                    ret = flb_input_set_property(ins, kv->key, kv->val->data.as_string);
+                } else if (kv->val->type == CFL_VARIANT_ARRAY) {
+                    for (i = 0; i < kv->val->data.as_array->entry_count; i++) {
+                        val = kv->val->data.as_array->entries[i];
+                        ret = flb_input_set_property(ins, kv->key, val->data.as_string);
+                    }
+                }
+            }
+            else if (type == FLB_CF_FILTER) {
+                 if (kv->val->type == CFL_VARIANT_STRING) {
+                    ret = flb_filter_set_property(ins, kv->key, kv->val->data.as_string);
+                } else if (kv->val->type == CFL_VARIANT_ARRAY) {
+                    for (i = 0; i < kv->val->data.as_array->entry_count; i++) {
+                        val = kv->val->data.as_array->entries[i];
+                        ret = flb_filter_set_property(ins, kv->key, val->data.as_string);
+                    }
+                }
+            }
+            else if (type == FLB_CF_OUTPUT) {
+                 if (kv->val->type == CFL_VARIANT_STRING) {
+                    ret = flb_output_set_property(ins, kv->key, kv->val->data.as_string);
+                } else if (kv->val->type == CFL_VARIANT_ARRAY) {
+                    for (i = 0; i < kv->val->data.as_array->entry_count; i++) {
+                        val = kv->val->data.as_array->entries[i];
+                        ret = flb_output_set_property(ins, kv->key, val->data.as_string);
+                    }
+                }
+            }
+
+            if (ret == -1) {
+                flb_error("[config] could not configure property '%s' on "
+                          "%s plugin with section name '%s'",
+                          kv->key, s_type, name);
+            }
+        }
+
+        /* Processors */
+        processors = flb_cf_group_get(cf, s, "processors");
+        if (processors) {
+            if (type == FLB_CF_INPUT) {
+                flb_processors_load_from_config_format_group(((struct flb_input_instance *) ins)->processor, processors);
+            }
+            else if (type == FLB_CF_OUTPUT) {
+                flb_processors_load_from_config_format_group(((struct flb_output_instance *) ins)->processor, processors);
+            }
+            else {
+                flb_error("[config] section '%s' does not support processors", s_type);
+            }
+        }
+    }
+
+    return 0;
+}
+/* Load a struct flb_config_format context into a flb_config instance */
+int flb_config_load_config_format(struct flb_config *config, struct flb_cf *cf)
+{
+    int ret;
+    struct flb_kv *kv;
+    struct mk_list *head;
+    struct cfl_kvpair *ckv;
+    struct cfl_list *chead;
+    struct flb_cf_section *s;
+
+    /* Process config environment vars */
+    mk_list_foreach(head, &cf->env) {
+        kv = mk_list_entry(head, struct flb_kv, _head);
+        ret = flb_env_set(config->env, kv->key, kv->val);
+        if (ret == -1) {
+            flb_error("could not set config environment variable '%s'", kv->key);
+            return -1;
+        }
+    }
+
+    /* Process all meta commands */
+    mk_list_foreach(head, &cf->metas) {
+        kv = mk_list_entry(head, struct flb_kv, _head);
+        flb_meta_run(config, kv->key, kv->val);
+    }
+
+    /* Validate sections */
+    mk_list_foreach(head, &cf->sections) {
+        s = mk_list_entry(head, struct flb_cf_section, _head);
+
+        if (strcasecmp(s->name, "env") == 0 ||
+            strcasecmp(s->name, "service") == 0 ||
+            strcasecmp(s->name, "custom") == 0 ||
+            strcasecmp(s->name, "input") == 0 ||
+            strcasecmp(s->name, "filter") == 0 ||
+            strcasecmp(s->name, "output") == 0) {
+
+            /* continue on valid sections */
+            continue;
+        }
+
+        /* Extra sanity checks */
+        if (strcasecmp(s->name, "parser") == 0 ||
+            strcasecmp(s->name, "multiline_parser") == 0) {
+            fprintf(stderr,
+                    "Sections 'multiline_parser' and 'parser' are not valid in "
+                    "the main configuration file. It belongs to \n"
+                    "the 'parsers_file' configuration files.\n");
+            return -1;
+        }
+    }
+
+    /* Read main 'service' section */
+    s = cf->service;
+    if (s) {
+        /* Iterate properties */
+        cfl_list_foreach(chead, &s->properties->list) {
+            ckv = cfl_list_entry(chead, struct cfl_kvpair, _head);
+            flb_config_set_property(config, ckv->key, ckv->val->data.as_string);
+        }
+    }
+
+    ret = configure_plugins_type(config, cf, FLB_CF_CUSTOM);
+    if (ret == -1) {
+        return -1;
+    }
+
+    ret = configure_plugins_type(config, cf, FLB_CF_INPUT);
+    if (ret == -1) {
+        return -1;
+    }
+    ret = configure_plugins_type(config, cf, FLB_CF_FILTER);
+    if (ret == -1) {
+        return -1;
+    }
+    ret = configure_plugins_type(config, cf, FLB_CF_OUTPUT);
+    if (ret == -1) {
+        return -1;
+    }
+
+    return 0;
+}