src/daemon/plugin.c

   1 /**
   2  * collectd - src/plugin.c
   3  * Copyright (C) 2005-2014  Florian octo Forster
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a
   6  * copy of this software and associated documentation files (the "Software"),
   7  * to deal in the Software without restriction, including without limitation
   8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9  * and/or sell copies of the Software, and to permit persons to whom the
  10  * Software is furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  21  * DEALINGS IN THE SOFTWARE.
  22  *
  23  * Authors:
  24  *   Florian octo Forster <octo at collectd.org>
  25  *   Sebastian Harl <sh at tokkee.org>
  26  **/
  27
  28 /* _GNU_SOURCE is needed in Linux to use pthread_setname_np */
  29 #define _GNU_SOURCE
  30
  31 #include "collectd.h"
  32
  33 #include "common.h"
  34 #include "configfile.h"
  35 #include "filter_chain.h"
  36 #include "plugin.h"
  37 #include "utils_avltree.h"
  38 #include "utils_cache.h"
  39 #include "utils_complain.h"
  40 #include "utils_heap.h"
  41 #include "utils_llist.h"
  42 #include "utils_random.h"
  43 #include "utils_time.h"
  44
  45 #if HAVE_PTHREAD_NP_H
  46 #include <pthread_np.h> /* for pthread_set_name_np(3) */
  47 #endif
  48
  49 #include <ltdl.h>
  50
  51 /*
  52  * Private structures
  53  */
  54 struct callback_func_s {
  55   void *cf_callback;
  56   user_data_t cf_udata;
  57   plugin_ctx_t cf_ctx;
  58 };
  59 typedef struct callback_func_s callback_func_t;
  60
  61 #define RF_SIMPLE 0
  62 #define RF_COMPLEX 1
  63 #define RF_REMOVE 65535
  64 struct read_func_s {
  65 /* `read_func_t' "inherits" from `callback_func_t'.
  66  * The `rf_super' member MUST be the first one in this structure! */
  67 #define rf_callback rf_super.cf_callback
  68 #define rf_udata rf_super.cf_udata
  69 #define rf_ctx rf_super.cf_ctx
  70   callback_func_t rf_super;
  71   char rf_group[DATA_MAX_NAME_LEN];
  72   char *rf_name;
  73   int rf_type;
  74   cdtime_t rf_interval;
  75   cdtime_t rf_effective_interval;
  76   cdtime_t rf_next_read;
  77 };
  78 typedef struct read_func_s read_func_t;
  79
  80 struct write_queue_s;
  81 typedef struct write_queue_s write_queue_t;
  82 struct write_queue_s {
  83   value_list_t *vl;
  84   plugin_ctx_t ctx;
  85   write_queue_t *next;
  86 };
  87
  88 struct flush_callback_s {
  89   char *name;
  90   cdtime_t timeout;
  91 };
  92 typedef struct flush_callback_s flush_callback_t;
  93
  94 /*
  95  * Private variables
  96  */
  97 static c_avl_tree_t *plugins_loaded = NULL;
  98
  99 static llist_t *list_init;
 100 static llist_t *list_write;
 101 static llist_t *list_flush;
 102 static llist_t *list_missing;
 103 static llist_t *list_shutdown;
 104 static llist_t *list_log;
 105 static llist_t *list_notification;
 106
 107 static fc_chain_t *pre_cache_chain = NULL;
 108 static fc_chain_t *post_cache_chain = NULL;
 109
 110 static c_avl_tree_t *data_sets;
 111
 112 static char *plugindir = NULL;
 113
 114 #ifndef DEFAULT_MAX_READ_INTERVAL
 115 #define DEFAULT_MAX_READ_INTERVAL TIME_T_TO_CDTIME_T_STATIC(86400)
 116 #endif
 117 static c_heap_t *read_heap = NULL;
 118 static llist_t *read_list;
 119 static int read_loop = 1;
 120 static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER;
 121 static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER;
 122 static pthread_t *read_threads = NULL;
 123 static size_t read_threads_num = 0;
 124 static cdtime_t max_read_interval = DEFAULT_MAX_READ_INTERVAL;
 125
 126 static write_queue_t *write_queue_head;
 127 static write_queue_t *write_queue_tail;
 128 static long write_queue_length = 0;
 129 static _Bool write_loop = 1;
 130 static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER;
 131 static pthread_cond_t write_cond = PTHREAD_COND_INITIALIZER;
 132 static pthread_t *write_threads = NULL;
 133 static size_t write_threads_num = 0;
 134
 135 static pthread_key_t plugin_ctx_key;
 136 static _Bool plugin_ctx_key_initialized = 0;
 137
 138 static long write_limit_high = 0;
 139 static long write_limit_low = 0;
 140
 141 static derive_t stats_values_dropped = 0;
 142 static _Bool record_statistics = 0;
 143
 144 /*
 145  * Static functions
 146  */
 147 static int plugin_dispatch_values_internal(value_list_t *vl);
 148
 149 static const char *plugin_get_dir(void) {
 150   if (plugindir == NULL)
 151     return (PLUGINDIR);
 152   else
 153     return (plugindir);
 154 }
 155
 156 static void plugin_update_internal_statistics(void) { /* {{{ */
 157
 158   gauge_t copy_write_queue_length = (gauge_t)write_queue_length;
 159
 160   /* Initialize `vl' */
 161   value_list_t vl = VALUE_LIST_INIT;
 162   sstrncpy(vl.host, hostname_g, sizeof(vl.host));
 163   sstrncpy(vl.plugin, "collectd", sizeof(vl.plugin));
 164
 165   /* Write queue */
 166   sstrncpy(vl.plugin_instance, "write_queue", sizeof(vl.plugin_instance));
 167
 168   /* Write queue : queue length */
 169   vl.values = &(value_t){.gauge = copy_write_queue_length};
 170   vl.values_len = 1;
 171   sstrncpy(vl.type, "queue_length", sizeof(vl.type));
 172   vl.type_instance[0] = 0;
 173   plugin_dispatch_values(&vl);
 174
 175   /* Write queue : Values dropped (queue length > low limit) */
 176   vl.values = &(value_t){.gauge = (gauge_t)stats_values_dropped};
 177   vl.values_len = 1;
 178   sstrncpy(vl.type, "derive", sizeof(vl.type));
 179   sstrncpy(vl.type_instance, "dropped", sizeof(vl.type_instance));
 180   plugin_dispatch_values(&vl);
 181
 182   /* Cache */
 183   sstrncpy(vl.plugin_instance, "cache", sizeof(vl.plugin_instance));
 184
 185   /* Cache : Nb entry in cache tree */
 186   vl.values = &(value_t){.gauge = (gauge_t)uc_get_size()};
 187   vl.values_len = 1;
 188   sstrncpy(vl.type, "cache_size", sizeof(vl.type));
 189   vl.type_instance[0] = 0;
 190   plugin_dispatch_values(&vl);
 191
 192   return;
 193 } /* }}} void plugin_update_internal_statistics */
 194
 195 static void destroy_callback(callback_func_t *cf) /* {{{ */
 196 {
 197   if (cf == NULL)
 198     return;
 199
 200   if ((cf->cf_udata.data != NULL) && (cf->cf_udata.free_func != NULL)) {
 201     cf->cf_udata.free_func(cf->cf_udata.data);
 202     cf->cf_udata.data = NULL;
 203     cf->cf_udata.free_func = NULL;
 204   }
 205   sfree(cf);
 206 } /* }}} void destroy_callback */
 207
 208 static void destroy_all_callbacks(llist_t **list) /* {{{ */
 209 {
 210   llentry_t *le;
 211
 212   if (*list == NULL)
 213     return;
 214
 215   le = llist_head(*list);
 216   while (le != NULL) {
 217     llentry_t *le_next;
 218
 219     le_next = le->next;
 220
 221     sfree(le->key);
 222     destroy_callback(le->value);
 223     le->value = NULL;
 224
 225     le = le_next;
 226   }
 227
 228   llist_destroy(*list);
 229   *list = NULL;
 230 } /* }}} void destroy_all_callbacks */
 231
 232 static void destroy_read_heap(void) /* {{{ */
 233 {
 234   if (read_heap == NULL)
 235     return;
 236
 237   while (42) {
 238     read_func_t *rf;
 239
 240     rf = c_heap_get_root(read_heap);
 241     if (rf == NULL)
 242       break;
 243     sfree(rf->rf_name);
 244     destroy_callback((callback_func_t *)rf);
 245   }
 246
 247   c_heap_destroy(read_heap);
 248   read_heap = NULL;
 249 } /* }}} void destroy_read_heap */
 250
 251 static int register_callback(llist_t **list, /* {{{ */
 252                              const char *name, callback_func_t *cf) {
 253   llentry_t *le;
 254   char *key;
 255
 256   if (*list == NULL) {
 257     *list = llist_create();
 258     if (*list == NULL) {
 259       ERROR("plugin: register_callback: "
 260             "llist_create failed.");
 261       destroy_callback(cf);
 262       return (-1);
 263     }
 264   }
 265
 266   key = strdup(name);
 267   if (key == NULL) {
 268     ERROR("plugin: register_callback: strdup failed.");
 269     destroy_callback(cf);
 270     return (-1);
 271   }
 272
 273   le = llist_search(*list, name);
 274   if (le == NULL) {
 275     le = llentry_create(key, cf);
 276     if (le == NULL) {
 277       ERROR("plugin: register_callback: "
 278             "llentry_create failed.");
 279       sfree(key);
 280       destroy_callback(cf);
 281       return (-1);
 282     }
 283
 284     llist_append(*list, le);
 285   } else {
 286     callback_func_t *old_cf;
 287
 288     old_cf = le->value;
 289     le->value = cf;
 290
 291     WARNING("plugin: register_callback: "
 292             "a callback named `%s' already exists - "
 293             "overwriting the old entry!",
 294             name);
 295
 296     destroy_callback(old_cf);
 297     sfree(key);
 298   }
 299
 300   return (0);
 301 } /* }}} int register_callback */
 302
 303 static void log_list_callbacks(llist_t **list, /* {{{ */
 304                                const char *comment) {
 305   char *str;
 306   int len;
 307   int i;
 308   llentry_t *le;
 309   int n;
 310   char **keys;
 311
 312   n = llist_size(*list);
 313   if (n == 0) {
 314     INFO("%s [none]", comment);
 315     return;
 316   }
 317
 318   keys = calloc(n, sizeof(char *));
 319
 320   if (keys == NULL) {
 321     ERROR("%s: failed to allocate memory for list of callbacks", comment);
 322
 323     return;
 324   }
 325
 326   for (le = llist_head(*list), i = 0, len = 0; le != NULL; le = le->next, i++) {
 327     keys[i] = le->key;
 328     len += strlen(le->key) + 6;
 329   }
 330   str = malloc(len + 10);
 331   if (str == NULL) {
 332     ERROR("%s: failed to allocate memory for list of callbacks", comment);
 333   } else {
 334     *str = '\0';
 335     strjoin(str, len, keys, n, "', '");
 336     INFO("%s ['%s']", comment, str);
 337     sfree(str);
 338   }
 339   sfree(keys);
 340 } /* }}} void log_list_callbacks */
 341
 342 static int create_register_callback(llist_t **list, /* {{{ */
 343                                     const char *name, void *callback,
 344                                     user_data_t const *ud) {
 345   callback_func_t *cf;
 346
 347   cf = calloc(1, sizeof(*cf));
 348   if (cf == NULL) {
 349     ERROR("plugin: create_register_callback: calloc failed.");
 350     return (-1);
 351   }
 352
 353   cf->cf_callback = callback;
 354   if (ud == NULL) {
 355     cf->cf_udata.data = NULL;
 356     cf->cf_udata.free_func = NULL;
 357   } else {
 358     cf->cf_udata = *ud;
 359   }
 360
 361   cf->cf_ctx = plugin_get_ctx();
 362
 363   return (register_callback(list, name, cf));
 364 } /* }}} int create_register_callback */
 365
 366 static int plugin_unregister(llist_t *list, const char *name) /* {{{ */
 367 {
 368   llentry_t *e;
 369
 370   if (list == NULL)
 371     return (-1);
 372
 373   e = llist_search(list, name);
 374   if (e == NULL)
 375     return (-1);
 376
 377   llist_remove(list, e);
 378
 379   sfree(e->key);
 380   destroy_callback(e->value);
 381
 382   llentry_destroy(e);
 383
 384   return (0);
 385 } /* }}} int plugin_unregister */
 386
 387 /*
 388  * (Try to) load the shared object `file'. Won't complain if it isn't a shared
 389  * object, but it will bitch about a shared object not having a
 390  * ``module_register'' symbol..
 391  */
 392 static int plugin_load_file(char *file, uint32_t flags) {
 393   lt_dlhandle dlh;
 394   void (*reg_handle)(void);
 395
 396   lt_dlinit();
 397   lt_dlerror(); /* clear errors */
 398
 399 #if LIBTOOL_VERSION == 2
 400   if (flags & PLUGIN_FLAGS_GLOBAL) {
 401     lt_dladvise advise;
 402     lt_dladvise_init(&advise);
 403     lt_dladvise_global(&advise);
 404     dlh = lt_dlopenadvise(file, advise);
 405     lt_dladvise_destroy(&advise);
 406   } else {
 407     dlh = lt_dlopen(file);
 408   }
 409 #else /* if LIBTOOL_VERSION == 1 */
 410   if (flags & PLUGIN_FLAGS_GLOBAL)
 411     WARNING("plugin_load_file: The global flag is not supported, "
 412             "libtool 2 is required for this.");
 413   dlh = lt_dlopen(file);
 414 #endif
 415
 416   if (dlh == NULL) {
 417     char errbuf[1024] = "";
 418
 419     ssnprintf(errbuf, sizeof(errbuf),
 420               "lt_dlopen (\"%s\") failed: %s. "
 421               "The most common cause for this problem is "
 422               "missing dependencies. Use ldd(1) to check "
 423               "the dependencies of the plugin "
 424               "/ shared object.",
 425               file, lt_dlerror());
 426
 427     ERROR("%s", errbuf);
 428     /* Make sure this is printed to STDERR in any case, but also
 429      * make sure it's printed only once. */
 430     if (list_log != NULL)
 431       fprintf(stderr, "ERROR: %s\n", errbuf);
 432
 433     return (1);
 434   }
 435
 436   if ((reg_handle = (void (*)(void))lt_dlsym(dlh, "module_register")) == NULL) {
 437     WARNING("Couldn't find symbol \"module_register\" in \"%s\": %s\n", file,
 438             lt_dlerror());
 439     lt_dlclose(dlh);
 440     return (-1);
 441   }
 442
 443   (*reg_handle)();
 444
 445   return (0);
 446 }
 447
 448 static void *plugin_read_thread(void __attribute__((unused)) * args) {
 449   while (read_loop != 0) {
 450     read_func_t *rf;
 451     plugin_ctx_t old_ctx;
 452     cdtime_t start;
 453     cdtime_t now;
 454     cdtime_t elapsed;
 455     int status;
 456     int rf_type;
 457     int rc;
 458
 459     /* Get the read function that needs to be read next.
 460      * We don't need to hold "read_lock" for the heap, but we need
 461      * to call c_heap_get_root() and pthread_cond_wait() in the
 462      * same protected block. */
 463     pthread_mutex_lock(&read_lock);
 464     rf = c_heap_get_root(read_heap);
 465     if (rf == NULL) {
 466       pthread_cond_wait(&read_cond, &read_lock);
 467       pthread_mutex_unlock(&read_lock);
 468       continue;
 469     }
 470     pthread_mutex_unlock(&read_lock);
 471
 472     if (rf->rf_interval == 0) {
 473       /* this should not happen, because the interval is set
 474        * for each plugin when loading it
 475        * XXX: issue a warning? */
 476       rf->rf_interval = plugin_get_interval();
 477       rf->rf_effective_interval = rf->rf_interval;
 478
 479       rf->rf_next_read = cdtime();
 480     }
 481
 482     /* sleep until this entry is due,
 483      * using pthread_cond_timedwait */
 484     pthread_mutex_lock(&read_lock);
 485     /* In pthread_cond_timedwait, spurious wakeups are possible
 486      * (and really happen, at least on NetBSD with > 1 CPU), thus
 487      * we need to re-evaluate the condition every time
 488      * pthread_cond_timedwait returns. */
 489     rc = 0;
 490     while ((read_loop != 0) && (cdtime() < rf->rf_next_read) && rc == 0) {
 491       rc = pthread_cond_timedwait(&read_cond, &read_lock,
 492                                   &CDTIME_T_TO_TIMESPEC(rf->rf_next_read));
 493     }
 494
 495     /* Must hold `read_lock' when accessing `rf->rf_type'. */
 496     rf_type = rf->rf_type;
 497     pthread_mutex_unlock(&read_lock);
 498
 499     /* Check if we're supposed to stop.. This may have interrupted
 500      * the sleep, too. */
 501     if (read_loop == 0) {
 502       /* Insert `rf' again, so it can be free'd correctly */
 503       c_heap_insert(read_heap, rf);
 504       break;
 505     }
 506
 507     /* The entry has been marked for deletion. The linked list
 508      * entry has already been removed by `plugin_unregister_read'.
 509      * All we have to do here is free the `read_func_t' and
 510      * continue. */
 511     if (rf_type == RF_REMOVE) {
 512       DEBUG("plugin_read_thread: Destroying the `%s' "
 513             "callback.",
 514             rf->rf_name);
 515       sfree(rf->rf_name);
 516       destroy_callback((callback_func_t *)rf);
 517       rf = NULL;
 518       continue;
 519     }
 520
 521     DEBUG("plugin_read_thread: Handling `%s'.", rf->rf_name);
 522
 523     start = cdtime();
 524
 525     old_ctx = plugin_set_ctx(rf->rf_ctx);
 526
 527     if (rf_type == RF_SIMPLE) {
 528       int (*callback)(void);
 529
 530       callback = rf->rf_callback;
 531       status = (*callback)();
 532     } else {
 533       plugin_read_cb callback;
 534
 535       assert(rf_type == RF_COMPLEX);
 536
 537       callback = rf->rf_callback;
 538       status = (*callback)(&rf->rf_udata);
 539     }
 540
 541     plugin_set_ctx(old_ctx);
 542
 543     /* If the function signals failure, we will increase the
 544      * intervals in which it will be called. */
 545     if (status != 0) {
 546       rf->rf_effective_interval *= 2;
 547       if (rf->rf_effective_interval > max_read_interval)
 548         rf->rf_effective_interval = max_read_interval;
 549
 550       NOTICE("read-function of plugin `%s' failed. "
 551              "Will suspend it for %.3f seconds.",
 552              rf->rf_name, CDTIME_T_TO_DOUBLE(rf->rf_effective_interval));
 553     } else {
 554       /* Success: Restore the interval, if it was changed. */
 555       rf->rf_effective_interval = rf->rf_interval;
 556     }
 557
 558     /* update the ``next read due'' field */
 559     now = cdtime();
 560
 561     /* calculate the time spent in the read function */
 562     elapsed = (now - start);
 563
 564     if (elapsed > rf->rf_effective_interval)
 565       WARNING(
 566           "plugin_read_thread: read-function of the `%s' plugin took %.3f "
 567           "seconds, which is above its read interval (%.3f seconds). You might "
 568           "want to adjust the `Interval' or `ReadThreads' settings.",
 569           rf->rf_name, CDTIME_T_TO_DOUBLE(elapsed),
 570           CDTIME_T_TO_DOUBLE(rf->rf_effective_interval));
 571
 572     DEBUG("plugin_read_thread: read-function of the `%s' plugin took "
 573           "%.6f seconds.",
 574           rf->rf_name, CDTIME_T_TO_DOUBLE(elapsed));
 575
 576     DEBUG("plugin_read_thread: Effective interval of the "
 577           "`%s' plugin is %.3f seconds.",
 578           rf->rf_name, CDTIME_T_TO_DOUBLE(rf->rf_effective_interval));
 579
 580     /* Calculate the next (absolute) time at which this function
 581      * should be called. */
 582     rf->rf_next_read += rf->rf_effective_interval;
 583
 584     /* Check, if `rf_next_read' is in the past. */
 585     if (rf->rf_next_read < now) {
 586       /* `rf_next_read' is in the past. Insert `now'
 587        * so this value doesn't trail off into the
 588        * past too much. */
 589       rf->rf_next_read = now;
 590     }
 591
 592     DEBUG("plugin_read_thread: Next read of the `%s' plugin at %.3f.",
 593           rf->rf_name, CDTIME_T_TO_DOUBLE(rf->rf_next_read));
 594
 595     /* Re-insert this read function into the heap again. */
 596     c_heap_insert(read_heap, rf);
 597   } /* while (read_loop) */
 598
 599   pthread_exit(NULL);
 600   return ((void *)0);
 601 } /* void *plugin_read_thread */
 602
 603 #ifdef PTHREAD_MAX_NAMELEN_NP
 604 #define THREAD_NAME_MAX PTHREAD_MAX_NAMELEN_NP
 605 #else
 606 #define THREAD_NAME_MAX 16
 607 #endif
 608
 609 static void set_thread_name(pthread_t tid, char const *name) {
 610 #if defined(HAVE_PTHREAD_SETNAME_NP) || defined(HAVE_PTHREAD_SET_NAME_NP)
 611
 612   /* glibc limits the length of the name and fails if the passed string
 613    * is too long, so we truncate it here. */
 614   char n[THREAD_NAME_MAX];
 615   if (strlen(name) >= THREAD_NAME_MAX)
 616     WARNING("set_thread_name(\"%s\"): name too long", name);
 617   sstrncpy(n, name, sizeof(n));
 618
 619 #if defined(HAVE_PTHREAD_SETNAME_NP)
 620   int status = pthread_setname_np(tid, n);
 621   if (status != 0) {
 622     char errbuf[1024];
 623     ERROR("set_thread_name(\"%s\"): %s", n,
 624           sstrerror(status, errbuf, sizeof(errbuf)));
 625   }
 626 #else /* if defined(HAVE_PTHREAD_SET_NAME_NP) */
 627   pthread_set_name_np(tid, n);
 628 #endif
 629
 630 #endif
 631 }
 632
 633 static void start_read_threads(size_t num) /* {{{ */
 634 {
 635   if (read_threads != NULL)
 636     return;
 637
 638   read_threads = (pthread_t *)calloc(num, sizeof(pthread_t));
 639   if (read_threads == NULL) {
 640     ERROR("plugin: start_read_threads: calloc failed.");
 641     return;
 642   }
 643
 644   read_threads_num = 0;
 645   for (size_t i = 0; i < num; i++) {
 646     int status = pthread_create(read_threads + read_threads_num,
 647                                 /* attr = */ NULL, plugin_read_thread,
 648                                 /* arg = */ NULL);
 649     if (status != 0) {
 650       char errbuf[1024];
 651       ERROR("plugin: start_read_threads: pthread_create failed "
 652             "with status %i (%s).",
 653             status, sstrerror(status, errbuf, sizeof(errbuf)));
 654       return;
 655     }
 656
 657     char name[THREAD_NAME_MAX];
 658     ssnprintf(name, sizeof(name), "reader#%zu", read_threads_num);
 659     set_thread_name(read_threads[read_threads_num], name);
 660
 661     read_threads_num++;
 662   } /* for (i) */
 663 } /* }}} void start_read_threads */
 664
 665 static void stop_read_threads(void) {
 666   if (read_threads == NULL)
 667     return;
 668
 669   INFO("collectd: Stopping %zu read threads.", read_threads_num);
 670
 671   pthread_mutex_lock(&read_lock);
 672   read_loop = 0;
 673   DEBUG("plugin: stop_read_threads: Signalling `read_cond'");
 674   pthread_cond_broadcast(&read_cond);
 675   pthread_mutex_unlock(&read_lock);
 676
 677   for (size_t i = 0; i < read_threads_num; i++) {
 678     if (pthread_join(read_threads[i], NULL) != 0) {
 679       ERROR("plugin: stop_read_threads: pthread_join failed.");
 680     }
 681     read_threads[i] = (pthread_t)0;
 682   }
 683   sfree(read_threads);
 684   read_threads_num = 0;
 685 } /* void stop_read_threads */
 686
 687 static void plugin_value_list_free(value_list_t *vl) /* {{{ */
 688 {
 689   if (vl == NULL)
 690     return;
 691
 692   meta_data_destroy(vl->meta);
 693   sfree(vl->values);
 694   sfree(vl);
 695 } /* }}} void plugin_value_list_free */
 696
 697 static value_list_t *
 698 plugin_value_list_clone(value_list_t const *vl_orig) /* {{{ */
 699 {
 700   value_list_t *vl;
 701
 702   if (vl_orig == NULL)
 703     return (NULL);
 704
 705   vl = malloc(sizeof(*vl));
 706   if (vl == NULL)
 707     return (NULL);
 708   memcpy(vl, vl_orig, sizeof(*vl));
 709
 710   if (vl->host[0] == 0)
 711     sstrncpy(vl->host, hostname_g, sizeof(vl->host));
 712
 713   vl->values = calloc(vl_orig->values_len, sizeof(*vl->values));
 714   if (vl->values == NULL) {
 715     plugin_value_list_free(vl);
 716     return (NULL);
 717   }
 718   memcpy(vl->values, vl_orig->values,
 719          vl_orig->values_len * sizeof(*vl->values));
 720
 721   vl->meta = meta_data_clone(vl->meta);
 722   if ((vl_orig->meta != NULL) && (vl->meta == NULL)) {
 723     plugin_value_list_free(vl);
 724     return (NULL);
 725   }
 726
 727   if (vl->time == 0)
 728     vl->time = cdtime();
 729
 730   /* Fill in the interval from the thread context, if it is zero. */
 731   if (vl->interval == 0) {
 732     plugin_ctx_t ctx = plugin_get_ctx();
 733
 734     if (ctx.interval != 0)
 735       vl->interval = ctx.interval;
 736     else {
 737       char name[6 * DATA_MAX_NAME_LEN];
 738       FORMAT_VL(name, sizeof(name), vl);
 739       ERROR("plugin_value_list_clone: Unable to determine "
 740             "interval from context for "
 741             "value list \"%s\". "
 742             "This indicates a broken plugin. "
 743             "Please report this problem to the "
 744             "collectd mailing list or at "
 745             "<http://collectd.org/bugs/>.",
 746             name);
 747       vl->interval = cf_get_default_interval();
 748     }
 749   }
 750
 751   return (vl);
 752 } /* }}} value_list_t *plugin_value_list_clone */
 753
 754 static int plugin_write_enqueue(value_list_t const *vl) /* {{{ */
 755 {
 756   write_queue_t *q;
 757
 758   q = malloc(sizeof(*q));
 759   if (q == NULL)
 760     return (ENOMEM);
 761   q->next = NULL;
 762
 763   q->vl = plugin_value_list_clone(vl);
 764   if (q->vl == NULL) {
 765     sfree(q);
 766     return (ENOMEM);
 767   }
 768
 769   /* Store context of caller (read plugin); otherwise, it would not be
 770    * available to the write plugins when actually dispatching the
 771    * value-list later on. */
 772   q->ctx = plugin_get_ctx();
 773
 774   pthread_mutex_lock(&write_lock);
 775
 776   if (write_queue_tail == NULL) {
 777     write_queue_head = q;
 778     write_queue_tail = q;
 779     write_queue_length = 1;
 780   } else {
 781     write_queue_tail->next = q;
 782     write_queue_tail = q;
 783     write_queue_length += 1;
 784   }
 785
 786   pthread_cond_signal(&write_cond);
 787   pthread_mutex_unlock(&write_lock);
 788
 789   return (0);
 790 } /* }}} int plugin_write_enqueue */
 791
 792 static value_list_t *plugin_write_dequeue(void) /* {{{ */
 793 {
 794   write_queue_t *q;
 795   value_list_t *vl;
 796
 797   pthread_mutex_lock(&write_lock);
 798
 799   while (write_loop && (write_queue_head == NULL))
 800     pthread_cond_wait(&write_cond, &write_lock);
 801
 802   if (write_queue_head == NULL) {
 803     pthread_mutex_unlock(&write_lock);
 804     return (NULL);
 805   }
 806
 807   q = write_queue_head;
 808   write_queue_head = q->next;
 809   write_queue_length -= 1;
 810   if (write_queue_head == NULL) {
 811     write_queue_tail = NULL;
 812     assert(0 == write_queue_length);
 813   }
 814
 815   pthread_mutex_unlock(&write_lock);
 816
 817   (void)plugin_set_ctx(q->ctx);
 818
 819   vl = q->vl;
 820   sfree(q);
 821   return (vl);
 822 } /* }}} value_list_t *plugin_write_dequeue */
 823
 824 static void *plugin_write_thread(void __attribute__((unused)) * args) /* {{{ */
 825 {
 826   while (write_loop) {
 827     value_list_t *vl = plugin_write_dequeue();
 828     if (vl == NULL)
 829       continue;
 830
 831     plugin_dispatch_values_internal(vl);
 832
 833     plugin_value_list_free(vl);
 834   }
 835
 836   pthread_exit(NULL);
 837   return ((void *)0);
 838 } /* }}} void *plugin_write_thread */
 839
 840 static void start_write_threads(size_t num) /* {{{ */
 841 {
 842   if (write_threads != NULL)
 843     return;
 844
 845   write_threads = (pthread_t *)calloc(num, sizeof(pthread_t));
 846   if (write_threads == NULL) {
 847     ERROR("plugin: start_write_threads: calloc failed.");
 848     return;
 849   }
 850
 851   write_threads_num = 0;
 852   for (size_t i = 0; i < num; i++) {
 853     int status = pthread_create(write_threads + write_threads_num,
 854                                 /* attr = */ NULL, plugin_write_thread,
 855                                 /* arg = */ NULL);
 856     if (status != 0) {
 857       char errbuf[1024];
 858       ERROR("plugin: start_write_threads: pthread_create failed "
 859             "with status %i (%s).",
 860             status, sstrerror(status, errbuf, sizeof(errbuf)));
 861       return;
 862     }
 863
 864     char name[THREAD_NAME_MAX];
 865     ssnprintf(name, sizeof(name), "writer#%zu", write_threads_num);
 866     set_thread_name(write_threads[write_threads_num], name);
 867
 868     write_threads_num++;
 869   } /* for (i) */
 870 } /* }}} void start_write_threads */
 871
 872 static void stop_write_threads(void) /* {{{ */
 873 {
 874   write_queue_t *q;
 875   size_t i;
 876
 877   if (write_threads == NULL)
 878     return;
 879
 880   INFO("collectd: Stopping %zu write threads.", write_threads_num);
 881
 882   pthread_mutex_lock(&write_lock);
 883   write_loop = 0;
 884   DEBUG("plugin: stop_write_threads: Signalling `write_cond'");
 885   pthread_cond_broadcast(&write_cond);
 886   pthread_mutex_unlock(&write_lock);
 887
 888   for (i = 0; i < write_threads_num; i++) {
 889     if (pthread_join(write_threads[i], NULL) != 0) {
 890       ERROR("plugin: stop_write_threads: pthread_join failed.");
 891     }
 892     write_threads[i] = (pthread_t)0;
 893   }
 894   sfree(write_threads);
 895   write_threads_num = 0;
 896
 897   pthread_mutex_lock(&write_lock);
 898   i = 0;
 899   for (q = write_queue_head; q != NULL;) {
 900     write_queue_t *q1 = q;
 901     plugin_value_list_free(q->vl);
 902     q = q->next;
 903     sfree(q1);
 904     i++;
 905   }
 906   write_queue_head = NULL;
 907   write_queue_tail = NULL;
 908   write_queue_length = 0;
 909   pthread_mutex_unlock(&write_lock);
 910
 911   if (i > 0) {
 912     WARNING("plugin: %zu value list%s left after shutting down "
 913             "the write threads.",
 914             i, (i == 1) ? " was" : "s were");
 915   }
 916 } /* }}} void stop_write_threads */
 917
 918 /*
 919  * Public functions
 920  */
 921 void plugin_set_dir(const char *dir) {
 922   sfree(plugindir);
 923
 924   if (dir == NULL) {
 925     plugindir = NULL;
 926     return;
 927   }
 928
 929   plugindir = strdup(dir);
 930   if (plugindir == NULL)
 931     ERROR("plugin_set_dir: strdup(\"%s\") failed", dir);
 932 }
 933
 934 static _Bool plugin_is_loaded(char const *name) {
 935   int status;
 936
 937   if (plugins_loaded == NULL)
 938     plugins_loaded =
 939         c_avl_create((int (*)(const void *, const void *))strcasecmp);
 940   assert(plugins_loaded != NULL);
 941
 942   status = c_avl_get(plugins_loaded, name, /* ret_value = */ NULL);
 943   return (status == 0);
 944 }
 945
 946 static int plugin_mark_loaded(char const *name) {
 947   char *name_copy;
 948   int status;
 949
 950   name_copy = strdup(name);
 951   if (name_copy == NULL)
 952     return (ENOMEM);
 953
 954   status = c_avl_insert(plugins_loaded,
 955                         /* key = */ name_copy, /* value = */ NULL);
 956   return (status);
 957 }
 958
 959 static void plugin_free_loaded(void) {
 960   void *key;
 961   void *value;
 962
 963   if (plugins_loaded == NULL)
 964     return;
 965
 966   while (c_avl_pick(plugins_loaded, &key, &value) == 0) {
 967     sfree(key);
 968     assert(value == NULL);
 969   }
 970
 971   c_avl_destroy(plugins_loaded);
 972   plugins_loaded = NULL;
 973 }
 974
 975 #define BUFSIZE 512
 976 int plugin_load(char const *plugin_name, uint32_t flags) {
 977   DIR *dh;
 978   const char *dir;
 979   char filename[BUFSIZE] = "";
 980   char typename[BUFSIZE];
 981   int ret;
 982   struct stat statbuf;
 983   struct dirent *de;
 984   int status;
 985
 986   if (plugin_name == NULL)
 987     return (EINVAL);
 988
 989   /* Check if plugin is already loaded and don't do anything in this
 990    * case. */
 991   if (plugin_is_loaded(plugin_name))
 992     return (0);
 993
 994   dir = plugin_get_dir();
 995   ret = 1;
 996
 997   /*
 998    * XXX: Magic at work:
 999    *
1000    * Some of the language bindings, for example the Python and Perl
1001    * plugins, need to be able to export symbols to the scripts they run.
1002    * For this to happen, the "Globals" flag needs to be set.
1003    * Unfortunately, this technical detail is hard to explain to the
1004    * average user and she shouldn't have to worry about this, ideally.
1005    * So in order to save everyone's sanity use a different default for a
1006    * handful of special plugins. --octo
1007    */
1008   if ((strcasecmp("perl", plugin_name) == 0) ||
1009       (strcasecmp("python", plugin_name) == 0))
1010     flags |= PLUGIN_FLAGS_GLOBAL;
1011
1012   /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
1013    * type when matching the filename */
1014   status = ssnprintf(typename, sizeof(typename), "%s.so", plugin_name);
1015   if ((status < 0) || ((size_t)status >= sizeof(typename))) {
1016     WARNING("plugin_load: Filename too long: \"%s.so\"", plugin_name);
1017     return (-1);
1018   }
1019
1020   if ((dh = opendir(dir)) == NULL) {
1021     char errbuf[1024];
1022     ERROR("plugin_load: opendir (%s) failed: %s", dir,
1023           sstrerror(errno, errbuf, sizeof(errbuf)));
1024     return (-1);
1025   }
1026
1027   while ((de = readdir(dh)) != NULL) {
1028     if (strcasecmp(de->d_name, typename))
1029       continue;
1030
1031     status = ssnprintf(filename, sizeof(filename), "%s/%s", dir, de->d_name);
1032     if ((status < 0) || ((size_t)status >= sizeof(filename))) {
1033       WARNING("plugin_load: Filename too long: \"%s/%s\"", dir, de->d_name);
1034       continue;
1035     }
1036
1037     if (lstat(filename, &statbuf) == -1) {
1038       char errbuf[1024];
1039       WARNING("plugin_load: stat (\"%s\") failed: %s", filename,
1040               sstrerror(errno, errbuf, sizeof(errbuf)));
1041       continue;
1042     } else if (!S_ISREG(statbuf.st_mode)) {
1043       /* don't follow symlinks */
1044       WARNING("plugin_load: %s is not a regular file.", filename);
1045       continue;
1046     }
1047
1048     status = plugin_load_file(filename, flags);
1049     if (status == 0) {
1050       /* success */
1051       plugin_mark_loaded(plugin_name);
1052       ret = 0;
1053       INFO("plugin_load: plugin \"%s\" successfully loaded.", plugin_name);
1054       break;
1055     } else {
1056       ERROR("plugin_load: Load plugin \"%s\" failed with "
1057             "status %i.",
1058             plugin_name, status);
1059     }
1060   }
1061
1062   closedir(dh);
1063
1064   if (filename[0] == 0)
1065     ERROR("plugin_load: Could not find plugin \"%s\" in %s", plugin_name, dir);
1066
1067   return (ret);
1068 }
1069
1070 /*
1071  * The `register_*' functions follow
1072  */
1073 int plugin_register_config(const char *name,
1074                            int (*callback)(const char *key, const char *val),
1075                            const char **keys, int keys_num) {
1076   cf_register(name, callback, keys, keys_num);
1077   return (0);
1078 } /* int plugin_register_config */
1079
1080 int plugin_register_complex_config(const char *type,
1081                                    int (*callback)(oconfig_item_t *)) {
1082   return (cf_register_complex(type, callback));
1083 } /* int plugin_register_complex_config */
1084
1085 int plugin_register_init(const char *name, int (*callback)(void)) {
1086   return (create_register_callback(&list_init, name, (void *)callback,
1087                                    /* user_data = */ NULL));
1088 } /* plugin_register_init */
1089
1090 static int plugin_compare_read_func(const void *arg0, const void *arg1) {
1091   const read_func_t *rf0;
1092   const read_func_t *rf1;
1093
1094   rf0 = arg0;
1095   rf1 = arg1;
1096
1097   if (rf0->rf_next_read < rf1->rf_next_read)
1098     return (-1);
1099   else if (rf0->rf_next_read > rf1->rf_next_read)
1100     return (1);
1101   else
1102     return (0);
1103 } /* int plugin_compare_read_func */
1104
1105 /* Add a read function to both, the heap and a linked list. The linked list if
1106  * used to look-up read functions, especially for the remove function. The heap
1107  * is used to determine which plugin to read next. */
1108 static int plugin_insert_read(read_func_t *rf) {
1109   int status;
1110   llentry_t *le;
1111
1112   rf->rf_next_read = cdtime();
1113   rf->rf_effective_interval = rf->rf_interval;
1114
1115   pthread_mutex_lock(&read_lock);
1116
1117   if (read_list == NULL) {
1118     read_list = llist_create();
1119     if (read_list == NULL) {
1120       pthread_mutex_unlock(&read_lock);
1121       ERROR("plugin_insert_read: read_list failed.");
1122       return (-1);
1123     }
1124   }
1125
1126   if (read_heap == NULL) {
1127     read_heap = c_heap_create(plugin_compare_read_func);
1128     if (read_heap == NULL) {
1129       pthread_mutex_unlock(&read_lock);
1130       ERROR("plugin_insert_read: c_heap_create failed.");
1131       return (-1);
1132     }
1133   }
1134
1135   le = llist_search(read_list, rf->rf_name);
1136   if (le != NULL) {
1137     pthread_mutex_unlock(&read_lock);
1138     WARNING("The read function \"%s\" is already registered. "
1139             "Check for duplicate \"LoadPlugin\" lines "
1140             "in your configuration!",
1141             rf->rf_name);
1142     return (EINVAL);
1143   }
1144
1145   le = llentry_create(rf->rf_name, rf);
1146   if (le == NULL) {
1147     pthread_mutex_unlock(&read_lock);
1148     ERROR("plugin_insert_read: llentry_create failed.");
1149     return (-1);
1150   }
1151
1152   status = c_heap_insert(read_heap, rf);
1153   if (status != 0) {
1154     pthread_mutex_unlock(&read_lock);
1155     ERROR("plugin_insert_read: c_heap_insert failed.");
1156     llentry_destroy(le);
1157     return (-1);
1158   }
1159
1160   /* This does not fail. */
1161   llist_append(read_list, le);
1162
1163   /* Wake up all the read threads. */
1164   pthread_cond_broadcast(&read_cond);
1165   pthread_mutex_unlock(&read_lock);
1166   return (0);
1167 } /* int plugin_insert_read */
1168
1169 int plugin_register_read(const char *name, int (*callback)(void)) {
1170   read_func_t *rf;
1171   int status;
1172
1173   rf = calloc(1, sizeof(*rf));
1174   if (rf == NULL) {
1175     ERROR("plugin_register_read: calloc failed.");
1176     return (ENOMEM);
1177   }
1178
1179   rf->rf_callback = (void *)callback;
1180   rf->rf_udata.data = NULL;
1181   rf->rf_udata.free_func = NULL;
1182   rf->rf_ctx = plugin_get_ctx();
1183   rf->rf_group[0] = '\0';
1184   rf->rf_name = strdup(name);
1185   rf->rf_type = RF_SIMPLE;
1186   rf->rf_interval = plugin_get_interval();
1187
1188   status = plugin_insert_read(rf);
1189   if (status != 0) {
1190     sfree(rf->rf_name);
1191     sfree(rf);
1192   }
1193
1194   return (status);
1195 } /* int plugin_register_read */
1196
1197 int plugin_register_complex_read(const char *group, const char *name,
1198                                  plugin_read_cb callback, cdtime_t interval,
1199                                  user_data_t const *user_data) {
1200   read_func_t *rf;
1201   int status;
1202
1203   rf = calloc(1, sizeof(*rf));
1204   if (rf == NULL) {
1205     ERROR("plugin_register_complex_read: calloc failed.");
1206     return (ENOMEM);
1207   }
1208
1209   rf->rf_callback = (void *)callback;
1210   if (group != NULL)
1211     sstrncpy(rf->rf_group, group, sizeof(rf->rf_group));
1212   else
1213     rf->rf_group[0] = '\0';
1214   rf->rf_name = strdup(name);
1215   rf->rf_type = RF_COMPLEX;
1216   rf->rf_interval = (interval != 0) ? interval : plugin_get_interval();
1217
1218   /* Set user data */
1219   if (user_data == NULL) {
1220     rf->rf_udata.data = NULL;
1221     rf->rf_udata.free_func = NULL;
1222   } else {
1223     rf->rf_udata = *user_data;
1224   }
1225
1226   rf->rf_ctx = plugin_get_ctx();
1227
1228   status = plugin_insert_read(rf);
1229   if (status != 0) {
1230     sfree(rf->rf_name);
1231     sfree(rf);
1232   }
1233
1234   return (status);
1235 } /* int plugin_register_complex_read */
1236
1237 int plugin_register_write(const char *name, plugin_write_cb callback,
1238                           user_data_t const *ud) {
1239   return (create_register_callback(&list_write, name, (void *)callback, ud));
1240 } /* int plugin_register_write */
1241
1242 static int plugin_flush_timeout_callback(user_data_t *ud) {
1243   flush_callback_t *cb = ud->data;
1244
1245   return plugin_flush(cb->name, cb->timeout, /* identifier = */ NULL);
1246 } /* static int plugin_flush_callback */
1247
1248 static void plugin_flush_timeout_callback_free(void *data) {
1249   flush_callback_t *cb = data;
1250
1251   if (cb == NULL)
1252     return;
1253
1254   sfree(cb->name);
1255   sfree(cb);
1256 } /* static void plugin_flush_callback_free */
1257
1258 static char *plugin_flush_callback_name(const char *name) {
1259   const char *flush_prefix = "flush/";
1260   size_t prefix_size;
1261   char *flush_name;
1262   size_t name_size;
1263
1264   prefix_size = strlen(flush_prefix);
1265   name_size = strlen(name);
1266
1267   flush_name = malloc(name_size + prefix_size + 1);
1268   if (flush_name == NULL) {
1269     ERROR("plugin_flush_callback_name: malloc failed.");
1270     return (NULL);
1271   }
1272
1273   sstrncpy(flush_name, flush_prefix, prefix_size + 1);
1274   sstrncpy(flush_name + prefix_size, name, name_size + 1);
1275
1276   return flush_name;
1277 } /* static char *plugin_flush_callback_name */
1278
1279 int plugin_register_flush(const char *name, plugin_flush_cb callback,
1280                           user_data_t const *ud) {
1281   int status;
1282   plugin_ctx_t ctx = plugin_get_ctx();
1283
1284   status = create_register_callback(&list_flush, name, (void *)callback, ud);
1285   if (status != 0)
1286     return status;
1287
1288   if (ctx.flush_interval != 0) {
1289     char *flush_name;
1290     flush_callback_t *cb;
1291
1292     flush_name = plugin_flush_callback_name(name);
1293     if (flush_name == NULL)
1294       return (-1);
1295
1296     cb = malloc(sizeof(*cb));
1297     if (cb == NULL) {
1298       ERROR("plugin_register_flush: malloc failed.");
1299       sfree(flush_name);
1300       return (-1);
1301     }
1302
1303     cb->name = strdup(name);
1304     if (cb->name == NULL) {
1305       ERROR("plugin_register_flush: strdup failed.");
1306       sfree(cb);
1307       sfree(flush_name);
1308       return (-1);
1309     }
1310     cb->timeout = ctx.flush_timeout;
1311
1312     status = plugin_register_complex_read(
1313         /* group     = */ "flush",
1314         /* name      = */ flush_name,
1315         /* callback  = */ plugin_flush_timeout_callback,
1316         /* interval  = */ ctx.flush_interval,
1317         /* user data = */ &(user_data_t){
1318             .data = cb, .free_func = plugin_flush_timeout_callback_free,
1319         });
1320
1321     sfree(flush_name);
1322     if (status != 0) {
1323       sfree(cb->name);
1324       sfree(cb);
1325       return status;
1326     }
1327   }
1328
1329   return 0;
1330 } /* int plugin_register_flush */
1331
1332 int plugin_register_missing(const char *name, plugin_missing_cb callback,
1333                             user_data_t const *ud) {
1334   return (create_register_callback(&list_missing, name, (void *)callback, ud));
1335 } /* int plugin_register_missing */
1336
1337 int plugin_register_shutdown(const char *name, int (*callback)(void)) {
1338   return (create_register_callback(&list_shutdown, name, (void *)callback,
1339                                    /* user_data = */ NULL));
1340 } /* int plugin_register_shutdown */
1341
1342 static void plugin_free_data_sets(void) {
1343   void *key;
1344   void *value;
1345
1346   if (data_sets == NULL)
1347     return;
1348
1349   while (c_avl_pick(data_sets, &key, &value) == 0) {
1350     data_set_t *ds = value;
1351     /* key is a pointer to ds->type */
1352
1353     sfree(ds->ds);
1354     sfree(ds);
1355   }
1356
1357   c_avl_destroy(data_sets);
1358   data_sets = NULL;
1359 } /* void plugin_free_data_sets */
1360
1361 int plugin_register_data_set(const data_set_t *ds) {
1362   data_set_t *ds_copy;
1363
1364   if ((data_sets != NULL) && (c_avl_get(data_sets, ds->type, NULL) == 0)) {
1365     NOTICE("Replacing DS `%s' with another version.", ds->type);
1366     plugin_unregister_data_set(ds->type);
1367   } else if (data_sets == NULL) {
1368     data_sets = c_avl_create((int (*)(const void *, const void *))strcmp);
1369     if (data_sets == NULL)
1370       return (-1);
1371   }
1372
1373   ds_copy = malloc(sizeof(*ds_copy));
1374   if (ds_copy == NULL)
1375     return (-1);
1376   memcpy(ds_copy, ds, sizeof(data_set_t));
1377
1378   ds_copy->ds = malloc(sizeof(*ds_copy->ds) * ds->ds_num);
1379   if (ds_copy->ds == NULL) {
1380     sfree(ds_copy);
1381     return (-1);
1382   }
1383
1384   for (size_t i = 0; i < ds->ds_num; i++)
1385     memcpy(ds_copy->ds + i, ds->ds + i, sizeof(data_source_t));
1386
1387   return (c_avl_insert(data_sets, (void *)ds_copy->type, (void *)ds_copy));
1388 } /* int plugin_register_data_set */
1389
1390 int plugin_register_log(const char *name, plugin_log_cb callback,
1391                         user_data_t const *ud) {
1392   return (create_register_callback(&list_log, name, (void *)callback, ud));
1393 } /* int plugin_register_log */
1394
1395 int plugin_register_notification(const char *name,
1396                                  plugin_notification_cb callback,
1397                                  user_data_t const *ud) {
1398   return (
1399       create_register_callback(&list_notification, name, (void *)callback, ud));
1400 } /* int plugin_register_log */
1401
1402 int plugin_unregister_config(const char *name) {
1403   cf_unregister(name);
1404   return (0);
1405 } /* int plugin_unregister_config */
1406
1407 int plugin_unregister_complex_config(const char *name) {
1408   cf_unregister_complex(name);
1409   return (0);
1410 } /* int plugin_unregister_complex_config */
1411
1412 int plugin_unregister_init(const char *name) {
1413   return (plugin_unregister(list_init, name));
1414 }
1415
1416 int plugin_unregister_read(const char *name) /* {{{ */
1417 {
1418   llentry_t *le;
1419   read_func_t *rf;
1420
1421   if (name == NULL)
1422     return (-ENOENT);
1423
1424   pthread_mutex_lock(&read_lock);
1425
1426   if (read_list == NULL) {
1427     pthread_mutex_unlock(&read_lock);
1428     return (-ENOENT);
1429   }
1430
1431   le = llist_search(read_list, name);
1432   if (le == NULL) {
1433     pthread_mutex_unlock(&read_lock);
1434     WARNING("plugin_unregister_read: No such read function: %s", name);
1435     return (-ENOENT);
1436   }
1437
1438   llist_remove(read_list, le);
1439
1440   rf = le->value;
1441   assert(rf != NULL);
1442   rf->rf_type = RF_REMOVE;
1443
1444   pthread_mutex_unlock(&read_lock);
1445
1446   llentry_destroy(le);
1447
1448   DEBUG("plugin_unregister_read: Marked `%s' for removal.", name);
1449
1450   return (0);
1451 } /* }}} int plugin_unregister_read */
1452
1453 void plugin_log_available_writers(void) {
1454   log_list_callbacks(&list_write, "Available write targets:");
1455 }
1456
1457 static int compare_read_func_group(llentry_t *e, void *ud) /* {{{ */
1458 {
1459   read_func_t *rf = e->value;
1460   char *group = ud;
1461
1462   return strcmp(rf->rf_group, (const char *)group);
1463 } /* }}} int compare_read_func_group */
1464
1465 int plugin_unregister_read_group(const char *group) /* {{{ */
1466 {
1467   llentry_t *le;
1468   read_func_t *rf;
1469
1470   int found = 0;
1471
1472   if (group == NULL)
1473     return (-ENOENT);
1474
1475   pthread_mutex_lock(&read_lock);
1476
1477   if (read_list == NULL) {
1478     pthread_mutex_unlock(&read_lock);
1479     return (-ENOENT);
1480   }
1481
1482   while (42) {
1483     le = llist_search_custom(read_list, compare_read_func_group, (void *)group);
1484
1485     if (le == NULL)
1486       break;
1487
1488     ++found;
1489
1490     llist_remove(read_list, le);
1491
1492     rf = le->value;
1493     assert(rf != NULL);
1494     rf->rf_type = RF_REMOVE;
1495
1496     llentry_destroy(le);
1497
1498     DEBUG("plugin_unregister_read_group: "
1499           "Marked `%s' (group `%s') for removal.",
1500           rf->rf_name, group);
1501   }
1502
1503   pthread_mutex_unlock(&read_lock);
1504
1505   if (found == 0) {
1506     WARNING("plugin_unregister_read_group: No such "
1507             "group of read function: %s",
1508             group);
1509     return (-ENOENT);
1510   }
1511
1512   return (0);
1513 } /* }}} int plugin_unregister_read_group */
1514
1515 int plugin_unregister_write(const char *name) {
1516   return (plugin_unregister(list_write, name));
1517 }
1518
1519 int plugin_unregister_flush(const char *name) {
1520   plugin_ctx_t ctx = plugin_get_ctx();
1521
1522   if (ctx.flush_interval != 0) {
1523     char *flush_name;
1524
1525     flush_name = plugin_flush_callback_name(name);
1526     if (flush_name != NULL) {
1527       plugin_unregister_read(flush_name);
1528       sfree(flush_name);
1529     }
1530   }
1531
1532   return plugin_unregister(list_flush, name);
1533 }
1534
1535 int plugin_unregister_missing(const char *name) {
1536   return (plugin_unregister(list_missing, name));
1537 }
1538
1539 int plugin_unregister_shutdown(const char *name) {
1540   return (plugin_unregister(list_shutdown, name));
1541 }
1542
1543 int plugin_unregister_data_set(const char *name) {
1544   data_set_t *ds;
1545
1546   if (data_sets == NULL)
1547     return (-1);
1548
1549   if (c_avl_remove(data_sets, name, NULL, (void *)&ds) != 0)
1550     return (-1);
1551
1552   sfree(ds->ds);
1553   sfree(ds);
1554
1555   return (0);
1556 } /* int plugin_unregister_data_set */
1557
1558 int plugin_unregister_log(const char *name) {
1559   return (plugin_unregister(list_log, name));
1560 }
1561
1562 int plugin_unregister_notification(const char *name) {
1563   return (plugin_unregister(list_notification, name));
1564 }
1565
1566 int plugin_init_all(void) {
1567   char const *chain_name;
1568   llentry_t *le;
1569   int status;
1570   int ret = 0;
1571
1572   /* Init the value cache */
1573   uc_init();
1574
1575   if (IS_TRUE(global_option_get("CollectInternalStats")))
1576     record_statistics = 1;
1577
1578   chain_name = global_option_get("PreCacheChain");
1579   pre_cache_chain = fc_chain_get_by_name(chain_name);
1580
1581   chain_name = global_option_get("PostCacheChain");
1582   post_cache_chain = fc_chain_get_by_name(chain_name);
1583
1584   write_limit_high = global_option_get_long("WriteQueueLimitHigh",
1585                                             /* default = */ 0);
1586   if (write_limit_high < 0) {
1587     ERROR("WriteQueueLimitHigh must be positive or zero.");
1588     write_limit_high = 0;
1589   }
1590
1591   write_limit_low =
1592       global_option_get_long("WriteQueueLimitLow",
1593                              /* default = */ write_limit_high / 2);
1594   if (write_limit_low < 0) {
1595     ERROR("WriteQueueLimitLow must be positive or zero.");
1596     write_limit_low = write_limit_high / 2;
1597   } else if (write_limit_low > write_limit_high) {
1598     ERROR("WriteQueueLimitLow must not be larger than "
1599           "WriteQueueLimitHigh.");
1600     write_limit_low = write_limit_high;
1601   }
1602
1603   write_threads_num = global_option_get_long("WriteThreads",
1604                                              /* default = */ 5);
1605   if (write_threads_num < 1) {
1606     ERROR("WriteThreads must be positive.");
1607     write_threads_num = 5;
1608   }
1609
1610   if ((list_init == NULL) && (read_heap == NULL))
1611     return ret;
1612
1613   /* Calling all init callbacks before checking if read callbacks
1614    * are available allows the init callbacks to register the read
1615    * callback. */
1616   le = llist_head(list_init);
1617   while (le != NULL) {
1618     callback_func_t *cf;
1619     plugin_init_cb callback;
1620     plugin_ctx_t old_ctx;
1621
1622     cf = le->value;
1623     old_ctx = plugin_set_ctx(cf->cf_ctx);
1624     callback = cf->cf_callback;
1625     status = (*callback)();
1626     plugin_set_ctx(old_ctx);
1627
1628     if (status != 0) {
1629       ERROR("Initialization of plugin `%s' "
1630             "failed with status %i. "
1631             "Plugin will be unloaded.",
1632             le->key, status);
1633       /* Plugins that register read callbacks from the init
1634        * callback should take care of appropriate error
1635        * handling themselves. */
1636       /* FIXME: Unload _all_ functions */
1637       plugin_unregister_read(le->key);
1638       ret = -1;
1639     }
1640
1641     le = le->next;
1642   }
1643
1644   start_write_threads((size_t)write_threads_num);
1645
1646   max_read_interval =
1647       global_option_get_time("MaxReadInterval", DEFAULT_MAX_READ_INTERVAL);
1648
1649   /* Start read-threads */
1650   if (read_heap != NULL) {
1651     const char *rt;
1652     int num;
1653
1654     rt = global_option_get("ReadThreads");
1655     num = atoi(rt);
1656     if (num != -1)
1657       start_read_threads((num > 0) ? ((size_t)num) : 5);
1658   }
1659   return ret;
1660 } /* void plugin_init_all */
1661
1662 /* TODO: Rename this function. */
1663 void plugin_read_all(void) {
1664   if (record_statistics) {
1665     plugin_update_internal_statistics();
1666   }
1667   uc_check_timeout();
1668
1669   return;
1670 } /* void plugin_read_all */
1671
1672 /* Read function called when the `-T' command line argument is given. */
1673 int plugin_read_all_once(void) {
1674   int status;
1675   int return_status = 0;
1676
1677   if (read_heap == NULL) {
1678     NOTICE("No read-functions are registered.");
1679     return (0);
1680   }
1681
1682   while (42) {
1683     read_func_t *rf;
1684     plugin_ctx_t old_ctx;
1685
1686     rf = c_heap_get_root(read_heap);
1687     if (rf == NULL)
1688       break;
1689
1690     old_ctx = plugin_set_ctx(rf->rf_ctx);
1691
1692     if (rf->rf_type == RF_SIMPLE) {
1693       int (*callback)(void);
1694
1695       callback = rf->rf_callback;
1696       status = (*callback)();
1697     } else {
1698       plugin_read_cb callback;
1699
1700       callback = rf->rf_callback;
1701       status = (*callback)(&rf->rf_udata);
1702     }
1703
1704     plugin_set_ctx(old_ctx);
1705
1706     if (status != 0) {
1707       NOTICE("read-function of plugin `%s' failed.", rf->rf_name);
1708       return_status = -1;
1709     }
1710
1711     sfree(rf->rf_name);
1712     destroy_callback((void *)rf);
1713   }
1714
1715   return (return_status);
1716 } /* int plugin_read_all_once */
1717
1718 int plugin_write(const char *plugin, /* {{{ */
1719                  const data_set_t *ds, const value_list_t *vl) {
1720   llentry_t *le;
1721   int status;
1722
1723   if (vl == NULL)
1724     return (EINVAL);
1725
1726   if (list_write == NULL)
1727     return (ENOENT);
1728
1729   if (ds == NULL) {
1730     ds = plugin_get_ds(vl->type);
1731     if (ds == NULL) {
1732       ERROR("plugin_write: Unable to lookup type `%s'.", vl->type);
1733       return (ENOENT);
1734     }
1735   }
1736
1737   if (plugin == NULL) {
1738     int success = 0;
1739     int failure = 0;
1740
1741     le = llist_head(list_write);
1742     while (le != NULL) {
1743       callback_func_t *cf = le->value;
1744       plugin_write_cb callback;
1745
1746       /* do not switch plugin context; rather keep the context (interval)
1747        * information of the calling read plugin */
1748
1749       DEBUG("plugin: plugin_write: Writing values via %s.", le->key);
1750       callback = cf->cf_callback;
1751       status = (*callback)(ds, vl, &cf->cf_udata);
1752       if (status != 0)
1753         failure++;
1754       else
1755         success++;
1756
1757       le = le->next;
1758     }
1759
1760     if ((success == 0) && (failure != 0))
1761       status = -1;
1762     else
1763       status = 0;
1764   } else /* plugin != NULL */
1765   {
1766     callback_func_t *cf;
1767     plugin_write_cb callback;
1768
1769     le = llist_head(list_write);
1770     while (le != NULL) {
1771       if (strcasecmp(plugin, le->key) == 0)
1772         break;
1773
1774       le = le->next;
1775     }
1776
1777     if (le == NULL)
1778       return (ENOENT);
1779
1780     cf = le->value;
1781
1782     /* do not switch plugin context; rather keep the context (interval)
1783      * information of the calling read plugin */
1784
1785     DEBUG("plugin: plugin_write: Writing values via %s.", le->key);
1786     callback = cf->cf_callback;
1787     status = (*callback)(ds, vl, &cf->cf_udata);
1788   }
1789
1790   return (status);
1791 } /* }}} int plugin_write */
1792
1793 int plugin_flush(const char *plugin, cdtime_t timeout, const char *identifier) {
1794   llentry_t *le;
1795
1796   if (list_flush == NULL)
1797     return (0);
1798
1799   le = llist_head(list_flush);
1800   while (le != NULL) {
1801     callback_func_t *cf;
1802     plugin_flush_cb callback;
1803     plugin_ctx_t old_ctx;
1804
1805     if ((plugin != NULL) && (strcmp(plugin, le->key) != 0)) {
1806       le = le->next;
1807       continue;
1808     }
1809
1810     cf = le->value;
1811     old_ctx = plugin_set_ctx(cf->cf_ctx);
1812     callback = cf->cf_callback;
1813
1814     (*callback)(timeout, identifier, &cf->cf_udata);
1815
1816     plugin_set_ctx(old_ctx);
1817
1818     le = le->next;
1819   }
1820   return (0);
1821 } /* int plugin_flush */
1822
1823 int plugin_shutdown_all(void) {
1824   llentry_t *le;
1825   int ret = 0; // Assume success.
1826
1827   destroy_all_callbacks(&list_init);
1828
1829   stop_read_threads();
1830
1831   pthread_mutex_lock(&read_lock);
1832   llist_destroy(read_list);
1833   read_list = NULL;
1834   pthread_mutex_unlock(&read_lock);
1835
1836   destroy_read_heap();
1837
1838   /* blocks until all write threads have shut down. */
1839   stop_write_threads();
1840
1841   /* ask all plugins to write out the state they kept. */
1842   plugin_flush(/* plugin = */ NULL,
1843                /* timeout = */ 0,
1844                /* identifier = */ NULL);
1845
1846   le = NULL;
1847   if (list_shutdown != NULL)
1848     le = llist_head(list_shutdown);
1849
1850   while (le != NULL) {
1851     callback_func_t *cf;
1852     plugin_shutdown_cb callback;
1853     plugin_ctx_t old_ctx;
1854
1855     cf = le->value;
1856     old_ctx = plugin_set_ctx(cf->cf_ctx);
1857     callback = cf->cf_callback;
1858
1859     /* Advance the pointer before calling the callback allows
1860      * shutdown functions to unregister themselves. If done the
1861      * other way around the memory `le' points to will be freed
1862      * after callback returns. */
1863     le = le->next;
1864
1865     if ((*callback)() != 0)
1866       ret = -1;
1867
1868     plugin_set_ctx(old_ctx);
1869   }
1870
1871   /* Write plugins which use the `user_data' pointer usually need the
1872    * same data available to the flush callback. If this is the case, set
1873    * the free_function to NULL when registering the flush callback and to
1874    * the real free function when registering the write callback. This way
1875    * the data isn't freed twice. */
1876   destroy_all_callbacks(&list_flush);
1877   destroy_all_callbacks(&list_missing);
1878   destroy_all_callbacks(&list_write);
1879
1880   destroy_all_callbacks(&list_notification);
1881   destroy_all_callbacks(&list_shutdown);
1882   destroy_all_callbacks(&list_log);
1883
1884   plugin_free_loaded();
1885   plugin_free_data_sets();
1886   return (ret);
1887 } /* void plugin_shutdown_all */
1888
1889 int plugin_dispatch_missing(const value_list_t *vl) /* {{{ */
1890 {
1891   llentry_t *le;
1892
1893   if (list_missing == NULL)
1894     return (0);
1895
1896   le = llist_head(list_missing);
1897   while (le != NULL) {
1898     callback_func_t *cf;
1899     plugin_missing_cb callback;
1900     plugin_ctx_t old_ctx;
1901     int status;
1902
1903     cf = le->value;
1904     old_ctx = plugin_set_ctx(cf->cf_ctx);
1905     callback = cf->cf_callback;
1906
1907     status = (*callback)(vl, &cf->cf_udata);
1908     plugin_set_ctx(old_ctx);
1909     if (status != 0) {
1910       if (status < 0) {
1911         ERROR("plugin_dispatch_missing: Callback function \"%s\" "
1912               "failed with status %i.",
1913               le->key, status);
1914         return (status);
1915       } else {
1916         return (0);
1917       }
1918     }
1919
1920     le = le->next;
1921   }
1922   return (0);
1923 } /* int }}} plugin_dispatch_missing */
1924
1925 static int plugin_dispatch_values_internal(value_list_t *vl) {
1926   int status;
1927   static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
1928
1929   data_set_t *ds;
1930
1931   _Bool free_meta_data = 0;
1932
1933   assert(vl != NULL);
1934
1935   /* These fields are initialized by plugin_value_list_clone() if needed: */
1936   assert(vl->host[0] != 0);
1937   assert(vl->time != 0); /* The time is determined at _enqueue_ time. */
1938   assert(vl->interval != 0);
1939
1940   if (vl->type[0] == 0 || vl->values == NULL || vl->values_len < 1) {
1941     ERROR("plugin_dispatch_values: Invalid value list "
1942           "from plugin %s.",
1943           vl->plugin);
1944     return (-1);
1945   }
1946
1947   /* Free meta data only if the calling function didn't specify any. In
1948    * this case matches and targets may add some and the calling function
1949    * may not expect (and therefore free) that data. */
1950   if (vl->meta == NULL)
1951     free_meta_data = 1;
1952
1953   if (list_write == NULL)
1954     c_complain_once(LOG_WARNING, &no_write_complaint,
1955                     "plugin_dispatch_values: No write callback has been "
1956                     "registered. Please load at least one output plugin, "
1957                     "if you want the collected data to be stored.");
1958
1959   if (data_sets == NULL) {
1960     ERROR("plugin_dispatch_values: No data sets registered. "
1961           "Could the types database be read? Check "
1962           "your `TypesDB' setting!");
1963     return (-1);
1964   }
1965
1966   if (c_avl_get(data_sets, vl->type, (void *)&ds) != 0) {
1967     char ident[6 * DATA_MAX_NAME_LEN];
1968
1969     FORMAT_VL(ident, sizeof(ident), vl);
1970     INFO("plugin_dispatch_values: Dataset not found: %s "
1971          "(from \"%s\"), check your types.db!",
1972          vl->type, ident);
1973     return (-1);
1974   }
1975
1976   DEBUG("plugin_dispatch_values: time = %.3f; interval = %.3f; "
1977         "host = %s; "
1978         "plugin = %s; plugin_instance = %s; "
1979         "type = %s; type_instance = %s;",
1980         CDTIME_T_TO_DOUBLE(vl->time), CDTIME_T_TO_DOUBLE(vl->interval),
1981         vl->host, vl->plugin, vl->plugin_instance, vl->type, vl->type_instance);
1982
1983 #if COLLECT_DEBUG
1984   assert(0 == strcmp(ds->type, vl->type));
1985 #else
1986   if (0 != strcmp(ds->type, vl->type))
1987     WARNING("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)",
1988             ds->type, vl->type);
1989 #endif
1990
1991 #if COLLECT_DEBUG
1992   assert(ds->ds_num == vl->values_len);
1993 #else
1994   if (ds->ds_num != vl->values_len) {
1995     ERROR("plugin_dispatch_values: ds->type = %s: "
1996           "(ds->ds_num = %zu) != "
1997           "(vl->values_len = %zu)",
1998           ds->type, ds->ds_num, vl->values_len);
1999     return (-1);
2000   }
2001 #endif
2002
2003   escape_slashes(vl->host, sizeof(vl->host));
2004   escape_slashes(vl->plugin, sizeof(vl->plugin));
2005   escape_slashes(vl->plugin_instance, sizeof(vl->plugin_instance));
2006   escape_slashes(vl->type, sizeof(vl->type));
2007   escape_slashes(vl->type_instance, sizeof(vl->type_instance));
2008
2009   if (pre_cache_chain != NULL) {
2010     status = fc_process_chain(ds, vl, pre_cache_chain);
2011     if (status < 0) {
2012       WARNING("plugin_dispatch_values: Running the "
2013               "pre-cache chain failed with "
2014               "status %i (%#x).",
2015               status, status);
2016     } else if (status == FC_TARGET_STOP)
2017       return (0);
2018   }
2019
2020   /* Update the value cache */
2021   uc_update(ds, vl);
2022
2023   if (post_cache_chain != NULL) {
2024     status = fc_process_chain(ds, vl, post_cache_chain);
2025     if (status < 0) {
2026       WARNING("plugin_dispatch_values: Running the "
2027               "post-cache chain failed with "
2028               "status %i (%#x).",
2029               status, status);
2030     }
2031   } else
2032     fc_default_action(ds, vl);
2033
2034   if ((free_meta_data != 0) && (vl->meta != NULL)) {
2035     meta_data_destroy(vl->meta);
2036     vl->meta = NULL;
2037   }
2038
2039   return (0);
2040 } /* int plugin_dispatch_values_internal */
2041
2042 static double get_drop_probability(void) /* {{{ */
2043 {
2044   long pos;
2045   long size;
2046   long wql;
2047
2048   pthread_mutex_lock(&write_lock);
2049   wql = write_queue_length;
2050   pthread_mutex_unlock(&write_lock);
2051
2052   if (wql < write_limit_low)
2053     return (0.0);
2054   if (wql >= write_limit_high)
2055     return (1.0);
2056
2057   pos = 1 + wql - write_limit_low;
2058   size = 1 + write_limit_high - write_limit_low;
2059
2060   return (((double)pos) / ((double)size));
2061 } /* }}} double get_drop_probability */
2062
2063 static _Bool check_drop_value(void) /* {{{ */
2064 {
2065   static cdtime_t last_message_time = 0;
2066   static pthread_mutex_t last_message_lock = PTHREAD_MUTEX_INITIALIZER;
2067
2068   double p;
2069   double q;
2070   int status;
2071
2072   if (write_limit_high == 0)
2073     return (0);
2074
2075   p = get_drop_probability();
2076   if (p == 0.0)
2077     return (0);
2078
2079   status = pthread_mutex_trylock(&last_message_lock);
2080   if (status == 0) {
2081     cdtime_t now;
2082
2083     now = cdtime();
2084     if ((now - last_message_time) > TIME_T_TO_CDTIME_T(1)) {
2085       last_message_time = now;
2086       ERROR("plugin_dispatch_values: Low water mark "
2087             "reached. Dropping %.0f%% of metrics.",
2088             100.0 * p);
2089     }
2090     pthread_mutex_unlock(&last_message_lock);
2091   }
2092
2093   if (p == 1.0)
2094     return (1);
2095
2096   q = cdrand_d();
2097   if (q > p)
2098     return (1);
2099   else
2100     return (0);
2101 } /* }}} _Bool check_drop_value */
2102
2103 int plugin_dispatch_values(value_list_t const *vl) {
2104   int status;
2105   static pthread_mutex_t statistics_lock = PTHREAD_MUTEX_INITIALIZER;
2106
2107   if (check_drop_value()) {
2108     if (record_statistics) {
2109       pthread_mutex_lock(&statistics_lock);
2110       stats_values_dropped++;
2111       pthread_mutex_unlock(&statistics_lock);
2112     }
2113     return (0);
2114   }
2115
2116   status = plugin_write_enqueue(vl);
2117   if (status != 0) {
2118     char errbuf[1024];
2119     ERROR("plugin_dispatch_values: plugin_write_enqueue failed "
2120           "with status %i (%s).",
2121           status, sstrerror(status, errbuf, sizeof(errbuf)));
2122     return (status);
2123   }
2124
2125   return (0);
2126 }
2127
2128 __attribute__((sentinel)) int
2129 plugin_dispatch_multivalue(value_list_t const *template, /* {{{ */
2130                            _Bool store_percentage, int store_type, ...) {
2131   value_list_t *vl;
2132   int failed = 0;
2133   gauge_t sum = 0.0;
2134   va_list ap;
2135
2136   assert(template->values_len == 1);
2137
2138   /* Calculate sum for Gauge to calculate percent if needed */
2139   if (DS_TYPE_GAUGE == store_type) {
2140     va_start(ap, store_type);
2141     while (42) {
2142       char const *name;
2143       gauge_t value;
2144
2145       name = va_arg(ap, char const *);
2146       if (name == NULL)
2147         break;
2148
2149       value = va_arg(ap, gauge_t);
2150       if (!isnan(value))
2151         sum += value;
2152     }
2153     va_end(ap);
2154   }
2155
2156   vl = plugin_value_list_clone(template);
2157   /* plugin_value_list_clone makes sure vl->time is set to non-zero. */
2158   if (store_percentage)
2159     sstrncpy(vl->type, "percent", sizeof(vl->type));
2160
2161   va_start(ap, store_type);
2162   while (42) {
2163     char const *name;
2164     int status;
2165
2166     /* Set the type instance. */
2167     name = va_arg(ap, char const *);
2168     if (name == NULL)
2169       break;
2170     sstrncpy(vl->type_instance, name, sizeof(vl->type_instance));
2171
2172     /* Set the value. */
2173     switch (store_type) {
2174     case DS_TYPE_GAUGE:
2175       vl->values[0].gauge = va_arg(ap, gauge_t);
2176       if (store_percentage)
2177         vl->values[0].gauge *= sum ? (100.0 / sum) : NAN;
2178       break;
2179     case DS_TYPE_ABSOLUTE:
2180       vl->values[0].absolute = va_arg(ap, absolute_t);
2181       break;
2182     case DS_TYPE_COUNTER:
2183       vl->values[0].counter = va_arg(ap, counter_t);
2184       break;
2185     case DS_TYPE_DERIVE:
2186       vl->values[0].derive = va_arg(ap, derive_t);
2187       break;
2188     default:
2189       ERROR("plugin_dispatch_multivalue: given store_type is incorrect.");
2190       failed++;
2191     }
2192
2193     status = plugin_write_enqueue(vl);
2194     if (status != 0)
2195       failed++;
2196   }
2197   va_end(ap);
2198
2199   plugin_value_list_free(vl);
2200   return (failed);
2201 } /* }}} int plugin_dispatch_multivalue */
2202
2203 int plugin_dispatch_notification(const notification_t *notif) {
2204   llentry_t *le;
2205   /* Possible TODO: Add flap detection here */
2206
2207   DEBUG("plugin_dispatch_notification: severity = %i; message = %s; "
2208         "time = %.3f; host = %s;",
2209         notif->severity, notif->message, CDTIME_T_TO_DOUBLE(notif->time),
2210         notif->host);
2211
2212   /* Nobody cares for notifications */
2213   if (list_notification == NULL)
2214     return (-1);
2215
2216   le = llist_head(list_notification);
2217   while (le != NULL) {
2218     callback_func_t *cf;
2219     plugin_notification_cb callback;
2220     int status;
2221
2222     /* do not switch plugin context; rather keep the context
2223      * (interval) information of the calling plugin */
2224
2225     cf = le->value;
2226     callback = cf->cf_callback;
2227     status = (*callback)(notif, &cf->cf_udata);
2228     if (status != 0) {
2229       WARNING("plugin_dispatch_notification: Notification "
2230               "callback %s returned %i.",
2231               le->key, status);
2232     }
2233
2234     le = le->next;
2235   }
2236
2237   return (0);
2238 } /* int plugin_dispatch_notification */
2239
2240 void plugin_log(int level, const char *format, ...) {
2241   char msg[1024];
2242   va_list ap;
2243   llentry_t *le;
2244
2245 #if !COLLECT_DEBUG
2246   if (level >= LOG_DEBUG)
2247     return;
2248 #endif
2249
2250   va_start(ap, format);
2251   vsnprintf(msg, sizeof(msg), format, ap);
2252   msg[sizeof(msg) - 1] = '\0';
2253   va_end(ap);
2254
2255   if (list_log == NULL) {
2256     fprintf(stderr, "%s\n", msg);
2257     return;
2258   }
2259
2260   le = llist_head(list_log);
2261   while (le != NULL) {
2262     callback_func_t *cf;
2263     plugin_log_cb callback;
2264
2265     cf = le->value;
2266     callback = cf->cf_callback;
2267
2268     /* do not switch plugin context; rather keep the context
2269      * (interval) information of the calling plugin */
2270
2271     (*callback)(level, msg, &cf->cf_udata);
2272
2273     le = le->next;
2274   }
2275 } /* void plugin_log */
2276
2277 int parse_log_severity(const char *severity) {
2278   int log_level = -1;
2279
2280   if ((0 == strcasecmp(severity, "emerg")) ||
2281       (0 == strcasecmp(severity, "alert")) ||
2282       (0 == strcasecmp(severity, "crit")) || (0 == strcasecmp(severity, "err")))
2283     log_level = LOG_ERR;
2284   else if (0 == strcasecmp(severity, "warning"))
2285     log_level = LOG_WARNING;
2286   else if (0 == strcasecmp(severity, "notice"))
2287     log_level = LOG_NOTICE;
2288   else if (0 == strcasecmp(severity, "info"))
2289     log_level = LOG_INFO;
2290 #if COLLECT_DEBUG
2291   else if (0 == strcasecmp(severity, "debug"))
2292     log_level = LOG_DEBUG;
2293 #endif /* COLLECT_DEBUG */
2294
2295   return (log_level);
2296 } /* int parse_log_severity */
2297
2298 int parse_notif_severity(const char *severity) {
2299   int notif_severity = -1;
2300
2301   if (strcasecmp(severity, "FAILURE") == 0)
2302     notif_severity = NOTIF_FAILURE;
2303   else if (strcmp(severity, "OKAY") == 0)
2304     notif_severity = NOTIF_OKAY;
2305   else if ((strcmp(severity, "WARNING") == 0) ||
2306            (strcmp(severity, "WARN") == 0))
2307     notif_severity = NOTIF_WARNING;
2308
2309   return (notif_severity);
2310 } /* int parse_notif_severity */
2311
2312 const data_set_t *plugin_get_ds(const char *name) {
2313   data_set_t *ds;
2314
2315   if (data_sets == NULL) {
2316     ERROR("plugin_get_ds: No data sets are defined yet.");
2317     return (NULL);
2318   }
2319
2320   if (c_avl_get(data_sets, name, (void *)&ds) != 0) {
2321     DEBUG("No such dataset registered: %s", name);
2322     return (NULL);
2323   }
2324
2325   return (ds);
2326 } /* data_set_t *plugin_get_ds */
2327
2328 static int plugin_notification_meta_add(notification_t *n, const char *name,
2329                                         enum notification_meta_type_e type,
2330                                         const void *value) {
2331   notification_meta_t *meta;
2332   notification_meta_t *tail;
2333
2334   if ((n == NULL) || (name == NULL) || (value == NULL)) {
2335     ERROR("plugin_notification_meta_add: A pointer is NULL!");
2336     return (-1);
2337   }
2338
2339   meta = calloc(1, sizeof(*meta));
2340   if (meta == NULL) {
2341     ERROR("plugin_notification_meta_add: calloc failed.");
2342     return (-1);
2343   }
2344
2345   sstrncpy(meta->name, name, sizeof(meta->name));
2346   meta->type = type;
2347
2348   switch (type) {
2349   case NM_TYPE_STRING: {
2350     meta->nm_value.nm_string = strdup((const char *)value);
2351     if (meta->nm_value.nm_string == NULL) {
2352       ERROR("plugin_notification_meta_add: strdup failed.");
2353       sfree(meta);
2354       return (-1);
2355     }
2356     break;
2357   }
2358   case NM_TYPE_SIGNED_INT: {
2359     meta->nm_value.nm_signed_int = *((int64_t *)value);
2360     break;
2361   }
2362   case NM_TYPE_UNSIGNED_INT: {
2363     meta->nm_value.nm_unsigned_int = *((uint64_t *)value);
2364     break;
2365   }
2366   case NM_TYPE_DOUBLE: {
2367     meta->nm_value.nm_double = *((double *)value);
2368     break;
2369   }
2370   case NM_TYPE_BOOLEAN: {
2371     meta->nm_value.nm_boolean = *((_Bool *)value);
2372     break;
2373   }
2374   default: {
2375     ERROR("plugin_notification_meta_add: Unknown type: %i", type);
2376     sfree(meta);
2377     return (-1);
2378   }
2379   } /* switch (type) */
2380
2381   meta->next = NULL;
2382   tail = n->meta;
2383   while ((tail != NULL) && (tail->next != NULL))
2384     tail = tail->next;
2385
2386   if (tail == NULL)
2387     n->meta = meta;
2388   else
2389     tail->next = meta;
2390
2391   return (0);
2392 } /* int plugin_notification_meta_add */
2393
2394 int plugin_notification_meta_add_string(notification_t *n, const char *name,
2395                                         const char *value) {
2396   return (plugin_notification_meta_add(n, name, NM_TYPE_STRING, value));
2397 }
2398
2399 int plugin_notification_meta_add_signed_int(notification_t *n, const char *name,
2400                                             int64_t value) {
2401   return (plugin_notification_meta_add(n, name, NM_TYPE_SIGNED_INT, &value));
2402 }
2403
2404 int plugin_notification_meta_add_unsigned_int(notification_t *n,
2405                                               const char *name,
2406                                               uint64_t value) {
2407   return (plugin_notification_meta_add(n, name, NM_TYPE_UNSIGNED_INT, &value));
2408 }
2409
2410 int plugin_notification_meta_add_double(notification_t *n, const char *name,
2411                                         double value) {
2412   return (plugin_notification_meta_add(n, name, NM_TYPE_DOUBLE, &value));
2413 }
2414
2415 int plugin_notification_meta_add_boolean(notification_t *n, const char *name,
2416                                          _Bool value) {
2417   return (plugin_notification_meta_add(n, name, NM_TYPE_BOOLEAN, &value));
2418 }
2419
2420 int plugin_notification_meta_copy(notification_t *dst,
2421                                   const notification_t *src) {
2422   assert(dst != NULL);
2423   assert(src != NULL);
2424   assert(dst != src);
2425   assert((src->meta == NULL) || (src->meta != dst->meta));
2426
2427   for (notification_meta_t *meta = src->meta; meta != NULL; meta = meta->next) {
2428     if (meta->type == NM_TYPE_STRING)
2429       plugin_notification_meta_add_string(dst, meta->name,
2430                                           meta->nm_value.nm_string);
2431     else if (meta->type == NM_TYPE_SIGNED_INT)
2432       plugin_notification_meta_add_signed_int(dst, meta->name,
2433                                               meta->nm_value.nm_signed_int);
2434     else if (meta->type == NM_TYPE_UNSIGNED_INT)
2435       plugin_notification_meta_add_unsigned_int(dst, meta->name,
2436                                                 meta->nm_value.nm_unsigned_int);
2437     else if (meta->type == NM_TYPE_DOUBLE)
2438       plugin_notification_meta_add_double(dst, meta->name,
2439                                           meta->nm_value.nm_double);
2440     else if (meta->type == NM_TYPE_BOOLEAN)
2441       plugin_notification_meta_add_boolean(dst, meta->name,
2442                                            meta->nm_value.nm_boolean);
2443   }
2444
2445   return (0);
2446 } /* int plugin_notification_meta_copy */
2447
2448 int plugin_notification_meta_free(notification_meta_t *n) {
2449   notification_meta_t *this;
2450   notification_meta_t *next;
2451
2452   if (n == NULL) {
2453     ERROR("plugin_notification_meta_free: n == NULL!");
2454     return (-1);
2455   }
2456
2457   this = n;
2458   while (this != NULL) {
2459     next = this->next;
2460
2461     if (this->type == NM_TYPE_STRING) {
2462       /* Assign to a temporary variable to work around nm_string's const
2463        * modifier. */
2464       void *tmp = (void *)this->nm_value.nm_string;
2465
2466       sfree(tmp);
2467       this->nm_value.nm_string = NULL;
2468     }
2469     sfree(this);
2470
2471     this = next;
2472   }
2473
2474   return (0);
2475 } /* int plugin_notification_meta_free */
2476
2477 static void plugin_ctx_destructor(void *ctx) {
2478   sfree(ctx);
2479 } /* void plugin_ctx_destructor */
2480
2481 static plugin_ctx_t ctx_init = {/* interval = */ 0};
2482
2483 static plugin_ctx_t *plugin_ctx_create(void) {
2484   plugin_ctx_t *ctx;
2485
2486   ctx = malloc(sizeof(*ctx));
2487   if (ctx == NULL) {
2488     char errbuf[1024];
2489     ERROR("Failed to allocate plugin context: %s",
2490           sstrerror(errno, errbuf, sizeof(errbuf)));
2491     return NULL;
2492   }
2493
2494   *ctx = ctx_init;
2495   assert(plugin_ctx_key_initialized);
2496   pthread_setspecific(plugin_ctx_key, ctx);
2497   DEBUG("Created new plugin context.");
2498   return (ctx);
2499 } /* int plugin_ctx_create */
2500
2501 void plugin_init_ctx(void) {
2502   pthread_key_create(&plugin_ctx_key, plugin_ctx_destructor);
2503   plugin_ctx_key_initialized = 1;
2504 } /* void plugin_init_ctx */
2505
2506 plugin_ctx_t plugin_get_ctx(void) {
2507   plugin_ctx_t *ctx;
2508
2509   assert(plugin_ctx_key_initialized);
2510   ctx = pthread_getspecific(plugin_ctx_key);
2511
2512   if (ctx == NULL) {
2513     ctx = plugin_ctx_create();
2514     /* this must no happen -- exit() instead? */
2515     if (ctx == NULL)
2516       return ctx_init;
2517   }
2518
2519   return (*ctx);
2520 } /* plugin_ctx_t plugin_get_ctx */
2521
2522 plugin_ctx_t plugin_set_ctx(plugin_ctx_t ctx) {
2523   plugin_ctx_t *c;
2524   plugin_ctx_t old;
2525
2526   assert(plugin_ctx_key_initialized);
2527   c = pthread_getspecific(plugin_ctx_key);
2528
2529   if (c == NULL) {
2530     c = plugin_ctx_create();
2531     /* this must no happen -- exit() instead? */
2532     if (c == NULL)
2533       return ctx_init;
2534   }
2535
2536   old = *c;
2537   *c = ctx;
2538
2539   return (old);
2540 } /* void plugin_set_ctx */
2541
2542 cdtime_t plugin_get_interval(void) {
2543   cdtime_t interval;
2544
2545   interval = plugin_get_ctx().interval;
2546   if (interval > 0)
2547     return interval;
2548
2549   return cf_get_default_interval();
2550 } /* cdtime_t plugin_get_interval */
2551
2552 typedef struct {
2553   plugin_ctx_t ctx;
2554   void *(*start_routine)(void *);
2555   void *arg;
2556 } plugin_thread_t;
2557
2558 static void *plugin_thread_start(void *arg) {
2559   plugin_thread_t *plugin_thread = arg;
2560
2561   void *(*start_routine)(void *) = plugin_thread->start_routine;
2562   void *plugin_arg = plugin_thread->arg;
2563
2564   plugin_set_ctx(plugin_thread->ctx);
2565
2566   sfree(plugin_thread);
2567
2568   return start_routine(plugin_arg);
2569 } /* void *plugin_thread_start */
2570
2571 int plugin_thread_create(pthread_t *thread, const pthread_attr_t *attr,
2572                          void *(*start_routine)(void *), void *arg,
2573                          char const *name) {
2574   plugin_thread_t *plugin_thread;
2575
2576   plugin_thread = malloc(sizeof(*plugin_thread));
2577   if (plugin_thread == NULL)
2578     return ENOMEM;
2579
2580   plugin_thread->ctx = plugin_get_ctx();
2581   plugin_thread->start_routine = start_routine;
2582   plugin_thread->arg = arg;
2583
2584   int ret = pthread_create(thread, attr, plugin_thread_start, plugin_thread);
2585   if (ret != 0) {
2586     sfree(plugin_thread);
2587     return ret;
2588   }
2589
2590   if (name != NULL)
2591     set_thread_name(*thread, name);
2592
2593   return 0;
2594 } /* int plugin_thread_create */
2595
2596 /* vim: set sw=8 ts=8 noet fdm=marker : */