src/plugin.c

   1 /**
   2  * collectd - src/plugin.c
   3  * Copyright (C) 2005-2009  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  *   Sebastian Harl <sh at tokkee.org>
  21  **/
  22
  23 #include "collectd.h"
  24 #include "utils_complain.h"
  25
  26 #include <ltdl.h>
  27
  28 #if HAVE_PTHREAD_H
  29 # include <pthread.h>
  30 #endif
  31
  32 #include "common.h"
  33 #include "plugin.h"
  34 #include "configfile.h"
  35 #include "utils_avltree.h"
  36 #include "utils_llist.h"
  37 #include "utils_cache.h"
  38 #include "utils_threshold.h"
  39 #include "filter_chain.h"
  40
  41 /*
  42  * Private structures
  43  */
  44 #define RF_SIMPLE  0
  45 #define RF_COMPLEX 1
  46 struct read_func_s
  47 {
  48         int wait_time;
  49         int wait_left;
  50         int type;
  51         union
  52         {
  53                 int (*cb_simple) (void);
  54                 plugin_read_cb cb_complex;
  55         } callback;
  56         enum { DONE = 0, TODO = 1, ACTIVE = 2 } needs_read;
  57         user_data_t udata;
  58 };
  59 typedef struct read_func_s read_func_t;
  60
  61 struct write_func_s
  62 {
  63         plugin_write_cb callback;
  64         user_data_t udata;
  65 };
  66 typedef struct write_func_s write_func_t;
  67
  68 struct flush_func_s
  69 {
  70         plugin_flush_cb callback;
  71         user_data_t udata;
  72 };
  73 typedef struct flush_func_s flush_func_t;
  74
  75 struct log_func_s
  76 {
  77         plugin_log_cb callback;
  78         user_data_t udata;
  79 };
  80 typedef struct log_func_s log_func_t;
  81
  82 /*
  83  * Private variables
  84  */
  85 static llist_t *list_init;
  86 static llist_t *list_read;
  87 static llist_t *list_write;
  88 static llist_t *list_flush;
  89 static llist_t *list_shutdown;
  90 static llist_t *list_log;
  91 static llist_t *list_notification;
  92
  93 static fc_chain_t *pre_cache_chain = NULL;
  94 static fc_chain_t *post_cache_chain = NULL;
  95
  96 static c_avl_tree_t *data_sets;
  97
  98 static char *plugindir = NULL;
  99
 100 static int             read_loop = 1;
 101 static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER;
 102 static pthread_cond_t  read_cond = PTHREAD_COND_INITIALIZER;
 103 static pthread_t      *read_threads = NULL;
 104 static int             read_threads_num = 0;
 105
 106 /*
 107  * Static functions
 108  */
 109 static const char *plugin_get_dir (void)
 110 {
 111         if (plugindir == NULL)
 112                 return (PLUGINDIR);
 113         else
 114                 return (plugindir);
 115 }
 116
 117 static int register_callback (llist_t **list, const char *name, void *callback)
 118 {
 119         llentry_t *le;
 120         char *key;
 121
 122         if ((*list == NULL)
 123                         && ((*list = llist_create ()) == NULL))
 124                 return (-1);
 125
 126         le = llist_search (*list, name);
 127         if (le == NULL)
 128         {
 129                 key = strdup (name);
 130                 if (key == NULL)
 131                         return (-1);
 132
 133                 le = llentry_create (key, callback);
 134                 if (le == NULL)
 135                 {
 136                         free (key);
 137                         return (-1);
 138                 }
 139
 140                 llist_append (*list, le);
 141         }
 142         else
 143         {
 144                 le->value = callback;
 145         }
 146
 147         return (0);
 148 } /* int register_callback */
 149
 150 static void plugin_user_data_destroy (user_data_t *ud)
 151 {
 152         if ((ud != NULL) && (ud->data != NULL) && (ud->free_func != NULL))
 153         {
 154                 ud->free_func (ud->data);
 155                 ud->data = NULL;
 156                 ud->free_func = NULL;
 157         }
 158 } /* void plugin_user_data_destroy */
 159
 160 static int plugin_unregister (llist_t *list, const char *name)
 161 {
 162         llentry_t *e;
 163
 164         e = llist_search (list, name);
 165
 166         if (e == NULL)
 167                 return (-1);
 168
 169         llist_remove (list, e);
 170         free (e->key);
 171         llentry_destroy (e);
 172
 173         return (0);
 174 } /* int plugin_unregister */
 175
 176 /*
 177  * (Try to) load the shared object `file'. Won't complain if it isn't a shared
 178  * object, but it will bitch about a shared object not having a
 179  * ``module_register'' symbol..
 180  */
 181 static int plugin_load_file (char *file)
 182 {
 183         lt_dlhandle dlh;
 184         void (*reg_handle) (void);
 185
 186         DEBUG ("file = %s", file);
 187
 188         lt_dlinit ();
 189         lt_dlerror (); /* clear errors */
 190
 191         if ((dlh = lt_dlopen (file)) == NULL)
 192         {
 193                 const char *error = lt_dlerror ();
 194
 195                 ERROR ("lt_dlopen (%s) failed: %s", file, error);
 196                 fprintf (stderr, "lt_dlopen (%s) failed: %s\n", file, error);
 197                 return (1);
 198         }
 199
 200         if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL)
 201         {
 202                 WARNING ("Couldn't find symbol `module_register' in `%s': %s\n",
 203                                 file, lt_dlerror ());
 204                 lt_dlclose (dlh);
 205                 return (-1);
 206         }
 207
 208         (*reg_handle) ();
 209
 210         return (0);
 211 }
 212
 213 static void *plugin_read_thread (void __attribute__((unused)) *args)
 214 {
 215         llentry_t   *le;
 216         read_func_t *rf;
 217         int          status;
 218         int          done;
 219
 220         pthread_mutex_lock (&read_lock);
 221
 222         while (read_loop != 0)
 223         {
 224                 le = llist_head (list_read);
 225                 done = 0;
 226
 227                 while ((read_loop != 0) && (le != NULL))
 228                 {
 229                         rf = (read_func_t *) le->value;
 230
 231                         if (rf->needs_read != TODO)
 232                         {
 233                                 le = le->next;
 234                                 continue;
 235                         }
 236
 237                         /* We will do this read function */
 238                         rf->needs_read = ACTIVE;
 239
 240                         DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Handling %s",
 241                                         (unsigned long int) pthread_self (), le->key);
 242                         pthread_mutex_unlock (&read_lock);
 243
 244                         if (rf->type == RF_SIMPLE)
 245                         {
 246                                 status = rf->callback.cb_simple ();
 247                         }
 248                         else
 249                         {
 250                                 assert (rf->type == RF_COMPLEX);
 251                                 status = rf->callback.cb_complex (&rf->udata);
 252                         }
 253                         done++;
 254
 255                         if (status != 0)
 256                         {
 257                                 if (rf->wait_time < interval_g)
 258                                         rf->wait_time = interval_g;
 259                                 rf->wait_left = rf->wait_time;
 260                                 rf->wait_time = rf->wait_time * 2;
 261                                 if (rf->wait_time > 86400)
 262                                         rf->wait_time = 86400;
 263
 264                                 NOTICE ("read-function of plugin `%s' "
 265                                                 "failed. Will suspend it for %i "
 266                                                 "seconds.", le->key, rf->wait_left);
 267                         }
 268                         else
 269                         {
 270                                 rf->wait_left = 0;
 271                                 rf->wait_time = interval_g;
 272                         }
 273
 274                         pthread_mutex_lock (&read_lock);
 275
 276                         rf->needs_read = DONE;
 277                         le = le->next;
 278                 } /* while (le != NULL) */
 279
 280                 if ((read_loop != 0) && (done == 0))
 281                 {
 282                         DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Waiting on read_cond.",
 283                                         (unsigned long int) pthread_self ());
 284                         pthread_cond_wait (&read_cond, &read_lock);
 285                 }
 286         } /* while (read_loop) */
 287
 288         pthread_mutex_unlock (&read_lock);
 289
 290         pthread_exit (NULL);
 291         return ((void *) 0);
 292 } /* void *plugin_read_thread */
 293
 294 static void start_read_threads (int num)
 295 {
 296         int i;
 297
 298         if (read_threads != NULL)
 299                 return;
 300
 301         read_threads = (pthread_t *) calloc (num, sizeof (pthread_t));
 302         if (read_threads == NULL)
 303         {
 304                 ERROR ("plugin: start_read_threads: calloc failed.");
 305                 return;
 306         }
 307
 308         read_threads_num = 0;
 309         for (i = 0; i < num; i++)
 310         {
 311                 if (pthread_create (read_threads + read_threads_num, NULL,
 312                                         plugin_read_thread, NULL) == 0)
 313                 {
 314                         read_threads_num++;
 315                 }
 316                 else
 317                 {
 318                         ERROR ("plugin: start_read_threads: pthread_create failed.");
 319                         return;
 320                 }
 321         } /* for (i) */
 322 } /* void start_read_threads */
 323
 324 static void stop_read_threads (void)
 325 {
 326         int i;
 327
 328         if (read_threads == NULL)
 329                 return;
 330
 331         pthread_mutex_lock (&read_lock);
 332         read_loop = 0;
 333         DEBUG ("plugin: stop_read_threads: Signalling `read_cond'");
 334         pthread_cond_broadcast (&read_cond);
 335         pthread_mutex_unlock (&read_lock);
 336
 337         for (i = 0; i < read_threads_num; i++)
 338         {
 339                 if (pthread_join (read_threads[i], NULL) != 0)
 340                 {
 341                         ERROR ("plugin: stop_read_threads: pthread_join failed.");
 342                 }
 343                 read_threads[i] = (pthread_t) 0;
 344         }
 345         sfree (read_threads);
 346         read_threads_num = 0;
 347 } /* void stop_read_threads */
 348
 349 static int remove_read_functions (void)
 350 {
 351         llentry_t *this;
 352
 353         if (list_read == NULL)
 354                 return (0);
 355
 356         this = llist_head (list_read);
 357         while (this != NULL)
 358         {
 359                 llentry_t *next;
 360                 read_func_t *rf;
 361
 362                 next = this->next;
 363                 rf = (read_func_t *) this->value;
 364
 365                 free (this->key);
 366
 367                 plugin_user_data_destroy (&rf->udata);
 368                 free (rf);
 369
 370                 this = next;
 371         }
 372
 373         llist_destroy (list_read);
 374         list_read = NULL;
 375
 376         return (0);
 377 } /* }}} int remove_read_functions */
 378
 379 /*
 380  * Public functions
 381  */
 382 void plugin_set_dir (const char *dir)
 383 {
 384         if (plugindir != NULL)
 385                 free (plugindir);
 386
 387         if (dir == NULL)
 388                 plugindir = NULL;
 389         else if ((plugindir = strdup (dir)) == NULL)
 390         {
 391                 char errbuf[1024];
 392                 ERROR ("strdup failed: %s",
 393                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 394         }
 395 }
 396
 397 #define BUFSIZE 512
 398 int plugin_load (const char *type)
 399 {
 400         DIR  *dh;
 401         const char *dir;
 402         char  filename[BUFSIZE] = "";
 403         char  typename[BUFSIZE];
 404         int   typename_len;
 405         int   ret;
 406         struct stat    statbuf;
 407         struct dirent *de;
 408         int status;
 409
 410         DEBUG ("type = %s", type);
 411
 412         dir = plugin_get_dir ();
 413         ret = 1;
 414
 415         /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
 416          * type when matching the filename */
 417         status = ssnprintf (typename, sizeof (typename), "%s.so", type);
 418         if ((status < 0) || ((size_t) status >= sizeof (typename)))
 419         {
 420                 WARNING ("snprintf: truncated: `%s.so'", type);
 421                 return (-1);
 422         }
 423         typename_len = strlen (typename);
 424
 425         if ((dh = opendir (dir)) == NULL)
 426         {
 427                 char errbuf[1024];
 428                 ERROR ("opendir (%s): %s", dir,
 429                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 430                 return (-1);
 431         }
 432
 433         while ((de = readdir (dh)) != NULL)
 434         {
 435                 if (strncasecmp (de->d_name, typename, typename_len))
 436                         continue;
 437
 438                 status = ssnprintf (filename, sizeof (filename),
 439                                 "%s/%s", dir, de->d_name);
 440                 if ((status < 0) || ((size_t) status >= sizeof (filename)))
 441                 {
 442                         WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name);
 443                         continue;
 444                 }
 445
 446                 if (lstat (filename, &statbuf) == -1)
 447                 {
 448                         char errbuf[1024];
 449                         WARNING ("stat %s: %s", filename,
 450                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 451                         continue;
 452                 }
 453                 else if (!S_ISREG (statbuf.st_mode))
 454                 {
 455                         /* don't follow symlinks */
 456                         continue;
 457                 }
 458
 459                 if (plugin_load_file (filename) == 0)
 460                 {
 461                         /* success */
 462                         ret = 0;
 463                         break;
 464                 }
 465                 else
 466                 {
 467                         fprintf (stderr, "Unable to load plugin %s.\n", type);
 468                 }
 469         }
 470
 471         closedir (dh);
 472
 473         if (filename[0] == '\0')
 474                 fprintf (stderr, "Could not find plugin %s.\n", type);
 475
 476         return (ret);
 477 }
 478
 479 /*
 480  * The `register_*' functions follow
 481  */
 482 int plugin_register_config (const char *name,
 483                 int (*callback) (const char *key, const char *val),
 484                 const char **keys, int keys_num)
 485 {
 486         cf_register (name, callback, keys, keys_num);
 487         return (0);
 488 } /* int plugin_register_config */
 489
 490 int plugin_register_complex_config (const char *type,
 491                 int (*callback) (oconfig_item_t *))
 492 {
 493         return (cf_register_complex (type, callback));
 494 } /* int plugin_register_complex_config */
 495
 496 int plugin_register_init (const char *name,
 497                 int (*callback) (void))
 498 {
 499         return (register_callback (&list_init, name, (void *) callback));
 500 } /* plugin_register_init */
 501
 502 int plugin_register_read (const char *name,
 503                 int (*callback) (void))
 504 {
 505         read_func_t *rf;
 506
 507         rf = (read_func_t *) malloc (sizeof (read_func_t));
 508         if (rf == NULL)
 509         {
 510                 char errbuf[1024];
 511                 ERROR ("plugin_register_read: malloc failed: %s",
 512                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 513                 return (-1);
 514         }
 515
 516         memset (rf, 0, sizeof (read_func_t));
 517         rf->wait_time = interval_g;
 518         rf->wait_left = 0;
 519         rf->type = RF_SIMPLE;
 520         rf->callback.cb_simple = callback;
 521         rf->needs_read = DONE;
 522         rf->udata.data = NULL;
 523         rf->udata.free_func = NULL;
 524
 525         return (register_callback (&list_read, name, (void *) rf));
 526 } /* int plugin_register_read */
 527
 528 int plugin_register_complex_read (const char *name,
 529                 plugin_read_cb callback, user_data_t *user_data)
 530 {
 531         read_func_t *rf;
 532
 533         rf = (read_func_t *) malloc (sizeof (read_func_t));
 534         if (rf == NULL)
 535         {
 536                 ERROR ("plugin_register_complex_read: malloc failed.");
 537                 return (-1);
 538         }
 539
 540         memset (rf, 0, sizeof (read_func_t));
 541         rf->wait_time = interval_g;
 542         rf->wait_left = 0;
 543         rf->type = RF_COMPLEX;
 544         rf->callback.cb_complex = callback;
 545         rf->needs_read = DONE;
 546
 547         /* Set user data */
 548         if (user_data == NULL)
 549         {
 550                 rf->udata.data = NULL;
 551                 rf->udata.free_func = NULL;
 552         }
 553         else
 554         {
 555                 rf->udata = *user_data;
 556         }
 557
 558         return (register_callback (&list_read, name, (void *) rf));
 559 } /* int plugin_register_complex_read */
 560
 561 int plugin_register_write (const char *name,
 562                 plugin_write_cb callback, user_data_t *user_data)
 563 {
 564         write_func_t *wf;
 565
 566         wf = (write_func_t *) malloc (sizeof (*wf));
 567         if (wf == NULL)
 568         {
 569                 ERROR ("plugin_register_write: malloc failed.");
 570                 return (-1);
 571         }
 572         memset (wf, 0, sizeof (*wf));
 573
 574         wf->callback = callback;
 575         if (user_data == NULL)
 576         {
 577                 wf->udata.data = NULL;
 578                 wf->udata.free_func = NULL;
 579         }
 580         else
 581         {
 582                 wf->udata = *user_data;
 583         }
 584
 585         return (register_callback (&list_write, name, (void *) wf));
 586 } /* int plugin_register_write */
 587
 588 int plugin_register_flush (const char *name,
 589                 plugin_flush_cb callback, user_data_t *user_data)
 590 {
 591         flush_func_t *ff;
 592
 593         ff = (flush_func_t *) malloc (sizeof (*ff));
 594         if (ff == NULL)
 595         {
 596                 ERROR ("plugin_register_flush: malloc failed.");
 597                 return (-1);
 598         }
 599         memset (ff, 0, sizeof (*ff));
 600
 601         ff->callback = callback;
 602         if (user_data == NULL)
 603         {
 604                 ff->udata.data = NULL;
 605                 ff->udata.free_func = NULL;
 606         }
 607         else
 608         {
 609                 ff->udata = *user_data;
 610         }
 611
 612         return (register_callback (&list_flush, name, (void *) ff));
 613 } /* int plugin_register_flush */
 614
 615 int plugin_register_shutdown (char *name,
 616                 int (*callback) (void))
 617 {
 618         return (register_callback (&list_shutdown, name, (void *) callback));
 619 } /* int plugin_register_shutdown */
 620
 621 int plugin_register_data_set (const data_set_t *ds)
 622 {
 623         data_set_t *ds_copy;
 624         int i;
 625
 626         if ((data_sets != NULL)
 627                         && (c_avl_get (data_sets, ds->type, NULL) == 0))
 628         {
 629                 NOTICE ("Replacing DS `%s' with another version.", ds->type);
 630                 plugin_unregister_data_set (ds->type);
 631         }
 632         else if (data_sets == NULL)
 633         {
 634                 data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp);
 635                 if (data_sets == NULL)
 636                         return (-1);
 637         }
 638
 639         ds_copy = (data_set_t *) malloc (sizeof (data_set_t));
 640         if (ds_copy == NULL)
 641                 return (-1);
 642         memcpy(ds_copy, ds, sizeof (data_set_t));
 643
 644         ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t)
 645                         * ds->ds_num);
 646         if (ds_copy->ds == NULL)
 647         {
 648                 free (ds_copy);
 649                 return (-1);
 650         }
 651
 652         for (i = 0; i < ds->ds_num; i++)
 653                 memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t));
 654
 655         return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy));
 656 } /* int plugin_register_data_set */
 657
 658 int plugin_register_log (const char *name,
 659                 plugin_log_cb callback, user_data_t *user_data)
 660 {
 661         log_func_t *lf;
 662
 663         lf = (log_func_t *) malloc (sizeof (*lf));
 664         if (lf == NULL)
 665         {
 666                 ERROR ("plugin_register_log: malloc failed.");
 667                 return (-1);
 668         }
 669         memset (lf, 0, sizeof (*lf));
 670
 671         lf->callback = callback;
 672         if (user_data == NULL)
 673         {
 674                 lf->udata.data = NULL;
 675                 lf->udata.free_func = NULL;
 676         }
 677         else
 678         {
 679                 lf->udata = *user_data;
 680         }
 681
 682         return (register_callback (&list_log, name, (void *) lf));
 683 } /* int plugin_register_log */
 684
 685 int plugin_register_notification (const char *name,
 686                 int (*callback) (const notification_t *notif))
 687 {
 688         return (register_callback (&list_notification, name, (void *) callback));
 689 } /* int plugin_register_log */
 690
 691 int plugin_unregister_config (const char *name)
 692 {
 693         cf_unregister (name);
 694         return (0);
 695 } /* int plugin_unregister_config */
 696
 697 int plugin_unregister_complex_config (const char *name)
 698 {
 699         cf_unregister_complex (name);
 700         return (0);
 701 } /* int plugin_unregister_complex_config */
 702
 703 int plugin_unregister_init (const char *name)
 704 {
 705         return (plugin_unregister (list_init, name));
 706 }
 707
 708 int plugin_unregister_read (const char *name)
 709 {
 710         llentry_t *e;
 711         read_func_t *rf;
 712
 713         e = llist_search (list_read, name);
 714
 715         if (e == NULL)
 716                 return (-1);
 717
 718         llist_remove (list_read, e);
 719
 720         rf = (read_func_t *) e->value;
 721         plugin_user_data_destroy (&rf->udata);
 722         free (rf);
 723         free (e->key);
 724
 725         llentry_destroy (e);
 726
 727         return (0);
 728 }
 729
 730 int plugin_unregister_write (const char *name)
 731 {
 732         llentry_t *e;
 733         write_func_t *wf;
 734
 735         e = llist_search (list_write, name);
 736
 737         if (e == NULL)
 738                 return (-1);
 739
 740         llist_remove (list_write, e);
 741
 742         wf = (write_func_t *) e->value;
 743         plugin_user_data_destroy (&wf->udata);
 744         free (wf);
 745         free (e->key);
 746
 747         llentry_destroy (e);
 748
 749         return (0);
 750 }
 751
 752 int plugin_unregister_flush (const char *name)
 753 {
 754         llentry_t *e;
 755         flush_func_t *ff;
 756
 757         e = llist_search (list_flush, name);
 758
 759         if (e == NULL)
 760                 return (-1);
 761
 762         llist_remove (list_flush, e);
 763
 764         ff = (flush_func_t *) e->value;
 765         plugin_user_data_destroy (&ff->udata);
 766         free (ff);
 767         free (e->key);
 768
 769         llentry_destroy (e);
 770
 771         return (0);
 772 }
 773
 774 int plugin_unregister_shutdown (const char *name)
 775 {
 776         return (plugin_unregister (list_shutdown, name));
 777 }
 778
 779 int plugin_unregister_data_set (const char *name)
 780 {
 781         data_set_t *ds;
 782
 783         if (data_sets == NULL)
 784                 return (-1);
 785
 786         if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0)
 787                 return (-1);
 788
 789         sfree (ds->ds);
 790         sfree (ds);
 791
 792         return (0);
 793 } /* int plugin_unregister_data_set */
 794
 795 int plugin_unregister_log (const char *name)
 796 {
 797         llentry_t *e;
 798         log_func_t *lf;
 799
 800         e = llist_search (list_log, name);
 801
 802         if (e == NULL)
 803                 return (-1);
 804
 805         llist_remove (list_log, e);
 806
 807         lf = (log_func_t *) e->value;
 808         plugin_user_data_destroy (&lf->udata);
 809         free (lf);
 810         free (e->key);
 811
 812         llentry_destroy (e);
 813
 814         return (0);
 815 }
 816
 817 int plugin_unregister_notification (const char *name)
 818 {
 819         return (plugin_unregister (list_notification, name));
 820 }
 821
 822 void plugin_init_all (void)
 823 {
 824         const char *chain_name;
 825         int (*callback) (void);
 826         llentry_t *le;
 827         int status;
 828
 829         /* Init the value cache */
 830         uc_init ();
 831
 832         chain_name = global_option_get ("PreCacheChain");
 833         pre_cache_chain = fc_chain_get_by_name (chain_name);
 834
 835         chain_name = global_option_get ("PostCacheChain");
 836         post_cache_chain = fc_chain_get_by_name (chain_name);
 837
 838
 839         if ((list_init == NULL) && (list_read == NULL))
 840                 return;
 841
 842         /* Calling all init callbacks before checking if read callbacks
 843          * are available allows the init callbacks to register the read
 844          * callback. */
 845         le = llist_head (list_init);
 846         while (le != NULL)
 847         {
 848                 callback = (int (*) (void)) le->value;
 849                 status = (*callback) ();
 850
 851                 if (status != 0)
 852                 {
 853                         ERROR ("Initialization of plugin `%s' "
 854                                         "failed with status %i. "
 855                                         "Plugin will be unloaded.",
 856                                         le->key, status);
 857                         /* Plugins that register read callbacks from the init
 858                          * callback should take care of appropriate error
 859                          * handling themselves. */
 860                         /* FIXME: Unload _all_ functions */
 861                         plugin_unregister_read (le->key);
 862                 }
 863
 864                 le = le->next;
 865         }
 866
 867         /* Start read-threads */
 868         if (list_read != NULL)
 869         {
 870                 const char *rt;
 871                 int num;
 872                 rt = global_option_get ("ReadThreads");
 873                 num = atoi (rt);
 874                 if (num != -1)
 875                         start_read_threads ((num > 0) ? num : 5);
 876         }
 877 } /* void plugin_init_all */
 878
 879 void plugin_read_all (void)
 880 {
 881         llentry_t   *le;
 882         read_func_t *rf;
 883
 884         uc_check_timeout ();
 885
 886         if (list_read == NULL)
 887                 return;
 888
 889         pthread_mutex_lock (&read_lock);
 890
 891         le = llist_head (list_read);
 892         while (le != NULL)
 893         {
 894                 rf = (read_func_t *) le->value;
 895
 896                 if (rf->needs_read != DONE)
 897                 {
 898                         le = le->next;
 899                         continue;
 900                 }
 901
 902                 if (rf->wait_left > 0)
 903                         rf->wait_left -= interval_g;
 904
 905                 if (rf->wait_left <= 0)
 906                 {
 907                         rf->needs_read = TODO;
 908                 }
 909
 910                 le = le->next;
 911         }
 912
 913         DEBUG ("plugin: plugin_read_all: Signalling `read_cond'");
 914         pthread_cond_broadcast (&read_cond);
 915         pthread_mutex_unlock (&read_lock);
 916 } /* void plugin_read_all */
 917
 918 /* Read function called when the `-T' command line argument is given. */
 919 int plugin_read_all_once (void)
 920 {
 921         llentry_t   *le;
 922         read_func_t *rf;
 923         int status;
 924         int return_status = 0;
 925
 926         if (list_read == NULL)
 927         {
 928                 NOTICE ("No read-functions are registered.");
 929                 return (0);
 930         }
 931
 932         for (le = llist_head (list_read);
 933              le != NULL;
 934              le = le->next)
 935         {
 936                 rf = (read_func_t *) le->value;
 937
 938                 if (rf->type == RF_SIMPLE)
 939                 {
 940                         status = rf->callback.cb_simple ();
 941                 }
 942                 else
 943                 {
 944                         assert (rf->type == RF_COMPLEX);
 945                         status = rf->callback.cb_complex (&rf->udata);
 946                 }
 947
 948                 if (status != 0)
 949                 {
 950                         NOTICE ("read-function of plugin `%s' failed.",
 951                                 le->key);
 952                         return_status = -1;
 953                 }
 954         }
 955
 956         return (return_status);
 957 } /* int plugin_read_all_once */
 958
 959 int plugin_write (const char *plugin, /* {{{ */
 960                 const data_set_t *ds, const value_list_t *vl)
 961 {
 962   llentry_t *le;
 963   int status;
 964
 965   if (vl == NULL)
 966     return (EINVAL);
 967
 968   if (list_write == NULL)
 969     return (ENOENT);
 970
 971   if (ds == NULL)
 972   {
 973     ds = plugin_get_ds (vl->type);
 974     if (ds == NULL)
 975     {
 976       ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type);
 977       return (ENOENT);
 978     }
 979   }
 980
 981   if (plugin == NULL)
 982   {
 983     int success = 0;
 984     int failure = 0;
 985
 986     le = llist_head (list_write);
 987     while (le != NULL)
 988     {
 989       write_func_t *wf = le->value;
 990
 991       DEBUG ("plugin: plugin_write: Writing values via %s.", le->key);
 992       status = wf->callback (ds, vl, &wf->udata);
 993       if (status != 0)
 994         failure++;
 995       else
 996         success++;
 997
 998       le = le->next;
 999     }
1000
1001     if ((success == 0) && (failure != 0))
1002       status = -1;
1003     else
1004       status = 0;
1005   }
1006   else /* plugin != NULL */
1007   {
1008     write_func_t *wf;
1009     le = llist_head (list_write);
1010     while (le != NULL)
1011     {
1012       if (strcasecmp (plugin, le->key) == 0)
1013         break;
1014
1015       le = le->next;
1016     }
1017
1018     if (le == NULL)
1019       return (ENOENT);
1020
1021     wf = le->value;
1022
1023     DEBUG ("plugin: plugin_write: Writing values via %s.", le->key);
1024     status = wf->callback (ds, vl, &wf->udata);
1025   }
1026
1027   return (status);
1028 } /* }}} int plugin_write */
1029
1030 int plugin_flush (const char *plugin, int timeout, const char *identifier)
1031 {
1032   llentry_t *le;
1033
1034   if (list_flush == NULL)
1035     return (0);
1036
1037   le = llist_head (list_flush);
1038   while (le != NULL)
1039   {
1040     flush_func_t *ff;
1041
1042     if ((plugin != NULL)
1043         && (strcmp (plugin, le->key) != 0))
1044     {
1045       le = le->next;
1046       continue;
1047     }
1048
1049     ff = (flush_func_t *) le->value;
1050
1051     ff->callback (timeout, identifier, &ff->udata);
1052
1053     le = le->next;
1054   }
1055   return (0);
1056 } /* int plugin_flush */
1057
1058 void plugin_shutdown_all (void)
1059 {
1060         int (*callback) (void);
1061         llentry_t *le;
1062
1063         stop_read_threads ();
1064         remove_read_functions ();
1065
1066         if (list_shutdown == NULL)
1067                 return;
1068
1069         le = llist_head (list_shutdown);
1070         while (le != NULL)
1071         {
1072                 callback = (int (*) (void)) le->value;
1073
1074                 /* Advance the pointer before calling the callback allows
1075                  * shutdown functions to unregister themselves. If done the
1076                  * other way around the memory `le' points to will be freed
1077                  * after callback returns. */
1078                 le = le->next;
1079
1080                 (*callback) ();
1081         }
1082 } /* void plugin_shutdown_all */
1083
1084 int plugin_dispatch_values (value_list_t *vl)
1085 {
1086         int status;
1087         static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
1088
1089         data_set_t *ds;
1090
1091         if ((vl == NULL) || (*vl->type == '\0')) {
1092                 ERROR ("plugin_dispatch_values: Invalid value list.");
1093                 return (-1);
1094         }
1095
1096         if (list_write == NULL)
1097                 c_complain_once (LOG_WARNING, &no_write_complaint,
1098                                 "plugin_dispatch_values: No write callback has been "
1099                                 "registered. Please load at least one output plugin, "
1100                                 "if you want the collected data to be stored.");
1101
1102         if (data_sets == NULL)
1103         {
1104                 ERROR ("plugin_dispatch_values: No data sets registered. "
1105                                 "Could the types database be read? Check "
1106                                 "your `TypesDB' setting!");
1107                 return (-1);
1108         }
1109
1110         if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0)
1111         {
1112                 INFO ("plugin_dispatch_values: Dataset not found: %s", vl->type);
1113                 return (-1);
1114         }
1115
1116         if (vl->time == 0)
1117                 vl->time = time (NULL);
1118
1119         DEBUG ("plugin_dispatch_values: time = %u; interval = %i; "
1120                         "host = %s; "
1121                         "plugin = %s; plugin_instance = %s; "
1122                         "type = %s; type_instance = %s;",
1123                         (unsigned int) vl->time, vl->interval,
1124                         vl->host,
1125                         vl->plugin, vl->plugin_instance,
1126                         vl->type, vl->type_instance);
1127
1128 #if COLLECT_DEBUG
1129         assert (0 == strcmp (ds->type, vl->type));
1130 #else
1131         if (0 != strcmp (ds->type, vl->type))
1132                 WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)",
1133                                 ds->type, vl->type);
1134 #endif
1135
1136 #if COLLECT_DEBUG
1137         assert (ds->ds_num == vl->values_len);
1138 #else
1139         if (ds->ds_num != vl->values_len)
1140         {
1141                 ERROR ("plugin_dispatch_values: ds->type = %s: "
1142                                 "(ds->ds_num = %i) != "
1143                                 "(vl->values_len = %i)",
1144                                 ds->type, ds->ds_num, vl->values_len);
1145                 return (-1);
1146         }
1147 #endif
1148
1149         escape_slashes (vl->host, sizeof (vl->host));
1150         escape_slashes (vl->plugin, sizeof (vl->plugin));
1151         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
1152         escape_slashes (vl->type, sizeof (vl->type));
1153         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
1154
1155         if (pre_cache_chain != NULL)
1156         {
1157                 status = fc_process_chain (ds, vl, pre_cache_chain);
1158                 if (status < 0)
1159                 {
1160                         WARNING ("plugin_dispatch_values: Running the "
1161                                         "pre-cache chain failed with "
1162                                         "status %i (%#x).",
1163                                         status, status);
1164                 }
1165                 else if (status == FC_TARGET_STOP)
1166                         return (0);
1167         }
1168
1169         /* Update the value cache */
1170         uc_update (ds, vl);
1171
1172         if (post_cache_chain != NULL)
1173         {
1174                 status = fc_process_chain (ds, vl, post_cache_chain);
1175                 if (status < 0)
1176                 {
1177                         WARNING ("plugin_dispatch_values: Running the "
1178                                         "post-cache chain failed with "
1179                                         "status %i (%#x).",
1180                                         status, status);
1181                 }
1182         }
1183         else
1184                 fc_default_action (ds, vl);
1185
1186         return (0);
1187 } /* int plugin_dispatch_values */
1188
1189 int plugin_dispatch_notification (const notification_t *notif)
1190 {
1191         int (*callback) (const notification_t *);
1192         llentry_t *le;
1193         /* Possible TODO: Add flap detection here */
1194
1195         DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; "
1196                         "time = %u; host = %s;",
1197                         notif->severity, notif->message,
1198                         (unsigned int) notif->time, notif->host);
1199
1200         /* Nobody cares for notifications */
1201         if (list_notification == NULL)
1202                 return (-1);
1203
1204         le = llist_head (list_notification);
1205         while (le != NULL)
1206         {
1207                 callback = (int (*) (const notification_t *)) le->value;
1208                 (*callback) (notif);
1209
1210                 le = le->next;
1211         }
1212
1213         return (0);
1214 } /* int plugin_dispatch_notification */
1215
1216 void plugin_log (int level, const char *format, ...)
1217 {
1218         char msg[1024];
1219         va_list ap;
1220         llentry_t *le;
1221
1222         if (list_log == NULL)
1223                 return;
1224
1225 #if !COLLECT_DEBUG
1226         if (level >= LOG_DEBUG)
1227                 return;
1228 #endif
1229
1230         va_start (ap, format);
1231         vsnprintf (msg, sizeof (msg), format, ap);
1232         msg[sizeof (msg) - 1] = '\0';
1233         va_end (ap);
1234
1235         le = llist_head (list_log);
1236         while (le != NULL)
1237         {
1238                 log_func_t *lf;
1239
1240                 lf = (log_func_t *) le->value;
1241
1242                 lf->callback (level, msg, &lf->udata);
1243
1244                 le = le->next;
1245         }
1246 } /* void plugin_log */
1247
1248 const data_set_t *plugin_get_ds (const char *name)
1249 {
1250         data_set_t *ds;
1251
1252         if (c_avl_get (data_sets, name, (void *) &ds) != 0)
1253         {
1254                 DEBUG ("No such dataset registered: %s", name);
1255                 return (NULL);
1256         }
1257
1258         return (ds);
1259 } /* data_set_t *plugin_get_ds */
1260
1261 static int plugin_notification_meta_add (notification_t *n,
1262     const char *name,
1263     enum notification_meta_type_e type,
1264     const void *value)
1265 {
1266   notification_meta_t *meta;
1267   notification_meta_t *tail;
1268
1269   if ((n == NULL) || (name == NULL) || (value == NULL))
1270   {
1271     ERROR ("plugin_notification_meta_add: A pointer is NULL!");
1272     return (-1);
1273   }
1274
1275   meta = (notification_meta_t *) malloc (sizeof (notification_meta_t));
1276   if (meta == NULL)
1277   {
1278     ERROR ("plugin_notification_meta_add: malloc failed.");
1279     return (-1);
1280   }
1281   memset (meta, 0, sizeof (notification_meta_t));
1282
1283   sstrncpy (meta->name, name, sizeof (meta->name));
1284   meta->type = type;
1285
1286   switch (type)
1287   {
1288     case NM_TYPE_STRING:
1289     {
1290       meta->nm_value.nm_string = strdup ((const char *) value);
1291       if (meta->nm_value.nm_string == NULL)
1292       {
1293         ERROR ("plugin_notification_meta_add: strdup failed.");
1294         sfree (meta);
1295         return (-1);
1296       }
1297       break;
1298     }
1299     case NM_TYPE_SIGNED_INT:
1300     {
1301       meta->nm_value.nm_signed_int = *((int64_t *) value);
1302       break;
1303     }
1304     case NM_TYPE_UNSIGNED_INT:
1305     {
1306       meta->nm_value.nm_unsigned_int = *((uint64_t *) value);
1307       break;
1308     }
1309     case NM_TYPE_DOUBLE:
1310     {
1311       meta->nm_value.nm_double = *((double *) value);
1312       break;
1313     }
1314     case NM_TYPE_BOOLEAN:
1315     {
1316       meta->nm_value.nm_boolean = *((bool *) value);
1317       break;
1318     }
1319     default:
1320     {
1321       ERROR ("plugin_notification_meta_add: Unknown type: %i", type);
1322       sfree (meta);
1323       return (-1);
1324     }
1325   } /* switch (type) */
1326
1327   meta->next = NULL;
1328   tail = n->meta;
1329   while ((tail != NULL) && (tail->next != NULL))
1330     tail = tail->next;
1331
1332   if (tail == NULL)
1333     n->meta = meta;
1334   else
1335     tail->next = meta;
1336
1337   return (0);
1338 } /* int plugin_notification_meta_add */
1339
1340 int plugin_notification_meta_add_string (notification_t *n,
1341     const char *name,
1342     const char *value)
1343 {
1344   return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value));
1345 }
1346
1347 int plugin_notification_meta_add_signed_int (notification_t *n,
1348     const char *name,
1349     int64_t value)
1350 {
1351   return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value));
1352 }
1353
1354 int plugin_notification_meta_add_unsigned_int (notification_t *n,
1355     const char *name,
1356     uint64_t value)
1357 {
1358   return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value));
1359 }
1360
1361 int plugin_notification_meta_add_double (notification_t *n,
1362     const char *name,
1363     double value)
1364 {
1365   return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value));
1366 }
1367
1368 int plugin_notification_meta_add_boolean (notification_t *n,
1369     const char *name,
1370     bool value)
1371 {
1372   return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value));
1373 }
1374
1375 int plugin_notification_meta_copy (notification_t *dst,
1376     const notification_t *src)
1377 {
1378   notification_meta_t *meta;
1379
1380   assert (dst != NULL);
1381   assert (src != NULL);
1382   assert (dst != src);
1383   assert ((src->meta == NULL) || (src->meta != dst->meta));
1384
1385   for (meta = src->meta; meta != NULL; meta = meta->next)
1386   {
1387     if (meta->type == NM_TYPE_STRING)
1388       plugin_notification_meta_add_string (dst, meta->name,
1389           meta->nm_value.nm_string);
1390     else if (meta->type == NM_TYPE_SIGNED_INT)
1391       plugin_notification_meta_add_signed_int (dst, meta->name,
1392           meta->nm_value.nm_signed_int);
1393     else if (meta->type == NM_TYPE_UNSIGNED_INT)
1394       plugin_notification_meta_add_unsigned_int (dst, meta->name,
1395           meta->nm_value.nm_unsigned_int);
1396     else if (meta->type == NM_TYPE_DOUBLE)
1397       plugin_notification_meta_add_double (dst, meta->name,
1398           meta->nm_value.nm_double);
1399     else if (meta->type == NM_TYPE_BOOLEAN)
1400       plugin_notification_meta_add_boolean (dst, meta->name,
1401           meta->nm_value.nm_boolean);
1402   }
1403
1404   return (0);
1405 } /* int plugin_notification_meta_copy */
1406
1407 int plugin_notification_meta_free (notification_meta_t *n)
1408 {
1409   notification_meta_t *this;
1410   notification_meta_t *next;
1411
1412   if (n == NULL)
1413   {
1414     ERROR ("plugin_notification_meta_free: n == NULL!");
1415     return (-1);
1416   }
1417
1418   this = n;
1419   while (this != NULL)
1420   {
1421     next = this->next;
1422
1423     if (this->type == NM_TYPE_STRING)
1424     {
1425       free ((char *)this->nm_value.nm_string);
1426       this->nm_value.nm_string = NULL;
1427     }
1428     sfree (this);
1429
1430     this = next;
1431   }
1432
1433   return (0);
1434 } /* int plugin_notification_meta_free */