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         int status;
 464
 465         if (values_num < 1)
 466                 return (0);
 467
 468         argc = values_num + 2;
 469         argv = (char **) malloc ((argc + 1) * sizeof (char *));
 470         if (argv == NULL)
 471                 return (-1);
 472
 473         argv[0] = "update";
 474         argv[1] = filename;
 475         memcpy (argv + 2, values, values_num * sizeof (char *));
 476         argv[argc] = NULL;
 477
 478         DEBUG ("rrd_update (argc = %i, argv = %p)", argc, (void *) argv);
 479
 480         optind = 0; /* bug in librrd? */
 481         rrd_clear_error ();
 482         status = rrd_update (argc, argv);
 483         if (status != 0)
 484         {
 485                 WARNING ("rrd_update failed: %s: %s",
 486                                 filename, rrd_get_error ());
 487                 status = -1;
 488         }
 489
 490         sfree (argv);
 491
 492         return (status);
 493 } /* int rrd_write_cache_entry */
 494
 495 static void *rrd_queue_thread (void *data)
 496 {
 497         while (42)
 498         {
 499                 rrd_queue_t *queue_entry;
 500                 rrd_cache_t *cache_entry;
 501                 char **values;
 502                 int    values_num;
 503                 int    i;
 504
 505                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 506                  * the same time, ALWAYS lock `cache_lock' first! */
 507
 508                 /* wait until an entry is available */
 509                 pthread_mutex_lock (&queue_lock);
 510                 while ((queue_head == NULL) && (do_shutdown == 0))
 511                         pthread_cond_wait (&queue_cond, &queue_lock);
 512
 513                 /* We're in the shutdown phase */
 514                 if (queue_head == NULL)
 515                 {
 516                         pthread_mutex_unlock (&queue_lock);
 517                         break;
 518                 }
 519
 520                 /* Dequeue the first entry */
 521                 queue_entry = queue_head;
 522                 if (queue_head == queue_tail)
 523                         queue_head = queue_tail = NULL;
 524                 else
 525                         queue_head = queue_head->next;
 526
 527                 /* Unlock the queue again */
 528                 pthread_mutex_unlock (&queue_lock);
 529
 530                 /* We now need the cache lock so the entry isn't updated while
 531                  * we make a copy of it's values */
 532                 pthread_mutex_lock (&cache_lock);
 533
 534                 avl_get (cache, queue_entry->filename, (void *) &cache_entry);
 535
 536                 values = cache_entry->values;
 537                 values_num = cache_entry->values_num;
 538
 539                 cache_entry->values = NULL;
 540                 cache_entry->values_num = 0;
 541                 cache_entry->flags = FLAG_NONE;
 542
 543                 pthread_mutex_unlock (&cache_lock);
 544
 545                 /* Write the values to the RRD-file */
 546                 rrd_write_to_file (queue_entry->filename, values, values_num);
 547
 548                 for (i = 0; i < values_num; i++)
 549                 {
 550                         sfree (values[i]);
 551                 }
 552                 sfree (values);
 553                 sfree (queue_entry->filename);
 554                 sfree (queue_entry);
 555         } /* while (42) */
 556
 557         pthread_mutex_lock (&cache_lock);
 558         avl_destroy (cache);
 559         cache = NULL;
 560         pthread_mutex_unlock (&cache_lock);
 561
 562         pthread_exit ((void *) 0);
 563         return ((void *) 0);
 564 } /* void *rrd_queue_thread */
 565
 566 static int rrd_queue_cache_entry (const char *filename)
 567 {
 568         rrd_queue_t *queue_entry;
 569
 570         queue_entry = (rrd_queue_t *) malloc (sizeof (rrd_queue_t));
 571         if (queue_entry == NULL)
 572                 return (-1);
 573
 574         queue_entry->filename = strdup (filename);
 575         if (queue_entry->filename == NULL)
 576         {
 577                 free (queue_entry);
 578                 return (-1);
 579         }
 580
 581         queue_entry->next = NULL;
 582
 583         pthread_mutex_lock (&queue_lock);
 584         if (queue_tail == NULL)
 585                 queue_head = queue_entry;
 586         else
 587                 queue_tail->next = queue_entry;
 588         queue_tail = queue_entry;
 589         pthread_cond_signal (&queue_cond);
 590         pthread_mutex_unlock (&queue_lock);
 591
 592         DEBUG ("rrdtool plugin: Put `%s' into the update queue", filename);
 593
 594         return (0);
 595 } /* int rrd_queue_cache_entry */
 596
 597 static void rrd_cache_flush (int timeout)
 598 {
 599         rrd_cache_t *rc;
 600         time_t       now;
 601
 602         char **keys = NULL;
 603         int    keys_num = 0;
 604
 605         char *key;
 606         avl_iterator_t *iter;
 607         int i;
 608
 609         DEBUG ("Flushing cache, timeout = %i", timeout);
 610
 611         now = time (NULL);
 612
 613         /* Build a list of entries to be flushed */
 614         iter = avl_get_iterator (cache);
 615         while (avl_iterator_next (iter, (void *) &key, (void *) &rc) == 0)
 616         {
 617                 DEBUG ("key = %s; age = %i;", key, now - rc->first_value);
 618
 619                 if (rc->flags == FLAG_QUEUED)
 620                         continue;
 621                 else if ((now - rc->first_value) < timeout)
 622                         continue;
 623                 else if (rc->values_num > 0)
 624                 {
 625                         if (rrd_queue_cache_entry (key) == 0)
 626                                 rc->flags = FLAG_QUEUED;
 627                 }
 628                 else /* ancient and no values -> waste of memory */
 629                 {
 630                         keys = (char **) realloc ((void *) keys,
 631                                         (keys_num + 1) * sizeof (char *));
 632                         if (keys == NULL)
 633                         {
 634                                 char errbuf[1024];
 635                                 ERROR ("rrdtool plugin: "
 636                                                 "realloc failed: %s",
 637                                                 sstrerror (errno, errbuf,
 638                                                         sizeof (errbuf)));
 639                                 avl_iterator_destroy (iter);
 640                                 return;
 641                         }
 642                         keys[keys_num] = key;
 643                         keys_num++;
 644                 }
 645         } /* while (avl_iterator_next) */
 646         avl_iterator_destroy (iter);
 647
 648         for (i = 0; i < keys_num; i++)
 649         {
 650                 if (avl_remove (cache, keys[i], (void *) &key, (void *) &rc) != 0)
 651                 {
 652                         DEBUG ("avl_remove (%s) failed.", keys[i]);
 653                         continue;
 654                 }
 655
 656                 assert (rc->values == NULL);
 657                 assert (rc->values_num == 0);
 658
 659                 sfree (rc);
 660                 sfree (key);
 661                 keys[i] = NULL;
 662         } /* for (i = 0..keys_num) */
 663
 664         free (keys);
 665         DEBUG ("Flushed %i value(s)", keys_num);
 666
 667         cache_flush_last = now;
 668 } /* void rrd_cache_flush */
 669
 670 static int rrd_cache_insert (const char *filename,
 671                 const char *value, time_t value_time)
 672 {
 673         rrd_cache_t *rc = NULL;
 674         int new_rc = 0;
 675         char **values_new;
 676
 677         pthread_mutex_lock (&cache_lock);
 678
 679         avl_get (cache, filename, (void *) &rc);
 680
 681         if (rc == NULL)
 682         {
 683                 rc = (rrd_cache_t *) malloc (sizeof (rrd_cache_t));
 684                 if (rc == NULL)
 685                         return (-1);
 686                 rc->values_num = 0;
 687                 rc->values = NULL;
 688                 rc->first_value = 0;
 689                 rc->last_value = 0;
 690                 rc->flags = FLAG_NONE;
 691                 new_rc = 1;
 692         }
 693
 694         if (rc->last_value >= value_time)
 695         {
 696                 pthread_mutex_unlock (&cache_lock);
 697                 WARNING ("rrdtool plugin: (rc->last_value = %u) >= (value_time = %u)",
 698                                 (unsigned int) rc->last_value,
 699                                 (unsigned int) value_time);
 700                 return (-1);
 701         }
 702
 703         values_new = (char **) realloc ((void *) rc->values,
 704                         (rc->values_num + 1) * sizeof (char *));
 705         if (values_new == NULL)
 706         {
 707                 char errbuf[1024];
 708                 void *cache_key = NULL;
 709
 710                 sstrerror (errno, errbuf, sizeof (errbuf));
 711
 712                 avl_remove (cache, filename, &cache_key, NULL);
 713                 pthread_mutex_unlock (&cache_lock);
 714
 715                 ERROR ("rrdtool plugin: realloc failed: %s", errbuf);
 716
 717                 sfree (cache_key);
 718                 sfree (rc->values);
 719                 sfree (rc);
 720                 return (-1);
 721         }
 722         rc->values = values_new;
 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         {
 760                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 761                  * the same time, ALWAYS lock `cache_lock' first! */
 762                 if (rc->flags != FLAG_QUEUED)
 763                 {
 764                         if (rrd_queue_cache_entry (filename) == 0)
 765                                 rc->flags = FLAG_QUEUED;
 766                 }
 767                 else
 768                 {
 769                         DEBUG ("rrdtool plugin: `%s' is already queued.", filename);
 770                 }
 771         }
 772
 773         if ((cache_timeout > 0) &&
 774                         ((time (NULL) - cache_flush_last) > cache_flush_timeout))
 775                 rrd_cache_flush (cache_flush_timeout);
 776
 777
 778         pthread_mutex_unlock (&cache_lock);
 779
 780         return (0);
 781 } /* int rrd_cache_insert */
 782
 783 static int rrd_write (const data_set_t *ds, const value_list_t *vl)
 784 {
 785         struct stat  statbuf;
 786         char         filename[512];
 787         char         values[512];
 788         int          status;
 789
 790         if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
 791                 return (-1);
 792
 793         if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
 794                 return (-1);
 795
 796         if (stat (filename, &statbuf) == -1)
 797         {
 798                 if (errno == ENOENT)
 799                 {
 800                         if (rrd_create_file (filename, ds))
 801                                 return (-1);
 802                 }
 803                 else
 804                 {
 805                         char errbuf[1024];
 806                         ERROR ("stat(%s) failed: %s", filename,
 807                                         sstrerror (errno, errbuf,
 808                                                 sizeof (errbuf)));
 809                         return (-1);
 810                 }
 811         }
 812         else if (!S_ISREG (statbuf.st_mode))
 813         {
 814                 ERROR ("stat(%s): Not a regular file!",
 815                                 filename);
 816                 return (-1);
 817         }
 818
 819         status = rrd_cache_insert (filename, values, vl->time);
 820
 821         return (status);
 822 } /* int rrd_write */
 823
 824 static int rrd_config (const char *key, const char *value)
 825 {
 826         if (strcasecmp ("CacheTimeout", key) == 0)
 827         {
 828                 int tmp = atoi (value);
 829                 if (tmp < 0)
 830                 {
 831                         fprintf (stderr, "rrdtool: `CacheTimeout' must "
 832                                         "be greater than 0.\n");
 833                         return (1);
 834                 }
 835                 cache_timeout = tmp;
 836         }
 837         else if (strcasecmp ("CacheFlush", key) == 0)
 838         {
 839                 int tmp = atoi (value);
 840                 if (tmp < 0)
 841                 {
 842                         fprintf (stderr, "rrdtool: `CacheFlush' must "
 843                                         "be greater than 0.\n");
 844                         return (1);
 845                 }
 846                 cache_flush_timeout = tmp;
 847         }
 848         else if (strcasecmp ("DataDir", key) == 0)
 849         {
 850                 if (datadir != NULL)
 851                         free (datadir);
 852                 datadir = strdup (value);
 853                 if (datadir != NULL)
 854                 {
 855                         int len = strlen (datadir);
 856                         while ((len > 0) && (datadir[len - 1] == '/'))
 857                         {
 858                                 len--;
 859                                 datadir[len] = '\0';
 860                         }
 861                         if (len <= 0)
 862                         {
 863                                 free (datadir);
 864                                 datadir = NULL;
 865                         }
 866                 }
 867         }
 868         else if (strcasecmp ("StepSize", key) == 0)
 869         {
 870                 int tmp = atoi (value);
 871                 if (tmp <= 0)
 872                 {
 873                         fprintf (stderr, "rrdtool: `StepSize' must "
 874                                         "be greater than 0.\n");
 875                         return (1);
 876                 }
 877                 stepsize = tmp;
 878         }
 879         else if (strcasecmp ("HeartBeat", key) == 0)
 880         {
 881                 int tmp = atoi (value);
 882                 if (tmp <= 0)
 883                 {
 884                         fprintf (stderr, "rrdtool: `HeartBeat' must "
 885                                         "be greater than 0.\n");
 886                         return (1);
 887                 }
 888                 heartbeat = tmp;
 889         }
 890         else if (strcasecmp ("RRARows", key) == 0)
 891         {
 892                 int tmp = atoi (value);
 893                 if (tmp <= 0)
 894                 {
 895                         fprintf (stderr, "rrdtool: `RRARows' must "
 896                                         "be greater than 0.\n");
 897                         return (1);
 898                 }
 899                 rrarows = tmp;
 900         }
 901         else if (strcasecmp ("RRATimespan", key) == 0)
 902         {
 903                 char *saveptr = NULL;
 904                 char *dummy;
 905                 char *ptr;
 906                 char *value_copy;
 907                 int *tmp_alloc;
 908
 909                 value_copy = strdup (value);
 910                 if (value_copy == NULL)
 911                         return (1);
 912
 913                 dummy = value_copy;
 914                 while ((ptr = strtok_r (dummy, ", \t", &saveptr)) != NULL)
 915                 {
 916                         dummy = NULL;
 917
 918                         tmp_alloc = realloc (rra_timespans_custom,
 919                                         sizeof (int) * (rra_timespans_custom_num + 1));
 920                         if (tmp_alloc == NULL)
 921                         {
 922                                 fprintf (stderr, "rrdtool: realloc failed.\n");
 923                                 free (value_copy);
 924                                 return (1);
 925                         }
 926                         rra_timespans_custom = tmp_alloc;
 927                         rra_timespans_custom[rra_timespans_custom_num] = atoi (ptr);
 928                         if (rra_timespans_custom[rra_timespans_custom_num] != 0)
 929                                 rra_timespans_custom_num++;
 930                 } /* while (strtok_r) */
 931                 free (value_copy);
 932         }
 933         else if (strcasecmp ("XFF", key) == 0)
 934         {
 935                 double tmp = atof (value);
 936                 if ((tmp < 0.0) || (tmp >= 1.0))
 937                 {
 938                         fprintf (stderr, "rrdtool: `XFF' must "
 939                                         "be in the range 0 to 1 (exclusive).");
 940                         return (1);
 941                 }
 942                 xff = tmp;
 943         }
 944         else
 945         {
 946                 return (-1);
 947         }
 948         return (0);
 949 } /* int rrd_config */
 950
 951 static int rrd_shutdown (void)
 952 {
 953         pthread_mutex_lock (&cache_lock);
 954         rrd_cache_flush (-1);
 955         pthread_mutex_unlock (&cache_lock);
 956
 957         pthread_mutex_lock (&queue_lock);
 958         do_shutdown = 1;
 959         pthread_cond_signal (&queue_cond);
 960         pthread_mutex_unlock (&queue_lock);
 961
 962         return (0);
 963 } /* int rrd_shutdown */
 964
 965 static int rrd_init (void)
 966 {
 967         int status;
 968
 969         if (stepsize <= 0)
 970                 stepsize = interval_g;
 971         if (heartbeat <= 0)
 972                 heartbeat = 2 * interval_g;
 973
 974         if (heartbeat < interval_g)
 975                 WARNING ("rrdtool plugin: Your `heartbeat' is "
 976                                 "smaller than your `interval'. This will "
 977                                 "likely cause problems.");
 978         else if (stepsize < interval_g)
 979                 WARNING ("rrdtool plugin: Your `stepsize' is "
 980                                 "smaller than your `interval'. This will "
 981                                 "create needlessly big RRD-files.");
 982
 983         /* Set the cache up */
 984         pthread_mutex_lock (&cache_lock);
 985
 986         cache = avl_create ((int (*) (const void *, const void *)) strcmp);
 987         if (cache == NULL)
 988         {
 989                 ERROR ("rrdtool plugin: avl_create failed.");
 990                 return (-1);
 991         }
 992
 993         cache_flush_last = time (NULL);
 994         if (cache_timeout < 2)
 995         {
 996                 cache_timeout = 0;
 997                 cache_flush_timeout = 0;
 998         }
 999         else if (cache_flush_timeout < cache_timeout)
1000                 cache_flush_timeout = 10 * cache_timeout;
1001
1002         pthread_mutex_unlock (&cache_lock);
1003
1004         status = pthread_create (&queue_thread, NULL, rrd_queue_thread, NULL);
1005         if (status != 0)
1006         {
1007                 ERROR ("rrdtool plugin: Cannot create queue-thread.");
1008                 return (-1);
1009         }
1010
1011         DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %i;"
1012                         " heartbeat = %i; rrarows = %i; xff = %lf;",
1013                         (datadir == NULL) ? "(null)" : datadir,
1014                         stepsize, heartbeat, rrarows, xff);
1015
1016         return (0);
1017 } /* int rrd_init */
1018
1019 void module_register (void)
1020 {
1021         plugin_register_config ("rrdtool", rrd_config,
1022                         config_keys, config_keys_num);
1023         plugin_register_init ("rrdtool", rrd_init);
1024         plugin_register_write ("rrdtool", rrd_write);
1025         plugin_register_shutdown ("rrdtool", rrd_shutdown);
1026 }