#include "common.h"
#include "utils_avltree.h"
+#include <rrd.h>
+
#if HAVE_PTHREAD_H
# include <pthread.h>
#endif
};
static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
+/* If datadir is zero, the daemon's basedir is used. If stepsize or heartbeat
+ * is zero a default, depending on the `interval' member of the value list is
+ * being used. */
static char *datadir = NULL;
static int stepsize = 0;
static int heartbeat = 0;
static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
+#if !HAVE_THREADSAFE_LIBRRD
+static pthread_mutex_t librrd_lock = PTHREAD_MUTEX_INITIALIZER;
+#endif
+
static int do_shutdown = 0;
/* * * * * * * * * *
* WARNING: Magic *
* * * * * * * * * */
-static int rra_get (char ***ret)
+
+static void rra_free (int rra_num, char **rra_def)
+{
+ int i;
+
+ for (i = 0; i < rra_num; i++)
+ {
+ sfree (rra_def[i]);
+ }
+ sfree (rra_def);
+} /* void rra_free */
+
+static int rra_get (char ***ret, const value_list_t *vl)
{
- static char **rra_def = NULL;
- static int rra_num = 0;
+ char **rra_def;
+ int rra_num;
int *rts;
int rts_num;
char buffer[64];
- if ((rra_num != 0) && (rra_def != NULL))
+ /* The stepsize we use here: If it is user-set, use it. If not, use the
+ * interval of the value-list. */
+ int ss;
+
+ if (rrarows <= 0)
{
- *ret = rra_def;
- return (rra_num);
+ *ret = NULL;
+ return (-1);
+ }
+
+ ss = (stepsize > 0) ? stepsize : vl->interval;
+ if (ss <= 0)
+ {
+ *ret = NULL;
+ return (-1);
}
/* Use the configured timespans or fall back to the built-in defaults */
if ((rra_def = (char **) malloc ((rra_max + 1) * sizeof (char *))) == NULL)
return (-1);
memset (rra_def, '\0', (rra_max + 1) * sizeof (char *));
-
- if ((stepsize <= 0) || (rrarows <= 0))
- {
- *ret = NULL;
- return (-1);
- }
+ rra_num = 0;
cdp_len = 0;
for (i = 0; i < rts_num; i++)
{
span = rts[i];
- if ((span / stepsize) < rrarows)
+ if ((span / ss) < rrarows)
continue;
if (cdp_len == 0)
cdp_len = 1;
else
cdp_len = (int) floor (((double) span)
- / ((double) (rrarows * stepsize)));
+ / ((double) (rrarows * ss)));
cdp_num = (int) ceil (((double) span)
- / ((double) (cdp_len * stepsize)));
+ / ((double) (cdp_len * ss)));
for (j = 0; j < rra_types_num; j++)
{
*ret = rra_def;
return (rra_num);
-}
+} /* int rra_get */
static void ds_free (int ds_num, char **ds_def)
{
free (ds_def);
}
-static int ds_get (char ***ret, const data_set_t *ds)
+static int ds_get (char ***ret, const data_set_t *ds, const value_list_t *vl)
{
char **ds_def;
int ds_num;
status = snprintf (buffer, sizeof (buffer),
"DS:%s:%s:%i:%s:%s",
- d->name, type, heartbeat,
+ d->name, type,
+ (heartbeat > 0) ? heartbeat : (2 * vl->interval),
min, max);
if ((status < 1) || (status >= sizeof (buffer)))
break;
return (ds_num);
}
-static int rrd_create_file (char *filename, const data_set_t *ds)
+#if HAVE_THREADSAFE_LIBRRD
+static int srrd_create (char *filename, unsigned long pdp_step, time_t last_up,
+ int argc, char **argv)
+{
+ int status;
+
+ optind = 0; /* bug in librrd? */
+ rrd_clear_error ();
+
+ status = rrd_create_r (filename, pdp_step, last_up, argc, argv);
+
+ if (status != 0)
+ {
+ WARNING ("rrdtool plugin: rrd_create_r (%s) failed: %s",
+ filename, rrd_get_error ());
+ }
+
+ return (status);
+} /* int srrd_create */
+
+static int srrd_update (char *filename, char *template, int argc, char **argv)
+{
+ int status;
+
+ optind = 0; /* bug in librrd? */
+ rrd_clear_error ();
+
+ status = rrd_update_r (filename, template, argc, argv);
+
+ if (status != 0)
+ {
+ WARNING ("rrdtool plugin: rrd_update_r (%s) failed: %s",
+ filename, rrd_get_error ());
+ }
+
+ return (status);
+} /* int srrd_update */
+/* #endif HAVE_THREADSAFE_LIBRRD */
+
+#else /* !HAVE_THREADSAFE_LIBRRD */
+static int srrd_create (char *filename, unsigned long pdp_step, time_t last_up,
+ int argc, char **argv)
+{
+ int status;
+
+ int new_argc;
+ char **new_argv;
+
+ char pdp_step_str[16];
+ char last_up_str[16];
+
+ new_argc = 6 + argc;
+ new_argv = (char **) malloc ((new_argc + 1) * sizeof (char *));
+ if (new_argv == NULL)
+ {
+ ERROR ("rrdtool plugin: malloc failed.");
+ return (-1);
+ }
+
+ if (last_up == 0)
+ last_up = time (NULL) - 10;
+
+ snprintf (pdp_step_str, sizeof (pdp_step_str), "%lu", pdp_step);
+ pdp_step_str[sizeof (pdp_step_str) - 1] = '\0';
+ snprintf (last_up_str, sizeof (last_up_str), "%u", (unsigned int) last_up);
+ last_up_str[sizeof (last_up_str) - 1] = '\0';
+
+ new_argv[0] = "create";
+ new_argv[1] = filename;
+ new_argv[2] = "-s";
+ new_argv[3] = pdp_step_str;
+ new_argv[4] = "-b";
+ new_argv[5] = last_up_str;
+
+ memcpy (new_argv + 6, argv, argc * sizeof (char *));
+ new_argv[new_argc] = NULL;
+
+ pthread_mutex_lock (&librrd_lock);
+ optind = 0; /* bug in librrd? */
+ rrd_clear_error ();
+
+ status = rrd_create (new_argc, new_argv);
+ pthread_mutex_unlock (&librrd_lock);
+
+ if (status != 0)
+ {
+ WARNING ("rrdtool plugin: rrd_create (%s) failed: %s",
+ filename, rrd_get_error ());
+ }
+
+ sfree (new_argv);
+
+ return (status);
+} /* int srrd_create */
+
+static int srrd_update (char *filename, char *template, int argc, char **argv)
+{
+ int status;
+
+ int new_argc;
+ char **new_argv;
+
+ assert (template == NULL);
+
+ new_argc = 2 + argc;
+ new_argv = (char **) malloc ((new_argc + 1) * sizeof (char *));
+ if (new_argv == NULL)
+ {
+ ERROR ("rrdtool plugin: malloc failed.");
+ return (-1);
+ }
+
+ new_argv[0] = "update";
+ new_argv[1] = filename;
+
+ memcpy (new_argv + 2, argv, argc * sizeof (char *));
+ new_argv[new_argc] = NULL;
+
+ pthread_mutex_lock (&librrd_lock);
+ optind = 0; /* bug in librrd? */
+ rrd_clear_error ();
+
+ status = rrd_update (new_argc, new_argv);
+ pthread_mutex_unlock (&librrd_lock);
+
+ if (status != 0)
+ {
+ WARNING ("rrdtool plugin: rrd_update_r failed: %s: %s",
+ argv[1], rrd_get_error ());
+ }
+
+ sfree (new_argv);
+
+ return (status);
+} /* int srrd_update */
+#endif /* !HAVE_THREADSAFE_LIBRRD */
+
+static int rrd_create_file (char *filename, const data_set_t *ds, const value_list_t *vl)
{
char **argv;
int argc;
int rra_num;
char **ds_def;
int ds_num;
- int i, j;
- char stepsize_str[16];
int status = 0;
if (check_create_dir (filename))
return (-1);
- if ((rra_num = rra_get (&rra_def)) < 1)
+ if ((rra_num = rra_get (&rra_def, vl)) < 1)
{
ERROR ("rrd_create_file failed: Could not calculate RRAs");
return (-1);
}
- if ((ds_num = ds_get (&ds_def, ds)) < 1)
+ if ((ds_num = ds_get (&ds_def, ds, vl)) < 1)
{
ERROR ("rrd_create_file failed: Could not calculate DSes");
return (-1);
}
- argc = ds_num + rra_num + 4;
+ argc = ds_num + rra_num;
if ((argv = (char **) malloc (sizeof (char *) * (argc + 1))) == NULL)
{
return (-1);
}
- status = snprintf (stepsize_str, sizeof (stepsize_str),
- "%i", stepsize);
- if ((status < 1) || (status >= sizeof (stepsize_str)))
- {
- ERROR ("rrdtool plugin: snprintf failed.");
- return (-1);
- }
+ memcpy (argv, ds_def, ds_num * sizeof (char *));
+ memcpy (argv + ds_num, rra_def, rra_num * sizeof (char *));
+ argv[ds_num + rra_num] = NULL;
- argv[0] = "create";
- argv[1] = filename;
- argv[2] = "-s";
- argv[3] = stepsize_str;
-
- j = 4;
- for (i = 0; i < ds_num; i++)
- argv[j++] = ds_def[i];
- for (i = 0; i < rra_num; i++)
- argv[j++] = rra_def[i];
- argv[j] = NULL;
-
- optind = 0; /* bug in librrd? */
- rrd_clear_error ();
- if (rrd_create (argc, argv) == -1)
- {
- ERROR ("rrd_create failed: %s: %s", filename, rrd_get_error ());
- status = -1;
- }
+ assert (vl->time > 10);
+ status = srrd_create (filename,
+ (stepsize > 0) ? stepsize : vl->interval,
+ vl->time - 10,
+ argc, argv);
free (argv);
ds_free (ds_num, ds_def);
+ rra_free (rra_num, rra_def);
return (status);
}
int status;
int i;
- memset (buffer, '\0', sizeof (buffer_len));
+ memset (buffer, '\0', buffer_len);
status = snprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
if ((status < 1) || (status >= buffer_len))
return (0);
} /* int value_list_to_filename */
-static int rrd_write_to_file (char *filename, char **values, int values_num)
-{
- char **argv;
- int argc;
- int status;
-
- if (values_num < 1)
- return (0);
-
- argc = values_num + 2;
- argv = (char **) malloc ((argc + 1) * sizeof (char *));
- if (argv == NULL)
- return (-1);
-
- argv[0] = "update";
- argv[1] = filename;
- memcpy (argv + 2, values, values_num * sizeof (char *));
- argv[argc] = NULL;
-
- DEBUG ("rrd_update (argc = %i, argv = %p)", argc, (void *) argv);
-
- optind = 0; /* bug in librrd? */
- rrd_clear_error ();
- status = rrd_update (argc, argv);
- if (status != 0)
- {
- WARNING ("rrd_update failed: %s: %s",
- filename, rrd_get_error ());
- status = -1;
- }
-
- sfree (argv);
-
- return (status);
-} /* int rrd_write_cache_entry */
-
static void *rrd_queue_thread (void *data)
{
while (42)
pthread_mutex_unlock (&cache_lock);
/* Write the values to the RRD-file */
- rrd_write_to_file (queue_entry->filename, values, values_num);
+ srrd_update (queue_entry->filename, NULL, values_num, values);
+ DEBUG ("rrdtool plugin: queue thread: Wrote %i values to %s",
+ values_num, queue_entry->filename);
for (i = 0; i < values_num; i++)
{
avl_iterator_t *iter;
int i;
- DEBUG ("Flushing cache, timeout = %i", timeout);
+ DEBUG ("rrdtool plugin: Flushing cache, timeout = %i", timeout);
now = time (NULL);
iter = avl_get_iterator (cache);
while (avl_iterator_next (iter, (void *) &key, (void *) &rc) == 0)
{
- DEBUG ("key = %s; age = %i;", key, now - rc->first_value);
-
if (rc->flags == FLAG_QUEUED)
continue;
else if ((now - rc->first_value) < timeout)
{
if (avl_remove (cache, keys[i], (void *) &key, (void *) &rc) != 0)
{
- DEBUG ("avl_remove (%s) failed.", keys[i]);
+ DEBUG ("rrdtool plugin: avl_remove (%s) failed.", keys[i]);
continue;
}
} /* for (i = 0..keys_num) */
free (keys);
- DEBUG ("Flushed %i value(s)", keys_num);
cache_flush_last = now;
} /* void rrd_cache_flush */
avl_insert (cache, cache_key, rc);
}
- DEBUG ("rrd_cache_insert (%s, %s, %u) = %p", filename, value,
- (unsigned int) value_time, (void *) rc);
+ DEBUG ("rrdtool plugin: rrd_cache_insert: file = %s; "
+ "values_num = %i; age = %u;",
+ filename, rc->values_num,
+ rc->last_value - rc->first_value);
if ((rc->last_value - rc->first_value) >= cache_timeout)
{
{
if (errno == ENOENT)
{
- if (rrd_create_file (filename, ds))
+ if (rrd_create_file (filename, ds, vl))
return (-1);
}
else
}
else if (strcasecmp ("StepSize", key) == 0)
{
- int tmp = atoi (value);
- if (tmp <= 0)
- {
- fprintf (stderr, "rrdtool: `StepSize' must "
- "be greater than 0.\n");
- return (1);
- }
- stepsize = tmp;
+ stepsize = atoi (value);
+ if (stepsize < 0)
+ stepsize = 0;
}
else if (strcasecmp ("HeartBeat", key) == 0)
{
- int tmp = atoi (value);
- if (tmp <= 0)
- {
- fprintf (stderr, "rrdtool: `HeartBeat' must "
- "be greater than 0.\n");
- return (1);
- }
- heartbeat = tmp;
+ heartbeat = atoi (value);
+ if (heartbeat < 0)
+ heartbeat = 0;
}
else if (strcasecmp ("RRARows", key) == 0)
{
pthread_cond_signal (&queue_cond);
pthread_mutex_unlock (&queue_lock);
+ /* Wait for all the values to be written to disk before returning. */
+ if (queue_thread != 0)
+ {
+ pthread_join (queue_thread, NULL);
+ queue_thread = 0;
+ DEBUG ("rrdtool plugin: queue_thread exited.");
+ }
+
return (0);
} /* int rrd_shutdown */
{
int status;
- if (stepsize <= 0)
- stepsize = interval_g;
+ if (stepsize < 0)
+ stepsize = 0;
if (heartbeat <= 0)
- heartbeat = 2 * interval_g;
+ {
+ if (stepsize > 0)
+ heartbeat = 2 * stepsize;
+ else
+ heartbeat = 0;
+ }
- if (heartbeat < interval_g)
+ if ((heartbeat > 0) && (heartbeat < interval_g))
WARNING ("rrdtool plugin: Your `heartbeat' is "
"smaller than your `interval'. This will "
"likely cause problems.");
- else if (stepsize < interval_g)
+ else if ((stepsize > 0) && (stepsize < interval_g))
WARNING ("rrdtool plugin: Your `stepsize' is "
"smaller than your `interval'. This will "
"create needlessly big RRD-files.");
projects / collectd.git / blobdiff