src/rrdtool.c

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