/**
* collectd - src/rrdtool.c
- * Copyright (C) 2006,2007 Florian octo Forster
+ * Copyright (C) 2006-2008 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
};
typedef struct rrd_cache_s rrd_cache_t;
+enum rrd_queue_dir_e
+{
+ QUEUE_INSERT_FRONT,
+ QUEUE_INSERT_BACK
+};
+typedef enum rrd_queue_dir_e rrd_queue_dir_t;
+
struct rrd_queue_s
{
char *filename;
if (rra_num >= rra_max)
break;
- if (snprintf (buffer, sizeof (buffer), "RRA:%s:%3.1f:%u:%u",
+ if (ssnprintf (buffer, sizeof (buffer), "RRA:%s:%3.1f:%u:%u",
rra_types[j], xff,
cdp_len, cdp_num) >= sizeof (buffer))
{
sstrncpy (min, "U", sizeof (min));
}
else
- {
- snprintf (min, sizeof (min), "%lf", d->min);
- min[sizeof (min) - 1] = '\0';
- }
+ ssnprintf (min, sizeof (min), "%lf", d->min);
if (isnan (d->max))
{
sstrncpy (max, "U", sizeof (max));
}
else
- {
- snprintf (max, sizeof (max), "%lf", d->max);
- max[sizeof (max) - 1] = '\0';
- }
+ ssnprintf (max, sizeof (max), "%lf", d->max);
- status = snprintf (buffer, sizeof (buffer),
+ status = ssnprintf (buffer, sizeof (buffer),
"DS:%s:%s:%i:%s:%s",
d->name, type,
(heartbeat > 0) ? heartbeat : (2 * vl->interval),
#if HAVE_THREADSAFE_LIBRRD
static int srrd_create (char *filename, unsigned long pdp_step, time_t last_up,
- int argc, char **argv)
+ int argc, const char **argv)
{
int status;
return (status);
} /* int srrd_create */
-static int srrd_update (char *filename, char *template, int argc, char **argv)
+static int srrd_update (char *filename, char *template,
+ int argc, const char **argv)
{
int status;
#else /* !HAVE_THREADSAFE_LIBRRD */
static int srrd_create (char *filename, unsigned long pdp_step, time_t last_up,
- int argc, char **argv)
+ int argc, const char **argv)
{
int status;
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';
+ ssnprintf (pdp_step_str, sizeof (pdp_step_str), "%lu", pdp_step);
+ ssnprintf (last_up_str, sizeof (last_up_str), "%u", (unsigned int) last_up);
new_argv[0] = "create";
new_argv[1] = filename;
return (status);
} /* int srrd_create */
-static int srrd_update (char *filename, char *template, int argc, char **argv)
+static int srrd_update (char *filename, char *template,
+ int argc, const char **argv)
{
int status;
status = srrd_create (filename,
(stepsize > 0) ? stepsize : vl->interval,
vl->time - 10,
- argc, argv);
+ argc, (const char **)argv);
free (argv);
ds_free (ds_num, ds_def);
memset (buffer, '\0', buffer_len);
- status = snprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
+ status = ssnprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
if ((status < 1) || (status >= buffer_len))
return (-1);
offset = status;
return (-1);
if (ds->ds[i].type == DS_TYPE_COUNTER)
- status = snprintf (buffer + offset, buffer_len - offset,
+ status = ssnprintf (buffer + offset, buffer_len - offset,
":%llu", vl->values[i].counter);
else
- status = snprintf (buffer + offset, buffer_len - offset,
+ status = ssnprintf (buffer + offset, buffer_len - offset,
":%lf", vl->values[i].gauge);
if ((status < 1) || (status >= (buffer_len - offset)))
if (datadir != NULL)
{
- status = snprintf (buffer + offset, buffer_len - offset,
+ status = ssnprintf (buffer + offset, buffer_len - offset,
"%s/", datadir);
if ((status < 1) || (status >= buffer_len - offset))
return (-1);
offset += status;
}
- status = snprintf (buffer + offset, buffer_len - offset,
+ status = ssnprintf (buffer + offset, buffer_len - offset,
"%s/", vl->host);
if ((status < 1) || (status >= buffer_len - offset))
return (-1);
offset += status;
if (strlen (vl->plugin_instance) > 0)
- status = snprintf (buffer + offset, buffer_len - offset,
+ status = ssnprintf (buffer + offset, buffer_len - offset,
"%s-%s/", vl->plugin, vl->plugin_instance);
else
- status = snprintf (buffer + offset, buffer_len - offset,
+ status = ssnprintf (buffer + offset, buffer_len - offset,
"%s/", vl->plugin);
if ((status < 1) || (status >= buffer_len - offset))
return (-1);
offset += status;
if (strlen (vl->type_instance) > 0)
- status = snprintf (buffer + offset, buffer_len - offset,
- "%s-%s.rrd", ds->type, vl->type_instance);
+ status = ssnprintf (buffer + offset, buffer_len - offset,
+ "%s-%s.rrd", vl->type, vl->type_instance);
else
- status = snprintf (buffer + offset, buffer_len - offset,
- "%s.rrd", ds->type);
+ status = ssnprintf (buffer + offset, buffer_len - offset,
+ "%s.rrd", vl->type);
if ((status < 1) || (status >= buffer_len - offset))
return (-1);
offset += status;
pthread_mutex_unlock (&cache_lock);
/* Write the values to the RRD-file */
- srrd_update (queue_entry->filename, NULL, values_num, values);
+ 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);
return ((void *) 0);
} /* void *rrd_queue_thread */
-static int rrd_queue_cache_entry (const char *filename)
+static int rrd_queue_cache_entry (const char *filename, rrd_queue_dir_t dir)
{
- rrd_queue_t *queue_entry;
-
- queue_entry = (rrd_queue_t *) malloc (sizeof (rrd_queue_t));
- if (queue_entry == NULL)
- return (-1);
-
- queue_entry->filename = strdup (filename);
- if (queue_entry->filename == NULL)
- {
- free (queue_entry);
- return (-1);
- }
-
- queue_entry->next = NULL;
-
- pthread_mutex_lock (&queue_lock);
- if (queue_tail == NULL)
- queue_head = queue_entry;
- else
- queue_tail->next = queue_entry;
- queue_tail = queue_entry;
- pthread_cond_signal (&queue_cond);
- pthread_mutex_unlock (&queue_lock);
-
- DEBUG ("rrdtool plugin: Put `%s' into the update queue", filename);
-
- return (0);
+ rrd_queue_t *queue_entry;
+
+ queue_entry = (rrd_queue_t *) malloc (sizeof (rrd_queue_t));
+ if (queue_entry == NULL)
+ return (-1);
+
+ queue_entry->filename = strdup (filename);
+ if (queue_entry->filename == NULL)
+ {
+ free (queue_entry);
+ return (-1);
+ }
+
+ queue_entry->next = NULL;
+
+ pthread_mutex_lock (&queue_lock);
+ if (dir == QUEUE_INSERT_FRONT)
+ {
+ queue_entry->next = queue_head;
+ queue_head = queue_entry;
+ if (queue_tail == NULL)
+ queue_tail = queue_head;
+ }
+ else /* (dir == QUEUE_INSERT_BACK) */
+ {
+ if (queue_tail == NULL)
+ queue_head = queue_entry;
+ else
+ queue_tail->next = queue_entry;
+ queue_tail = queue_entry;
+ }
+ pthread_cond_signal (&queue_cond);
+ pthread_mutex_unlock (&queue_lock);
+
+ DEBUG ("rrdtool plugin: Put `%s' into the update queue", filename);
+
+ return (0);
} /* int rrd_queue_cache_entry */
+static int rrd_queue_move_to_front (const char *filename)
+{
+ rrd_queue_t *this;
+ rrd_queue_t *prev;
+
+ this = NULL;
+ prev = NULL;
+ pthread_mutex_lock (&queue_lock);
+ for (this = queue_head; this != NULL; this = this->next)
+ {
+ if (strcmp (this->filename, filename) == 0)
+ break;
+ prev = this;
+ }
+
+ /* Check if we found the entry and if it is NOT the first entry. */
+ if ((this != NULL) && (prev != NULL))
+ {
+ prev->next = this->next;
+ this->next = queue_head;
+ queue_head = this;
+ }
+ pthread_mutex_unlock (&queue_lock);
+
+ return (0);
+} /* int rrd_queue_move_to_front */
+
static void rrd_cache_flush (int timeout)
{
rrd_cache_t *rc;
continue;
else if (rc->values_num > 0)
{
- if (rrd_queue_cache_entry (key) == 0)
+ if (rrd_queue_cache_entry (key, QUEUE_INSERT_BACK) == 0)
rc->flags = FLAG_QUEUED;
}
else /* ancient and no values -> waste of memory */
cache_flush_last = now;
} /* void rrd_cache_flush */
+static int rrd_cache_flush_identifier (int timeout, const char *identifier)
+{
+ rrd_cache_t *rc;
+ time_t now;
+ int status;
+ char key[2048];
+
+ if (identifier == NULL)
+ {
+ rrd_cache_flush (timeout);
+ return (0);
+ }
+
+ now = time (NULL);
+
+ if (datadir == NULL)
+ snprintf (key, sizeof (key), "%s.rrd",
+ identifier);
+ else
+ snprintf (key, sizeof (key), "%s/%s.rrd",
+ datadir, identifier);
+ key[sizeof (key) - 1] = 0;
+
+ status = c_avl_get (cache, key, (void *) &rc);
+ if (status != 0)
+ {
+ WARNING ("rrdtool plugin: rrd_cache_flush_identifier: "
+ "c_avl_get (%s) failed. Does that file really exist?",
+ key);
+ return (status);
+ }
+
+ if (rc->flags == FLAG_QUEUED)
+ status = rrd_queue_move_to_front (key);
+ else if ((now - rc->first_value) < timeout)
+ status = 0;
+ else if (rc->values_num > 0)
+ {
+ status = rrd_queue_cache_entry (key, QUEUE_INSERT_FRONT);
+ if (status == 0)
+ rc->flags = FLAG_QUEUED;
+ }
+
+ return (status);
+} /* int rrd_cache_flush_identifier */
+
static int rrd_cache_insert (const char *filename,
const char *value, time_t value_time)
{
}
DEBUG ("rrdtool plugin: rrd_cache_insert: file = %s; "
- "values_num = %i; age = %u;",
+ "values_num = %i; age = %lu;",
filename, rc->values_num,
- rc->last_value - rc->first_value);
+ (unsigned long)(rc->last_value - rc->first_value));
if ((rc->last_value - rc->first_value) >= cache_timeout)
{
* the same time, ALWAYS lock `cache_lock' first! */
if (rc->flags != FLAG_QUEUED)
{
- if (rrd_queue_cache_entry (filename) == 0)
+ if (rrd_queue_cache_entry (filename, QUEUE_INSERT_BACK) == 0)
rc->flags = FLAG_QUEUED;
}
else
char values[512];
int status;
+ if (0 != strcmp (ds->type, vl->type)) {
+ ERROR ("rrdtool plugin: DS type does not match value list type");
+ return -1;
+ }
+
if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
return (-1);
return (status);
} /* int rrd_write */
-static int rrd_flush (const int timeout)
+static int rrd_flush (int timeout, const char *identifier)
{
pthread_mutex_lock (&cache_lock);
return (0);
}
- rrd_cache_flush (timeout);
+ rrd_cache_flush_identifier (timeout, identifier);
+
pthread_mutex_unlock (&cache_lock);
return (0);
} /* int rrd_flush */
projects / collectd.git / blobdiff