src/rrdtool.c

   1 /**
   2  * collectd - src/rrdtool.c
   3  * Copyright (C) 2006-2013  Florian octo Forster
   4  * Copyright (C) 2008-2008  Sebastian Harl
   5  * Copyright (C) 2009       Mariusz Gronczewski
   6  *
   7  * This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the GNU General Public License as published by the
   9  * Free Software Foundation; only version 2 of the License is applicable.
  10  *
  11  * This program is distributed in the hope that it will be useful, but
  12  * WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License along
  17  * with this program; if not, write to the Free Software Foundation, Inc.,
  18  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  19  *
  20  * Authors:
  21  *   Florian octo Forster <octo at collectd.org>
  22  *   Sebastian Harl <sh at tokkee.org>
  23  *   Mariusz Gronczewski <xani666 at gmail.com>
  24  **/
  25
  26 #include "collectd.h"
  27 #include "plugin.h"
  28 #include "common.h"
  29 #include "utils_avltree.h"
  30 #include "utils_rrdcreate.h"
  31
  32 #include <rrd.h>
  33
  34 #if HAVE_PTHREAD_H
  35 # include <pthread.h>
  36 #endif
  37
  38 /*
  39  * Private types
  40  */
  41 struct rrd_cache_s
  42 {
  43         int      values_num;
  44         char   **values;
  45         cdtime_t first_value;
  46         cdtime_t last_value;
  47         int64_t  random_variation;
  48         enum
  49         {
  50                 FLAG_NONE   = 0x00,
  51                 FLAG_QUEUED = 0x01,
  52                 FLAG_FLUSHQ = 0x02
  53         } flags;
  54 };
  55 typedef struct rrd_cache_s rrd_cache_t;
  56
  57 enum rrd_queue_dir_e
  58 {
  59   QUEUE_INSERT_FRONT,
  60   QUEUE_INSERT_BACK
  61 };
  62 typedef enum rrd_queue_dir_e rrd_queue_dir_t;
  63
  64 struct rrd_queue_s
  65 {
  66         char *filename;
  67         struct rrd_queue_s *next;
  68 };
  69 typedef struct rrd_queue_s rrd_queue_t;
  70
  71 /*
  72  * Private variables
  73  */
  74 static const char *config_keys[] =
  75 {
  76         "CacheTimeout",
  77         "CacheFlush",
  78         "DataDir",
  79         "StepSize",
  80         "HeartBeat",
  81         "RRARows",
  82         "RRATimespan",
  83         "XFF",
  84         "WritesPerSecond",
  85         "RandomTimeout"
  86 };
  87 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
  88
  89 /* If datadir is zero, the daemon's basedir is used. If stepsize or heartbeat
  90  * is zero a default, depending on the `interval' member of the value list is
  91  * being used. */
  92 static char *datadir   = NULL;
  93 static double write_rate = 0.0;
  94 static rrdcreate_config_t rrdcreate_config =
  95 {
  96         /* stepsize = */ 0,
  97         /* heartbeat = */ 0,
  98         /* rrarows = */ 1200,
  99         /* xff = */ 0.1,
 100
 101         /* timespans = */ NULL,
 102         /* timespans_num = */ 0,
 103
 104         /* consolidation_functions = */ NULL,
 105         /* consolidation_functions_num = */ 0
 106 };
 107
 108 /* XXX: If you need to lock both, cache_lock and queue_lock, at the same time,
 109  * ALWAYS lock `cache_lock' first! */
 110 static cdtime_t    cache_timeout = 0;
 111 static cdtime_t    cache_flush_timeout = 0;
 112 static cdtime_t    random_timeout = TIME_T_TO_CDTIME_T (1);
 113 static cdtime_t    cache_flush_last;
 114 static c_avl_tree_t *cache = NULL;
 115 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 116
 117 static rrd_queue_t    *queue_head = NULL;
 118 static rrd_queue_t    *queue_tail = NULL;
 119 static rrd_queue_t    *flushq_head = NULL;
 120 static rrd_queue_t    *flushq_tail = NULL;
 121 static pthread_t       queue_thread;
 122 static int             queue_thread_running = 1;
 123 static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER;
 124 static pthread_cond_t  queue_cond = PTHREAD_COND_INITIALIZER;
 125
 126 #if !HAVE_THREADSAFE_LIBRRD
 127 static pthread_mutex_t librrd_lock = PTHREAD_MUTEX_INITIALIZER;
 128 #endif
 129
 130 static int do_shutdown = 0;
 131
 132 #if HAVE_THREADSAFE_LIBRRD
 133 static int srrd_update (char *filename, char *template,
 134                 int argc, const char **argv)
 135 {
 136         int status;
 137
 138         optind = 0; /* bug in librrd? */
 139         rrd_clear_error ();
 140
 141         status = rrd_update_r (filename, template, argc, (void *) argv);
 142
 143         if (status != 0)
 144         {
 145                 WARNING ("rrdtool plugin: rrd_update_r (%s) failed: %s",
 146                                 filename, rrd_get_error ());
 147         }
 148
 149         return (status);
 150 } /* int srrd_update */
 151 /* #endif HAVE_THREADSAFE_LIBRRD */
 152
 153 #else /* !HAVE_THREADSAFE_LIBRRD */
 154 static int srrd_update (char *filename, char *template,
 155                 int argc, const char **argv)
 156 {
 157         int status;
 158
 159         int new_argc;
 160         char **new_argv;
 161
 162         assert (template == NULL);
 163
 164         new_argc = 2 + argc;
 165         new_argv = (char **) malloc ((new_argc + 1) * sizeof (char *));
 166         if (new_argv == NULL)
 167         {
 168                 ERROR ("rrdtool plugin: malloc failed.");
 169                 return (-1);
 170         }
 171
 172         new_argv[0] = "update";
 173         new_argv[1] = filename;
 174
 175         memcpy (new_argv + 2, argv, argc * sizeof (char *));
 176         new_argv[new_argc] = NULL;
 177
 178         pthread_mutex_lock (&librrd_lock);
 179         optind = 0; /* bug in librrd? */
 180         rrd_clear_error ();
 181
 182         status = rrd_update (new_argc, new_argv);
 183         pthread_mutex_unlock (&librrd_lock);
 184
 185         if (status != 0)
 186         {
 187                 WARNING ("rrdtool plugin: rrd_update_r failed: %s: %s",
 188                                 filename, rrd_get_error ());
 189         }
 190
 191         sfree (new_argv);
 192
 193         return (status);
 194 } /* int srrd_update */
 195 #endif /* !HAVE_THREADSAFE_LIBRRD */
 196
 197 static int value_list_to_string (char *buffer, int buffer_len,
 198                 const data_set_t *ds, const value_list_t *vl)
 199 {
 200         int offset;
 201         int status;
 202         time_t tt;
 203         int i;
 204
 205         memset (buffer, '\0', buffer_len);
 206
 207         tt = CDTIME_T_TO_TIME_T (vl->time);
 208         status = ssnprintf (buffer, buffer_len, "%u", (unsigned int) tt);
 209         if ((status < 1) || (status >= buffer_len))
 210                 return (-1);
 211         offset = status;
 212
 213         for (i = 0; i < ds->ds_num; i++)
 214         {
 215                 if ((ds->ds[i].type != DS_TYPE_COUNTER)
 216                                 && (ds->ds[i].type != DS_TYPE_GAUGE)
 217                                 && (ds->ds[i].type != DS_TYPE_DERIVE)
 218                                 && (ds->ds[i].type != DS_TYPE_ABSOLUTE))
 219                         return (-1);
 220
 221                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 222                         status = ssnprintf (buffer + offset, buffer_len - offset,
 223                                         ":%llu", vl->values[i].counter);
 224                 else if (ds->ds[i].type == DS_TYPE_GAUGE)
 225                         status = ssnprintf (buffer + offset, buffer_len - offset,
 226                                         ":%lf", vl->values[i].gauge);
 227                 else if (ds->ds[i].type == DS_TYPE_DERIVE)
 228                         status = ssnprintf (buffer + offset, buffer_len - offset,
 229                                         ":%"PRIi64, vl->values[i].derive);
 230                 else /*if (ds->ds[i].type == DS_TYPE_ABSOLUTE) */
 231                         status = ssnprintf (buffer + offset, buffer_len - offset,
 232                                         ":%"PRIu64, vl->values[i].absolute);
 233
 234                 if ((status < 1) || (status >= (buffer_len - offset)))
 235                         return (-1);
 236
 237                 offset += status;
 238         } /* for ds->ds_num */
 239
 240         return (0);
 241 } /* int value_list_to_string */
 242
 243 static int value_list_to_filename (char *buffer, size_t buffer_size,
 244                 value_list_t const *vl)
 245 {
 246         char const suffix[] = ".rrd";
 247         int status;
 248         size_t len;
 249
 250         if (datadir != NULL)
 251         {
 252                 size_t datadir_len = strlen (datadir) + 1;
 253
 254                 if (datadir_len >= buffer_size)
 255                         return (ENOMEM);
 256
 257                 sstrncpy (buffer, datadir, buffer_size);
 258                 buffer[datadir_len - 1] = '/';
 259                 buffer[datadir_len] = 0;
 260
 261                 buffer += datadir_len;
 262                 buffer_size -= datadir_len;
 263         }
 264
 265         status = FORMAT_VL (buffer, buffer_size, vl);
 266         if (status != 0)
 267                 return (status);
 268
 269         len = strlen (buffer);
 270         assert (len < buffer_size);
 271         buffer += len;
 272         buffer_size -= len;
 273
 274         if (buffer_size <= sizeof (suffix))
 275                 return (ENOMEM);
 276
 277         memcpy (buffer, suffix, sizeof (suffix));
 278         return (0);
 279 } /* int value_list_to_filename */
 280
 281 static void *rrd_queue_thread (void __attribute__((unused)) *data)
 282 {
 283         struct timeval tv_next_update;
 284         struct timeval tv_now;
 285
 286         gettimeofday (&tv_next_update, /* timezone = */ NULL);
 287
 288         while (42)
 289         {
 290                 rrd_queue_t *queue_entry;
 291                 rrd_cache_t *cache_entry;
 292                 char **values;
 293                 int    values_num;
 294                 int    status;
 295                 int    i;
 296
 297                 values = NULL;
 298                 values_num = 0;
 299
 300                 pthread_mutex_lock (&queue_lock);
 301                 /* Wait for values to arrive */
 302                 while (42)
 303                 {
 304                   struct timespec ts_wait;
 305
 306                   while ((flushq_head == NULL) && (queue_head == NULL)
 307                       && (do_shutdown == 0))
 308                     pthread_cond_wait (&queue_cond, &queue_lock);
 309
 310                   if ((flushq_head == NULL) && (queue_head == NULL))
 311                     break;
 312
 313                   /* Don't delay if there's something to flush */
 314                   if (flushq_head != NULL)
 315                     break;
 316
 317                   /* Don't delay if we're shutting down */
 318                   if (do_shutdown != 0)
 319                     break;
 320
 321                   /* Don't delay if no delay was configured. */
 322                   if (write_rate <= 0.0)
 323                     break;
 324
 325                   gettimeofday (&tv_now, /* timezone = */ NULL);
 326                   status = timeval_cmp (tv_next_update, tv_now, NULL);
 327                   /* We're good to go */
 328                   if (status <= 0)
 329                     break;
 330
 331                   /* We're supposed to wait a bit with this update, so we'll
 332                    * wait for the next addition to the queue or to the end of
 333                    * the wait period - whichever comes first. */
 334                   ts_wait.tv_sec = tv_next_update.tv_sec;
 335                   ts_wait.tv_nsec = 1000 * tv_next_update.tv_usec;
 336
 337                   status = pthread_cond_timedwait (&queue_cond, &queue_lock,
 338                       &ts_wait);
 339                   if (status == ETIMEDOUT)
 340                     break;
 341                 } /* while (42) */
 342
 343                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 344                  * the same time, ALWAYS lock `cache_lock' first! */
 345
 346                 /* We're in the shutdown phase */
 347                 if ((flushq_head == NULL) && (queue_head == NULL))
 348                 {
 349                   pthread_mutex_unlock (&queue_lock);
 350                   break;
 351                 }
 352
 353                 if (flushq_head != NULL)
 354                 {
 355                   /* Dequeue the first flush entry */
 356                   queue_entry = flushq_head;
 357                   if (flushq_head == flushq_tail)
 358                     flushq_head = flushq_tail = NULL;
 359                   else
 360                     flushq_head = flushq_head->next;
 361                 }
 362                 else /* if (queue_head != NULL) */
 363                 {
 364                   /* Dequeue the first regular entry */
 365                   queue_entry = queue_head;
 366                   if (queue_head == queue_tail)
 367                     queue_head = queue_tail = NULL;
 368                   else
 369                     queue_head = queue_head->next;
 370                 }
 371
 372                 /* Unlock the queue again */
 373                 pthread_mutex_unlock (&queue_lock);
 374
 375                 /* We now need the cache lock so the entry isn't updated while
 376                  * we make a copy of it's values */
 377                 pthread_mutex_lock (&cache_lock);
 378
 379                 status = c_avl_get (cache, queue_entry->filename,
 380                                 (void *) &cache_entry);
 381
 382                 if (status == 0)
 383                 {
 384                         values = cache_entry->values;
 385                         values_num = cache_entry->values_num;
 386
 387                         cache_entry->values = NULL;
 388                         cache_entry->values_num = 0;
 389                         cache_entry->flags = FLAG_NONE;
 390                 }
 391
 392                 pthread_mutex_unlock (&cache_lock);
 393
 394                 if (status != 0)
 395                 {
 396                         sfree (queue_entry->filename);
 397                         sfree (queue_entry);
 398                         continue;
 399                 }
 400
 401                 /* Update `tv_next_update' */
 402                 if (write_rate > 0.0)
 403                 {
 404                   gettimeofday (&tv_now, /* timezone = */ NULL);
 405                   tv_next_update.tv_sec = tv_now.tv_sec;
 406                   tv_next_update.tv_usec = tv_now.tv_usec
 407                     + ((suseconds_t) (1000000 * write_rate));
 408                   while (tv_next_update.tv_usec > 1000000)
 409                   {
 410                     tv_next_update.tv_sec++;
 411                     tv_next_update.tv_usec -= 1000000;
 412                   }
 413                 }
 414
 415                 /* Write the values to the RRD-file */
 416                 srrd_update (queue_entry->filename, NULL,
 417                                 values_num, (const char **)values);
 418                 DEBUG ("rrdtool plugin: queue thread: Wrote %i value%s to %s",
 419                                 values_num, (values_num == 1) ? "" : "s",
 420                                 queue_entry->filename);
 421
 422                 for (i = 0; i < values_num; i++)
 423                 {
 424                         sfree (values[i]);
 425                 }
 426                 sfree (values);
 427                 sfree (queue_entry->filename);
 428                 sfree (queue_entry);
 429         } /* while (42) */
 430
 431         pthread_exit ((void *) 0);
 432         return ((void *) 0);
 433 } /* void *rrd_queue_thread */
 434
 435 static int rrd_queue_enqueue (const char *filename,
 436     rrd_queue_t **head, rrd_queue_t **tail)
 437 {
 438   rrd_queue_t *queue_entry;
 439
 440   queue_entry = (rrd_queue_t *) malloc (sizeof (rrd_queue_t));
 441   if (queue_entry == NULL)
 442     return (-1);
 443
 444   queue_entry->filename = strdup (filename);
 445   if (queue_entry->filename == NULL)
 446   {
 447     free (queue_entry);
 448     return (-1);
 449   }
 450
 451   queue_entry->next = NULL;
 452
 453   pthread_mutex_lock (&queue_lock);
 454
 455   if (*tail == NULL)
 456     *head = queue_entry;
 457   else
 458     (*tail)->next = queue_entry;
 459   *tail = queue_entry;
 460
 461   pthread_cond_signal (&queue_cond);
 462   pthread_mutex_unlock (&queue_lock);
 463
 464   return (0);
 465 } /* int rrd_queue_enqueue */
 466
 467 static int rrd_queue_dequeue (const char *filename,
 468     rrd_queue_t **head, rrd_queue_t **tail)
 469 {
 470   rrd_queue_t *this;
 471   rrd_queue_t *prev;
 472
 473   pthread_mutex_lock (&queue_lock);
 474
 475   prev = NULL;
 476   this = *head;
 477
 478   while (this != NULL)
 479   {
 480     if (strcmp (this->filename, filename) == 0)
 481       break;
 482
 483     prev = this;
 484     this = this->next;
 485   }
 486
 487   if (this == NULL)
 488   {
 489     pthread_mutex_unlock (&queue_lock);
 490     return (-1);
 491   }
 492
 493   if (prev == NULL)
 494     *head = this->next;
 495   else
 496     prev->next = this->next;
 497
 498   if (this->next == NULL)
 499     *tail = prev;
 500
 501   pthread_mutex_unlock (&queue_lock);
 502
 503   sfree (this->filename);
 504   sfree (this);
 505
 506   return (0);
 507 } /* int rrd_queue_dequeue */
 508
 509 /* XXX: You must hold "cache_lock" when calling this function! */
 510 static void rrd_cache_flush (cdtime_t timeout)
 511 {
 512         rrd_cache_t *rc;
 513         cdtime_t     now;
 514
 515         char **keys = NULL;
 516         int    keys_num = 0;
 517
 518         char *key;
 519         c_avl_iterator_t *iter;
 520         int i;
 521
 522         DEBUG ("rrdtool plugin: Flushing cache, timeout = %.3f",
 523                         CDTIME_T_TO_DOUBLE (timeout));
 524
 525         now = cdtime ();
 526         timeout = TIME_T_TO_CDTIME_T (timeout);
 527
 528         /* Build a list of entries to be flushed */
 529         iter = c_avl_get_iterator (cache);
 530         while (c_avl_iterator_next (iter, (void *) &key, (void *) &rc) == 0)
 531         {
 532                 if (rc->flags != FLAG_NONE)
 533                         continue;
 534                 /* timeout == 0  =>  flush everything */
 535                 else if ((timeout != 0)
 536                                 && ((now - rc->first_value) < timeout))
 537                         continue;
 538                 else if (rc->values_num > 0)
 539                 {
 540                         int status;
 541
 542                         status = rrd_queue_enqueue (key, &queue_head,  &queue_tail);
 543                         if (status == 0)
 544                                 rc->flags = FLAG_QUEUED;
 545                 }
 546                 else /* ancient and no values -> waste of memory */
 547                 {
 548                         char **tmp = (char **) realloc ((void *) keys,
 549                                         (keys_num + 1) * sizeof (char *));
 550                         if (tmp == NULL)
 551                         {
 552                                 char errbuf[1024];
 553                                 ERROR ("rrdtool plugin: "
 554                                                 "realloc failed: %s",
 555                                                 sstrerror (errno, errbuf,
 556                                                         sizeof (errbuf)));
 557                                 c_avl_iterator_destroy (iter);
 558                                 sfree (keys);
 559                                 return;
 560                         }
 561                         keys = tmp;
 562                         keys[keys_num] = key;
 563                         keys_num++;
 564                 }
 565         } /* while (c_avl_iterator_next) */
 566         c_avl_iterator_destroy (iter);
 567
 568         for (i = 0; i < keys_num; i++)
 569         {
 570                 if (c_avl_remove (cache, keys[i], (void *) &key, (void *) &rc) != 0)
 571                 {
 572                         DEBUG ("rrdtool plugin: c_avl_remove (%s) failed.", keys[i]);
 573                         continue;
 574                 }
 575
 576                 assert (rc->values == NULL);
 577                 assert (rc->values_num == 0);
 578
 579                 sfree (rc);
 580                 sfree (key);
 581                 keys[i] = NULL;
 582         } /* for (i = 0..keys_num) */
 583
 584         sfree (keys);
 585
 586         cache_flush_last = now;
 587 } /* void rrd_cache_flush */
 588
 589 static int rrd_cache_flush_identifier (cdtime_t timeout,
 590     const char *identifier)
 591 {
 592   rrd_cache_t *rc;
 593   cdtime_t now;
 594   int status;
 595   char key[2048];
 596
 597   if (identifier == NULL)
 598   {
 599     rrd_cache_flush (timeout);
 600     return (0);
 601   }
 602
 603   now = cdtime ();
 604
 605   if (datadir == NULL)
 606     snprintf (key, sizeof (key), "%s.rrd",
 607         identifier);
 608   else
 609     snprintf (key, sizeof (key), "%s/%s.rrd",
 610         datadir, identifier);
 611   key[sizeof (key) - 1] = 0;
 612
 613   status = c_avl_get (cache, key, (void *) &rc);
 614   if (status != 0)
 615   {
 616     INFO ("rrdtool plugin: rrd_cache_flush_identifier: "
 617         "c_avl_get (%s) failed. Does that file really exist?",
 618         key);
 619     return (status);
 620   }
 621
 622   if (rc->flags == FLAG_FLUSHQ)
 623   {
 624     status = 0;
 625   }
 626   else if (rc->flags == FLAG_QUEUED)
 627   {
 628     rrd_queue_dequeue (key, &queue_head, &queue_tail);
 629     status = rrd_queue_enqueue (key, &flushq_head, &flushq_tail);
 630     if (status == 0)
 631       rc->flags = FLAG_FLUSHQ;
 632   }
 633   else if ((now - rc->first_value) < timeout)
 634   {
 635     status = 0;
 636   }
 637   else if (rc->values_num > 0)
 638   {
 639     status = rrd_queue_enqueue (key, &flushq_head, &flushq_tail);
 640     if (status == 0)
 641       rc->flags = FLAG_FLUSHQ;
 642   }
 643
 644   return (status);
 645 } /* int rrd_cache_flush_identifier */
 646
 647 static int64_t rrd_get_random_variation (void)
 648 {
 649   double dbl_timeout;
 650   cdtime_t ctm_timeout;
 651   double rand_fact;
 652   _Bool negative;
 653   int64_t ret;
 654
 655   if (random_timeout <= 0)
 656     return (0);
 657
 658   /* Assure that "cache_timeout + random_variation" is never negative. */
 659   if (random_timeout > cache_timeout)
 660   {
 661           INFO ("rrdtool plugin: Adjusting \"RandomTimeout\" to %.3f seconds.",
 662                           CDTIME_T_TO_DOUBLE (cache_timeout));
 663           random_timeout = cache_timeout;
 664   }
 665
 666   /* This seems a bit complicated, but "random_timeout" is likely larger than
 667    * RAND_MAX, so we can't simply use modulo here. */
 668   dbl_timeout = CDTIME_T_TO_DOUBLE (random_timeout);
 669   rand_fact = ((double) random ())
 670     / ((double) RAND_MAX);
 671   negative = (_Bool) (random () % 2);
 672
 673   ctm_timeout = DOUBLE_TO_CDTIME_T (dbl_timeout * rand_fact);
 674
 675   ret = (int64_t) ctm_timeout;
 676   if (negative)
 677     ret *= -1;
 678
 679   return (ret);
 680 } /* int64_t rrd_get_random_variation */
 681
 682 static int rrd_cache_insert (const char *filename,
 683                 const char *value, cdtime_t value_time)
 684 {
 685         rrd_cache_t *rc = NULL;
 686         int new_rc = 0;
 687         char **values_new;
 688
 689         pthread_mutex_lock (&cache_lock);
 690
 691         /* This shouldn't happen, but it did happen at least once, so we'll be
 692          * careful. */
 693         if (cache == NULL)
 694         {
 695                 pthread_mutex_unlock (&cache_lock);
 696                 WARNING ("rrdtool plugin: cache == NULL.");
 697                 return (-1);
 698         }
 699
 700         c_avl_get (cache, filename, (void *) &rc);
 701
 702         if (rc == NULL)
 703         {
 704                 rc = malloc (sizeof (*rc));
 705                 if (rc == NULL)
 706                         return (-1);
 707                 rc->values_num = 0;
 708                 rc->values = NULL;
 709                 rc->first_value = 0;
 710                 rc->last_value = 0;
 711                 rc->random_variation = rrd_get_random_variation ();
 712                 rc->flags = FLAG_NONE;
 713                 new_rc = 1;
 714         }
 715
 716         if (rc->last_value >= value_time)
 717         {
 718                 pthread_mutex_unlock (&cache_lock);
 719                 DEBUG ("rrdtool plugin: (rc->last_value = %"PRIu64") "
 720                                 ">= (value_time = %"PRIu64")",
 721                                 rc->last_value, value_time);
 722                 return (-1);
 723         }
 724
 725         values_new = (char **) realloc ((void *) rc->values,
 726                         (rc->values_num + 1) * sizeof (char *));
 727         if (values_new == NULL)
 728         {
 729                 char errbuf[1024];
 730                 void *cache_key = NULL;
 731
 732                 sstrerror (errno, errbuf, sizeof (errbuf));
 733
 734                 c_avl_remove (cache, filename, &cache_key, NULL);
 735                 pthread_mutex_unlock (&cache_lock);
 736
 737                 ERROR ("rrdtool plugin: realloc failed: %s", errbuf);
 738
 739                 sfree (cache_key);
 740                 sfree (rc->values);
 741                 sfree (rc);
 742                 return (-1);
 743         }
 744         rc->values = values_new;
 745
 746         rc->values[rc->values_num] = strdup (value);
 747         if (rc->values[rc->values_num] != NULL)
 748                 rc->values_num++;
 749
 750         if (rc->values_num == 1)
 751                 rc->first_value = value_time;
 752         rc->last_value = value_time;
 753
 754         /* Insert if this is the first value */
 755         if (new_rc == 1)
 756         {
 757                 void *cache_key = strdup (filename);
 758
 759                 if (cache_key == NULL)
 760                 {
 761                         char errbuf[1024];
 762                         sstrerror (errno, errbuf, sizeof (errbuf));
 763
 764                         pthread_mutex_unlock (&cache_lock);
 765
 766                         ERROR ("rrdtool plugin: strdup failed: %s", errbuf);
 767
 768                         sfree (rc->values[0]);
 769                         sfree (rc->values);
 770                         sfree (rc);
 771                         return (-1);
 772                 }
 773
 774                 c_avl_insert (cache, cache_key, rc);
 775         }
 776
 777         DEBUG ("rrdtool plugin: rrd_cache_insert: file = %s; "
 778                         "values_num = %i; age = %.3f;",
 779                         filename, rc->values_num,
 780                         CDTIME_T_TO_DOUBLE (rc->last_value - rc->first_value));
 781
 782         if ((rc->last_value - rc->first_value) >= (cache_timeout + rc->random_variation))
 783         {
 784                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 785                  * the same time, ALWAYS lock `cache_lock' first! */
 786                 if (rc->flags == FLAG_NONE)
 787                 {
 788                         int status;
 789
 790                         status = rrd_queue_enqueue (filename, &queue_head, &queue_tail);
 791                         if (status == 0)
 792                                 rc->flags = FLAG_QUEUED;
 793
 794                         rc->random_variation = rrd_get_random_variation ();
 795                 }
 796                 else
 797                 {
 798                         DEBUG ("rrdtool plugin: `%s' is already queued.", filename);
 799                 }
 800         }
 801
 802         if ((cache_timeout > 0) &&
 803                         ((cdtime () - cache_flush_last) > cache_flush_timeout))
 804                 rrd_cache_flush (cache_flush_timeout);
 805
 806         pthread_mutex_unlock (&cache_lock);
 807
 808         return (0);
 809 } /* int rrd_cache_insert */
 810
 811 static int rrd_cache_destroy (void) /* {{{ */
 812 {
 813   void *key = NULL;
 814   void *value = NULL;
 815
 816   int non_empty = 0;
 817
 818   pthread_mutex_lock (&cache_lock);
 819
 820   if (cache == NULL)
 821   {
 822     pthread_mutex_unlock (&cache_lock);
 823     return (0);
 824   }
 825
 826   while (c_avl_pick (cache, &key, &value) == 0)
 827   {
 828     rrd_cache_t *rc;
 829     int i;
 830
 831     sfree (key);
 832     key = NULL;
 833
 834     rc = value;
 835     value = NULL;
 836
 837     if (rc->values_num > 0)
 838       non_empty++;
 839
 840     for (i = 0; i < rc->values_num; i++)
 841       sfree (rc->values[i]);
 842     sfree (rc->values);
 843     sfree (rc);
 844   }
 845
 846   c_avl_destroy (cache);
 847   cache = NULL;
 848
 849   if (non_empty > 0)
 850   {
 851     INFO ("rrdtool plugin: %i cache %s had values when destroying the cache.",
 852         non_empty, (non_empty == 1) ? "entry" : "entries");
 853   }
 854   else
 855   {
 856     DEBUG ("rrdtool plugin: No values have been lost "
 857         "when destroying the cache.");
 858   }
 859
 860   pthread_mutex_unlock (&cache_lock);
 861   return (0);
 862 } /* }}} int rrd_cache_destroy */
 863
 864 static int rrd_compare_numeric (const void *a_ptr, const void *b_ptr)
 865 {
 866         int a = *((int *) a_ptr);
 867         int b = *((int *) b_ptr);
 868
 869         if (a < b)
 870                 return (-1);
 871         else if (a > b)
 872                 return (1);
 873         else
 874                 return (0);
 875 } /* int rrd_compare_numeric */
 876
 877 static int rrd_write (const data_set_t *ds, const value_list_t *vl,
 878                 user_data_t __attribute__((unused)) *user_data)
 879 {
 880         struct stat  statbuf;
 881         char         filename[512];
 882         char         values[512];
 883         int          status;
 884
 885         if (do_shutdown)
 886                 return (0);
 887
 888         if (0 != strcmp (ds->type, vl->type)) {
 889                 ERROR ("rrdtool plugin: DS type does not match value list type");
 890                 return -1;
 891         }
 892
 893         if (value_list_to_filename (filename, sizeof (filename), vl) != 0)
 894                 return (-1);
 895
 896         if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
 897                 return (-1);
 898
 899         if (stat (filename, &statbuf) == -1)
 900         {
 901                 if (errno == ENOENT)
 902                 {
 903                         status = cu_rrd_create_file (filename,
 904                                         ds, vl, &rrdcreate_config);
 905                         if (status != 0)
 906                                 return (-1);
 907                 }
 908                 else
 909                 {
 910                         char errbuf[1024];
 911                         ERROR ("stat(%s) failed: %s", filename,
 912                                         sstrerror (errno, errbuf,
 913                                                 sizeof (errbuf)));
 914                         return (-1);
 915                 }
 916         }
 917         else if (!S_ISREG (statbuf.st_mode))
 918         {
 919                 ERROR ("stat(%s): Not a regular file!",
 920                                 filename);
 921                 return (-1);
 922         }
 923
 924         status = rrd_cache_insert (filename, values, vl->time);
 925
 926         return (status);
 927 } /* int rrd_write */
 928
 929 static int rrd_flush (cdtime_t timeout, const char *identifier,
 930                 __attribute__((unused)) user_data_t *user_data)
 931 {
 932         pthread_mutex_lock (&cache_lock);
 933
 934         if (cache == NULL) {
 935                 pthread_mutex_unlock (&cache_lock);
 936                 return (0);
 937         }
 938
 939         rrd_cache_flush_identifier (timeout, identifier);
 940
 941         pthread_mutex_unlock (&cache_lock);
 942         return (0);
 943 } /* int rrd_flush */
 944
 945 static int rrd_config (const char *key, const char *value)
 946 {
 947         if (strcasecmp ("CacheTimeout", key) == 0)
 948         {
 949                 double tmp = atof (value);
 950                 if (tmp < 0)
 951                 {
 952                         fprintf (stderr, "rrdtool: `CacheTimeout' must "
 953                                         "be greater than 0.\n");
 954                         ERROR ("rrdtool: `CacheTimeout' must "
 955                                         "be greater than 0.\n");
 956                         return (1);
 957                 }
 958                 cache_timeout = DOUBLE_TO_CDTIME_T (tmp);
 959         }
 960         else if (strcasecmp ("CacheFlush", key) == 0)
 961         {
 962                 int tmp = atoi (value);
 963                 if (tmp < 0)
 964                 {
 965                         fprintf (stderr, "rrdtool: `CacheFlush' must "
 966                                         "be greater than 0.\n");
 967                         ERROR ("rrdtool: `CacheFlush' must "
 968                                         "be greater than 0.\n");
 969                         return (1);
 970                 }
 971                 cache_flush_timeout = tmp;
 972         }
 973         else if (strcasecmp ("DataDir", key) == 0)
 974         {
 975                 if (datadir != NULL)
 976                         free (datadir);
 977                 datadir = strdup (value);
 978                 if (datadir != NULL)
 979                 {
 980                         int len = strlen (datadir);
 981                         while ((len > 0) && (datadir[len - 1] == '/'))
 982                         {
 983                                 len--;
 984                                 datadir[len] = '\0';
 985                         }
 986                         if (len <= 0)
 987                         {
 988                                 free (datadir);
 989                                 datadir = NULL;
 990                         }
 991                 }
 992         }
 993         else if (strcasecmp ("StepSize", key) == 0)
 994         {
 995                 unsigned long temp = strtoul (value, NULL, 0);
 996                 if (temp > 0)
 997                         rrdcreate_config.stepsize = temp;
 998         }
 999         else if (strcasecmp ("HeartBeat", key) == 0)
1000         {
1001                 int temp = atoi (value);
1002                 if (temp > 0)
1003                         rrdcreate_config.heartbeat = temp;
1004         }
1005         else if (strcasecmp ("RRARows", key) == 0)
1006         {
1007                 int tmp = atoi (value);
1008                 if (tmp <= 0)
1009                 {
1010                         fprintf (stderr, "rrdtool: `RRARows' must "
1011                                         "be greater than 0.\n");
1012                         ERROR ("rrdtool: `RRARows' must "
1013                                         "be greater than 0.\n");
1014                         return (1);
1015                 }
1016                 rrdcreate_config.rrarows = tmp;
1017         }
1018         else if (strcasecmp ("RRATimespan", key) == 0)
1019         {
1020                 char *saveptr = NULL;
1021                 char *dummy;
1022                 char *ptr;
1023                 char *value_copy;
1024                 int *tmp_alloc;
1025
1026                 value_copy = strdup (value);
1027                 if (value_copy == NULL)
1028                         return (1);
1029
1030                 dummy = value_copy;
1031                 while ((ptr = strtok_r (dummy, ", \t", &saveptr)) != NULL)
1032                 {
1033                         dummy = NULL;
1034
1035                         tmp_alloc = realloc (rrdcreate_config.timespans,
1036                                         sizeof (int) * (rrdcreate_config.timespans_num + 1));
1037                         if (tmp_alloc == NULL)
1038                         {
1039                                 fprintf (stderr, "rrdtool: realloc failed.\n");
1040                                 ERROR ("rrdtool: realloc failed.\n");
1041                                 free (value_copy);
1042                                 return (1);
1043                         }
1044                         rrdcreate_config.timespans = tmp_alloc;
1045                         rrdcreate_config.timespans[rrdcreate_config.timespans_num] = atoi (ptr);
1046                         if (rrdcreate_config.timespans[rrdcreate_config.timespans_num] != 0)
1047                                 rrdcreate_config.timespans_num++;
1048                 } /* while (strtok_r) */
1049
1050                 qsort (/* base = */ rrdcreate_config.timespans,
1051                                 /* nmemb  = */ rrdcreate_config.timespans_num,
1052                                 /* size   = */ sizeof (rrdcreate_config.timespans[0]),
1053                                 /* compar = */ rrd_compare_numeric);
1054
1055                 free (value_copy);
1056         }
1057         else if (strcasecmp ("XFF", key) == 0)
1058         {
1059                 double tmp = atof (value);
1060                 if ((tmp < 0.0) || (tmp >= 1.0))
1061                 {
1062                         fprintf (stderr, "rrdtool: `XFF' must "
1063                                         "be in the range 0 to 1 (exclusive).");
1064                         ERROR ("rrdtool: `XFF' must "
1065                                         "be in the range 0 to 1 (exclusive).");
1066                         return (1);
1067                 }
1068                 rrdcreate_config.xff = tmp;
1069         }
1070         else if (strcasecmp ("WritesPerSecond", key) == 0)
1071         {
1072                 double wps = atof (value);
1073
1074                 if (wps < 0.0)
1075                 {
1076                         fprintf (stderr, "rrdtool: `WritesPerSecond' must be "
1077                                         "greater than or equal to zero.");
1078                         return (1);
1079                 }
1080                 else if (wps == 0.0)
1081                 {
1082                         write_rate = 0.0;
1083                 }
1084                 else
1085                 {
1086                         write_rate = 1.0 / wps;
1087                 }
1088         }
1089         else if (strcasecmp ("RandomTimeout", key) == 0)
1090         {
1091                 double tmp;
1092
1093                 tmp = atof (value);
1094                 if (tmp < 0.0)
1095                 {
1096                         fprintf (stderr, "rrdtool: `RandomTimeout' must "
1097                                         "be greater than or equal to zero.\n");
1098                         ERROR ("rrdtool: `RandomTimeout' must "
1099                                         "be greater then or equal to zero.");
1100                 }
1101                 else
1102                 {
1103                         random_timeout = DOUBLE_TO_CDTIME_T (tmp);
1104                 }
1105         }
1106         else
1107         {
1108                 return (-1);
1109         }
1110         return (0);
1111 } /* int rrd_config */
1112
1113 static int rrd_shutdown (void)
1114 {
1115         pthread_mutex_lock (&cache_lock);
1116         rrd_cache_flush (0);
1117         pthread_mutex_unlock (&cache_lock);
1118
1119         pthread_mutex_lock (&queue_lock);
1120         do_shutdown = 1;
1121         pthread_cond_signal (&queue_cond);
1122         pthread_mutex_unlock (&queue_lock);
1123
1124         if ((queue_thread_running != 0)
1125                         && ((queue_head != NULL) || (flushq_head != NULL)))
1126         {
1127                 INFO ("rrdtool plugin: Shutting down the queue thread. "
1128                                 "This may take a while.");
1129         }
1130         else if (queue_thread_running != 0)
1131         {
1132                 INFO ("rrdtool plugin: Shutting down the queue thread.");
1133         }
1134
1135         /* Wait for all the values to be written to disk before returning. */
1136         if (queue_thread_running != 0)
1137         {
1138                 pthread_join (queue_thread, NULL);
1139                 memset (&queue_thread, 0, sizeof (queue_thread));
1140                 queue_thread_running = 0;
1141                 DEBUG ("rrdtool plugin: queue_thread exited.");
1142         }
1143
1144         rrd_cache_destroy ();
1145
1146         return (0);
1147 } /* int rrd_shutdown */
1148
1149 static int rrd_init (void)
1150 {
1151         static int init_once = 0;
1152         int status;
1153
1154         if (init_once != 0)
1155                 return (0);
1156         init_once = 1;
1157
1158         if (rrdcreate_config.heartbeat <= 0)
1159                 rrdcreate_config.heartbeat = 2 * rrdcreate_config.stepsize;
1160
1161         /* Set the cache up */
1162         pthread_mutex_lock (&cache_lock);
1163
1164         cache = c_avl_create ((int (*) (const void *, const void *)) strcmp);
1165         if (cache == NULL)
1166         {
1167                 ERROR ("rrdtool plugin: c_avl_create failed.");
1168                 return (-1);
1169         }
1170
1171         cache_flush_last = cdtime ();
1172         if (cache_timeout == 0)
1173         {
1174                 cache_flush_timeout = 0;
1175         }
1176         else if (cache_flush_timeout < cache_timeout)
1177                 cache_flush_timeout = 10 * cache_timeout;
1178
1179         pthread_mutex_unlock (&cache_lock);
1180
1181         status = plugin_thread_create (&queue_thread, /* attr = */ NULL,
1182                         rrd_queue_thread, /* args = */ NULL);
1183         if (status != 0)
1184         {
1185                 ERROR ("rrdtool plugin: Cannot create queue-thread.");
1186                 return (-1);
1187         }
1188         queue_thread_running = 1;
1189
1190         DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %lu;"
1191                         " heartbeat = %i; rrarows = %i; xff = %lf;",
1192                         (datadir == NULL) ? "(null)" : datadir,
1193                         rrdcreate_config.stepsize,
1194                         rrdcreate_config.heartbeat,
1195                         rrdcreate_config.rrarows,
1196                         rrdcreate_config.xff);
1197
1198         return (0);
1199 } /* int rrd_init */
1200
1201 void module_register (void)
1202 {
1203         plugin_register_config ("rrdtool", rrd_config,
1204                         config_keys, config_keys_num);
1205         plugin_register_init ("rrdtool", rrd_init);
1206         plugin_register_write ("rrdtool", rrd_write, /* user_data = */ NULL);
1207         plugin_register_flush ("rrdtool", rrd_flush, /* user_data = */ NULL);
1208         plugin_register_shutdown ("rrdtool", rrd_shutdown);
1209 }