From: Ruben Kerkhof Date: Thu, 3 Mar 2016 20:32:38 +0000 (+0100) Subject: Merge branch 'collectd-5.4' into collectd-5.5 X-Git-Tag: collectd-5.5.2~41 X-Git-Url: https://git.octo.it/?p=collectd.git;a=commitdiff_plain;h=aedf67e69cbe93a6925fd1004a7c8b990aed4b0a Merge branch 'collectd-5.4' into collectd-5.5 --- aedf67e69cbe93a6925fd1004a7c8b990aed4b0a diff --cc src/daemon/plugin.c index 29716185,00000000..73d01568 mode 100644,000000..100644 --- a/src/daemon/plugin.c +++ b/src/daemon/plugin.c @@@ -1,2732 -1,0 +1,2731 @@@ +/** + * collectd - src/plugin.c + * Copyright (C) 2005-2014 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + * Sebastian Harl + **/ + +#include "collectd.h" +#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_time.h" +#include "utils_random.h" + +#include + +/* + * Private structures + */ +struct callback_func_s +{ + void *cf_callback; + user_data_t cf_udata; + plugin_ctx_t cf_ctx; +}; +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'. + * The `rf_super' member MUST be the first one in this structure! */ +#define rf_callback rf_super.cf_callback +#define rf_udata rf_super.cf_udata +#define rf_ctx rf_super.cf_ctx + callback_func_t rf_super; + char rf_group[DATA_MAX_NAME_LEN]; + char *rf_name; + int rf_type; + cdtime_t rf_interval; + cdtime_t rf_effective_interval; + cdtime_t rf_next_read; +}; +typedef struct read_func_s read_func_t; + +struct write_queue_s; +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; +static llist_t *list_missing; +static llist_t *list_shutdown; +static llist_t *list_log; +static llist_t *list_notification; + +static fc_chain_t *pre_cache_chain = NULL; +static fc_chain_t *post_cache_chain = NULL; + +static c_avl_tree_t *data_sets; + +static char *plugindir = NULL; + +#ifndef DEFAULT_MAX_READ_INTERVAL +# define DEFAULT_MAX_READ_INTERVAL TIME_T_TO_CDTIME_T (86400) +#endif +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; +static pthread_t *read_threads = NULL; +static int read_threads_num = 0; +static cdtime_t max_read_interval = DEFAULT_MAX_READ_INTERVAL; + +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; +static pthread_t *write_threads = NULL; +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; + +static derive_t stats_values_dropped = 0; +static _Bool record_statistics = 0; + +/* + * Static functions + */ +static int plugin_dispatch_values_internal (value_list_t *vl); + +static const char *plugin_get_dir (void) +{ + if (plugindir == NULL) + return (PLUGINDIR); + else + 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; + + /* Initialize `vl' */ + vl.values = values; + vl.values_len = 2; + vl.time = 0; + sstrncpy (vl.host, hostname_g, sizeof (vl.host)); + sstrncpy (vl.plugin, "collectd", sizeof (vl.plugin)); + + vl.type_instance[0] = 0; + vl.values_len = 1; + + /* 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; + 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; + sstrncpy (vl.type, "derive", sizeof (vl.type)); + sstrncpy (vl.type_instance, "dropped", sizeof (vl.type_instance)); + plugin_dispatch_values (&vl); + + /* Cache */ + sstrncpy (vl.plugin_instance, "cache", + sizeof (vl.plugin_instance)); + + /* Cache : Nb entry in cache tree */ + vl.values[0].gauge = (gauge_t) uc_get_size(); + sstrncpy (vl.type, "cache_size", sizeof (vl.type)); + vl.type_instance[0] = 0; + plugin_dispatch_values (&vl); + + return; +} /* }}} void plugin_update_internal_statistics */ + +static void destroy_callback (callback_func_t *cf) /* {{{ */ +{ + if (cf == 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; + } + sfree (cf); +} /* }}} void destroy_callback */ + +static void destroy_all_callbacks (llist_t **list) /* {{{ */ +{ + llentry_t *le; + + if (*list == NULL) + return; + + le = llist_head (*list); + while (le != NULL) + { + llentry_t *le_next; + + le_next = le->next; + + sfree (le->key); + destroy_callback (le->value); + le->value = NULL; + + le = le_next; + } + + llist_destroy (*list); + *list = NULL; +} /* }}} void destroy_all_callbacks */ + +static void destroy_read_heap (void) /* {{{ */ +{ + if (read_heap == NULL) + return; + + while (42) + { + read_func_t *rf; + + rf = c_heap_get_root (read_heap); + if (rf == NULL) + break; + sfree (rf->rf_name); + destroy_callback ((callback_func_t *) rf); + } + + 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) +{ + llentry_t *le; + char *key; + + if (*list == NULL) + { + *list = llist_create (); + if (*list == NULL) + { + ERROR ("plugin: register_callback: " + "llist_create failed."); + destroy_callback (cf); + return (-1); + } + } + + key = strdup (name); + if (key == NULL) + { + ERROR ("plugin: register_callback: strdup failed."); + destroy_callback (cf); + return (-1); + } + + le = llist_search (*list, name); + if (le == NULL) + { + le = llentry_create (key, cf); + if (le == NULL) + { + ERROR ("plugin: register_callback: " + "llentry_create failed."); + free (key); + destroy_callback (cf); + return (-1); + } + + llist_append (*list, le); + } + else + { + callback_func_t *old_cf; + + 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); + } + + return (0); +} /* }}} int register_callback */ + +static void log_list_callbacks (llist_t **list, /* {{{ */ + const char *comment) +{ + char *str; + int len; + llentry_t *le; + int i; + int n; + char **keys; + + n = llist_size(*list); + if (n == 0) + { + INFO("%s [none]", comment); + return; + } + + keys = calloc(n, sizeof(char*)); + + if (keys == NULL) + { + ERROR("%s: failed to allocate memory for list of callbacks", + comment); + + return; + } + + for (le = llist_head (*list), i = 0, len = 0; + le != NULL; + le = le->next, i++) + { + keys[i] = le->key; + len += strlen(le->key) + 6; + } + str = malloc(len + 10); + if (str == NULL) + { + ERROR("%s: failed to allocate memory for list of callbacks", + comment); + } + else + { + *str = '\0'; + strjoin(str, len, keys, n, "', '"); + INFO("%s ['%s']", comment, str); + free(str); + } + free(keys); +} /* }}} void log_list_callbacks */ + +static int create_register_callback (llist_t **list, /* {{{ */ + const char *name, void *callback, user_data_t *ud) +{ + callback_func_t *cf; + + cf = (callback_func_t *) malloc (sizeof (*cf)); + if (cf == NULL) + { + ERROR ("plugin: create_register_callback: malloc failed."); + return (-1); + } + memset (cf, 0, sizeof (*cf)); + + cf->cf_callback = callback; + if (ud == NULL) + { + cf->cf_udata.data = NULL; + cf->cf_udata.free_func = NULL; + } + else + { + cf->cf_udata = *ud; + } + + cf->cf_ctx = plugin_get_ctx (); + + 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); + + e = llist_search (list, name); + if (e == NULL) + return (-1); + + llist_remove (list, e); + + sfree (e->key); + destroy_callback (e->value); + + llentry_destroy (e); + + return (0); +} /* }}} int plugin_unregister */ + +/* + * (Try to) load the shared object `file'. Won't complain if it isn't a shared + * 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; + void (*reg_handle) (void); + + lt_dlinit (); + lt_dlerror (); /* clear errors */ + +#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) + { + 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 ()); + + 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); + + return (1); + } + + if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL) + { + WARNING ("Couldn't find symbol \"module_register\" in \"%s\": %s\n", + file, lt_dlerror ()); + lt_dlclose (dlh); + return (-1); + } + + (*reg_handle) (); + + return (0); +} + +static void *plugin_read_thread (void __attribute__((unused)) *args) +{ + while (read_loop != 0) + { + read_func_t *rf; + plugin_ctx_t old_ctx; + cdtime_t now; + int status; + int rf_type; + int rc; + + /* Get the read function that needs to be read next. + * We don't need to hold "read_lock" for the heap, but we need + * to call c_heap_get_root() and pthread_cond_wait() in the + * same protected block. */ + pthread_mutex_lock (&read_lock); + rf = c_heap_get_root (read_heap); + if (rf == NULL) + { + pthread_cond_wait (&read_cond, &read_lock); + pthread_mutex_unlock (&read_lock); + continue; + } + pthread_mutex_unlock (&read_lock); + + if (rf->rf_interval == 0) + { + /* this should not happen, because the interval is set + * for each plugin when loading it + * XXX: issue a warning? */ + rf->rf_interval = plugin_get_interval (); + rf->rf_effective_interval = rf->rf_interval; + + rf->rf_next_read = cdtime (); + } + + /* 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) + && (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); + } + + /* 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); + sfree (rf->rf_name); + destroy_callback ((callback_func_t *) rf); + rf = NULL; + continue; + } + + DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + + old_ctx = plugin_set_ctx (rf->rf_ctx); + + 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); + } + + plugin_set_ctx (old_ctx); + + /* If the function signals failure, we will increase the + * intervals in which it will be called. */ + if (status != 0) + { + rf->rf_effective_interval *= 2; + if (rf->rf_effective_interval > max_read_interval) + rf->rf_effective_interval = max_read_interval; + + NOTICE ("read-function of plugin `%s' failed. " + "Will suspend it for %.3f seconds.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); + } + else + { + /* Success: Restore the interval, if it was changed. */ + rf->rf_effective_interval = rf->rf_interval; + } + + /* update the ``next read due'' field */ + now = cdtime (); + + DEBUG ("plugin_read_thread: Effective interval of the " + "%s plugin is %.3f seconds.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); + + /* Calculate the next (absolute) time at which this function + * should be called. */ + rf->rf_next_read += rf->rf_effective_interval; + + /* Check, if `rf_next_read' is in the past. */ + 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. */ + rf->rf_next_read = now; + } + + DEBUG ("plugin_read_thread: Next read of the %s plugin at %.3f.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_next_read)); + + /* Re-insert this read function into the heap again. */ + c_heap_insert (read_heap, rf); + } /* while (read_loop) */ + + pthread_exit (NULL); + return ((void *) 0); +} /* void *plugin_read_thread */ + +static void start_read_threads (int num) +{ + int i; + + if (read_threads != NULL) + return; + + read_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); + if (read_threads == NULL) + { + ERROR ("plugin: start_read_threads: calloc failed."); + return; + } + + read_threads_num = 0; + for (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."); + return; + } + } /* for (i) */ +} /* void start_read_threads */ + +static void stop_read_threads (void) +{ + int i; + + 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'"); + pthread_cond_broadcast (&read_cond); + pthread_mutex_unlock (&read_lock); + + for (i = 0; i < read_threads_num; i++) + { + if (pthread_join (read_threads[i], NULL) != 0) + { + ERROR ("plugin: stop_read_threads: pthread_join failed."); + } + read_threads[i] = (pthread_t) 0; + } + sfree (read_threads); + read_threads_num = 0; +} /* void stop_read_threads */ + +static void plugin_value_list_free (value_list_t *vl) /* {{{ */ +{ + if (vl == NULL) + return; + + meta_data_destroy (vl->meta); + sfree (vl->values); + sfree (vl); +} /* }}} void plugin_value_list_free */ + +static value_list_t *plugin_value_list_clone (value_list_t const *vl_orig) /* {{{ */ +{ + value_list_t *vl; + + if (vl_orig == NULL) + return (NULL); + + vl = malloc (sizeof (*vl)); + if (vl == NULL) + return (NULL); + memcpy (vl, vl_orig, sizeof (*vl)); + + vl->values = calloc (vl_orig->values_len, sizeof (*vl->values)); + if (vl->values == NULL) + { + plugin_value_list_free (vl); + 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); + } + + 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 */ + +static int plugin_write_enqueue (value_list_t const *vl) /* {{{ */ +{ + write_queue_t *q; + + q = malloc (sizeof (*q)); + if (q == NULL) + return (ENOMEM); + q->next = NULL; + + q->vl = plugin_value_list_clone (vl); + if (q->vl == NULL) + { + sfree (q); + 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); + pthread_mutex_unlock (&write_lock); + + return (0); +} /* }}} int plugin_write_enqueue */ + +static value_list_t *plugin_write_dequeue (void) /* {{{ */ +{ + write_queue_t *q; + value_list_t *vl; + + pthread_mutex_lock (&write_lock); + + while (write_loop && (write_queue_head == NULL)) + pthread_cond_wait (&write_cond, &write_lock); + + if (write_queue_head == NULL) + { + pthread_mutex_unlock (&write_lock); + return (NULL); + } + + q = write_queue_head; + write_queue_head = q->next; + 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); +} /* }}} value_list_t *plugin_write_dequeue */ + +static void *plugin_write_thread (void __attribute__((unused)) *args) /* {{{ */ +{ + while (write_loop) + { + value_list_t *vl = plugin_write_dequeue (); + if (vl == NULL) + continue; + + plugin_dispatch_values_internal (vl); + + plugin_value_list_free (vl); + } + + pthread_exit (NULL); + return ((void *) 0); +} /* }}} void *plugin_write_thread */ + +static void start_write_threads (size_t num) /* {{{ */ +{ + size_t i; + + if (write_threads != NULL) + return; + + write_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); + if (write_threads == NULL) + { + ERROR ("plugin: start_write_threads: calloc failed."); + return; + } + + write_threads_num = 0; + for (i = 0; i < num; i++) + { + int status; + + 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 " + "with status %i (%s).", status, + sstrerror (status, errbuf, sizeof (errbuf))); + return; + } + + write_threads_num++; + } /* for (i) */ +} /* }}} void start_write_threads */ + +static void stop_write_threads (void) /* {{{ */ +{ + write_queue_t *q; + int i; + + if (write_threads == NULL) + return; + + INFO ("collectd: Stopping %zu write threads.", write_threads_num); + + pthread_mutex_lock (&write_lock); + write_loop = 0; + DEBUG ("plugin: stop_write_threads: Signalling `write_cond'"); + pthread_cond_broadcast (&write_cond); + pthread_mutex_unlock (&write_lock); + + for (i = 0; i < write_threads_num; i++) + { + if (pthread_join (write_threads[i], NULL) != 0) + { + ERROR ("plugin: stop_write_threads: pthread_join failed."); + } + write_threads[i] = (pthread_t) 0; + } + sfree (write_threads); + write_threads_num = 0; + + pthread_mutex_lock (&write_lock); + i = 0; + for (q = write_queue_head; q != NULL; ) + { + write_queue_t *q1 = q; + plugin_value_list_free (q->vl); + 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) + { + WARNING ("plugin: %i value list%s left after shutting down " + "the write threads.", + i, (i == 1) ? " was" : "s were"); + } +} /* }}} void stop_write_threads */ + +/* + * Public functions + */ +void plugin_set_dir (const char *dir) +{ + if (plugindir != NULL) + free (plugindir); + + if (dir == NULL) + plugindir = NULL; + else if ((plugindir = strdup (dir)) == NULL) + { + char errbuf[1024]; + ERROR ("strdup failed: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + } +} + +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 (char const *plugin_name, uint32_t flags) +{ + DIR *dh; + const char *dir; + char filename[BUFSIZE] = ""; + char typename[BUFSIZE]; + int ret; + struct stat statbuf; + struct dirent *de; + int status; + + 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", plugin_name); + if ((status < 0) || ((size_t) status >= sizeof (typename))) + { + WARNING ("plugin_load: Filename too long: \"%s.so\"", plugin_name); + 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); + } + + while ((de = readdir (dh)) != NULL) + { + if (strcasecmp (de->d_name, typename)) + continue; + + status = ssnprintf (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; + } + + if (lstat (filename, &statbuf) == -1) + { + char errbuf[1024]; + 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 ("plugin_load: %s is not a regular file.", + filename); + continue; + } + + status = plugin_load_file (filename, flags); + if (status == 0) + { + /* success */ + plugin_mark_loaded (plugin_name); + ret = 0; + break; + } + else + { + ERROR ("plugin_load: Load plugin \"%s\" failed with " + "status %i.", plugin_name, status); + } + } + + closedir (dh); + + if (filename[0] == 0) + ERROR ("plugin_load: Could not find plugin \"%s\" in %s", + plugin_name, dir); + + return (ret); +} + +/* + * 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) +{ + 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 *)) +{ + 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)); +} /* 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 < rf1->rf_next_read) + return (-1); + else if (rf0->rf_next_read > rf1->rf_next_read) + 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; + + rf->rf_next_read = cdtime (); + rf->rf_effective_interval = rf->rf_interval; + + 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); + + /* Wake up all the read threads. */ + pthread_cond_broadcast (&read_cond); + 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 = malloc (sizeof (*rf)); + if (rf == NULL) + { + 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_ctx = plugin_get_ctx (); + rf->rf_group[0] = '\0'; + rf->rf_name = strdup (name); + rf->rf_type = RF_SIMPLE; + rf->rf_interval = plugin_get_interval (); + + status = plugin_insert_read (rf); + if (status != 0) { + sfree (rf->rf_name); + sfree (rf); + } + + return (status); +} /* int plugin_register_read */ + +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 = malloc (sizeof (*rf)); + if (rf == NULL) + { + ERROR ("plugin_register_complex_read: malloc failed."); + return (ENOMEM); + } + + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + if (group != NULL) + sstrncpy (rf->rf_group, group, sizeof (rf->rf_group)); + else + rf->rf_group[0] = '\0'; + rf->rf_name = strdup (name); + rf->rf_type = RF_COMPLEX; + if (interval != NULL) + rf->rf_interval = TIMESPEC_TO_CDTIME_T (interval); + else + rf->rf_interval = plugin_get_interval (); + + /* Set user data */ + if (user_data == NULL) + { + rf->rf_udata.data = NULL; + rf->rf_udata.free_func = NULL; + } + else + { + rf->rf_udata = *user_data; + } + + rf->rf_ctx = plugin_get_ctx (); + + status = plugin_insert_read (rf); + if (status != 0) { + sfree (rf->rf_name); + sfree (rf); + } + + return (status); +} /* int plugin_register_complex_read */ + +int plugin_register_write (const char *name, + plugin_write_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_write, name, + (void *) callback, ud)); +} /* int plugin_register_write */ + +int plugin_register_flush (const char *name, + plugin_flush_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_flush, name, + (void *) callback, ud)); +} /* int plugin_register_flush */ + +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, + (void *) callback, /* user_data = */ NULL)); +} /* int plugin_register_shutdown */ + +static void plugin_free_data_sets (void) +{ + void *key; + void *value; + + if (data_sets == NULL) + return; + + while (c_avl_pick (data_sets, &key, &value) == 0) + { + data_set_t *ds = value; + /* key is a pointer to ds->type */ + + sfree (ds->ds); + sfree (ds); + } + + c_avl_destroy (data_sets); + data_sets = NULL; +} /* void plugin_free_data_sets */ + +int plugin_register_data_set (const data_set_t *ds) +{ + data_set_t *ds_copy; + int i; + + if ((data_sets != NULL) + && (c_avl_get (data_sets, ds->type, NULL) == 0)) + { + NOTICE ("Replacing DS `%s' with another version.", ds->type); + plugin_unregister_data_set (ds->type); + } + else if (data_sets == NULL) + { + data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp); + if (data_sets == NULL) + return (-1); + } + + ds_copy = (data_set_t *) malloc (sizeof (data_set_t)); + if (ds_copy == NULL) + return (-1); + memcpy(ds_copy, ds, sizeof (data_set_t)); + + ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t) + * ds->ds_num); + if (ds_copy->ds == NULL) + { + free (ds_copy); + return (-1); + } + + for (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)); +} /* int plugin_register_data_set */ + +int plugin_register_log (const char *name, + plugin_log_cb callback, user_data_t *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 *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); +} /* int plugin_unregister_config */ + +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) +{ + return (plugin_unregister (list_init, name)); +} + +int plugin_unregister_read (const char *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 */ + +void plugin_log_available_writers (void) +{ + log_list_callbacks (&list_write, "Available write targets:"); +} + +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) +{ + return (plugin_unregister (list_write, name)); +} + +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)); +} + +int plugin_unregister_data_set (const char *name) +{ + data_set_t *ds; + + if (data_sets == NULL) + return (-1); + + if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0) + return (-1); + + sfree (ds->ds); + sfree (ds); + + return (0); +} /* int plugin_unregister_data_set */ + +int plugin_unregister_log (const char *name) +{ + return (plugin_unregister (list_log, name)); +} + +int plugin_unregister_notification (const char *name) +{ + return (plugin_unregister (list_notification, name)); +} + +void plugin_init_all (void) +{ + char const *chain_name; - long write_threads_num; + llentry_t *le; + int status; + + /* Init the value cache */ + uc_init (); + + if (IS_TRUE (global_option_get ("CollectInternalStats"))) + record_statistics = 1; + + chain_name = global_option_get ("PreCacheChain"); + pre_cache_chain = fc_chain_get_by_name (chain_name); + + 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) + { + ERROR ("WriteQueueLimitHigh must be positive or zero."); + write_limit_high = 0; + } + + 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; + } + + write_threads_num = global_option_get_long ("WriteThreads", + /* default = */ 5); + if (write_threads_num < 1) + { + ERROR ("WriteThreads must be positive."); + write_threads_num = 5; + } + + if ((list_init == NULL) && (read_heap == NULL)) + return; + + /* Calling all init callbacks before checking if read callbacks + * are available allows the init callbacks to register the read + * callback. */ + le = llist_head (list_init); + while (le != NULL) + { + callback_func_t *cf; + plugin_init_cb callback; + plugin_ctx_t old_ctx; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + status = (*callback) (); + plugin_set_ctx (old_ctx); + + if (status != 0) + { + ERROR ("Initialization of plugin `%s' " + "failed with status %i. " + "Plugin will be unloaded.", + le->key, status); + /* Plugins that register read callbacks from the init + * callback should take care of appropriate error + * handling themselves. */ + /* FIXME: Unload _all_ functions */ + plugin_unregister_read (le->key); + } + + le = le->next; + } + + start_write_threads ((size_t) write_threads_num); + + max_read_interval = global_option_get_time ("MaxReadInterval", + DEFAULT_MAX_READ_INTERVAL); + + /* Start read-threads */ + if (read_heap != NULL) + { + const char *rt; + int num; + + rt = global_option_get ("ReadThreads"); + num = atoi (rt); + if (num != -1) + start_read_threads ((num > 0) ? num : 5); + } +} /* void plugin_init_all */ + +/* TODO: Rename this function. */ +void plugin_read_all (void) +{ + if(record_statistics) { + plugin_update_internal_statistics (); + } + 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) +{ + int status; + int return_status = 0; + + if (read_heap == NULL) + { + NOTICE ("No read-functions are registered."); + return (0); + } + + while (42) + { + read_func_t *rf; + plugin_ctx_t old_ctx; + + rf = c_heap_get_root (read_heap); + if (rf == NULL) + break; + + old_ctx = plugin_set_ctx (rf->rf_ctx); + + if (rf->rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + plugin_set_ctx (old_ctx); + + if (status != 0) + { + NOTICE ("read-function of plugin `%s' failed.", + rf->rf_name); + return_status = -1; + } + + sfree (rf->rf_name); + destroy_callback ((void *) rf); + } + + return (return_status); +} /* int plugin_read_all_once */ + +int plugin_write (const char *plugin, /* {{{ */ + const data_set_t *ds, const value_list_t *vl) +{ + llentry_t *le; + int status; + + if (vl == NULL) + return (EINVAL); + + if (list_write == NULL) + 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); + } + } + + if (plugin == NULL) + { + int success = 0; + int failure = 0; + + le = llist_head (list_write); + while (le != NULL) + { + callback_func_t *cf = le->value; + plugin_write_cb callback; + + /* do not switch plugin context; rather keep the context (interval) + * information of the calling read plugin */ + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); + if (status != 0) + failure++; + else + success++; + + le = le->next; + } + + if ((success == 0) && (failure != 0)) + status = -1; + else + status = 0; + } + else /* plugin != NULL */ + { + callback_func_t *cf; + plugin_write_cb callback; + + le = llist_head (list_write); + while (le != NULL) + { + if (strcasecmp (plugin, le->key) == 0) + break; + + le = le->next; + } + + if (le == NULL) + return (ENOENT); + + cf = le->value; + + /* do not switch plugin context; rather keep the context (interval) + * information of the calling read plugin */ + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); + } + + 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); + + le = llist_head (list_flush); + while (le != NULL) + { + callback_func_t *cf; + plugin_flush_cb callback; + plugin_ctx_t old_ctx; + + if ((plugin != NULL) + && (strcmp (plugin, le->key) != 0)) + { + le = le->next; + continue; + } + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + (*callback) (timeout, identifier, &cf->cf_udata); + + plugin_set_ctx (old_ctx); + + le = le->next; + } + return (0); +} /* int plugin_flush */ + +void plugin_shutdown_all (void) +{ + llentry_t *le; + + stop_read_threads (); + + destroy_all_callbacks (&list_init); + + 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; + if (list_shutdown != NULL) + le = llist_head (list_shutdown); + + while (le != NULL) + { + callback_func_t *cf; + plugin_shutdown_cb callback; + plugin_ctx_t old_ctx; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + /* Advance the pointer before calling the callback allows + * shutdown functions to unregister themselves. If done the + * other way around the memory `le' points to will be freed + * after callback returns. */ + le = le->next; + + (*callback) (); + + plugin_set_ctx (old_ctx); + } + + stop_write_threads (); + + /* 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); + + plugin_free_loaded (); + plugin_free_data_sets (); +} /* 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; + plugin_ctx_t old_ctx; + int status; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + 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); + } + else + { + return (0); + } + } + + le = le->next; + } + 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; + + 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 " + "registered. Please load at least one output plugin, " + "if you want the collected data to be stored."); + + if (data_sets == NULL) + { + ERROR ("plugin_dispatch_values: No data sets registered. " + "Could the types database be read? Check " + "your `TypesDB' setting!"); + return (-1); + } + + if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0) + { + 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); + } + + /* 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; " + "type = %s; type_instance = %s;", + CDTIME_T_TO_DOUBLE (vl->time), + CDTIME_T_TO_DOUBLE (vl->interval), + vl->host, + vl->plugin, vl->plugin_instance, + vl->type, vl->type_instance); + +#if COLLECT_DEBUG + assert (0 == strcmp (ds->type, vl->type)); +#else + if (0 != strcmp (ds->type, vl->type)) + WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)", + ds->type, vl->type); +#endif + +#if COLLECT_DEBUG + assert (ds->ds_num == vl->values_len); +#else + if (ds->ds_num != vl->values_len) + { + ERROR ("plugin_dispatch_values: ds->type = %s: " + "(ds->ds_num = %i) != " + "(vl->values_len = %i)", + ds->type, ds->ds_num, vl->values_len); + return (-1); + } +#endif + + escape_slashes (vl->host, sizeof (vl->host)); + escape_slashes (vl->plugin, sizeof (vl->plugin)); + escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance)); + 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) + { + WARNING ("plugin_dispatch_values: Running the " + "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) + { + free (vl->values); + vl->values = saved_values; + vl->values_len = saved_values_len; + } + return (0); + } + } + + /* Update the value cache */ + uc_update (ds, vl); + + if (post_cache_chain != NULL) + { + status = fc_process_chain (ds, vl, post_cache_chain); + if (status < 0) + { + WARNING ("plugin_dispatch_values: Running the " + "post-cache chain failed with " + "status %i (%#x).", + status, status); + } + } + 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_internal */ + +static double get_drop_probability (void) /* {{{ */ +{ + long pos; + long size; + long wql; + + pthread_mutex_lock (&write_lock); + wql = write_queue_length; + pthread_mutex_unlock (&write_lock); + + if (wql < write_limit_low) + return (0.0); + if (wql >= write_limit_high) + return (1.0); + + pos = 1 + wql - write_limit_low; + size = 1 + write_limit_high - write_limit_low; + + return (((double) pos) / ((double) size)); +} /* }}} double get_drop_probability */ + +static _Bool check_drop_value (void) /* {{{ */ +{ + static cdtime_t last_message_time = 0; + static pthread_mutex_t last_message_lock = PTHREAD_MUTEX_INITIALIZER; + + double p; + double q; + int status; + + if (write_limit_high == 0) + return (0); + + p = get_drop_probability (); + if (p == 0.0) + return (0); + + status = pthread_mutex_trylock (&last_message_lock); + if (status == 0) + { + cdtime_t now; + + now = cdtime (); + if ((now - last_message_time) > TIME_T_TO_CDTIME_T (1)) + { + last_message_time = now; + ERROR ("plugin_dispatch_values: Low water mark " + "reached. Dropping %.0f%% of metrics.", + 100.0 * p); + } + pthread_mutex_unlock (&last_message_lock); + } + + if (p == 1.0) + return (1); + + q = cdrand_d (); + if (q > p) + return (1); + else + return (0); +} /* }}} _Bool check_drop_value */ + +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) { + pthread_mutex_lock(&statistics_lock); + stats_values_dropped++; + pthread_mutex_unlock(&statistics_lock); + } + return (0); + } + + status = plugin_write_enqueue (vl); + if (status != 0) + { + char errbuf[1024]; + ERROR ("plugin_dispatch_values: plugin_write_enqueue failed " + "with status %i (%s).", status, + sstrerror (status, errbuf, sizeof (errbuf))); + return (status); + } + + return (0); +} + +__attribute__((sentinel)) +int plugin_dispatch_multivalue (value_list_t const *template, /* {{{ */ + _Bool store_percentage, int store_type, ...) +{ + value_list_t *vl; + int failed = 0; + gauge_t sum = 0.0; + va_list ap; + + assert (template->values_len == 1); + + /* Calculate sum for Gauge to calculate percent if needed */ + if (DS_TYPE_GAUGE == store_type) { + va_start (ap, store_type); + while (42) + { + char const *name; + gauge_t value; + + name = va_arg (ap, char const *); + if (name == NULL) + break; + + value = va_arg (ap, gauge_t); + if (!isnan (value)) + sum += value; + } + va_end (ap); + } + + + vl = plugin_value_list_clone (template); + /* plugin_value_list_clone makes sure vl->time is set to non-zero. */ + if (store_percentage) + sstrncpy (vl->type, "percent", sizeof (vl->type)); + + va_start (ap, store_type); + while (42) + { + char const *name; + int status; + + /* Set the type instance. */ + name = va_arg (ap, char const *); + if (name == NULL) + break; + sstrncpy (vl->type_instance, name, sizeof (vl->type_instance)); + + /* Set the value. */ + switch (store_type) + { + case DS_TYPE_GAUGE: + vl->values[0].gauge = va_arg (ap, gauge_t); + if (store_percentage) + vl->values[0].gauge *= sum ? (100.0 / sum) : 0; + break; + case DS_TYPE_ABSOLUTE: + vl->values[0].absolute = va_arg (ap, absolute_t); + break; + case DS_TYPE_COUNTER: + vl->values[0].counter = va_arg (ap, counter_t); + break; + case DS_TYPE_DERIVE: + vl->values[0].derive = va_arg (ap, derive_t); + break; + default: + ERROR ("plugin_dispatch_multivalue: given store_type is incorrect."); + failed++; + } + + + status = plugin_write_enqueue (vl); + if (status != 0) + failed++; + } + va_end (ap); + + plugin_value_list_free (vl); + return (failed); +} /* }}} int plugin_dispatch_multivalue */ + +int plugin_dispatch_notification (const notification_t *notif) +{ + llentry_t *le; + /* Possible TODO: Add flap detection here */ + + DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; " + "time = %.3f; host = %s;", + notif->severity, notif->message, + CDTIME_T_TO_DOUBLE (notif->time), notif->host); + + /* Nobody cares for notifications */ + if (list_notification == NULL) + return (-1); + + le = llist_head (list_notification); + while (le != NULL) + { + callback_func_t *cf; + plugin_notification_cb callback; + int status; + + /* do not switch plugin context; rather keep the context + * (interval) information of the calling plugin */ + + 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; + } + + return (0); +} /* int plugin_dispatch_notification */ + +void plugin_log (int level, const char *format, ...) +{ + char msg[1024]; + va_list ap; + llentry_t *le; + +#if !COLLECT_DEBUG + if (level >= LOG_DEBUG) + return; +#endif + + va_start (ap, format); + vsnprintf (msg, sizeof (msg), format, ap); + 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) + { + callback_func_t *cf; + plugin_log_cb callback; + + cf = le->value; + callback = cf->cf_callback; + + /* do not switch plugin context; rather keep the context + * (interval) information of the calling plugin */ + + (*callback) (level, msg, &cf->cf_udata); + + le = le->next; + } +} /* void plugin_log */ + +int parse_log_severity (const char *severity) +{ + int log_level = -1; + + if ((0 == strcasecmp (severity, "emerg")) + || (0 == strcasecmp (severity, "alert")) + || (0 == strcasecmp (severity, "crit")) + || (0 == strcasecmp (severity, "err"))) + log_level = LOG_ERR; + else if (0 == strcasecmp (severity, "warning")) + log_level = LOG_WARNING; + else if (0 == strcasecmp (severity, "notice")) + log_level = LOG_NOTICE; + else if (0 == strcasecmp (severity, "info")) + log_level = LOG_INFO; +#if COLLECT_DEBUG + else if (0 == strcasecmp (severity, "debug")) + log_level = LOG_DEBUG; +#endif /* COLLECT_DEBUG */ + + return (log_level); +} /* int parse_log_severity */ + +int parse_notif_severity (const char *severity) +{ + int notif_severity = -1; + + if (strcasecmp (severity, "FAILURE") == 0) + notif_severity = NOTIF_FAILURE; + else if (strcmp (severity, "OKAY") == 0) + notif_severity = NOTIF_OKAY; + else if ((strcmp (severity, "WARNING") == 0) + || (strcmp (severity, "WARN") == 0)) + notif_severity = NOTIF_WARNING; + + return (notif_severity); +} /* int parse_notif_severity */ + +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); + return (NULL); + } + + return (ds); +} /* data_set_t *plugin_get_ds */ + +static int plugin_notification_meta_add (notification_t *n, + const char *name, + enum notification_meta_type_e type, + const void *value) +{ + notification_meta_t *meta; + notification_meta_t *tail; + + if ((n == NULL) || (name == NULL) || (value == NULL)) + { + ERROR ("plugin_notification_meta_add: A pointer is NULL!"); + return (-1); + } + + meta = (notification_meta_t *) malloc (sizeof (notification_meta_t)); + if (meta == NULL) + { + ERROR ("plugin_notification_meta_add: malloc failed."); + return (-1); + } + memset (meta, 0, sizeof (notification_meta_t)); + + sstrncpy (meta->name, name, sizeof (meta->name)); + meta->type = type; + + switch (type) + { + case NM_TYPE_STRING: + { + meta->nm_value.nm_string = strdup ((const char *) value); + if (meta->nm_value.nm_string == NULL) + { + ERROR ("plugin_notification_meta_add: strdup failed."); + sfree (meta); + return (-1); + } + break; + } + case NM_TYPE_SIGNED_INT: + { + meta->nm_value.nm_signed_int = *((int64_t *) value); + break; + } + case NM_TYPE_UNSIGNED_INT: + { + meta->nm_value.nm_unsigned_int = *((uint64_t *) value); + break; + } + case NM_TYPE_DOUBLE: + { + meta->nm_value.nm_double = *((double *) value); + break; + } + case NM_TYPE_BOOLEAN: + { + meta->nm_value.nm_boolean = *((_Bool *) value); + break; + } + default: + { + ERROR ("plugin_notification_meta_add: Unknown type: %i", type); + sfree (meta); + return (-1); + } + } /* switch (type) */ + + meta->next = NULL; + tail = n->meta; + while ((tail != NULL) && (tail->next != NULL)) + tail = tail->next; + + if (tail == NULL) + n->meta = meta; + else + tail->next = meta; + + 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)); +} + +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)); +} + +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)); +} + +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)); +} + +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)); +} + +int plugin_notification_meta_copy (notification_t *dst, + const notification_t *src) +{ + notification_meta_t *meta; + + assert (dst != NULL); + assert (src != NULL); + assert (dst != src); + assert ((src->meta == NULL) || (src->meta != dst->meta)); + + for (meta = src->meta; meta != NULL; meta = meta->next) + { + if (meta->type == NM_TYPE_STRING) + plugin_notification_meta_add_string (dst, meta->name, + meta->nm_value.nm_string); + else if (meta->type == NM_TYPE_SIGNED_INT) + plugin_notification_meta_add_signed_int (dst, meta->name, + meta->nm_value.nm_signed_int); + else if (meta->type == NM_TYPE_UNSIGNED_INT) + plugin_notification_meta_add_unsigned_int (dst, meta->name, + meta->nm_value.nm_unsigned_int); + else if (meta->type == NM_TYPE_DOUBLE) + plugin_notification_meta_add_double (dst, meta->name, + meta->nm_value.nm_double); + else if (meta->type == NM_TYPE_BOOLEAN) + plugin_notification_meta_add_boolean (dst, meta->name, + meta->nm_value.nm_boolean); + } + + return (0); +} /* int plugin_notification_meta_copy */ + +int plugin_notification_meta_free (notification_meta_t *n) +{ + notification_meta_t *this; + notification_meta_t *next; + + if (n == NULL) + { + ERROR ("plugin_notification_meta_free: n == NULL!"); + return (-1); + } + + this = n; + while (this != NULL) + { + next = this->next; + + if (this->type == NM_TYPE_STRING) + { + free ((char *)this->nm_value.nm_string); + this->nm_value.nm_string = NULL; + } + sfree (this); + + this = next; + } + + return (0); +} /* int plugin_notification_meta_free */ + +static void plugin_ctx_destructor (void *ctx) +{ + sfree (ctx); +} /* void plugin_ctx_destructor */ + +static plugin_ctx_t ctx_init = { /* interval = */ 0 }; + +static plugin_ctx_t *plugin_ctx_create (void) +{ + plugin_ctx_t *ctx; + + ctx = malloc (sizeof (*ctx)); + if (ctx == NULL) { + char errbuf[1024]; + ERROR ("Failed to allocate plugin context: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return NULL; + } + + *ctx = ctx_init; + assert (plugin_ctx_key_initialized); + pthread_setspecific (plugin_ctx_key, ctx); + DEBUG("Created new plugin context."); + return (ctx); +} /* int plugin_ctx_create */ + +void plugin_init_ctx (void) +{ + pthread_key_create (&plugin_ctx_key, plugin_ctx_destructor); + plugin_ctx_key_initialized = 1; +} /* void plugin_init_ctx */ + +plugin_ctx_t plugin_get_ctx (void) +{ + plugin_ctx_t *ctx; + + assert (plugin_ctx_key_initialized); + ctx = pthread_getspecific (plugin_ctx_key); + + if (ctx == NULL) { + ctx = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (ctx == NULL) + return ctx_init; + } + + return (*ctx); +} /* plugin_ctx_t plugin_get_ctx */ + +plugin_ctx_t plugin_set_ctx (plugin_ctx_t ctx) +{ + plugin_ctx_t *c; + plugin_ctx_t old; + + assert (plugin_ctx_key_initialized); + c = pthread_getspecific (plugin_ctx_key); + + if (c == NULL) { + c = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (c == NULL) + return ctx_init; + } + + old = *c; + *c = ctx; + + return (old); +} /* void plugin_set_ctx */ + +cdtime_t plugin_get_interval (void) +{ + cdtime_t interval; + + interval = plugin_get_ctx().interval; + if (interval > 0) + return interval; + + return cf_get_default_interval (); +} /* cdtime_t plugin_get_interval */ + +typedef struct { + plugin_ctx_t ctx; + void *(*start_routine) (void *); + void *arg; +} plugin_thread_t; + +static void *plugin_thread_start (void *arg) +{ + plugin_thread_t *plugin_thread = arg; + + void *(*start_routine) (void *) = plugin_thread->start_routine; + void *plugin_arg = plugin_thread->arg; + + plugin_set_ctx (plugin_thread->ctx); + + free (plugin_thread); + + return start_routine (plugin_arg); +} /* void *plugin_thread_start */ + +int plugin_thread_create (pthread_t *thread, const pthread_attr_t *attr, + void *(*start_routine) (void *), void *arg) +{ + plugin_thread_t *plugin_thread; + + plugin_thread = malloc (sizeof (*plugin_thread)); + if (plugin_thread == NULL) + return -1; + + 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 */ + +/* vim: set sw=8 ts=8 noet fdm=marker : */ diff --cc src/daemon/utils_cache.c index 8c5db316,00000000..45444c7f mode 100644,000000..100644 --- a/src/daemon/utils_cache.c +++ b/src/daemon/utils_cache.c @@@ -1,998 -1,0 +1,997 @@@ +/** + * collectd - src/utils_cache.c + * Copyright (C) 2007-2010 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + **/ + +#include "collectd.h" +#include "common.h" +#include "plugin.h" +#include "utils_avltree.h" +#include "utils_cache.h" +#include "meta_data.h" + +#include +#include + +typedef struct cache_entry_s +{ + char name[6 * DATA_MAX_NAME_LEN]; + int values_num; + gauge_t *values_gauge; + value_t *values_raw; + /* Time contained in the package + * (for calculating rates) */ + cdtime_t last_time; + /* Time according to the local clock + * (for purging old entries) */ + cdtime_t last_update; + /* Interval in which the data is collected + * (for purding old entries) */ + cdtime_t interval; + int state; + int hits; + + /* + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! 0 ! 1 ! 2 ! 3 ! 4 ! 5 ! 6 ! 7 ! 8 ! ... + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ... + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! t = 0 ! t = 1 ! t = 2 ! ... + * +-----------------+-----------------+-----------------+---- + */ + gauge_t *history; + size_t history_index; /* points to the next position to write to. */ + size_t history_length; + + meta_data_t *meta; +} cache_entry_t; + +static c_avl_tree_t *cache_tree = NULL; +static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; + +static int cache_compare (const cache_entry_t *a, const cache_entry_t *b) +{ +#if COLLECT_DEBUG + assert ((a != NULL) && (b != NULL)); +#endif + return (strcmp (a->name, b->name)); +} /* int cache_compare */ + +static cache_entry_t *cache_alloc (int values_num) +{ + cache_entry_t *ce; + + ce = (cache_entry_t *) malloc (sizeof (cache_entry_t)); + if (ce == NULL) + { + ERROR ("utils_cache: cache_alloc: malloc failed."); + return (NULL); + } + memset (ce, '\0', sizeof (cache_entry_t)); + ce->values_num = values_num; + + ce->values_gauge = calloc (values_num, sizeof (*ce->values_gauge)); + ce->values_raw = calloc (values_num, sizeof (*ce->values_raw)); + if ((ce->values_gauge == NULL) || (ce->values_raw == NULL)) + { + sfree (ce->values_gauge); + sfree (ce->values_raw); + sfree (ce); + ERROR ("utils_cache: cache_alloc: calloc failed."); + return (NULL); + } + + ce->history = NULL; + ce->history_length = 0; + ce->meta = NULL; + + return (ce); +} /* cache_entry_t *cache_alloc */ + +static void cache_free (cache_entry_t *ce) +{ + if (ce == NULL) + return; + + sfree (ce->values_gauge); + sfree (ce->values_raw); + sfree (ce->history); + if (ce->meta != NULL) + { + meta_data_destroy (ce->meta); + ce->meta = NULL; + } + sfree (ce); +} /* void cache_free */ + +static void uc_check_range (const data_set_t *ds, cache_entry_t *ce) +{ + int i; + + for (i = 0; i < ds->ds_num; i++) + { + if (isnan (ce->values_gauge[i])) + continue; + else if (ce->values_gauge[i] < ds->ds[i].min) + ce->values_gauge[i] = NAN; + else if (ce->values_gauge[i] > ds->ds[i].max) + ce->values_gauge[i] = NAN; + } +} /* void uc_check_range */ + +static int uc_insert (const data_set_t *ds, const value_list_t *vl, + const char *key) +{ + int i; + char *key_copy; + cache_entry_t *ce; + + /* `cache_lock' has been locked by `uc_update' */ + + key_copy = strdup (key); + if (key_copy == NULL) + { + ERROR ("uc_insert: strdup failed."); + return (-1); + } + + ce = cache_alloc (ds->ds_num); + if (ce == NULL) + { + sfree (key_copy); + ERROR ("uc_insert: cache_alloc (%i) failed.", ds->ds_num); + return (-1); + } + + sstrncpy (ce->name, key, sizeof (ce->name)); + + for (i = 0; i < ds->ds_num; i++) + { + switch (ds->ds[i].type) + { + case DS_TYPE_COUNTER: + ce->values_gauge[i] = NAN; + ce->values_raw[i].counter = vl->values[i].counter; + break; + + case DS_TYPE_GAUGE: + ce->values_gauge[i] = vl->values[i].gauge; + ce->values_raw[i].gauge = vl->values[i].gauge; + break; + + case DS_TYPE_DERIVE: + ce->values_gauge[i] = NAN; + ce->values_raw[i].derive = vl->values[i].derive; + break; + + case DS_TYPE_ABSOLUTE: + ce->values_gauge[i] = NAN; + if (vl->interval > 0) + ce->values_gauge[i] = ((double) vl->values[i].absolute) + / CDTIME_T_TO_DOUBLE (vl->interval); + ce->values_raw[i].absolute = vl->values[i].absolute; + break; + + default: + /* This shouldn't happen. */ + ERROR ("uc_insert: Don't know how to handle data source type %i.", + ds->ds[i].type); + sfree (key_copy); + cache_free (ce); + return (-1); + } /* switch (ds->ds[i].type) */ + } /* for (i) */ + + /* Prune invalid gauge data */ + uc_check_range (ds, ce); + + ce->last_time = vl->time; + ce->last_update = cdtime (); + ce->interval = vl->interval; + ce->state = STATE_OKAY; + + if (c_avl_insert (cache_tree, key_copy, ce) != 0) + { + sfree (key_copy); + ERROR ("uc_insert: c_avl_insert failed."); + return (-1); + } + + DEBUG ("uc_insert: Added %s to the cache.", key); + return (0); +} /* int uc_insert */ + +int uc_init (void) +{ + if (cache_tree == NULL) + cache_tree = c_avl_create ((int (*) (const void *, const void *)) + cache_compare); + + return (0); +} /* int uc_init */ + +int uc_check_timeout (void) +{ + cdtime_t now; + cache_entry_t *ce; + + char **keys = NULL; + cdtime_t *keys_time = NULL; + cdtime_t *keys_interval = NULL; + int keys_len = 0; + + char *key; + c_avl_iterator_t *iter; + + int status; + int i; + + pthread_mutex_lock (&cache_lock); + + now = cdtime (); + + /* Build a list of entries to be flushed */ + iter = c_avl_get_iterator (cache_tree); + while (c_avl_iterator_next (iter, (void *) &key, (void *) &ce) == 0) + { + char **tmp; + cdtime_t *tmp_time; + + /* If the entry is fresh enough, continue. */ + if ((now - ce->last_update) < (ce->interval * timeout_g)) + continue; + + /* If entry has not been updated, add to `keys' array */ + tmp = (char **) realloc ((void *) keys, + (keys_len + 1) * sizeof (char *)); + if (tmp == NULL) + { + ERROR ("uc_check_timeout: realloc failed."); + continue; + } + keys = tmp; + + tmp_time = realloc (keys_time, (keys_len + 1) * sizeof (*keys_time)); + if (tmp_time == NULL) + { + ERROR ("uc_check_timeout: realloc failed."); + continue; + } + keys_time = tmp_time; + + tmp_time = realloc (keys_interval, (keys_len + 1) * sizeof (*keys_interval)); + if (tmp_time == NULL) + { + ERROR ("uc_check_timeout: realloc failed."); + continue; + } + keys_interval = tmp_time; + + keys[keys_len] = strdup (key); + if (keys[keys_len] == NULL) + { + ERROR ("uc_check_timeout: strdup failed."); + continue; + } + keys_time[keys_len] = ce->last_time; + keys_interval[keys_len] = ce->interval; + + keys_len++; + } /* while (c_avl_iterator_next) */ + + c_avl_iterator_destroy (iter); + pthread_mutex_unlock (&cache_lock); + + if (keys_len == 0) + { + /* realloc() may have been called for these. */ + sfree (keys); + sfree (keys_time); + sfree (keys_interval); + return (0); + } + + /* Call the "missing" callback for each value. Do this before removing the + * value from the cache, so that callbacks can still access the data stored, + * including plugin specific meta data, rates, history, …. This must be done + * without holding the lock, otherwise we will run into a deadlock if a + * plugin calls the cache interface. */ + for (i = 0; i < keys_len; i++) + { + value_list_t vl = VALUE_LIST_INIT; + + vl.values = NULL; + vl.values_len = 0; + vl.meta = NULL; + + status = parse_identifier_vl (keys[i], &vl); + if (status != 0) + { + ERROR ("uc_check_timeout: parse_identifier_vl (\"%s\") failed.", keys[i]); + continue; + } + + vl.time = keys_time[i]; + vl.interval = keys_interval[i]; + + plugin_dispatch_missing (&vl); + } /* for (i = 0; i < keys_len; i++) */ + + /* Now actually remove all the values from the cache. We don't re-evaluate + * the timestamp again, so in theory it is possible we remove a value after + * it is updated here. */ + pthread_mutex_lock (&cache_lock); + for (i = 0; i < keys_len; i++) + { + key = NULL; + ce = NULL; + + status = c_avl_remove (cache_tree, keys[i], + (void *) &key, (void *) &ce); + if (status != 0) + { + ERROR ("uc_check_timeout: c_avl_remove (\"%s\") failed.", keys[i]); + sfree (keys[i]); + continue; + } + + sfree (keys[i]); + sfree (key); + cache_free (ce); + } /* for (i = 0; i < keys_len; i++) */ + pthread_mutex_unlock (&cache_lock); + + sfree (keys); + sfree (keys_time); + sfree (keys_interval); + + return (0); +} /* int uc_check_timeout */ + +int uc_update (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int status; + int i; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_update: FORMAT_VL failed."); + return (-1); + } + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) /* entry does not yet exist */ + { + status = uc_insert (ds, vl, name); + pthread_mutex_unlock (&cache_lock); + return (status); + } + + assert (ce != NULL); + assert (ce->values_num == ds->ds_num); + + if (ce->last_time >= vl->time) + { + pthread_mutex_unlock (&cache_lock); + NOTICE ("uc_update: Value too old: name = %s; value time = %.3f; " + "last cache update = %.3f;", + name, + CDTIME_T_TO_DOUBLE (vl->time), + CDTIME_T_TO_DOUBLE (ce->last_time)); + return (-1); + } + + for (i = 0; i < ds->ds_num; i++) + { + switch (ds->ds[i].type) + { + case DS_TYPE_COUNTER: + { + counter_t diff; + + /* check if the counter has wrapped around */ + if (vl->values[i].counter < ce->values_raw[i].counter) + { + if (ce->values_raw[i].counter <= 4294967295U) + diff = (4294967295U - ce->values_raw[i].counter) + + vl->values[i].counter; + else + diff = (18446744073709551615ULL - ce->values_raw[i].counter) + + vl->values[i].counter; + } + else /* counter has NOT wrapped around */ + { + diff = vl->values[i].counter - ce->values_raw[i].counter; + } + + ce->values_gauge[i] = ((double) diff) + / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time)); + ce->values_raw[i].counter = vl->values[i].counter; + } + break; + + case DS_TYPE_GAUGE: + ce->values_raw[i].gauge = vl->values[i].gauge; + ce->values_gauge[i] = vl->values[i].gauge; + break; + + case DS_TYPE_DERIVE: + { + derive_t diff; + + diff = vl->values[i].derive - ce->values_raw[i].derive; + + ce->values_gauge[i] = ((double) diff) + / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time)); + ce->values_raw[i].derive = vl->values[i].derive; + } + break; + + case DS_TYPE_ABSOLUTE: + ce->values_gauge[i] = ((double) vl->values[i].absolute) + / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time)); + ce->values_raw[i].absolute = vl->values[i].absolute; + break; + + default: + /* This shouldn't happen. */ + pthread_mutex_unlock (&cache_lock); + ERROR ("uc_update: Don't know how to handle data source type %i.", + ds->ds[i].type); + return (-1); + } /* switch (ds->ds[i].type) */ + + DEBUG ("uc_update: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]); + } /* for (i) */ + + /* Update the history if it exists. */ + if (ce->history != NULL) + { + assert (ce->history_index < ce->history_length); + for (i = 0; i < ce->values_num; i++) + { + size_t hist_idx = (ce->values_num * ce->history_index) + i; + ce->history[hist_idx] = ce->values_gauge[i]; + } + + assert (ce->history_length > 0); + ce->history_index = (ce->history_index + 1) % ce->history_length; + } + + /* Prune invalid gauge data */ + uc_check_range (ds, ce); + + ce->last_time = vl->time; + ce->last_update = cdtime (); + ce->interval = vl->interval; + + pthread_mutex_unlock (&cache_lock); + + return (0); +} /* int uc_update */ + +int uc_get_rate_by_name (const char *name, gauge_t **ret_values, size_t *ret_values_num) +{ + gauge_t *ret = NULL; + size_t ret_num = 0; + cache_entry_t *ce = NULL; + int status = 0; + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + + /* remove missing values from getval */ + if (ce->state == STATE_MISSING) + { + status = -1; + } + else + { + ret_num = ce->values_num; + ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t)); + if (ret == NULL) + { + ERROR ("utils_cache: uc_get_rate_by_name: malloc failed."); + status = -1; + } + else + { + memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t)); + } + } + } + else + { + DEBUG ("utils_cache: uc_get_rate_by_name: No such value: %s", name); + status = -1; + } + + pthread_mutex_unlock (&cache_lock); + + if (status == 0) + { + *ret_values = ret; + *ret_values_num = ret_num; + } + + return (status); +} /* gauge_t *uc_get_rate_by_name */ + +gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + gauge_t *ret = NULL; + size_t ret_num = 0; + int status; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("utils_cache: uc_get_rate: FORMAT_VL failed."); + return (NULL); + } + + status = uc_get_rate_by_name (name, &ret, &ret_num); + if (status != 0) + return (NULL); + + /* This is important - the caller has no other way of knowing how many + * values are returned. */ + if (ret_num != (size_t) ds->ds_num) + { + ERROR ("utils_cache: uc_get_rate: ds[%s] has %i values, " + "but uc_get_rate_by_name returned %zu.", + ds->type, ds->ds_num, ret_num); + sfree (ret); + return (NULL); + } + + return (ret); +} /* gauge_t *uc_get_rate */ + +size_t uc_get_size (void) { + size_t size_arrays = 0; + + pthread_mutex_lock (&cache_lock); + size_arrays = (size_t) c_avl_size (cache_tree); + pthread_mutex_unlock (&cache_lock); + + return (size_arrays); +} + +int uc_get_names (char ***ret_names, cdtime_t **ret_times, size_t *ret_number) +{ + c_avl_iterator_t *iter; + char *key; + cache_entry_t *value; + + char **names = NULL; + cdtime_t *times = NULL; + size_t number = 0; + size_t size_arrays = 0; + + int status = 0; + + if ((ret_names == NULL) || (ret_number == NULL)) + return (-1); + + pthread_mutex_lock (&cache_lock); + + size_arrays = (size_t) c_avl_size (cache_tree); + if (size_arrays < 1) + { + /* Handle the "no values" case here, to avoid the error message when + * calloc() returns NULL. */ + pthread_mutex_unlock (&cache_lock); + return (0); + } + + names = calloc (size_arrays, sizeof (*names)); + times = calloc (size_arrays, sizeof (*times)); + if ((names == NULL) || (times == NULL)) + { + ERROR ("uc_get_names: calloc failed."); + sfree (names); + sfree (times); + pthread_mutex_unlock (&cache_lock); + return (ENOMEM); + } + + iter = c_avl_get_iterator (cache_tree); + while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0) + { + /* remove missing values when list values */ + if (value->state == STATE_MISSING) + continue; + + /* c_avl_size does not return a number smaller than the number of elements + * returned by c_avl_iterator_next. */ + assert (number < size_arrays); + + if (ret_times != NULL) + times[number] = value->last_time; + + names[number] = strdup (key); + if (names[number] == NULL) + { + status = -1; + break; + } + + number++; + } /* while (c_avl_iterator_next) */ + + c_avl_iterator_destroy (iter); + pthread_mutex_unlock (&cache_lock); + + if (status != 0) + { + size_t i; + + for (i = 0; i < number; i++) + { + sfree (names[i]); + } + sfree (names); + sfree (times); + + return (-1); + } + + *ret_names = names; + if (ret_times != NULL) + *ret_times = times; + else + sfree (times); + *ret_number = number; + + return (0); +} /* int uc_get_names */ + +int uc_get_state (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = STATE_ERROR; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->state; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_get_state */ + +int uc_set_state (const data_set_t *ds, const value_list_t *vl, int state) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->state; + ce->state = state; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_set_state */ + +int uc_get_history_by_name (const char *name, + gauge_t *ret_history, size_t num_steps, size_t num_ds) +{ + cache_entry_t *ce = NULL; + size_t i; + int status = 0; + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + return (-ENOENT); + } + + if (((size_t) ce->values_num) != num_ds) + { + pthread_mutex_unlock (&cache_lock); + return (-EINVAL); + } + + /* Check if there are enough values available. If not, increase the buffer + * size. */ + if (ce->history_length < num_steps) + { + gauge_t *tmp; - size_t i; + + tmp = realloc (ce->history, sizeof (*ce->history) + * num_steps * ce->values_num); + if (tmp == NULL) + { + pthread_mutex_unlock (&cache_lock); + return (-ENOMEM); + } + + for (i = ce->history_length * ce->values_num; + i < (num_steps * ce->values_num); + i++) + tmp[i] = NAN; + + ce->history = tmp; + ce->history_length = num_steps; + } /* if (ce->history_length < num_steps) */ + + /* Copy the values to the output buffer. */ + for (i = 0; i < num_steps; i++) + { + size_t src_index; + size_t dst_index; + + if (i < ce->history_index) + src_index = ce->history_index - (i + 1); + else + src_index = ce->history_length + ce->history_index - (i + 1); + src_index = src_index * num_ds; + + dst_index = i * num_ds; + + memcpy (ret_history + dst_index, ce->history + src_index, + sizeof (*ret_history) * num_ds); + } + + pthread_mutex_unlock (&cache_lock); + + return (0); +} /* int uc_get_history_by_name */ + +int uc_get_history (const data_set_t *ds, const value_list_t *vl, + gauge_t *ret_history, size_t num_steps, size_t num_ds) +{ + char name[6 * DATA_MAX_NAME_LEN]; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("utils_cache: uc_get_history: FORMAT_VL failed."); + return (-1); + } + + return (uc_get_history_by_name (name, ret_history, num_steps, num_ds)); +} /* int uc_get_history */ + +int uc_get_hits (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = STATE_ERROR; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_get_hits */ + +int uc_set_hits (const data_set_t *ds, const value_list_t *vl, int hits) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + ce->hits = hits; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_set_hits */ + +int uc_inc_hits (const data_set_t *ds, const value_list_t *vl, int step) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + ce->hits = ret + step; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_inc_hits */ + +/* + * Meta data interface + */ +/* XXX: This function will acquire `cache_lock' but will not free it! */ +static meta_data_t *uc_get_meta (const value_list_t *vl) /* {{{ */ +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int status; + + status = FORMAT_VL (name, sizeof (name), vl); + if (status != 0) + { + ERROR ("utils_cache: uc_get_meta: FORMAT_VL failed."); + return (NULL); + } + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + return (NULL); + } + assert (ce != NULL); + + if (ce->meta == NULL) + ce->meta = meta_data_create (); + + if (ce->meta == NULL) + pthread_mutex_unlock (&cache_lock); + + return (ce->meta); +} /* }}} meta_data_t *uc_get_meta */ + +/* Sorry about this preprocessor magic, but it really makes this file much + * shorter.. */ +#define UC_WRAP(wrap_function) { \ + meta_data_t *meta; \ + int status; \ + meta = uc_get_meta (vl); \ + if (meta == NULL) return (-1); \ + status = wrap_function (meta, key); \ + pthread_mutex_unlock (&cache_lock); \ + return (status); \ +} +int uc_meta_data_exists (const value_list_t *vl, const char *key) + UC_WRAP (meta_data_exists) + +int uc_meta_data_delete (const value_list_t *vl, const char *key) + UC_WRAP (meta_data_delete) +#undef UC_WRAP + +/* We need a new version of this macro because the following functions take + * two argumetns. */ +#define UC_WRAP(wrap_function) { \ + meta_data_t *meta; \ + int status; \ + meta = uc_get_meta (vl); \ + if (meta == NULL) return (-1); \ + status = wrap_function (meta, key, value); \ + pthread_mutex_unlock (&cache_lock); \ + return (status); \ +} +int uc_meta_data_add_string (const value_list_t *vl, + const char *key, + const char *value) + UC_WRAP(meta_data_add_string) +int uc_meta_data_add_signed_int (const value_list_t *vl, + const char *key, + int64_t value) + UC_WRAP(meta_data_add_signed_int) +int uc_meta_data_add_unsigned_int (const value_list_t *vl, + const char *key, + uint64_t value) + UC_WRAP(meta_data_add_unsigned_int) +int uc_meta_data_add_double (const value_list_t *vl, + const char *key, + double value) + UC_WRAP(meta_data_add_double) +int uc_meta_data_add_boolean (const value_list_t *vl, + const char *key, + _Bool value) + UC_WRAP(meta_data_add_boolean) + +int uc_meta_data_get_string (const value_list_t *vl, + const char *key, + char **value) + UC_WRAP(meta_data_get_string) +int uc_meta_data_get_signed_int (const value_list_t *vl, + const char *key, + int64_t *value) + UC_WRAP(meta_data_get_signed_int) +int uc_meta_data_get_unsigned_int (const value_list_t *vl, + const char *key, + uint64_t *value) + UC_WRAP(meta_data_get_unsigned_int) +int uc_meta_data_get_double (const value_list_t *vl, + const char *key, + double *value) + UC_WRAP(meta_data_get_double) +int uc_meta_data_get_boolean (const value_list_t *vl, + const char *key, + _Bool *value) + UC_WRAP(meta_data_get_boolean) +#undef UC_WRAP + +/* vim: set sw=2 ts=8 sts=2 tw=78 : */