X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Frrdtool.c;h=ee5d70c8255ed7631e1c6f770ce5b74b7f5fd5f2;hb=4858a84e3e51830a103a5ff4aee6b844b708ce62;hp=b80e1109cfcdfed8e27623a3390e548e0ca9e7e5;hpb=de0fdb208de123fe753c5fcf03533833777a5b4a;p=collectd.git diff --git a/src/rrdtool.c b/src/rrdtool.c index b80e1109..ee5d70c8 100644 --- a/src/rrdtool.c +++ b/src/rrdtool.c @@ -1,6 +1,8 @@ /** * collectd - src/rrdtool.c * Copyright (C) 2006-2008 Florian octo Forster + * Copyright (C) 2008-2008 Sebastian Harl + * Copyright (C) 2009 Mariusz Gronczewski * * 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 @@ -17,6 +19,8 @@ * * Authors: * Florian octo Forster + * Sebastian Harl + * Mariusz Gronczewski **/ #include "collectd.h" @@ -40,6 +44,7 @@ struct rrd_cache_s char **values; time_t first_value; time_t last_value; + int random_variation; enum { FLAG_NONE = 0x00, @@ -76,7 +81,8 @@ static const char *config_keys[] = "RRARows", "RRATimespan", "XFF", - "WritesPerSecond" + "WritesPerSecond", + "RandomTimeout" }; static int config_keys_num = STATIC_ARRAY_SIZE (config_keys); @@ -103,6 +109,7 @@ static rrdcreate_config_t rrdcreate_config = * ALWAYS lock `cache_lock' first! */ static int cache_timeout = 0; static int cache_flush_timeout = 0; +static int random_timeout = 1; static time_t cache_flush_last; static c_avl_tree_t *cache = NULL; static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; @@ -111,7 +118,8 @@ static rrd_queue_t *queue_head = NULL; static rrd_queue_t *queue_tail = NULL; static rrd_queue_t *flushq_head = NULL; static rrd_queue_t *flushq_tail = NULL; -static pthread_t queue_thread = 0; +static pthread_t queue_thread; +static int queue_thread_running = 1; static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; @@ -203,15 +211,23 @@ static int value_list_to_string (char *buffer, int buffer_len, for (i = 0; i < ds->ds_num; i++) { if ((ds->ds[i].type != DS_TYPE_COUNTER) - && (ds->ds[i].type != DS_TYPE_GAUGE)) + && (ds->ds[i].type != DS_TYPE_GAUGE) + && (ds->ds[i].type != DS_TYPE_DERIVE) + && (ds->ds[i].type != DS_TYPE_ABSOLUTE)) return (-1); if (ds->ds[i].type == DS_TYPE_COUNTER) status = ssnprintf (buffer + offset, buffer_len - offset, ":%llu", vl->values[i].counter); - else + else if (ds->ds[i].type == DS_TYPE_GAUGE) status = ssnprintf (buffer + offset, buffer_len - offset, ":%lf", vl->values[i].gauge); + else if (ds->ds[i].type == DS_TYPE_DERIVE) + status = ssnprintf (buffer + offset, buffer_len - offset, + ":%"PRIi64, vl->values[i].derive); + else /*if (ds->ds[i].type == DS_TYPE_ABSOLUTE) */ + status = ssnprintf (buffer + offset, buffer_len - offset, + ":%"PRIu64, vl->values[i].absolute); if ((status < 1) || (status >= (buffer_len - offset))) return (-1); @@ -282,6 +298,9 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data) int status; int i; + values = NULL; + values_num = 0; + pthread_mutex_lock (&queue_lock); /* Wait for values to arrive */ while (true) @@ -400,8 +419,9 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data) /* Write the values to the RRD-file */ srrd_update (queue_entry->filename, NULL, values_num, (const char **)values); - DEBUG ("rrdtool plugin: queue thread: Wrote %i values to %s", - values_num, queue_entry->filename); + DEBUG ("rrdtool plugin: queue thread: Wrote %i value%s to %s", + values_num, (values_num == 1) ? "" : "s", + queue_entry->filename); for (i = 0; i < values_num; i++) { @@ -596,7 +616,7 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier) status = c_avl_get (cache, key, (void *) &rc); if (status != 0) { - WARNING ("rrdtool plugin: rrd_cache_flush_identifier: " + INFO ("rrdtool plugin: rrd_cache_flush_identifier: " "c_avl_get (%s) failed. Does that file really exist?", key); return (status); @@ -726,7 +746,7 @@ static int rrd_cache_insert (const char *filename, filename, rc->values_num, (unsigned long)(rc->last_value - rc->first_value)); - if ((rc->last_value - rc->first_value) >= cache_timeout) + if ((rc->last_value + rc->random_variation - rc->first_value) >= cache_timeout) { /* XXX: If you need to lock both, cache_lock and queue_lock, at * the same time, ALWAYS lock `cache_lock' first! */ @@ -737,6 +757,18 @@ static int rrd_cache_insert (const char *filename, status = rrd_queue_enqueue (filename, &queue_head, &queue_tail); if (status == 0) rc->flags = FLAG_QUEUED; + + /* Update the jitter value. Negative values are + * slightly preferred. */ + if (random_timeout > 0) + { + rc->random_variation = (rand () % (2 * random_timeout)) + - random_timeout; + } + else + { + rc->random_variation = 0; + } } else { @@ -766,7 +798,8 @@ static int rrd_compare_numeric (const void *a_ptr, const void *b_ptr) return (0); } /* int rrd_compare_numeric */ -static int rrd_write (const data_set_t *ds, const value_list_t *vl) +static int rrd_write (const data_set_t *ds, const value_list_t *vl, + user_data_t __attribute__((unused)) *user_data) { struct stat statbuf; char filename[512]; @@ -814,7 +847,8 @@ static int rrd_write (const data_set_t *ds, const value_list_t *vl) return (status); } /* int rrd_write */ -static int rrd_flush (int timeout, const char *identifier) +static int rrd_flush (int timeout, const char *identifier, + user_data_t __attribute__((unused)) *user_data) { pthread_mutex_lock (&cache_lock); @@ -973,6 +1007,23 @@ static int rrd_config (const char *key, const char *value) write_rate = 1.0 / wps; } } + else if (strcasecmp ("RandomTimeout", key) == 0) + { + int tmp; + + tmp = atoi (value); + if (tmp < 0) + { + fprintf (stderr, "rrdtool: `RandomTimeout' must " + "be greater than or equal to zero.\n"); + ERROR ("rrdtool: `RandomTimeout' must " + "be greater then or equal to zero."); + } + else + { + random_timeout = tmp; + } + } else { return (-1); @@ -991,21 +1042,40 @@ static int rrd_shutdown (void) pthread_cond_signal (&queue_cond); pthread_mutex_unlock (&queue_lock); + if ((queue_thread_running != 0) + && ((queue_head != NULL) || (flushq_head != NULL))) + { + INFO ("rrdtool plugin: Shutting down the queue thread. " + "This may take a while."); + } + else if (queue_thread_running != 0) + { + INFO ("rrdtool plugin: Shutting down the queue thread."); + } + /* Wait for all the values to be written to disk before returning. */ - if (queue_thread != 0) + if (queue_thread_running != 0) { pthread_join (queue_thread, NULL); - queue_thread = 0; + memset (&queue_thread, 0, sizeof (queue_thread)); + queue_thread_running = 0; DEBUG ("rrdtool plugin: queue_thread exited."); } + /* TODO: Maybe it'd be a good idea to free the cache here.. */ + return (0); } /* int rrd_shutdown */ static int rrd_init (void) { + static int init_once = 0; int status; + if (init_once != 0) + return (0); + init_once = 1; + if (rrdcreate_config.stepsize < 0) rrdcreate_config.stepsize = 0; if (rrdcreate_config.heartbeat <= 0) @@ -1043,12 +1113,14 @@ static int rrd_init (void) pthread_mutex_unlock (&cache_lock); - status = pthread_create (&queue_thread, NULL, rrd_queue_thread, NULL); + status = pthread_create (&queue_thread, /* attr = */ NULL, + rrd_queue_thread, /* args = */ NULL); if (status != 0) { ERROR ("rrdtool plugin: Cannot create queue-thread."); return (-1); } + queue_thread_running = 1; DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %i;" " heartbeat = %i; rrarows = %i; xff = %lf;", @@ -1066,7 +1138,7 @@ void module_register (void) plugin_register_config ("rrdtool", rrd_config, config_keys, config_keys_num); plugin_register_init ("rrdtool", rrd_init); - plugin_register_write ("rrdtool", rrd_write); - plugin_register_flush ("rrdtool", rrd_flush); + plugin_register_write ("rrdtool", rrd_write, /* user_data = */ NULL); + plugin_register_flush ("rrdtool", rrd_flush, /* user_data = */ NULL); plugin_register_shutdown ("rrdtool", rrd_shutdown); }