X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fplugin.c;h=6139baf0de05a2145ac5020d73d2bf64ef2814ae;hb=5a7145dd5626503f411a4aed87f30fc0f9689c90;hp=f222e3acb01cd41f681618bf6507decaf797de4b;hpb=9272eec7d5dc98eaa6c96f22c002e77b4f37fdf0;p=collectd.git diff --git a/src/plugin.c b/src/plugin.c index f222e3ac..6139baf0 100644 --- a/src/plugin.c +++ b/src/plugin.c @@ -1,6 +1,6 @@ /** * collectd - src/plugin.c - * Copyright (C) 2005-2008 Florian octo Forster + * Copyright (C) 2005-2009 Florian octo Forster * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the @@ -34,18 +34,35 @@ #include "configfile.h" #include "utils_avltree.h" #include "utils_llist.h" +#include "utils_heap.h" #include "utils_cache.h" #include "utils_threshold.h" +#include "filter_chain.h" /* * Private structures */ +struct callback_func_s +{ + void *cf_callback; + user_data_t cf_udata; +}; +typedef struct callback_func_s callback_func_t; + +#define RF_SIMPLE 0 +#define RF_COMPLEX 1 struct read_func_s { - int wait_time; - int wait_left; - int (*callback) (void); - enum { DONE = 0, TODO = 1, ACTIVE = 2 } needs_read; + /* `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 + callback_func_t rf_super; + char rf_name[DATA_MAX_NAME_LEN]; + int rf_type; + struct timespec rf_interval; + struct timespec rf_effective_interval; + struct timespec rf_next_read; }; typedef struct read_func_s read_func_t; @@ -53,18 +70,20 @@ typedef struct read_func_s read_func_t; * Private variables */ static llist_t *list_init; -static llist_t *list_read; static llist_t *list_write; -static llist_t *list_filter; static llist_t *list_flush; 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; +static c_heap_t *read_heap = NULL; static int read_loop = 1; static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER; @@ -82,26 +101,101 @@ static const char *plugin_get_dir (void) return (plugindir); } -static int register_callback (llist_t **list, const char *name, void *callback) +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) + { + callback_func_t *cf; + + cf = c_head_get_root (read_heap); + if (cf == NULL) + break; + + destroy_callback (cf); + } + + c_heap_destroy (read_heap); + read_heap = NULL; +} /* }}} void destroy_read_heap */ + +static int register_callback (llist_t **list, /* {{{ */ + const char *name, callback_func_t *cf) { llentry_t *le; char *key; - if ((*list == NULL) - && ((*list = llist_create ()) == NULL)) + if (*list == NULL) + { + *list = llist_create (); + if (*list == NULL) + { + ERROR ("plugin: create_register_callback: " + "llist_create failed."); + destroy_callback (cf); + return (-1); + } + } + + key = strdup (name); + if (key == NULL) + { + ERROR ("plugin: create_register_callback: strdup failed."); + destroy_callback (cf); return (-1); + } le = llist_search (*list, name); if (le == NULL) { - key = strdup (name); - if (key == NULL) - return (-1); - - le = llentry_create (key, callback); + le = llentry_create (key, cf); if (le == NULL) { + ERROR ("plugin: create_register_callback: " + "llentry_create failed."); free (key); + destroy_callback (cf); return (-1); } @@ -109,27 +203,65 @@ static int register_callback (llist_t **list, const char *name, void *callback) } else { - le->value = callback; + callback_func_t *old_cf; + + old_cf = le->value; + le->value = cf; + + destroy_callback (old_cf); + sfree (key); } return (0); -} /* int register_callback */ +} /* }}} int register_callback */ -static int plugin_unregister (llist_t *list, const char *name) +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; + } + + return (register_callback (list, name, cf)); +} /* }}} int create_register_callback */ + +static int plugin_unregister (llist_t *list, const char *name) /* {{{ */ { llentry_t *e; - e = llist_search (list, name); + if (list == NULL) + return (-1); + e = llist_search (list, name); if (e == NULL) return (-1); llist_remove (list, e); - free (e->key); + + sfree (e->key); + destroy_callback (e->value); + llentry_destroy (e); return (0); -} /* int plugin_unregister */ +} /* }}} int plugin_unregister */ /* * (Try to) load the shared object `file'. Won't complain if it isn't a shared @@ -168,80 +300,146 @@ static int plugin_load_file (char *file) return (0); } -static void *plugin_read_thread (void *args) +static void *plugin_read_thread (void __attribute__((unused)) *args) { - llentry_t *le; - read_func_t *rf; - int status; - int done; - - pthread_mutex_lock (&read_lock); - while (read_loop != 0) { - le = llist_head (list_read); - done = 0; + read_func_t *rf; + struct timeval now; + int status; - while ((read_loop != 0) && (le != NULL)) + /* Get the read function that needs to be read next. */ + rf = c_head_get_root (read_heap); + if (rf == NULL) { - rf = (read_func_t *) le->value; + struct timespec abstime; - if (rf->needs_read != TODO) - { - le = le->next; - continue; - } + gettimeofday (&now, /* timezone = */ NULL); - /* We will do this read function */ - rf->needs_read = ACTIVE; + abstime.tv_sec = now.tv_sec + interval_g; + abstime.tv_nsec = 1000 * now.tv_usec; - DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Handling %s", - (unsigned long int) pthread_self (), le->key); + pthread_mutex_lock (&read_lock); + pthread_cond_timedwait (&read_cond, &read_lock, + &abstime); pthread_mutex_unlock (&read_lock); + continue; + } - status = rf->callback (); - done++; + if ((rf->rf_interval.tv_sec == 0) && (rf->rf_interval.tv_nsec == 0)) + { + gettimeofday (&now, /* timezone = */ NULL); - if (status != 0) - { - if (rf->wait_time < interval_g) - rf->wait_time = interval_g; - rf->wait_left = rf->wait_time; - rf->wait_time = rf->wait_time * 2; - if (rf->wait_time > 86400) - rf->wait_time = 86400; - - NOTICE ("read-function of plugin `%s' " - "failed. Will suspend it for %i " - "seconds.", le->key, rf->wait_left); - } - else + rf->rf_interval.tv_sec = interval_g; + rf->rf_interval.tv_nsec = 0; + + rf->rf_effective_interval = rf->rf_interval; + + rf->rf_next_read.tv_sec = now.tv_sec; + rf->rf_next_read.tv_nsec = 1000 * now.tv_usec; + } + + /* sleep until this entry is due, + * using pthread_cond_timedwait */ + pthread_mutex_lock (&read_lock); + pthread_cond_timedwait (&read_cond, &read_lock, + &rf->rf_next_read); + 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; + } + + DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + + 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); + } + + /* If the function signals failure, we will increase the + * intervals in which it will be called. */ + if (status != 0) + { + rf->rf_effective_interval.tv_sec *= 2; + rf->rf_effective_interval.tv_nsec *= 2; + NORMALIZE_TIMESPEC (rf->rf_effective_interval); + + if (rf->rf_effective_interval.tv_sec >= 86400) { - rf->wait_left = 0; - rf->wait_time = interval_g; + rf->rf_effective_interval.tv_sec = 86400; + rf->rf_effective_interval.tv_nsec = 0; } - pthread_mutex_lock (&read_lock); - - rf->needs_read = DONE; - le = le->next; - } /* while (le != NULL) */ + NOTICE ("read-function of plugin `%s' failed. " + "Will suspend it for %i seconds.", + rf->rf_name, + (int) rf->rf_effective_interval.tv_sec); + } + else + { + /* Success: Restore the interval, if it was changed. */ + rf->rf_effective_interval = rf->rf_interval; + } - if ((read_loop != 0) && (done == 0)) + /* update the ``next read due'' field */ + gettimeofday (&now, /* timezone = */ NULL); + + DEBUG ("plugin_read_thread: Effective interval of the " + "%s plugin is %i.%09i.", + rf->rf_name, + (int) rf->rf_effective_interval.tv_sec, + (int) rf->rf_effective_interval.tv_nsec); + + /* Calculate the next (absolute) time at which this function + * should be called. */ + rf->rf_next_read.tv_sec = rf->rf_next_read.tv_sec + + rf->rf_effective_interval.tv_sec; + rf->rf_next_read.tv_nsec = rf->rf_next_read.tv_nsec + + rf->rf_effective_interval.tv_nsec; + NORMALIZE_TIMESPEC (rf->rf_next_read); + + /* Check, if `rf_next_read' is in the past. */ + if ((rf->rf_next_read.tv_sec < now.tv_sec) + || ((rf->rf_next_read.tv_sec == now.tv_sec) + && (rf->rf_next_read.tv_nsec < (1000 * now.tv_usec)))) { - DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Waiting on read_cond.", - (unsigned long int) pthread_self ()); - pthread_cond_wait (&read_cond, &read_lock); + /* `rf_next_read' is in the past. Insert `now' + * so this value doesn't trail off into the + * past too much. */ + rf->rf_next_read.tv_sec = now.tv_sec; + rf->rf_next_read.tv_nsec = 1000 * now.tv_usec; } - } /* while (read_loop) */ - pthread_mutex_unlock (&read_lock); + DEBUG ("plugin_read_thread: Next read of the %s plugin at %i.%09i.", + rf->rf_name, + (int) rf->rf_next_read.tv_sec, + (int) rf->rf_next_read.tv_nsec); + + /* Re-insert this read function into the heap again. */ + c_heap_insert (read_heap, rf); + } /* while (read_loop) */ pthread_exit (NULL); return ((void *) 0); } /* void *plugin_read_thread */ -static void start_threads (int num) +static void start_read_threads (int num) { int i; @@ -251,7 +449,7 @@ static void start_threads (int num) read_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); if (read_threads == NULL) { - ERROR ("plugin: start_threads: calloc failed."); + ERROR ("plugin: start_read_threads: calloc failed."); return; } @@ -265,19 +463,24 @@ static void start_threads (int num) } else { - ERROR ("plugin: start_threads: pthread_create failed."); + ERROR ("plugin: start_read_threads: pthread_create failed."); return; } } /* for (i) */ -} /* void start_threads */ +} /* void start_read_threads */ -static void stop_threads (void) +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_threads: Signalling `read_cond'"); + DEBUG ("plugin: stop_read_threads: Signalling `read_cond'"); pthread_cond_broadcast (&read_cond); pthread_mutex_unlock (&read_lock); @@ -285,13 +488,13 @@ static void stop_threads (void) { if (pthread_join (read_threads[i], NULL) != 0) { - ERROR ("plugin: stop_threads: pthread_join failed."); + ERROR ("plugin: stop_read_threads: pthread_join failed."); } read_threads[i] = (pthread_t) 0; } sfree (read_threads); read_threads_num = 0; -} /* void stop_threads */ +} /* void stop_read_threads */ /* * Public functions @@ -322,6 +525,7 @@ int plugin_load (const char *type) int ret; struct stat statbuf; struct dirent *de; + int status; DEBUG ("type = %s", type); @@ -330,8 +534,8 @@ int plugin_load (const char *type) /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the * type when matching the filename */ - if (ssnprintf (typename, sizeof (typename), - "%s.so", type) >= sizeof (typename)) + status = ssnprintf (typename, sizeof (typename), "%s.so", type); + if ((status < 0) || ((size_t) status >= sizeof (typename))) { WARNING ("snprintf: truncated: `%s.so'", type); return (-1); @@ -351,8 +555,9 @@ int plugin_load (const char *type) if (strncasecmp (de->d_name, typename, typename_len)) continue; - if (ssnprintf (filename, sizeof (filename), - "%s/%s", dir, de->d_name) >= sizeof (filename)) + status = ssnprintf (filename, sizeof (filename), + "%s/%s", dir, de->d_name); + if ((status < 0) || ((size_t) status >= sizeof (filename))) { WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name); continue; @@ -368,6 +573,7 @@ int plugin_load (const char *type) else if (!S_ISREG (statbuf.st_mode)) { /* don't follow symlinks */ + WARNING ("stat %s: not a regular file", filename); continue; } @@ -411,14 +617,46 @@ int plugin_register_complex_config (const char *type, int plugin_register_init (const char *name, int (*callback) (void)) { - return (register_callback (&list_init, name, (void *) callback)); + 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.tv_sec < rf1->rf_next_read.tv_sec) + return (-1); + else if (rf0->rf_next_read.tv_sec > rf1->rf_next_read.tv_sec) + return (1); + else if (rf0->rf_next_read.tv_nsec < rf1->rf_next_read.tv_nsec) + return (-1); + else if (rf0->rf_next_read.tv_nsec > rf1->rf_next_read.tv_nsec) + return (1); + else + return (0); +} /* int plugin_compare_read_func */ + int plugin_register_read (const char *name, int (*callback) (void)) { read_func_t *rf; + if (read_heap == NULL) + { + read_heap = c_heap_create (plugin_compare_read_func); + if (read_heap == NULL) + { + ERROR ("plugin_register_read: " + "c_heap_create failed."); + return (-1); + } + } + rf = (read_func_t *) malloc (sizeof (read_func_t)); if (rf == NULL) { @@ -428,37 +666,86 @@ int plugin_register_read (const char *name, return (-1); } - memset (rf, '\0', sizeof (read_func_t)); - rf->wait_time = interval_g; - rf->wait_left = 0; - rf->callback = callback; - rf->needs_read = DONE; - - return (register_callback (&list_read, name, (void *) rf)); + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + rf->rf_udata.data = NULL; + rf->rf_udata.free_func = NULL; + sstrncpy (rf->rf_name, name, sizeof (rf->rf_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; + + return (c_heap_insert (read_heap, rf)); } /* int plugin_register_read */ -int plugin_register_write (const char *name, - int (*callback) (const data_set_t *ds, const value_list_t *vl)) +int plugin_register_complex_read (const char *name, + plugin_read_cb callback, + const struct timespec *interval, + user_data_t *user_data) { - return (register_callback (&list_write, name, (void *) callback)); -} /* int plugin_register_write */ + read_func_t *rf; -int plugin_register_filter (const char *name, - int (*callback) (const data_set_t *ds, value_list_t *vl)) + if (read_heap == NULL) + { + read_heap = c_heap_create (plugin_compare_read_func); + if (read_heap == NULL) + { + ERROR ("plugin_register_read: c_heap_create failed."); + return (-1); + } + } + + rf = (read_func_t *) malloc (sizeof (read_func_t)); + if (rf == NULL) + { + ERROR ("plugin_register_complex_read: malloc failed."); + return (-1); + } + + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); + rf->rf_type = RF_COMPLEX; + if (interval != NULL) + { + rf->rf_interval = *interval; + } + rf->rf_effective_interval = rf->rf_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; + } + + return (c_heap_insert (read_heap, rf)); +} /* int plugin_register_complex_read */ + +int plugin_register_write (const char *name, + plugin_write_cb callback, user_data_t *ud) { - return (register_callback (&list_filter, name, (void *) callback)); -} /* int plugin_register_filter */ + return (create_register_callback (&list_write, name, + (void *) callback, ud)); +} /* int plugin_register_write */ int plugin_register_flush (const char *name, - int (*callback) (const int timeout, const char *identifier)) + plugin_flush_cb callback, user_data_t *ud) { - return (register_callback (&list_flush, name, (void *) callback)); + return (create_register_callback (&list_flush, name, + (void *) callback, ud)); } /* int plugin_register_flush */ int plugin_register_shutdown (char *name, int (*callback) (void)) { - return (register_callback (&list_shutdown, name, (void *) callback)); + return (create_register_callback (&list_shutdown, name, + (void *) callback, /* user_data = */ NULL)); } /* int plugin_register_shutdown */ int plugin_register_data_set (const data_set_t *ds) @@ -498,16 +785,18 @@ int plugin_register_data_set (const data_set_t *ds) return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy)); } /* int plugin_register_data_set */ -int plugin_register_log (char *name, - void (*callback) (int priority, const char *msg)) +int plugin_register_log (const char *name, + plugin_log_cb callback, user_data_t *ud) { - return (register_callback (&list_log, name, (void *) callback)); + return (create_register_callback (&list_log, name, + (void *) callback, ud)); } /* int plugin_register_log */ int plugin_register_notification (const char *name, - int (*callback) (const notification_t *notif)) + plugin_notification_cb callback, user_data_t *ud) { - return (register_callback (&list_notification, name, (void *) callback)); + return (create_register_callback (&list_notification, name, + (void *) callback, ud)); } /* int plugin_register_log */ int plugin_unregister_config (const char *name) @@ -529,19 +818,9 @@ int plugin_unregister_init (const char *name) int plugin_unregister_read (const char *name) { - llentry_t *e; - - e = llist_search (list_read, name); - - if (e == NULL) - return (-1); - - llist_remove (list_read, e); - free (e->value); - free (e->key); - llentry_destroy (e); - - return (0); + /* TODO: Implement removal of a specific key from the heap. */ + assert (0); + return (-1); } int plugin_unregister_write (const char *name) @@ -549,11 +828,6 @@ int plugin_unregister_write (const char *name) return (plugin_unregister (list_write, name)); } -int plugin_unregister_filter (const char *name) -{ - return (plugin_unregister (list_filter, name)); -} - int plugin_unregister_flush (const char *name) { return (plugin_unregister (list_flush, name)); @@ -592,14 +866,21 @@ int plugin_unregister_notification (const char *name) void plugin_init_all (void) { - int (*callback) (void); + const char *chain_name; llentry_t *le; int status; /* Init the value cache */ uc_init (); - if ((list_init == NULL) && (list_read == NULL)) + 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); + + + if ((list_init == NULL) && (read_heap == NULL)) return; /* Calling all init callbacks before checking if read callbacks @@ -608,7 +889,11 @@ void plugin_init_all (void) le = llist_head (list_init); while (le != NULL) { - callback = (int (*) (void)) le->value; + callback_func_t *cf; + plugin_init_cb callback; + + cf = le->value; + callback = cf->cf_callback; status = (*callback) (); if (status != 0) @@ -628,59 +913,76 @@ void plugin_init_all (void) } /* Start read-threads */ - if (list_read != NULL) + if (read_heap != NULL) { const char *rt; int num; rt = global_option_get ("ReadThreads"); num = atoi (rt); - start_threads ((num > 0) ? num : 5); + if (num != -1) + start_read_threads ((num > 0) ? num : 5); } } /* void plugin_init_all */ +/* TODO: Rename this function. */ void plugin_read_all (void) { - llentry_t *le; - read_func_t *rf; - uc_check_timeout (); - if (list_read == NULL) - return; + return; +} /* void plugin_read_all */ - pthread_mutex_lock (&read_lock); +/* Read function called when the `-T' command line argument is given. */ +int plugin_read_all_once (void) +{ + int status; + int return_status = 0; - le = llist_head (list_read); - while (le != NULL) + if (read_heap == NULL) { - rf = (read_func_t *) le->value; + NOTICE ("No read-functions are registered."); + return (0); + } + + while (42) + { + read_func_t *rf; + + rf = c_head_get_root (read_heap); + if (rf == NULL) + break; - if (rf->needs_read != DONE) + if (rf->rf_type == RF_SIMPLE) { - le = le->next; - continue; + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); } + else + { + plugin_read_cb callback; - if (rf->wait_left > 0) - rf->wait_left -= interval_g; + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } - if (rf->wait_left <= 0) + if (status != 0) { - rf->needs_read = TODO; + NOTICE ("read-function of plugin `%s' failed.", + rf->rf_name); + return_status = -1; } - le = le->next; + destroy_callback ((void *) rf); } - DEBUG ("plugin: plugin_read_all: Signalling `read_cond'"); - pthread_cond_broadcast (&read_cond); - pthread_mutex_unlock (&read_lock); -} /* void plugin_read_all */ + return (return_status); +} /* int plugin_read_all_once */ int plugin_write (const char *plugin, /* {{{ */ const data_set_t *ds, const value_list_t *vl) { - int (*callback) (const data_set_t *ds, const value_list_t *vl); llentry_t *le; int status; @@ -708,8 +1010,12 @@ int plugin_write (const char *plugin, /* {{{ */ le = llist_head (list_write); while (le != NULL) { - callback = le->value; - status = (*callback) (ds, vl); + callback_func_t *cf = le->value; + plugin_write_cb callback; + + 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 @@ -725,6 +1031,9 @@ int plugin_write (const char *plugin, /* {{{ */ } else /* plugin != NULL */ { + callback_func_t *cf; + plugin_write_cb callback; + le = llist_head (list_write); while (le != NULL) { @@ -737,8 +1046,11 @@ int plugin_write (const char *plugin, /* {{{ */ if (le == NULL) return (ENOENT); - callback = le->value; - status = (*callback) (ds, vl); + cf = le->value; + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); } return (status); @@ -746,7 +1058,6 @@ int plugin_write (const char *plugin, /* {{{ */ int plugin_flush (const char *plugin, int timeout, const char *identifier) { - int (*callback) (int timeout, const char *identifier); llentry_t *le; if (list_flush == NULL) @@ -755,6 +1066,9 @@ int plugin_flush (const char *plugin, int timeout, const char *identifier) le = llist_head (list_flush); while (le != NULL) { + callback_func_t *cf; + plugin_flush_cb callback; + if ((plugin != NULL) && (strcmp (plugin, le->key) != 0)) { @@ -762,8 +1076,10 @@ int plugin_flush (const char *plugin, int timeout, const char *identifier) continue; } - callback = (int (*) (int, const char *)) le->value; - (*callback) (timeout, identifier); + cf = le->value; + callback = cf->cf_callback; + + (*callback) (timeout, identifier, &cf->cf_udata); le = le->next; } @@ -772,18 +1088,27 @@ int plugin_flush (const char *plugin, int timeout, const char *identifier) void plugin_shutdown_all (void) { - int (*callback) (void); llentry_t *le; - stop_threads (); + stop_read_threads (); - if (list_shutdown == NULL) - return; + destroy_all_callbacks (&list_init); + destroy_read_heap (); + + plugin_flush (/* plugin = */ NULL, /* timeout = */ -1, + /* identifier = */ NULL); + + le = NULL; + if (list_shutdown != NULL) + le = llist_head (list_shutdown); - le = llist_head (list_shutdown); while (le != NULL) { - callback = (int (*) (void)) le->value; + callback_func_t *cf; + plugin_shutdown_cb callback; + + cf = le->value; + callback = cf->cf_callback; /* Advance the pointer before calling the callback allows * shutdown functions to unregister themselves. If done the @@ -793,18 +1118,27 @@ void plugin_shutdown_all (void) (*callback) (); } + + destroy_all_callbacks (&list_write); + destroy_all_callbacks (&list_flush); + destroy_all_callbacks (&list_notification); + destroy_all_callbacks (&list_shutdown); + destroy_all_callbacks (&list_log); } /* void plugin_shutdown_all */ int plugin_dispatch_values (value_list_t *vl) { + int status; static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC; - data_set_t *ds; - llentry_t *le; + value_t *saved_values; + int saved_values_len; - int filter = 0; + data_set_t *ds; - if ((vl == NULL) || (*vl->type == '\0')) { + if ((vl == NULL) || (vl->type[0] == 0) + || (vl->values == NULL) || (vl->values_len < 1)) + { ERROR ("plugin_dispatch_values: Invalid value list."); return (-1); } @@ -829,6 +1163,12 @@ int plugin_dispatch_values (value_list_t *vl) return (-1); } + if (vl->time == 0) + vl->time = time (NULL); + + if (vl->interval <= 0) + vl->interval = interval_g; + DEBUG ("plugin_dispatch_values: time = %u; interval = %i; " "host = %s; " "plugin = %s; plugin_instance = %s; " @@ -865,38 +1205,82 @@ int plugin_dispatch_values (value_list_t *vl) escape_slashes (vl->type, sizeof (vl->type)); escape_slashes (vl->type_instance, sizeof (vl->type_instance)); - le = llist_head (list_filter); - while (le != NULL) + /* 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)) { - int (*filter_callback) (const data_set_t *, value_list_t *) = - (int (*) (const data_set_t *, value_list_t *)) le->value; + saved_values = vl->values; + saved_values_len = vl->values_len; - filter |= (*filter_callback) (ds, vl); - - if (filter == FILTER_IGNORE) + 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; + } - le = le->next; + 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 ((filter & FILTER_NOTHRESHOLD_CHECK) == 0) - ut_check_threshold (ds, vl); + /* Initiate threshold checking */ + ut_check_threshold (ds, vl); - if (filter & FILTER_NOWRITE) - return (0); - - le = llist_head (list_write); - while (le != NULL) + if (post_cache_chain != NULL) { - int (*write_callback) (const data_set_t *, const value_list_t *) = - (int (*) (const data_set_t *, const value_list_t *)) le->value; - - (*write_callback) (ds, vl); + 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); - le = le->next; + /* 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); @@ -904,7 +1288,6 @@ int plugin_dispatch_values (value_list_t *vl) int plugin_dispatch_notification (const notification_t *notif) { - int (*callback) (const notification_t *); llentry_t *le; /* Possible TODO: Add flap detection here */ @@ -920,8 +1303,19 @@ int plugin_dispatch_notification (const notification_t *notif) le = llist_head (list_notification); while (le != NULL) { - callback = (int (*) (const notification_t *)) le->value; - (*callback) (notif); + callback_func_t *cf; + plugin_notification_cb callback; + int status; + + cf = le->value; + callback = cf->cf_callback; + status = (*callback) (notif, &cf->cf_udata); + if (status != 0) + { + WARNING ("plugin_dispatch_notification: Notification " + "callback %s returned %i.", + le->key, status); + } le = le->next; } @@ -933,8 +1327,6 @@ void plugin_log (int level, const char *format, ...) { char msg[1024]; va_list ap; - - void (*callback) (int, const char *); llentry_t *le; if (list_log == NULL) @@ -953,8 +1345,13 @@ void plugin_log (int level, const char *format, ...) le = llist_head (list_log); while (le != NULL) { - callback = (void (*) (int, const char *)) le->value; - (*callback) (level, msg); + callback_func_t *cf; + plugin_log_cb callback; + + cf = le->value; + callback = cf->cf_callback; + + (*callback) (level, msg, &cf->cf_udata); le = le->next; } @@ -1002,8 +1399,8 @@ static int plugin_notification_meta_add (notification_t *n, { case NM_TYPE_STRING: { - meta->value_string = strdup ((const char *) value); - if (meta->value_string == NULL) + 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); @@ -1013,22 +1410,22 @@ static int plugin_notification_meta_add (notification_t *n, } case NM_TYPE_SIGNED_INT: { - meta->value_signed_int = *((int64_t *) value); + meta->nm_value.nm_signed_int = *((int64_t *) value); break; } case NM_TYPE_UNSIGNED_INT: { - meta->value_unsigned_int = *((uint64_t *) value); + meta->nm_value.nm_unsigned_int = *((uint64_t *) value); break; } case NM_TYPE_DOUBLE: { - meta->value_double = *((double *) value); + meta->nm_value.nm_double = *((double *) value); break; } case NM_TYPE_BOOLEAN: { - meta->value_boolean = *((bool *) value); + meta->nm_value.nm_boolean = *((bool *) value); break; } default: @@ -1101,25 +1498,25 @@ int plugin_notification_meta_copy (notification_t *dst, { if (meta->type == NM_TYPE_STRING) plugin_notification_meta_add_string (dst, meta->name, - meta->value_string); + meta->nm_value.nm_string); else if (meta->type == NM_TYPE_SIGNED_INT) plugin_notification_meta_add_signed_int (dst, meta->name, - meta->value_signed_int); + meta->nm_value.nm_signed_int); else if (meta->type == NM_TYPE_UNSIGNED_INT) plugin_notification_meta_add_unsigned_int (dst, meta->name, - meta->value_unsigned_int); + meta->nm_value.nm_unsigned_int); else if (meta->type == NM_TYPE_DOUBLE) plugin_notification_meta_add_double (dst, meta->name, - meta->value_double); + meta->nm_value.nm_double); else if (meta->type == NM_TYPE_BOOLEAN) plugin_notification_meta_add_boolean (dst, meta->name, - meta->value_boolean); + meta->nm_value.nm_boolean); } return (0); } /* int plugin_notification_meta_copy */ -int plugin_notification_meta_free (notification_t *n) +int plugin_notification_meta_free (notification_meta_t *n) { notification_meta_t *this; notification_meta_t *next; @@ -1130,16 +1527,15 @@ int plugin_notification_meta_free (notification_t *n) return (-1); } - this = n->meta; - n->meta = NULL; + this = n; while (this != NULL) { next = this->next; if (this->type == NM_TYPE_STRING) { - free ((char *)this->value_string); - this->value_string = NULL; + free ((char *)this->nm_value.nm_string); + this->nm_value.nm_string = NULL; } sfree (this); @@ -1148,3 +1544,5 @@ int plugin_notification_meta_free (notification_t *n) return (0); } /* int plugin_notification_meta_free */ + +/* vim: set sw=8 ts=8 noet fdm=marker : */