netdata/libnetdata/dyn_conf/dyn_conf.c

// SPDX-License-Identifier: GPL-3.0-or-later

#include "dyn_conf.h"

#define DYN_CONF_PATH_MAX (4096)
#define DYN_CONF_DIR VARLIB_DIR "/dynconf"

#define DYN_CONF_JOB_SCHEMA "job_schema"
#define DYN_CONF_SCHEMA "schema"
#define DYN_CONF_MODULE_LIST "modules"
#define DYN_CONF_JOB_LIST "jobs"
#define DYN_CONF_CFG_EXT ".cfg"

void job_flags_wallkthrough(dyncfg_job_flg_t flags, void (*cb)(const char *str, void *data), void *data)
{
    if (flags & JOB_FLG_PS_LOADED)
        cb("JOB_FLG_PS_LOADED", data);
    if (flags & JOB_FLG_PLUGIN_PUSHED)
        cb("JOB_FLG_PLUGIN_PUSHED", data);
    if (flags & JOB_FLG_STREAMING_PUSHED)
        cb("JOB_FLG_STREAMING_PUSHED", data);
    if (flags & JOB_FLG_USER_CREATED)
        cb("JOB_FLG_USER_CREATED", data);
}

struct deferred_cfg_send {
    DICTIONARY *plugins_dict;
    char *plugin_name;
    char *module_name;
    char *job_name;
    struct deferred_cfg_send *next;
};

bool dyncfg_shutdown = false;
struct deferred_cfg_send *deferred_configs = NULL;
pthread_mutex_t deferred_configs_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t deferred_configs_cond = PTHREAD_COND_INITIALIZER;

static void deferred_config_free(struct deferred_cfg_send *dcs)
{
    freez(dcs->plugin_name);
    freez(dcs->module_name);
    freez(dcs->job_name);
    freez(dcs);
}

static void deferred_config_push_back(DICTIONARY *plugins_dict, const char *plugin_name, const char *module_name, const char *job_name)
{
    struct deferred_cfg_send *deferred = callocz(1, sizeof(struct deferred_cfg_send));
    deferred->plugin_name = strdupz(plugin_name);
    if (module_name != NULL) {
        deferred->module_name = strdupz(module_name);
        if (job_name != NULL)
            deferred->job_name = strdupz(job_name);
    }
    deferred->plugins_dict = plugins_dict;
    pthread_mutex_lock(&deferred_configs_lock);
    if (dyncfg_shutdown) {
        pthread_mutex_unlock(&deferred_configs_lock);
        deferred_config_free(deferred);
        return;
    }
    struct deferred_cfg_send *last = deferred_configs;
    if (last == NULL)
        deferred_configs = deferred;
    else {
        while (last->next != NULL)
            last = last->next;
        last->next = deferred;
    }
    pthread_cond_signal(&deferred_configs_cond);
    pthread_mutex_unlock(&deferred_configs_lock);
}

static void deferred_configs_unlock()
{
    dyncfg_shutdown = true;
    // if we get cancelled in pthread_cond_wait
    // we will arrive at cancelled cleanup handler
    // with mutex locked we need to unlock it
    pthread_mutex_unlock(&deferred_configs_lock);
}

static struct deferred_cfg_send *deferred_config_pop(void *ptr)
{
    pthread_mutex_lock(&deferred_configs_lock);
    while (deferred_configs == NULL) {
        netdata_thread_cleanup_push(deferred_configs_unlock, ptr);
        pthread_cond_wait(&deferred_configs_cond, &deferred_configs_lock);
        netdata_thread_cleanup_pop(0);
    }
    struct deferred_cfg_send *deferred = deferred_configs;
    deferred_configs = deferred_configs->next;
    pthread_mutex_unlock(&deferred_configs_lock);
    return deferred;
}

static int _get_list_of_plugins_json_cb(const DICTIONARY_ITEM *item, void *entry, void *data)
{
    UNUSED(item);
    json_object *obj = (json_object *)data;
    struct configurable_plugin *plugin = (struct configurable_plugin *)entry;

    json_object *plugin_name = json_object_new_string(plugin->name);
    json_object_array_add(obj, plugin_name);

    return 0;
}

json_object *get_list_of_plugins_json(DICTIONARY *plugins_dict)
{
    json_object *obj = json_object_new_array();

    dictionary_walkthrough_read(plugins_dict, _get_list_of_plugins_json_cb, obj);

    return obj;
}

static int _get_list_of_modules_json_cb(const DICTIONARY_ITEM *item, void *entry, void *data)
{
    UNUSED(item);
    json_object *obj = (json_object *)data;
    struct module *module = (struct module *)entry;

    json_object *json_module = json_object_new_object();

    json_object *json_item = json_object_new_string(module->name);
    json_object_object_add(json_module, "name", json_item);
    const char *module_type = module_type2str(module->type);
    json_item = json_object_new_string(module_type);
    json_object_object_add(json_module, "type", json_item);

    json_object_array_add(obj, json_module);

    return 0;
}

json_object *get_list_of_modules_json(struct configurable_plugin *plugin)
{
    json_object *obj = json_object_new_array();

    pthread_mutex_lock(&plugin->lock);

    dictionary_walkthrough_read(plugin->modules, _get_list_of_modules_json_cb, obj);

    pthread_mutex_unlock(&plugin->lock);

    return obj;
}

const char *job_status2str(enum job_status status)
{
    switch (status) {
        case JOB_STATUS_UNKNOWN:
            return "unknown";
        case JOB_STATUS_STOPPED:
            return "stopped";
        case JOB_STATUS_RUNNING:
            return "running";
        case JOB_STATUS_ERROR:
            return "error";
        default:
            return "unknown";
    }
}

static void _job_flags2str_cb(const char *str, void *data)
{
    json_object *json_item = json_object_new_string(str);
    json_object_array_add((json_object *)data, json_item);
}

json_object *job2json(struct job *job) {
    json_object *json_job = json_object_new_object();

    json_object *json_item = json_object_new_string(job->name);
    json_object_object_add(json_job, "name", json_item);

    json_item = json_object_new_string(job_type2str(job->type));
    json_object_object_add(json_job, "type", json_item);

    netdata_mutex_lock(&job->lock);
    json_item = json_object_new_string(job_status2str(job->status));
    json_object_object_add(json_job, "status", json_item);

    json_item = json_object_new_int(job->state);
    json_object_object_add(json_job, "state", json_item);

    json_item = job->reason == NULL ? NULL : json_object_new_string(job->reason);
    json_object_object_add(json_job, "reason", json_item);

    int64_t last_state_update_s  = job->last_state_update / USEC_PER_SEC;
    int64_t last_state_update_us = job->last_state_update % USEC_PER_SEC;

    json_item = json_object_new_int64(last_state_update_s);
    json_object_object_add(json_job, "last_state_update_s", json_item);

    json_item = json_object_new_int64(last_state_update_us);
    json_object_object_add(json_job, "last_state_update_us", json_item);

    json_item = json_object_new_array();
    job_flags_wallkthrough(job->flags, _job_flags2str_cb, json_item);
    json_object_object_add(json_job, "flags", json_item);

    netdata_mutex_unlock(&job->lock);

    return json_job;
}

static int _get_list_of_jobs_json_cb(const DICTIONARY_ITEM *item, void *entry, void *data)
{
    UNUSED(item);
    json_object *obj = (json_object *)data;

    json_object *json_job = job2json((struct job *)entry);

    json_object_array_add(obj, json_job);

    return 0;
}

json_object *get_list_of_jobs_json(struct module *module)
{
    json_object *obj = json_object_new_array();

    pthread_mutex_lock(&module->lock);

    dictionary_walkthrough_read(module->jobs, _get_list_of_jobs_json_cb, obj);

    pthread_mutex_unlock(&module->lock);

    return obj;
}

struct job *get_job_by_name(struct module *module, const char *job_name)
{
    return dictionary_get(module->jobs, job_name);
}

void unlink_job(const char *plugin_name, const char *module_name, const char *job_name)
{
    // as we are going to do unlink here we better make sure we have all to build proper path
    if (unlikely(job_name == NULL || module_name == NULL || plugin_name == NULL))
        return;
    BUFFER *buffer = buffer_create(DYN_CONF_PATH_MAX, NULL);
    buffer_sprintf(buffer, DYN_CONF_DIR "/%s/%s/%s" DYN_CONF_CFG_EXT, plugin_name, module_name, job_name);
    if (unlink(buffer_tostring(buffer)))
        netdata_log_error("Cannot remove file %s", buffer_tostring(buffer));

    buffer_free(buffer);
}

void delete_job(struct configurable_plugin *plugin, const char *module_name, const char *job_name)
{
    struct module *module = get_module_by_name(plugin, module_name);
    if (module == NULL) {
        error_report("DYNCFG module \"%s\" not found", module_name);
        return;
    }

    struct job  *job_item = get_job_by_name(module, job_name);
    if (job_item == NULL) {
        error_report("DYNCFG job \"%s\" not found", job_name);
        return;
    }

    dictionary_del(module->jobs, job_name);
}

void delete_job_pname(DICTIONARY *plugins_dict, const char *plugin_name, const char *module_name, const char *job_name)
{
    const DICTIONARY_ITEM *plugin_item = dictionary_get_and_acquire_item(plugins_dict, plugin_name);
    if (plugin_item == NULL) {
        error_report("DYNCFG plugin \"%s\" not found", plugin_name);
        return;
    }
    struct configurable_plugin *plugin  = dictionary_acquired_item_value(plugin_item);

    delete_job(plugin, module_name, job_name);

    dictionary_acquired_item_release(plugins_dict, plugin_item);
}

int remove_job(struct module *module, struct job *job)
{
    enum set_config_result rc = module->delete_job_cb(module->job_config_cb_usr_ctx, module->plugin->name, module->name, job->name);

    if (rc != SET_CONFIG_ACCEPTED) {
        error_report("DYNCFG module \"%s\" rejected delete job for \"%s\"", module->name, job->name);
        return 0;
    }
    return 1;
}

struct module *get_module_by_name(struct configurable_plugin *plugin, const char *module_name)
{
    return dictionary_get(plugin->modules, module_name);
}

inline struct configurable_plugin *get_plugin_by_name(DICTIONARY *plugins_dict, const char *name)
{
    return dictionary_get(plugins_dict, name);
}

static int store_config(const char *module_name, const char *submodule_name, const char *cfg_idx, dyncfg_config_t cfg)
{
    BUFFER *filename = buffer_create(DYN_CONF_PATH_MAX, NULL);
    buffer_sprintf(filename, DYN_CONF_DIR "/%s", module_name);
    if (mkdir(buffer_tostring(filename), 0755) == -1) {
        if (errno != EEXIST) {
            netdata_log_error("DYNCFG store_config: failed to create module directory %s", buffer_tostring(filename));
            buffer_free(filename);
            return 1;
        }
    }

    if (submodule_name != NULL) {
        buffer_sprintf(filename, "/%s", submodule_name);
        if (mkdir(buffer_tostring(filename), 0755) == -1) {
            if (errno != EEXIST) {
                netdata_log_error("DYNCFG store_config: failed to create submodule directory %s", buffer_tostring(filename));
                buffer_free(filename);
                return 1;
            }
        }
    }

    if (cfg_idx != NULL)
        buffer_sprintf(filename, "/%s", cfg_idx);

    buffer_strcat(filename, DYN_CONF_CFG_EXT);


    error_report("DYNCFG store_config: %s", buffer_tostring(filename));

    //write to file
    FILE *f = fopen(buffer_tostring(filename), "w");
    if (f == NULL) {
        error_report("DYNCFG store_config: failed to open %s for writing", buffer_tostring(filename));
        buffer_free(filename);
        return 1;
    }

    fwrite(cfg.data, cfg.data_size, 1, f);
    fclose(f);

    buffer_free(filename);
    return 0;
}

#ifdef NETDATA_DEV_MODE
#define netdata_dev_fatal(...) fatal(__VA_ARGS__)
#else
#define netdata_dev_fatal(...) error_report(__VA_ARGS__)
#endif

void dyn_conf_store_config(const char *function, const char *payload, struct configurable_plugin *plugin) {
    dyncfg_config_t config = {
        .data = (char*)payload,
        .data_size = strlen(payload)
    };

    char *fnc = strdupz(function);
    // split fnc to words
    char *words[DYNCFG_MAX_WORDS];
    size_t words_c = quoted_strings_splitter(fnc, words, DYNCFG_MAX_WORDS, isspace_map_pluginsd);

    const char *fnc_name = get_word(words, words_c, 0);
    if (fnc_name == NULL) {
        error_report("Function name expected \"%s\"", function);
        goto CLEANUP;
    }
    if (strncmp(fnc_name, FUNCTION_NAME_SET_PLUGIN_CONFIG, strlen(FUNCTION_NAME_SET_PLUGIN_CONFIG)) == 0) {
        store_config(plugin->name, NULL, NULL, config);
        goto CLEANUP;
    }

    if (words_c < 2) {
        error_report("Module name expected \"%s\"", function);
        goto CLEANUP;
    }
    const char *module_name = get_word(words, words_c, 1);
    if (strncmp(fnc_name, FUNCTION_NAME_SET_MODULE_CONFIG, strlen(FUNCTION_NAME_SET_MODULE_CONFIG)) == 0) {
        store_config(plugin->name, module_name, NULL, config);
        goto CLEANUP;
    }

    if (words_c < 3) {
        error_report("Job name expected \"%s\"", function);
        goto CLEANUP;
    }
    const char *job_name = get_word(words, words_c, 2);
    if (strncmp(fnc_name, FUNCTION_NAME_SET_JOB_CONFIG, strlen(FUNCTION_NAME_SET_JOB_CONFIG)) == 0) {
        store_config(plugin->name, module_name, job_name, config);
        goto CLEANUP;
    }

    netdata_dev_fatal("Unknown function \"%s\"", function);

CLEANUP:
    freez(fnc);
}

dyncfg_config_t load_config(const char *plugin_name, const char *module_name, const char *job_id)
{
    BUFFER *filename = buffer_create(DYN_CONF_PATH_MAX, NULL);
    buffer_sprintf(filename, DYN_CONF_DIR "/%s", plugin_name);
    if (module_name != NULL)
        buffer_sprintf(filename, "/%s", module_name);

    if (job_id != NULL)
        buffer_sprintf(filename, "/%s", job_id);

    buffer_strcat(filename, DYN_CONF_CFG_EXT);

    dyncfg_config_t config;
    long bytes;
    config.data = read_by_filename(buffer_tostring(filename), &bytes);

    if (config.data == NULL)
        error_report("DYNCFG load_config: failed to load config from %s", buffer_tostring(filename));

    config.data_size = bytes;

    buffer_free(filename);

    return config;
}

char *set_plugin_config(struct configurable_plugin *plugin, dyncfg_config_t cfg)
{
    enum set_config_result rc = plugin->set_config_cb(plugin->cb_usr_ctx, plugin->name, &cfg);
    if (rc != SET_CONFIG_ACCEPTED) {
        error_report("DYNCFG plugin \"%s\" rejected config", plugin->name);
        return "plugin rejected config";
    }

    return NULL;
}

static char *set_module_config(struct module *mod, dyncfg_config_t cfg)
{
    struct configurable_plugin *plugin = mod->plugin;

    enum set_config_result rc = mod->set_config_cb(mod->config_cb_usr_ctx, plugin->name, mod->name, &cfg);
    if (rc != SET_CONFIG_ACCEPTED) {
        error_report("DYNCFG module \"%s\" rejected config", plugin->name);
        return "module rejected config";
    }

    return NULL;
}

struct job *job_new(const char *job_id)
{
    struct job *job = callocz(1, sizeof(struct job));
    job->state = JOB_STATUS_UNKNOWN;
    job->last_state_update = now_realtime_usec();
    job->name = strdupz(job_id);
    netdata_mutex_init(&job->lock);
    return job;
}

static inline void job_del(struct job *job)
{
    netdata_mutex_destroy(&job->lock);
    freez(job->reason);
    freez((void*)job->name);
    freez(job);
}

void job_del_cb(const DICTIONARY_ITEM *item, void *value, void *data)
{
    UNUSED(item);
    UNUSED(data);
    job_del((struct job *)value);
}

void module_del_cb(const DICTIONARY_ITEM *item, void *value, void *data)
{
    UNUSED(item);
    UNUSED(data);
    struct module *mod = (struct module *)value;
    dictionary_destroy(mod->jobs);
    freez(mod->name);
    freez(mod);
}

const DICTIONARY_ITEM *register_plugin(DICTIONARY *plugins_dict, struct configurable_plugin *plugin, bool localhost)
{
    if (get_plugin_by_name(plugins_dict, plugin->name) != NULL) {
        error_report("DYNCFG plugin \"%s\" already registered", plugin->name);
        return NULL;
    }

    if (plugin->set_config_cb == NULL) {
        error_report("DYNCFG plugin \"%s\" has no set_config_cb", plugin->name);
        return NULL;
    }

    pthread_mutex_init(&plugin->lock, NULL);

    plugin->modules = dictionary_create(DICT_OPTION_VALUE_LINK_DONT_CLONE);
    dictionary_register_delete_callback(plugin->modules, module_del_cb, NULL);

    if (localhost)
        deferred_config_push_back(plugins_dict, plugin->name, NULL, NULL);

    dictionary_set(plugins_dict, plugin->name, plugin, sizeof(plugin));

    // the plugin keeps the pointer to the dictionary item, so we need to acquire it
    return dictionary_get_and_acquire_item(plugins_dict, plugin->name);
}

void unregister_plugin(DICTIONARY *plugins_dict, const DICTIONARY_ITEM *plugin)
{
    struct configurable_plugin *plug = dictionary_acquired_item_value(plugin);
    dictionary_acquired_item_release(plugins_dict, plugin);
    dictionary_del(plugins_dict, plug->name);
}

int register_module(DICTIONARY *plugins_dict, struct configurable_plugin *plugin, struct module *module, bool localhost)
{
    if (get_module_by_name(plugin, module->name) != NULL) {
        error_report("DYNCFG module \"%s\" already registered", module->name);
        return 1;
    }

    pthread_mutex_init(&module->lock, NULL);

    if (localhost)
        deferred_config_push_back(plugins_dict, plugin->name, module->name, NULL);

    module->plugin = plugin;

    if (module->type == MOD_TYPE_ARRAY) {
        module->jobs = dictionary_create(DICT_OPTION_VALUE_LINK_DONT_CLONE);
        dictionary_register_delete_callback(module->jobs, job_del_cb, NULL);

        if (localhost) {
            // load all jobs from disk
            BUFFER *path = buffer_create(DYN_CONF_PATH_MAX, NULL);
            buffer_sprintf(path, "%s/%s/%s", DYN_CONF_DIR, plugin->name, module->name);
            DIR *dir = opendir(buffer_tostring(path));
            if (dir != NULL) {
                struct dirent *ent;
                while ((ent = readdir(dir)) != NULL) {
                    if (ent->d_name[0] == '.')
                        continue;
                    if (ent->d_type != DT_REG)
                        continue;
                    size_t len = strnlen(ent->d_name, NAME_MAX);
                    if (len <= strlen(DYN_CONF_CFG_EXT))
                        continue;
                    if (strcmp(ent->d_name + len - strlen(DYN_CONF_CFG_EXT), DYN_CONF_CFG_EXT) != 0)
                        continue;
                    ent->d_name[len - strlen(DYN_CONF_CFG_EXT)] = '\0';

                    struct job *job = job_new(ent->d_name);
                    job->module = module;
                    job->flags = JOB_FLG_PS_LOADED;
                    job->type = JOB_TYPE_USER;

                    dictionary_set(module->jobs, job->name, job, sizeof(job));

                    deferred_config_push_back(plugins_dict, plugin->name, module->name, ent->d_name);
                }
                closedir(dir);
            }
            buffer_free(path);
        }
    }

    dictionary_set(plugin->modules, module->name, module, sizeof(module));

    return 0;
}

int register_job(DICTIONARY *plugins_dict, const char *plugin_name, const char *module_name, const char *job_name, enum job_type job_type, dyncfg_job_flg_t flags, int ignore_existing)
{
    int rc = 1;
    const DICTIONARY_ITEM *plugin_item = dictionary_get_and_acquire_item(plugins_dict, plugin_name);
    if (plugin_item == NULL) {
        error_report("plugin \"%s\" not registered", plugin_name);
        return rc;
    }
    struct configurable_plugin *plugin = dictionary_acquired_item_value(plugin_item);
    struct module *mod = get_module_by_name(plugin, module_name);
    if (mod == NULL) {
        error_report("module \"%s\" not registered", module_name);
        goto ERR_EXIT;
    }
    if (mod->type != MOD_TYPE_ARRAY) {
        error_report("module \"%s\" is not an array", module_name);
        goto ERR_EXIT;
    }
    if (get_job_by_name(mod, job_name) != NULL) {
        if (!ignore_existing)
            error_report("job \"%s\" already registered", job_name);
        goto ERR_EXIT;
    }

    struct job *job = job_new(job_name);
    job->module = mod;
    job->flags = flags;
    job->type = job_type;

    dictionary_set(mod->jobs, job->name, job, sizeof(job));

    rc = 0;
ERR_EXIT:
    dictionary_acquired_item_release(plugins_dict, plugin_item);
    return rc;
}

void freez_dyncfg(void *ptr) {
    freez(ptr);
}

#ifdef NETDATA_TEST_DYNCFG
static void handle_dyncfg_root(DICTIONARY *plugins_dict, struct uni_http_response *resp, int method)
{
    if (method != HTTP_METHOD_GET) {
        resp->content = "method not allowed";
        resp->content_length = strlen(resp->content);
        resp->status = HTTP_RESP_METHOD_NOT_ALLOWED;
        return;
    }
    json_object *obj = get_list_of_plugins_json(plugins_dict);
    json_object *wrapper = json_object_new_object();
    json_object_object_add(wrapper, "configurable_plugins", obj);
    resp->content = strdupz(json_object_to_json_string_ext(wrapper, JSON_C_TO_STRING_PRETTY));
    json_object_put(wrapper);
    resp->status = HTTP_RESP_OK;
    resp->content_type = CT_APPLICATION_JSON;
    resp->content_free = freez_dyncfg;
    resp->content_length = strlen(resp->content);
}

static void handle_plugin_root(struct uni_http_response *resp, int method, struct configurable_plugin *plugin, void *post_payload, size_t post_payload_size)
{
    switch(method) {
        case HTTP_METHOD_GET:
        {
            dyncfg_config_t cfg = plugin->get_config_cb(plugin->cb_usr_ctx, plugin->name);
            resp->content = mallocz(cfg.data_size);
            memcpy(resp->content, cfg.data, cfg.data_size);
            resp->status = HTTP_RESP_OK;
            resp->content_free = freez_dyncfg;
            resp->content_length = cfg.data_size;
            return;
        }
        case HTTP_METHOD_PUT:
        {
            char *response;
            if (post_payload == NULL) {
                resp->content = "no payload";
                resp->content_length = strlen(resp->content);
                resp->status = HTTP_RESP_BAD_REQUEST;
                return;
            }
            dyncfg_config_t cont = {
                .data = post_payload,
                .data_size = post_payload_size
            };
            response = set_plugin_config(plugin, cont);
            if (response == NULL) {
                resp->status = HTTP_RESP_OK;
                resp->content = "OK";
                resp->content_length = strlen(resp->content);
            } else {
                resp->status = HTTP_RESP_BAD_REQUEST;
                resp->content = response;
                resp->content_length = strlen(resp->content);
            }
            return;
        }
        default:
            resp->content = "method not allowed";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_METHOD_NOT_ALLOWED;
            return;
    }
}
#endif

void handle_module_root(struct uni_http_response *resp, int method, struct configurable_plugin *plugin, const char *module, void *post_payload, size_t post_payload_size)
{
    if (strncmp(module, DYN_CONF_SCHEMA, sizeof(DYN_CONF_SCHEMA)) == 0) {
        dyncfg_config_t cfg = plugin->get_config_schema_cb(plugin->cb_usr_ctx, plugin->name);
        resp->content = mallocz(cfg.data_size);
        memcpy(resp->content, cfg.data, cfg.data_size);
        resp->status = HTTP_RESP_OK;
        resp->content_free = freez_dyncfg;
        resp->content_length = cfg.data_size;
        return;
    }
    if (strncmp(module, DYN_CONF_MODULE_LIST, sizeof(DYN_CONF_MODULE_LIST)) == 0) {
        if (method != HTTP_METHOD_GET) {
            resp->content = "method not allowed (only GET)";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_METHOD_NOT_ALLOWED;
            return;
        }
        json_object *obj = get_list_of_modules_json(plugin);
        json_object *wrapper = json_object_new_object();
        json_object_object_add(wrapper, "modules", obj);
        resp->content = strdupz(json_object_to_json_string_ext(wrapper, JSON_C_TO_STRING_PRETTY));
        json_object_put(wrapper);
        resp->status = HTTP_RESP_OK;
        resp->content_type = CT_APPLICATION_JSON;
        resp->content_free = freez_dyncfg;
        resp->content_length = strlen(resp->content);
        return;
    }
    struct module *mod = get_module_by_name(plugin, module);
    if (mod == NULL) {
        resp->content = "module not found";
        resp->content_length = strlen(resp->content);
        resp->status = HTTP_RESP_NOT_FOUND;
        return;
    }
    if (method == HTTP_METHOD_GET) {
        dyncfg_config_t cfg = mod->get_config_cb(mod->config_cb_usr_ctx, plugin->name, mod->name);
        resp->content = mallocz(cfg.data_size);
        memcpy(resp->content, cfg.data, cfg.data_size);
        resp->status = HTTP_RESP_OK;
        resp->content_free = freez_dyncfg;
        resp->content_length = cfg.data_size;
        return;
    } else if (method == HTTP_METHOD_PUT) {
        char *response;
        if (post_payload == NULL) {
            resp->content = "no payload";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_BAD_REQUEST;
            return;
        }
        dyncfg_config_t cont = {
            .data = post_payload,
            .data_size = post_payload_size
        };
        response = set_module_config(mod, cont);
        if (response == NULL) {
            resp->status = HTTP_RESP_OK;
            resp->content = "OK";
            resp->content_length = strlen(resp->content);
        } else {
            resp->status = HTTP_RESP_BAD_REQUEST;
            resp->content = response;
            resp->content_length = strlen(resp->content);
        }
        return;
    }
    resp->content = "method not allowed";
    resp->content_length = strlen(resp->content);
    resp->status = HTTP_RESP_METHOD_NOT_ALLOWED;
}

static inline void _handle_job_root(struct uni_http_response *resp, int method, struct module *mod, const char *job_id, void *post_payload, size_t post_payload_size, struct job *job)
{
    if (method == HTTP_METHOD_POST) {
        if (job != NULL) {
            resp->content = "can't POST, job already exists (use PUT to update?)";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_BAD_REQUEST;
            return;
        }
        if (post_payload == NULL) {
            resp->content = "no payload";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_BAD_REQUEST;
            return;
        }
        dyncfg_config_t cont = {
            .data = post_payload,
            .data_size = post_payload_size
        };
        if (mod->set_job_config_cb(mod->job_config_cb_usr_ctx, mod->plugin->name, mod->name, job_id, &cont)) {
            resp->content = "failed to add job";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_INTERNAL_SERVER_ERROR;
            return;
        }
        resp->status = HTTP_RESP_OK;
        resp->content = "OK";
        resp->content_length = strlen(resp->content);
        return;
    }
    if (job == NULL) {
        resp->content = "job not found";
        resp->content_length = strlen(resp->content);
        resp->status = HTTP_RESP_NOT_FOUND;
        return;
    }
    switch (method) {
        case HTTP_METHOD_GET:
        {
            dyncfg_config_t cfg = mod->get_job_config_cb(mod->job_config_cb_usr_ctx, mod->plugin->name, mod->name, job->name);
            resp->content = mallocz(cfg.data_size);
            memcpy(resp->content, cfg.data, cfg.data_size);
            resp->status = HTTP_RESP_OK;
            resp->content_free = freez_dyncfg;
            resp->content_length = cfg.data_size;
            return;
        }
        case HTTP_METHOD_PUT:
        {
            if (post_payload == NULL) {
                resp->content = "missing payload";
                resp->content_length = strlen(resp->content);
                resp->status = HTTP_RESP_BAD_REQUEST;
                return;
            }
            dyncfg_config_t cont = {
                .data = post_payload,
                .data_size = post_payload_size
            };
            if (mod->set_job_config_cb(mod->job_config_cb_usr_ctx, mod->plugin->name, mod->name, job->name, &cont) != SET_CONFIG_ACCEPTED) {
                error_report("DYNCFG module \"%s\" rejected config for job \"%s\"", mod->name, job->name);
                resp->content = "failed to set job config";
                resp->content_length = strlen(resp->content);
                resp->status = HTTP_RESP_INTERNAL_SERVER_ERROR;
                return;
            }
            resp->status = HTTP_RESP_OK;
            resp->content = "OK";
            resp->content_length = strlen(resp->content);
            return;
        }
        case HTTP_METHOD_DELETE:
        {
            if (!remove_job(mod, job)) {
                resp->content = "failed to remove job";
                resp->content_length = strlen(resp->content);
                resp->status = HTTP_RESP_INTERNAL_SERVER_ERROR;
                return;
            }
            resp->status = HTTP_RESP_OK;
            resp->content = "OK";
            resp->content_length = strlen(resp->content);
            return;
        }
        default:
            resp->content = "method not allowed (only GET, PUT, DELETE)";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_METHOD_NOT_ALLOWED;
            return;
    }
}

void handle_job_root(struct uni_http_response *resp, int method, struct module *mod, const char *job_id, void *post_payload, size_t post_payload_size)
{
    if (strncmp(job_id, DYN_CONF_SCHEMA, sizeof(DYN_CONF_SCHEMA)) == 0) {
        dyncfg_config_t cfg = mod->get_config_schema_cb(mod->config_cb_usr_ctx, mod->plugin->name, mod->name);
        resp->content = mallocz(cfg.data_size);
        memcpy(resp->content, cfg.data, cfg.data_size);
        resp->status = HTTP_RESP_OK;
        resp->content_free = freez_dyncfg;
        resp->content_length = cfg.data_size;
        return;
    }
    if (strncmp(job_id, DYN_CONF_JOB_SCHEMA, sizeof(DYN_CONF_JOB_SCHEMA)) == 0) {
        dyncfg_config_t cfg = mod->get_job_config_schema_cb(mod->job_config_cb_usr_ctx, mod->plugin->name, mod->name);
        resp->content = mallocz(cfg.data_size);
        memcpy(resp->content, cfg.data, cfg.data_size);
        resp->status = HTTP_RESP_OK;
        resp->content_free = freez_dyncfg;
        resp->content_length = cfg.data_size;
        return;
    }
    if (strncmp(job_id, DYN_CONF_JOB_LIST, sizeof(DYN_CONF_JOB_LIST)) == 0) {
        if (mod->type != MOD_TYPE_ARRAY) {
            resp->content = "module type is not job_array (can't get the list of jobs)";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_NOT_FOUND;
            return;
        }
        if (method != HTTP_METHOD_GET) {
            resp->content = "method not allowed (only GET)";
            resp->content_length = strlen(resp->content);
            resp->status = HTTP_RESP_METHOD_NOT_ALLOWED;
            return;
        }
        json_object *obj = get_list_of_jobs_json(mod);
        json_object *wrapper = json_object_new_object();
        json_object_object_add(wrapper, "jobs", obj);
        resp->content = strdupz(json_object_to_json_string_ext(wrapper, JSON_C_TO_STRING_PRETTY));
        json_object_put(wrapper);
        resp->status = HTTP_RESP_OK;
        resp->content_type = CT_APPLICATION_JSON;
        resp->content_free = freez_dyncfg;
        resp->content_length = strlen(resp->content);
        return;
    }
    const DICTIONARY_ITEM *job_item = dictionary_get_and_acquire_item(mod->jobs, job_id);
    struct job *job = dictionary_acquired_item_value(job_item);

    _handle_job_root(resp, method, mod, job_id, post_payload, post_payload_size, job);

    dictionary_acquired_item_release(mod->jobs, job_item);
}

struct uni_http_response dyn_conf_process_http_request(
    DICTIONARY *plugins_dict __maybe_unused,
    int method __maybe_unused,
    const char *plugin __maybe_unused,
    const char *module __maybe_unused,
    const char *job_id __maybe_unused,
    void *post_payload __maybe_unused,
    size_t post_payload_size __maybe_unused)
{
    struct uni_http_response resp = {
        .status = HTTP_RESP_INTERNAL_SERVER_ERROR,
        .content_type = CT_TEXT_PLAIN,
        .content = HTTP_RESP_INTERNAL_SERVER_ERROR_STR,
        .content_free = NULL,
        .content_length = 0
    };
#ifndef NETDATA_TEST_DYNCFG
    resp.content = "DYNCFG is disabled (as it is for now developer only feature). This will be enabled by default when ready for technical preview.";
    resp.content_length = strlen(resp.content);
    resp.status = HTTP_RESP_PRECOND_FAIL;
    return resp;
#else
    if (plugin == NULL) {
        handle_dyncfg_root(plugins_dict, &resp, method);
        return resp;
    }
    const DICTIONARY_ITEM *plugin_item = dictionary_get_and_acquire_item(plugins_dict, plugin);
    if (plugin_item == NULL) {
        resp.content = "plugin not found";
        resp.content_length = strlen(resp.content);
        resp.status = HTTP_RESP_NOT_FOUND;
        return resp;
    }
    struct configurable_plugin *plug = dictionary_acquired_item_value(plugin_item);
    if (module == NULL) {
        handle_plugin_root(&resp, method, plug, post_payload, post_payload_size);
        goto EXIT_PLUGIN;
    }
    if (job_id == NULL) {
        handle_module_root(&resp, method, plug, module, post_payload, post_payload_size);
        goto EXIT_PLUGIN;
    }
    // for modules we do not do get_and_acquire as modules are never removed (only together with the plugin)
    struct module *mod = get_module_by_name(plug, module);
    if (mod == NULL) {
        resp.content = "module not found";
        resp.content_length = strlen(resp.content);
        resp.status = HTTP_RESP_NOT_FOUND;
        goto EXIT_PLUGIN;
    }
    if (mod->type != MOD_TYPE_ARRAY) {
        resp.content = "400 - this module is not array type";
        resp.content_length = strlen(resp.content);
        resp.status = HTTP_RESP_BAD_REQUEST;
        goto EXIT_PLUGIN;
    }
    handle_job_root(&resp, method, mod, job_id, post_payload, post_payload_size);

EXIT_PLUGIN:
    dictionary_acquired_item_release(plugins_dict, plugin_item);
    return resp;
#endif
}

void plugin_del_cb(const DICTIONARY_ITEM *item, void *value, void *data)
{
    UNUSED(item);
    UNUSED(data);
    struct configurable_plugin *plugin = (struct configurable_plugin *)value;
    dictionary_destroy(plugin->modules);
    freez(plugin->name);
    freez(plugin);
}

// on failure - return NULL - all unlocked, nothing acquired
// on success - return pointer to job item - keep job and plugin acquired and locked!!!
//              for caller convenience (to prevent another lock and races)
//              caller is responsible to unlock the job and release it when not needed anymore
//              this also avoids dependency creep
const DICTIONARY_ITEM *report_job_status_acq_lock(DICTIONARY *plugins_dict, const DICTIONARY_ITEM **plugin_acq_item, DICTIONARY **job_dict, const char *plugin_name, const char *module_name, const char *job_name, enum job_status status, int status_code, char *reason)
{
    *plugin_acq_item = dictionary_get_and_acquire_item(plugins_dict, plugin_name);
    if (*plugin_acq_item == NULL) {
        netdata_log_error("plugin %s not found", plugin_name);
        return NULL;
    }

    struct configurable_plugin *plug = dictionary_acquired_item_value(*plugin_acq_item);
    struct module *mod = get_module_by_name(plug, module_name);
    if (mod == NULL) {
        netdata_log_error("module %s not found", module_name);
        dictionary_acquired_item_release(plugins_dict, *plugin_acq_item);
        return NULL;
    }
    if (mod->type != MOD_TYPE_ARRAY) {
        netdata_log_error("module %s is not array", module_name);
        dictionary_acquired_item_release(plugins_dict, *plugin_acq_item);
        return NULL;
    }
    *job_dict = mod->jobs;
    const DICTIONARY_ITEM *job_item = dictionary_get_and_acquire_item(mod->jobs, job_name);
    if (job_item == NULL) {
        netdata_log_error("job %s not found", job_name);
        dictionary_acquired_item_release(plugins_dict, *plugin_acq_item);
        return NULL;
    }
    struct job *job = dictionary_acquired_item_value(job_item);

    pthread_mutex_lock(&job->lock);
    job->status = status;
    job->state = status_code;
    if (job->reason != NULL) {
        freez(job->reason);
    }
    job->reason = reason != NULL ? strdupz(reason) : NULL; // reason is optional
    job->last_state_update = now_realtime_usec();

    job->dirty = true;

    // no unlock and acquired_item_release on success on purpose
    return job_item;
}

int dyn_conf_init(void)
{
    if (mkdir(DYN_CONF_DIR, 0755) == -1) {
        if (errno != EEXIST) {
            netdata_log_error("failed to create directory for dynamic configuration");
            return 1;
        }
    }

    return 0;
}

static void dyncfg_cleanup(void *ptr) {
    struct netdata_static_thread *static_thread = (struct netdata_static_thread *) ptr;
    static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

    netdata_log_info("cleaning up...");

    pthread_mutex_lock(&deferred_configs_lock);
    dyncfg_shutdown = true;
    while (deferred_configs != NULL) {
        struct deferred_cfg_send *dcs = deferred_configs;
        deferred_configs = dcs->next;
        deferred_config_free(dcs);
    }
    pthread_mutex_unlock(&deferred_configs_lock);

    static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
}

void *dyncfg_main(void *ptr)
{
    netdata_thread_cleanup_push(dyncfg_cleanup, ptr);

    while (!netdata_exit) {
        struct deferred_cfg_send *dcs = deferred_config_pop(ptr);
        DICTIONARY *plugins_dict = dcs->plugins_dict;
#ifdef NETDATA_INTERNAL_CHECKS
        if (plugins_dict == NULL) {
            fatal("DYNCFG, plugins_dict is NULL");
            deferred_config_free(dcs);
            continue;
        }
#endif

        const DICTIONARY_ITEM *plugin_item = dictionary_get_and_acquire_item(plugins_dict, dcs->plugin_name);
        if (plugin_item == NULL) {
            error_report("DYNCFG, plugin %s not found", dcs->plugin_name);
            deferred_config_free(dcs);
            continue;
        }
        struct configurable_plugin *plugin = dictionary_acquired_item_value(plugin_item);
        if (dcs->module_name == NULL) {
            dyncfg_config_t cfg = load_config(dcs->plugin_name, NULL, NULL);
            if (cfg.data != NULL) {
                plugin->set_config_cb(plugin->cb_usr_ctx, plugin->name, &cfg);
                freez(cfg.data);
            }
        } else if (dcs->job_name == NULL) {
            dyncfg_config_t cfg = load_config(dcs->plugin_name, dcs->module_name, NULL);
            if (cfg.data != NULL) {
                struct module *mod = get_module_by_name(plugin, dcs->module_name);
                mod->set_config_cb(mod->config_cb_usr_ctx, plugin->name, mod->name, &cfg);
                freez(cfg.data);
            }
        } else {
            dyncfg_config_t cfg = load_config(dcs->plugin_name, dcs->module_name, dcs->job_name);
            if (cfg.data != NULL) {
                struct module *mod = get_module_by_name(plugin, dcs->module_name);
                mod->set_job_config_cb(mod->job_config_cb_usr_ctx, plugin->name, mod->name, dcs->job_name, &cfg);
                freez(cfg.data);
            }
        }
        deferred_config_free(dcs);
        dictionary_acquired_item_release(plugins_dict, plugin_item);
    }

    netdata_thread_cleanup_pop(1);
    return NULL;
}

bool is_dyncfg_function(const char *function_name, uint8_t type) {
    // TODO add hash to speed things up
    if (type & (DYNCFG_FUNCTION_TYPE_GET | DYNCFG_FUNCTION_TYPE_REGULAR)) {
        if (strncmp(function_name, FUNCTION_NAME_GET_PLUGIN_CONFIG, strlen(FUNCTION_NAME_GET_PLUGIN_CONFIG)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_GET_PLUGIN_CONFIG_SCHEMA, strlen(FUNCTION_NAME_GET_PLUGIN_CONFIG_SCHEMA)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_GET_MODULE_CONFIG, strlen(FUNCTION_NAME_GET_MODULE_CONFIG)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_GET_MODULE_CONFIG_SCHEMA, strlen(FUNCTION_NAME_GET_MODULE_CONFIG_SCHEMA)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_GET_JOB_CONFIG, strlen(FUNCTION_NAME_GET_JOB_CONFIG)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_GET_JOB_CONFIG_SCHEMA, strlen(FUNCTION_NAME_GET_JOB_CONFIG_SCHEMA)) == 0)
            return true;
    }

    if (type & (DYNCFG_FUNCTION_TYPE_SET | DYNCFG_FUNCTION_TYPE_PAYLOAD)) {
        if (strncmp(function_name, FUNCTION_NAME_SET_PLUGIN_CONFIG, strlen(FUNCTION_NAME_SET_PLUGIN_CONFIG)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_SET_MODULE_CONFIG, strlen(FUNCTION_NAME_SET_MODULE_CONFIG)) == 0)
            return true;
        if (strncmp(function_name, FUNCTION_NAME_SET_JOB_CONFIG, strlen(FUNCTION_NAME_SET_JOB_CONFIG)) == 0)
            return true;
    }

    if (type & (DYNCFG_FUNCTION_TYPE_DELETE | DYNCFG_FUNCTION_TYPE_REGULAR)) {
        if (strncmp(function_name, FUNCTION_NAME_DELETE_JOB, strlen(FUNCTION_NAME_DELETE_JOB)) == 0)
            return true;
    }

    return false;
}