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, const 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,
 354                 int argc, const char **argv)
 355 {
 356         int status;
 357
 358         optind = 0; /* bug in librrd? */
 359         rrd_clear_error ();
 360
 361         status = rrd_update_r (filename, template, argc, argv);
 362
 363         if (status != 0)
 364         {
 365                 WARNING ("rrdtool plugin: rrd_update_r (%s) failed: %s",
 366                                 filename, rrd_get_error ());
 367         }
 368
 369         return (status);
 370 } /* int srrd_update */
 371 /* #endif HAVE_THREADSAFE_LIBRRD */
 372
 373 #else /* !HAVE_THREADSAFE_LIBRRD */
 374 static int srrd_create (char *filename, unsigned long pdp_step, time_t last_up,
 375                 int argc, const char **argv)
 376 {
 377         int status;
 378
 379         int new_argc;
 380         char **new_argv;
 381
 382         char pdp_step_str[16];
 383         char last_up_str[16];
 384
 385         new_argc = 6 + argc;
 386         new_argv = (char **) malloc ((new_argc + 1) * sizeof (char *));
 387         if (new_argv == NULL)
 388         {
 389                 ERROR ("rrdtool plugin: malloc failed.");
 390                 return (-1);
 391         }
 392
 393         if (last_up == 0)
 394                 last_up = time (NULL) - 10;
 395
 396         ssnprintf (pdp_step_str, sizeof (pdp_step_str), "%lu", pdp_step);
 397         ssnprintf (last_up_str, sizeof (last_up_str), "%u", (unsigned int) last_up);
 398
 399         new_argv[0] = "create";
 400         new_argv[1] = filename;
 401         new_argv[2] = "-s";
 402         new_argv[3] = pdp_step_str;
 403         new_argv[4] = "-b";
 404         new_argv[5] = last_up_str;
 405
 406         memcpy (new_argv + 6, argv, argc * sizeof (char *));
 407         new_argv[new_argc] = NULL;
 408
 409         pthread_mutex_lock (&librrd_lock);
 410         optind = 0; /* bug in librrd? */
 411         rrd_clear_error ();
 412
 413         status = rrd_create (new_argc, new_argv);
 414         pthread_mutex_unlock (&librrd_lock);
 415
 416         if (status != 0)
 417         {
 418                 WARNING ("rrdtool plugin: rrd_create (%s) failed: %s",
 419                                 filename, rrd_get_error ());
 420         }
 421
 422         sfree (new_argv);
 423
 424         return (status);
 425 } /* int srrd_create */
 426
 427 static int srrd_update (char *filename, char *template,
 428                 int argc, const char **argv)
 429 {
 430         int status;
 431
 432         int new_argc;
 433         char **new_argv;
 434
 435         assert (template == NULL);
 436
 437         new_argc = 2 + argc;
 438         new_argv = (char **) malloc ((new_argc + 1) * sizeof (char *));
 439         if (new_argv == NULL)
 440         {
 441                 ERROR ("rrdtool plugin: malloc failed.");
 442                 return (-1);
 443         }
 444
 445         new_argv[0] = "update";
 446         new_argv[1] = filename;
 447
 448         memcpy (new_argv + 2, argv, argc * sizeof (char *));
 449         new_argv[new_argc] = NULL;
 450
 451         pthread_mutex_lock (&librrd_lock);
 452         optind = 0; /* bug in librrd? */
 453         rrd_clear_error ();
 454
 455         status = rrd_update (new_argc, new_argv);
 456         pthread_mutex_unlock (&librrd_lock);
 457
 458         if (status != 0)
 459         {
 460                 WARNING ("rrdtool plugin: rrd_update_r failed: %s: %s",
 461                                 argv[1], rrd_get_error ());
 462         }
 463
 464         sfree (new_argv);
 465
 466         return (status);
 467 } /* int srrd_update */
 468 #endif /* !HAVE_THREADSAFE_LIBRRD */
 469
 470 static int rrd_create_file (char *filename, const data_set_t *ds, const value_list_t *vl)
 471 {
 472         char **argv;
 473         int argc;
 474         char **rra_def;
 475         int rra_num;
 476         char **ds_def;
 477         int ds_num;
 478         int status = 0;
 479
 480         if (check_create_dir (filename))
 481                 return (-1);
 482
 483         if ((rra_num = rra_get (&rra_def, vl)) < 1)
 484         {
 485                 ERROR ("rrd_create_file failed: Could not calculate RRAs");
 486                 return (-1);
 487         }
 488
 489         if ((ds_num = ds_get (&ds_def, ds, vl)) < 1)
 490         {
 491                 ERROR ("rrd_create_file failed: Could not calculate DSes");
 492                 return (-1);
 493         }
 494
 495         argc = ds_num + rra_num;
 496
 497         if ((argv = (char **) malloc (sizeof (char *) * (argc + 1))) == NULL)
 498         {
 499                 char errbuf[1024];
 500                 ERROR ("rrd_create failed: %s",
 501                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 502                 return (-1);
 503         }
 504
 505         memcpy (argv, ds_def, ds_num * sizeof (char *));
 506         memcpy (argv + ds_num, rra_def, rra_num * sizeof (char *));
 507         argv[ds_num + rra_num] = NULL;
 508
 509         assert (vl->time > 10);
 510         status = srrd_create (filename,
 511                         (stepsize > 0) ? stepsize : vl->interval,
 512                         vl->time - 10,
 513                         argc, (const char **)argv);
 514
 515         free (argv);
 516         ds_free (ds_num, ds_def);
 517         rra_free (rra_num, rra_def);
 518
 519         return (status);
 520 }
 521
 522 static int value_list_to_string (char *buffer, int buffer_len,
 523                 const data_set_t *ds, const value_list_t *vl)
 524 {
 525         int offset;
 526         int status;
 527         int i;
 528
 529         memset (buffer, '\0', buffer_len);
 530
 531         status = ssnprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
 532         if ((status < 1) || (status >= buffer_len))
 533                 return (-1);
 534         offset = status;
 535
 536         for (i = 0; i < ds->ds_num; i++)
 537         {
 538                 if ((ds->ds[i].type != DS_TYPE_COUNTER)
 539                                 && (ds->ds[i].type != DS_TYPE_GAUGE))
 540                         return (-1);
 541
 542                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 543                         status = ssnprintf (buffer + offset, buffer_len - offset,
 544                                         ":%llu", vl->values[i].counter);
 545                 else
 546                         status = ssnprintf (buffer + offset, buffer_len - offset,
 547                                         ":%lf", vl->values[i].gauge);
 548
 549                 if ((status < 1) || (status >= (buffer_len - offset)))
 550                         return (-1);
 551
 552                 offset += status;
 553         } /* for ds->ds_num */
 554
 555         return (0);
 556 } /* int value_list_to_string */
 557
 558 static int value_list_to_filename (char *buffer, int buffer_len,
 559                 const data_set_t *ds, const value_list_t *vl)
 560 {
 561         int offset = 0;
 562         int status;
 563
 564         if (datadir != NULL)
 565         {
 566                 status = ssnprintf (buffer + offset, buffer_len - offset,
 567                                 "%s/", datadir);
 568                 if ((status < 1) || (status >= buffer_len - offset))
 569                         return (-1);
 570                 offset += status;
 571         }
 572
 573         status = ssnprintf (buffer + offset, buffer_len - offset,
 574                         "%s/", vl->host);
 575         if ((status < 1) || (status >= buffer_len - offset))
 576                 return (-1);
 577         offset += status;
 578
 579         if (strlen (vl->plugin_instance) > 0)
 580                 status = ssnprintf (buffer + offset, buffer_len - offset,
 581                                 "%s-%s/", vl->plugin, vl->plugin_instance);
 582         else
 583                 status = ssnprintf (buffer + offset, buffer_len - offset,
 584                                 "%s/", vl->plugin);
 585         if ((status < 1) || (status >= buffer_len - offset))
 586                 return (-1);
 587         offset += status;
 588
 589         if (strlen (vl->type_instance) > 0)
 590                 status = ssnprintf (buffer + offset, buffer_len - offset,
 591                                 "%s-%s.rrd", vl->type, vl->type_instance);
 592         else
 593                 status = ssnprintf (buffer + offset, buffer_len - offset,
 594                                 "%s.rrd", vl->type);
 595         if ((status < 1) || (status >= buffer_len - offset))
 596                 return (-1);
 597         offset += status;
 598
 599         return (0);
 600 } /* int value_list_to_filename */
 601
 602 static void *rrd_queue_thread (void *data)
 603 {
 604         while (42)
 605         {
 606                 rrd_queue_t *queue_entry;
 607                 rrd_cache_t *cache_entry;
 608                 char **values;
 609                 int    values_num;
 610                 int    i;
 611
 612                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 613                  * the same time, ALWAYS lock `cache_lock' first! */
 614
 615                 /* wait until an entry is available */
 616                 pthread_mutex_lock (&queue_lock);
 617                 while ((queue_head == NULL) && (do_shutdown == 0))
 618                         pthread_cond_wait (&queue_cond, &queue_lock);
 619
 620                 /* We're in the shutdown phase */
 621                 if (queue_head == NULL)
 622                 {
 623                         pthread_mutex_unlock (&queue_lock);
 624                         break;
 625                 }
 626
 627                 /* Dequeue the first entry */
 628                 queue_entry = queue_head;
 629                 if (queue_head == queue_tail)
 630                         queue_head = queue_tail = NULL;
 631                 else
 632                         queue_head = queue_head->next;
 633
 634                 /* Unlock the queue again */
 635                 pthread_mutex_unlock (&queue_lock);
 636
 637                 /* We now need the cache lock so the entry isn't updated while
 638                  * we make a copy of it's values */
 639                 pthread_mutex_lock (&cache_lock);
 640
 641                 c_avl_get (cache, queue_entry->filename, (void *) &cache_entry);
 642
 643                 values = cache_entry->values;
 644                 values_num = cache_entry->values_num;
 645
 646                 cache_entry->values = NULL;
 647                 cache_entry->values_num = 0;
 648                 cache_entry->flags = FLAG_NONE;
 649
 650                 pthread_mutex_unlock (&cache_lock);
 651
 652                 /* Write the values to the RRD-file */
 653                 srrd_update (queue_entry->filename, NULL,
 654                                 values_num, (const char **)values);
 655                 DEBUG ("rrdtool plugin: queue thread: Wrote %i values to %s",
 656                                 values_num, queue_entry->filename);
 657
 658                 for (i = 0; i < values_num; i++)
 659                 {
 660                         sfree (values[i]);
 661                 }
 662                 sfree (values);
 663                 sfree (queue_entry->filename);
 664                 sfree (queue_entry);
 665         } /* while (42) */
 666
 667         pthread_mutex_lock (&cache_lock);
 668         c_avl_destroy (cache);
 669         cache = NULL;
 670         pthread_mutex_unlock (&cache_lock);
 671
 672         pthread_exit ((void *) 0);
 673         return ((void *) 0);
 674 } /* void *rrd_queue_thread */
 675
 676 static int rrd_queue_cache_entry (const char *filename, rrd_queue_dir_t dir)
 677 {
 678   rrd_queue_t *queue_entry;
 679
 680   queue_entry = (rrd_queue_t *) malloc (sizeof (rrd_queue_t));
 681   if (queue_entry == NULL)
 682     return (-1);
 683
 684   queue_entry->filename = strdup (filename);
 685   if (queue_entry->filename == NULL)
 686   {
 687     free (queue_entry);
 688     return (-1);
 689   }
 690
 691   queue_entry->next = NULL;
 692
 693   pthread_mutex_lock (&queue_lock);
 694   if (dir == QUEUE_INSERT_FRONT)
 695   {
 696     queue_entry->next = queue_head;
 697     queue_head = queue_entry;
 698     if (queue_tail == NULL)
 699       queue_tail = queue_head;
 700   }
 701   else /* (dir == QUEUE_INSERT_BACK) */
 702   {
 703     if (queue_tail == NULL)
 704       queue_head = queue_entry;
 705     else
 706       queue_tail->next = queue_entry;
 707     queue_tail = queue_entry;
 708   }
 709   pthread_cond_signal (&queue_cond);
 710   pthread_mutex_unlock (&queue_lock);
 711
 712   DEBUG ("rrdtool plugin: Put `%s' into the update queue", filename);
 713
 714   return (0);
 715 } /* int rrd_queue_cache_entry */
 716
 717 static int rrd_queue_move_to_front (const char *filename)
 718 {
 719   rrd_queue_t *this;
 720   rrd_queue_t *prev;
 721
 722   this = NULL;
 723   prev = NULL;
 724   pthread_mutex_lock (&queue_lock);
 725   for (this = queue_head; this != NULL; this = this->next)
 726   {
 727     if (strcmp (this->filename, filename) == 0)
 728       break;
 729     prev = this;
 730   }
 731
 732   /* Check if we found the entry and if it is NOT the first entry. */
 733   if ((this != NULL) && (prev != NULL))
 734   {
 735     prev->next = this->next;
 736     this->next = queue_head;
 737     queue_head = this;
 738   }
 739   pthread_mutex_unlock (&queue_lock);
 740
 741   return (0);
 742 } /* int rrd_queue_move_to_front */
 743
 744 static void rrd_cache_flush (int timeout)
 745 {
 746         rrd_cache_t *rc;
 747         time_t       now;
 748
 749         char **keys = NULL;
 750         int    keys_num = 0;
 751
 752         char *key;
 753         c_avl_iterator_t *iter;
 754         int i;
 755
 756         DEBUG ("rrdtool plugin: Flushing cache, timeout = %i", timeout);
 757
 758         now = time (NULL);
 759
 760         /* Build a list of entries to be flushed */
 761         iter = c_avl_get_iterator (cache);
 762         while (c_avl_iterator_next (iter, (void *) &key, (void *) &rc) == 0)
 763         {
 764                 if (rc->flags == FLAG_QUEUED)
 765                         continue;
 766                 else if ((now - rc->first_value) < timeout)
 767                         continue;
 768                 else if (rc->values_num > 0)
 769                 {
 770                         if (rrd_queue_cache_entry (key, QUEUE_INSERT_BACK) == 0)
 771                                 rc->flags = FLAG_QUEUED;
 772                 }
 773                 else /* ancient and no values -> waste of memory */
 774                 {
 775                         char **tmp = (char **) realloc ((void *) keys,
 776                                         (keys_num + 1) * sizeof (char *));
 777                         if (tmp == NULL)
 778                         {
 779                                 char errbuf[1024];
 780                                 ERROR ("rrdtool plugin: "
 781                                                 "realloc failed: %s",
 782                                                 sstrerror (errno, errbuf,
 783                                                         sizeof (errbuf)));
 784                                 c_avl_iterator_destroy (iter);
 785                                 sfree (keys);
 786                                 return;
 787                         }
 788                         keys = tmp;
 789                         keys[keys_num] = key;
 790                         keys_num++;
 791                 }
 792         } /* while (c_avl_iterator_next) */
 793         c_avl_iterator_destroy (iter);
 794
 795         for (i = 0; i < keys_num; i++)
 796         {
 797                 if (c_avl_remove (cache, keys[i], (void *) &key, (void *) &rc) != 0)
 798                 {
 799                         DEBUG ("rrdtool plugin: c_avl_remove (%s) failed.", keys[i]);
 800                         continue;
 801                 }
 802
 803                 assert (rc->values == NULL);
 804                 assert (rc->values_num == 0);
 805
 806                 sfree (rc);
 807                 sfree (key);
 808                 keys[i] = NULL;
 809         } /* for (i = 0..keys_num) */
 810
 811         sfree (keys);
 812
 813         cache_flush_last = now;
 814 } /* void rrd_cache_flush */
 815
 816 static int rrd_cache_flush_identifier (int timeout, const char *identifier)
 817 {
 818   rrd_cache_t *rc;
 819   time_t now;
 820   int status;
 821   char key[2048];
 822
 823   if (identifier == NULL)
 824   {
 825     rrd_cache_flush (timeout);
 826     return (0);
 827   }
 828
 829   now = time (NULL);
 830
 831   if (datadir == NULL)
 832           snprintf (key, sizeof (key), "%s.rrd",
 833                           identifier);
 834   else
 835           snprintf (key, sizeof (key), "%s/%s.rrd",
 836                           datadir, identifier);
 837   key[sizeof (key) - 1] = 0;
 838
 839   status = c_avl_get (cache, key, (void *) &rc);
 840   if (status != 0)
 841   {
 842     WARNING ("rrdtool plugin: rrd_cache_flush_identifier: "
 843         "c_avl_get (%s) failed. Does that file really exist?",
 844         key);
 845     return (status);
 846   }
 847
 848   if (rc->flags == FLAG_QUEUED)
 849     status = rrd_queue_move_to_front (key);
 850   else if ((now - rc->first_value) < timeout)
 851     status = 0;
 852   else if (rc->values_num > 0)
 853   {
 854     status = rrd_queue_cache_entry (key, QUEUE_INSERT_FRONT);
 855     if (status == 0)
 856       rc->flags = FLAG_QUEUED;
 857   }
 858
 859   return (status);
 860 } /* int rrd_cache_flush_identifier */
 861
 862 static int rrd_cache_insert (const char *filename,
 863                 const char *value, time_t value_time)
 864 {
 865         rrd_cache_t *rc = NULL;
 866         int new_rc = 0;
 867         char **values_new;
 868
 869         pthread_mutex_lock (&cache_lock);
 870
 871         c_avl_get (cache, filename, (void *) &rc);
 872
 873         if (rc == NULL)
 874         {
 875                 rc = (rrd_cache_t *) malloc (sizeof (rrd_cache_t));
 876                 if (rc == NULL)
 877                         return (-1);
 878                 rc->values_num = 0;
 879                 rc->values = NULL;
 880                 rc->first_value = 0;
 881                 rc->last_value = 0;
 882                 rc->flags = FLAG_NONE;
 883                 new_rc = 1;
 884         }
 885
 886         if (rc->last_value >= value_time)
 887         {
 888                 pthread_mutex_unlock (&cache_lock);
 889                 WARNING ("rrdtool plugin: (rc->last_value = %u) >= (value_time = %u)",
 890                                 (unsigned int) rc->last_value,
 891                                 (unsigned int) value_time);
 892                 return (-1);
 893         }
 894
 895         values_new = (char **) realloc ((void *) rc->values,
 896                         (rc->values_num + 1) * sizeof (char *));
 897         if (values_new == NULL)
 898         {
 899                 char errbuf[1024];
 900                 void *cache_key = NULL;
 901
 902                 sstrerror (errno, errbuf, sizeof (errbuf));
 903
 904                 c_avl_remove (cache, filename, &cache_key, NULL);
 905                 pthread_mutex_unlock (&cache_lock);
 906
 907                 ERROR ("rrdtool plugin: realloc failed: %s", errbuf);
 908
 909                 sfree (cache_key);
 910                 sfree (rc->values);
 911                 sfree (rc);
 912                 return (-1);
 913         }
 914         rc->values = values_new;
 915
 916         rc->values[rc->values_num] = strdup (value);
 917         if (rc->values[rc->values_num] != NULL)
 918                 rc->values_num++;
 919
 920         if (rc->values_num == 1)
 921                 rc->first_value = value_time;
 922         rc->last_value = value_time;
 923
 924         /* Insert if this is the first value */
 925         if (new_rc == 1)
 926         {
 927                 void *cache_key = strdup (filename);
 928
 929                 if (cache_key == NULL)
 930                 {
 931                         char errbuf[1024];
 932                         sstrerror (errno, errbuf, sizeof (errbuf));
 933
 934                         pthread_mutex_unlock (&cache_lock);
 935
 936                         ERROR ("rrdtool plugin: strdup failed: %s", errbuf);
 937
 938                         sfree (rc->values[0]);
 939                         sfree (rc->values);
 940                         sfree (rc);
 941                         return (-1);
 942                 }
 943
 944                 c_avl_insert (cache, cache_key, rc);
 945         }
 946
 947         DEBUG ("rrdtool plugin: rrd_cache_insert: file = %s; "
 948                         "values_num = %i; age = %lu;",
 949                         filename, rc->values_num,
 950                         (unsigned long)(rc->last_value - rc->first_value));
 951
 952         if ((rc->last_value - rc->first_value) >= cache_timeout)
 953         {
 954                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
 955                  * the same time, ALWAYS lock `cache_lock' first! */
 956                 if (rc->flags != FLAG_QUEUED)
 957                 {
 958                         if (rrd_queue_cache_entry (filename, QUEUE_INSERT_BACK) == 0)
 959                                 rc->flags = FLAG_QUEUED;
 960                 }
 961                 else
 962                 {
 963                         DEBUG ("rrdtool plugin: `%s' is already queued.", filename);
 964                 }
 965         }
 966
 967         if ((cache_timeout > 0) &&
 968                         ((time (NULL) - cache_flush_last) > cache_flush_timeout))
 969                 rrd_cache_flush (cache_flush_timeout);
 970
 971
 972         pthread_mutex_unlock (&cache_lock);
 973
 974         return (0);
 975 } /* int rrd_cache_insert */
 976
 977 static int rrd_compare_numeric (const void *a_ptr, const void *b_ptr)
 978 {
 979         int a = *((int *) a_ptr);
 980         int b = *((int *) b_ptr);
 981
 982         if (a < b)
 983                 return (-1);
 984         else if (a > b)
 985                 return (1);
 986         else
 987                 return (0);
 988 } /* int rrd_compare_numeric */
 989
 990 static int rrd_write (const data_set_t *ds, const value_list_t *vl)
 991 {
 992         struct stat  statbuf;
 993         char         filename[512];
 994         char         values[512];
 995         int          status;
 996
 997         if (0 != strcmp (ds->type, vl->type)) {
 998                 ERROR ("rrdtool plugin: DS type does not match value list type");
 999                 return -1;
1000         }
1001
1002         if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
1003                 return (-1);
1004
1005         if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
1006                 return (-1);
1007
1008         if (stat (filename, &statbuf) == -1)
1009         {
1010                 if (errno == ENOENT)
1011                 {
1012                         if (rrd_create_file (filename, ds, vl))
1013                                 return (-1);
1014                 }
1015                 else
1016                 {
1017                         char errbuf[1024];
1018                         ERROR ("stat(%s) failed: %s", filename,
1019                                         sstrerror (errno, errbuf,
1020                                                 sizeof (errbuf)));
1021                         return (-1);
1022                 }
1023         }
1024         else if (!S_ISREG (statbuf.st_mode))
1025         {
1026                 ERROR ("stat(%s): Not a regular file!",
1027                                 filename);
1028                 return (-1);
1029         }
1030
1031         status = rrd_cache_insert (filename, values, vl->time);
1032
1033         return (status);
1034 } /* int rrd_write */
1035
1036 static int rrd_flush (int timeout, const char *identifier)
1037 {
1038         pthread_mutex_lock (&cache_lock);
1039
1040         if (cache == NULL) {
1041                 pthread_mutex_unlock (&cache_lock);
1042                 return (0);
1043         }
1044
1045         rrd_cache_flush_identifier (timeout, identifier);
1046
1047         pthread_mutex_unlock (&cache_lock);
1048         return (0);
1049 } /* int rrd_flush */
1050
1051 static int rrd_config (const char *key, const char *value)
1052 {
1053         if (strcasecmp ("CacheTimeout", key) == 0)
1054         {
1055                 int tmp = atoi (value);
1056                 if (tmp < 0)
1057                 {
1058                         fprintf (stderr, "rrdtool: `CacheTimeout' must "
1059                                         "be greater than 0.\n");
1060                         return (1);
1061                 }
1062                 cache_timeout = tmp;
1063         }
1064         else if (strcasecmp ("CacheFlush", key) == 0)
1065         {
1066                 int tmp = atoi (value);
1067                 if (tmp < 0)
1068                 {
1069                         fprintf (stderr, "rrdtool: `CacheFlush' must "
1070                                         "be greater than 0.\n");
1071                         return (1);
1072                 }
1073                 cache_flush_timeout = tmp;
1074         }
1075         else if (strcasecmp ("DataDir", key) == 0)
1076         {
1077                 if (datadir != NULL)
1078                         free (datadir);
1079                 datadir = strdup (value);
1080                 if (datadir != NULL)
1081                 {
1082                         int len = strlen (datadir);
1083                         while ((len > 0) && (datadir[len - 1] == '/'))
1084                         {
1085                                 len--;
1086                                 datadir[len] = '\0';
1087                         }
1088                         if (len <= 0)
1089                         {
1090                                 free (datadir);
1091                                 datadir = NULL;
1092                         }
1093                 }
1094         }
1095         else if (strcasecmp ("StepSize", key) == 0)
1096         {
1097                 stepsize = atoi (value);
1098                 if (stepsize < 0)
1099                         stepsize = 0;
1100         }
1101         else if (strcasecmp ("HeartBeat", key) == 0)
1102         {
1103                 heartbeat = atoi (value);
1104                 if (heartbeat < 0)
1105                         heartbeat = 0;
1106         }
1107         else if (strcasecmp ("RRARows", key) == 0)
1108         {
1109                 int tmp = atoi (value);
1110                 if (tmp <= 0)
1111                 {
1112                         fprintf (stderr, "rrdtool: `RRARows' must "
1113                                         "be greater than 0.\n");
1114                         return (1);
1115                 }
1116                 rrarows = tmp;
1117         }
1118         else if (strcasecmp ("RRATimespan", key) == 0)
1119         {
1120                 char *saveptr = NULL;
1121                 char *dummy;
1122                 char *ptr;
1123                 char *value_copy;
1124                 int *tmp_alloc;
1125
1126                 value_copy = strdup (value);
1127                 if (value_copy == NULL)
1128                         return (1);
1129
1130                 dummy = value_copy;
1131                 while ((ptr = strtok_r (dummy, ", \t", &saveptr)) != NULL)
1132                 {
1133                         dummy = NULL;
1134
1135                         tmp_alloc = realloc (rra_timespans_custom,
1136                                         sizeof (int) * (rra_timespans_custom_num + 1));
1137                         if (tmp_alloc == NULL)
1138                         {
1139                                 fprintf (stderr, "rrdtool: realloc failed.\n");
1140                                 free (value_copy);
1141                                 return (1);
1142                         }
1143                         rra_timespans_custom = tmp_alloc;
1144                         rra_timespans_custom[rra_timespans_custom_num] = atoi (ptr);
1145                         if (rra_timespans_custom[rra_timespans_custom_num] != 0)
1146                                 rra_timespans_custom_num++;
1147                 } /* while (strtok_r) */
1148
1149                 qsort (/* base = */ rra_timespans_custom,
1150                                 /* nmemb  = */ rra_timespans_custom_num,
1151                                 /* size   = */ sizeof (rra_timespans_custom[0]),
1152                                 /* compar = */ rrd_compare_numeric);
1153
1154                 free (value_copy);
1155         }
1156         else if (strcasecmp ("XFF", key) == 0)
1157         {
1158                 double tmp = atof (value);
1159                 if ((tmp < 0.0) || (tmp >= 1.0))
1160                 {
1161                         fprintf (stderr, "rrdtool: `XFF' must "
1162                                         "be in the range 0 to 1 (exclusive).");
1163                         return (1);
1164                 }
1165                 xff = tmp;
1166         }
1167         else
1168         {
1169                 return (-1);
1170         }
1171         return (0);
1172 } /* int rrd_config */
1173
1174 static int rrd_shutdown (void)
1175 {
1176         pthread_mutex_lock (&cache_lock);
1177         rrd_cache_flush (-1);
1178         pthread_mutex_unlock (&cache_lock);
1179
1180         pthread_mutex_lock (&queue_lock);
1181         do_shutdown = 1;
1182         pthread_cond_signal (&queue_cond);
1183         pthread_mutex_unlock (&queue_lock);
1184
1185         /* Wait for all the values to be written to disk before returning. */
1186         if (queue_thread != 0)
1187         {
1188                 pthread_join (queue_thread, NULL);
1189                 queue_thread = 0;
1190                 DEBUG ("rrdtool plugin: queue_thread exited.");
1191         }
1192
1193         return (0);
1194 } /* int rrd_shutdown */
1195
1196 static int rrd_init (void)
1197 {
1198         int status;
1199
1200         if (stepsize < 0)
1201                 stepsize = 0;
1202         if (heartbeat <= 0)
1203                 heartbeat = 2 * stepsize;
1204
1205         if ((heartbeat > 0) && (heartbeat < interval_g))
1206                 WARNING ("rrdtool plugin: Your `heartbeat' is "
1207                                 "smaller than your `interval'. This will "
1208                                 "likely cause problems.");
1209         else if ((stepsize > 0) && (stepsize < interval_g))
1210                 WARNING ("rrdtool plugin: Your `stepsize' is "
1211                                 "smaller than your `interval'. This will "
1212                                 "create needlessly big RRD-files.");
1213
1214         /* Set the cache up */
1215         pthread_mutex_lock (&cache_lock);
1216
1217         cache = c_avl_create ((int (*) (const void *, const void *)) strcmp);
1218         if (cache == NULL)
1219         {
1220                 ERROR ("rrdtool plugin: c_avl_create failed.");
1221                 return (-1);
1222         }
1223
1224         cache_flush_last = time (NULL);
1225         if (cache_timeout < 2)
1226         {
1227                 cache_timeout = 0;
1228                 cache_flush_timeout = 0;
1229         }
1230         else if (cache_flush_timeout < cache_timeout)
1231                 cache_flush_timeout = 10 * cache_timeout;
1232
1233         pthread_mutex_unlock (&cache_lock);
1234
1235         status = pthread_create (&queue_thread, NULL, rrd_queue_thread, NULL);
1236         if (status != 0)
1237         {
1238                 ERROR ("rrdtool plugin: Cannot create queue-thread.");
1239                 return (-1);
1240         }
1241
1242         DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %i;"
1243                         " heartbeat = %i; rrarows = %i; xff = %lf;",
1244                         (datadir == NULL) ? "(null)" : datadir,
1245                         stepsize, heartbeat, rrarows, xff);
1246
1247         return (0);
1248 } /* int rrd_init */
1249
1250 void module_register (void)
1251 {
1252         plugin_register_config ("rrdtool", rrd_config,
1253                         config_keys, config_keys_num);
1254         plugin_register_init ("rrdtool", rrd_init);
1255         plugin_register_write ("rrdtool", rrd_write);
1256         plugin_register_flush ("rrdtool", rrd_flush);
1257         plugin_register_shutdown ("rrdtool", rrd_shutdown);
1258 }