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