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 int plugin_read_all_once (void)
 686 {
 687         llentry_t   *le;
 688         read_func_t *rf;
 689         int status;
 690
 691         if (list_read == NULL)
 692                 return (0);
 693
 694         for (le = llist_head (list_read);
 695              le != NULL;
 696              le = le->next)
 697         {
 698                 rf = (read_func_t *) le->value;
 699                 status = rf->callback ();
 700                 if (status != 0)
 701                 {
 702                         NOTICE ("read-function of plugin `%s' failed.",
 703                                 le->key);
 704                         return status;
 705                 }
 706         }
 707
 708         return (0);
 709 } /* void plugin_read_all_once */
 710
 711 int plugin_write (const char *plugin, /* {{{ */
 712                 const data_set_t *ds, const value_list_t *vl)
 713 {
 714   int (*callback) (const data_set_t *ds, const value_list_t *vl);
 715   llentry_t *le;
 716   int status;
 717
 718   if (vl == NULL)
 719     return (EINVAL);
 720
 721   if (list_write == NULL)
 722     return (ENOENT);
 723
 724   if (ds == NULL)
 725   {
 726     ds = plugin_get_ds (vl->type);
 727     if (ds == NULL)
 728     {
 729       ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type);
 730       return (ENOENT);
 731     }
 732   }
 733
 734   if (plugin == NULL)
 735   {
 736     int success = 0;
 737     int failure = 0;
 738
 739     le = llist_head (list_write);
 740     while (le != NULL)
 741     {
 742       callback = le->value;
 743       status = (*callback) (ds, vl);
 744       if (status != 0)
 745         failure++;
 746       else
 747         success++;
 748
 749       le = le->next;
 750     }
 751
 752     if ((success == 0) && (failure != 0))
 753       status = -1;
 754     else
 755       status = 0;
 756   }
 757   else /* plugin != NULL */
 758   {
 759     le = llist_head (list_write);
 760     while (le != NULL)
 761     {
 762       if (strcasecmp (plugin, le->key) == 0)
 763         break;
 764
 765       le = le->next;
 766     }
 767
 768     if (le == NULL)
 769       return (ENOENT);
 770
 771     callback = le->value;
 772     status = (*callback) (ds, vl);
 773   }
 774
 775   return (status);
 776 } /* }}} int plugin_write */
 777
 778 int plugin_flush (const char *plugin, int timeout, const char *identifier)
 779 {
 780   int (*callback) (int timeout, const char *identifier);
 781   llentry_t *le;
 782
 783   if (list_flush == NULL)
 784     return (0);
 785
 786   le = llist_head (list_flush);
 787   while (le != NULL)
 788   {
 789     if ((plugin != NULL)
 790         && (strcmp (plugin, le->key) != 0))
 791     {
 792       le = le->next;
 793       continue;
 794     }
 795
 796     callback = (int (*) (int, const char *)) le->value;
 797     (*callback) (timeout, identifier);
 798
 799     le = le->next;
 800   }
 801   return (0);
 802 } /* int plugin_flush */
 803
 804 void plugin_shutdown_all (void)
 805 {
 806         int (*callback) (void);
 807         llentry_t *le;
 808
 809         stop_threads ();
 810
 811         if (list_shutdown == NULL)
 812                 return;
 813
 814         le = llist_head (list_shutdown);
 815         while (le != NULL)
 816         {
 817                 callback = (int (*) (void)) le->value;
 818
 819                 /* Advance the pointer before calling the callback allows
 820                  * shutdown functions to unregister themselves. If done the
 821                  * other way around the memory `le' points to will be freed
 822                  * after callback returns. */
 823                 le = le->next;
 824
 825                 (*callback) ();
 826         }
 827 } /* void plugin_shutdown_all */
 828
 829 int plugin_dispatch_values (value_list_t *vl)
 830 {
 831         static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
 832
 833         data_set_t *ds;
 834
 835         if ((vl == NULL) || (*vl->type == '\0')) {
 836                 ERROR ("plugin_dispatch_values: Invalid value list.");
 837                 return (-1);
 838         }
 839
 840         if (list_write == NULL)
 841                 c_complain_once (LOG_WARNING, &no_write_complaint,
 842                                 "plugin_dispatch_values: No write callback has been "
 843                                 "registered. Please load at least one output plugin, "
 844                                 "if you want the collected data to be stored.");
 845
 846         if (data_sets == NULL)
 847         {
 848                 ERROR ("plugin_dispatch_values: No data sets registered. "
 849                                 "Could the types database be read? Check "
 850                                 "your `TypesDB' setting!");
 851                 return (-1);
 852         }
 853
 854         if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0)
 855         {
 856                 INFO ("plugin_dispatch_values: Dataset not found: %s", vl->type);
 857                 return (-1);
 858         }
 859
 860         DEBUG ("plugin_dispatch_values: time = %u; interval = %i; "
 861                         "host = %s; "
 862                         "plugin = %s; plugin_instance = %s; "
 863                         "type = %s; type_instance = %s;",
 864                         (unsigned int) vl->time, vl->interval,
 865                         vl->host,
 866                         vl->plugin, vl->plugin_instance,
 867                         vl->type, vl->type_instance);
 868
 869 #if COLLECT_DEBUG
 870         assert (0 == strcmp (ds->type, vl->type));
 871 #else
 872         if (0 != strcmp (ds->type, vl->type))
 873                 WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)",
 874                                 ds->type, vl->type);
 875 #endif
 876
 877 #if COLLECT_DEBUG
 878         assert (ds->ds_num == vl->values_len);
 879 #else
 880         if (ds->ds_num != vl->values_len)
 881         {
 882                 ERROR ("plugin_dispatch_values: ds->type = %s: "
 883                                 "(ds->ds_num = %i) != "
 884                                 "(vl->values_len = %i)",
 885                                 ds->type, ds->ds_num, vl->values_len);
 886                 return (-1);
 887         }
 888 #endif
 889
 890         escape_slashes (vl->host, sizeof (vl->host));
 891         escape_slashes (vl->plugin, sizeof (vl->plugin));
 892         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
 893         escape_slashes (vl->type, sizeof (vl->type));
 894         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
 895
 896         /* Update the value cache */
 897         uc_update (ds, vl);
 898
 899         fc_process (ds, vl);
 900
 901         return (0);
 902 } /* int plugin_dispatch_values */
 903
 904 int plugin_dispatch_notification (const notification_t *notif)
 905 {
 906         int (*callback) (const notification_t *);
 907         llentry_t *le;
 908         /* Possible TODO: Add flap detection here */
 909
 910         DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; "
 911                         "time = %u; host = %s;",
 912                         notif->severity, notif->message,
 913                         (unsigned int) notif->time, notif->host);
 914
 915         /* Nobody cares for notifications */
 916         if (list_notification == NULL)
 917                 return (-1);
 918
 919         le = llist_head (list_notification);
 920         while (le != NULL)
 921         {
 922                 callback = (int (*) (const notification_t *)) le->value;
 923                 (*callback) (notif);
 924
 925                 le = le->next;
 926         }
 927
 928         return (0);
 929 } /* int plugin_dispatch_notification */
 930
 931 void plugin_log (int level, const char *format, ...)
 932 {
 933         char msg[1024];
 934         va_list ap;
 935
 936         void (*callback) (int, const char *);
 937         llentry_t *le;
 938
 939         if (list_log == NULL)
 940                 return;
 941
 942 #if !COLLECT_DEBUG
 943         if (level >= LOG_DEBUG)
 944                 return;
 945 #endif
 946
 947         va_start (ap, format);
 948         vsnprintf (msg, sizeof (msg), format, ap);
 949         msg[sizeof (msg) - 1] = '\0';
 950         va_end (ap);
 951
 952         le = llist_head (list_log);
 953         while (le != NULL)
 954         {
 955                 callback = (void (*) (int, const char *)) le->value;
 956                 (*callback) (level, msg);
 957
 958                 le = le->next;
 959         }
 960 } /* void plugin_log */
 961
 962 const data_set_t *plugin_get_ds (const char *name)
 963 {
 964         data_set_t *ds;
 965
 966         if (c_avl_get (data_sets, name, (void *) &ds) != 0)
 967         {
 968                 DEBUG ("No such dataset registered: %s", name);
 969                 return (NULL);
 970         }
 971
 972         return (ds);
 973 } /* data_set_t *plugin_get_ds */
 974
 975 static int plugin_notification_meta_add (notification_t *n,
 976     const char *name,
 977     enum notification_meta_type_e type,
 978     const void *value)
 979 {
 980   notification_meta_t *meta;
 981   notification_meta_t *tail;
 982
 983   if ((n == NULL) || (name == NULL) || (value == NULL))
 984   {
 985     ERROR ("plugin_notification_meta_add: A pointer is NULL!");
 986     return (-1);
 987   }
 988
 989   meta = (notification_meta_t *) malloc (sizeof (notification_meta_t));
 990   if (meta == NULL)
 991   {
 992     ERROR ("plugin_notification_meta_add: malloc failed.");
 993     return (-1);
 994   }
 995   memset (meta, 0, sizeof (notification_meta_t));
 996
 997   sstrncpy (meta->name, name, sizeof (meta->name));
 998   meta->type = type;
 999
1000   switch (type)
1001   {
1002     case NM_TYPE_STRING:
1003     {
1004       meta->nm_value.nm_string = strdup ((const char *) value);
1005       if (meta->nm_value.nm_string == NULL)
1006       {
1007         ERROR ("plugin_notification_meta_add: strdup failed.");
1008         sfree (meta);
1009         return (-1);
1010       }
1011       break;
1012     }
1013     case NM_TYPE_SIGNED_INT:
1014     {
1015       meta->nm_value.nm_signed_int = *((int64_t *) value);
1016       break;
1017     }
1018     case NM_TYPE_UNSIGNED_INT:
1019     {
1020       meta->nm_value.nm_unsigned_int = *((uint64_t *) value);
1021       break;
1022     }
1023     case NM_TYPE_DOUBLE:
1024     {
1025       meta->nm_value.nm_double = *((double *) value);
1026       break;
1027     }
1028     case NM_TYPE_BOOLEAN:
1029     {
1030       meta->nm_value.nm_boolean = *((bool *) value);
1031       break;
1032     }
1033     default:
1034     {
1035       ERROR ("plugin_notification_meta_add: Unknown type: %i", type);
1036       sfree (meta);
1037       return (-1);
1038     }
1039   } /* switch (type) */
1040
1041   meta->next = NULL;
1042   tail = n->meta;
1043   while ((tail != NULL) && (tail->next != NULL))
1044     tail = tail->next;
1045
1046   if (tail == NULL)
1047     n->meta = meta;
1048   else
1049     tail->next = meta;
1050
1051   return (0);
1052 } /* int plugin_notification_meta_add */
1053
1054 int plugin_notification_meta_add_string (notification_t *n,
1055     const char *name,
1056     const char *value)
1057 {
1058   return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value));
1059 }
1060
1061 int plugin_notification_meta_add_signed_int (notification_t *n,
1062     const char *name,
1063     int64_t value)
1064 {
1065   return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value));
1066 }
1067
1068 int plugin_notification_meta_add_unsigned_int (notification_t *n,
1069     const char *name,
1070     uint64_t value)
1071 {
1072   return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value));
1073 }
1074
1075 int plugin_notification_meta_add_double (notification_t *n,
1076     const char *name,
1077     double value)
1078 {
1079   return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value));
1080 }
1081
1082 int plugin_notification_meta_add_boolean (notification_t *n,
1083     const char *name,
1084     bool value)
1085 {
1086   return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value));
1087 }
1088
1089 int plugin_notification_meta_copy (notification_t *dst,
1090     const notification_t *src)
1091 {
1092   notification_meta_t *meta;
1093
1094   assert (dst != NULL);
1095   assert (src != NULL);
1096   assert (dst != src);
1097   assert ((src->meta == NULL) || (src->meta != dst->meta));
1098
1099   for (meta = src->meta; meta != NULL; meta = meta->next)
1100   {
1101     if (meta->type == NM_TYPE_STRING)
1102       plugin_notification_meta_add_string (dst, meta->name,
1103           meta->nm_value.nm_string);
1104     else if (meta->type == NM_TYPE_SIGNED_INT)
1105       plugin_notification_meta_add_signed_int (dst, meta->name,
1106           meta->nm_value.nm_signed_int);
1107     else if (meta->type == NM_TYPE_UNSIGNED_INT)
1108       plugin_notification_meta_add_unsigned_int (dst, meta->name,
1109           meta->nm_value.nm_unsigned_int);
1110     else if (meta->type == NM_TYPE_DOUBLE)
1111       plugin_notification_meta_add_double (dst, meta->name,
1112           meta->nm_value.nm_double);
1113     else if (meta->type == NM_TYPE_BOOLEAN)
1114       plugin_notification_meta_add_boolean (dst, meta->name,
1115           meta->nm_value.nm_boolean);
1116   }
1117
1118   return (0);
1119 } /* int plugin_notification_meta_copy */
1120
1121 int plugin_notification_meta_free (notification_t *n)
1122 {
1123   notification_meta_t *this;
1124   notification_meta_t *next;
1125
1126   if (n == NULL)
1127   {
1128     ERROR ("plugin_notification_meta_free: n == NULL!");
1129     return (-1);
1130   }
1131
1132   this = n->meta;
1133   n->meta = NULL;
1134   while (this != NULL)
1135   {
1136     next = this->next;
1137
1138     if (this->type == NM_TYPE_STRING)
1139     {
1140       free ((char *)this->nm_value.nm_string);
1141       this->nm_value.nm_string = NULL;
1142     }
1143     sfree (this);
1144
1145     this = next;
1146   }
1147
1148   return (0);
1149 } /* int plugin_notification_meta_free */