src/rrdtool.c

   1 /**
   2  * collectd - src/rrdtool.c
   3  * Copyright (C) 2006,2007  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 #if HAVE_PTHREAD_H
  28 # include <pthread.h>
  29 #endif
  30
  31 /*
  32  * Private types
  33  */
  34 struct rrd_cache_s
  35 {
  36         int    values_num;
  37         char **values;
  38         time_t first_value;
  39         time_t last_value;
  40         enum
  41         {
  42                 FLAG_NONE   = 0x00,
  43                 FLAG_QUEUED = 0x01
  44         } flags;
  45 };
  46 typedef struct rrd_cache_s rrd_cache_t;
  47
  48 struct rrd_queue_s
  49 {
  50         char *filename;
  51         struct rrd_queue_s *next;
  52 };
  53 typedef struct rrd_queue_s rrd_queue_t;
  54
  55 /*
  56  * Private variables
  57  */
  58 static int rra_timespans[] =
  59 {
  60         3600,
  61         86400,
  62         604800,
  63         2678400,
  64         31622400
  65 };
  66 static int rra_timespans_num = STATIC_ARRAY_SIZE (rra_timespans);
  67
  68 static int *rra_timespans_custom = NULL;
  69 static int rra_timespans_custom_num = 0;
  70
  71 static char *rra_types[] =
  72 {
  73         "AVERAGE",
  74         "MIN",
  75         "MAX"
  76 };
  77 static int rra_types_num = STATIC_ARRAY_SIZE (rra_types);
  78
  79 static const char *config_keys[] =
  80 {
  81         "CacheTimeout",
  82         "CacheFlush",
  83         "DataDir",
  84         "StepSize",
  85         "HeartBeat",
  86         "RRARows",
  87         "RRATimespan",
  88         "XFF"
  89 };
  90 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
  91
  92 static char   *datadir   = NULL;
  93 static int     stepsize  = 0;
  94 static int     heartbeat = 0;
  95 static int     rrarows   = 1200;
  96 static double  xff       = 0.1;
  97
  98 /* XXX: If you need to lock both, cache_lock and queue_lock, at the same time,
  99  * ALWAYS lock `cache_lock' first! */
 100 static int         cache_timeout = 0;
 101 static int         cache_flush_timeout = 0;
 102 static time_t      cache_flush_last;
 103 static avl_tree_t *cache = NULL;
 104 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 105
 106 static rrd_queue_t    *queue_head = NULL;
 107 static rrd_queue_t    *queue_tail = NULL;
 108 static pthread_t       queue_thread = 0;
 109 static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER;
 110 static pthread_cond_t  queue_cond = PTHREAD_COND_INITIALIZER;
 111
 112 static int do_shutdown = 0;
 113
 114 /* * * * * * * * * *
 115  * WARNING:  Magic *
 116  * * * * * * * * * */
 117 static int rra_get (char ***ret)
 118 {
 119         static char **rra_def = NULL;
 120         static int rra_num = 0;
 121
 122         int *rts;
 123         int  rts_num;
 124
 125         int rra_max;
 126
 127         int span;
 128
 129         int cdp_num;
 130         int cdp_len;
 131         int i, j;
 132
 133         char buffer[64];
 134
 135         if ((rra_num != 0) && (rra_def != NULL))
 136         {
 137                 *ret = rra_def;
 138                 return (rra_num);
 139         }
 140
 141         /* Use the configured timespans or fall back to the built-in defaults */
 142         if (rra_timespans_custom_num != 0)
 143         {
 144                 rts = rra_timespans_custom;
 145                 rts_num = rra_timespans_custom_num;
 146         }
 147         else
 148         {
 149                 rts = rra_timespans;
 150                 rts_num = rra_timespans_num;
 151         }
 152
 153         rra_max = rts_num * rra_types_num;
 154
 155         if ((rra_def = (char **) malloc ((rra_max + 1) * sizeof (char *))) == NULL)
 156                 return (-1);
 157         memset (rra_def, '\0', (rra_max + 1) * sizeof (char *));
 158
 159         if ((stepsize <= 0) || (rrarows <= 0))
 160         {
 161                 *ret = NULL;
 162                 return (-1);
 163         }
 164
 165         cdp_len = 0;
 166         for (i = 0; i < rts_num; i++)
 167         {
 168                 span = rts[i];
 169
 170                 if ((span / stepsize) < rrarows)
 171                         continue;
 172
 173                 if (cdp_len == 0)
 174                         cdp_len = 1;
 175                 else
 176                         cdp_len = (int) floor (((double) span)
 177                                         / ((double) (rrarows * stepsize)));
 178
 179                 cdp_num = (int) ceil (((double) span)
 180                                 / ((double) (cdp_len * stepsize)));
 181
 182                 for (j = 0; j < rra_types_num; j++)
 183                 {
 184                         if (rra_num >= rra_max)
 185                                 break;
 186
 187                         if (snprintf (buffer, sizeof (buffer), "RRA:%s:%3.1f:%u:%u",
 188                                                 rra_types[j], xff,
 189                                                 cdp_len, cdp_num) >= sizeof (buffer))
 190                         {
 191                                 ERROR ("rra_get: Buffer would have been truncated.");
 192                                 continue;
 193                         }
 194
 195                         rra_def[rra_num++] = sstrdup (buffer);
 196                 }
 197         }
 198
 199 #if COLLECT_DEBUG
 200         DEBUG ("rra_num = %i", rra_num);
 201         for (i = 0; i < rra_num; i++)
 202                 DEBUG ("  %s", rra_def[i]);
 203 #endif
 204
 205         *ret = rra_def;
 206         return (rra_num);
 207 }
 208
 209 static void ds_free (int ds_num, char **ds_def)
 210 {
 211         int i;
 212
 213         for (i = 0; i < ds_num; i++)
 214                 if (ds_def[i] != NULL)
 215                         free (ds_def[i]);
 216         free (ds_def);
 217 }
 218
 219 static int ds_get (char ***ret, const data_set_t *ds)
 220 {
 221         char **ds_def;
 222         int ds_num;
 223
 224         char min[32];
 225         char max[32];
 226         char buffer[128];
 227
 228         DEBUG ("ds->ds_num = %i", ds->ds_num);
 229
 230         ds_def = (char **) malloc (ds->ds_num * sizeof (char *));
 231         if (ds_def == NULL)
 232         {
 233                 char errbuf[1024];
 234                 ERROR ("rrdtool plugin: malloc failed: %s",
 235                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 236                 return (-1);
 237         }
 238         memset (ds_def, '\0', ds->ds_num * sizeof (char *));
 239
 240         for (ds_num = 0; ds_num < ds->ds_num; ds_num++)
 241         {
 242                 data_source_t *d = ds->ds + ds_num;
 243                 char *type;
 244                 int status;
 245
 246                 ds_def[ds_num] = NULL;
 247
 248                 if (d->type == DS_TYPE_COUNTER)
 249                         type = "COUNTER";
 250                 else if (d->type == DS_TYPE_GAUGE)
 251                         type = "GAUGE";
 252                 else
 253                 {
 254                         ERROR ("rrdtool plugin: Unknown DS type: %i",
 255                                         d->type);
 256                         break;
 257                 }
 258
 259                 if (isnan (d->min))
 260                 {
 261                         strcpy (min, "U");
 262                 }
 263                 else
 264                 {
 265                         snprintf (min, sizeof (min), "%lf", d->min);
 266                         min[sizeof (min) - 1] = '\0';
 267                 }
 268
 269                 if (isnan (d->max))
 270                 {
 271                         strcpy (max, "U");
 272                 }
 273                 else
 274                 {
 275                         snprintf (max, sizeof (max), "%lf", d->max);
 276                         max[sizeof (max) - 1] = '\0';
 277                 }
 278
 279                 status = snprintf (buffer, sizeof (buffer),
 280                                 "DS:%s:%s:%i:%s:%s",
 281                                 d->name, type, heartbeat,
 282                                 min, max);
 283                 if ((status < 1) || (status >= sizeof (buffer)))
 284                         break;
 285
 286                 ds_def[ds_num] = sstrdup (buffer);
 287         } /* for ds_num = 0 .. ds->ds_num */
 288
 289 #if COLLECT_DEBUG
 290 {
 291         int i;
 292         DEBUG ("ds_num = %i", ds_num);
 293         for (i = 0; i < ds_num; i++)
 294                 DEBUG ("  %s", ds_def[i]);
 295 }
 296 #endif
 297
 298         if (ds_num != ds->ds_num)
 299         {
 300                 ds_free (ds_num, ds_def);
 301                 return (-1);
 302         }
 303
 304         *ret = ds_def;
 305         return (ds_num);
 306 }
 307
 308 static int rrd_create_file (char *filename, const data_set_t *ds)
 309 {
 310         char **argv;
 311         int argc;
 312         char **rra_def;
 313         int rra_num;
 314         char **ds_def;
 315         int ds_num;
 316         int i, j;
 317         char stepsize_str[16];
 318         int status = 0;
 319
 320         if (check_create_dir (filename))
 321                 return (-1);
 322
 323         if ((rra_num = rra_get (&rra_def)) < 1)
 324         {
 325                 ERROR ("rrd_create_file failed: Could not calculate RRAs");
 326                 return (-1);
 327         }
 328
 329         if ((ds_num = ds_get (&ds_def, ds)) < 1)
 330         {
 331                 ERROR ("rrd_create_file failed: Could not calculate DSes");
 332                 return (-1);
 333         }
 334
 335         argc = ds_num + rra_num + 4;
 336
 337         if ((argv = (char **) malloc (sizeof (char *) * (argc + 1))) == NULL)
 338         {
 339                 char errbuf[1024];
 340                 ERROR ("rrd_create failed: %s",
 341                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 342                 return (-1);
 343         }
 344
 345         status = snprintf (stepsize_str, sizeof (stepsize_str),
 346                         "%i", stepsize);
 347         if ((status < 1) || (status >= sizeof (stepsize_str)))
 348         {
 349                 ERROR ("rrdtool plugin: snprintf failed.");
 350                 return (-1);
 351         }
 352
 353         argv[0] = "create";
 354         argv[1] = filename;
 355         argv[2] = "-s";
 356         argv[3] = stepsize_str;
 357
 358         j = 4;
 359         for (i = 0; i < ds_num; i++)
 360                 argv[j++] = ds_def[i];
 361         for (i = 0; i < rra_num; i++)
 362                 argv[j++] = rra_def[i];
 363         argv[j] = NULL;
 364
 365         optind = 0; /* bug in librrd? */
 366         rrd_clear_error ();
 367         if (rrd_create (argc, argv) == -1)
 368         {
 369                 ERROR ("rrd_create failed: %s: %s", filename, rrd_get_error ());
 370                 status = -1;
 371         }
 372
 373         free (argv);
 374         ds_free (ds_num, ds_def);
 375
 376         return (status);
 377 }
 378
 379 static int value_list_to_string (char *buffer, int buffer_len,
 380                 const data_set_t *ds, const value_list_t *vl)
 381 {
 382         int offset;
 383         int status;
 384         int i;
 385
 386         memset (buffer, '\0', sizeof (buffer_len));
 387
 388         status = snprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
 389         if ((status < 1) || (status >= buffer_len))
 390                 return (-1);
 391         offset = status;
 392
 393         for (i = 0; i < ds->ds_num; i++)
 394         {
 395                 if ((ds->ds[i].type != DS_TYPE_COUNTER)
 396                                 && (ds->ds[i].type != DS_TYPE_GAUGE))
 397                         return (-1);
 398
 399                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 400                         status = snprintf (buffer + offset, buffer_len - offset,
 401                                         ":%llu", vl->values[i].counter);
 402                 else
 403                         status = snprintf (buffer + offset, buffer_len - offset,
 404                                         ":%lf", vl->values[i].gauge);
 405
 406                 if ((status < 1) || (status >= (buffer_len - offset)))
 407                         return (-1);
 408
 409                 offset += status;
 410         } /* for ds->ds_num */
 411
 412         return (0);
 413 } /* int value_list_to_string */
 414
 415 static int value_list_to_filename (char *buffer, int buffer_len,
 416                 const data_set_t *ds, const value_list_t *vl)
 417 {
 418         int offset = 0;
 419         int status;
 420
 421         if (datadir != NULL)
 422         {
 423                 status = snprintf (buffer + offset, buffer_len - offset,
 424                                 "%s/", datadir);
 425                 if ((status < 1) || (status >= buffer_len - offset))
 426                         return (-1);
 427                 offset += status;
 428         }
 429
 430         status = snprintf (buffer + offset, buffer_len - offset,
 431                         "%s/", vl->host);
 432         if ((status < 1) || (status >= buffer_len - offset))
 433                 return (-1);
 434         offset += status;
 435
 436         if (strlen (vl->plugin_instance) > 0)
 437                 status = snprintf (buffer + offset, buffer_len - offset,
 438                                 "%s-%s/", vl->plugin, vl->plugin_instance);
 439         else
 440                 status = snprintf (buffer + offset, buffer_len - offset,
 441                                 "%s/", vl->plugin);
 442         if ((status < 1) || (status >= buffer_len - offset))
 443                 return (-1);
 444         offset += status;
 445
 446         if (strlen (vl->type_instance) > 0)
 447                 status = snprintf (buffer + offset, buffer_len - offset,
 448                                 "%s-%s.rrd", ds->type, vl->type_instance);
 449         else
 450                 status = snprintf (buffer + offset, buffer_len - offset,
 451                                 "%s.rrd", ds->type);
 452         if ((status < 1) || (status >= buffer_len - offset))
 453                 return (-1);
 454         offset += status;
 455
 456         return (0);
 457 } /* int value_list_to_filename */
 458
 459 static int rrd_write_to_file (char *filename, char **values, int values_num)
 460 {
 461         char **argv;
 462         int    argc;
 463
 464         char *fn;
 465         int status;
 466
 467         if (values_num < 1)
 468                 return (0);
 469
 470         argc = values_num + 2;
 471         argv = (char **) malloc ((argc + 1) * sizeof (char *));
 472         if (argv == NULL)
 473                 return (-1);
 474
 475         argv[0] = "update";
 476         argv[1] = filename;
 477         memcpy (argv + 2, values, values_num * sizeof (char *));
 478         argv[argc] = NULL;
 479
 480         DEBUG ("rrd_update (argc = %i, argv = %p)", argc, (void *) argv);
 481
 482         optind = 0; /* bug in librrd? */
 483         rrd_clear_error ();
 484         status = rrd_update (argc, argv);
 485         if (status != 0)
 486         {
 487                 WARNING ("rrd_update failed: %s: %s",
 488                                 filename, rrd_get_error ());
 489                 status = -1;
 490         }
 491
 492         free (argv);
 493         free (fn);
 494
 495         return (status);
 496 } /* int rrd_write_cache_entry */
 497
 498 static void *rrd_queue_thread (void *data)
 499 {
 500         while (42)
 501         {
 502                 rrd_queue_t *queue_entry;
 503                 rrd_cache_t *cache_entry;
 504                 char **values;
 505                 int    values_num;
 506                 int    i;
 507
 508                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 509                  * the same time, ALWAYS lock `cache_lock' first! */
 510
 511                 /* wait until an entry is available */
 512                 pthread_mutex_lock (&queue_lock);
 513                 while ((queue_head == NULL) && (do_shutdown == 0))
 514                         pthread_cond_wait (&queue_cond, &queue_lock);
 515
 516                 /* We're in the shutdown phase */
 517                 if (queue_head == NULL)
 518                 {
 519                         pthread_mutex_unlock (&queue_lock);
 520                         break;
 521                 }
 522
 523                 /* Dequeue the first entry */
 524                 queue_entry = queue_head;
 525                 if (queue_head == queue_tail)
 526                         queue_head = queue_tail = NULL;
 527                 else
 528                         queue_head = queue_head->next;
 529
 530                 /* Unlock the queue again */
 531                 pthread_mutex_unlock (&queue_lock);
 532
 533                 /* We now need the cache lock so the entry isn't updated while
 534                  * we make a copy of it's values */
 535                 pthread_mutex_lock (&cache_lock);
 536
 537                 avl_get (cache, queue_entry->filename, (void *) &cache_entry);
 538
 539                 values = cache_entry->values;
 540                 values_num = cache_entry->values_num;
 541
 542                 cache_entry->values = NULL;
 543                 cache_entry->values_num = 0;
 544                 cache_entry->flags = FLAG_NONE;
 545
 546                 pthread_mutex_unlock (&cache_lock);
 547
 548                 /* Write the values to the RRD-file */
 549                 rrd_write_to_file (queue_entry->filename, values, values_num);
 550
 551                 for (i = 0; i < values_num; i++)
 552                 {
 553                         sfree (values[i]);
 554                 }
 555                 sfree (values);
 556                 sfree (queue_entry->filename);
 557                 sfree (queue_entry);
 558         } /* while (42) */
 559
 560         pthread_mutex_lock (&cache_lock);
 561         avl_destroy (cache);
 562         cache = NULL;
 563         pthread_mutex_unlock (&cache_lock);
 564
 565         pthread_exit ((void *) 0);
 566         return ((void *) 0);
 567 } /* void *rrd_queue_thread */
 568
 569 static int rrd_queue_cache_entry (const char *filename)
 570 {
 571         rrd_queue_t *queue_entry;
 572
 573         queue_entry = (rrd_queue_t *) malloc (sizeof (rrd_queue_t));
 574         if (queue_entry == NULL)
 575                 return (-1);
 576
 577         queue_entry->filename = strdup (filename);
 578         if (queue_entry->filename == NULL)
 579         {
 580                 free (queue_entry);
 581                 return (-1);
 582         }
 583
 584         queue_entry->next = NULL;
 585
 586         pthread_mutex_lock (&queue_lock);
 587         if (queue_tail == NULL)
 588                 queue_head = queue_entry;
 589         else
 590                 queue_tail->next = queue_entry;
 591         queue_tail = queue_entry;
 592         pthread_cond_signal (&queue_cond);
 593         pthread_mutex_unlock (&queue_lock);
 594
 595         return (0);
 596 } /* int rrd_queue_cache_entry */
 597
 598 static void rrd_cache_flush (int timeout)
 599 {
 600         rrd_cache_t *rc;
 601         time_t       now;
 602
 603         char **keys = NULL;
 604         int    keys_num = 0;
 605
 606         char *key;
 607         avl_iterator_t *iter;
 608         int i;
 609
 610         DEBUG ("Flushing cache, timeout = %i", timeout);
 611
 612         now = time (NULL);
 613
 614         /* Build a list of entries to be flushed */
 615         iter = avl_get_iterator (cache);
 616         while (avl_iterator_next (iter, (void *) &key, (void *) &rc) == 0)
 617         {
 618                 DEBUG ("key = %s; age = %i;", key, now - rc->first_value);
 619
 620                 if (rc->flags == FLAG_QUEUED)
 621                         continue;
 622                 else if ((now - rc->first_value) < timeout)
 623                         continue;
 624                 else if (rc->values_num > 0)
 625                 {
 626                         if (rrd_queue_cache_entry (key) == 0)
 627                                 rc->flags = FLAG_QUEUED;
 628                 }
 629                 else /* ancient and no values -> waste of memory */
 630                 {
 631                         keys = (char **) realloc ((void *) keys,
 632                                         (keys_num + 1) * sizeof (char *));
 633                         if (keys == NULL)
 634                         {
 635                                 char errbuf[1024];
 636                                 ERROR ("rrdtool plugin: "
 637                                                 "realloc failed: %s",
 638                                                 sstrerror (errno, errbuf,
 639                                                         sizeof (errbuf)));
 640                                 avl_iterator_destroy (iter);
 641                                 return;
 642                         }
 643                         keys[keys_num] = key;
 644                         keys_num++;
 645                 }
 646         } /* while (avl_iterator_next) */
 647         avl_iterator_destroy (iter);
 648
 649         for (i = 0; i < keys_num; i++)
 650         {
 651                 if (avl_remove (cache, keys[i], (void *) &key, (void *) &rc) != 0)
 652                 {
 653                         DEBUG ("avl_remove (%s) failed.", keys[i]);
 654                         continue;
 655                 }
 656
 657                 assert (rc->values == NULL);
 658                 assert (rc->values_num == 0);
 659
 660                 sfree (rc);
 661                 sfree (key);
 662                 keys[i] = NULL;
 663         } /* for (i = 0..keys_num) */
 664
 665         free (keys);
 666         DEBUG ("Flushed %i value(s)", keys_num);
 667
 668         cache_flush_last = now;
 669 } /* void rrd_cache_flush */
 670
 671 static int rrd_cache_insert (const char *filename,
 672                 const char *value, time_t value_time)
 673 {
 674         rrd_cache_t *rc = NULL;
 675         int new_rc = 0;
 676         char **values_new;
 677
 678         pthread_mutex_lock (&cache_lock);
 679
 680         avl_get (cache, filename, (void *) &rc);
 681
 682         if (rc == NULL)
 683         {
 684                 rc = (rrd_cache_t *) malloc (sizeof (rrd_cache_t));
 685                 if (rc == NULL)
 686                         return (-1);
 687                 rc->values_num = 0;
 688                 rc->values = NULL;
 689                 rc->first_value = 0;
 690                 rc->last_value = 0;
 691                 rc->flags = FLAG_NONE;
 692                 new_rc = 1;
 693         }
 694
 695         if (rc->last_value >= value_time)
 696         {
 697                 pthread_mutex_unlock (&cache_lock);
 698                 WARNING ("rrdtool plugin: (rc->last_value = %u) >= (value_time = %u)",
 699                                 (unsigned int) rc->last_value,
 700                                 (unsigned int) value_time);
 701                 return (-1);
 702         }
 703
 704         values_new = (char **) realloc ((void *) rc->values,
 705                         (rc->values_num + 1) * sizeof (char *));
 706         if (values_new == NULL)
 707         {
 708                 char errbuf[1024];
 709                 void *cache_key = NULL;
 710
 711                 sstrerror (errno, errbuf, sizeof (errbuf));
 712
 713                 avl_remove (cache, filename, &cache_key, NULL);
 714                 pthread_mutex_unlock (&cache_lock);
 715
 716                 ERROR ("rrdtool plugin: realloc failed: %s", errbuf);
 717
 718                 sfree (cache_key);
 719                 sfree (rc->values);
 720                 sfree (rc);
 721                 return (-1);
 722         }
 723
 724         rc->values[rc->values_num] = strdup (value);
 725         if (rc->values[rc->values_num] != NULL)
 726                 rc->values_num++;
 727
 728         if (rc->values_num == 1)
 729                 rc->first_value = value_time;
 730         rc->last_value = value_time;
 731
 732         /* Insert if this is the first value */
 733         if (new_rc == 1)
 734         {
 735                 void *cache_key = strdup (filename);
 736
 737                 if (cache_key == NULL)
 738                 {
 739                         char errbuf[1024];
 740                         sstrerror (errno, errbuf, sizeof (errbuf));
 741
 742                         pthread_mutex_unlock (&cache_lock);
 743
 744                         ERROR ("rrdtool plugin: strdup failed: %s", errbuf);
 745
 746                         sfree (rc->values[0]);
 747                         sfree (rc->values);
 748                         sfree (rc);
 749                         return (-1);
 750                 }
 751
 752                 avl_insert (cache, cache_key, rc);
 753         }
 754
 755         DEBUG ("rrd_cache_insert (%s, %s, %u) = %p", filename, value,
 756                         (unsigned int) value_time, (void *) rc);
 757
 758         if (((rc->last_value - rc->first_value) >= cache_timeout)
 759                         && (rc->flags != FLAG_QUEUED))
 760         {
 761                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 762                  * the same time, ALWAYS lock `cache_lock' first! */
 763                 if (rrd_queue_cache_entry (filename) == 0)
 764                         rc->flags = FLAG_QUEUED;
 765         }
 766
 767         if ((cache_timeout > 0) &&
 768                         ((time (NULL) - cache_flush_last) > cache_flush_timeout))
 769                 rrd_cache_flush (cache_flush_timeout);
 770
 771
 772         pthread_mutex_unlock (&cache_lock);
 773
 774         return (0);
 775 } /* int rrd_cache_insert */
 776
 777 static int rrd_write (const data_set_t *ds, const value_list_t *vl)
 778 {
 779         struct stat  statbuf;
 780         char         filename[512];
 781         char         values[512];
 782         int          status;
 783
 784         if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
 785                 return (-1);
 786
 787         if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
 788                 return (-1);
 789
 790         if (stat (filename, &statbuf) == -1)
 791         {
 792                 if (errno == ENOENT)
 793                 {
 794                         if (rrd_create_file (filename, ds))
 795                                 return (-1);
 796                 }
 797                 else
 798                 {
 799                         char errbuf[1024];
 800                         ERROR ("stat(%s) failed: %s", filename,
 801                                         sstrerror (errno, errbuf,
 802                                                 sizeof (errbuf)));
 803                         return (-1);
 804                 }
 805         }
 806         else if (!S_ISREG (statbuf.st_mode))
 807         {
 808                 ERROR ("stat(%s): Not a regular file!",
 809                                 filename);
 810                 return (-1);
 811         }
 812
 813         status = rrd_cache_insert (filename, values, vl->time);
 814
 815         return (status);
 816 } /* int rrd_write */
 817
 818 static int rrd_config (const char *key, const char *value)
 819 {
 820         if (strcasecmp ("CacheTimeout", key) == 0)
 821         {
 822                 int tmp = atoi (value);
 823                 if (tmp < 0)
 824                 {
 825                         fprintf (stderr, "rrdtool: `CacheTimeout' must "
 826                                         "be greater than 0.\n");
 827                         return (1);
 828                 }
 829                 cache_timeout = tmp;
 830         }
 831         else if (strcasecmp ("CacheFlush", key) == 0)
 832         {
 833                 int tmp = atoi (value);
 834                 if (tmp < 0)
 835                 {
 836                         fprintf (stderr, "rrdtool: `CacheFlush' must "
 837                                         "be greater than 0.\n");
 838                         return (1);
 839                 }
 840                 cache_flush_timeout = tmp;
 841         }
 842         else if (strcasecmp ("DataDir", key) == 0)
 843         {
 844                 if (datadir != NULL)
 845                         free (datadir);
 846                 datadir = strdup (value);
 847                 if (datadir != NULL)
 848                 {
 849                         int len = strlen (datadir);
 850                         while ((len > 0) && (datadir[len - 1] == '/'))
 851                         {
 852                                 len--;
 853                                 datadir[len] = '\0';
 854                         }
 855                         if (len <= 0)
 856                         {
 857                                 free (datadir);
 858                                 datadir = NULL;
 859                         }
 860                 }
 861         }
 862         else if (strcasecmp ("StepSize", key) == 0)
 863         {
 864                 int tmp = atoi (value);
 865                 if (tmp <= 0)
 866                 {
 867                         fprintf (stderr, "rrdtool: `StepSize' must "
 868                                         "be greater than 0.\n");
 869                         return (1);
 870                 }
 871                 stepsize = tmp;
 872         }
 873         else if (strcasecmp ("HeartBeat", key) == 0)
 874         {
 875                 int tmp = atoi (value);
 876                 if (tmp <= 0)
 877                 {
 878                         fprintf (stderr, "rrdtool: `HeartBeat' must "
 879                                         "be greater than 0.\n");
 880                         return (1);
 881                 }
 882                 heartbeat = tmp;
 883         }
 884         else if (strcasecmp ("RRARows", key) == 0)
 885         {
 886                 int tmp = atoi (value);
 887                 if (tmp <= 0)
 888                 {
 889                         fprintf (stderr, "rrdtool: `RRARows' must "
 890                                         "be greater than 0.\n");
 891                         return (1);
 892                 }
 893                 rrarows = tmp;
 894         }
 895         else if (strcasecmp ("RRATimespan", key) == 0)
 896         {
 897                 char *saveptr = NULL;
 898                 char *dummy;
 899                 char *ptr;
 900                 char *value_copy;
 901                 int *tmp_alloc;
 902
 903                 value_copy = strdup (value);
 904                 if (value_copy == NULL)
 905                         return (1);
 906
 907                 dummy = value_copy;
 908                 while ((ptr = strtok_r (dummy, ", \t", &saveptr)) != NULL)
 909                 {
 910                         dummy = NULL;
 911
 912                         tmp_alloc = realloc (rra_timespans_custom,
 913                                         sizeof (int) * (rra_timespans_custom_num + 1));
 914                         if (tmp_alloc == NULL)
 915                         {
 916                                 fprintf (stderr, "rrdtool: realloc failed.\n");
 917                                 free (value_copy);
 918                                 return (1);
 919                         }
 920                         rra_timespans_custom = tmp_alloc;
 921                         rra_timespans_custom[rra_timespans_custom_num] = atoi (ptr);
 922                         if (rra_timespans_custom[rra_timespans_custom_num] != 0)
 923                                 rra_timespans_custom_num++;
 924                 } /* while (strtok_r) */
 925                 free (value_copy);
 926         }
 927         else if (strcasecmp ("XFF", key) == 0)
 928         {
 929                 double tmp = atof (value);
 930                 if ((tmp < 0.0) || (tmp >= 1.0))
 931                 {
 932                         fprintf (stderr, "rrdtool: `XFF' must "
 933                                         "be in the range 0 to 1 (exclusive).");
 934                         return (1);
 935                 }
 936                 xff = tmp;
 937         }
 938         else
 939         {
 940                 return (-1);
 941         }
 942         return (0);
 943 } /* int rrd_config */
 944
 945 static int rrd_shutdown (void)
 946 {
 947         pthread_mutex_lock (&cache_lock);
 948         rrd_cache_flush (-1);
 949         pthread_mutex_unlock (&cache_lock);
 950
 951         pthread_mutex_lock (&queue_lock);
 952         do_shutdown = 1;
 953         pthread_cond_signal (&queue_cond);
 954         pthread_mutex_unlock (&queue_lock);
 955
 956         return (0);
 957 } /* int rrd_shutdown */
 958
 959 static int rrd_init (void)
 960 {
 961         int status;
 962
 963         if (stepsize <= 0)
 964                 stepsize = interval_g;
 965         if (heartbeat <= 0)
 966                 heartbeat = 2 * interval_g;
 967
 968         if (heartbeat < interval_g)
 969                 WARNING ("rrdtool plugin: Your `heartbeat' is "
 970                                 "smaller than your `interval'. This will "
 971                                 "likely cause problems.");
 972         else if (stepsize < interval_g)
 973                 WARNING ("rrdtool plugin: Your `stepsize' is "
 974                                 "smaller than your `interval'. This will "
 975                                 "create needlessly big RRD-files.");
 976
 977         /* Set the cache up */
 978         pthread_mutex_lock (&cache_lock);
 979
 980         cache = avl_create ((int (*) (const void *, const void *)) strcmp);
 981         if (cache == NULL)
 982         {
 983                 ERROR ("rrdtool plugin: avl_create failed.");
 984                 return (-1);
 985         }
 986
 987         cache_flush_last = time (NULL);
 988         if (cache_timeout < 2)
 989         {
 990                 cache_timeout = 0;
 991                 cache_flush_timeout = 0;
 992         }
 993         else if (cache_flush_timeout < cache_timeout)
 994                 cache_flush_timeout = 10 * cache_timeout;
 995
 996         pthread_mutex_unlock (&cache_lock);
 997
 998         status = pthread_create (&queue_thread, NULL, rrd_queue_thread, NULL);
 999         if (status != 0)
1000         {
1001                 ERROR ("rrdtool plugin: Cannot create queue-thread.");
1002                 return (-1);
1003         }
1004
1005         DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %i;"
1006                         " heartbeat = %i; rrarows = %i; xff = %lf;",
1007                         (datadir == NULL) ? "(null)" : datadir,
1008                         stepsize, heartbeat, rrarows, xff);
1009
1010         return (0);
1011 } /* int rrd_init */
1012
1013 void module_register (void)
1014 {
1015         plugin_register_config ("rrdtool", rrd_config,
1016                         config_keys, config_keys_num);
1017         plugin_register_init ("rrdtool", rrd_init);
1018         plugin_register_write ("rrdtool", rrd_write);
1019         plugin_register_shutdown ("rrdtool", rrd_shutdown);
1020 }