![[tokkee]](http://tokkee.org/images/avatar.png){kind=avatar}
diff --git a/src/plugin.c b/src/plugin.c
index b2a8b86511efd1e905fc687301a547f83280eb0b..4e106a5b1eb5193c3188ff87934feff0f4d4fa49 100644 (file)
--- a/src/plugin.c
+++ b/src/plugin.c
/*
* 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 write_queue_t *write_queue_head;
static write_queue_t *write_queue_tail;
+static long write_queue_size = 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_key_t plugin_ctx_key;
static _Bool plugin_ctx_key_initialized = 0;
+static long writequeuelengthlimit_high = 0;
+static long writequeuelengthlimit_low = 0;
+
+static unsigned int random_seed;
+
/*
* Static functions
*/
{
write_queue_head = q;
write_queue_tail = q;
+ write_queue_size = 1;
}
else
{
write_queue_tail->next = q;
write_queue_tail = q;
+ write_queue_size += 1;
}
pthread_cond_signal (&write_cond);
q = write_queue_head;
write_queue_head = q->next;
- if (write_queue_head == NULL)
+ write_queue_size -= 1;
+ if (write_queue_head == NULL) {
write_queue_tail = NULL;
+ write_queue_size = 0; /* Maybe instead of setting write_queue_size to 0, we should assert(write_queue_size == 0) ? */
+ }
pthread_mutex_unlock (&write_lock);
}
write_queue_head = NULL;
write_queue_tail = NULL;
+ write_queue_size = 0;
pthread_mutex_unlock (&write_lock);
if (i > 0)
}
}
+static _Bool plugin_is_loaded (char const *name)
+{
+ int status;
+
+ if (plugins_loaded == NULL)
+ plugins_loaded = c_avl_create ((void *) strcasecmp);
+ assert (plugins_loaded != NULL);
+
+ status = c_avl_get (plugins_loaded, name, /* ret_value = */ NULL);
+ return (status == 0);
+}
+
+static int plugin_mark_loaded (char const *name)
+{
+ char *name_copy;
+ int status;
+
+ name_copy = strdup (name);
+ if (name_copy == NULL)
+ return (ENOMEM);
+
+ status = c_avl_insert (plugins_loaded,
+ /* key = */ name_copy, /* value = */ NULL);
+ return (status);
+}
+
+static void plugin_free_loaded ()
+{
+ void *key;
+ void *value;
+
+ if (plugins_loaded == NULL)
+ return;
+
+ while (c_avl_pick (plugins_loaded, &key, &value) == 0)
+ {
+ sfree (key);
+ assert (value == NULL);
+ }
+
+ c_avl_destroy (plugins_loaded);
+ plugins_loaded = NULL;
+}
+
#define BUFSIZE 512
-int plugin_load (const char *type, uint32_t flags)
+int plugin_load (char const *plugin_name, uint32_t flags)
{
DIR *dh;
const char *dir;
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", type);
+ 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\"", type);
+ WARNING ("plugin_load: Filename too long: \"%s.so\"", plugin_name);
return (-1);
}
typename_len = strlen (typename);
if (status == 0)
{
/* success */
+ plugin_mark_loaded (plugin_name);
ret = 0;
break;
}
else
{
ERROR ("plugin_load: Load plugin \"%s\" failed with "
- "status %i.", type, status);
+ "status %i.", plugin_name, status);
}
}
if (filename[0] == 0)
ERROR ("plugin_load: Could not find plugin \"%s\" in %s",
- type, dir);
+ plugin_name, dir);
return (ret);
}
chain_name = global_option_get ("PostCacheChain");
post_cache_chain = fc_chain_get_by_name (chain_name);
+ {
+ const char *str_writequeuelengthlimithigh = global_option_get ("WriteQueueLengthLimitHigh");
+ const char *str_writequeuelengthlimitlow = global_option_get ("WriteQueueLengthLimitLow");
+
+ writequeuelengthlimit_high = 0;
+ writequeuelengthlimit_low = 0;
+
+ if(NULL != str_writequeuelengthlimithigh) {
+ errno = 0;
+ /* get high limit */
+ writequeuelengthlimit_high = strtol(str_writequeuelengthlimithigh, NULL, 10);
+ if ((errno == ERANGE && (writequeuelengthlimit_high == LONG_MAX || writequeuelengthlimit_high == LONG_MIN))
+ || (errno != 0 && writequeuelengthlimit_high == 0)
+ ) {
+ writequeuelengthlimit_high = 0;
+ ERROR("Config 'WriteQueueLengthLimitHigh' accepts one integer value only. Running with no limit !");
+ }
+ if(writequeuelengthlimit_high < 0) {
+ ERROR("Config 'WriteQueueLengthLimitHigh' accepts positive values only. Running with no limit !");
+ writequeuelengthlimit_high = 0;
+ }
+ }
+
+ if((writequeuelengthlimit_high > 0) && (NULL != str_writequeuelengthlimitlow)) {
+ errno = 0;
+ /* get low limit */
+ writequeuelengthlimit_low = strtol(str_writequeuelengthlimitlow, NULL, 10);
+ if ((errno == ERANGE && (writequeuelengthlimit_low == LONG_MAX || writequeuelengthlimit_low == LONG_MIN))
+ || (errno != 0 && writequeuelengthlimit_low == 0)
+ ) {
+ writequeuelengthlimit_low = 0;
+ ERROR("Config 'WriteQueueLengthLimitLow' accepts one integer value only. Using default low limit instead");
+ }
+
+ if(writequeuelengthlimit_low < 0) {
+ ERROR("Config 'WriteQueueLengthLimitLow' accepts positive values only. Using default low limit instead");
+ writequeuelengthlimit_low = 0;
+ } else if(writequeuelengthlimit_low > writequeuelengthlimit_high) {
+ ERROR("Config 'WriteQueueLengthLimitLow' (%ld) cannot be bigger than high limit (%ld). Using default low limit instead",
+ writequeuelengthlimit_low, writequeuelengthlimit_high);
+ writequeuelengthlimit_low = 0;
+ }
+ }
+ /* Check/compute low limit if not/badly defined */
+ if(writequeuelengthlimit_high > 0) {
+ if(0 == writequeuelengthlimit_low) {
+ writequeuelengthlimit_low = .5 * writequeuelengthlimit_high;
+ }
+ INFO("Config 'WriteQueueLengthLimit*' : Running with limits high=%ld low=%ld", writequeuelengthlimit_high, writequeuelengthlimit_low);
+ random_seed = time(0);
+ } else {
+ writequeuelengthlimit_low = 0; /* This should be useless, but in case... */
+ }
+ }
+
{
char const *tmp = global_option_get ("WriteThreads");
int num = atoi (tmp);
destroy_all_callbacks (&list_notification);
destroy_all_callbacks (&list_shutdown);
destroy_all_callbacks (&list_log);
+
+ plugin_free_loaded ();
} /* void plugin_shutdown_all */
int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */
int plugin_dispatch_values (value_list_t const *vl)
{
int status;
+ int wq_size = write_queue_size;
+ /* We store write_queue_size in a local variable because other threads may update write_queue_size.
+ * Having this in a local variable (like a cache) is better : we do not need a lock */
+ short metric_will_be_dropped = 0;
+
+ if((writequeuelengthlimit_high > 0) && (wq_size > writequeuelengthlimit_low)) {
+ if(wq_size >= writequeuelengthlimit_high) {
+ /* if high == low, we come here too */
+ metric_will_be_dropped = 1;
+ } else {
+ /* here, high != low */
+ long probability_to_drop;
+ long n;
+
+ probability_to_drop = (wq_size - writequeuelengthlimit_low);
+
+ /* We use rand_r with its bad RNG because it's enough for playing dices.
+ * There is no security consideration here so rand_r() should be enough here.
+ */
+ n = rand_r(&random_seed) % (writequeuelengthlimit_high - writequeuelengthlimit_low) ;
+
+ /* Let's have X = high - low.
+ * n is in range [0..X]
+ * probability_to_drop is in range [1..X[
+ * probability_to_drop gets bigger when wq_size gets bigger.
+ */
+ if(n <= probability_to_drop) {
+ metric_will_be_dropped = 1;
+ }
+ }
+ }
- status = plugin_write_enqueue (vl);
- if (status != 0)
+ if( ! metric_will_be_dropped) {
+ 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);
+ }
+ }
+ else
{
- char errbuf[1024];
- ERROR ("plugin_dispatch_values: plugin_write_enqueue failed "
- "with status %i (%s).", status,
- sstrerror (status, errbuf, sizeof (errbuf)));
- return (status);
+ /* If you want to count dropped metrics, don't forget to add a lock here */
+ /* dropped_metrics++; */
+ ERROR ("plugin_dispatch_values: value dropped (write queue %ld > %ld) : time = %.3f; interval = %.3f ; %s/%s%s%s/%s%s%s",
+ write_queue_size, writequeuelengthlimit_low,
+ CDTIME_T_TO_DOUBLE (vl->time),
+ CDTIME_T_TO_DOUBLE (vl->interval),
+ vl->host,
+ vl->plugin, vl->plugin_instance[0]?"-":"", vl->plugin_instance,
+ vl->type, vl->type_instance[0]?"-":"", vl->type_instance);
}
return (0);