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_read_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_read_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_read_threads: pthread_create failed.");
 305                         return;
 306                 }
 307         } /* for (i) */
 308 } /* void start_read_threads */
 309
 310 static void stop_read_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_read_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_read_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_read_threads */
 334
 335 static int remove_read_functions (void)
 336 {
 337         llentry_t *this;
 338
 339         if (list_read == NULL)
 340                 return (0);
 341
 342         this = llist_head (list_read);
 343         while (this != NULL)
 344         {
 345                 llentry_t *next;
 346                 read_func_t *rf;
 347
 348                 next = this->next;
 349                 rf = (read_func_t *) this->value;
 350
 351                 free (this->key);
 352
 353                 plugin_user_data_destroy (&rf->udata);
 354                 free (rf);
 355
 356                 this = next;
 357         }
 358
 359         llist_destroy (list_read);
 360         list_read = NULL;
 361
 362         return (0);
 363 } /* }}} int remove_read_functions */
 364
 365 /*
 366  * Public functions
 367  */
 368 void plugin_set_dir (const char *dir)
 369 {
 370         if (plugindir != NULL)
 371                 free (plugindir);
 372
 373         if (dir == NULL)
 374                 plugindir = NULL;
 375         else if ((plugindir = strdup (dir)) == NULL)
 376         {
 377                 char errbuf[1024];
 378                 ERROR ("strdup failed: %s",
 379                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 380         }
 381 }
 382
 383 #define BUFSIZE 512
 384 int plugin_load (const char *type)
 385 {
 386         DIR  *dh;
 387         const char *dir;
 388         char  filename[BUFSIZE] = "";
 389         char  typename[BUFSIZE];
 390         int   typename_len;
 391         int   ret;
 392         struct stat    statbuf;
 393         struct dirent *de;
 394         int status;
 395
 396         DEBUG ("type = %s", type);
 397
 398         dir = plugin_get_dir ();
 399         ret = 1;
 400
 401         /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
 402          * type when matching the filename */
 403         status = ssnprintf (typename, sizeof (typename), "%s.so", type);
 404         if ((status < 0) || ((size_t) status >= sizeof (typename)))
 405         {
 406                 WARNING ("snprintf: truncated: `%s.so'", type);
 407                 return (-1);
 408         }
 409         typename_len = strlen (typename);
 410
 411         if ((dh = opendir (dir)) == NULL)
 412         {
 413                 char errbuf[1024];
 414                 ERROR ("opendir (%s): %s", dir,
 415                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 416                 return (-1);
 417         }
 418
 419         while ((de = readdir (dh)) != NULL)
 420         {
 421                 if (strncasecmp (de->d_name, typename, typename_len))
 422                         continue;
 423
 424                 status = ssnprintf (filename, sizeof (filename),
 425                                 "%s/%s", dir, de->d_name);
 426                 if ((status < 0) || ((size_t) status >= sizeof (filename)))
 427                 {
 428                         WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name);
 429                         continue;
 430                 }
 431
 432                 if (lstat (filename, &statbuf) == -1)
 433                 {
 434                         char errbuf[1024];
 435                         WARNING ("stat %s: %s", filename,
 436                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 437                         continue;
 438                 }
 439                 else if (!S_ISREG (statbuf.st_mode))
 440                 {
 441                         /* don't follow symlinks */
 442                         continue;
 443                 }
 444
 445                 if (plugin_load_file (filename) == 0)
 446                 {
 447                         /* success */
 448                         ret = 0;
 449                         break;
 450                 }
 451                 else
 452                 {
 453                         fprintf (stderr, "Unable to load plugin %s.\n", type);
 454                 }
 455         }
 456
 457         closedir (dh);
 458
 459         if (filename[0] == '\0')
 460                 fprintf (stderr, "Could not find plugin %s.\n", type);
 461
 462         return (ret);
 463 }
 464
 465 /*
 466  * The `register_*' functions follow
 467  */
 468 int plugin_register_config (const char *name,
 469                 int (*callback) (const char *key, const char *val),
 470                 const char **keys, int keys_num)
 471 {
 472         cf_register (name, callback, keys, keys_num);
 473         return (0);
 474 } /* int plugin_register_config */
 475
 476 int plugin_register_complex_config (const char *type,
 477                 int (*callback) (oconfig_item_t *))
 478 {
 479         return (cf_register_complex (type, callback));
 480 } /* int plugin_register_complex_config */
 481
 482 int plugin_register_init (const char *name,
 483                 int (*callback) (void))
 484 {
 485         return (register_callback (&list_init, name, (void *) callback));
 486 } /* plugin_register_init */
 487
 488 int plugin_register_read (const char *name,
 489                 int (*callback) (void))
 490 {
 491         read_func_t *rf;
 492
 493         rf = (read_func_t *) malloc (sizeof (read_func_t));
 494         if (rf == NULL)
 495         {
 496                 char errbuf[1024];
 497                 ERROR ("plugin_register_read: malloc failed: %s",
 498                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 499                 return (-1);
 500         }
 501
 502         memset (rf, 0, sizeof (read_func_t));
 503         rf->wait_time = interval_g;
 504         rf->wait_left = 0;
 505         rf->type = RF_SIMPLE;
 506         rf->callback.cb_simple = callback;
 507         rf->needs_read = DONE;
 508         rf->udata.data = NULL;
 509         rf->udata.free_func = NULL;
 510
 511         return (register_callback (&list_read, name, (void *) rf));
 512 } /* int plugin_register_read */
 513
 514 int plugin_register_complex_read (const char *name,
 515                 plugin_read_cb callback, user_data_t *user_data)
 516 {
 517         read_func_t *rf;
 518
 519         rf = (read_func_t *) malloc (sizeof (read_func_t));
 520         if (rf == NULL)
 521         {
 522                 ERROR ("plugin_register_complex_read: malloc failed.");
 523                 return (-1);
 524         }
 525
 526         memset (rf, 0, sizeof (read_func_t));
 527         rf->wait_time = interval_g;
 528         rf->wait_left = 0;
 529         rf->type = RF_COMPLEX;
 530         rf->callback.cb_complex = callback;
 531         rf->needs_read = DONE;
 532
 533         /* Set user data */
 534         if (user_data == NULL)
 535         {
 536                 rf->udata.data = NULL;
 537                 rf->udata.free_func = NULL;
 538         }
 539         else
 540         {
 541                 rf->udata = *user_data;
 542         }
 543
 544         return (register_callback (&list_read, name, (void *) rf));
 545 } /* int plugin_register_complex_read */
 546
 547 int plugin_register_write (const char *name,
 548                 plugin_write_cb callback, user_data_t *user_data)
 549 {
 550         write_func_t *wr;
 551
 552         wr = (write_func_t *) malloc (sizeof (*wr));
 553         if (wr == NULL)
 554         {
 555                 ERROR ("plugin_register_write: malloc failed.");
 556                 return (-1);
 557         }
 558         memset (wr, 0, sizeof (*wr));
 559
 560         wr->callback = callback;
 561         if (user_data == NULL)
 562         {
 563                 wr->udata.data = NULL;
 564                 wr->udata.free_func = NULL;
 565         }
 566         else
 567         {
 568                 wr->udata = *user_data;
 569         }
 570
 571         return (register_callback (&list_write, name, (void *) wr));
 572 } /* int plugin_register_write */
 573
 574 int plugin_register_flush (const char *name,
 575                 int (*callback) (const int timeout, const char *identifier))
 576 {
 577         return (register_callback (&list_flush, name, (void *) callback));
 578 } /* int plugin_register_flush */
 579
 580 int plugin_register_shutdown (char *name,
 581                 int (*callback) (void))
 582 {
 583         return (register_callback (&list_shutdown, name, (void *) callback));
 584 } /* int plugin_register_shutdown */
 585
 586 int plugin_register_data_set (const data_set_t *ds)
 587 {
 588         data_set_t *ds_copy;
 589         int i;
 590
 591         if ((data_sets != NULL)
 592                         && (c_avl_get (data_sets, ds->type, NULL) == 0))
 593         {
 594                 NOTICE ("Replacing DS `%s' with another version.", ds->type);
 595                 plugin_unregister_data_set (ds->type);
 596         }
 597         else if (data_sets == NULL)
 598         {
 599                 data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp);
 600                 if (data_sets == NULL)
 601                         return (-1);
 602         }
 603
 604         ds_copy = (data_set_t *) malloc (sizeof (data_set_t));
 605         if (ds_copy == NULL)
 606                 return (-1);
 607         memcpy(ds_copy, ds, sizeof (data_set_t));
 608
 609         ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t)
 610                         * ds->ds_num);
 611         if (ds_copy->ds == NULL)
 612         {
 613                 free (ds_copy);
 614                 return (-1);
 615         }
 616
 617         for (i = 0; i < ds->ds_num; i++)
 618                 memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t));
 619
 620         return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy));
 621 } /* int plugin_register_data_set */
 622
 623 int plugin_register_log (char *name,
 624                 void (*callback) (int priority, const char *msg))
 625 {
 626         return (register_callback (&list_log, name, (void *) callback));
 627 } /* int plugin_register_log */
 628
 629 int plugin_register_notification (const char *name,
 630                 int (*callback) (const notification_t *notif))
 631 {
 632         return (register_callback (&list_notification, name, (void *) callback));
 633 } /* int plugin_register_log */
 634
 635 int plugin_unregister_config (const char *name)
 636 {
 637         cf_unregister (name);
 638         return (0);
 639 } /* int plugin_unregister_config */
 640
 641 int plugin_unregister_complex_config (const char *name)
 642 {
 643         cf_unregister_complex (name);
 644         return (0);
 645 } /* int plugin_unregister_complex_config */
 646
 647 int plugin_unregister_init (const char *name)
 648 {
 649         return (plugin_unregister (list_init, name));
 650 }
 651
 652 int plugin_unregister_read (const char *name)
 653 {
 654         llentry_t *e;
 655         read_func_t *rf;
 656
 657         e = llist_search (list_read, name);
 658
 659         if (e == NULL)
 660                 return (-1);
 661
 662         llist_remove (list_read, e);
 663
 664         rf = (read_func_t *) e->value;
 665         plugin_user_data_destroy (&rf->udata);
 666         free (rf);
 667         free (e->key);
 668
 669         llentry_destroy (e);
 670
 671         return (0);
 672 }
 673
 674 int plugin_unregister_write (const char *name)
 675 {
 676         llentry_t *e;
 677         write_func_t *wf;
 678
 679         e = llist_search (list_write, name);
 680
 681         if (e == NULL)
 682                 return (-1);
 683
 684         llist_remove (list_write, e);
 685
 686         wf = (write_func_t *) e->value;
 687         plugin_user_data_destroy (&wf->udata);
 688         free (wf);
 689         free (e->key);
 690
 691         llentry_destroy (e);
 692
 693         return (0);
 694 }
 695
 696 int plugin_unregister_flush (const char *name)
 697 {
 698         return (plugin_unregister (list_flush, name));
 699 }
 700
 701 int plugin_unregister_shutdown (const char *name)
 702 {
 703         return (plugin_unregister (list_shutdown, name));
 704 }
 705
 706 int plugin_unregister_data_set (const char *name)
 707 {
 708         data_set_t *ds;
 709
 710         if (data_sets == NULL)
 711                 return (-1);
 712
 713         if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0)
 714                 return (-1);
 715
 716         sfree (ds->ds);
 717         sfree (ds);
 718
 719         return (0);
 720 } /* int plugin_unregister_data_set */
 721
 722 int plugin_unregister_log (const char *name)
 723 {
 724         return (plugin_unregister (list_log, name));
 725 }
 726
 727 int plugin_unregister_notification (const char *name)
 728 {
 729         return (plugin_unregister (list_notification, name));
 730 }
 731
 732 void plugin_init_all (void)
 733 {
 734         const char *chain_name;
 735         int (*callback) (void);
 736         llentry_t *le;
 737         int status;
 738
 739         /* Init the value cache */
 740         uc_init ();
 741
 742         chain_name = global_option_get ("PreCacheChain");
 743         pre_cache_chain = fc_chain_get_by_name (chain_name);
 744
 745         chain_name = global_option_get ("PostCacheChain");
 746         post_cache_chain = fc_chain_get_by_name (chain_name);
 747
 748
 749         if ((list_init == NULL) && (list_read == NULL))
 750                 return;
 751
 752         /* Calling all init callbacks before checking if read callbacks
 753          * are available allows the init callbacks to register the read
 754          * callback. */
 755         le = llist_head (list_init);
 756         while (le != NULL)
 757         {
 758                 callback = (int (*) (void)) le->value;
 759                 status = (*callback) ();
 760
 761                 if (status != 0)
 762                 {
 763                         ERROR ("Initialization of plugin `%s' "
 764                                         "failed with status %i. "
 765                                         "Plugin will be unloaded.",
 766                                         le->key, status);
 767                         /* Plugins that register read callbacks from the init
 768                          * callback should take care of appropriate error
 769                          * handling themselves. */
 770                         /* FIXME: Unload _all_ functions */
 771                         plugin_unregister_read (le->key);
 772                 }
 773
 774                 le = le->next;
 775         }
 776
 777         /* Start read-threads */
 778         if (list_read != NULL)
 779         {
 780                 const char *rt;
 781                 int num;
 782                 rt = global_option_get ("ReadThreads");
 783                 num = atoi (rt);
 784                 if (num != -1)
 785                         start_read_threads ((num > 0) ? num : 5);
 786         }
 787 } /* void plugin_init_all */
 788
 789 void plugin_read_all (void)
 790 {
 791         llentry_t   *le;
 792         read_func_t *rf;
 793
 794         uc_check_timeout ();
 795
 796         if (list_read == NULL)
 797                 return;
 798
 799         pthread_mutex_lock (&read_lock);
 800
 801         le = llist_head (list_read);
 802         while (le != NULL)
 803         {
 804                 rf = (read_func_t *) le->value;
 805
 806                 if (rf->needs_read != DONE)
 807                 {
 808                         le = le->next;
 809                         continue;
 810                 }
 811
 812                 if (rf->wait_left > 0)
 813                         rf->wait_left -= interval_g;
 814
 815                 if (rf->wait_left <= 0)
 816                 {
 817                         rf->needs_read = TODO;
 818                 }
 819
 820                 le = le->next;
 821         }
 822
 823         DEBUG ("plugin: plugin_read_all: Signalling `read_cond'");
 824         pthread_cond_broadcast (&read_cond);
 825         pthread_mutex_unlock (&read_lock);
 826 } /* void plugin_read_all */
 827
 828 /* Read function called when the `-T' command line argument is given. */
 829 int plugin_read_all_once (void)
 830 {
 831         llentry_t   *le;
 832         read_func_t *rf;
 833         int status;
 834         int return_status = 0;
 835
 836         if (list_read == NULL)
 837         {
 838                 NOTICE ("No read-functions are registered.");
 839                 return (0);
 840         }
 841
 842         for (le = llist_head (list_read);
 843              le != NULL;
 844              le = le->next)
 845         {
 846                 rf = (read_func_t *) le->value;
 847
 848                 if (rf->type == RF_SIMPLE)
 849                 {
 850                         status = rf->callback.cb_simple ();
 851                 }
 852                 else
 853                 {
 854                         assert (rf->type == RF_COMPLEX);
 855                         status = rf->callback.cb_complex (&rf->udata);
 856                 }
 857
 858                 if (status != 0)
 859                 {
 860                         NOTICE ("read-function of plugin `%s' failed.",
 861                                 le->key);
 862                         return_status = -1;
 863                 }
 864         }
 865
 866         return (return_status);
 867 } /* int plugin_read_all_once */
 868
 869 int plugin_write (const char *plugin, /* {{{ */
 870                 const data_set_t *ds, const value_list_t *vl)
 871 {
 872   llentry_t *le;
 873   int status;
 874
 875   if (vl == NULL)
 876     return (EINVAL);
 877
 878   if (list_write == NULL)
 879     return (ENOENT);
 880
 881   if (ds == NULL)
 882   {
 883     ds = plugin_get_ds (vl->type);
 884     if (ds == NULL)
 885     {
 886       ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type);
 887       return (ENOENT);
 888     }
 889   }
 890
 891   if (plugin == NULL)
 892   {
 893     int success = 0;
 894     int failure = 0;
 895
 896     le = llist_head (list_write);
 897     while (le != NULL)
 898     {
 899       write_func_t *wf = le->value;
 900
 901       DEBUG ("plugin: plugin_write: Writing values via %s.", le->key);
 902       status = wf->callback (ds, vl, &wf->udata);
 903       if (status != 0)
 904         failure++;
 905       else
 906         success++;
 907
 908       le = le->next;
 909     }
 910
 911     if ((success == 0) && (failure != 0))
 912       status = -1;
 913     else
 914       status = 0;
 915   }
 916   else /* plugin != NULL */
 917   {
 918     write_func_t *wf;
 919     le = llist_head (list_write);
 920     while (le != NULL)
 921     {
 922       if (strcasecmp (plugin, le->key) == 0)
 923         break;
 924
 925       le = le->next;
 926     }
 927
 928     if (le == NULL)
 929       return (ENOENT);
 930
 931     wf = le->value;
 932
 933     DEBUG ("plugin: plugin_write: Writing values via %s.", le->key);
 934     status = wf->callback (ds, vl, &wf->udata);
 935   }
 936
 937   return (status);
 938 } /* }}} int plugin_write */
 939
 940 int plugin_flush (const char *plugin, int timeout, const char *identifier)
 941 {
 942   int (*callback) (int timeout, const char *identifier);
 943   llentry_t *le;
 944
 945   if (list_flush == NULL)
 946     return (0);
 947
 948   le = llist_head (list_flush);
 949   while (le != NULL)
 950   {
 951     if ((plugin != NULL)
 952         && (strcmp (plugin, le->key) != 0))
 953     {
 954       le = le->next;
 955       continue;
 956     }
 957
 958     callback = (int (*) (int, const char *)) le->value;
 959     (*callback) (timeout, identifier);
 960
 961     le = le->next;
 962   }
 963   return (0);
 964 } /* int plugin_flush */
 965
 966 void plugin_shutdown_all (void)
 967 {
 968         int (*callback) (void);
 969         llentry_t *le;
 970
 971         stop_read_threads ();
 972         remove_read_functions ();
 973
 974         if (list_shutdown == NULL)
 975                 return;
 976
 977         le = llist_head (list_shutdown);
 978         while (le != NULL)
 979         {
 980                 callback = (int (*) (void)) le->value;
 981
 982                 /* Advance the pointer before calling the callback allows
 983                  * shutdown functions to unregister themselves. If done the
 984                  * other way around the memory `le' points to will be freed
 985                  * after callback returns. */
 986                 le = le->next;
 987
 988                 (*callback) ();
 989         }
 990 } /* void plugin_shutdown_all */
 991
 992 int plugin_dispatch_values (value_list_t *vl)
 993 {
 994         int status;
 995         static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
 996
 997         data_set_t *ds;
 998
 999         if ((vl == NULL) || (*vl->type == '\0')) {
1000                 ERROR ("plugin_dispatch_values: Invalid value list.");
1001                 return (-1);
1002         }
1003
1004         if (list_write == NULL)
1005                 c_complain_once (LOG_WARNING, &no_write_complaint,
1006                                 "plugin_dispatch_values: No write callback has been "
1007                                 "registered. Please load at least one output plugin, "
1008                                 "if you want the collected data to be stored.");
1009
1010         if (data_sets == NULL)
1011         {
1012                 ERROR ("plugin_dispatch_values: No data sets registered. "
1013                                 "Could the types database be read? Check "
1014                                 "your `TypesDB' setting!");
1015                 return (-1);
1016         }
1017
1018         if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0)
1019         {
1020                 INFO ("plugin_dispatch_values: Dataset not found: %s", vl->type);
1021                 return (-1);
1022         }
1023
1024         if (vl->time == 0)
1025                 vl->time = time (NULL);
1026
1027         DEBUG ("plugin_dispatch_values: time = %u; interval = %i; "
1028                         "host = %s; "
1029                         "plugin = %s; plugin_instance = %s; "
1030                         "type = %s; type_instance = %s;",
1031                         (unsigned int) vl->time, vl->interval,
1032                         vl->host,
1033                         vl->plugin, vl->plugin_instance,
1034                         vl->type, vl->type_instance);
1035
1036 #if COLLECT_DEBUG
1037         assert (0 == strcmp (ds->type, vl->type));
1038 #else
1039         if (0 != strcmp (ds->type, vl->type))
1040                 WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)",
1041                                 ds->type, vl->type);
1042 #endif
1043
1044 #if COLLECT_DEBUG
1045         assert (ds->ds_num == vl->values_len);
1046 #else
1047         if (ds->ds_num != vl->values_len)
1048         {
1049                 ERROR ("plugin_dispatch_values: ds->type = %s: "
1050                                 "(ds->ds_num = %i) != "
1051                                 "(vl->values_len = %i)",
1052                                 ds->type, ds->ds_num, vl->values_len);
1053                 return (-1);
1054         }
1055 #endif
1056
1057         escape_slashes (vl->host, sizeof (vl->host));
1058         escape_slashes (vl->plugin, sizeof (vl->plugin));
1059         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
1060         escape_slashes (vl->type, sizeof (vl->type));
1061         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
1062
1063         if (pre_cache_chain != NULL)
1064         {
1065                 status = fc_process_chain (ds, vl, pre_cache_chain);
1066                 if (status < 0)
1067                 {
1068                         WARNING ("plugin_dispatch_values: Running the "
1069                                         "pre-cache chain failed with "
1070                                         "status %i (%#x).",
1071                                         status, status);
1072                 }
1073                 else if (status == FC_TARGET_STOP)
1074                         return (0);
1075         }
1076
1077         /* Update the value cache */
1078         uc_update (ds, vl);
1079
1080         if (post_cache_chain != NULL)
1081         {
1082                 status = fc_process_chain (ds, vl, post_cache_chain);
1083                 if (status < 0)
1084                 {
1085                         WARNING ("plugin_dispatch_values: Running the "
1086                                         "post-cache chain failed with "
1087                                         "status %i (%#x).",
1088                                         status, status);
1089                 }
1090         }
1091         else
1092                 fc_default_action (ds, vl);
1093
1094         return (0);
1095 } /* int plugin_dispatch_values */
1096
1097 int plugin_dispatch_notification (const notification_t *notif)
1098 {
1099         int (*callback) (const notification_t *);
1100         llentry_t *le;
1101         /* Possible TODO: Add flap detection here */
1102
1103         DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; "
1104                         "time = %u; host = %s;",
1105                         notif->severity, notif->message,
1106                         (unsigned int) notif->time, notif->host);
1107
1108         /* Nobody cares for notifications */
1109         if (list_notification == NULL)
1110                 return (-1);
1111
1112         le = llist_head (list_notification);
1113         while (le != NULL)
1114         {
1115                 callback = (int (*) (const notification_t *)) le->value;
1116                 (*callback) (notif);
1117
1118                 le = le->next;
1119         }
1120
1121         return (0);
1122 } /* int plugin_dispatch_notification */
1123
1124 void plugin_log (int level, const char *format, ...)
1125 {
1126         char msg[1024];
1127         va_list ap;
1128
1129         void (*callback) (int, const char *);
1130         llentry_t *le;
1131
1132         if (list_log == NULL)
1133                 return;
1134
1135 #if !COLLECT_DEBUG
1136         if (level >= LOG_DEBUG)
1137                 return;
1138 #endif
1139
1140         va_start (ap, format);
1141         vsnprintf (msg, sizeof (msg), format, ap);
1142         msg[sizeof (msg) - 1] = '\0';
1143         va_end (ap);
1144
1145         le = llist_head (list_log);
1146         while (le != NULL)
1147         {
1148                 callback = (void (*) (int, const char *)) le->value;
1149                 (*callback) (level, msg);
1150
1151                 le = le->next;
1152         }
1153 } /* void plugin_log */
1154
1155 const data_set_t *plugin_get_ds (const char *name)
1156 {
1157         data_set_t *ds;
1158
1159         if (c_avl_get (data_sets, name, (void *) &ds) != 0)
1160         {
1161                 DEBUG ("No such dataset registered: %s", name);
1162                 return (NULL);
1163         }
1164
1165         return (ds);
1166 } /* data_set_t *plugin_get_ds */
1167
1168 static int plugin_notification_meta_add (notification_t *n,
1169     const char *name,
1170     enum notification_meta_type_e type,
1171     const void *value)
1172 {
1173   notification_meta_t *meta;
1174   notification_meta_t *tail;
1175
1176   if ((n == NULL) || (name == NULL) || (value == NULL))
1177   {
1178     ERROR ("plugin_notification_meta_add: A pointer is NULL!");
1179     return (-1);
1180   }
1181
1182   meta = (notification_meta_t *) malloc (sizeof (notification_meta_t));
1183   if (meta == NULL)
1184   {
1185     ERROR ("plugin_notification_meta_add: malloc failed.");
1186     return (-1);
1187   }
1188   memset (meta, 0, sizeof (notification_meta_t));
1189
1190   sstrncpy (meta->name, name, sizeof (meta->name));
1191   meta->type = type;
1192
1193   switch (type)
1194   {
1195     case NM_TYPE_STRING:
1196     {
1197       meta->nm_value.nm_string = strdup ((const char *) value);
1198       if (meta->nm_value.nm_string == NULL)
1199       {
1200         ERROR ("plugin_notification_meta_add: strdup failed.");
1201         sfree (meta);
1202         return (-1);
1203       }
1204       break;
1205     }
1206     case NM_TYPE_SIGNED_INT:
1207     {
1208       meta->nm_value.nm_signed_int = *((int64_t *) value);
1209       break;
1210     }
1211     case NM_TYPE_UNSIGNED_INT:
1212     {
1213       meta->nm_value.nm_unsigned_int = *((uint64_t *) value);
1214       break;
1215     }
1216     case NM_TYPE_DOUBLE:
1217     {
1218       meta->nm_value.nm_double = *((double *) value);
1219       break;
1220     }
1221     case NM_TYPE_BOOLEAN:
1222     {
1223       meta->nm_value.nm_boolean = *((bool *) value);
1224       break;
1225     }
1226     default:
1227     {
1228       ERROR ("plugin_notification_meta_add: Unknown type: %i", type);
1229       sfree (meta);
1230       return (-1);
1231     }
1232   } /* switch (type) */
1233
1234   meta->next = NULL;
1235   tail = n->meta;
1236   while ((tail != NULL) && (tail->next != NULL))
1237     tail = tail->next;
1238
1239   if (tail == NULL)
1240     n->meta = meta;
1241   else
1242     tail->next = meta;
1243
1244   return (0);
1245 } /* int plugin_notification_meta_add */
1246
1247 int plugin_notification_meta_add_string (notification_t *n,
1248     const char *name,
1249     const char *value)
1250 {
1251   return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value));
1252 }
1253
1254 int plugin_notification_meta_add_signed_int (notification_t *n,
1255     const char *name,
1256     int64_t value)
1257 {
1258   return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value));
1259 }
1260
1261 int plugin_notification_meta_add_unsigned_int (notification_t *n,
1262     const char *name,
1263     uint64_t value)
1264 {
1265   return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value));
1266 }
1267
1268 int plugin_notification_meta_add_double (notification_t *n,
1269     const char *name,
1270     double value)
1271 {
1272   return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value));
1273 }
1274
1275 int plugin_notification_meta_add_boolean (notification_t *n,
1276     const char *name,
1277     bool value)
1278 {
1279   return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value));
1280 }
1281
1282 int plugin_notification_meta_copy (notification_t *dst,
1283     const notification_t *src)
1284 {
1285   notification_meta_t *meta;
1286
1287   assert (dst != NULL);
1288   assert (src != NULL);
1289   assert (dst != src);
1290   assert ((src->meta == NULL) || (src->meta != dst->meta));
1291
1292   for (meta = src->meta; meta != NULL; meta = meta->next)
1293   {
1294     if (meta->type == NM_TYPE_STRING)
1295       plugin_notification_meta_add_string (dst, meta->name,
1296           meta->nm_value.nm_string);
1297     else if (meta->type == NM_TYPE_SIGNED_INT)
1298       plugin_notification_meta_add_signed_int (dst, meta->name,
1299           meta->nm_value.nm_signed_int);
1300     else if (meta->type == NM_TYPE_UNSIGNED_INT)
1301       plugin_notification_meta_add_unsigned_int (dst, meta->name,
1302           meta->nm_value.nm_unsigned_int);
1303     else if (meta->type == NM_TYPE_DOUBLE)
1304       plugin_notification_meta_add_double (dst, meta->name,
1305           meta->nm_value.nm_double);
1306     else if (meta->type == NM_TYPE_BOOLEAN)
1307       plugin_notification_meta_add_boolean (dst, meta->name,
1308           meta->nm_value.nm_boolean);
1309   }
1310
1311   return (0);
1312 } /* int plugin_notification_meta_copy */
1313
1314 int plugin_notification_meta_free (notification_meta_t *n)
1315 {
1316   notification_meta_t *this;
1317   notification_meta_t *next;
1318
1319   if (n == NULL)
1320   {
1321     ERROR ("plugin_notification_meta_free: n == NULL!");
1322     return (-1);
1323   }
1324
1325   this = n;
1326   while (this != NULL)
1327   {
1328     next = this->next;
1329
1330     if (this->type == NM_TYPE_STRING)
1331     {
1332       free ((char *)this->nm_value.nm_string);
1333       this->nm_value.nm_string = NULL;
1334     }
1335     sfree (this);
1336
1337     this = next;
1338   }
1339
1340   return (0);
1341 } /* int plugin_notification_meta_free */