X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fplugin.c;h=7ccfc4dd4c2d1803cb049f7bca78a8d9e6d723c6;hb=17f4b4ccf2c32190566f592f1dfb953ecbd9b779;hp=547b5eb56da30f590cfbc3dc74c760d162bf5518;hpb=8a9ac11ae8f0f9999f75508b4413fe5c95bfe497;p=collectd.git diff --git a/src/plugin.c b/src/plugin.c index 547b5eb5..7ccfc4dd 100644 --- a/src/plugin.c +++ b/src/plugin.c @@ -21,22 +21,23 @@ **/ #include "collectd.h" -#include "utils_complain.h" - -#include - -#if HAVE_PTHREAD_H -# include -#endif - #include "common.h" #include "plugin.h" #include "configfile.h" +#include "filter_chain.h" #include "utils_avltree.h" +#include "utils_cache.h" +#include "utils_complain.h" #include "utils_llist.h" #include "utils_heap.h" -#include "utils_cache.h" -#include "filter_chain.h" +#include "utils_time.h" +#include "utils_random.h" + +#if HAVE_PTHREAD_H +# include +#endif + +#include /* * Private structures @@ -61,11 +62,11 @@ struct read_func_s #define rf_ctx rf_super.cf_ctx callback_func_t rf_super; char rf_group[DATA_MAX_NAME_LEN]; - char rf_name[DATA_MAX_NAME_LEN]; + char *rf_name; int rf_type; - struct timespec rf_interval; - struct timespec rf_effective_interval; - struct timespec rf_next_read; + cdtime_t rf_interval; + cdtime_t rf_effective_interval; + cdtime_t rf_next_read; }; typedef struct read_func_s read_func_t; @@ -74,12 +75,15 @@ typedef struct write_queue_s write_queue_t; struct write_queue_s { value_list_t *vl; + plugin_ctx_t ctx; write_queue_t *next; }; /* * Private variables */ +static c_avl_tree_t *plugins_loaded = NULL; + static llist_t *list_init; static llist_t *list_write; static llist_t *list_flush; @@ -105,6 +109,7 @@ static int read_threads_num = 0; static write_queue_t *write_queue_head; static write_queue_t *write_queue_tail; +static long write_queue_length = 0; static _Bool write_loop = 1; static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t write_cond = PTHREAD_COND_INITIALIZER; @@ -114,6 +119,10 @@ static size_t write_threads_num = 0; static pthread_key_t plugin_ctx_key; static _Bool plugin_ctx_key_initialized = 0; +static long write_limit_high = 0; +static long write_limit_low = 0; +cdtime_t last_drop_time = 0; + /* * Static functions */ @@ -359,13 +368,6 @@ static int plugin_load_file (char *file, uint32_t flags) return (0); } -static _Bool timeout_reached(struct timespec timeout) -{ - 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) @@ -391,18 +393,15 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) } pthread_mutex_unlock (&read_lock); - if ((rf->rf_interval.tv_sec == 0) && (rf->rf_interval.tv_nsec == 0)) + if (rf->rf_interval == 0) { /* this should not happen, because the interval is set * for each plugin when loading it * XXX: issue a warning? */ - now = cdtime (); - - CDTIME_T_TO_TIMESPEC (plugin_get_interval (), &rf->rf_interval); - + rf->rf_interval = plugin_get_interval (); rf->rf_effective_interval = rf->rf_interval; - CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read); + rf->rf_next_read = cdtime (); } /* sleep until this entry is due, @@ -414,11 +413,15 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) * pthread_cond_timedwait returns. */ rc = 0; while ((read_loop != 0) - && !timeout_reached(rf->rf_next_read) + && (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, - &rf->rf_next_read); + &ts); } /* Must hold `read_lock' when accessing `rf->rf_type'. */ @@ -442,6 +445,7 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) { DEBUG ("plugin_read_thread: Destroying the `%s' " "callback.", rf->rf_name); + sfree (rf->rf_name); destroy_callback ((callback_func_t *) rf); rf = NULL; continue; @@ -474,20 +478,14 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) * 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_effective_interval.tv_sec = 86400; - rf->rf_effective_interval.tv_nsec = 0; - } + rf->rf_effective_interval *= 2; + if (rf->rf_effective_interval > TIME_T_TO_CDTIME_T (86400)) + rf->rf_effective_interval = TIME_T_TO_CDTIME_T (86400); NOTICE ("read-function of plugin `%s' failed. " - "Will suspend it for %i seconds.", + "Will suspend it for %.3f seconds.", rf->rf_name, - (int) rf->rf_effective_interval.tv_sec); + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); } else { @@ -499,32 +497,26 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) now = cdtime (); DEBUG ("plugin_read_thread: Effective interval of the " - "%s plugin is %i.%09i.", + "%s plugin is %.3f seconds.", rf->rf_name, - (int) rf->rf_effective_interval.tv_sec, - (int) rf->rf_effective_interval.tv_nsec); + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); /* 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); + rf->rf_next_read += rf->rf_effective_interval; /* Check, if `rf_next_read' is in the past. */ - if (TIMESPEC_TO_CDTIME_T (&rf->rf_next_read) < now) + if (rf->rf_next_read < now) { /* `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); + rf->rf_next_read = now; } - DEBUG ("plugin_read_thread: Next read of the %s plugin at %i.%09i.", + DEBUG ("plugin_read_thread: Next read of the %s plugin at %.3f.", rf->rf_name, - (int) rf->rf_next_read.tv_sec, - (int) rf->rf_next_read.tv_nsec); + CDTIME_T_TO_DOUBLE (rf->rf_next_read)); /* Re-insert this read function into the heap again. */ c_heap_insert (read_heap, rf); @@ -629,6 +621,31 @@ static value_list_t *plugin_value_list_clone (value_list_t const *vl_orig) /* {{ return (NULL); } + if (vl->time == 0) + vl->time = cdtime (); + + /* Fill in the interval from the thread context, if it is zero. */ + if (vl->interval == 0) + { + plugin_ctx_t ctx = plugin_get_ctx (); + + if (ctx.interval != 0) + vl->interval = ctx.interval; + else + { + char name[6 * DATA_MAX_NAME_LEN]; + FORMAT_VL (name, sizeof (name), vl); + ERROR ("plugin_value_list_clone: Unable to determine " + "interval from context for " + "value list \"%s\". " + "This indicates a broken plugin. " + "Please report this problem to the " + "collectd mailing list or at " + ".", name); + vl->interval = cf_get_default_interval (); + } + } + return (vl); } /* }}} value_list_t *plugin_value_list_clone */ @@ -648,17 +665,24 @@ static int plugin_write_enqueue (value_list_t const *vl) /* {{{ */ return (ENOMEM); } + /* Store context of caller (read plugin); otherwise, it would not be + * available to the write plugins when actually dispatching the + * value-list later on. */ + q->ctx = plugin_get_ctx (); + pthread_mutex_lock (&write_lock); if (write_queue_tail == NULL) { write_queue_head = q; write_queue_tail = q; + write_queue_length = 1; } else { write_queue_tail->next = q; write_queue_tail = q; + write_queue_length += 1; } pthread_cond_signal (&write_cond); @@ -685,11 +709,16 @@ static value_list_t *plugin_write_dequeue (void) /* {{{ */ q = write_queue_head; write_queue_head = q->next; - if (write_queue_head == NULL) + write_queue_length -= 1; + if (write_queue_head == NULL) { write_queue_tail = NULL; + assert(0 == write_queue_length); + } pthread_mutex_unlock (&write_lock); + (void) plugin_set_ctx (q->ctx); + vl = q->vl; sfree (q); return (vl); @@ -777,14 +806,17 @@ static void stop_write_threads (void) /* {{{ */ pthread_mutex_lock (&write_lock); i = 0; - for (q = write_queue_head; q != NULL; q = q->next) + for (q = write_queue_head; q != NULL; ) { + write_queue_t *q1 = q; plugin_value_list_free (q->vl); - sfree (q); + q = q->next; + sfree (q1); i++; } write_queue_head = NULL; write_queue_tail = NULL; + write_queue_length = 0; pthread_mutex_unlock (&write_lock); if (i > 0) @@ -813,8 +845,52 @@ void plugin_set_dir (const char *dir) } } +static _Bool plugin_is_loaded (char const *name) +{ + int status; + + if (plugins_loaded == NULL) + plugins_loaded = c_avl_create ((void *) strcasecmp); + assert (plugins_loaded != NULL); + + status = c_avl_get (plugins_loaded, name, /* ret_value = */ NULL); + return (status == 0); +} + +static int plugin_mark_loaded (char const *name) +{ + char *name_copy; + int status; + + name_copy = strdup (name); + if (name_copy == NULL) + return (ENOMEM); + + status = c_avl_insert (plugins_loaded, + /* key = */ name_copy, /* value = */ NULL); + return (status); +} + +static void plugin_free_loaded () +{ + void *key; + void *value; + + if (plugins_loaded == NULL) + return; + + while (c_avl_pick (plugins_loaded, &key, &value) == 0) + { + sfree (key); + assert (value == NULL); + } + + c_avl_destroy (plugins_loaded); + plugins_loaded = NULL; +} + #define BUFSIZE 512 -int plugin_load (const char *type, uint32_t flags) +int plugin_load (char const *plugin_name, uint32_t flags) { DIR *dh; const char *dir; @@ -826,17 +902,38 @@ int plugin_load (const char *type, uint32_t flags) struct dirent *de; int status; - DEBUG ("type = %s", type); + if (plugin_name == NULL) + return (EINVAL); + + /* Check if plugin is already loaded and don't do anything in this + * case. */ + if (plugin_is_loaded (plugin_name)) + return (0); dir = plugin_get_dir (); ret = 1; + /* + * XXX: Magic at work: + * + * Some of the language bindings, for example the Python and Perl + * plugins, need to be able to export symbols to the scripts they run. + * For this to happen, the "Globals" flag needs to be set. + * Unfortunately, this technical detail is hard to explain to the + * average user and she shouldn't have to worry about this, ideally. + * So in order to save everyone's sanity use a different default for a + * handful of special plugins. --octo + */ + if ((strcasecmp ("perl", plugin_name) == 0) + || (strcasecmp ("python", plugin_name) == 0)) + flags |= PLUGIN_FLAGS_GLOBAL; + /* `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", type); + status = ssnprintf (typename, sizeof (typename), "%s.so", plugin_name); if ((status < 0) || ((size_t) status >= sizeof (typename))) { - WARNING ("snprintf: truncated: `%s.so'", type); + WARNING ("plugin_load: Filename too long: \"%s.so\"", plugin_name); return (-1); } typename_len = strlen (typename); @@ -844,7 +941,7 @@ int plugin_load (const char *type, uint32_t flags) if ((dh = opendir (dir)) == NULL) { char errbuf[1024]; - ERROR ("opendir (%s): %s", dir, + ERROR ("plugin_load: opendir (%s) failed: %s", dir, sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } @@ -858,40 +955,47 @@ int plugin_load (const char *type, uint32_t flags) "%s/%s", dir, de->d_name); if ((status < 0) || ((size_t) status >= sizeof (filename))) { - WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name); + WARNING ("plugin_load: Filename too long: \"%s/%s\"", + dir, de->d_name); continue; } if (lstat (filename, &statbuf) == -1) { char errbuf[1024]; - WARNING ("stat %s: %s", filename, + WARNING ("plugin_load: stat (\"%s\") failed: %s", + filename, sstrerror (errno, errbuf, sizeof (errbuf))); continue; } else if (!S_ISREG (statbuf.st_mode)) { /* don't follow symlinks */ - WARNING ("stat %s: not a regular file", filename); + WARNING ("plugin_load: %s is not a regular file.", + filename); continue; } - if (plugin_load_file (filename, flags) == 0) + status = plugin_load_file (filename, flags); + if (status == 0) { /* success */ + plugin_mark_loaded (plugin_name); ret = 0; break; } else { - fprintf (stderr, "Unable to load plugin %s.\n", type); + ERROR ("plugin_load: Load plugin \"%s\" failed with " + "status %i.", plugin_name, status); } } closedir (dh); - if (filename[0] == '\0') - fprintf (stderr, "Could not find plugin %s.\n", type); + if (filename[0] == 0) + ERROR ("plugin_load: Could not find plugin \"%s\" in %s", + plugin_name, dir); return (ret); } @@ -928,13 +1032,9 @@ static int plugin_compare_read_func (const void *arg0, const void *arg1) 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) + if (rf0->rf_next_read < rf1->rf_next_read) return (-1); - else if (rf0->rf_next_read.tv_nsec > rf1->rf_next_read.tv_nsec) + else if (rf0->rf_next_read > rf1->rf_next_read) return (1); else return (0); @@ -948,6 +1048,9 @@ static int plugin_insert_read (read_func_t *rf) int status; llentry_t *le; + rf->rf_next_read = cdtime (); + rf->rf_effective_interval = rf->rf_interval; + pthread_mutex_lock (&read_lock); if (read_list == NULL) @@ -1009,43 +1112,12 @@ static int plugin_insert_read (read_func_t *rf) return (0); } /* int plugin_insert_read */ -static int read_cb_wrapper (user_data_t *ud) -{ - int (*callback) (void); - - if (ud == NULL) - return -1; - - callback = ud->data; - return callback(); -} /* int read_cb_wrapper */ - int plugin_register_read (const char *name, int (*callback) (void)) { read_func_t *rf; - plugin_ctx_t ctx = plugin_get_ctx (); int status; - if (ctx.interval != 0) { - /* If ctx.interval is not zero (== use the plugin or global - * interval), we need to use the "complex" read callback, - * because only that allows to specify a different interval. - * Wrap the callback using read_cb_wrapper(). */ - struct timespec interval; - user_data_t user_data; - - user_data.data = callback; - user_data.free_func = NULL; - - CDTIME_T_TO_TIMESPEC (ctx.interval, &interval); - return plugin_register_complex_read (/* group = */ NULL, - name, read_cb_wrapper, &interval, &user_data); - } - - DEBUG ("plugin_register_read: default_interval = %.3f", - CDTIME_T_TO_DOUBLE(plugin_get_interval ())); - rf = malloc (sizeof (*rf)); if (rf == NULL) { @@ -1057,13 +1129,11 @@ int plugin_register_read (const char *name, rf->rf_callback = (void *) callback; rf->rf_udata.data = NULL; rf->rf_udata.free_func = NULL; - rf->rf_ctx = ctx; + rf->rf_ctx = plugin_get_ctx (); rf->rf_group[0] = '\0'; - sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); + rf->rf_name = strdup (name); rf->rf_type = RF_SIMPLE; - rf->rf_interval.tv_sec = 0; - rf->rf_interval.tv_nsec = 0; - rf->rf_effective_interval = rf->rf_interval; + rf->rf_interval = plugin_get_interval (); status = plugin_insert_read (rf); if (status != 0) @@ -1078,7 +1148,6 @@ int plugin_register_complex_read (const char *group, const char *name, user_data_t *user_data) { read_func_t *rf; - plugin_ctx_t ctx = plugin_get_ctx (); int status; rf = malloc (sizeof (*rf)); @@ -1094,21 +1163,12 @@ int plugin_register_complex_read (const char *group, const char *name, 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_name = strdup (name); rf->rf_type = RF_COMPLEX; if (interval != NULL) - { - rf->rf_interval = *interval; - } - else if (ctx.interval != 0) - { - CDTIME_T_TO_TIMESPEC (ctx.interval, &rf->rf_interval); - } - rf->rf_effective_interval = rf->rf_interval; - - DEBUG ("plugin_register_read: interval = %i.%09i", - (int) rf->rf_interval.tv_sec, - (int) rf->rf_interval.tv_nsec); + rf->rf_interval = TIMESPEC_TO_CDTIME_T (interval); + else + rf->rf_interval = plugin_get_interval (); /* Set user data */ if (user_data == NULL) @@ -1121,7 +1181,7 @@ int plugin_register_complex_read (const char *group, const char *name, rf->rf_udata = *user_data; } - rf->rf_ctx = ctx; + rf->rf_ctx = plugin_get_ctx (); status = plugin_insert_read (rf); if (status != 0) @@ -1375,7 +1435,8 @@ int plugin_unregister_notification (const char *name) void plugin_init_all (void) { - const char *chain_name; + char const *chain_name; + long write_threads_num; llentry_t *le; int status; @@ -1388,16 +1449,38 @@ void plugin_init_all (void) chain_name = global_option_get ("PostCacheChain"); post_cache_chain = fc_chain_get_by_name (chain_name); + write_limit_high = global_option_get_long ("WriteQueueLimitHigh", + /* default = */ 0); + if (write_limit_high < 0) { - char const *tmp = global_option_get ("WriteThreads"); - int num = atoi (tmp); + ERROR ("WriteQueueLimitHigh must be positive or zero."); + write_limit_high = 0; + } - if (num < 1) - num = 5; + write_limit_low = global_option_get_long ("WriteQueueLimitLow", + /* default = */ write_limit_high / 2); + if (write_limit_low < 0) + { + ERROR ("WriteQueueLimitLow must be positive or zero."); + write_limit_low = write_limit_high / 2; + } + else if (write_limit_low > write_limit_high) + { + ERROR ("WriteQueueLimitLow must not be larger than " + "WriteQueueLimitHigh."); + write_limit_low = write_limit_high; + } - start_write_threads ((size_t) num); + write_threads_num = global_option_get_long ("WriteThreads", + /* default = */ 5); + if (write_threads_num < 1) + { + ERROR ("WriteThreads must be positive."); + write_threads_num = 5; } + start_write_threads ((size_t) write_threads_num); + if ((list_init == NULL) && (read_heap == NULL)) return; @@ -1680,6 +1763,8 @@ void plugin_shutdown_all (void) destroy_all_callbacks (&list_notification); destroy_all_callbacks (&list_shutdown); destroy_all_callbacks (&list_log); + + plugin_free_loaded (); } /* void plugin_shutdown_all */ int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */ @@ -1774,29 +1859,10 @@ static int plugin_dispatch_values_internal (value_list_t *vl) return (-1); } - if (vl->time == 0) - vl->time = cdtime (); - - if (vl->interval <= 0) - { - plugin_ctx_t ctx = plugin_get_ctx (); - - if (ctx.interval != 0) - vl->interval = ctx.interval; - else - { - char name[6 * DATA_MAX_NAME_LEN]; - FORMAT_VL (name, sizeof (name), vl); - ERROR ("plugin_dispatch_values: Unable to determine " - "interval from context for " - "value list \"%s\". " - "This indicates a broken plugin. " - "Please report this problem to the " - "collectd mailing list or at " - ".", name); - vl->interval = cf_get_default_interval (); - } - } + /* 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; " @@ -1919,10 +1985,55 @@ static int plugin_dispatch_values_internal (value_list_t *vl) return (0); } /* int plugin_dispatch_values_internal */ +static _Bool drop_metric(void) { + _Bool drop = 0; + int wq_len = write_queue_length; + /* We store write_queue_length in a local variable because other threads may update write_queue_length. + * Having this in a local variable (like a cache) is better : we do not need a lock */ + + if(wq_len < write_limit_low) return(0); + + if((write_limit_high > 0) && (wq_len > write_limit_low)) { + if(wq_len >= write_limit_high) { + /* if high == low, we come here too */ + drop = 1; + } else { + /* here, high != low */ + long probability_to_drop; + long n; + + probability_to_drop = (wq_len - write_limit_low); + + n = cdrand_range(write_limit_low, write_limit_high); + + /* Let's have X = high - low. + * n is in range [0..X] + * probability_to_drop is in range [1..X[ + * probability_to_drop gets bigger when wq_len gets bigger. + */ + if(n <= probability_to_drop) { + drop = 1; + } + } + } + if(drop) { + cdtime_t now = cdtime(); + if((now - last_drop_time) > TIME_T_TO_CDTIME_T (60)) { + last_drop_time = now; + /* If you want to count dropped metrics, don't forget to add a lock here */ + /* dropped_metrics++; */ + ERROR ("plugin_dispatch_values : Low water mark reached, dropping a metric"); + } + } + return(drop); +} + int plugin_dispatch_values (value_list_t const *vl) { int status; + if(drop_metric ()) return(0); + status = plugin_write_enqueue (vl); if (status != 0) { @@ -2058,6 +2169,12 @@ const data_set_t *plugin_get_ds (const char *name) { data_set_t *ds; + if (data_sets == NULL) + { + ERROR ("plugin_get_ds: No data sets are defined yet."); + return (NULL); + } + if (c_avl_get (data_sets, name, (void *) &ds) != 0) { DEBUG ("No such dataset registered: %s", name);