src/plugin.c

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