X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fplugin.c;h=24b269865da47cd74a1ea631b3ecd1c8c5eabb5f;hb=9e7b3a035836474cb4af253b248de30d5eb97f4d;hp=4e35bf2ba56e584dd908e53f2ba12681ca285d8a;hpb=9c8848b1b4bc307b18eabc22f913313fecdd36af;p=collectd.git diff --git a/src/plugin.c b/src/plugin.c index 4e35bf2b..91c40b6b 100644 --- a/src/plugin.c +++ b/src/plugin.c @@ -1,6 +1,6 @@ /** * collectd - src/plugin.c - * Copyright (C) 2005-2009 Florian octo Forster + * Copyright (C) 2005-2011 Florian octo Forster * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the @@ -16,7 +16,7 @@ * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA * * Authors: - * Florian octo Forster + * Florian octo Forster * Sebastian Harl **/ @@ -34,8 +34,8 @@ #include "configfile.h" #include "utils_avltree.h" #include "utils_llist.h" +#include "utils_heap.h" #include "utils_cache.h" -#include "utils_threshold.h" #include "filter_chain.h" /* @@ -50,6 +50,7 @@ typedef struct callback_func_s callback_func_t; #define RF_SIMPLE 0 #define RF_COMPLEX 1 +#define RF_REMOVE 65535 struct read_func_s { /* `read_func_t' "inherits" from `callback_func_t'. @@ -57,10 +58,12 @@ struct read_func_s #define rf_callback rf_super.cf_callback #define rf_udata rf_super.cf_udata callback_func_t rf_super; + char rf_group[DATA_MAX_NAME_LEN]; + char rf_name[DATA_MAX_NAME_LEN]; int rf_type; - int rf_wait_time; - int rf_wait_left; - enum { DONE = 0, TODO = 1, ACTIVE = 2 } rf_needs_read; + struct timespec rf_interval; + struct timespec rf_effective_interval; + struct timespec rf_next_read; }; typedef struct read_func_s read_func_t; @@ -68,9 +71,9 @@ typedef struct read_func_s read_func_t; * Private variables */ static llist_t *list_init; -static llist_t *list_read; static llist_t *list_write; static llist_t *list_flush; +static llist_t *list_missing; static llist_t *list_shutdown; static llist_t *list_log; static llist_t *list_notification; @@ -82,6 +85,8 @@ static c_avl_tree_t *data_sets; static char *plugindir = NULL; +static c_heap_t *read_heap = NULL; +static llist_t *read_list; static int read_loop = 1; static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER; @@ -138,6 +143,26 @@ static void destroy_all_callbacks (llist_t **list) /* {{{ */ *list = NULL; } /* }}} void destroy_all_callbacks */ +static void destroy_read_heap (void) /* {{{ */ +{ + if (read_heap == NULL) + return; + + while (42) + { + callback_func_t *cf; + + cf = c_heap_get_root (read_heap); + if (cf == NULL) + break; + + destroy_callback (cf); + } + + c_heap_destroy (read_heap); + read_heap = NULL; +} /* }}} void destroy_read_heap */ + static int register_callback (llist_t **list, /* {{{ */ const char *name, callback_func_t *cf) { @@ -149,7 +174,7 @@ static int register_callback (llist_t **list, /* {{{ */ *list = llist_create (); if (*list == NULL) { - ERROR ("plugin: create_register_callback: " + ERROR ("plugin: register_callback: " "llist_create failed."); destroy_callback (cf); return (-1); @@ -159,7 +184,7 @@ static int register_callback (llist_t **list, /* {{{ */ key = strdup (name); if (key == NULL) { - ERROR ("plugin: create_register_callback: strdup failed."); + ERROR ("plugin: register_callback: strdup failed."); destroy_callback (cf); return (-1); } @@ -170,7 +195,7 @@ static int register_callback (llist_t **list, /* {{{ */ le = llentry_create (key, cf); if (le == NULL) { - ERROR ("plugin: create_register_callback: " + ERROR ("plugin: register_callback: " "llentry_create failed."); free (key); destroy_callback (cf); @@ -186,6 +211,10 @@ static int register_callback (llist_t **list, /* {{{ */ old_cf = le->value; le->value = cf; + WARNING ("plugin: register_callback: " + "a callback named `%s' already exists - " + "overwriting the old entry!", name); + destroy_callback (old_cf); sfree (key); } @@ -196,14 +225,12 @@ static int register_callback (llist_t **list, /* {{{ */ static int create_register_callback (llist_t **list, /* {{{ */ const char *name, void *callback, user_data_t *ud) { - char *key; callback_func_t *cf; cf = (callback_func_t *) malloc (sizeof (*cf)); if (cf == NULL) { ERROR ("plugin: create_register_callback: malloc failed."); - sfree (key); return (-1); } memset (cf, 0, sizeof (*cf)); @@ -248,28 +275,55 @@ static int plugin_unregister (llist_t *list, const char *name) /* {{{ */ * object, but it will bitch about a shared object not having a * ``module_register'' symbol.. */ -static int plugin_load_file (char *file) +static int plugin_load_file (char *file, uint32_t flags) { lt_dlhandle dlh; void (*reg_handle) (void); - DEBUG ("file = %s", file); - lt_dlinit (); lt_dlerror (); /* clear errors */ - if ((dlh = lt_dlopen (file)) == NULL) +#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 + + if (dlh == NULL) { - const char *error = lt_dlerror (); + char errbuf[1024] = ""; + + ssnprintf (errbuf, sizeof (errbuf), + "lt_dlopen (\"%s\") failed: %s. " + "The most common cause for this problem are " + "missing dependencies. Use ldd(1) to check " + "the dependencies of the plugin " + "/ shared object.", + file, lt_dlerror ()); + + ERROR ("%s", errbuf); + /* Make sure this is printed to STDERR in any case, but also + * make sure it's printed only once. */ + if (list_log != NULL) + fprintf (stderr, "ERROR: %s\n", errbuf); - ERROR ("lt_dlopen (%s) failed: %s", file, error); - fprintf (stderr, "lt_dlopen (%s) failed: %s\n", file, error); return (1); } if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL) { - WARNING ("Couldn't find symbol `module_register' in `%s': %s\n", + WARNING ("Couldn't find symbol \"module_register\" in \"%s\": %s\n", file, lt_dlerror ()); lt_dlclose (dlh); return (-1); @@ -280,88 +334,171 @@ static int plugin_load_file (char *file) return (0); } -static void *plugin_read_thread (void __attribute__((unused)) *args) +static _Bool timeout_reached(struct timespec timeout) { - llentry_t *le; - read_func_t *rf; - int status; - int done; - - pthread_mutex_lock (&read_lock); + struct timeval now; + gettimeofday(&now, NULL); + return (now.tv_sec >= timeout.tv_sec && now.tv_usec >= (timeout.tv_nsec / 1000)); +} +static void *plugin_read_thread (void __attribute__((unused)) *args) +{ while (read_loop != 0) { - le = llist_head (list_read); - done = 0; - - while ((read_loop != 0) && (le != NULL)) + read_func_t *rf; + cdtime_t now; + int status; + int rf_type; + int rc; + + /* Get the read function that needs to be read next. */ + rf = c_heap_get_root (read_heap); + if (rf == NULL) { - rf = (read_func_t *) le->value; + struct timespec abstime; - if (rf->rf_needs_read != TODO) - { - le = le->next; - continue; - } + now = cdtime (); - /* We will do this read function */ - rf->rf_needs_read = ACTIVE; + CDTIME_T_TO_TIMESPEC (now + interval_g, &abstime); - DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Handling %s", - (unsigned long int) pthread_self (), le->key); + pthread_mutex_lock (&read_lock); + pthread_cond_timedwait (&read_cond, &read_lock, + &abstime); pthread_mutex_unlock (&read_lock); + continue; + } - if (rf->rf_type == RF_SIMPLE) - { - int (*callback) (void); + if ((rf->rf_interval.tv_sec == 0) && (rf->rf_interval.tv_nsec == 0)) + { + now = cdtime (); - callback = rf->rf_callback; - status = (*callback) (); - } - else - { - plugin_read_cb callback; + CDTIME_T_TO_TIMESPEC (interval_g, &rf->rf_interval); - callback = rf->rf_callback; - status = (*callback) (&rf->rf_udata); - } + rf->rf_effective_interval = rf->rf_interval; + + CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read); + } - done++; + /* sleep until this entry is due, + * using pthread_cond_timedwait */ + pthread_mutex_lock (&read_lock); + /* In pthread_cond_timedwait, spurious wakeups are possible + * (and really happen, at least on NetBSD with > 1 CPU), thus + * we need to re-evaluate the condition every time + * pthread_cond_timedwait returns. */ + rc = 0; + while ((read_loop != 0) + && !timeout_reached(rf->rf_next_read) + && rc == 0) + { + rc = pthread_cond_timedwait (&read_cond, &read_lock, + &rf->rf_next_read); + } - if (status != 0) - { - if (rf->rf_wait_time < interval_g) - rf->rf_wait_time = interval_g; - rf->rf_wait_left = rf->rf_wait_time; - rf->rf_wait_time = rf->rf_wait_time * 2; - if (rf->rf_wait_time > 86400) - rf->rf_wait_time = 86400; - - NOTICE ("read-function of plugin `%s' " - "failed. Will suspend it for %i " - "seconds.", le->key, rf->rf_wait_left); - } - else + /* Must hold `read_lock' when accessing `rf->rf_type'. */ + rf_type = rf->rf_type; + pthread_mutex_unlock (&read_lock); + + /* Check if we're supposed to stop.. This may have interrupted + * the sleep, too. */ + if (read_loop == 0) + { + /* Insert `rf' again, so it can be free'd correctly */ + c_heap_insert (read_heap, rf); + break; + } + + /* The entry has been marked for deletion. The linked list + * entry has already been removed by `plugin_unregister_read'. + * All we have to do here is free the `read_func_t' and + * continue. */ + if (rf_type == RF_REMOVE) + { + DEBUG ("plugin_read_thread: Destroying the `%s' " + "callback.", rf->rf_name); + destroy_callback ((callback_func_t *) rf); + rf = NULL; + continue; + } + + DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + + if (rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + assert (rf_type == RF_COMPLEX); + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + /* If the function signals failure, we will increase the + * intervals in which it will be called. */ + if (status != 0) + { + rf->rf_effective_interval.tv_sec *= 2; + rf->rf_effective_interval.tv_nsec *= 2; + NORMALIZE_TIMESPEC (rf->rf_effective_interval); + + if (rf->rf_effective_interval.tv_sec >= 86400) { - rf->rf_wait_left = 0; - rf->rf_wait_time = interval_g; + rf->rf_effective_interval.tv_sec = 86400; + rf->rf_effective_interval.tv_nsec = 0; } - pthread_mutex_lock (&read_lock); - - rf->rf_needs_read = DONE; - le = le->next; - } /* while (le != NULL) */ + NOTICE ("read-function of plugin `%s' failed. " + "Will suspend it for %i seconds.", + rf->rf_name, + (int) rf->rf_effective_interval.tv_sec); + } + else + { + /* Success: Restore the interval, if it was changed. */ + rf->rf_effective_interval = rf->rf_interval; + } - if ((read_loop != 0) && (done == 0)) + /* update the ``next read due'' field */ + now = cdtime (); + + DEBUG ("plugin_read_thread: Effective interval of the " + "%s plugin is %i.%09i.", + rf->rf_name, + (int) rf->rf_effective_interval.tv_sec, + (int) rf->rf_effective_interval.tv_nsec); + + /* Calculate the next (absolute) time at which this function + * should be called. */ + rf->rf_next_read.tv_sec = rf->rf_next_read.tv_sec + + rf->rf_effective_interval.tv_sec; + rf->rf_next_read.tv_nsec = rf->rf_next_read.tv_nsec + + rf->rf_effective_interval.tv_nsec; + NORMALIZE_TIMESPEC (rf->rf_next_read); + + /* Check, if `rf_next_read' is in the past. */ + if (TIMESPEC_TO_CDTIME_T (&rf->rf_next_read) < now) { - DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Waiting on read_cond.", - (unsigned long int) pthread_self ()); - pthread_cond_wait (&read_cond, &read_lock); + /* `rf_next_read' is in the past. Insert `now' + * so this value doesn't trail off into the + * past too much. */ + CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read); } - } /* while (read_loop) */ - pthread_mutex_unlock (&read_lock); + DEBUG ("plugin_read_thread: Next read of the %s plugin at %i.%09i.", + rf->rf_name, + (int) rf->rf_next_read.tv_sec, + (int) rf->rf_next_read.tv_nsec); + + /* Re-insert this read function into the heap again. */ + c_heap_insert (read_heap, rf); + } /* while (read_loop) */ pthread_exit (NULL); return ((void *) 0); @@ -404,6 +541,8 @@ static void stop_read_threads (void) if (read_threads == NULL) return; + INFO ("collectd: Stopping %i read threads.", read_threads_num); + pthread_mutex_lock (&read_lock); read_loop = 0; DEBUG ("plugin: stop_read_threads: Signalling `read_cond'"); @@ -441,7 +580,7 @@ void plugin_set_dir (const char *dir) } #define BUFSIZE 512 -int plugin_load (const char *type) +int plugin_load (const char *type, uint32_t flags) { DIR *dh; const char *dir; @@ -499,10 +638,11 @@ int plugin_load (const char *type) else if (!S_ISREG (statbuf.st_mode)) { /* don't follow symlinks */ + WARNING ("stat %s: not a regular file", filename); continue; } - if (plugin_load_file (filename) == 0) + if (plugin_load_file (filename, flags) == 0) { /* success */ ret = 0; @@ -546,50 +686,152 @@ int plugin_register_init (const char *name, /* user_data = */ NULL)); } /* plugin_register_init */ +static int plugin_compare_read_func (const void *arg0, const void *arg1) +{ + const read_func_t *rf0; + const read_func_t *rf1; + + rf0 = arg0; + rf1 = arg1; + + if (rf0->rf_next_read.tv_sec < rf1->rf_next_read.tv_sec) + return (-1); + else if (rf0->rf_next_read.tv_sec > rf1->rf_next_read.tv_sec) + return (1); + else if (rf0->rf_next_read.tv_nsec < rf1->rf_next_read.tv_nsec) + return (-1); + else if (rf0->rf_next_read.tv_nsec > rf1->rf_next_read.tv_nsec) + return (1); + else + return (0); +} /* int plugin_compare_read_func */ + +/* Add a read function to both, the heap and a linked list. The linked list if + * used to look-up read functions, especially for the remove function. The heap + * is used to determine which plugin to read next. */ +static int plugin_insert_read (read_func_t *rf) +{ + int status; + llentry_t *le; + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + read_list = llist_create (); + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: read_list failed."); + return (-1); + } + } + + if (read_heap == NULL) + { + read_heap = c_heap_create (plugin_compare_read_func); + if (read_heap == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_create failed."); + return (-1); + } + } + + le = llist_search (read_list, rf->rf_name); + if (le != NULL) + { + pthread_mutex_unlock (&read_lock); + WARNING ("The read function \"%s\" is already registered. " + "Check for duplicate \"LoadPlugin\" lines " + "in your configuration!", + rf->rf_name); + 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); + } + + status = c_heap_insert (read_heap, rf); + if (status != 0) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_insert failed."); + llentry_destroy (le); + return (-1); + } + + /* This does not fail. */ + llist_append (read_list, le); + + pthread_mutex_unlock (&read_lock); + return (0); +} /* int plugin_insert_read */ + int plugin_register_read (const char *name, int (*callback) (void)) { read_func_t *rf; + int status; - rf = (read_func_t *) malloc (sizeof (read_func_t)); + rf = malloc (sizeof (*rf)); if (rf == NULL) { - char errbuf[1024]; - ERROR ("plugin_register_read: malloc failed: %s", - sstrerror (errno, errbuf, sizeof (errbuf))); - return (-1); + ERROR ("plugin_register_read: malloc failed."); + return (ENOMEM); } memset (rf, 0, sizeof (read_func_t)); rf->rf_callback = (void *) callback; rf->rf_udata.data = NULL; rf->rf_udata.free_func = NULL; - rf->rf_wait_time = interval_g; - rf->rf_wait_left = 0; + rf->rf_group[0] = '\0'; + sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); rf->rf_type = RF_SIMPLE; - rf->rf_needs_read = DONE; + rf->rf_interval.tv_sec = 0; + rf->rf_interval.tv_nsec = 0; + rf->rf_effective_interval = rf->rf_interval; + + status = plugin_insert_read (rf); + if (status != 0) + sfree (rf); - return (register_callback (&list_read, name, (callback_func_t *) rf)); + return (status); } /* int plugin_register_read */ -int plugin_register_complex_read (const char *name, - plugin_read_cb callback, user_data_t *user_data) +int plugin_register_complex_read (const char *group, const char *name, + plugin_read_cb callback, + const struct timespec *interval, + user_data_t *user_data) { read_func_t *rf; + int status; - rf = (read_func_t *) malloc (sizeof (read_func_t)); + rf = malloc (sizeof (*rf)); if (rf == NULL) { ERROR ("plugin_register_complex_read: malloc failed."); - return (-1); + return (ENOMEM); } memset (rf, 0, sizeof (read_func_t)); rf->rf_callback = (void *) callback; - rf->rf_wait_time = interval_g; - rf->rf_wait_left = 0; + if (group != NULL) + sstrncpy (rf->rf_group, group, sizeof (rf->rf_group)); + else + rf->rf_group[0] = '\0'; + sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); rf->rf_type = RF_COMPLEX; - rf->rf_needs_read = DONE; + if (interval != NULL) + { + rf->rf_interval = *interval; + } + rf->rf_effective_interval = rf->rf_interval; /* Set user data */ if (user_data == NULL) @@ -602,7 +844,11 @@ int plugin_register_complex_read (const char *name, rf->rf_udata = *user_data; } - return (register_callback (&list_read, name, (callback_func_t *) rf)); + status = plugin_insert_read (rf); + if (status != 0) + sfree (rf); + + return (status); } /* int plugin_register_complex_read */ int plugin_register_write (const char *name, @@ -619,7 +865,14 @@ int plugin_register_flush (const char *name, (void *) callback, ud)); } /* int plugin_register_flush */ -int plugin_register_shutdown (char *name, +int plugin_register_missing (const char *name, + plugin_missing_cb callback, user_data_t *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, @@ -694,10 +947,106 @@ int plugin_unregister_init (const char *name) return (plugin_unregister (list_init, name)); } -int plugin_unregister_read (const char *name) +int plugin_unregister_read (const char *name) /* {{{ */ { - return (plugin_unregister (list_read, name)); -} + llentry_t *le; + read_func_t *rf; + + if (name == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + 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); + } + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + pthread_mutex_unlock (&read_lock); + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read: Marked `%s' for removal.", name); + + return (0); +} /* }}} int plugin_unregister_read */ + +static int compare_read_func_group (llentry_t *e, void *ud) /* {{{ */ +{ + read_func_t *rf = e->value; + char *group = ud; + + return strcmp (rf->rf_group, (const char *)group); +} /* }}} int compare_read_func_group */ + +int plugin_unregister_read_group (const char *group) /* {{{ */ +{ + llentry_t *le; + read_func_t *rf; + + int found = 0; + + if (group == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + return (-ENOENT); + } + + while (42) + { + le = llist_search_custom (read_list, + compare_read_func_group, (void *)group); + + if (le == NULL) + break; + + ++found; + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read_group: " + "Marked `%s' (group `%s') for removal.", + rf->rf_name, group); + } + + pthread_mutex_unlock (&read_lock); + + if (found == 0) + { + WARNING ("plugin_unregister_read_group: No such " + "group of read function: %s", group); + return (-ENOENT); + } + + return (0); +} /* }}} int plugin_unregister_read_group */ int plugin_unregister_write (const char *name) { @@ -709,6 +1058,11 @@ int plugin_unregister_flush (const char *name) return (plugin_unregister (list_flush, name)); } +int plugin_unregister_missing (const char *name) +{ + return (plugin_unregister (list_missing, name)); +} + int plugin_unregister_shutdown (const char *name) { return (plugin_unregister (list_shutdown, name)); @@ -756,7 +1110,7 @@ void plugin_init_all (void) post_cache_chain = fc_chain_get_by_name (chain_name); - if ((list_init == NULL) && (list_read == NULL)) + if ((list_init == NULL) && (read_heap == NULL)) return; /* Calling all init callbacks before checking if read callbacks @@ -789,7 +1143,7 @@ void plugin_init_all (void) } /* Start read-threads */ - if (list_read != NULL) + if (read_heap != NULL) { const char *rt; int num; @@ -800,64 +1154,33 @@ void plugin_init_all (void) } } /* void plugin_init_all */ +/* TODO: Rename this function. */ void plugin_read_all (void) { - llentry_t *le; - read_func_t *rf; - uc_check_timeout (); - if (list_read == NULL) - return; - - pthread_mutex_lock (&read_lock); - - le = llist_head (list_read); - while (le != NULL) - { - rf = (read_func_t *) le->value; - - if (rf->rf_needs_read != DONE) - { - le = le->next; - continue; - } - - if (rf->rf_wait_left > 0) - rf->rf_wait_left -= interval_g; - - if (rf->rf_wait_left <= 0) - { - rf->rf_needs_read = TODO; - } - - le = le->next; - } - - DEBUG ("plugin: plugin_read_all: Signalling `read_cond'"); - pthread_cond_broadcast (&read_cond); - pthread_mutex_unlock (&read_lock); + return; } /* void plugin_read_all */ /* Read function called when the `-T' command line argument is given. */ int plugin_read_all_once (void) { - llentry_t *le; - read_func_t *rf; int status; int return_status = 0; - if (list_read == NULL) + if (read_heap == NULL) { NOTICE ("No read-functions are registered."); return (0); } - for (le = llist_head (list_read); - le != NULL; - le = le->next) + while (42) { - rf = (read_func_t *) le->value; + read_func_t *rf; + + rf = c_heap_get_root (read_heap); + if (rf == NULL) + break; if (rf->rf_type == RF_SIMPLE) { @@ -877,9 +1200,11 @@ int plugin_read_all_once (void) if (status != 0) { NOTICE ("read-function of plugin `%s' failed.", - le->key); + rf->rf_name); return_status = -1; } + + destroy_callback ((void *) rf); } return (return_status); @@ -961,7 +1286,7 @@ int plugin_write (const char *plugin, /* {{{ */ return (status); } /* }}} int plugin_write */ -int plugin_flush (const char *plugin, int timeout, const char *identifier) +int plugin_flush (const char *plugin, cdtime_t timeout, const char *identifier) { llentry_t *le; @@ -998,9 +1323,16 @@ void plugin_shutdown_all (void) stop_read_threads (); destroy_all_callbacks (&list_init); - destroy_all_callbacks (&list_read); - plugin_flush (/* plugin = */ NULL, /* timeout = */ -1, + pthread_mutex_lock (&read_lock); + llist_destroy (read_list); + read_list = NULL; + pthread_mutex_unlock (&read_lock); + + destroy_read_heap (); + + plugin_flush (/* plugin = */ NULL, + /* timeout = */ 0, /* identifier = */ NULL); le = NULL; @@ -1024,25 +1356,84 @@ void plugin_shutdown_all (void) (*callback) (); } - destroy_all_callbacks (&list_write); + /* Write plugins which use the `user_data' pointer usually need the + * same data available to the flush callback. If this is the case, set + * the free_function to NULL when registering the flush callback and to + * the real free function when registering the write callback. This way + * the data isn't freed twice. */ destroy_all_callbacks (&list_flush); + destroy_all_callbacks (&list_missing); + destroy_all_callbacks (&list_write); + destroy_all_callbacks (&list_notification); destroy_all_callbacks (&list_shutdown); destroy_all_callbacks (&list_log); } /* void plugin_shutdown_all */ +int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */ +{ + llentry_t *le; + + if (list_missing == NULL) + return (0); + + le = llist_head (list_missing); + while (le != NULL) + { + callback_func_t *cf; + plugin_missing_cb callback; + int status; + + cf = le->value; + callback = cf->cf_callback; + + status = (*callback) (vl, &cf->cf_udata); + if (status != 0) + { + if (status < 0) + { + ERROR ("plugin_dispatch_missing: Callback function \"%s\" " + "failed with status %i.", + le->key, status); + return (status); + } + else + { + return (0); + } + } + + le = le->next; + } + return (0); +} /* int }}} plugin_dispatch_missing */ + int plugin_dispatch_values (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; - if ((vl == NULL) || (*vl->type == '\0')) { - ERROR ("plugin_dispatch_values: Invalid value list."); + int free_meta_data = 0; + + if ((vl == NULL) || (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); } + /* Free meta data only if the calling function didn't specify any. In + * this case matches and targets may add some and the calling function + * may not expect (and therefore free) that data. */ + if (vl->meta == NULL) + free_meta_data = 1; + if (list_write == NULL) c_complain_once (LOG_WARNING, &no_write_complaint, "plugin_dispatch_values: No write callback has been " @@ -1059,18 +1450,27 @@ int plugin_dispatch_values (value_list_t *vl) if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0) { - INFO ("plugin_dispatch_values: Dataset not found: %s", vl->type); + char ident[6 * DATA_MAX_NAME_LEN]; + + FORMAT_VL (ident, sizeof (ident), vl); + INFO ("plugin_dispatch_values: Dataset not found: %s " + "(from \"%s\"), check your types.db!", + vl->type, ident); return (-1); } if (vl->time == 0) - vl->time = time (NULL); + vl->time = cdtime (); + + if (vl->interval <= 0) + vl->interval = interval_g; - DEBUG ("plugin_dispatch_values: time = %u; interval = %i; " + DEBUG ("plugin_dispatch_values: time = %.3f; interval = %.3f; " "host = %s; " "plugin = %s; plugin_instance = %s; " "type = %s; type_instance = %s;", - (unsigned int) vl->time, vl->interval, + CDTIME_T_TO_DOUBLE (vl->time), + CDTIME_T_TO_DOUBLE (vl->interval), vl->host, vl->plugin, vl->plugin_instance, vl->type, vl->type_instance); @@ -1102,6 +1502,31 @@ int plugin_dispatch_values (value_list_t *vl) 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); @@ -1113,7 +1538,17 @@ int plugin_dispatch_values (value_list_t *vl) 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) + { + free (vl->values); + vl->values = saved_values; + vl->values_len = saved_values_len; + } return (0); + } } /* Update the value cache */ @@ -1133,19 +1568,79 @@ int plugin_dispatch_values (value_list_t *vl) else fc_default_action (ds, vl); + /* Restore the state of the value_list so that plugins don't get + * confused.. */ + if (saved_values != NULL) + { + free (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); } /* int plugin_dispatch_values */ +int plugin_dispatch_values_secure (const value_list_t *vl) +{ + value_list_t vl_copy; + int status; + + if (vl == NULL) + return EINVAL; + + memcpy (&vl_copy, vl, sizeof (vl_copy)); + + /* Write callbacks must not change the values and meta pointers, so we can + * savely skip copying those and make this more efficient. */ + if ((pre_cache_chain == NULL) && (post_cache_chain == NULL)) + return (plugin_dispatch_values (&vl_copy)); + + /* Set pointers to NULL, just to be on the save side. */ + vl_copy.values = NULL; + vl_copy.meta = NULL; + + vl_copy.values = malloc (sizeof (*vl_copy.values) * vl->values_len); + if (vl_copy.values == NULL) + { + ERROR ("plugin_dispatch_values_secure: malloc failed."); + return (ENOMEM); + } + memcpy (vl_copy.values, vl->values, sizeof (*vl_copy.values) * vl->values_len); + + if (vl->meta != NULL) + { + vl_copy.meta = meta_data_clone (vl->meta); + if (vl_copy.meta == NULL) + { + ERROR ("plugin_dispatch_values_secure: meta_data_clone failed."); + free (vl_copy.values); + return (ENOMEM); + } + } /* if (vl->meta) */ + + status = plugin_dispatch_values (&vl_copy); + + meta_data_destroy (vl_copy.meta); + free (vl_copy.values); + + return (status); +} /* int plugin_dispatch_values_secure */ + int plugin_dispatch_notification (const notification_t *notif) { - int (*callback) (const notification_t *); llentry_t *le; /* Possible TODO: Add flap detection here */ DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; " - "time = %u; host = %s;", + "time = %.3f; host = %s;", notif->severity, notif->message, - (unsigned int) notif->time, notif->host); + CDTIME_T_TO_DOUBLE (notif->time), notif->host); /* Nobody cares for notifications */ if (list_notification == NULL) @@ -1154,8 +1649,19 @@ int plugin_dispatch_notification (const notification_t *notif) le = llist_head (list_notification); while (le != NULL) { - callback = (int (*) (const notification_t *)) le->value; - (*callback) (notif); + callback_func_t *cf; + plugin_notification_cb callback; + int status; + + cf = le->value; + callback = cf->cf_callback; + status = (*callback) (notif, &cf->cf_udata); + if (status != 0) + { + WARNING ("plugin_dispatch_notification: Notification " + "callback %s returned %i.", + le->key, status); + } le = le->next; } @@ -1169,9 +1675,6 @@ void plugin_log (int level, const char *format, ...) va_list ap; llentry_t *le; - if (list_log == NULL) - return; - #if !COLLECT_DEBUG if (level >= LOG_DEBUG) return; @@ -1182,6 +1685,12 @@ void plugin_log (int level, const char *format, ...) msg[sizeof (msg) - 1] = '\0'; va_end (ap); + if (list_log == NULL) + { + fprintf (stderr, "%s\n", msg); + return; + } + le = llist_head (list_log); while (le != NULL) { @@ -1265,7 +1774,7 @@ static int plugin_notification_meta_add (notification_t *n, } case NM_TYPE_BOOLEAN: { - meta->nm_value.nm_boolean = *((bool *) value); + meta->nm_value.nm_boolean = *((_Bool *) value); break; } default: @@ -1319,7 +1828,7 @@ int plugin_notification_meta_add_double (notification_t *n, int plugin_notification_meta_add_boolean (notification_t *n, const char *name, - bool value) + _Bool value) { return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value)); }