Code

Merge remote-tracking branch 'github/pr/387'
[collectd.git] / src / rrdtool.c
index 9ef0fa1f032f40c495e8565bd8877771b0d0d899..2f28329f0fbef43f17c70b3b7136a6b4f187e2b8 100644 (file)
@@ -1,6 +1,6 @@
 /**
  * collectd - src/rrdtool.c
- * Copyright (C) 2006-2008  Florian octo Forster
+ * Copyright (C) 2006-2013  Florian octo Forster
  * Copyright (C) 2008-2008  Sebastian Harl
  * Copyright (C) 2009       Mariusz Gronczewski
  *
@@ -18,7 +18,7 @@
  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  *
  * Authors:
- *   Florian octo Forster <octo at verplant.org>
+ *   Florian octo Forster <octo at collectd.org>
  *   Sebastian Harl <sh at tokkee.org>
  *   Mariusz Gronczewski <xani666 at gmail.com>
  **/
@@ -27,6 +27,7 @@
 #include "plugin.h"
 #include "common.h"
 #include "utils_avltree.h"
+#include "utils_random.h"
 #include "utils_rrdcreate.h"
 
 #include <rrd.h>
@@ -75,6 +76,7 @@ static const char *config_keys[] =
 {
        "CacheTimeout",
        "CacheFlush",
+       "CreateFilesAsync",
        "DataDir",
        "StepSize",
        "HeartBeat",
@@ -102,14 +104,16 @@ static rrdcreate_config_t rrdcreate_config =
        /* timespans_num = */ 0,
 
        /* consolidation_functions = */ NULL,
-       /* consolidation_functions_num = */ 0
+       /* consolidation_functions_num = */ 0,
+
+       /* async = */ 0
 };
 
 /* XXX: If you need to lock both, cache_lock and queue_lock, at the same time,
  * ALWAYS lock `cache_lock' first! */
 static cdtime_t    cache_timeout = 0;
 static cdtime_t    cache_flush_timeout = 0;
-static cdtime_t    random_timeout = 1;
+static cdtime_t    random_timeout = TIME_T_TO_CDTIME_T (1);
 static cdtime_t    cache_flush_last;
 static c_avl_tree_t *cache = NULL;
 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
@@ -185,7 +189,7 @@ static int srrd_update (char *filename, char *template,
        if (status != 0)
        {
                WARNING ("rrdtool plugin: rrd_update_r failed: %s: %s",
-                               argv[1], rrd_get_error ());
+                               filename, rrd_get_error ());
        }
 
        sfree (new_argv);
@@ -194,7 +198,7 @@ static int srrd_update (char *filename, char *template,
 } /* int srrd_update */
 #endif /* !HAVE_THREADSAFE_LIBRRD */
 
-static int value_list_to_string (char *buffer, int buffer_len,
+static int value_list_to_string_multiple (char *buffer, int buffer_len,
                const data_set_t *ds, const value_list_t *vl)
 {
        int offset;
@@ -238,49 +242,67 @@ static int value_list_to_string (char *buffer, int buffer_len,
        } /* for ds->ds_num */
 
        return (0);
-} /* int value_list_to_string */
+} /* int value_list_to_string_multiple */
 
-static int value_list_to_filename (char *buffer, int buffer_len,
-               const data_set_t __attribute__((unused)) *ds, const value_list_t *vl)
+static int value_list_to_string (char *buffer, int buffer_len,
+               const data_set_t *ds, const value_list_t *vl)
 {
-       int offset = 0;
        int status;
+       time_t tt;
 
-       if (datadir != NULL)
+       if (ds->ds_num != 1)
+               return (value_list_to_string_multiple (buffer, buffer_len,
+                                       ds, vl));
+
+       tt = CDTIME_T_TO_TIME_T (vl->time);
+       switch (ds->ds[0].type)
        {
-               status = ssnprintf (buffer + offset, buffer_len - offset,
-                               "%s/", datadir);
-               if ((status < 1) || (status >= buffer_len - offset))
-                       return (-1);
-               offset += status;
+               case DS_TYPE_DERIVE:
+                       status = ssnprintf (buffer, buffer_len, "%u:%"PRIi64,
+                               (unsigned) tt, vl->values[0].derive);
+                       break;
+               case DS_TYPE_GAUGE:
+                       status = ssnprintf (buffer, buffer_len, "%u:%lf",
+                               (unsigned) tt, vl->values[0].gauge);
+                       break;
+               case DS_TYPE_COUNTER:
+                       status = ssnprintf (buffer, buffer_len, "%u:%llu",
+                               (unsigned) tt, vl->values[0].counter);
+                       break;
+               case DS_TYPE_ABSOLUTE:
+                       status = ssnprintf (buffer, buffer_len, "%u:%"PRIu64,
+                               (unsigned) tt, vl->values[0].absolute);
+                       break;
+               default:
+                       return (EINVAL);
        }
 
-       status = ssnprintf (buffer + offset, buffer_len - offset,
-                       "%s/", vl->host);
-       if ((status < 1) || (status >= buffer_len - offset))
-               return (-1);
-       offset += status;
+       if ((status < 1) || (status >= buffer_len))
+               return (ENOMEM);
 
-       if (strlen (vl->plugin_instance) > 0)
-               status = ssnprintf (buffer + offset, buffer_len - offset,
-                               "%s-%s/", vl->plugin, vl->plugin_instance);
-       else
-               status = ssnprintf (buffer + offset, buffer_len - offset,
-                               "%s/", vl->plugin);
-       if ((status < 1) || (status >= buffer_len - offset))
-               return (-1);
-       offset += status;
+       return (0);
+} /* int value_list_to_string */
 
-       if (strlen (vl->type_instance) > 0)
-               status = ssnprintf (buffer + offset, buffer_len - offset,
-                               "%s-%s.rrd", vl->type, vl->type_instance);
-       else
-               status = ssnprintf (buffer + offset, buffer_len - offset,
-                               "%s.rrd", vl->type);
-       if ((status < 1) || (status >= buffer_len - offset))
-               return (-1);
-       offset += status;
+static int value_list_to_filename (char *buffer, size_t buffer_size,
+               value_list_t const *vl)
+{
+       char const suffix[] = ".rrd";
+       int status;
+       size_t len;
+
+       status = FORMAT_VL (buffer, buffer_size, vl);
+       if (status != 0)
+               return (status);
 
+       len = strlen (buffer);
+       assert (len < buffer_size);
+       buffer += len;
+       buffer_size -= len;
+
+       if (buffer_size <= sizeof (suffix))
+               return (ENOMEM);
+
+       memcpy (buffer, suffix, sizeof (suffix));
        return (0);
 } /* int value_list_to_filename */
 
@@ -305,7 +327,7 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
 
                 pthread_mutex_lock (&queue_lock);
                 /* Wait for values to arrive */
-                while (true)
+                while (42)
                 {
                   struct timespec ts_wait;
 
@@ -344,7 +366,7 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
                       &ts_wait);
                   if (status == ETIMEDOUT)
                     break;
-                } /* while (true) */
+                } /* while (42) */
 
                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
                  * the same time, ALWAYS lock `cache_lock' first! */
@@ -512,10 +534,11 @@ static int rrd_queue_dequeue (const char *filename,
   return (0);
 } /* int rrd_queue_dequeue */
 
-static void rrd_cache_flush (int timeout)
+/* XXX: You must hold "cache_lock" when calling this function! */
+static void rrd_cache_flush (cdtime_t timeout)
 {
        rrd_cache_t *rc;
-       time_t       now;
+       cdtime_t     now;
 
        char **keys = NULL;
        int    keys_num = 0;
@@ -524,9 +547,11 @@ static void rrd_cache_flush (int timeout)
        c_avl_iterator_t *iter;
        int i;
 
-       DEBUG ("rrdtool plugin: Flushing cache, timeout = %i", timeout);
+       DEBUG ("rrdtool plugin: Flushing cache, timeout = %.3f",
+                       CDTIME_T_TO_DOUBLE (timeout));
 
-       now = time (NULL);
+       now = cdtime ();
+       timeout = TIME_T_TO_CDTIME_T (timeout);
 
        /* Build a list of entries to be flushed */
        iter = c_avl_get_iterator (cache);
@@ -534,7 +559,9 @@ static void rrd_cache_flush (int timeout)
        {
                if (rc->flags != FLAG_NONE)
                        continue;
-               else if ((now - rc->first_value) < timeout)
+               /* timeout == 0  =>  flush everything */
+               else if ((timeout != 0)
+                               && ((now - rc->first_value) < timeout))
                        continue;
                else if (rc->values_num > 0)
                {
@@ -587,10 +614,11 @@ static void rrd_cache_flush (int timeout)
        cache_flush_last = now;
 } /* void rrd_cache_flush */
 
-static int rrd_cache_flush_identifier (int timeout, const char *identifier)
+static int rrd_cache_flush_identifier (cdtime_t timeout,
+    const char *identifier)
 {
   rrd_cache_t *rc;
-  time_t now;
+  cdtime_t now;
   int status;
   char key[2048];
 
@@ -600,7 +628,7 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier)
     return (0);
   }
 
-  now = time (NULL);
+  now = cdtime ();
 
   if (datadir == NULL)
     snprintf (key, sizeof (key), "%s.rrd",
@@ -646,32 +674,24 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier)
 
 static int64_t rrd_get_random_variation (void)
 {
-  double dbl_timeout;
-  cdtime_t ctm_timeout;
-  double rand_fact;
-  _Bool negative;
-  int64_t ret;
+  long min;
+  long max;
 
   if (random_timeout <= 0)
     return (0);
 
-  /* This seems a bit complicated, but "random_timeout" is likely larger than
-   * RAND_MAX, so we can't simply use modulo here. */
-  dbl_timeout = CDTIME_T_TO_DOUBLE (random_timeout);
-  rand_fact = ((double) random ())
-    / ((double) RAND_MAX);
-  negative = (_Bool) (random () % 2);
-
-  ctm_timeout = DOUBLE_TO_CDTIME_T (dbl_timeout * rand_fact);
-
-  ret = (int64_t) ctm_timeout;
-  if (negative)
-    ret *= -1;
+  /* Assure that "cache_timeout + random_variation" is never negative. */
+  if (random_timeout > cache_timeout)
+  {
+         INFO ("rrdtool plugin: Adjusting \"RandomTimeout\" to %.3f seconds.",
+                         CDTIME_T_TO_DOUBLE (cache_timeout));
+         random_timeout = cache_timeout;
+  }
 
-  DEBUG ("rrdtool plugin: random_variation = %.3f s",
-      (negative ? -1.0 : 1.0) * dbl_timeout * rand_fact);
+  max = (long) (random_timeout / 2);
+  min = max - ((long) random_timeout);
 
-  return (ret);
+  return ((int64_t) cdrand_range (min, max));
 } /* int64_t rrd_get_random_variation */
 
 static int rrd_cache_insert (const char *filename,
@@ -703,7 +723,7 @@ static int rrd_cache_insert (const char *filename,
                rc->values = NULL;
                rc->first_value = 0;
                rc->last_value = 0;
-               rc->random_variation = 0;
+               rc->random_variation = rrd_get_random_variation ();
                rc->flags = FLAG_NONE;
                new_rc = 1;
        }
@@ -774,7 +794,7 @@ static int rrd_cache_insert (const char *filename,
                        filename, rc->values_num,
                        CDTIME_T_TO_DOUBLE (rc->last_value - rc->first_value));
 
-       if ((rc->last_value + rc->random_variation - rc->first_value) >= cache_timeout)
+       if ((rc->last_value - rc->first_value) >= (cache_timeout + rc->random_variation))
        {
                /* XXX: If you need to lock both, cache_lock and queue_lock, at
                 * the same time, ALWAYS lock `cache_lock' first! */
@@ -885,7 +905,7 @@ static int rrd_write (const data_set_t *ds, const value_list_t *vl,
                return -1;
        }
 
-       if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
+       if (value_list_to_filename (filename, sizeof (filename), vl) != 0)
                return (-1);
 
        if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
@@ -899,6 +919,8 @@ static int rrd_write (const data_set_t *ds, const value_list_t *vl,
                                        ds, vl, &rrdcreate_config);
                        if (status != 0)
                                return (-1);
+                       else if (rrdcreate_config.async)
+                               return (0);
                }
                else
                {
@@ -921,8 +943,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,
-               user_data_t __attribute__((unused)) *user_data)
+static int rrd_flush (cdtime_t timeout, const char *identifier,
+               __attribute__((unused)) user_data_t *user_data)
 {
        pthread_mutex_lock (&cache_lock);
 
@@ -987,7 +1009,7 @@ static int rrd_config (const char *key, const char *value)
        }
        else if (strcasecmp ("StepSize", key) == 0)
        {
-               int temp = atoi (value);
+               unsigned long temp = strtoul (value, NULL, 0);
                if (temp > 0)
                        rrdcreate_config.stepsize = temp;
        }
@@ -997,6 +1019,13 @@ static int rrd_config (const char *key, const char *value)
                if (temp > 0)
                        rrdcreate_config.heartbeat = temp;
        }
+       else if (strcasecmp ("CreateFilesAsync", key) == 0)
+       {
+               if (IS_TRUE (value))
+                       rrdcreate_config.async = 1;
+               else
+                       rrdcreate_config.async = 0;
+       }
        else if (strcasecmp ("RRARows", key) == 0)
        {
                int tmp = atoi (value);
@@ -1095,7 +1124,7 @@ static int rrd_config (const char *key, const char *value)
                }
                else
                {
-                       cache_timeout = DOUBLE_TO_CDTIME_T (tmp);
+                       random_timeout = DOUBLE_TO_CDTIME_T (tmp);
                }
        }
        else
@@ -1108,7 +1137,7 @@ static int rrd_config (const char *key, const char *value)
 static int rrd_shutdown (void)
 {
        pthread_mutex_lock (&cache_lock);
-       rrd_cache_flush (-1);
+       rrd_cache_flush (0);
        pthread_mutex_unlock (&cache_lock);
 
        pthread_mutex_lock (&queue_lock);
@@ -1150,22 +1179,9 @@ static int rrd_init (void)
                return (0);
        init_once = 1;
 
-       if (rrdcreate_config.stepsize < 0)
-               rrdcreate_config.stepsize = 0;
        if (rrdcreate_config.heartbeat <= 0)
                rrdcreate_config.heartbeat = 2 * rrdcreate_config.stepsize;
 
-       if ((rrdcreate_config.heartbeat > 0)
-                       && (rrdcreate_config.heartbeat < interval_g))
-               WARNING ("rrdtool plugin: Your `heartbeat' is "
-                               "smaller than your `interval'. This will "
-                               "likely cause problems.");
-       else if ((rrdcreate_config.stepsize > 0)
-                       && (rrdcreate_config.stepsize < interval_g))
-               WARNING ("rrdtool plugin: Your `stepsize' is "
-                               "smaller than your `interval'. This will "
-                               "create needlessly big RRD-files.");
-
        /* Set the cache up */
        pthread_mutex_lock (&cache_lock);
 
@@ -1176,10 +1192,9 @@ static int rrd_init (void)
                return (-1);
        }
 
-       cache_flush_last = time (NULL);
-       if (cache_timeout < 2)
+       cache_flush_last = cdtime ();
+       if (cache_timeout == 0)
        {
-               cache_timeout = 0;
                cache_flush_timeout = 0;
        }
        else if (cache_flush_timeout < cache_timeout)
@@ -1187,7 +1202,7 @@ static int rrd_init (void)
 
        pthread_mutex_unlock (&cache_lock);
 
-       status = pthread_create (&queue_thread, /* attr = */ NULL,
+       status = plugin_thread_create (&queue_thread, /* attr = */ NULL,
                        rrd_queue_thread, /* args = */ NULL);
        if (status != 0)
        {
@@ -1196,7 +1211,7 @@ static int rrd_init (void)
        }
        queue_thread_running = 1;
 
-       DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %i;"
+       DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %lu;"
                        " heartbeat = %i; rrarows = %i; xff = %lf;",
                        (datadir == NULL) ? "(null)" : datadir,
                        rrdcreate_config.stepsize,