#include "collectd.h"
-#include "common.h"
#include "configfile.h"
#include "filter_chain.h"
#include "plugin.h"
-#include "utils_avltree.h"
+#include "utils/avltree/avltree.h"
+#include "utils/common/common.h"
+#include "utils/heap/heap.h"
#include "utils_cache.h"
#include "utils_complain.h"
-#include "utils_heap.h"
#include "utils_llist.h"
#include "utils_random.h"
#include "utils_time.h"
+#ifdef WIN32
+#define EXPORT __declspec(dllexport)
+#include <sys/stat.h>
+#include <unistd.h>
+#else
+#define EXPORT
+#endif
+
#if HAVE_PTHREAD_NP_H
#include <pthread_np.h> /* for pthread_set_name_np(3) */
#endif
};
typedef struct read_func_s read_func_t;
+struct cache_event_func_s {
+ plugin_cache_event_cb callback;
+ char *name;
+ user_data_t user_data;
+ plugin_ctx_t plugin_ctx;
+};
+typedef struct cache_event_func_s cache_event_func_t;
+
struct write_queue_s;
typedef struct write_queue_s write_queue_t;
struct write_queue_s {
static llist_t *list_log;
static llist_t *list_notification;
+static size_t list_cache_event_num;
+static cache_event_func_t list_cache_event[32];
+
static fc_chain_t *pre_cache_chain;
static fc_chain_t *post_cache_chain;
static long write_limit_high;
static long write_limit_low;
+static pthread_mutex_t statistics_lock = PTHREAD_MUTEX_INITIALIZER;
static derive_t stats_values_dropped;
static bool record_statistics;
static int register_callback(llist_t **list, /* {{{ */
const char *name, callback_func_t *cf) {
- llentry_t *le;
- char *key;
if (*list == NULL) {
*list = llist_create();
}
}
- key = strdup(name);
+ char *key = strdup(name);
if (key == NULL) {
ERROR("plugin: register_callback: strdup failed.");
destroy_callback(cf);
return -1;
}
- le = llist_search(*list, name);
+ llentry_t *le = llist_search(*list, name);
if (le == NULL) {
le = llentry_create(key, cf);
if (le == NULL) {
llist_append(*list, le);
} else {
- callback_func_t *old_cf;
-
- old_cf = le->value;
+ callback_func_t *old_cf = le->value;
le->value = cf;
P_WARNING("register_callback: "
int i;
llentry_t *le;
int n;
- char **keys;
n = llist_size(*list);
if (n == 0) {
return;
}
- keys = calloc(n, sizeof(char *));
-
+ char **keys = calloc(n, sizeof(*keys));
if (keys == NULL) {
ERROR("%s: failed to allocate memory for list of callbacks", comment);
-
return;
}
cf->cf_callback = callback;
if (ud == NULL) {
cf->cf_udata = (user_data_t){
- .data = NULL, .free_func = NULL,
+ .data = NULL,
+ .free_func = NULL,
};
} else {
cf->cf_udata = *ud;
if (read_threads != NULL)
return;
- read_threads = (pthread_t *)calloc(num, sizeof(pthread_t));
+ read_threads = calloc(num, sizeof(*read_threads));
if (read_threads == NULL) {
ERROR("plugin: start_read_threads: calloc failed.");
return;
}
char name[THREAD_NAME_MAX];
- snprintf(name, sizeof(name), "reader#%" PRIsz, read_threads_num);
+ ssnprintf(name, sizeof(name), "reader#%" PRIu64,
+ (uint64_t)read_threads_num);
set_thread_name(read_threads[read_threads_num], name);
read_threads_num++;
if (write_threads != NULL)
return;
- write_threads = (pthread_t *)calloc(num, sizeof(pthread_t));
+ write_threads = calloc(num, sizeof(*write_threads));
if (write_threads == NULL) {
ERROR("plugin: start_write_threads: calloc failed.");
return;
}
char name[THREAD_NAME_MAX];
- snprintf(name, sizeof(name), "writer#%" PRIsz, write_threads_num);
+ ssnprintf(name, sizeof(name), "writer#%" PRIu64,
+ (uint64_t)write_threads_num);
set_thread_name(write_threads[write_threads_num], name);
write_threads_num++;
ERROR("plugin_set_dir: strdup(\"%s\") failed", dir);
}
-static bool plugin_is_loaded(char const *name) {
- int status;
-
+bool plugin_is_loaded(char const *name) {
if (plugins_loaded == NULL)
plugins_loaded =
c_avl_create((int (*)(const void *, const void *))strcasecmp);
assert(plugins_loaded != NULL);
- status = c_avl_get(plugins_loaded, name, /* ret_value = */ NULL);
+ int status = c_avl_get(plugins_loaded, name, /* ret_value = */ NULL);
return status == 0;
}
}
#define BUFSIZE 512
+#ifdef WIN32
+#define SHLIB_SUFFIX ".dll"
+#else
+#define SHLIB_SUFFIX ".so"
+#endif
int plugin_load(char const *plugin_name, bool global) {
DIR *dh;
const char *dir;
(strcasecmp("python", plugin_name) == 0))
global = true;
- /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
+ /* `cpu' should not match `cpufreq'. To solve this we add SHLIB_SUFFIX to the
* type when matching the filename */
- status = snprintf(typename, sizeof(typename), "%s.so", plugin_name);
+ status = snprintf(typename, sizeof(typename), "%s" SHLIB_SUFFIX, plugin_name);
if ((status < 0) || ((size_t)status >= sizeof(typename))) {
- WARNING("plugin_load: Filename too long: \"%s.so\"", plugin_name);
+ WARNING("plugin_load: Filename too long: \"%s" SHLIB_SUFFIX "\"",
+ plugin_name);
return -1;
}
/*
* The `register_*' functions follow
*/
-int plugin_register_config(const char *name,
- int (*callback)(const char *key, const char *val),
- const char **keys, int keys_num) {
+EXPORT int plugin_register_config(const char *name,
+ int (*callback)(const char *key,
+ const char *val),
+ const char **keys, int keys_num) {
cf_register(name, callback, keys, keys_num);
return 0;
} /* int plugin_register_config */
-int plugin_register_complex_config(const char *type,
- int (*callback)(oconfig_item_t *)) {
+EXPORT int plugin_register_complex_config(const char *type,
+ int (*callback)(oconfig_item_t *)) {
return cf_register_complex(type, callback);
} /* int plugin_register_complex_config */
-int plugin_register_init(const char *name, int (*callback)(void)) {
+EXPORT int plugin_register_init(const char *name, int (*callback)(void)) {
return create_register_callback(&list_init, name, (void *)callback, NULL);
} /* plugin_register_init */
return 0;
} /* int plugin_insert_read */
-int plugin_register_read(const char *name, int (*callback)(void)) {
+EXPORT int plugin_register_read(const char *name, int (*callback)(void)) {
read_func_t *rf;
int status;
return status;
} /* int plugin_register_read */
-int plugin_register_complex_read(const char *group, const char *name,
- plugin_read_cb callback, cdtime_t interval,
- user_data_t const *user_data) {
+EXPORT int plugin_register_complex_read(const char *group, const char *name,
+ plugin_read_cb callback,
+ cdtime_t interval,
+ user_data_t const *user_data) {
read_func_t *rf;
int status;
return status;
} /* int plugin_register_complex_read */
-int plugin_register_write(const char *name, plugin_write_cb callback,
- user_data_t const *ud) {
+EXPORT int plugin_register_write(const char *name, plugin_write_cb callback,
+ user_data_t const *ud) {
return create_register_callback(&list_write, name, (void *)callback, ud);
} /* int plugin_register_write */
return flush_name;
} /* static char *plugin_flush_callback_name */
-int plugin_register_flush(const char *name, plugin_flush_cb callback,
- user_data_t const *ud) {
+EXPORT int plugin_register_flush(const char *name, plugin_flush_cb callback,
+ user_data_t const *ud) {
int status;
plugin_ctx_t ctx = plugin_get_ctx();
/* name = */ flush_name,
/* callback = */ plugin_flush_timeout_callback,
/* interval = */ ctx.flush_interval,
- /* user data = */ &(user_data_t){
- .data = cb, .free_func = plugin_flush_timeout_callback_free,
+ /* user data = */
+ &(user_data_t){
+ .data = cb,
+ .free_func = plugin_flush_timeout_callback_free,
});
sfree(flush_name);
return 0;
} /* int plugin_register_flush */
-int plugin_register_missing(const char *name, plugin_missing_cb callback,
- user_data_t const *ud) {
+EXPORT int plugin_register_missing(const char *name, plugin_missing_cb callback,
+ user_data_t const *ud) {
return create_register_callback(&list_missing, name, (void *)callback, ud);
} /* int plugin_register_missing */
-int plugin_register_shutdown(const char *name, int (*callback)(void)) {
+EXPORT int plugin_register_cache_event(const char *name,
+ plugin_cache_event_cb callback,
+ user_data_t const *ud) {
+
+ if (name == NULL || callback == NULL)
+ return EINVAL;
+
+ char *name_copy = strdup(name);
+ if (name_copy == NULL) {
+ P_ERROR("plugin_register_cache_event: strdup failed.");
+ free_userdata(ud);
+ return ENOMEM;
+ }
+
+ if (list_cache_event_num >= 32) {
+ P_ERROR("plugin_register_cache_event: Too much cache event callbacks tried "
+ "to be registered.");
+ free_userdata(ud);
+ return ENOMEM;
+ }
+
+ for (size_t i = 0; i < list_cache_event_num; i++) {
+ cache_event_func_t *cef = &list_cache_event[i];
+ if (!cef->callback)
+ continue;
+
+ if (strcmp(name, cef->name) == 0) {
+ P_ERROR("plugin_register_cache_event: a callback named `%s' already "
+ "registered!",
+ name);
+ free_userdata(ud);
+ return -1;
+ }
+ }
+
+ user_data_t user_data;
+ if (ud == NULL) {
+ user_data = (user_data_t){
+ .data = NULL, .free_func = NULL,
+ };
+ } else {
+ user_data = *ud;
+ }
+
+ list_cache_event[list_cache_event_num] =
+ (cache_event_func_t){.callback = callback,
+ .name = name_copy,
+ .user_data = user_data,
+ .plugin_ctx = plugin_get_ctx()};
+ list_cache_event_num++;
+
+ return 0;
+} /* int plugin_register_cache_event */
+
+EXPORT int plugin_register_shutdown(const char *name, int (*callback)(void)) {
return create_register_callback(&list_shutdown, name, (void *)callback, NULL);
} /* int plugin_register_shutdown */
data_sets = NULL;
} /* void plugin_free_data_sets */
-int plugin_register_data_set(const data_set_t *ds) {
+EXPORT int plugin_register_data_set(const data_set_t *ds) {
data_set_t *ds_copy;
if ((data_sets != NULL) && (c_avl_get(data_sets, ds->type, NULL) == 0)) {
return c_avl_insert(data_sets, (void *)ds_copy->type, (void *)ds_copy);
} /* int plugin_register_data_set */
-int plugin_register_log(const char *name, plugin_log_cb callback,
- user_data_t const *ud) {
+EXPORT int plugin_register_log(const char *name, plugin_log_cb callback,
+ user_data_t const *ud) {
return create_register_callback(&list_log, name, (void *)callback, ud);
} /* int plugin_register_log */
-int plugin_register_notification(const char *name,
- plugin_notification_cb callback,
- user_data_t const *ud) {
+EXPORT int plugin_register_notification(const char *name,
+ plugin_notification_cb callback,
+ user_data_t const *ud) {
return create_register_callback(&list_notification, name, (void *)callback,
ud);
} /* int plugin_register_log */
-int plugin_unregister_config(const char *name) {
+EXPORT int plugin_unregister_config(const char *name) {
cf_unregister(name);
return 0;
} /* int plugin_unregister_config */
-int plugin_unregister_complex_config(const char *name) {
+EXPORT int plugin_unregister_complex_config(const char *name) {
cf_unregister_complex(name);
return 0;
} /* int plugin_unregister_complex_config */
-int plugin_unregister_init(const char *name) {
+EXPORT int plugin_unregister_init(const char *name) {
return plugin_unregister(list_init, name);
}
-int plugin_unregister_read(const char *name) /* {{{ */
+EXPORT int plugin_unregister_read(const char *name) /* {{{ */
{
llentry_t *le;
read_func_t *rf;
return 0;
} /* }}} int plugin_unregister_read */
-void plugin_log_available_writers(void) {
+EXPORT void plugin_log_available_writers(void) {
log_list_callbacks(&list_write, "Available write targets:");
}
return strcmp(rf->rf_group, (const char *)group);
} /* }}} int compare_read_func_group */
-int plugin_unregister_read_group(const char *group) /* {{{ */
+EXPORT int plugin_unregister_read_group(const char *group) /* {{{ */
{
llentry_t *le;
read_func_t *rf;
return 0;
} /* }}} int plugin_unregister_read_group */
-int plugin_unregister_write(const char *name) {
+EXPORT int plugin_unregister_write(const char *name) {
return plugin_unregister(list_write, name);
}
-int plugin_unregister_flush(const char *name) {
+EXPORT int plugin_unregister_flush(const char *name) {
plugin_ctx_t ctx = plugin_get_ctx();
if (ctx.flush_interval != 0) {
return plugin_unregister(list_flush, name);
}
-int plugin_unregister_missing(const char *name) {
+EXPORT int plugin_unregister_missing(const char *name) {
return plugin_unregister(list_missing, name);
}
-int plugin_unregister_shutdown(const char *name) {
+EXPORT int plugin_unregister_cache_event(const char *name) {
+ for (size_t i = 0; i < list_cache_event_num; i++) {
+ cache_event_func_t *cef = &list_cache_event[i];
+ if (!cef->callback)
+ continue;
+ if (strcmp(name, cef->name) == 0) {
+ /* Mark callback as inactive, so mask in cache entries remains actual */
+ cef->callback = NULL;
+ sfree(cef->name);
+ free_userdata(&cef->user_data);
+ }
+ }
+ return 0;
+}
+
+static void destroy_cache_event_callbacks() {
+ for (size_t i = 0; i < list_cache_event_num; i++) {
+ cache_event_func_t *cef = &list_cache_event[i];
+ if (!cef->callback)
+ continue;
+ cef->callback = NULL;
+ sfree(cef->name);
+ free_userdata(&cef->user_data);
+ }
+}
+
+EXPORT int plugin_unregister_shutdown(const char *name) {
return plugin_unregister(list_shutdown, name);
}
-int plugin_unregister_data_set(const char *name) {
+EXPORT int plugin_unregister_data_set(const char *name) {
data_set_t *ds;
if (data_sets == NULL)
return 0;
} /* int plugin_unregister_data_set */
-int plugin_unregister_log(const char *name) {
+EXPORT int plugin_unregister_log(const char *name) {
return plugin_unregister(list_log, name);
}
-int plugin_unregister_notification(const char *name) {
+EXPORT int plugin_unregister_notification(const char *name) {
return plugin_unregister(list_notification, name);
}
-int plugin_init_all(void) {
+EXPORT int plugin_init_all(void) {
char const *chain_name;
llentry_t *le;
int status;
} /* void plugin_init_all */
/* TODO: Rename this function. */
-void plugin_read_all(void) {
+EXPORT void plugin_read_all(void) {
uc_check_timeout();
return;
} /* void plugin_read_all */
/* Read function called when the `-T' command line argument is given. */
-int plugin_read_all_once(void) {
+EXPORT int plugin_read_all_once(void) {
int status;
int return_status = 0;
return return_status;
} /* int plugin_read_all_once */
-int plugin_write(const char *plugin, /* {{{ */
- const data_set_t *ds, const value_list_t *vl) {
+EXPORT int plugin_write(const char *plugin, /* {{{ */
+ const data_set_t *ds, const value_list_t *vl) {
llentry_t *le;
int status;
return status;
} /* }}} int plugin_write */
-int plugin_flush(const char *plugin, cdtime_t timeout, const char *identifier) {
+EXPORT int plugin_flush(const char *plugin, cdtime_t timeout,
+ const char *identifier) {
llentry_t *le;
if (list_flush == NULL)
return 0;
} /* int plugin_flush */
-int plugin_shutdown_all(void) {
+EXPORT int plugin_shutdown_all(void) {
llentry_t *le;
int ret = 0; // Assume success.
* the data isn't freed twice. */
destroy_all_callbacks(&list_flush);
destroy_all_callbacks(&list_missing);
+ destroy_cache_event_callbacks();
destroy_all_callbacks(&list_write);
destroy_all_callbacks(&list_notification);
return ret;
} /* void plugin_shutdown_all */
-int plugin_dispatch_missing(const value_list_t *vl) /* {{{ */
+EXPORT int plugin_dispatch_missing(const value_list_t *vl) /* {{{ */
{
if (list_missing == NULL)
return 0;
return 0;
} /* int }}} plugin_dispatch_missing */
+void plugin_dispatch_cache_event(enum cache_event_type_e event_type,
+ unsigned long callbacks_mask, const char *name,
+ const value_list_t *vl) {
+ switch (event_type) {
+ case CE_VALUE_NEW:
+ callbacks_mask = 0;
+ for (size_t i = 0; i < list_cache_event_num; i++) {
+ cache_event_func_t *cef = &list_cache_event[i];
+ plugin_cache_event_cb callback = cef->callback;
+
+ if (!callback)
+ continue;
+
+ cache_event_t event = (cache_event_t){.type = event_type,
+ .value_list = vl,
+ .value_list_name = name,
+ .ret = 0};
+
+ plugin_ctx_t old_ctx = plugin_set_ctx(cef->plugin_ctx);
+ int status = (*callback)(&event, &cef->user_data);
+ plugin_set_ctx(old_ctx);
+
+ if (status != 0) {
+ ERROR("plugin_dispatch_cache_event: Callback \"%s\" failed with status "
+ "%i for event NEW.",
+ cef->name, status);
+ } else {
+ if (event.ret) {
+ DEBUG(
+ "plugin_dispatch_cache_event: Callback \"%s\" subscribed to %s.",
+ cef->name, name);
+ callbacks_mask |= (1 << (i));
+ } else {
+ DEBUG("plugin_dispatch_cache_event: Callback \"%s\" ignores %s.",
+ cef->name, name);
+ }
+ }
+ }
+
+ if (callbacks_mask)
+ uc_set_callbacks_mask(name, callbacks_mask);
+
+ break;
+ case CE_VALUE_UPDATE:
+ case CE_VALUE_EXPIRED:
+ for (size_t i = 0; i < list_cache_event_num; i++) {
+ cache_event_func_t *cef = &list_cache_event[i];
+ plugin_cache_event_cb callback = cef->callback;
+
+ if (!callback)
+ continue;
+
+ if (callbacks_mask && (1 << (i)) == 0)
+ continue;
+
+ cache_event_t event = (cache_event_t){.type = event_type,
+ .value_list = vl,
+ .value_list_name = name,
+ .ret = 0};
+
+ plugin_ctx_t old_ctx = plugin_set_ctx(cef->plugin_ctx);
+ int status = (*callback)(&event, &cef->user_data);
+ plugin_set_ctx(old_ctx);
+
+ if (status != 0) {
+ ERROR("plugin_dispatch_cache_event: Callback \"%s\" failed with status "
+ "%i for event %s.",
+ cef->name, status,
+ ((event_type == CE_VALUE_UPDATE) ? "UPDATE" : "EXPIRED"));
+ }
+ }
+ break;
+ }
+ return;
+}
+
static int plugin_dispatch_values_internal(value_list_t *vl) {
int status;
static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
return false;
} /* }}} bool check_drop_value */
-int plugin_dispatch_values(value_list_t const *vl) {
+EXPORT int plugin_dispatch_values(value_list_t const *vl) {
int status;
- static pthread_mutex_t statistics_lock = PTHREAD_MUTEX_INITIALIZER;
if (check_drop_value()) {
if (record_statistics) {
gauge_t sum = 0.0;
va_list ap;
+ if (check_drop_value()) {
+ if (record_statistics) {
+ pthread_mutex_lock(&statistics_lock);
+ stats_values_dropped++;
+ pthread_mutex_unlock(&statistics_lock);
+ }
+ return 0;
+ }
+
assert(template->values_len == 1);
/* Calculate sum for Gauge to calculate percent if needed */
return failed;
} /* }}} int plugin_dispatch_multivalue */
-int plugin_dispatch_notification(const notification_t *notif) {
+EXPORT int plugin_dispatch_notification(const notification_t *notif) {
llentry_t *le;
/* Possible TODO: Add flap detection here */
return 0;
} /* int plugin_dispatch_notification */
-void plugin_log(int level, const char *format, ...) {
+EXPORT void plugin_log(int level, const char *format, ...) {
char msg[1024];
va_list ap;
llentry_t *le;
return log_level;
} /* int parse_log_severity */
-int parse_notif_severity(const char *severity) {
+EXPORT int parse_notif_severity(const char *severity) {
int notif_severity = -1;
if (strcasecmp(severity, "FAILURE") == 0)
return notif_severity;
} /* int parse_notif_severity */
-const data_set_t *plugin_get_ds(const char *name) {
+EXPORT const data_set_t *plugin_get_ds(const char *name) {
data_set_t *ds;
if (data_sets == NULL) {
return ctx;
} /* int plugin_ctx_create */
-void plugin_init_ctx(void) {
+EXPORT void plugin_init_ctx(void) {
pthread_key_create(&plugin_ctx_key, plugin_ctx_destructor);
plugin_ctx_key_initialized = true;
} /* void plugin_init_ctx */
-plugin_ctx_t plugin_get_ctx(void) {
+EXPORT plugin_ctx_t plugin_get_ctx(void) {
plugin_ctx_t *ctx;
assert(plugin_ctx_key_initialized);
return *ctx;
} /* plugin_ctx_t plugin_get_ctx */
-plugin_ctx_t plugin_set_ctx(plugin_ctx_t ctx) {
+EXPORT plugin_ctx_t plugin_set_ctx(plugin_ctx_t ctx) {
plugin_ctx_t *c;
plugin_ctx_t old;
return old;
} /* void plugin_set_ctx */
-cdtime_t plugin_get_interval(void) {
+EXPORT cdtime_t plugin_get_interval(void) {
cdtime_t interval;
interval = plugin_get_ctx().interval;
projects / collectd.git / blobdiff