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         pthread_mutex_lock (&read_lock);
 280         read_loop = 0;
 281         DEBUG ("plugin: stop_threads: Signalling `read_cond'");
 282         pthread_cond_broadcast (&read_cond);
 283         pthread_mutex_unlock (&read_lock);
 284
 285         for (i = 0; i < read_threads_num; i++)
 286         {
 287                 if (pthread_join (read_threads[i], NULL) != 0)
 288                 {
 289                         ERROR ("plugin: stop_threads: pthread_join failed.");
 290                 }
 291                 read_threads[i] = (pthread_t) 0;
 292         }
 293         sfree (read_threads);
 294         read_threads_num = 0;
 295 } /* void stop_threads */
 296
 297 /*
 298  * Public functions
 299  */
 300 void plugin_set_dir (const char *dir)
 301 {
 302         if (plugindir != NULL)
 303                 free (plugindir);
 304
 305         if (dir == NULL)
 306                 plugindir = NULL;
 307         else if ((plugindir = strdup (dir)) == NULL)
 308         {
 309                 char errbuf[1024];
 310                 ERROR ("strdup failed: %s",
 311                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 312         }
 313 }
 314
 315 #define BUFSIZE 512
 316 int plugin_load (const char *type)
 317 {
 318         DIR  *dh;
 319         const char *dir;
 320         char  filename[BUFSIZE] = "";
 321         char  typename[BUFSIZE];
 322         int   typename_len;
 323         int   ret;
 324         struct stat    statbuf;
 325         struct dirent *de;
 326
 327         DEBUG ("type = %s", type);
 328
 329         dir = plugin_get_dir ();
 330         ret = 1;
 331
 332         /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
 333          * type when matching the filename */
 334         if (ssnprintf (typename, sizeof (typename),
 335                         "%s.so", type) >= sizeof (typename))
 336         {
 337                 WARNING ("snprintf: truncated: `%s.so'", type);
 338                 return (-1);
 339         }
 340         typename_len = strlen (typename);
 341
 342         if ((dh = opendir (dir)) == NULL)
 343         {
 344                 char errbuf[1024];
 345                 ERROR ("opendir (%s): %s", dir,
 346                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 347                 return (-1);
 348         }
 349
 350         while ((de = readdir (dh)) != NULL)
 351         {
 352                 if (strncasecmp (de->d_name, typename, typename_len))
 353                         continue;
 354
 355                 if (ssnprintf (filename, sizeof (filename),
 356                                 "%s/%s", dir, de->d_name) >= sizeof (filename))
 357                 {
 358                         WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name);
 359                         continue;
 360                 }
 361
 362                 if (lstat (filename, &statbuf) == -1)
 363                 {
 364                         char errbuf[1024];
 365                         WARNING ("stat %s: %s", filename,
 366                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 367                         continue;
 368                 }
 369                 else if (!S_ISREG (statbuf.st_mode))
 370                 {
 371                         /* don't follow symlinks */
 372                         continue;
 373                 }
 374
 375                 if (plugin_load_file (filename) == 0)
 376                 {
 377                         /* success */
 378                         ret = 0;
 379                         break;
 380                 }
 381                 else
 382                 {
 383                         fprintf (stderr, "Unable to load plugin %s.\n", type);
 384                 }
 385         }
 386
 387         closedir (dh);
 388
 389         if (filename[0] == '\0')
 390                 fprintf (stderr, "Could not find plugin %s.\n", type);
 391
 392         return (ret);
 393 }
 394
 395 /*
 396  * The `register_*' functions follow
 397  */
 398 int plugin_register_config (const char *name,
 399                 int (*callback) (const char *key, const char *val),
 400                 const char **keys, int keys_num)
 401 {
 402         cf_register (name, callback, keys, keys_num);
 403         return (0);
 404 } /* int plugin_register_config */
 405
 406 int plugin_register_complex_config (const char *type,
 407                 int (*callback) (oconfig_item_t *))
 408 {
 409         return (cf_register_complex (type, callback));
 410 } /* int plugin_register_complex_config */
 411
 412 int plugin_register_init (const char *name,
 413                 int (*callback) (void))
 414 {
 415         return (register_callback (&list_init, name, (void *) callback));
 416 } /* plugin_register_init */
 417
 418 int plugin_register_read (const char *name,
 419                 int (*callback) (void))
 420 {
 421         read_func_t *rf;
 422
 423         rf = (read_func_t *) malloc (sizeof (read_func_t));
 424         if (rf == NULL)
 425         {
 426                 char errbuf[1024];
 427                 ERROR ("plugin_register_read: malloc failed: %s",
 428                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 429                 return (-1);
 430         }
 431
 432         memset (rf, '\0', sizeof (read_func_t));
 433         rf->wait_time = interval_g;
 434         rf->wait_left = 0;
 435         rf->callback = callback;
 436         rf->needs_read = DONE;
 437
 438         return (register_callback (&list_read, name, (void *) rf));
 439 } /* int plugin_register_read */
 440
 441 int plugin_register_write (const char *name,
 442                 int (*callback) (const data_set_t *ds, const value_list_t *vl))
 443 {
 444         return (register_callback (&list_write, name, (void *) callback));
 445 } /* int plugin_register_write */
 446
 447 int plugin_register_filter (const char *name,
 448                 int (*callback) (const data_set_t *ds, value_list_t *vl))
 449 {
 450         return (register_callback (&list_filter, name, (void *) callback));
 451 } /* int plugin_register_filter */
 452
 453 int plugin_register_flush (const char *name,
 454                 int (*callback) (const int timeout, const char *identifier))
 455 {
 456         return (register_callback (&list_flush, name, (void *) callback));
 457 } /* int plugin_register_flush */
 458
 459 int plugin_register_shutdown (char *name,
 460                 int (*callback) (void))
 461 {
 462         return (register_callback (&list_shutdown, name, (void *) callback));
 463 } /* int plugin_register_shutdown */
 464
 465 int plugin_register_data_set (const data_set_t *ds)
 466 {
 467         data_set_t *ds_copy;
 468         int i;
 469
 470         if ((data_sets != NULL)
 471                         && (c_avl_get (data_sets, ds->type, NULL) == 0))
 472         {
 473                 NOTICE ("Replacing DS `%s' with another version.", ds->type);
 474                 plugin_unregister_data_set (ds->type);
 475         }
 476         else if (data_sets == NULL)
 477         {
 478                 data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp);
 479                 if (data_sets == NULL)
 480                         return (-1);
 481         }
 482
 483         ds_copy = (data_set_t *) malloc (sizeof (data_set_t));
 484         if (ds_copy == NULL)
 485                 return (-1);
 486         memcpy(ds_copy, ds, sizeof (data_set_t));
 487
 488         ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t)
 489                         * ds->ds_num);
 490         if (ds_copy->ds == NULL)
 491         {
 492                 free (ds_copy);
 493                 return (-1);
 494         }
 495
 496         for (i = 0; i < ds->ds_num; i++)
 497                 memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t));
 498
 499         return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy));
 500 } /* int plugin_register_data_set */
 501
 502 int plugin_register_log (char *name,
 503                 void (*callback) (int priority, const char *msg))
 504 {
 505         return (register_callback (&list_log, name, (void *) callback));
 506 } /* int plugin_register_log */
 507
 508 int plugin_register_notification (const char *name,
 509                 int (*callback) (const notification_t *notif))
 510 {
 511         return (register_callback (&list_notification, name, (void *) callback));
 512 } /* int plugin_register_log */
 513
 514 int plugin_unregister_config (const char *name)
 515 {
 516         cf_unregister (name);
 517         return (0);
 518 } /* int plugin_unregister_config */
 519
 520 int plugin_unregister_complex_config (const char *name)
 521 {
 522         cf_unregister_complex (name);
 523         return (0);
 524 } /* int plugin_unregister_complex_config */
 525
 526 int plugin_unregister_init (const char *name)
 527 {
 528         return (plugin_unregister (list_init, name));
 529 }
 530
 531 int plugin_unregister_read (const char *name)
 532 {
 533         llentry_t *e;
 534
 535         e = llist_search (list_read, name);
 536
 537         if (e == NULL)
 538                 return (-1);
 539
 540         llist_remove (list_read, e);
 541         free (e->value);
 542         free (e->key);
 543         llentry_destroy (e);
 544
 545         return (0);
 546 }
 547
 548 int plugin_unregister_write (const char *name)
 549 {
 550         return (plugin_unregister (list_write, name));
 551 }
 552
 553 int plugin_unregister_filter (const char *name)
 554 {
 555         return (plugin_unregister (list_filter, name));
 556 }
 557
 558 int plugin_unregister_flush (const char *name)
 559 {
 560         return (plugin_unregister (list_flush, name));
 561 }
 562
 563 int plugin_unregister_shutdown (const char *name)
 564 {
 565         return (plugin_unregister (list_shutdown, name));
 566 }
 567
 568 int plugin_unregister_data_set (const char *name)
 569 {
 570         data_set_t *ds;
 571
 572         if (data_sets == NULL)
 573                 return (-1);
 574
 575         if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0)
 576                 return (-1);
 577
 578         sfree (ds->ds);
 579         sfree (ds);
 580
 581         return (0);
 582 } /* int plugin_unregister_data_set */
 583
 584 int plugin_unregister_log (const char *name)
 585 {
 586         return (plugin_unregister (list_log, name));
 587 }
 588
 589 int plugin_unregister_notification (const char *name)
 590 {
 591         return (plugin_unregister (list_notification, name));
 592 }
 593
 594 void plugin_init_all (void)
 595 {
 596         int (*callback) (void);
 597         llentry_t *le;
 598         int status;
 599
 600         /* Init the value cache */
 601         uc_init ();
 602
 603         if ((list_init == NULL) && (list_read == NULL))
 604                 return;
 605
 606         /* Calling all init callbacks before checking if read callbacks
 607          * are available allows the init callbacks to register the read
 608          * callback. */
 609         le = llist_head (list_init);
 610         while (le != NULL)
 611         {
 612                 callback = (int (*) (void)) le->value;
 613                 status = (*callback) ();
 614
 615                 if (status != 0)
 616                 {
 617                         ERROR ("Initialization of plugin `%s' "
 618                                         "failed with status %i. "
 619                                         "Plugin will be unloaded.",
 620                                         le->key, status);
 621                         /* Plugins that register read callbacks from the init
 622                          * callback should take care of appropriate error
 623                          * handling themselves. */
 624                         /* FIXME: Unload _all_ functions */
 625                         plugin_unregister_read (le->key);
 626                 }
 627
 628                 le = le->next;
 629         }
 630
 631         /* Start read-threads */
 632         if (list_read != NULL)
 633         {
 634                 const char *rt;
 635                 int num;
 636                 rt = global_option_get ("ReadThreads");
 637                 num = atoi (rt);
 638                 start_threads ((num > 0) ? num : 5);
 639         }
 640 } /* void plugin_init_all */
 641
 642 void plugin_read_all (void)
 643 {
 644         llentry_t   *le;
 645         read_func_t *rf;
 646
 647         uc_check_timeout ();
 648
 649         if (list_read == NULL)
 650                 return;
 651
 652         pthread_mutex_lock (&read_lock);
 653
 654         le = llist_head (list_read);
 655         while (le != NULL)
 656         {
 657                 rf = (read_func_t *) le->value;
 658
 659                 if (rf->needs_read != DONE)
 660                 {
 661                         le = le->next;
 662                         continue;
 663                 }
 664
 665                 if (rf->wait_left > 0)
 666                         rf->wait_left -= interval_g;
 667
 668                 if (rf->wait_left <= 0)
 669                 {
 670                         rf->needs_read = TODO;
 671                 }
 672
 673                 le = le->next;
 674         }
 675
 676         DEBUG ("plugin: plugin_read_all: Signalling `read_cond'");
 677         pthread_cond_broadcast (&read_cond);
 678         pthread_mutex_unlock (&read_lock);
 679 } /* void plugin_read_all */
 680
 681 int plugin_write (const char *plugin, /* {{{ */
 682                 const data_set_t *ds, const value_list_t *vl)
 683 {
 684   int (*callback) (const data_set_t *ds, const value_list_t *vl);
 685   llentry_t *le;
 686   int status;
 687
 688   if (vl == NULL)
 689     return (EINVAL);
 690
 691   if (list_write == NULL)
 692     return (ENOENT);
 693
 694   if (ds == NULL)
 695   {
 696     ds = plugin_get_ds (vl->type);
 697     if (ds == NULL)
 698     {
 699       ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type);
 700       return (ENOENT);
 701     }
 702   }
 703
 704   if (plugin == NULL)
 705   {
 706     int success = 0;
 707     int failure = 0;
 708
 709     le = llist_head (list_write);
 710     while (le != NULL)
 711     {
 712       callback = le->value;
 713       status = (*callback) (ds, vl);
 714       if (status != 0)
 715         failure++;
 716       else
 717         success++;
 718
 719       le = le->next;
 720     }
 721
 722     if ((success == 0) && (failure != 0))
 723       status = -1;
 724     else
 725       status = 0;
 726   }
 727   else /* plugin != NULL */
 728   {
 729     le = llist_head (list_write);
 730     while (le != NULL)
 731     {
 732       if (strcasecmp (plugin, le->key) == 0)
 733         break;
 734
 735       le = le->next;
 736     }
 737
 738     if (le == NULL)
 739       return (ENOENT);
 740
 741     callback = le->value;
 742     status = (*callback) (ds, vl);
 743   }
 744
 745   return (status);
 746 } /* }}} int plugin_write */
 747
 748 int plugin_flush (const char *plugin, int timeout, const char *identifier)
 749 {
 750   int (*callback) (int timeout, const char *identifier);
 751   llentry_t *le;
 752
 753   if (list_flush == NULL)
 754     return (0);
 755
 756   le = llist_head (list_flush);
 757   while (le != NULL)
 758   {
 759     if ((plugin != NULL)
 760         && (strcmp (plugin, le->key) != 0))
 761     {
 762       le = le->next;
 763       continue;
 764     }
 765
 766     callback = (int (*) (int, const char *)) le->value;
 767     (*callback) (timeout, identifier);
 768
 769     le = le->next;
 770   }
 771   return (0);
 772 } /* int plugin_flush */
 773
 774 void plugin_shutdown_all (void)
 775 {
 776         int (*callback) (void);
 777         llentry_t *le;
 778
 779         stop_threads ();
 780
 781         if (list_shutdown == NULL)
 782                 return;
 783
 784         le = llist_head (list_shutdown);
 785         while (le != NULL)
 786         {
 787                 callback = (int (*) (void)) le->value;
 788
 789                 /* Advance the pointer before calling the callback allows
 790                  * shutdown functions to unregister themselves. If done the
 791                  * other way around the memory `le' points to will be freed
 792                  * after callback returns. */
 793                 le = le->next;
 794
 795                 (*callback) ();
 796         }
 797 } /* void plugin_shutdown_all */
 798
 799 int plugin_dispatch_values (value_list_t *vl)
 800 {
 801         static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
 802
 803         data_set_t *ds;
 804
 805         if ((vl == NULL) || (*vl->type == '\0')) {
 806                 ERROR ("plugin_dispatch_values: Invalid value list.");
 807                 return (-1);
 808         }
 809
 810         if (list_write == NULL)
 811                 c_complain_once (LOG_WARNING, &no_write_complaint,
 812                                 "plugin_dispatch_values: No write callback has been "
 813                                 "registered. Please load at least one output plugin, "
 814                                 "if you want the collected data to be stored.");
 815
 816         if (data_sets == NULL)
 817         {
 818                 ERROR ("plugin_dispatch_values: No data sets registered. "
 819                                 "Could the types database be read? Check "
 820                                 "your `TypesDB' setting!");
 821                 return (-1);
 822         }
 823
 824         if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0)
 825         {
 826                 INFO ("plugin_dispatch_values: Dataset not found: %s", vl->type);
 827                 return (-1);
 828         }
 829
 830         DEBUG ("plugin_dispatch_values: time = %u; interval = %i; "
 831                         "host = %s; "
 832                         "plugin = %s; plugin_instance = %s; "
 833                         "type = %s; type_instance = %s;",
 834                         (unsigned int) vl->time, vl->interval,
 835                         vl->host,
 836                         vl->plugin, vl->plugin_instance,
 837                         vl->type, vl->type_instance);
 838
 839 #if COLLECT_DEBUG
 840         assert (0 == strcmp (ds->type, vl->type));
 841 #else
 842         if (0 != strcmp (ds->type, vl->type))
 843                 WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)",
 844                                 ds->type, vl->type);
 845 #endif
 846
 847 #if COLLECT_DEBUG
 848         assert (ds->ds_num == vl->values_len);
 849 #else
 850         if (ds->ds_num != vl->values_len)
 851         {
 852                 ERROR ("plugin_dispatch_values: ds->type = %s: "
 853                                 "(ds->ds_num = %i) != "
 854                                 "(vl->values_len = %i)",
 855                                 ds->type, ds->ds_num, vl->values_len);
 856                 return (-1);
 857         }
 858 #endif
 859
 860         escape_slashes (vl->host, sizeof (vl->host));
 861         escape_slashes (vl->plugin, sizeof (vl->plugin));
 862         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
 863         escape_slashes (vl->type, sizeof (vl->type));
 864         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
 865
 866         /* Update the value cache */
 867         uc_update (ds, vl);
 868
 869         fc_process (ds, vl);
 870
 871         return (0);
 872 } /* int plugin_dispatch_values */
 873
 874 int plugin_dispatch_notification (const notification_t *notif)
 875 {
 876         int (*callback) (const notification_t *);
 877         llentry_t *le;
 878         /* Possible TODO: Add flap detection here */
 879
 880         DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; "
 881                         "time = %u; host = %s;",
 882                         notif->severity, notif->message,
 883                         (unsigned int) notif->time, notif->host);
 884
 885         /* Nobody cares for notifications */
 886         if (list_notification == NULL)
 887                 return (-1);
 888
 889         le = llist_head (list_notification);
 890         while (le != NULL)
 891         {
 892                 callback = (int (*) (const notification_t *)) le->value;
 893                 (*callback) (notif);
 894
 895                 le = le->next;
 896         }
 897
 898         return (0);
 899 } /* int plugin_dispatch_notification */
 900
 901 void plugin_log (int level, const char *format, ...)
 902 {
 903         char msg[1024];
 904         va_list ap;
 905
 906         void (*callback) (int, const char *);
 907         llentry_t *le;
 908
 909         if (list_log == NULL)
 910                 return;
 911
 912 #if !COLLECT_DEBUG
 913         if (level >= LOG_DEBUG)
 914                 return;
 915 #endif
 916
 917         va_start (ap, format);
 918         vsnprintf (msg, sizeof (msg), format, ap);
 919         msg[sizeof (msg) - 1] = '\0';
 920         va_end (ap);
 921
 922         le = llist_head (list_log);
 923         while (le != NULL)
 924         {
 925                 callback = (void (*) (int, const char *)) le->value;
 926                 (*callback) (level, msg);
 927
 928                 le = le->next;
 929         }
 930 } /* void plugin_log */
 931
 932 const data_set_t *plugin_get_ds (const char *name)
 933 {
 934         data_set_t *ds;
 935
 936         if (c_avl_get (data_sets, name, (void *) &ds) != 0)
 937         {
 938                 DEBUG ("No such dataset registered: %s", name);
 939                 return (NULL);
 940         }
 941
 942         return (ds);
 943 } /* data_set_t *plugin_get_ds */
 944
 945 static int plugin_notification_meta_add (notification_t *n,
 946     const char *name,
 947     enum notification_meta_type_e type,
 948     const void *value)
 949 {
 950   notification_meta_t *meta;
 951   notification_meta_t *tail;
 952
 953   if ((n == NULL) || (name == NULL) || (value == NULL))
 954   {
 955     ERROR ("plugin_notification_meta_add: A pointer is NULL!");
 956     return (-1);
 957   }
 958
 959   meta = (notification_meta_t *) malloc (sizeof (notification_meta_t));
 960   if (meta == NULL)
 961   {
 962     ERROR ("plugin_notification_meta_add: malloc failed.");
 963     return (-1);
 964   }
 965   memset (meta, 0, sizeof (notification_meta_t));
 966
 967   sstrncpy (meta->name, name, sizeof (meta->name));
 968   meta->type = type;
 969
 970   switch (type)
 971   {
 972     case NM_TYPE_STRING:
 973     {
 974       meta->nm_value.nm_string = strdup ((const char *) value);
 975       if (meta->nm_value.nm_string == NULL)
 976       {
 977         ERROR ("plugin_notification_meta_add: strdup failed.");
 978         sfree (meta);
 979         return (-1);
 980       }
 981       break;
 982     }
 983     case NM_TYPE_SIGNED_INT:
 984     {
 985       meta->nm_value.nm_signed_int = *((int64_t *) value);
 986       break;
 987     }
 988     case NM_TYPE_UNSIGNED_INT:
 989     {
 990       meta->nm_value.nm_unsigned_int = *((uint64_t *) value);
 991       break;
 992     }
 993     case NM_TYPE_DOUBLE:
 994     {
 995       meta->nm_value.nm_double = *((double *) value);
 996       break;
 997     }
 998     case NM_TYPE_BOOLEAN:
 999     {
1000       meta->nm_value.nm_boolean = *((bool *) value);
1001       break;
1002     }
1003     default:
1004     {
1005       ERROR ("plugin_notification_meta_add: Unknown type: %i", type);
1006       sfree (meta);
1007       return (-1);
1008     }
1009   } /* switch (type) */
1010
1011   meta->next = NULL;
1012   tail = n->meta;
1013   while ((tail != NULL) && (tail->next != NULL))
1014     tail = tail->next;
1015
1016   if (tail == NULL)
1017     n->meta = meta;
1018   else
1019     tail->next = meta;
1020
1021   return (0);
1022 } /* int plugin_notification_meta_add */
1023
1024 int plugin_notification_meta_add_string (notification_t *n,
1025     const char *name,
1026     const char *value)
1027 {
1028   return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value));
1029 }
1030
1031 int plugin_notification_meta_add_signed_int (notification_t *n,
1032     const char *name,
1033     int64_t value)
1034 {
1035   return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value));
1036 }
1037
1038 int plugin_notification_meta_add_unsigned_int (notification_t *n,
1039     const char *name,
1040     uint64_t value)
1041 {
1042   return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value));
1043 }
1044
1045 int plugin_notification_meta_add_double (notification_t *n,
1046     const char *name,
1047     double value)
1048 {
1049   return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value));
1050 }
1051
1052 int plugin_notification_meta_add_boolean (notification_t *n,
1053     const char *name,
1054     bool value)
1055 {
1056   return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value));
1057 }
1058
1059 int plugin_notification_meta_copy (notification_t *dst,
1060     const notification_t *src)
1061 {
1062   notification_meta_t *meta;
1063
1064   assert (dst != NULL);
1065   assert (src != NULL);
1066   assert (dst != src);
1067   assert ((src->meta == NULL) || (src->meta != dst->meta));
1068
1069   for (meta = src->meta; meta != NULL; meta = meta->next)
1070   {
1071     if (meta->type == NM_TYPE_STRING)
1072       plugin_notification_meta_add_string (dst, meta->name,
1073           meta->nm_value.nm_string);
1074     else if (meta->type == NM_TYPE_SIGNED_INT)
1075       plugin_notification_meta_add_signed_int (dst, meta->name,
1076           meta->nm_value.nm_signed_int);
1077     else if (meta->type == NM_TYPE_UNSIGNED_INT)
1078       plugin_notification_meta_add_unsigned_int (dst, meta->name,
1079           meta->nm_value.nm_unsigned_int);
1080     else if (meta->type == NM_TYPE_DOUBLE)
1081       plugin_notification_meta_add_double (dst, meta->name,
1082           meta->nm_value.nm_double);
1083     else if (meta->type == NM_TYPE_BOOLEAN)
1084       plugin_notification_meta_add_boolean (dst, meta->name,
1085           meta->nm_value.nm_boolean);
1086   }
1087
1088   return (0);
1089 } /* int plugin_notification_meta_copy */
1090
1091 int plugin_notification_meta_free (notification_t *n)
1092 {
1093   notification_meta_t *this;
1094   notification_meta_t *next;
1095
1096   if (n == NULL)
1097   {
1098     ERROR ("plugin_notification_meta_free: n == NULL!");
1099     return (-1);
1100   }
1101
1102   this = n->meta;
1103   n->meta = NULL;
1104   while (this != NULL)
1105   {
1106     next = this->next;
1107
1108     if (this->type == NM_TYPE_STRING)
1109     {
1110       free ((char *)this->nm_value.nm_string);
1111       this->nm_value.nm_string = NULL;
1112     }
1113     sfree (this);
1114
1115     this = next;
1116   }
1117
1118   return (0);
1119 } /* int plugin_notification_meta_free */