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