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