Merge branch 'collectd-4.0' into collectd-4.1
[collectd.git] / src / plugin.c
1 /**
2  * collectd - src/plugin.c
3  * Copyright (C) 2005,2006  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
24 #include <ltdl.h>
25
26 #if HAVE_PTHREAD_H
27 # include <pthread.h>
28 #endif
29
30 #include "common.h"
31 #include "plugin.h"
32 #include "configfile.h"
33 #include "utils_llist.h"
34
35 /*
36  * Private structures
37  */
38 struct read_func_s
39 {
40         int wait_time;
41         int wait_left;
42         int (*callback) (void);
43         enum { DONE = 0, TODO = 1, ACTIVE = 2 } needs_read;
44 };
45 typedef struct read_func_s read_func_t;
46
47 /*
48  * Private variables
49  */
50 static llist_t *list_init;
51 static llist_t *list_read;
52 static llist_t *list_write;
53 static llist_t *list_shutdown;
54 static llist_t *list_data_set;
55 static llist_t *list_log;
56
57 static char *plugindir = NULL;
58
59 static int             read_loop = 1;
60 static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER;
61 static pthread_cond_t  read_cond = PTHREAD_COND_INITIALIZER;
62 static pthread_t      *read_threads = NULL;
63 static int             read_threads_num = 0;
64
65 /*
66  * Static functions
67  */
68 static const char *plugin_get_dir (void)
69 {
70         if (plugindir == NULL)
71                 return (PLUGINDIR);
72         else
73                 return (plugindir);
74 }
75
76 static int register_callback (llist_t **list, const char *name, void *callback)
77 {
78         llentry_t *le;
79
80         if ((*list == NULL)
81                         && ((*list = llist_create ()) == NULL))
82                 return (-1);
83
84         le = llist_search (*list, name);
85         if (le == NULL)
86         {
87                 le = llentry_create (name, callback);
88                 if (le == NULL)
89                         return (-1);
90
91                 llist_append (*list, le);
92         }
93         else
94         {
95                 le->value = callback;
96         }
97
98         return (0);
99 } /* int register_callback */
100
101 static int plugin_unregister (llist_t *list, const char *name)
102 {
103         llentry_t *e;
104
105         e = llist_search (list, name);
106
107         if (e == NULL)
108                 return (-1);
109
110         llist_remove (list, e);
111         llentry_destroy (e);
112
113         return (0);
114 } /* int plugin_unregister */
115
116 /*
117  * (Try to) load the shared object `file'. Won't complain if it isn't a shared
118  * object, but it will bitch about a shared object not having a
119  * ``module_register'' symbol..
120  */
121 static int plugin_load_file (char *file)
122 {
123         lt_dlhandle dlh;
124         void (*reg_handle) (void);
125
126         DEBUG ("file = %s", file);
127
128         lt_dlinit ();
129         lt_dlerror (); /* clear errors */
130
131         if ((dlh = lt_dlopen (file)) == NULL)
132         {
133                 const char *error = lt_dlerror ();
134
135                 ERROR ("lt_dlopen failed: %s", error);
136                 return (1);
137         }
138
139         if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL)
140         {
141                 WARNING ("Couldn't find symbol ``module_register'' in ``%s'': %s\n",
142                                 file, lt_dlerror ());
143                 lt_dlclose (dlh);
144                 return (-1);
145         }
146
147         (*reg_handle) ();
148
149         return (0);
150 }
151
152 static void *plugin_read_thread (void *args)
153 {
154         llentry_t   *le;
155         read_func_t *rf;
156         int          status;
157         int          done;
158
159         pthread_mutex_lock (&read_lock);
160
161         while (read_loop != 0)
162         {
163                 le = llist_head (list_read);
164                 done = 0;
165
166                 while ((read_loop != 0) && (le != NULL))
167                 {
168                         rf = (read_func_t *) le->value;
169
170                         if (rf->needs_read != TODO)
171                         {
172                                 le = le->next;
173                                 continue;
174                         }
175
176                         /* We will do this read function */
177                         rf->needs_read = ACTIVE;
178
179                         DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Handling %s",
180                                         (unsigned long int) pthread_self (), le->key);
181                         pthread_mutex_unlock (&read_lock);
182
183                         status = rf->callback ();
184                         done++;
185
186                         if (status != 0)
187                         {
188                                 if (rf->wait_time < interval_g)
189                                         rf->wait_time = interval_g;
190                                 rf->wait_left = rf->wait_time;
191                                 rf->wait_time = rf->wait_time * 2;
192                                 if (rf->wait_time > 86400)
193                                         rf->wait_time = 86400;
194
195                                 NOTICE ("read-function of plugin `%s' "
196                                                 "failed. Will suspend it for %i "
197                                                 "seconds.", le->key, rf->wait_left);
198                         }
199                         else
200                         {
201                                 rf->wait_left = 0;
202                                 rf->wait_time = interval_g;
203                         }
204
205                         pthread_mutex_lock (&read_lock);
206
207                         rf->needs_read = DONE;
208                         le = le->next;
209                 } /* while (le != NULL) */
210
211                 if ((read_loop != 0) && (done == 0))
212                 {
213                         DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Waiting on read_cond.",
214                                         (unsigned long int) pthread_self ());
215                         pthread_cond_wait (&read_cond, &read_lock);
216                 }
217         } /* while (read_loop) */
218
219         pthread_mutex_unlock (&read_lock);
220
221         pthread_exit (NULL);
222 } /* void *plugin_read_thread */
223
224 static void start_threads (int num)
225 {
226         int i;
227
228         if (read_threads != NULL)
229                 return;
230
231         read_threads = (pthread_t *) calloc (num, sizeof (pthread_t));
232         if (read_threads == NULL)
233         {
234                 ERROR ("plugin: start_threads: calloc failed.");
235                 return;
236         }
237
238         read_threads_num = 0;
239         for (i = 0; i < num; i++)
240         {
241                 if (pthread_create (read_threads + read_threads_num, NULL,
242                                         plugin_read_thread, NULL) == 0)
243                 {
244                         read_threads_num++;
245                 }
246                 else
247                 {
248                         ERROR ("plugin: start_threads: pthread_create failed.");
249                         return;
250                 }
251         } /* for (i) */
252 } /* void start_threads */
253
254 static void stop_threads (void)
255 {
256         int i;
257
258         pthread_mutex_lock (&read_lock);
259         read_loop = 0;
260         DEBUG ("plugin: stop_threads: Signalling `read_cond'");
261         pthread_cond_broadcast (&read_cond);
262         pthread_mutex_unlock (&read_lock);
263
264         for (i = 0; i < read_threads_num; i++)
265         {
266                 if (pthread_join (read_threads[i], NULL) != 0)
267                 {
268                         ERROR ("plugin: stop_threads: pthread_join failed.");
269                 }
270                 read_threads[i] = (pthread_t) 0;
271         }
272         sfree (read_threads);
273         read_threads_num = 0;
274 } /* void stop_threads */
275
276 /*
277  * Public functions
278  */
279 void plugin_set_dir (const char *dir)
280 {
281         if (plugindir != NULL)
282                 free (plugindir);
283
284         if (dir == NULL)
285                 plugindir = NULL;
286         else if ((plugindir = strdup (dir)) == NULL)
287         {
288                 char errbuf[1024];
289                 ERROR ("strdup failed: %s",
290                                 sstrerror (errno, errbuf, sizeof (errbuf)));
291         }
292 }
293
294 #define BUFSIZE 512
295 int plugin_load (const char *type)
296 {
297         DIR  *dh;
298         const char *dir;
299         char  filename[BUFSIZE];
300         char  typename[BUFSIZE];
301         int   typename_len;
302         int   ret;
303         struct stat    statbuf;
304         struct dirent *de;
305
306         DEBUG ("type = %s", type);
307
308         dir = plugin_get_dir ();
309         ret = 1;
310
311         /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
312          * type when matching the filename */
313         if (snprintf (typename, BUFSIZE, "%s.so", type) >= BUFSIZE)
314         {
315                 WARNING ("snprintf: truncated: `%s.so'", type);
316                 return (-1);
317         }
318         typename_len = strlen (typename);
319
320         if ((dh = opendir (dir)) == NULL)
321         {
322                 char errbuf[1024];
323                 ERROR ("opendir (%s): %s", dir,
324                                 sstrerror (errno, errbuf, sizeof (errbuf)));
325                 return (-1);
326         }
327
328         while ((de = readdir (dh)) != NULL)
329         {
330                 if (strncasecmp (de->d_name, typename, typename_len))
331                         continue;
332
333                 if (snprintf (filename, BUFSIZE, "%s/%s", dir, de->d_name) >= BUFSIZE)
334                 {
335                         WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name);
336                         continue;
337                 }
338
339                 if (lstat (filename, &statbuf) == -1)
340                 {
341                         char errbuf[1024];
342                         WARNING ("stat %s: %s", filename,
343                                         sstrerror (errno, errbuf, sizeof (errbuf)));
344                         continue;
345                 }
346                 else if (!S_ISREG (statbuf.st_mode))
347                 {
348                         /* don't follow symlinks */
349                         continue;
350                 }
351
352                 if (plugin_load_file (filename) == 0)
353                 {
354                         /* success */
355                         ret = 0;
356                         break;
357                 }
358         }
359
360         closedir (dh);
361
362         return (ret);
363 }
364
365 /*
366  * The `register_*' functions follow
367  */
368 int plugin_register_config (const char *name,
369                 int (*callback) (const char *key, const char *val),
370                 const char **keys, int keys_num)
371 {
372         cf_register (name, callback, keys, keys_num);
373         return (0);
374 } /* int plugin_register_config */
375
376 int plugin_register_complex_config (const char *type,
377                 int (*callback) (oconfig_item_t *))
378 {
379         return (cf_register_complex (type, callback));
380 } /* int plugin_register_complex_config */
381
382 int plugin_register_init (const char *name,
383                 int (*callback) (void))
384 {
385         return (register_callback (&list_init, name, (void *) callback));
386 } /* plugin_register_init */
387
388 int plugin_register_read (const char *name,
389                 int (*callback) (void))
390 {
391         read_func_t *rf;
392
393         rf = (read_func_t *) malloc (sizeof (read_func_t));
394         if (rf == NULL)
395         {
396                 char errbuf[1024];
397                 ERROR ("plugin_register_read: malloc failed: %s",
398                                 sstrerror (errno, errbuf, sizeof (errbuf)));
399                 return (-1);
400         }
401
402         memset (rf, '\0', sizeof (read_func_t));
403         rf->wait_time = interval_g;
404         rf->wait_left = 0;
405         rf->callback = callback;
406         rf->needs_read = DONE;
407
408         return (register_callback (&list_read, name, (void *) rf));
409 } /* int plugin_register_read */
410
411 int plugin_register_write (const char *name,
412                 int (*callback) (const data_set_t *ds, const value_list_t *vl))
413 {
414         return (register_callback (&list_write, name, (void *) callback));
415 } /* int plugin_register_write */
416
417 int plugin_register_shutdown (char *name,
418                 int (*callback) (void))
419 {
420         return (register_callback (&list_shutdown, name, (void *) callback));
421 } /* int plugin_register_shutdown */
422
423 int plugin_register_data_set (const data_set_t *ds)
424 {
425         data_set_t *ds_copy;
426         int i;
427
428         if ((list_data_set != NULL)
429                         && (llist_search (list_data_set, ds->type) != NULL))
430         {
431                 NOTICE ("Replacing DS `%s' with another version.", ds->type);
432                 plugin_unregister_data_set (ds->type);
433         }
434
435         ds_copy = (data_set_t *) malloc (sizeof (data_set_t));
436         if (ds_copy == NULL)
437                 return (-1);
438         memcpy(ds_copy, ds, sizeof (data_set_t));
439
440         ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t)
441                         * ds->ds_num);
442         if (ds_copy->ds == NULL)
443         {
444                 free (ds_copy);
445                 return (-1);
446         }
447
448         for (i = 0; i < ds->ds_num; i++)
449                 memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t));
450
451         return (register_callback (&list_data_set, ds->type, (void *) ds_copy));
452 } /* int plugin_register_data_set */
453
454 int plugin_register_log (char *name,
455                 void (*callback) (int priority, const char *msg))
456 {
457         return (register_callback (&list_log, name, (void *) callback));
458 } /* int plugin_register_log */
459
460 int plugin_unregister_config (const char *name)
461 {
462         cf_unregister (name);
463         return (0);
464 } /* int plugin_unregister_config */
465
466 int plugin_unregister_complex_config (const char *name)
467 {
468         cf_unregister_complex (name);
469         return (0);
470 } /* int plugin_unregister_complex_config */
471
472 int plugin_unregister_init (const char *name)
473 {
474         return (plugin_unregister (list_init, name));
475 }
476
477 int plugin_unregister_read (const char *name)
478 {
479         llentry_t *e;
480
481         e = llist_search (list_read, name);
482
483         if (e == NULL)
484                 return (-1);
485
486         llist_remove (list_read, e);
487         free (e->value);
488         llentry_destroy (e);
489
490         return (0);
491 }
492
493 int plugin_unregister_write (const char *name)
494 {
495         return (plugin_unregister (list_write, name));
496 }
497
498 int plugin_unregister_shutdown (const char *name)
499 {
500         return (plugin_unregister (list_shutdown, name));
501 }
502
503 int plugin_unregister_data_set (const char *name)
504 {
505         llentry_t  *e;
506         data_set_t *ds;
507
508         if (list_data_set == NULL)
509                 return (-1);
510
511         e = llist_search (list_data_set, name);
512
513         if (e == NULL)
514                 return (-1);
515
516         llist_remove (list_data_set, e);
517         ds = (data_set_t *) e->value;
518         llentry_destroy (e);
519
520         sfree (ds->ds);
521         sfree (ds);
522
523         return (0);
524 } /* int plugin_unregister_data_set */
525
526 int plugin_unregister_log (const char *name)
527 {
528         return (plugin_unregister (list_log, name));
529 }
530
531 void plugin_init_all (void)
532 {
533         int (*callback) (void);
534         llentry_t *le;
535         int status;
536
537         /* Start read-threads */
538         if (list_read != NULL)
539         {
540                 const char *rt;
541                 int num;
542                 rt = global_option_get ("ReadThreads");
543                 num = atoi (rt);
544                 start_threads ((num > 0) ? num : 5);
545         }
546
547         if (list_init == NULL)
548                 return;
549
550         le = llist_head (list_init);
551         while (le != NULL)
552         {
553                 callback = (int (*) (void)) le->value;
554                 status = (*callback) ();
555
556                 if (status != 0)
557                 {
558                         ERROR ("Initialization of plugin `%s' "
559                                         "failed with status %i. "
560                                         "Plugin will be unloaded.",
561                                         le->key, status);
562                         /* FIXME: Unload _all_ functions */
563                         plugin_unregister_read (le->key);
564                 }
565
566                 le = le->next;
567         }
568 } /* void plugin_init_all */
569
570 void plugin_read_all (const int *loop)
571 {
572         llentry_t   *le;
573         read_func_t *rf;
574
575         if (list_read == NULL)
576                 return;
577
578         pthread_mutex_lock (&read_lock);
579
580         le = llist_head (list_read);
581         while (le != NULL)
582         {
583                 rf = (read_func_t *) le->value;
584
585                 if (rf->needs_read != DONE)
586                 {
587                         le = le->next;
588                         continue;
589                 }
590
591                 if (rf->wait_left > 0)
592                         rf->wait_left -= interval_g;
593
594                 if (rf->wait_left <= 0)
595                 {
596                         rf->needs_read = TODO;
597                 }
598
599                 le = le->next;
600         }
601
602         DEBUG ("plugin: plugin_read_all: Signalling `read_cond'");
603         pthread_cond_broadcast (&read_cond);
604         pthread_mutex_unlock (&read_lock);
605 } /* void plugin_read_all */
606
607 void plugin_shutdown_all (void)
608 {
609         int (*callback) (void);
610         llentry_t *le;
611
612         stop_threads ();
613
614         if (list_shutdown == NULL)
615                 return;
616
617         le = llist_head (list_shutdown);
618         while (le != NULL)
619         {
620                 callback = (int (*) (void)) le->value;
621
622                 /* Advance the pointer before calling the callback allows
623                  * shutdown functions to unregister themselves. If done the
624                  * other way around the memory `le' points to will be freed
625                  * after callback returns. */
626                 le = le->next;
627
628                 (*callback) ();
629         }
630 } /* void plugin_shutdown_all */
631
632 int plugin_dispatch_values (const char *name, value_list_t *vl)
633 {
634         int (*callback) (const data_set_t *, const value_list_t *);
635         data_set_t *ds;
636         llentry_t *le;
637
638         if ((list_write == NULL) || (list_data_set == NULL))
639                 return (-1);
640
641         le = llist_search (list_data_set, name);
642         if (le == NULL)
643         {
644                 DEBUG ("No such dataset registered: %s", name);
645                 return (-1);
646         }
647
648         ds = (data_set_t *) le->value;
649
650         DEBUG ("plugin: plugin_dispatch_values: time = %u; interval = %i; "
651                         "host = %s; "
652                         "plugin = %s; plugin_instance = %s; "
653                         "type = %s; type_instance = %s;",
654                         (unsigned int) vl->time, vl->interval,
655                         vl->host,
656                         vl->plugin, vl->plugin_instance,
657                         ds->type, vl->type_instance);
658
659 #if COLLECT_DEBUG
660         assert (ds->ds_num == vl->values_len);
661 #else
662         if (ds->ds_num != vl->values_len)
663         {
664                 ERROR ("plugin: ds->type = %s: (ds->ds_num = %i) != "
665                                 "(vl->values_len = %i)",
666                                 ds->type, ds->ds_num, vl->values_len);
667                 return (-1);
668         }
669 #endif
670
671         escape_slashes (vl->host, sizeof (vl->host));
672         escape_slashes (vl->plugin, sizeof (vl->plugin));
673         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
674         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
675
676         le = llist_head (list_write);
677         while (le != NULL)
678         {
679                 callback = (int (*) (const data_set_t *, const value_list_t *)) le->value;
680                 (*callback) (ds, vl);
681
682                 le = le->next;
683         }
684
685         return (0);
686 } /* int plugin_dispatch_values */
687
688 void plugin_log (int level, const char *format, ...)
689 {
690         char msg[512];
691         va_list ap;
692
693         void (*callback) (int, const char *);
694         llentry_t *le;
695
696         if (list_log == NULL)
697                 return;
698
699 #if !COLLECT_DEBUG
700         if (level >= LOG_DEBUG)
701                 return;
702 #endif
703
704         va_start (ap, format);
705         vsnprintf (msg, 512, format, ap);
706         msg[511] = '\0';
707         va_end (ap);
708
709         le = llist_head (list_log);
710         while (le != NULL)
711         {
712                 callback = (void (*) (int, const char *)) le->value;
713                 (*callback) (level, msg);
714
715                 le = le->next;
716         }
717 } /* void plugin_log */
718
719 void plugin_complain (int level, complain_t *c, const char *format, ...)
720 {
721         char message[512];
722         va_list ap;
723
724         if (c->delay > 0)
725         {
726                 c->delay--;
727                 return;
728         }
729
730         if (c->interval < interval_g)
731                 c->interval = interval_g;
732         else
733                 c->interval *= 2;
734
735         if (c->interval > 86400)
736                 c->interval = 86400;
737
738         c->delay = c->interval / interval_g;
739
740         va_start (ap, format);
741         vsnprintf (message, 512, format, ap);
742         message[511] = '\0';
743         va_end (ap);
744
745         plugin_log (level, message);
746 }
747
748 void plugin_relief (int level, complain_t *c, const char *format, ...)
749 {
750         char message[512];
751         va_list ap;
752
753         if (c->interval == 0)
754                 return;
755
756         c->interval = 0;
757
758         va_start (ap, format);
759         vsnprintf (message, 512, format, ap);
760         message[511] = '\0';
761         va_end (ap);
762
763         plugin_log (level, message);
764 }
765
766 const data_set_t *plugin_get_ds (const char *name)
767 {
768         data_set_t *ds;
769         llentry_t *le;
770
771         le = llist_search (list_data_set, name);
772         if (le == NULL)
773         {
774                 DEBUG ("No such dataset registered: %s", name);
775                 return (NULL);
776         }
777
778         ds = (data_set_t *) le->value;
779
780         return (ds);
781 } /* data_set_t *plugin_get_ds */