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