* Sebastian Harl <sh at tokkee.org>
**/
+/* _GNU_SOURCE is needed in Linux to use pthread_setname_np */
+#define _GNU_SOURCE
+
#include "collectd.h"
#include "common.h"
#include "utils_random.h"
#include "utils_time.h"
-#include <ltdl.h>
+#if HAVE_PTHREAD_NP_H
+#include <pthread_np.h> /* for pthread_set_name_np(3) */
+#endif
+
+#include <dlfcn.h>
/*
* Private structures
static char *plugindir = NULL;
#ifndef DEFAULT_MAX_READ_INTERVAL
-#define DEFAULT_MAX_READ_INTERVAL TIME_T_TO_CDTIME_T(86400)
+#define DEFAULT_MAX_READ_INTERVAL TIME_T_TO_CDTIME_T_STATIC(86400)
#endif
static c_heap_t *read_heap = NULL;
static llist_t *read_list;
static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER;
static pthread_t *read_threads = NULL;
-static int read_threads_num = 0;
+static size_t read_threads_num = 0;
static cdtime_t max_read_interval = DEFAULT_MAX_READ_INTERVAL;
static write_queue_t *write_queue_head;
static const char *plugin_get_dir(void) {
if (plugindir == NULL)
- return (PLUGINDIR);
+ return PLUGINDIR;
else
- return (plugindir);
+ return plugindir;
}
-static void plugin_update_internal_statistics(void) { /* {{{ */
- derive_t copy_write_queue_length;
- value_list_t vl = VALUE_LIST_INIT;
- value_t values[2];
-
- copy_write_queue_length = write_queue_length;
+static int plugin_update_internal_statistics(void) { /* {{{ */
+ gauge_t copy_write_queue_length = (gauge_t)write_queue_length;
/* Initialize `vl' */
- vl.values = values;
- vl.values_len = 2;
- vl.time = 0;
- sstrncpy(vl.host, hostname_g, sizeof(vl.host));
+ value_list_t vl = VALUE_LIST_INIT;
sstrncpy(vl.plugin, "collectd", sizeof(vl.plugin));
-
- vl.type_instance[0] = 0;
- vl.values_len = 1;
+ vl.interval = plugin_get_interval();
/* Write queue */
sstrncpy(vl.plugin_instance, "write_queue", sizeof(vl.plugin_instance));
/* Write queue : queue length */
- vl.values[0].gauge = (gauge_t)copy_write_queue_length;
+ vl.values = &(value_t){.gauge = copy_write_queue_length};
+ vl.values_len = 1;
sstrncpy(vl.type, "queue_length", sizeof(vl.type));
vl.type_instance[0] = 0;
plugin_dispatch_values(&vl);
/* Write queue : Values dropped (queue length > low limit) */
- vl.values[0].derive = (derive_t)stats_values_dropped;
+ vl.values = &(value_t){.gauge = (gauge_t)stats_values_dropped};
+ vl.values_len = 1;
sstrncpy(vl.type, "derive", sizeof(vl.type));
sstrncpy(vl.type_instance, "dropped", sizeof(vl.type_instance));
plugin_dispatch_values(&vl);
sstrncpy(vl.plugin_instance, "cache", sizeof(vl.plugin_instance));
/* Cache : Nb entry in cache tree */
- vl.values[0].gauge = (gauge_t)uc_get_size();
+ vl.values = &(value_t){.gauge = (gauge_t)uc_get_size()};
+ vl.values_len = 1;
sstrncpy(vl.type, "cache_size", sizeof(vl.type));
vl.type_instance[0] = 0;
plugin_dispatch_values(&vl);
- return;
-} /* }}} void plugin_update_internal_statistics */
+ return 0;
+} /* }}} int plugin_update_internal_statistics */
-static void destroy_callback(callback_func_t *cf) /* {{{ */
+static void free_userdata(user_data_t const *ud) /* {{{ */
{
- if (cf == NULL)
+ if (ud == NULL)
return;
- if ((cf->cf_udata.data != NULL) && (cf->cf_udata.free_func != NULL)) {
- cf->cf_udata.free_func(cf->cf_udata.data);
- cf->cf_udata.data = NULL;
- cf->cf_udata.free_func = NULL;
+ if ((ud->data != NULL) && (ud->free_func != NULL)) {
+ ud->free_func(ud->data);
}
+} /* }}} void free_userdata */
+
+static void destroy_callback(callback_func_t *cf) /* {{{ */
+{
+ if (cf == NULL)
+ return;
+ free_userdata(&cf->cf_udata);
sfree(cf);
} /* }}} void destroy_callback */
ERROR("plugin: register_callback: "
"llist_create failed.");
destroy_callback(cf);
- return (-1);
+ return -1;
}
}
if (key == NULL) {
ERROR("plugin: register_callback: strdup failed.");
destroy_callback(cf);
- return (-1);
+ return -1;
}
le = llist_search(*list, name);
"llentry_create failed.");
sfree(key);
destroy_callback(cf);
- return (-1);
+ return -1;
}
llist_append(*list, le);
sfree(key);
}
- return (0);
+ return 0;
} /* }}} int register_callback */
static void log_list_callbacks(llist_t **list, /* {{{ */
cf = calloc(1, sizeof(*cf));
if (cf == NULL) {
+ free_userdata(ud);
ERROR("plugin: create_register_callback: calloc failed.");
- return (-1);
+ return -1;
}
cf->cf_callback = callback;
cf->cf_ctx = plugin_get_ctx();
- return (register_callback(list, name, cf));
+ return register_callback(list, name, cf);
} /* }}} int create_register_callback */
static int plugin_unregister(llist_t *list, const char *name) /* {{{ */
llentry_t *e;
if (list == NULL)
- return (-1);
+ return -1;
e = llist_search(list, name);
if (e == NULL)
- return (-1);
+ return -1;
llist_remove(list, e);
llentry_destroy(e);
- return (0);
+ return 0;
} /* }}} int plugin_unregister */
/*
* object, but it will bitch about a shared object not having a
* ``module_register'' symbol..
*/
-static int plugin_load_file(char *file, uint32_t flags) {
- lt_dlhandle dlh;
+static int plugin_load_file(const char *file, _Bool global) {
void (*reg_handle)(void);
- lt_dlinit();
- lt_dlerror(); /* clear errors */
+ int flags = RTLD_NOW;
+ if (global)
+ flags |= RTLD_GLOBAL;
-#if LIBTOOL_VERSION == 2
- if (flags & PLUGIN_FLAGS_GLOBAL) {
- lt_dladvise advise;
- lt_dladvise_init(&advise);
- lt_dladvise_global(&advise);
- dlh = lt_dlopenadvise(file, advise);
- lt_dladvise_destroy(&advise);
- } else {
- dlh = lt_dlopen(file);
- }
-#else /* if LIBTOOL_VERSION == 1 */
- if (flags & PLUGIN_FLAGS_GLOBAL)
- WARNING("plugin_load_file: The global flag is not supported, "
- "libtool 2 is required for this.");
- dlh = lt_dlopen(file);
-#endif
+ void *dlh = dlopen(file, flags);
if (dlh == NULL) {
char errbuf[1024] = "";
- ssnprintf(errbuf, sizeof(errbuf),
- "lt_dlopen (\"%s\") failed: %s. "
- "The most common cause for this problem is "
- "missing dependencies. Use ldd(1) to check "
- "the dependencies of the plugin "
- "/ shared object.",
- file, lt_dlerror());
+ snprintf(errbuf, sizeof(errbuf),
+ "dlopen (\"%s\") failed: %s. "
+ "The most common cause for this problem is "
+ "missing dependencies. Use ldd(1) to check "
+ "the dependencies of the plugin "
+ "/ shared object.",
+ file, dlerror());
ERROR("%s", errbuf);
/* Make sure this is printed to STDERR in any case, but also
if (list_log != NULL)
fprintf(stderr, "ERROR: %s\n", errbuf);
- return (1);
+ return 1;
}
- if ((reg_handle = (void (*)(void))lt_dlsym(dlh, "module_register")) == NULL) {
+ reg_handle = (void (*)(void))dlsym(dlh, "module_register");
+ if (reg_handle == NULL) {
WARNING("Couldn't find symbol \"module_register\" in \"%s\": %s\n", file,
- lt_dlerror());
- lt_dlclose(dlh);
- return (-1);
+ dlerror());
+ dlclose(dlh);
+ return -1;
}
(*reg_handle)();
- return (0);
+ return 0;
}
static void *plugin_read_thread(void __attribute__((unused)) * args) {
* pthread_cond_timedwait returns. */
rc = 0;
while ((read_loop != 0) && (cdtime() < rf->rf_next_read) && rc == 0) {
- struct timespec ts = {0};
-
- CDTIME_T_TO_TIMESPEC(rf->rf_next_read, &ts);
-
- rc = pthread_cond_timedwait(&read_cond, &read_lock, &ts);
+ rc = pthread_cond_timedwait(&read_cond, &read_lock,
+ &CDTIME_T_TO_TIMESPEC(rf->rf_next_read));
}
/* Must hold `read_lock' when accessing `rf->rf_type'. */
} /* while (read_loop) */
pthread_exit(NULL);
- return ((void *)0);
+ return (void *)0;
} /* void *plugin_read_thread */
-static void start_read_threads(int num) {
+#ifdef PTHREAD_MAX_NAMELEN_NP
+#define THREAD_NAME_MAX PTHREAD_MAX_NAMELEN_NP
+#else
+#define THREAD_NAME_MAX 16
+#endif
+
+static void set_thread_name(pthread_t tid, char const *name) {
+#if defined(HAVE_PTHREAD_SETNAME_NP) || defined(HAVE_PTHREAD_SET_NAME_NP)
+
+ /* glibc limits the length of the name and fails if the passed string
+ * is too long, so we truncate it here. */
+ char n[THREAD_NAME_MAX];
+ if (strlen(name) >= THREAD_NAME_MAX)
+ WARNING("set_thread_name(\"%s\"): name too long", name);
+ sstrncpy(n, name, sizeof(n));
+
+#if defined(HAVE_PTHREAD_SETNAME_NP)
+ int status = pthread_setname_np(tid, n);
+ if (status != 0) {
+ char errbuf[1024];
+ ERROR("set_thread_name(\"%s\"): %s", n,
+ sstrerror(status, errbuf, sizeof(errbuf)));
+ }
+#else /* if defined(HAVE_PTHREAD_SET_NAME_NP) */
+ pthread_set_name_np(tid, n);
+#endif
+
+#endif
+}
+
+static void start_read_threads(size_t num) /* {{{ */
+{
if (read_threads != NULL)
return;
}
read_threads_num = 0;
- for (int i = 0; i < num; i++) {
- if (pthread_create(read_threads + read_threads_num, NULL,
- plugin_read_thread, NULL) == 0) {
- read_threads_num++;
- } else {
- ERROR("plugin: start_read_threads: pthread_create failed.");
+ for (size_t i = 0; i < num; i++) {
+ int status = pthread_create(read_threads + read_threads_num,
+ /* attr = */ NULL, plugin_read_thread,
+ /* arg = */ NULL);
+ if (status != 0) {
+ char errbuf[1024];
+ ERROR("plugin: start_read_threads: pthread_create failed "
+ "with status %i (%s).",
+ status, sstrerror(status, errbuf, sizeof(errbuf)));
return;
}
+
+ char name[THREAD_NAME_MAX];
+ snprintf(name, sizeof(name), "reader#%" PRIsz, read_threads_num);
+ set_thread_name(read_threads[read_threads_num], name);
+
+ read_threads_num++;
} /* for (i) */
-} /* void start_read_threads */
+} /* }}} void start_read_threads */
static void stop_read_threads(void) {
if (read_threads == NULL)
return;
- INFO("collectd: Stopping %i read threads.", read_threads_num);
+ INFO("collectd: Stopping %" PRIsz " read threads.", read_threads_num);
pthread_mutex_lock(&read_lock);
read_loop = 0;
pthread_cond_broadcast(&read_cond);
pthread_mutex_unlock(&read_lock);
- for (int i = 0; i < read_threads_num; i++) {
+ for (size_t i = 0; i < read_threads_num; i++) {
if (pthread_join(read_threads[i], NULL) != 0) {
ERROR("plugin: stop_read_threads: pthread_join failed.");
}
value_list_t *vl;
if (vl_orig == NULL)
- return (NULL);
+ return NULL;
vl = malloc(sizeof(*vl));
if (vl == NULL)
- return (NULL);
+ return NULL;
memcpy(vl, vl_orig, sizeof(*vl));
+ if (vl->host[0] == 0)
+ sstrncpy(vl->host, hostname_g, sizeof(vl->host));
+
vl->values = calloc(vl_orig->values_len, sizeof(*vl->values));
if (vl->values == NULL) {
plugin_value_list_free(vl);
- return (NULL);
+ return NULL;
}
memcpy(vl->values, vl_orig->values,
vl_orig->values_len * sizeof(*vl->values));
vl->meta = meta_data_clone(vl->meta);
if ((vl_orig->meta != NULL) && (vl->meta == NULL)) {
plugin_value_list_free(vl);
- return (NULL);
+ return NULL;
}
if (vl->time == 0)
}
}
- return (vl);
+ return vl;
} /* }}} value_list_t *plugin_value_list_clone */
static int plugin_write_enqueue(value_list_t const *vl) /* {{{ */
q = malloc(sizeof(*q));
if (q == NULL)
- return (ENOMEM);
+ return ENOMEM;
q->next = NULL;
q->vl = plugin_value_list_clone(vl);
if (q->vl == NULL) {
sfree(q);
- return (ENOMEM);
+ return ENOMEM;
}
/* Store context of caller (read plugin); otherwise, it would not be
pthread_cond_signal(&write_cond);
pthread_mutex_unlock(&write_lock);
- return (0);
+ return 0;
} /* }}} int plugin_write_enqueue */
static value_list_t *plugin_write_dequeue(void) /* {{{ */
if (write_queue_head == NULL) {
pthread_mutex_unlock(&write_lock);
- return (NULL);
+ return NULL;
}
q = write_queue_head;
vl = q->vl;
sfree(q);
- return (vl);
+ return vl;
} /* }}} value_list_t *plugin_write_dequeue */
static void *plugin_write_thread(void __attribute__((unused)) * args) /* {{{ */
}
pthread_exit(NULL);
- return ((void *)0);
+ return (void *)0;
} /* }}} void *plugin_write_thread */
static void start_write_threads(size_t num) /* {{{ */
write_threads_num = 0;
for (size_t i = 0; i < num; i++) {
- int status;
-
- status = pthread_create(write_threads + write_threads_num,
- /* attr = */ NULL, plugin_write_thread,
- /* arg = */ NULL);
+ int status = pthread_create(write_threads + write_threads_num,
+ /* attr = */ NULL, plugin_write_thread,
+ /* arg = */ NULL);
if (status != 0) {
char errbuf[1024];
ERROR("plugin: start_write_threads: pthread_create failed "
return;
}
+ char name[THREAD_NAME_MAX];
+ snprintf(name, sizeof(name), "writer#%" PRIsz, write_threads_num);
+ set_thread_name(write_threads[write_threads_num], name);
+
write_threads_num++;
} /* for (i) */
} /* }}} void start_write_threads */
if (write_threads == NULL)
return;
- INFO("collectd: Stopping %zu write threads.", write_threads_num);
+ INFO("collectd: Stopping %" PRIsz " write threads.", write_threads_num);
pthread_mutex_lock(&write_lock);
write_loop = 0;
pthread_mutex_unlock(&write_lock);
if (i > 0) {
- WARNING("plugin: %zu value list%s left after shutting down "
+ WARNING("plugin: %" PRIsz " value list%s left after shutting down "
"the write threads.",
i, (i == 1) ? " was" : "s were");
}
assert(plugins_loaded != NULL);
status = c_avl_get(plugins_loaded, name, /* ret_value = */ NULL);
- return (status == 0);
+ return status == 0;
}
static int plugin_mark_loaded(char const *name) {
name_copy = strdup(name);
if (name_copy == NULL)
- return (ENOMEM);
+ return ENOMEM;
status = c_avl_insert(plugins_loaded,
/* key = */ name_copy, /* value = */ NULL);
- return (status);
+ return status;
}
static void plugin_free_loaded(void) {
}
#define BUFSIZE 512
-int plugin_load(char const *plugin_name, uint32_t flags) {
+int plugin_load(char const *plugin_name, _Bool global) {
DIR *dh;
const char *dir;
char filename[BUFSIZE] = "";
int status;
if (plugin_name == NULL)
- return (EINVAL);
+ return EINVAL;
/* Check if plugin is already loaded and don't do anything in this
* case. */
if (plugin_is_loaded(plugin_name))
- return (0);
+ return 0;
dir = plugin_get_dir();
ret = 1;
*/
if ((strcasecmp("perl", plugin_name) == 0) ||
(strcasecmp("python", plugin_name) == 0))
- flags |= PLUGIN_FLAGS_GLOBAL;
+ global = 1;
/* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
* type when matching the filename */
- status = ssnprintf(typename, sizeof(typename), "%s.so", plugin_name);
+ status = snprintf(typename, sizeof(typename), "%s.so", plugin_name);
if ((status < 0) || ((size_t)status >= sizeof(typename))) {
WARNING("plugin_load: Filename too long: \"%s.so\"", plugin_name);
- return (-1);
+ return -1;
}
if ((dh = opendir(dir)) == NULL) {
char errbuf[1024];
ERROR("plugin_load: opendir (%s) failed: %s", dir,
sstrerror(errno, errbuf, sizeof(errbuf)));
- return (-1);
+ return -1;
}
while ((de = readdir(dh)) != NULL) {
if (strcasecmp(de->d_name, typename))
continue;
- status = ssnprintf(filename, sizeof(filename), "%s/%s", dir, de->d_name);
+ status = snprintf(filename, sizeof(filename), "%s/%s", dir, de->d_name);
if ((status < 0) || ((size_t)status >= sizeof(filename))) {
WARNING("plugin_load: Filename too long: \"%s/%s\"", dir, de->d_name);
continue;
continue;
}
- status = plugin_load_file(filename, flags);
+ status = plugin_load_file(filename, global);
if (status == 0) {
/* success */
plugin_mark_loaded(plugin_name);
if (filename[0] == 0)
ERROR("plugin_load: Could not find plugin \"%s\" in %s", plugin_name, dir);
- return (ret);
+ return ret;
}
/*
int (*callback)(const char *key, const char *val),
const char **keys, int keys_num) {
cf_register(name, callback, keys, keys_num);
- return (0);
+ return 0;
} /* int plugin_register_config */
int plugin_register_complex_config(const char *type,
int (*callback)(oconfig_item_t *)) {
- return (cf_register_complex(type, callback));
+ return cf_register_complex(type, callback);
} /* int plugin_register_complex_config */
int plugin_register_init(const char *name, int (*callback)(void)) {
- return (create_register_callback(&list_init, name, (void *)callback,
- /* user_data = */ NULL));
+ return create_register_callback(&list_init, name, (void *)callback, NULL);
} /* plugin_register_init */
static int plugin_compare_read_func(const void *arg0, const void *arg1) {
rf1 = arg1;
if (rf0->rf_next_read < rf1->rf_next_read)
- return (-1);
+ return -1;
else if (rf0->rf_next_read > rf1->rf_next_read)
- return (1);
+ return 1;
else
- return (0);
+ return 0;
} /* int plugin_compare_read_func */
/* Add a read function to both, the heap and a linked list. The linked list if
if (read_list == NULL) {
pthread_mutex_unlock(&read_lock);
ERROR("plugin_insert_read: read_list failed.");
- return (-1);
+ return -1;
}
}
if (read_heap == NULL) {
pthread_mutex_unlock(&read_lock);
ERROR("plugin_insert_read: c_heap_create failed.");
- return (-1);
+ return -1;
}
}
if (le != NULL) {
pthread_mutex_unlock(&read_lock);
WARNING("The read function \"%s\" is already registered. "
- "Check for duplicate \"LoadPlugin\" lines "
- "in your configuration!",
+ "Check for duplicates in your configuration!",
rf->rf_name);
- return (EINVAL);
+ return EINVAL;
}
le = llentry_create(rf->rf_name, rf);
if (le == NULL) {
pthread_mutex_unlock(&read_lock);
ERROR("plugin_insert_read: llentry_create failed.");
- return (-1);
+ return -1;
}
status = c_heap_insert(read_heap, rf);
pthread_mutex_unlock(&read_lock);
ERROR("plugin_insert_read: c_heap_insert failed.");
llentry_destroy(le);
- return (-1);
+ return -1;
}
/* This does not fail. */
/* Wake up all the read threads. */
pthread_cond_broadcast(&read_cond);
pthread_mutex_unlock(&read_lock);
- return (0);
+ return 0;
} /* int plugin_insert_read */
int plugin_register_read(const char *name, int (*callback)(void)) {
rf = calloc(1, sizeof(*rf));
if (rf == NULL) {
ERROR("plugin_register_read: calloc failed.");
- return (ENOMEM);
+ return ENOMEM;
}
rf->rf_callback = (void *)callback;
sfree(rf);
}
- return (status);
+ return status;
} /* int plugin_register_read */
int plugin_register_complex_read(const char *group, const char *name,
rf = calloc(1, sizeof(*rf));
if (rf == NULL) {
+ free_userdata(user_data);
ERROR("plugin_register_complex_read: calloc failed.");
- return (ENOMEM);
+ return ENOMEM;
}
rf->rf_callback = (void *)callback;
status = plugin_insert_read(rf);
if (status != 0) {
+ free_userdata(&rf->rf_udata);
sfree(rf->rf_name);
sfree(rf);
}
- return (status);
+ return status;
} /* int plugin_register_complex_read */
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));
+ return create_register_callback(&list_write, name, (void *)callback, ud);
} /* int plugin_register_write */
static int plugin_flush_timeout_callback(user_data_t *ud) {
flush_callback_t *cb = ud->data;
- return plugin_flush(cb->name, cb->timeout, /* identifier = */ NULL);
+ return plugin_flush(cb->name, cb->timeout, NULL);
} /* static int plugin_flush_callback */
static void plugin_flush_timeout_callback_free(void *data) {
flush_name = malloc(name_size + prefix_size + 1);
if (flush_name == NULL) {
ERROR("plugin_flush_callback_name: malloc failed.");
- return (NULL);
+ return NULL;
}
sstrncpy(flush_name, flush_prefix, prefix_size + 1);
flush_name = plugin_flush_callback_name(name);
if (flush_name == NULL)
- return (-1);
+ return -1;
cb = malloc(sizeof(*cb));
if (cb == NULL) {
ERROR("plugin_register_flush: malloc failed.");
sfree(flush_name);
- return (-1);
+ return -1;
}
cb->name = strdup(name);
ERROR("plugin_register_flush: strdup failed.");
sfree(cb);
sfree(flush_name);
- return (-1);
+ return -1;
}
cb->timeout = ctx.flush_timeout;
});
sfree(flush_name);
- if (status != 0) {
- sfree(cb->name);
- sfree(cb);
- return status;
- }
+ return status;
}
return 0;
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));
+ return create_register_callback(&list_missing, name, (void *)callback, ud);
} /* int plugin_register_missing */
int plugin_register_shutdown(const char *name, int (*callback)(void)) {
- return (create_register_callback(&list_shutdown, name, (void *)callback,
- /* user_data = */ NULL));
+ return create_register_callback(&list_shutdown, name, (void *)callback, NULL);
} /* int plugin_register_shutdown */
static void plugin_free_data_sets(void) {
} else if (data_sets == NULL) {
data_sets = c_avl_create((int (*)(const void *, const void *))strcmp);
if (data_sets == NULL)
- return (-1);
+ return -1;
}
ds_copy = malloc(sizeof(*ds_copy));
if (ds_copy == NULL)
- return (-1);
+ return -1;
memcpy(ds_copy, ds, sizeof(data_set_t));
ds_copy->ds = malloc(sizeof(*ds_copy->ds) * ds->ds_num);
if (ds_copy->ds == NULL) {
sfree(ds_copy);
- return (-1);
+ return -1;
}
for (size_t i = 0; i < ds->ds_num; i++)
memcpy(ds_copy->ds + i, ds->ds + i, sizeof(data_source_t));
- return (c_avl_insert(data_sets, (void *)ds_copy->type, (void *)ds_copy));
+ 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) {
- return (create_register_callback(&list_log, name, (void *)callback, 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) {
- return (
- create_register_callback(&list_notification, name, (void *)callback, ud));
+ return create_register_callback(&list_notification, name, (void *)callback,
+ ud);
} /* int plugin_register_log */
int plugin_unregister_config(const char *name) {
cf_unregister(name);
- return (0);
+ return 0;
} /* int plugin_unregister_config */
int plugin_unregister_complex_config(const char *name) {
cf_unregister_complex(name);
- return (0);
+ return 0;
} /* int plugin_unregister_complex_config */
int plugin_unregister_init(const char *name) {
- return (plugin_unregister(list_init, name));
+ return plugin_unregister(list_init, name);
}
int plugin_unregister_read(const char *name) /* {{{ */
read_func_t *rf;
if (name == NULL)
- return (-ENOENT);
+ return -ENOENT;
pthread_mutex_lock(&read_lock);
if (read_list == NULL) {
pthread_mutex_unlock(&read_lock);
- return (-ENOENT);
+ return -ENOENT;
}
le = llist_search(read_list, name);
if (le == NULL) {
pthread_mutex_unlock(&read_lock);
WARNING("plugin_unregister_read: No such read function: %s", name);
- return (-ENOENT);
+ return -ENOENT;
}
llist_remove(read_list, le);
DEBUG("plugin_unregister_read: Marked `%s' for removal.", name);
- return (0);
+ return 0;
} /* }}} int plugin_unregister_read */
void plugin_log_available_writers(void) {
int found = 0;
if (group == NULL)
- return (-ENOENT);
+ return -ENOENT;
pthread_mutex_lock(&read_lock);
if (read_list == NULL) {
pthread_mutex_unlock(&read_lock);
- return (-ENOENT);
+ return -ENOENT;
}
while (42) {
WARNING("plugin_unregister_read_group: No such "
"group of read function: %s",
group);
- return (-ENOENT);
+ return -ENOENT;
}
- return (0);
+ return 0;
} /* }}} int plugin_unregister_read_group */
int plugin_unregister_write(const char *name) {
- return (plugin_unregister(list_write, name));
+ return plugin_unregister(list_write, name);
}
int plugin_unregister_flush(const char *name) {
}
int plugin_unregister_missing(const char *name) {
- return (plugin_unregister(list_missing, name));
+ return plugin_unregister(list_missing, name);
}
int plugin_unregister_shutdown(const char *name) {
- return (plugin_unregister(list_shutdown, name));
+ return plugin_unregister(list_shutdown, name);
}
int plugin_unregister_data_set(const char *name) {
data_set_t *ds;
if (data_sets == NULL)
- return (-1);
+ return -1;
if (c_avl_remove(data_sets, name, NULL, (void *)&ds) != 0)
- return (-1);
+ return -1;
sfree(ds->ds);
sfree(ds);
- return (0);
+ return 0;
} /* int plugin_unregister_data_set */
int plugin_unregister_log(const char *name) {
- return (plugin_unregister(list_log, name));
+ return plugin_unregister(list_log, name);
}
int plugin_unregister_notification(const char *name) {
- return (plugin_unregister(list_notification, name));
+ return plugin_unregister(list_notification, name);
}
int plugin_init_all(void) {
/* Init the value cache */
uc_init();
- if (IS_TRUE(global_option_get("CollectInternalStats")))
+ if (IS_TRUE(global_option_get("CollectInternalStats"))) {
record_statistics = 1;
+ plugin_register_read("collectd", plugin_update_internal_statistics);
+ }
chain_name = global_option_get("PreCacheChain");
pre_cache_chain = fc_chain_get_by_name(chain_name);
rt = global_option_get("ReadThreads");
num = atoi(rt);
if (num != -1)
- start_read_threads((num > 0) ? num : 5);
+ start_read_threads((num > 0) ? ((size_t)num) : 5);
}
return ret;
} /* void plugin_init_all */
/* TODO: Rename this function. */
void plugin_read_all(void) {
- if (record_statistics) {
- plugin_update_internal_statistics();
- }
uc_check_timeout();
return;
if (read_heap == NULL) {
NOTICE("No read-functions are registered.");
- return (0);
+ return 0;
}
while (42) {
destroy_callback((void *)rf);
}
- return (return_status);
+ return return_status;
} /* int plugin_read_all_once */
int plugin_write(const char *plugin, /* {{{ */
int status;
if (vl == NULL)
- return (EINVAL);
+ return EINVAL;
if (list_write == NULL)
- return (ENOENT);
+ return ENOENT;
if (ds == NULL) {
ds = plugin_get_ds(vl->type);
if (ds == NULL) {
ERROR("plugin_write: Unable to lookup type `%s'.", vl->type);
- return (ENOENT);
+ return ENOENT;
}
}
}
if (le == NULL)
- return (ENOENT);
+ return ENOENT;
cf = le->value;
status = (*callback)(ds, vl, &cf->cf_udata);
}
- return (status);
+ return status;
} /* }}} int plugin_write */
int plugin_flush(const char *plugin, cdtime_t timeout, const char *identifier) {
llentry_t *le;
if (list_flush == NULL)
- return (0);
+ return 0;
le = llist_head(list_flush);
while (le != NULL) {
le = le->next;
}
- return (0);
+ return 0;
} /* int plugin_flush */
int plugin_shutdown_all(void) {
plugin_free_loaded();
plugin_free_data_sets();
- return (ret);
+ return ret;
} /* void plugin_shutdown_all */
int plugin_dispatch_missing(const value_list_t *vl) /* {{{ */
{
- llentry_t *le;
-
if (list_missing == NULL)
- return (0);
+ return 0;
- le = llist_head(list_missing);
+ llentry_t *le = llist_head(list_missing);
while (le != NULL) {
- callback_func_t *cf;
- plugin_missing_cb callback;
- plugin_ctx_t old_ctx;
- int status;
-
- cf = le->value;
- old_ctx = plugin_set_ctx(cf->cf_ctx);
- callback = cf->cf_callback;
+ callback_func_t *cf = le->value;
+ plugin_ctx_t old_ctx = plugin_set_ctx(cf->cf_ctx);
+ plugin_missing_cb callback = cf->cf_callback;
- status = (*callback)(vl, &cf->cf_udata);
+ int status = (*callback)(vl, &cf->cf_udata);
plugin_set_ctx(old_ctx);
if (status != 0) {
if (status < 0) {
ERROR("plugin_dispatch_missing: Callback function \"%s\" "
"failed with status %i.",
le->key, status);
- return (status);
+ return status;
} else {
- return (0);
+ return 0;
}
}
le = le->next;
}
- return (0);
+ return 0;
} /* int }}} plugin_dispatch_missing */
static int plugin_dispatch_values_internal(value_list_t *vl) {
int status;
static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
- value_t *saved_values;
- int saved_values_len;
-
- data_set_t *ds;
+ _Bool free_meta_data = 0;
- int free_meta_data = 0;
+ assert(vl != NULL);
- assert(vl);
- assert(vl->plugin);
+ /* These fields are initialized by plugin_value_list_clone() if needed: */
+ assert(vl->host[0] != 0);
+ assert(vl->time != 0); /* The time is determined at _enqueue_ time. */
+ assert(vl->interval != 0);
if (vl->type[0] == 0 || vl->values == NULL || vl->values_len < 1) {
ERROR("plugin_dispatch_values: Invalid value list "
"from plugin %s.",
vl->plugin);
- return (-1);
+ return -1;
}
/* Free meta data only if the calling function didn't specify any. In
ERROR("plugin_dispatch_values: No data sets registered. "
"Could the types database be read? Check "
"your `TypesDB' setting!");
- return (-1);
+ return -1;
}
+ data_set_t *ds = NULL;
if (c_avl_get(data_sets, vl->type, (void *)&ds) != 0) {
char ident[6 * DATA_MAX_NAME_LEN];
INFO("plugin_dispatch_values: Dataset not found: %s "
"(from \"%s\"), check your types.db!",
vl->type, ident);
- return (-1);
+ return -1;
}
- /* Assured by plugin_value_list_clone(). The time is determined at
- * _enqueue_ time. */
- assert(vl->time != 0);
- assert(vl->interval != 0);
-
DEBUG("plugin_dispatch_values: time = %.3f; interval = %.3f; "
"host = %s; "
"plugin = %s; plugin_instance = %s; "
#else
if (ds->ds_num != vl->values_len) {
ERROR("plugin_dispatch_values: ds->type = %s: "
- "(ds->ds_num = %zu) != "
- "(vl->values_len = %zu)",
+ "(ds->ds_num = %" PRIsz ") != "
+ "(vl->values_len = %" PRIsz ")",
ds->type, ds->ds_num, vl->values_len);
- return (-1);
+ return -1;
}
#endif
escape_slashes(vl->type, sizeof(vl->type));
escape_slashes(vl->type_instance, sizeof(vl->type_instance));
- /* Copy the values. This way, we can assure `targets' that they get
- * dynamically allocated values, which they can free and replace if
- * they like. */
- if ((pre_cache_chain != NULL) || (post_cache_chain != NULL)) {
- saved_values = vl->values;
- saved_values_len = vl->values_len;
-
- vl->values = (value_t *)calloc(vl->values_len, sizeof(*vl->values));
- if (vl->values == NULL) {
- ERROR("plugin_dispatch_values: calloc failed.");
- vl->values = saved_values;
- return (-1);
- }
- memcpy(vl->values, saved_values, vl->values_len * sizeof(*vl->values));
- } else /* if ((pre == NULL) && (post == NULL)) */
- {
- saved_values = NULL;
- saved_values_len = 0;
- }
-
if (pre_cache_chain != NULL) {
status = fc_process_chain(ds, vl, pre_cache_chain);
if (status < 0) {
"pre-cache chain failed with "
"status %i (%#x).",
status, status);
- } else if (status == FC_TARGET_STOP) {
- /* Restore the state of the value_list so that plugins
- * don't get confused.. */
- if (saved_values != NULL) {
- sfree(vl->values);
- vl->values = saved_values;
- vl->values_len = saved_values_len;
- }
- return (0);
- }
+ } else if (status == FC_TARGET_STOP)
+ return 0;
}
/* Update the value cache */
} else
fc_default_action(ds, vl);
- /* Restore the state of the value_list so that plugins don't get
- * confused.. */
- if (saved_values != NULL) {
- sfree(vl->values);
- vl->values = saved_values;
- vl->values_len = saved_values_len;
- }
-
if ((free_meta_data != 0) && (vl->meta != NULL)) {
meta_data_destroy(vl->meta);
vl->meta = NULL;
}
- return (0);
+ return 0;
} /* int plugin_dispatch_values_internal */
static double get_drop_probability(void) /* {{{ */
pthread_mutex_unlock(&write_lock);
if (wql < write_limit_low)
- return (0.0);
+ return 0.0;
if (wql >= write_limit_high)
- return (1.0);
+ return 1.0;
pos = 1 + wql - write_limit_low;
size = 1 + write_limit_high - write_limit_low;
- return (((double)pos) / ((double)size));
+ return (double)pos / (double)size;
} /* }}} double get_drop_probability */
static _Bool check_drop_value(void) /* {{{ */
int status;
if (write_limit_high == 0)
- return (0);
+ return 0;
p = get_drop_probability();
if (p == 0.0)
- return (0);
+ return 0;
status = pthread_mutex_trylock(&last_message_lock);
if (status == 0) {
}
if (p == 1.0)
- return (1);
+ return 1;
q = cdrand_d();
if (q > p)
- return (1);
+ return 1;
else
- return (0);
+ return 0;
} /* }}} _Bool check_drop_value */
int plugin_dispatch_values(value_list_t const *vl) {
stats_values_dropped++;
pthread_mutex_unlock(&statistics_lock);
}
- return (0);
+ return 0;
}
status = plugin_write_enqueue(vl);
ERROR("plugin_dispatch_values: plugin_write_enqueue failed "
"with status %i (%s).",
status, sstrerror(status, errbuf, sizeof(errbuf)));
- return (status);
+ return status;
}
- return (0);
+ return 0;
}
__attribute__((sentinel)) int
va_end(ap);
plugin_value_list_free(vl);
- return (failed);
+ return failed;
} /* }}} int plugin_dispatch_multivalue */
int plugin_dispatch_notification(const notification_t *notif) {
/* Nobody cares for notifications */
if (list_notification == NULL)
- return (-1);
+ return -1;
le = llist_head(list_notification);
while (le != NULL) {
le = le->next;
}
- return (0);
+ return 0;
} /* int plugin_dispatch_notification */
void plugin_log(int level, const char *format, ...) {
log_level = LOG_DEBUG;
#endif /* COLLECT_DEBUG */
- return (log_level);
+ return log_level;
} /* int parse_log_severity */
int parse_notif_severity(const char *severity) {
(strcmp(severity, "WARN") == 0))
notif_severity = NOTIF_WARNING;
- return (notif_severity);
+ return notif_severity;
} /* int parse_notif_severity */
const data_set_t *plugin_get_ds(const char *name) {
if (data_sets == NULL) {
ERROR("plugin_get_ds: No data sets are defined yet.");
- return (NULL);
+ return NULL;
}
if (c_avl_get(data_sets, name, (void *)&ds) != 0) {
DEBUG("No such dataset registered: %s", name);
- return (NULL);
+ return NULL;
}
- return (ds);
+ return ds;
} /* data_set_t *plugin_get_ds */
static int plugin_notification_meta_add(notification_t *n, const char *name,
if ((n == NULL) || (name == NULL) || (value == NULL)) {
ERROR("plugin_notification_meta_add: A pointer is NULL!");
- return (-1);
+ return -1;
}
meta = calloc(1, sizeof(*meta));
if (meta == NULL) {
ERROR("plugin_notification_meta_add: calloc failed.");
- return (-1);
+ return -1;
}
sstrncpy(meta->name, name, sizeof(meta->name));
if (meta->nm_value.nm_string == NULL) {
ERROR("plugin_notification_meta_add: strdup failed.");
sfree(meta);
- return (-1);
+ return -1;
}
break;
}
default: {
ERROR("plugin_notification_meta_add: Unknown type: %i", type);
sfree(meta);
- return (-1);
+ return -1;
}
} /* switch (type) */
else
tail->next = meta;
- return (0);
+ return 0;
} /* int plugin_notification_meta_add */
int plugin_notification_meta_add_string(notification_t *n, const char *name,
const char *value) {
- return (plugin_notification_meta_add(n, name, NM_TYPE_STRING, value));
+ return plugin_notification_meta_add(n, name, NM_TYPE_STRING, value);
}
int plugin_notification_meta_add_signed_int(notification_t *n, const char *name,
int64_t value) {
- return (plugin_notification_meta_add(n, name, NM_TYPE_SIGNED_INT, &value));
+ return plugin_notification_meta_add(n, name, NM_TYPE_SIGNED_INT, &value);
}
int plugin_notification_meta_add_unsigned_int(notification_t *n,
const char *name,
uint64_t value) {
- return (plugin_notification_meta_add(n, name, NM_TYPE_UNSIGNED_INT, &value));
+ return plugin_notification_meta_add(n, name, NM_TYPE_UNSIGNED_INT, &value);
}
int plugin_notification_meta_add_double(notification_t *n, const char *name,
double value) {
- return (plugin_notification_meta_add(n, name, NM_TYPE_DOUBLE, &value));
+ return plugin_notification_meta_add(n, name, NM_TYPE_DOUBLE, &value);
}
int plugin_notification_meta_add_boolean(notification_t *n, const char *name,
_Bool value) {
- return (plugin_notification_meta_add(n, name, NM_TYPE_BOOLEAN, &value));
+ return plugin_notification_meta_add(n, name, NM_TYPE_BOOLEAN, &value);
}
int plugin_notification_meta_copy(notification_t *dst,
meta->nm_value.nm_boolean);
}
- return (0);
+ return 0;
} /* int plugin_notification_meta_copy */
int plugin_notification_meta_free(notification_meta_t *n) {
if (n == NULL) {
ERROR("plugin_notification_meta_free: n == NULL!");
- return (-1);
+ return -1;
}
this = n;
this = next;
}
- return (0);
+ return 0;
} /* int plugin_notification_meta_free */
static void plugin_ctx_destructor(void *ctx) {
assert(plugin_ctx_key_initialized);
pthread_setspecific(plugin_ctx_key, ctx);
DEBUG("Created new plugin context.");
- return (ctx);
+ return ctx;
} /* int plugin_ctx_create */
void plugin_init_ctx(void) {
return ctx_init;
}
- return (*ctx);
+ return *ctx;
} /* plugin_ctx_t plugin_get_ctx */
plugin_ctx_t plugin_set_ctx(plugin_ctx_t ctx) {
old = *c;
*c = ctx;
- return (old);
+ return old;
} /* void plugin_set_ctx */
cdtime_t plugin_get_interval(void) {
} /* void *plugin_thread_start */
int plugin_thread_create(pthread_t *thread, const pthread_attr_t *attr,
- void *(*start_routine)(void *), void *arg) {
+ void *(*start_routine)(void *), void *arg,
+ char const *name) {
plugin_thread_t *plugin_thread;
plugin_thread = malloc(sizeof(*plugin_thread));
if (plugin_thread == NULL)
- return -1;
+ return ENOMEM;
plugin_thread->ctx = plugin_get_ctx();
plugin_thread->start_routine = start_routine;
plugin_thread->arg = arg;
- return pthread_create(thread, attr, plugin_thread_start, plugin_thread);
-} /* int plugin_thread_create */
+ int ret = pthread_create(thread, attr, plugin_thread_start, plugin_thread);
+ if (ret != 0) {
+ sfree(plugin_thread);
+ return ret;
+ }
+
+ if (name != NULL)
+ set_thread_name(*thread, name);
-/* vim: set sw=8 ts=8 noet fdm=marker : */
+ return 0;
+} /* int plugin_thread_create */
projects / collectd.git / blobdiff