From: Marc Fournier Date: Sat, 1 Aug 2015 07:30:22 +0000 (+0200) Subject: Merge branch 'collectd-5.4' into collectd-5.5 X-Git-Tag: collectd-5.5.1~76 X-Git-Url: https://git.octo.it/?p=collectd.git;a=commitdiff_plain;h=8c864f5e634118fce2446ebbba96a1cb7ec39dda Merge branch 'collectd-5.4' into collectd-5.5 --- 8c864f5e634118fce2446ebbba96a1cb7ec39dda diff --cc src/Makefile.am index c718621e,36b6bd77..07ffdda9 --- a/src/Makefile.am +++ b/src/Makefile.am @@@ -99,11 -152,14 +99,12 @@@ if BUILD_PLUGIN_AMQ pkglib_LTLIBRARIES += amqp.la amqp_la_SOURCES = amqp.c \ utils_cmd_putval.c utils_cmd_putval.h \ + utils_parse_option.c utils_parse_option.h \ utils_format_graphite.c utils_format_graphite.h \ utils_format_json.c utils_format_json.h -amqp_la_LDFLAGS = -module -avoid-version $(BUILD_WITH_LIBRABBITMQ_LDFLAGS) +amqp_la_LDFLAGS = $(PLUGIN_LDFLAGS) $(BUILD_WITH_LIBRABBITMQ_LDFLAGS) amqp_la_CPPFLAGS = $(AM_CPPFLAGS) $(BUILD_WITH_LIBRABBITMQ_CPPFLAGS) amqp_la_LIBADD = $(BUILD_WITH_LIBRABBITMQ_LIBS) -collectd_LDADD += "-dlopen" amqp.la -collectd_DEPENDENCIES += amqp.la endif if BUILD_PLUGIN_APACHE diff --cc src/daemon/collectd.c index 46e13b3f,00000000..88c38ace mode 100644,000000..100644 --- a/src/daemon/collectd.c +++ b/src/daemon/collectd.c @@@ -1,719 -1,0 +1,721 @@@ +/** + * collectd - src/collectd.c + * Copyright (C) 2005-2007 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + * Alvaro Barcellos + **/ + +#include "collectd.h" +#include "common.h" + +#include "plugin.h" +#include "configfile.h" + +#include +#include +#include +#include + +#include + +#if HAVE_LOCALE_H +# include +#endif + +#if HAVE_STATGRAB_H +# include +#endif + +#ifndef COLLECTD_LOCALE +# define COLLECTD_LOCALE "C" +#endif + +/* + * Global variables + */ +char hostname_g[DATA_MAX_NAME_LEN]; +cdtime_t interval_g; +int pidfile_from_cli = 0; +int timeout_g; +#if HAVE_LIBKSTAT +kstat_ctl_t *kc; +#endif /* HAVE_LIBKSTAT */ + +static int loop = 0; + +static void *do_flush (void __attribute__((unused)) *arg) +{ + INFO ("Flushing all data."); + plugin_flush (/* plugin = */ NULL, + /* timeout = */ 0, + /* ident = */ NULL); + INFO ("Finished flushing all data."); + pthread_exit (NULL); + return NULL; +} + +static void sig_int_handler (int __attribute__((unused)) signal) +{ + loop++; +} + +static void sig_term_handler (int __attribute__((unused)) signal) +{ + loop++; +} + +static void sig_usr1_handler (int __attribute__((unused)) signal) +{ + pthread_t thread; + pthread_attr_t attr; + + /* flushing the data might take a while, + * so it should be done asynchronously */ + pthread_attr_init (&attr); + pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED); + pthread_create (&thread, &attr, do_flush, NULL); + pthread_attr_destroy (&attr); +} + +static int init_hostname (void) +{ + const char *str; + + struct addrinfo ai_hints; + struct addrinfo *ai_list; + struct addrinfo *ai_ptr; + int status; + + str = global_option_get ("Hostname"); + if (str != NULL) + { + sstrncpy (hostname_g, str, sizeof (hostname_g)); + return (0); + } + + if (gethostname (hostname_g, sizeof (hostname_g)) != 0) + { + fprintf (stderr, "`gethostname' failed and no " + "hostname was configured.\n"); + return (-1); + } + + str = global_option_get ("FQDNLookup"); + if (IS_FALSE (str)) + return (0); + + memset (&ai_hints, '\0', sizeof (ai_hints)); + ai_hints.ai_flags = AI_CANONNAME; + + status = getaddrinfo (hostname_g, NULL, &ai_hints, &ai_list); + if (status != 0) + { + ERROR ("Looking up \"%s\" failed. You have set the " + "\"FQDNLookup\" option, but I cannot resolve " + "my hostname to a fully qualified domain " + "name. Please fix the network " + "configuration.", hostname_g); + return (-1); + } + + for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) + { + if (ai_ptr->ai_canonname == NULL) + continue; + + sstrncpy (hostname_g, ai_ptr->ai_canonname, sizeof (hostname_g)); + break; + } + + freeaddrinfo (ai_list); + return (0); +} /* int init_hostname */ + +static int init_global_variables (void) +{ + char const *str; + + interval_g = cf_get_default_interval (); + assert (interval_g > 0); + DEBUG ("interval_g = %.3f;", CDTIME_T_TO_DOUBLE (interval_g)); + + str = global_option_get ("Timeout"); + if (str == NULL) + str = "2"; + timeout_g = atoi (str); + if (timeout_g <= 1) + { + fprintf (stderr, "Cannot set the timeout to a correct value.\n" + "Please check your settings.\n"); + return (-1); + } + DEBUG ("timeout_g = %i;", timeout_g); + + if (init_hostname () != 0) + return (-1); + DEBUG ("hostname_g = %s;", hostname_g); + + return (0); +} /* int init_global_variables */ + +static int change_basedir (const char *orig_dir) +{ + char *dir; + size_t dirlen; + int status; + + dir = strdup (orig_dir); + if (dir == NULL) + { + char errbuf[1024]; + ERROR ("strdup failed: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return (-1); + } + + dirlen = strlen (dir); + while ((dirlen > 0) && (dir[dirlen - 1] == '/')) + dir[--dirlen] = '\0'; + - if (dirlen <= 0) ++ if (dirlen <= 0) { ++ free (dir); + return (-1); ++ } + + status = chdir (dir); + if (status == 0) + { + free (dir); + return (0); + } + else if (errno != ENOENT) + { + char errbuf[1024]; + ERROR ("change_basedir: chdir (%s): %s", dir, + sstrerror (errno, errbuf, sizeof (errbuf))); + free (dir); + return (-1); + } + + status = mkdir (dir, S_IRWXU | S_IRWXG | S_IRWXO); + if (status != 0) + { + char errbuf[1024]; + ERROR ("change_basedir: mkdir (%s): %s", dir, + sstrerror (errno, errbuf, sizeof (errbuf))); + free (dir); + return (-1); + } + + status = chdir (dir); + if (status != 0) + { + char errbuf[1024]; + ERROR ("change_basedir: chdir (%s): %s", dir, + sstrerror (errno, errbuf, sizeof (errbuf))); + free (dir); + return (-1); + } + + free (dir); + return (0); +} /* static int change_basedir (char *dir) */ + +#if HAVE_LIBKSTAT +static void update_kstat (void) +{ + if (kc == NULL) + { + if ((kc = kstat_open ()) == NULL) + ERROR ("Unable to open kstat control structure"); + } + else + { + kid_t kid; + kid = kstat_chain_update (kc); + if (kid > 0) + { + INFO ("kstat chain has been updated"); + plugin_init_all (); + } + else if (kid < 0) + ERROR ("kstat chain update failed"); + /* else: everything works as expected */ + } + + return; +} /* static void update_kstat (void) */ +#endif /* HAVE_LIBKSTAT */ + +/* TODO + * Remove all settings but `-f' and `-C' + */ +static void exit_usage (int status) +{ + printf ("Usage: "PACKAGE_NAME" [OPTIONS]\n\n" + + "Available options:\n" + " General:\n" + " -C Configuration file.\n" + " Default: "CONFIGFILE"\n" + " -t Test config and exit.\n" + " -T Test plugin read and exit.\n" + " -P PID-file.\n" + " Default: "PIDFILE"\n" +#if COLLECT_DAEMON + " -f Don't fork to the background.\n" +#endif + " -h Display help (this message)\n" + "\nBuiltin defaults:\n" + " Config file "CONFIGFILE"\n" + " PID file "PIDFILE"\n" + " Plugin directory "PLUGINDIR"\n" + " Data directory "PKGLOCALSTATEDIR"\n" + "\n"PACKAGE_NAME" "PACKAGE_VERSION", http://collectd.org/\n" + "by Florian octo Forster \n" + "for contributions see `AUTHORS'\n"); + exit (status); +} /* static void exit_usage (int status) */ + +static int do_init (void) +{ +#if HAVE_SETLOCALE + if (setlocale (LC_NUMERIC, COLLECTD_LOCALE) == NULL) + WARNING ("setlocale (\"%s\") failed.", COLLECTD_LOCALE); +#endif + +#if HAVE_LIBKSTAT + kc = NULL; + update_kstat (); +#endif + +#if HAVE_LIBSTATGRAB + if (sg_init ( +# if HAVE_LIBSTATGRAB_0_90 + 0 +# endif + )) + { + ERROR ("sg_init: %s", sg_str_error (sg_get_error ())); + return (-1); + } + + if (sg_drop_privileges ()) + { + ERROR ("sg_drop_privileges: %s", sg_str_error (sg_get_error ())); + return (-1); + } +#endif + + plugin_init_all (); + + return (0); +} /* int do_init () */ + + +static int do_loop (void) +{ + cdtime_t interval = cf_get_default_interval (); + cdtime_t wait_until; + + wait_until = cdtime () + interval; + + while (loop == 0) + { + struct timespec ts_wait = { 0, 0 }; + cdtime_t now; + +#if HAVE_LIBKSTAT + update_kstat (); +#endif + + /* Issue all plugins */ + plugin_read_all (); + + now = cdtime (); + if (now >= wait_until) + { + WARNING ("Not sleeping because the next interval is " + "%.3f seconds in the past!", + CDTIME_T_TO_DOUBLE (now - wait_until)); + wait_until = now + interval; + continue; + } + + CDTIME_T_TO_TIMESPEC (wait_until - now, &ts_wait); + wait_until = wait_until + interval; + + while ((loop == 0) && (nanosleep (&ts_wait, &ts_wait) != 0)) + { + if (errno != EINTR) + { + char errbuf[1024]; + ERROR ("nanosleep failed: %s", + sstrerror (errno, errbuf, + sizeof (errbuf))); + return (-1); + } + } + } /* while (loop == 0) */ + + return (0); +} /* int do_loop */ + +static int do_shutdown (void) +{ + plugin_shutdown_all (); + return (0); +} /* int do_shutdown */ + +#if COLLECT_DAEMON +static int pidfile_create (void) +{ + FILE *fh; + const char *file = global_option_get ("PIDFile"); + + if ((fh = fopen (file, "w")) == NULL) + { + char errbuf[1024]; + ERROR ("fopen (%s): %s", file, + sstrerror (errno, errbuf, sizeof (errbuf))); + return (1); + } + + fprintf (fh, "%i\n", (int) getpid ()); + fclose(fh); + + return (0); +} /* static int pidfile_create (const char *file) */ + +static int pidfile_remove (void) +{ + const char *file = global_option_get ("PIDFile"); + if (file == NULL) + return 0; + + return (unlink (file)); +} /* static int pidfile_remove (const char *file) */ +#endif /* COLLECT_DAEMON */ + +#ifdef KERNEL_LINUX +int notify_upstart (void) +{ + const char *upstart_job = getenv("UPSTART_JOB"); + + if (upstart_job == NULL) + return 0; + + if (strcmp(upstart_job, "collectd") != 0) + return 0; + + WARNING ("supervised by upstart, will stop to signal readyness"); + raise(SIGSTOP); + unsetenv("UPSTART_JOB"); + + return 1; +} + +int notify_systemd (void) +{ + int fd = -1; + const char *notifysocket = getenv("NOTIFY_SOCKET"); + struct sockaddr_un su; + struct iovec iov; + struct msghdr hdr; + + if (notifysocket == NULL) + return 0; + + if ((strchr("@/", notifysocket[0])) == NULL || + strlen(notifysocket) < 2) + return 0; + + WARNING ("supervised by systemd, will signal readyness"); + if ((fd = socket(AF_UNIX, SOCK_DGRAM, 0)) < 0) { + WARNING ("cannot contact systemd socket %s", notifysocket); + return 0; + } + + bzero(&su, sizeof(su)); + su.sun_family = AF_UNIX; + sstrncpy (su.sun_path, notifysocket, sizeof(su.sun_path)); + + if (notifysocket[0] == '@') + su.sun_path[0] = 0; + + bzero(&iov, sizeof(iov)); + iov.iov_base = "READY=1"; + iov.iov_len = strlen("READY=1"); + + bzero(&hdr, sizeof(hdr)); + hdr.msg_name = &su; + hdr.msg_namelen = offsetof(struct sockaddr_un, sun_path) + + strlen(notifysocket); + hdr.msg_iov = &iov; + hdr.msg_iovlen = 1; + + unsetenv("NOTIFY_SOCKET"); + if (sendmsg(fd, &hdr, MSG_NOSIGNAL) < 0) { + WARNING ("cannot send notification to systemd"); + close(fd); + return 0; + } + close(fd); + return 1; +} +#endif /* KERNEL_LINUX */ + +int main (int argc, char **argv) +{ + struct sigaction sig_int_action; + struct sigaction sig_term_action; + struct sigaction sig_usr1_action; + struct sigaction sig_pipe_action; + char *configfile = CONFIGFILE; + int test_config = 0; + int test_readall = 0; + const char *basedir; +#if COLLECT_DAEMON + struct sigaction sig_chld_action; + pid_t pid; + int daemonize = 1; +#endif + int exit_status = 0; + + /* read options */ + while (1) + { + int c; + + c = getopt (argc, argv, "htTC:" +#if COLLECT_DAEMON + "fP:" +#endif + ); + + if (c == -1) + break; + + switch (c) + { + case 'C': + configfile = optarg; + break; + case 't': + test_config = 1; + break; + case 'T': + test_readall = 1; + global_option_set ("ReadThreads", "-1"); +#if COLLECT_DAEMON + daemonize = 0; +#endif /* COLLECT_DAEMON */ + break; +#if COLLECT_DAEMON + case 'P': + global_option_set ("PIDFile", optarg); + pidfile_from_cli = 1; + break; + case 'f': + daemonize = 0; + break; +#endif /* COLLECT_DAEMON */ + case 'h': + exit_usage (0); + break; + default: + exit_usage (1); + } /* switch (c) */ + } /* while (1) */ + + if (optind < argc) + exit_usage (1); + + plugin_init_ctx (); + + /* + * Read options from the config file, the environment and the command + * line (in that order, with later options overwriting previous ones in + * general). + * Also, this will automatically load modules. + */ + if (cf_read (configfile)) + { + fprintf (stderr, "Error: Reading the config file failed!\n" + "Read the syslog for details.\n"); + return (1); + } + + /* + * Change directory. We do this _after_ reading the config and loading + * modules to relative paths work as expected. + */ + if ((basedir = global_option_get ("BaseDir")) == NULL) + { + fprintf (stderr, "Don't have a basedir to use. This should not happen. Ever."); + return (1); + } + else if (change_basedir (basedir)) + { + fprintf (stderr, "Error: Unable to change to directory `%s'.\n", basedir); + return (1); + } + + /* + * Set global variables or, if that failes, exit. We cannot run with + * them being uninitialized. If nothing is configured, then defaults + * are being used. So this means that the user has actually done + * something wrong. + */ + if (init_global_variables () != 0) + return (1); + + if (test_config) + return (0); + +#if COLLECT_DAEMON + /* + * fork off child + */ + memset (&sig_chld_action, '\0', sizeof (sig_chld_action)); + sig_chld_action.sa_handler = SIG_IGN; + sigaction (SIGCHLD, &sig_chld_action, NULL); + + /* + * Only daemonize if we're not being supervised + * by upstart or systemd (when using Linux). + */ + if (daemonize +#ifdef KERNEL_LINUX + && notify_upstart() == 0 && notify_systemd() == 0 +#endif + ) + { + if ((pid = fork ()) == -1) + { + /* error */ + char errbuf[1024]; + fprintf (stderr, "fork: %s", + sstrerror (errno, errbuf, + sizeof (errbuf))); + return (1); + } + else if (pid != 0) + { + /* parent */ + /* printf ("Running (PID %i)\n", pid); */ + return (0); + } + + /* Detach from session */ + setsid (); + + /* Write pidfile */ + if (pidfile_create ()) + exit (2); + + /* close standard descriptors */ + close (2); + close (1); + close (0); + + if (open ("/dev/null", O_RDWR) != 0) + { + ERROR ("Error: Could not connect `STDIN' to `/dev/null'"); + return (1); + } + if (dup (0) != 1) + { + ERROR ("Error: Could not connect `STDOUT' to `/dev/null'"); + return (1); + } + if (dup (0) != 2) + { + ERROR ("Error: Could not connect `STDERR' to `/dev/null'"); + return (1); + } + } /* if (daemonize) */ +#endif /* COLLECT_DAEMON */ + + memset (&sig_pipe_action, '\0', sizeof (sig_pipe_action)); + sig_pipe_action.sa_handler = SIG_IGN; + sigaction (SIGPIPE, &sig_pipe_action, NULL); + + /* + * install signal handlers + */ + memset (&sig_int_action, '\0', sizeof (sig_int_action)); + sig_int_action.sa_handler = sig_int_handler; + if (0 != sigaction (SIGINT, &sig_int_action, NULL)) { + char errbuf[1024]; + ERROR ("Error: Failed to install a signal handler for signal INT: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return (1); + } + + memset (&sig_term_action, '\0', sizeof (sig_term_action)); + sig_term_action.sa_handler = sig_term_handler; + if (0 != sigaction (SIGTERM, &sig_term_action, NULL)) { + char errbuf[1024]; + ERROR ("Error: Failed to install a signal handler for signal TERM: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return (1); + } + + memset (&sig_usr1_action, '\0', sizeof (sig_usr1_action)); + sig_usr1_action.sa_handler = sig_usr1_handler; + if (0 != sigaction (SIGUSR1, &sig_usr1_action, NULL)) { + char errbuf[1024]; + ERROR ("Error: Failed to install a signal handler for signal USR1: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return (1); + } + + /* + * run the actual loops + */ + do_init (); + + if (test_readall) + { + if (plugin_read_all_once () != 0) + exit_status = 1; + } + else + { + INFO ("Initialization complete, entering read-loop."); + do_loop (); + } + + /* close syslog */ + INFO ("Exiting normally."); + + do_shutdown (); + +#if COLLECT_DAEMON + if (daemonize) + pidfile_remove (); +#endif /* COLLECT_DAEMON */ + + return (exit_status); +} /* int main */ diff --cc src/daemon/configfile.c index 8fc6f7d7,00000000..ae3e7989 mode 100644,000000..100644 --- a/src/daemon/configfile.c +++ b/src/daemon/configfile.c @@@ -1,1369 -1,0 +1,1372 @@@ +/** + * collectd - src/configfile.c + * Copyright (C) 2005-2011 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + * Sebastian tokkee Harl + **/ + +#include "collectd.h" + +#include "liboconfig/oconfig.h" + +#include "common.h" +#include "plugin.h" +#include "configfile.h" +#include "types_list.h" +#include "filter_chain.h" + +#if HAVE_WORDEXP_H +# include +#endif /* HAVE_WORDEXP_H */ + +#if HAVE_FNMATCH_H +# include +#endif /* HAVE_FNMATCH_H */ + +#if HAVE_LIBGEN_H +# include +#endif /* HAVE_LIBGEN_H */ + +#define ESCAPE_NULL(str) ((str) == NULL ? "(null)" : (str)) + +/* + * Private types + */ +typedef struct cf_callback +{ + const char *type; + int (*callback) (const char *, const char *); + const char **keys; + int keys_num; + plugin_ctx_t ctx; + struct cf_callback *next; +} cf_callback_t; + +typedef struct cf_complex_callback_s +{ + char *type; + int (*callback) (oconfig_item_t *); + plugin_ctx_t ctx; + struct cf_complex_callback_s *next; +} cf_complex_callback_t; + +typedef struct cf_value_map_s +{ + char *key; + int (*func) (oconfig_item_t *); +} cf_value_map_t; + +typedef struct cf_global_option_s +{ + char *key; + char *value; + char *def; +} cf_global_option_t; + +/* + * Prototypes of callback functions + */ +static int dispatch_value_typesdb (oconfig_item_t *ci); +static int dispatch_value_plugindir (oconfig_item_t *ci); +static int dispatch_loadplugin (oconfig_item_t *ci); +static int dispatch_block_plugin (oconfig_item_t *ci); + +/* + * Private variables + */ +static cf_callback_t *first_callback = NULL; +static cf_complex_callback_t *complex_callback_head = NULL; + +static cf_value_map_t cf_value_map[] = +{ + {"TypesDB", dispatch_value_typesdb}, + {"PluginDir", dispatch_value_plugindir}, + {"LoadPlugin", dispatch_loadplugin}, + {"Plugin", dispatch_block_plugin} +}; +static int cf_value_map_num = STATIC_ARRAY_SIZE (cf_value_map); + +static cf_global_option_t cf_global_options[] = +{ + {"BaseDir", NULL, PKGLOCALSTATEDIR}, + {"PIDFile", NULL, PIDFILE}, + {"Hostname", NULL, NULL}, + {"FQDNLookup", NULL, "true"}, + {"Interval", NULL, NULL}, + {"ReadThreads", NULL, "5"}, + {"WriteThreads", NULL, "5"}, + {"WriteQueueLimitHigh", NULL, NULL}, + {"WriteQueueLimitLow", NULL, NULL}, + {"Timeout", NULL, "2"}, + {"AutoLoadPlugin", NULL, "false"}, + {"CollectInternalStats", NULL, "false"}, + {"PreCacheChain", NULL, "PreCache"}, + {"PostCacheChain", NULL, "PostCache"}, + {"MaxReadInterval", NULL, "86400"} +}; +static int cf_global_options_num = STATIC_ARRAY_SIZE (cf_global_options); + +static int cf_default_typesdb = 1; + +/* + * Functions to handle register/unregister, search, and other plugin related + * stuff + */ +static cf_callback_t *cf_search (const char *type) +{ + cf_callback_t *cf_cb; + + if (type == NULL) + return (NULL); + + for (cf_cb = first_callback; cf_cb != NULL; cf_cb = cf_cb->next) + if (strcasecmp (cf_cb->type, type) == 0) + break; + + return (cf_cb); +} + +static int cf_dispatch (const char *type, const char *orig_key, + const char *orig_value) +{ + cf_callback_t *cf_cb; + plugin_ctx_t old_ctx; + char *key; + char *value; + int ret; + int i; + + if (orig_key == NULL) + return (EINVAL); + + DEBUG ("type = %s, key = %s, value = %s", + ESCAPE_NULL(type), + orig_key, + ESCAPE_NULL(orig_value)); + + if ((cf_cb = cf_search (type)) == NULL) + { + WARNING ("Found a configuration for the `%s' plugin, but " + "the plugin isn't loaded or didn't register " + "a configuration callback.", type); + return (-1); + } + + if ((key = strdup (orig_key)) == NULL) + return (1); + if ((value = strdup (orig_value)) == NULL) + { + free (key); + return (2); + } + + ret = -1; + + old_ctx = plugin_set_ctx (cf_cb->ctx); + + for (i = 0; i < cf_cb->keys_num; i++) + { + if ((cf_cb->keys[i] != NULL) + && (strcasecmp (cf_cb->keys[i], key) == 0)) + { + ret = (*cf_cb->callback) (key, value); + break; + } + } + + plugin_set_ctx (old_ctx); + + if (i >= cf_cb->keys_num) + WARNING ("Plugin `%s' did not register for value `%s'.", type, key); + + free (key); + free (value); + + return (ret); +} /* int cf_dispatch */ + +static int dispatch_global_option (const oconfig_item_t *ci) +{ + if (ci->values_num != 1) + return (-1); + if (ci->values[0].type == OCONFIG_TYPE_STRING) + return (global_option_set (ci->key, ci->values[0].value.string)); + else if (ci->values[0].type == OCONFIG_TYPE_NUMBER) + { + char tmp[128]; + ssnprintf (tmp, sizeof (tmp), "%lf", ci->values[0].value.number); + return (global_option_set (ci->key, tmp)); + } + else if (ci->values[0].type == OCONFIG_TYPE_BOOLEAN) + { + if (ci->values[0].value.boolean) + return (global_option_set (ci->key, "true")); + else + return (global_option_set (ci->key, "false")); + } + + return (-1); +} /* int dispatch_global_option */ + +static int dispatch_value_typesdb (oconfig_item_t *ci) +{ + int i = 0; + + assert (strcasecmp (ci->key, "TypesDB") == 0); + + cf_default_typesdb = 0; + + if (ci->values_num < 1) { + ERROR ("configfile: `TypesDB' needs at least one argument."); + return (-1); + } + + for (i = 0; i < ci->values_num; ++i) + { + if (OCONFIG_TYPE_STRING != ci->values[i].type) { + WARNING ("configfile: TypesDB: Skipping %i. argument which " + "is not a string.", i + 1); + continue; + } + + read_types_list (ci->values[i].value.string); + } + return (0); +} /* int dispatch_value_typesdb */ + +static int dispatch_value_plugindir (oconfig_item_t *ci) +{ + assert (strcasecmp (ci->key, "PluginDir") == 0); + + if (ci->values_num != 1) + return (-1); + if (ci->values[0].type != OCONFIG_TYPE_STRING) + return (-1); + + plugin_set_dir (ci->values[0].value.string); + return (0); +} + +static int dispatch_loadplugin (oconfig_item_t *ci) +{ + int i; + const char *name; + unsigned int flags = 0; + plugin_ctx_t ctx; + plugin_ctx_t old_ctx; + int ret_val; + + assert (strcasecmp (ci->key, "LoadPlugin") == 0); + + if (ci->values_num != 1) + return (-1); + if (ci->values[0].type != OCONFIG_TYPE_STRING) + return (-1); + + name = ci->values[0].value.string; + if (strcmp ("libvirt", name) == 0) + name = "virt"; + + /* default to the global interval set before loading this plugin */ + memset (&ctx, 0, sizeof (ctx)); + ctx.interval = cf_get_default_interval (); + + for (i = 0; i < ci->children_num; ++i) { + if (strcasecmp("Globals", ci->children[i].key) == 0) + cf_util_get_flag (ci->children + i, &flags, PLUGIN_FLAGS_GLOBAL); + else if (strcasecmp ("Interval", ci->children[i].key) == 0) { + if (cf_util_get_cdtime (ci->children + i, &ctx.interval) != 0) { + /* cf_util_get_cdtime will log an error */ + continue; + } + } + else { + WARNING("Ignoring unknown LoadPlugin option \"%s\" " + "for plugin \"%s\"", + ci->children[i].key, ci->values[0].value.string); + } + } + + old_ctx = plugin_set_ctx (ctx); + ret_val = plugin_load (name, (uint32_t) flags); + /* reset to the "global" context */ + plugin_set_ctx (old_ctx); + + return (ret_val); +} /* int dispatch_value_loadplugin */ + +static int dispatch_value_plugin (const char *plugin, oconfig_item_t *ci) +{ + char buffer[4096]; + char *buffer_ptr; + int buffer_free; + int i; + + buffer_ptr = buffer; + buffer_free = sizeof (buffer); + + for (i = 0; i < ci->values_num; i++) + { + int status = -1; + + if (ci->values[i].type == OCONFIG_TYPE_STRING) + status = ssnprintf (buffer_ptr, buffer_free, " %s", + ci->values[i].value.string); + else if (ci->values[i].type == OCONFIG_TYPE_NUMBER) + status = ssnprintf (buffer_ptr, buffer_free, " %lf", + ci->values[i].value.number); + else if (ci->values[i].type == OCONFIG_TYPE_BOOLEAN) + status = ssnprintf (buffer_ptr, buffer_free, " %s", + ci->values[i].value.boolean + ? "true" : "false"); + + if ((status < 0) || (status >= buffer_free)) + return (-1); + buffer_free -= status; + buffer_ptr += status; + } + /* skip the initial space */ + buffer_ptr = buffer + 1; + + return (cf_dispatch (plugin, ci->key, buffer_ptr)); +} /* int dispatch_value_plugin */ + +static int dispatch_value (oconfig_item_t *ci) +{ + int ret = -2; + int i; + + for (i = 0; i < cf_value_map_num; i++) + if (strcasecmp (cf_value_map[i].key, ci->key) == 0) + { + ret = cf_value_map[i].func (ci); + break; + } + + for (i = 0; i < cf_global_options_num; i++) + if (strcasecmp (cf_global_options[i].key, ci->key) == 0) + { + ret = dispatch_global_option (ci); + break; + } + + return (ret); +} /* int dispatch_value */ + +static int dispatch_block_plugin (oconfig_item_t *ci) +{ + int i; + char *name; + + cf_complex_callback_t *cb; + + if (strcasecmp (ci->key, "Plugin") != 0) + return (-1); + if (ci->values_num < 1) + return (-1); + if (ci->values[0].type != OCONFIG_TYPE_STRING) + return (-1); + + name = ci->values[0].value.string; + if (strcmp ("libvirt", name) == 0) + { + /* TODO(octo): Remove this legacy. */ + WARNING ("The \"libvirt\" plugin has been renamed to \"virt\" to avoid problems with the build system. " + "Your configuration is still using the old name. " + "Please change it to use \"virt\" as soon as possible. " + "This compatibility code will go away eventually."); + name = "virt"; + } + + if (IS_TRUE (global_option_get ("AutoLoadPlugin"))) + { + plugin_ctx_t ctx; + plugin_ctx_t old_ctx; + int status; + + /* default to the global interval set before loading this plugin */ + memset (&ctx, 0, sizeof (ctx)); + ctx.interval = cf_get_default_interval (); + + old_ctx = plugin_set_ctx (ctx); + status = plugin_load (name, /* flags = */ 0); + /* reset to the "global" context */ + plugin_set_ctx (old_ctx); + + if (status != 0) + { + ERROR ("Automatically loading plugin \"%s\" failed " + "with status %i.", name, status); + return (status); + } + } + + /* Check for a complex callback first */ + for (cb = complex_callback_head; cb != NULL; cb = cb->next) + { + if (strcasecmp (name, cb->type) == 0) + { + plugin_ctx_t old_ctx; + int ret_val; + + old_ctx = plugin_set_ctx (cb->ctx); + ret_val = (cb->callback (ci)); + plugin_set_ctx (old_ctx); + return (ret_val); + } + } + + /* Hm, no complex plugin found. Dispatch the values one by one */ + for (i = 0; i < ci->children_num; i++) + { + if (ci->children[i].children == NULL) + dispatch_value_plugin (name, ci->children + i); + else + { + WARNING ("There is a `%s' block within the " + "configuration for the %s plugin. " + "The plugin either only expects " + "\"simple\" configuration statements " + "or wasn't loaded using `LoadPlugin'." + " Please check your configuration.", + ci->children[i].key, name); + } + } + + return (0); +} + + +static int dispatch_block (oconfig_item_t *ci) +{ + if (strcasecmp (ci->key, "LoadPlugin") == 0) + return (dispatch_loadplugin (ci)); + else if (strcasecmp (ci->key, "Plugin") == 0) + return (dispatch_block_plugin (ci)); + else if (strcasecmp (ci->key, "Chain") == 0) + return (fc_configure (ci)); + + return (0); +} + +static int cf_ci_replace_child (oconfig_item_t *dst, oconfig_item_t *src, + int offset) +{ + oconfig_item_t *temp; + int i; + + assert (offset >= 0); + assert (dst->children_num > offset); + + /* Free the memory used by the replaced child. Usually that's the + * `Include "blah"' statement. */ + temp = dst->children + offset; + for (i = 0; i < temp->values_num; i++) + { + if (temp->values[i].type == OCONFIG_TYPE_STRING) + { + sfree (temp->values[i].value.string); + } + } + sfree (temp->values); + temp = NULL; + + /* If (src->children_num == 0) the array size is decreased. If offset + * is _not_ the last element, (offset < (dst->children_num - 1)), then + * we need to move the trailing elements before resizing the array. */ + if ((src->children_num == 0) && (offset < (dst->children_num - 1))) + { + int nmemb = dst->children_num - (offset + 1); + memmove (dst->children + offset, dst->children + offset + 1, + sizeof (oconfig_item_t) * nmemb); + } + + /* Resize the memory containing the children to be big enough to hold + * all children. */ + if (dst->children_num + src->children_num - 1 == 0) + { + dst->children_num = 0; + return (0); + } + + temp = (oconfig_item_t *) realloc (dst->children, + sizeof (oconfig_item_t) + * (dst->children_num + src->children_num - 1)); + if (temp == NULL) + { + ERROR ("configfile: realloc failed."); + return (-1); + } + dst->children = temp; + + /* If there are children behind the include statement, and they have + * not yet been moved because (src->children_num == 0), then move them + * to the end of the list, so that the new children have room before + * them. */ + if ((src->children_num > 0) + && ((dst->children_num - (offset + 1)) > 0)) + { + int nmemb = dst->children_num - (offset + 1); + int old_offset = offset + 1; + int new_offset = offset + src->children_num; + + memmove (dst->children + new_offset, + dst->children + old_offset, + sizeof (oconfig_item_t) * nmemb); + } + + /* Last but not least: If there are new children, copy them to the + * memory reserved for them. */ + if (src->children_num > 0) + { + memcpy (dst->children + offset, + src->children, + sizeof (oconfig_item_t) * src->children_num); + } + + /* Update the number of children. */ + dst->children_num += (src->children_num - 1); + + return (0); +} /* int cf_ci_replace_child */ + +static int cf_ci_append_children (oconfig_item_t *dst, oconfig_item_t *src) +{ + oconfig_item_t *temp; + + if ((src == NULL) || (src->children_num == 0)) + return (0); + + temp = (oconfig_item_t *) realloc (dst->children, + sizeof (oconfig_item_t) + * (dst->children_num + src->children_num)); + if (temp == NULL) + { + ERROR ("configfile: realloc failed."); + return (-1); + } + dst->children = temp; + + memcpy (dst->children + dst->children_num, + src->children, + sizeof (oconfig_item_t) + * src->children_num); + dst->children_num += src->children_num; + + return (0); +} /* int cf_ci_append_children */ + +#define CF_MAX_DEPTH 8 +static oconfig_item_t *cf_read_generic (const char *path, + const char *pattern, int depth); + +static int cf_include_all (oconfig_item_t *root, int depth) +{ + int i; + + for (i = 0; i < root->children_num; i++) + { + oconfig_item_t *new; + oconfig_item_t *old; + + char *pattern = NULL; + + int j; + + if (strcasecmp (root->children[i].key, "Include") != 0) + continue; + + old = root->children + i; + + if ((old->values_num != 1) + || (old->values[0].type != OCONFIG_TYPE_STRING)) + { + ERROR ("configfile: `Include' needs exactly one string argument."); + continue; + } + + for (j = 0; j < old->children_num; ++j) + { + oconfig_item_t *child = old->children + j; + + if (strcasecmp (child->key, "Filter") == 0) + cf_util_get_string (child, &pattern); + else + ERROR ("configfile: Option `%s' not allowed in block.", + child->key); + } + + new = cf_read_generic (old->values[0].value.string, pattern, depth + 1); + sfree (pattern); + + if (new == NULL) + return (-1); + + /* Now replace the i'th child in `root' with `new'. */ - if (cf_ci_replace_child (root, new, i) < 0) ++ if (cf_ci_replace_child (root, new, i) < 0) { ++ sfree (new->values); ++ sfree (new); + return (-1); ++ } + + /* ... and go back to the new i'th child. */ + --i; + + sfree (new->values); + sfree (new); + } /* for (i = 0; i < root->children_num; i++) */ + + return (0); +} /* int cf_include_all */ + +static oconfig_item_t *cf_read_file (const char *file, + const char *pattern, int depth) +{ + oconfig_item_t *root; + int status; + + assert (depth < CF_MAX_DEPTH); + + if (pattern != NULL) { +#if HAVE_FNMATCH_H && HAVE_LIBGEN_H + char *tmp = sstrdup (file); + char *filename = basename (tmp); + + if ((filename != NULL) && (fnmatch (pattern, filename, 0) != 0)) { + DEBUG ("configfile: Not including `%s' because it " + "does not match pattern `%s'.", + filename, pattern); + free (tmp); + return (NULL); + } + + free (tmp); +#else + ERROR ("configfile: Cannot apply pattern filter '%s' " + "to file '%s': functions basename() and / or " + "fnmatch() not available.", pattern, file); +#endif /* HAVE_FNMATCH_H && HAVE_LIBGEN_H */ + } + + root = oconfig_parse_file (file); + if (root == NULL) + { + ERROR ("configfile: Cannot read file `%s'.", file); + return (NULL); + } + + status = cf_include_all (root, depth); + if (status != 0) + { + oconfig_free (root); + return (NULL); + } + + return (root); +} /* oconfig_item_t *cf_read_file */ + +static int cf_compare_string (const void *p1, const void *p2) +{ + return strcmp (*(const char **) p1, *(const char **) p2); +} + +static oconfig_item_t *cf_read_dir (const char *dir, + const char *pattern, int depth) +{ + oconfig_item_t *root = NULL; + DIR *dh; + struct dirent *de; + char **filenames = NULL; + int filenames_num = 0; + int status; + int i; + + assert (depth < CF_MAX_DEPTH); + + dh = opendir (dir); + if (dh == NULL) + { + char errbuf[1024]; + ERROR ("configfile: opendir failed: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return (NULL); + } + + root = (oconfig_item_t *) malloc (sizeof (oconfig_item_t)); + if (root == NULL) + { + ERROR ("configfile: malloc failed."); + return (NULL); + } + memset (root, 0, sizeof (oconfig_item_t)); + + while ((de = readdir (dh)) != NULL) + { + char name[1024]; + char **tmp; + + if ((de->d_name[0] == '.') || (de->d_name[0] == 0)) + continue; + + status = ssnprintf (name, sizeof (name), "%s/%s", + dir, de->d_name); + if ((status < 0) || ((size_t) status >= sizeof (name))) + { + ERROR ("configfile: Not including `%s/%s' because its" + " name is too long.", + dir, de->d_name); + for (i = 0; i < filenames_num; ++i) + free (filenames[i]); + free (filenames); + free (root); + return (NULL); + } + + ++filenames_num; + tmp = (char **) realloc (filenames, + filenames_num * sizeof (*filenames)); + if (tmp == NULL) { + ERROR ("configfile: realloc failed."); + for (i = 0; i < filenames_num - 1; ++i) + free (filenames[i]); + free (filenames); + free (root); + return (NULL); + } + filenames = tmp; + + filenames[filenames_num - 1] = sstrdup (name); + } + + if (filenames == NULL) + return (root); + + qsort ((void *) filenames, filenames_num, sizeof (*filenames), + cf_compare_string); + + for (i = 0; i < filenames_num; ++i) + { + oconfig_item_t *temp; + char *name = filenames[i]; + + temp = cf_read_generic (name, pattern, depth); + if (temp == NULL) + { + /* An error should already have been reported. */ + sfree (name); + continue; + } + + cf_ci_append_children (root, temp); + sfree (temp->children); + sfree (temp); + + free (name); + } + + free(filenames); + return (root); +} /* oconfig_item_t *cf_read_dir */ + +/* + * cf_read_generic + * + * Path is stat'ed and either cf_read_file or cf_read_dir is called + * accordingly. + * + * There are two versions of this function: If `wordexp' exists shell wildcards + * will be expanded and the function will include all matches found. If + * `wordexp' (or, more precisely, it's header file) is not available the + * simpler function is used which does not do any such expansion. + */ +#if HAVE_WORDEXP_H +static oconfig_item_t *cf_read_generic (const char *path, + const char *pattern, int depth) +{ + oconfig_item_t *root = NULL; + int status; + const char *path_ptr; + wordexp_t we; + size_t i; + + if (depth >= CF_MAX_DEPTH) + { + ERROR ("configfile: Not including `%s' because the maximum " + "nesting depth has been reached.", path); + return (NULL); + } + + status = wordexp (path, &we, WRDE_NOCMD); + if (status != 0) + { + ERROR ("configfile: wordexp (%s) failed.", path); + return (NULL); + } + + root = (oconfig_item_t *) malloc (sizeof (oconfig_item_t)); + if (root == NULL) + { + ERROR ("configfile: malloc failed."); + return (NULL); + } + memset (root, '\0', sizeof (oconfig_item_t)); + + /* wordexp() might return a sorted list already. That's not + * documented though, so let's make sure we get what we want. */ + qsort ((void *) we.we_wordv, we.we_wordc, sizeof (*we.we_wordv), + cf_compare_string); + + for (i = 0; i < we.we_wordc; i++) + { + oconfig_item_t *temp; + struct stat statbuf; + + path_ptr = we.we_wordv[i]; + + status = stat (path_ptr, &statbuf); + if (status != 0) + { + char errbuf[1024]; + WARNING ("configfile: stat (%s) failed: %s", + path_ptr, + sstrerror (errno, errbuf, sizeof (errbuf))); + continue; + } + + if (S_ISREG (statbuf.st_mode)) + temp = cf_read_file (path_ptr, pattern, depth); + else if (S_ISDIR (statbuf.st_mode)) + temp = cf_read_dir (path_ptr, pattern, depth); + else + { + WARNING ("configfile: %s is neither a file nor a " + "directory.", path); + continue; + } + + if (temp == NULL) { + oconfig_free (root); + return (NULL); + } + + cf_ci_append_children (root, temp); + sfree (temp->children); + sfree (temp); + } + + wordfree (&we); + + return (root); +} /* oconfig_item_t *cf_read_generic */ +/* #endif HAVE_WORDEXP_H */ + +#else /* if !HAVE_WORDEXP_H */ +static oconfig_item_t *cf_read_generic (const char *path, + const char *pattern, int depth) +{ + struct stat statbuf; + int status; + + if (depth >= CF_MAX_DEPTH) + { + ERROR ("configfile: Not including `%s' because the maximum " + "nesting depth has been reached.", path); + return (NULL); + } + + status = stat (path, &statbuf); + if (status != 0) + { + char errbuf[1024]; + ERROR ("configfile: stat (%s) failed: %s", + path, + sstrerror (errno, errbuf, sizeof (errbuf))); + return (NULL); + } + + if (S_ISREG (statbuf.st_mode)) + return (cf_read_file (path, pattern, depth)); + else if (S_ISDIR (statbuf.st_mode)) + return (cf_read_dir (path, pattern, depth)); + + ERROR ("configfile: %s is neither a file nor a directory.", path); + return (NULL); +} /* oconfig_item_t *cf_read_generic */ +#endif /* !HAVE_WORDEXP_H */ + +/* + * Public functions + */ +int global_option_set (const char *option, const char *value) +{ + int i; + + DEBUG ("option = %s; value = %s;", option, value); + + for (i = 0; i < cf_global_options_num; i++) + if (strcasecmp (cf_global_options[i].key, option) == 0) + break; + + if (i >= cf_global_options_num) + return (-1); + + if (strcasecmp (option, "PIDFile") == 0 && pidfile_from_cli == 1) + { + DEBUG ("Configfile: Ignoring `PIDFILE' option because " + "command-line option `-P' take precedence."); + return (0); + } + + sfree (cf_global_options[i].value); + + if (value != NULL) + cf_global_options[i].value = strdup (value); + else + cf_global_options[i].value = NULL; + + return (0); +} + +const char *global_option_get (const char *option) +{ + int i; + + for (i = 0; i < cf_global_options_num; i++) + if (strcasecmp (cf_global_options[i].key, option) == 0) + break; + + if (i >= cf_global_options_num) + return (NULL); + + return ((cf_global_options[i].value != NULL) + ? cf_global_options[i].value + : cf_global_options[i].def); +} /* char *global_option_get */ + +long global_option_get_long (const char *option, long default_value) +{ + const char *str; + long value; + + str = global_option_get (option); + if (NULL == str) + return (default_value); + + errno = 0; + value = strtol (str, /* endptr = */ NULL, /* base = */ 0); + if (errno != 0) + return (default_value); + + return (value); +} /* char *global_option_get_long */ + +cdtime_t global_option_get_time (const char *name, cdtime_t def) /* {{{ */ +{ + char const *optstr; + char *endptr = NULL; + double v; + + optstr = global_option_get (name); + if (optstr == NULL) + return (def); + + errno = 0; + v = strtod (optstr, &endptr); + if ((endptr == NULL) || (*endptr != 0) || (errno != 0)) + return (def); + else if (v <= 0.0) + return (def); + + return (DOUBLE_TO_CDTIME_T (v)); +} /* }}} cdtime_t global_option_get_time */ + +cdtime_t cf_get_default_interval (void) +{ + return (global_option_get_time ("Interval", + DOUBLE_TO_CDTIME_T (COLLECTD_DEFAULT_INTERVAL))); +} + +void cf_unregister (const char *type) +{ + cf_callback_t *this, *prev; + + for (prev = NULL, this = first_callback; + this != NULL; + prev = this, this = this->next) + if (strcasecmp (this->type, type) == 0) + { + if (prev == NULL) + first_callback = this->next; + else + prev->next = this->next; + + free (this); + break; + } +} /* void cf_unregister */ + +void cf_unregister_complex (const char *type) +{ + cf_complex_callback_t *this, *prev; + + for (prev = NULL, this = complex_callback_head; + this != NULL; + prev = this, this = this->next) + if (strcasecmp (this->type, type) == 0) + { + if (prev == NULL) + complex_callback_head = this->next; + else + prev->next = this->next; + + sfree (this->type); + sfree (this); + break; + } +} /* void cf_unregister */ + +void cf_register (const char *type, + int (*callback) (const char *, const char *), + const char **keys, int keys_num) +{ + cf_callback_t *cf_cb; + + /* Remove this module from the list, if it already exists */ + cf_unregister (type); + + /* This pointer will be free'd in `cf_unregister' */ + if ((cf_cb = (cf_callback_t *) malloc (sizeof (cf_callback_t))) == NULL) + return; + + cf_cb->type = type; + cf_cb->callback = callback; + cf_cb->keys = keys; + cf_cb->keys_num = keys_num; + cf_cb->ctx = plugin_get_ctx (); + + cf_cb->next = first_callback; + first_callback = cf_cb; +} /* void cf_register */ + +int cf_register_complex (const char *type, int (*callback) (oconfig_item_t *)) +{ + cf_complex_callback_t *new; + + new = (cf_complex_callback_t *) malloc (sizeof (cf_complex_callback_t)); + if (new == NULL) + return (-1); + + new->type = strdup (type); + if (new->type == NULL) + { + sfree (new); + return (-1); + } + + new->callback = callback; + new->next = NULL; + + new->ctx = plugin_get_ctx (); + + if (complex_callback_head == NULL) + { + complex_callback_head = new; + } + else + { + cf_complex_callback_t *last = complex_callback_head; + while (last->next != NULL) + last = last->next; + last->next = new; + } + + return (0); +} /* int cf_register_complex */ + +int cf_read (char *filename) +{ + oconfig_item_t *conf; + int i; + + conf = cf_read_generic (filename, /* pattern = */ NULL, /* depth = */ 0); + if (conf == NULL) + { + ERROR ("Unable to read config file %s.", filename); + return (-1); + } + else if (conf->children_num == 0) + { + ERROR ("Configuration file %s is empty.", filename); + oconfig_free (conf); + return (-1); + } + + for (i = 0; i < conf->children_num; i++) + { + if (conf->children[i].children == NULL) + dispatch_value (conf->children + i); + else + dispatch_block (conf->children + i); + } + + oconfig_free (conf); + + /* Read the default types.db if no `TypesDB' option was given. */ + if (cf_default_typesdb) + read_types_list (PKGDATADIR"/types.db"); + + return (0); +} /* int cf_read */ + +/* Assures the config option is a string, duplicates it and returns the copy in + * "ret_string". If necessary "*ret_string" is freed first. Returns zero upon + * success. */ +int cf_util_get_string (const oconfig_item_t *ci, char **ret_string) /* {{{ */ +{ + char *string; + + if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) + { + ERROR ("cf_util_get_string: The %s option requires " + "exactly one string argument.", ci->key); + return (-1); + } + + string = strdup (ci->values[0].value.string); + if (string == NULL) + return (-1); + + if (*ret_string != NULL) + sfree (*ret_string); + *ret_string = string; + + return (0); +} /* }}} int cf_util_get_string */ + +/* Assures the config option is a string and copies it to the provided buffer. + * Assures null-termination. */ +int cf_util_get_string_buffer (const oconfig_item_t *ci, char *buffer, /* {{{ */ + size_t buffer_size) +{ + if ((ci == NULL) || (buffer == NULL) || (buffer_size < 1)) + return (EINVAL); + + if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) + { + ERROR ("cf_util_get_string_buffer: The %s option requires " + "exactly one string argument.", ci->key); + return (-1); + } + + strncpy (buffer, ci->values[0].value.string, buffer_size); + buffer[buffer_size - 1] = 0; + + return (0); +} /* }}} int cf_util_get_string_buffer */ + +/* Assures the config option is a number and returns it as an int. */ +int cf_util_get_int (const oconfig_item_t *ci, int *ret_value) /* {{{ */ +{ + if ((ci == NULL) || (ret_value == NULL)) + return (EINVAL); + + if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_NUMBER)) + { + ERROR ("cf_util_get_int: The %s option requires " + "exactly one numeric argument.", ci->key); + return (-1); + } + + *ret_value = (int) ci->values[0].value.number; + + return (0); +} /* }}} int cf_util_get_int */ + +int cf_util_get_double (const oconfig_item_t *ci, double *ret_value) /* {{{ */ +{ + if ((ci == NULL) || (ret_value == NULL)) + return (EINVAL); + + if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_NUMBER)) + { + ERROR ("cf_util_get_double: The %s option requires " + "exactly one numeric argument.", ci->key); + return (-1); + } + + *ret_value = ci->values[0].value.number; + + return (0); +} /* }}} int cf_util_get_double */ + +int cf_util_get_boolean (const oconfig_item_t *ci, _Bool *ret_bool) /* {{{ */ +{ + if ((ci == NULL) || (ret_bool == NULL)) + return (EINVAL); + + if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_BOOLEAN)) + { + ERROR ("cf_util_get_boolean: The %s option requires " + "exactly one boolean argument.", ci->key); + return (-1); + } + + *ret_bool = ci->values[0].value.boolean ? 1 : 0; + + return (0); +} /* }}} int cf_util_get_boolean */ + +int cf_util_get_flag (const oconfig_item_t *ci, /* {{{ */ + unsigned int *ret_value, unsigned int flag) +{ + int status; + _Bool b; + + if (ret_value == NULL) + return (EINVAL); + + b = 0; + status = cf_util_get_boolean (ci, &b); + if (status != 0) + return (status); + + if (b) + { + *ret_value |= flag; + } + else + { + *ret_value &= ~flag; + } + + return (0); +} /* }}} int cf_util_get_flag */ + +/* Assures that the config option is a string or a number if the correct range + * of 1-65535. The string is then converted to a port number using + * `service_name_to_port_number' and returned. + * Returns the port number in the range [1-65535] or less than zero upon + * failure. */ +int cf_util_get_port_number (const oconfig_item_t *ci) /* {{{ */ +{ + int tmp; + + if ((ci->values_num != 1) + || ((ci->values[0].type != OCONFIG_TYPE_STRING) + && (ci->values[0].type != OCONFIG_TYPE_NUMBER))) + { + ERROR ("cf_util_get_port_number: The \"%s\" option requires " + "exactly one string argument.", ci->key); + return (-1); + } + + if (ci->values[0].type == OCONFIG_TYPE_STRING) + return (service_name_to_port_number (ci->values[0].value.string)); + + assert (ci->values[0].type == OCONFIG_TYPE_NUMBER); + tmp = (int) (ci->values[0].value.number + 0.5); + if ((tmp < 1) || (tmp > 65535)) + { + ERROR ("cf_util_get_port_number: The \"%s\" option requires " + "a service name or a port number. The number " + "you specified, %i, is not in the valid " + "range of 1-65535.", + ci->key, tmp); + return (-1); + } + + return (tmp); +} /* }}} int cf_util_get_port_number */ + +int cf_util_get_service (const oconfig_item_t *ci, char **ret_string) /* {{{ */ +{ + int port; + char *service; + int status; + + if (ci->values_num != 1) + { + ERROR ("cf_util_get_service: The %s option requires exactly " + "one argument.", ci->key); + return (-1); + } + + if (ci->values[0].type == OCONFIG_TYPE_STRING) + return (cf_util_get_string (ci, ret_string)); + if (ci->values[0].type != OCONFIG_TYPE_NUMBER) + { + ERROR ("cf_util_get_service: The %s option requires " + "exactly one string or numeric argument.", + ci->key); + } + + port = 0; + status = cf_util_get_int (ci, &port); + if (status != 0) + return (status); + else if ((port < 1) || (port > 65535)) + { + ERROR ("cf_util_get_service: The port number given " + "for the %s option is out of " + "range (%i).", ci->key, port); + return (-1); + } + + service = malloc (6); + if (service == NULL) + { + ERROR ("cf_util_get_service: Out of memory."); + return (-1); + } + ssnprintf (service, 6, "%i", port); + + sfree (*ret_string); + *ret_string = service; + + return (0); +} /* }}} int cf_util_get_service */ + +int cf_util_get_cdtime (const oconfig_item_t *ci, cdtime_t *ret_value) /* {{{ */ +{ + if ((ci == NULL) || (ret_value == NULL)) + return (EINVAL); + + if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_NUMBER)) + { + ERROR ("cf_util_get_cdtime: The %s option requires " + "exactly one numeric argument.", ci->key); + return (-1); + } + + if (ci->values[0].value.number < 0.0) + { + ERROR ("cf_util_get_cdtime: The numeric argument of the %s " + "option must not be negative.", ci->key); + return (-1); + } + + *ret_value = DOUBLE_TO_CDTIME_T (ci->values[0].value.number); + + return (0); +} /* }}} int cf_util_get_cdtime */ + diff --cc src/daemon/plugin.c index 3d364458,00000000..bfd22017 mode 100644,000000..100644 --- a/src/daemon/plugin.c +++ b/src/daemon/plugin.c @@@ -1,2727 -1,0 +1,2732 @@@ +/** + * collectd - src/plugin.c + * Copyright (C) 2005-2014 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + * Sebastian Harl + **/ + +#include "collectd.h" +#include "common.h" +#include "plugin.h" +#include "configfile.h" +#include "filter_chain.h" +#include "utils_avltree.h" +#include "utils_cache.h" +#include "utils_complain.h" +#include "utils_llist.h" +#include "utils_heap.h" +#include "utils_time.h" +#include "utils_random.h" + +#include + +/* + * Private structures + */ +struct callback_func_s +{ + void *cf_callback; + user_data_t cf_udata; + plugin_ctx_t cf_ctx; +}; +typedef struct callback_func_s callback_func_t; + +#define RF_SIMPLE 0 +#define RF_COMPLEX 1 +#define RF_REMOVE 65535 +struct read_func_s +{ + /* `read_func_t' "inherits" from `callback_func_t'. + * The `rf_super' member MUST be the first one in this structure! */ +#define rf_callback rf_super.cf_callback +#define rf_udata rf_super.cf_udata +#define rf_ctx rf_super.cf_ctx + callback_func_t rf_super; + char rf_group[DATA_MAX_NAME_LEN]; + char *rf_name; + int rf_type; + cdtime_t rf_interval; + cdtime_t rf_effective_interval; + cdtime_t rf_next_read; +}; +typedef struct read_func_s read_func_t; + +struct write_queue_s; +typedef struct write_queue_s write_queue_t; +struct write_queue_s +{ + value_list_t *vl; + plugin_ctx_t ctx; + write_queue_t *next; +}; + +/* + * Private variables + */ +static c_avl_tree_t *plugins_loaded = NULL; + +static llist_t *list_init; +static llist_t *list_write; +static llist_t *list_flush; +static llist_t *list_missing; +static llist_t *list_shutdown; +static llist_t *list_log; +static llist_t *list_notification; + +static fc_chain_t *pre_cache_chain = NULL; +static fc_chain_t *post_cache_chain = NULL; + +static c_avl_tree_t *data_sets; + +static char *plugindir = NULL; + +#ifndef DEFAULT_MAX_READ_INTERVAL +# define DEFAULT_MAX_READ_INTERVAL TIME_T_TO_CDTIME_T (86400) +#endif +static c_heap_t *read_heap = NULL; +static llist_t *read_list; +static int read_loop = 1; +static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER; +static pthread_t *read_threads = NULL; +static int read_threads_num = 0; +static cdtime_t max_read_interval = DEFAULT_MAX_READ_INTERVAL; + +static write_queue_t *write_queue_head; +static write_queue_t *write_queue_tail; +static long write_queue_length = 0; +static _Bool write_loop = 1; +static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t write_cond = PTHREAD_COND_INITIALIZER; +static pthread_t *write_threads = NULL; +static size_t write_threads_num = 0; + +static pthread_key_t plugin_ctx_key; +static _Bool plugin_ctx_key_initialized = 0; + +static long write_limit_high = 0; +static long write_limit_low = 0; + +static derive_t stats_values_dropped = 0; +static _Bool record_statistics = 0; + +/* + * Static functions + */ +static int plugin_dispatch_values_internal (value_list_t *vl); + +static const char *plugin_get_dir (void) +{ + if (plugindir == NULL) + return (PLUGINDIR); + else + return (plugindir); +} + +static void plugin_update_internal_statistics (void) { /* {{{ */ + derive_t copy_write_queue_length; + value_list_t vl = VALUE_LIST_INIT; + value_t values[2]; + + copy_write_queue_length = write_queue_length; + + /* Initialize `vl' */ + vl.values = values; + vl.values_len = 2; + vl.time = 0; + sstrncpy (vl.host, hostname_g, sizeof (vl.host)); + sstrncpy (vl.plugin, "collectd", sizeof (vl.plugin)); + + vl.type_instance[0] = 0; + vl.values_len = 1; + + /* Write queue */ + sstrncpy (vl.plugin_instance, "write_queue", + sizeof (vl.plugin_instance)); + + /* Write queue : queue length */ + vl.values[0].gauge = (gauge_t) copy_write_queue_length; + sstrncpy (vl.type, "queue_length", sizeof (vl.type)); + vl.type_instance[0] = 0; + plugin_dispatch_values (&vl); + + /* Write queue : Values dropped (queue length > low limit) */ + vl.values[0].derive = (derive_t) stats_values_dropped; + sstrncpy (vl.type, "derive", sizeof (vl.type)); + sstrncpy (vl.type_instance, "dropped", sizeof (vl.type_instance)); + plugin_dispatch_values (&vl); + + /* Cache */ + sstrncpy (vl.plugin_instance, "cache", + sizeof (vl.plugin_instance)); + + /* Cache : Nb entry in cache tree */ + vl.values[0].gauge = (gauge_t) uc_get_size(); + sstrncpy (vl.type, "cache_size", sizeof (vl.type)); + vl.type_instance[0] = 0; + plugin_dispatch_values (&vl); + + return; +} /* }}} void plugin_update_internal_statistics */ + +static void destroy_callback (callback_func_t *cf) /* {{{ */ +{ + if (cf == NULL) + return; + + if ((cf->cf_udata.data != NULL) && (cf->cf_udata.free_func != NULL)) + { + cf->cf_udata.free_func (cf->cf_udata.data); + cf->cf_udata.data = NULL; + cf->cf_udata.free_func = NULL; + } + sfree (cf); +} /* }}} void destroy_callback */ + +static void destroy_all_callbacks (llist_t **list) /* {{{ */ +{ + llentry_t *le; + + if (*list == NULL) + return; + + le = llist_head (*list); + while (le != NULL) + { + llentry_t *le_next; + + le_next = le->next; + + sfree (le->key); + destroy_callback (le->value); + le->value = NULL; + + le = le_next; + } + + llist_destroy (*list); + *list = NULL; +} /* }}} void destroy_all_callbacks */ + +static void destroy_read_heap (void) /* {{{ */ +{ + if (read_heap == NULL) + return; + + while (42) + { + callback_func_t *cf; + + cf = c_heap_get_root (read_heap); + if (cf == NULL) + break; + + destroy_callback (cf); + } + + c_heap_destroy (read_heap); + read_heap = NULL; +} /* }}} void destroy_read_heap */ + +static int register_callback (llist_t **list, /* {{{ */ + const char *name, callback_func_t *cf) +{ + llentry_t *le; + char *key; + + if (*list == NULL) + { + *list = llist_create (); + if (*list == NULL) + { + ERROR ("plugin: register_callback: " + "llist_create failed."); + destroy_callback (cf); + return (-1); + } + } + + key = strdup (name); + if (key == NULL) + { + ERROR ("plugin: register_callback: strdup failed."); + destroy_callback (cf); + return (-1); + } + + le = llist_search (*list, name); + if (le == NULL) + { + le = llentry_create (key, cf); + if (le == NULL) + { + ERROR ("plugin: register_callback: " + "llentry_create failed."); + free (key); + destroy_callback (cf); + return (-1); + } + + llist_append (*list, le); + } + else + { + callback_func_t *old_cf; + + old_cf = le->value; + le->value = cf; + + WARNING ("plugin: register_callback: " + "a callback named `%s' already exists - " + "overwriting the old entry!", name); + + destroy_callback (old_cf); + sfree (key); + } + + return (0); +} /* }}} int register_callback */ + +static void log_list_callbacks (llist_t **list, /* {{{ */ + const char *comment) +{ + char *str; + int len; + llentry_t *le; + int i; + int n; + char **keys; + + n = llist_size(*list); + if (n == 0) + { + INFO("%s [none]", comment); + return; + } + + keys = calloc(n, sizeof(char*)); + + if (keys == NULL) + { + ERROR("%s: failed to allocate memory for list of callbacks", + comment); + + return; + } + + for (le = llist_head (*list), i = 0, len = 0; + le != NULL; + le = le->next, i++) + { + keys[i] = le->key; + len += strlen(le->key) + 6; + } + str = malloc(len + 10); + if (str == NULL) + { + ERROR("%s: failed to allocate memory for list of callbacks", + comment); + } + else + { + *str = '\0'; + strjoin(str, len, keys, n, "', '"); + INFO("%s ['%s']", comment, str); + free(str); + } + free(keys); +} /* }}} void log_list_callbacks */ + +static int create_register_callback (llist_t **list, /* {{{ */ + const char *name, void *callback, user_data_t *ud) +{ + callback_func_t *cf; + + cf = (callback_func_t *) malloc (sizeof (*cf)); + if (cf == NULL) + { + ERROR ("plugin: create_register_callback: malloc failed."); + return (-1); + } + memset (cf, 0, sizeof (*cf)); + + cf->cf_callback = callback; + if (ud == NULL) + { + cf->cf_udata.data = NULL; + cf->cf_udata.free_func = NULL; + } + else + { + cf->cf_udata = *ud; + } + + cf->cf_ctx = plugin_get_ctx (); + + return (register_callback (list, name, cf)); +} /* }}} int create_register_callback */ + +static int plugin_unregister (llist_t *list, const char *name) /* {{{ */ +{ + llentry_t *e; + + if (list == NULL) + return (-1); + + e = llist_search (list, name); + if (e == NULL) + return (-1); + + llist_remove (list, e); + + sfree (e->key); + destroy_callback (e->value); + + llentry_destroy (e); + + return (0); +} /* }}} int plugin_unregister */ + +/* + * (Try to) load the shared object `file'. Won't complain if it isn't a shared + * object, but it will bitch about a shared object not having a + * ``module_register'' symbol.. + */ +static int plugin_load_file (char *file, uint32_t flags) +{ + lt_dlhandle dlh; + void (*reg_handle) (void); + + lt_dlinit (); + lt_dlerror (); /* clear errors */ + +#if LIBTOOL_VERSION == 2 + if (flags & PLUGIN_FLAGS_GLOBAL) { + lt_dladvise advise; + lt_dladvise_init(&advise); + lt_dladvise_global(&advise); + dlh = lt_dlopenadvise(file, advise); + lt_dladvise_destroy(&advise); + } else { + dlh = lt_dlopen (file); + } +#else /* if LIBTOOL_VERSION == 1 */ + if (flags & PLUGIN_FLAGS_GLOBAL) + WARNING ("plugin_load_file: The global flag is not supported, " + "libtool 2 is required for this."); + dlh = lt_dlopen (file); +#endif + + if (dlh == NULL) + { + char errbuf[1024] = ""; + + ssnprintf (errbuf, sizeof (errbuf), + "lt_dlopen (\"%s\") failed: %s. " + "The most common cause for this problem is " + "missing dependencies. Use ldd(1) to check " + "the dependencies of the plugin " + "/ shared object.", + file, lt_dlerror ()); + + ERROR ("%s", errbuf); + /* Make sure this is printed to STDERR in any case, but also + * make sure it's printed only once. */ + if (list_log != NULL) + fprintf (stderr, "ERROR: %s\n", errbuf); + + return (1); + } + + if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL) + { + WARNING ("Couldn't find symbol \"module_register\" in \"%s\": %s\n", + file, lt_dlerror ()); + lt_dlclose (dlh); + return (-1); + } + + (*reg_handle) (); + + return (0); +} + +static void *plugin_read_thread (void __attribute__((unused)) *args) +{ + while (read_loop != 0) + { + read_func_t *rf; + plugin_ctx_t old_ctx; + cdtime_t now; + int status; + int rf_type; + int rc; + + /* Get the read function that needs to be read next. + * We don't need to hold "read_lock" for the heap, but we need + * to call c_heap_get_root() and pthread_cond_wait() in the + * same protected block. */ + pthread_mutex_lock (&read_lock); + rf = c_heap_get_root (read_heap); + if (rf == NULL) + { + pthread_cond_wait (&read_cond, &read_lock); + pthread_mutex_unlock (&read_lock); + continue; + } + pthread_mutex_unlock (&read_lock); + + if (rf->rf_interval == 0) + { + /* this should not happen, because the interval is set + * for each plugin when loading it + * XXX: issue a warning? */ + rf->rf_interval = plugin_get_interval (); + rf->rf_effective_interval = rf->rf_interval; + + rf->rf_next_read = cdtime (); + } + + /* sleep until this entry is due, + * using pthread_cond_timedwait */ + pthread_mutex_lock (&read_lock); + /* In pthread_cond_timedwait, spurious wakeups are possible + * (and really happen, at least on NetBSD with > 1 CPU), thus + * we need to re-evaluate the condition every time + * pthread_cond_timedwait returns. */ + rc = 0; + while ((read_loop != 0) + && (cdtime () < rf->rf_next_read) + && rc == 0) + { + struct timespec ts = { 0 }; + + CDTIME_T_TO_TIMESPEC (rf->rf_next_read, &ts); + + rc = pthread_cond_timedwait (&read_cond, &read_lock, + &ts); + } + + /* Must hold `read_lock' when accessing `rf->rf_type'. */ + rf_type = rf->rf_type; + pthread_mutex_unlock (&read_lock); + + /* Check if we're supposed to stop.. This may have interrupted + * the sleep, too. */ + if (read_loop == 0) + { + /* Insert `rf' again, so it can be free'd correctly */ + c_heap_insert (read_heap, rf); + break; + } + + /* The entry has been marked for deletion. The linked list + * entry has already been removed by `plugin_unregister_read'. + * All we have to do here is free the `read_func_t' and + * continue. */ + if (rf_type == RF_REMOVE) + { + DEBUG ("plugin_read_thread: Destroying the `%s' " + "callback.", rf->rf_name); + sfree (rf->rf_name); + destroy_callback ((callback_func_t *) rf); + rf = NULL; + continue; + } + + DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + + old_ctx = plugin_set_ctx (rf->rf_ctx); + + if (rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + assert (rf_type == RF_COMPLEX); + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + plugin_set_ctx (old_ctx); + + /* If the function signals failure, we will increase the + * intervals in which it will be called. */ + if (status != 0) + { + rf->rf_effective_interval *= 2; + if (rf->rf_effective_interval > max_read_interval) + rf->rf_effective_interval = max_read_interval; + + NOTICE ("read-function of plugin `%s' failed. " + "Will suspend it for %.3f seconds.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); + } + else + { + /* Success: Restore the interval, if it was changed. */ + rf->rf_effective_interval = rf->rf_interval; + } + + /* update the ``next read due'' field */ + now = cdtime (); + + DEBUG ("plugin_read_thread: Effective interval of the " + "%s plugin is %.3f seconds.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); + + /* Calculate the next (absolute) time at which this function + * should be called. */ + rf->rf_next_read += rf->rf_effective_interval; + + /* Check, if `rf_next_read' is in the past. */ + if (rf->rf_next_read < now) + { + /* `rf_next_read' is in the past. Insert `now' + * so this value doesn't trail off into the + * past too much. */ + rf->rf_next_read = now; + } + + DEBUG ("plugin_read_thread: Next read of the %s plugin at %.3f.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_next_read)); + + /* Re-insert this read function into the heap again. */ + c_heap_insert (read_heap, rf); + } /* while (read_loop) */ + + pthread_exit (NULL); + return ((void *) 0); +} /* void *plugin_read_thread */ + +static void start_read_threads (int num) +{ + int i; + + if (read_threads != NULL) + return; + + read_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); + if (read_threads == NULL) + { + ERROR ("plugin: start_read_threads: calloc failed."); + return; + } + + read_threads_num = 0; + for (i = 0; i < num; i++) + { + if (pthread_create (read_threads + read_threads_num, NULL, + plugin_read_thread, NULL) == 0) + { + read_threads_num++; + } + else + { + ERROR ("plugin: start_read_threads: pthread_create failed."); + return; + } + } /* for (i) */ +} /* void start_read_threads */ + +static void stop_read_threads (void) +{ + int i; + + if (read_threads == NULL) + return; + + INFO ("collectd: Stopping %i read threads.", read_threads_num); + + pthread_mutex_lock (&read_lock); + read_loop = 0; + DEBUG ("plugin: stop_read_threads: Signalling `read_cond'"); + pthread_cond_broadcast (&read_cond); + pthread_mutex_unlock (&read_lock); + + for (i = 0; i < read_threads_num; i++) + { + if (pthread_join (read_threads[i], NULL) != 0) + { + ERROR ("plugin: stop_read_threads: pthread_join failed."); + } + read_threads[i] = (pthread_t) 0; + } + sfree (read_threads); + read_threads_num = 0; +} /* void stop_read_threads */ + +static void plugin_value_list_free (value_list_t *vl) /* {{{ */ +{ + if (vl == NULL) + return; + + meta_data_destroy (vl->meta); + sfree (vl->values); + sfree (vl); +} /* }}} void plugin_value_list_free */ + +static value_list_t *plugin_value_list_clone (value_list_t const *vl_orig) /* {{{ */ +{ + value_list_t *vl; + + if (vl_orig == NULL) + return (NULL); + + vl = malloc (sizeof (*vl)); + if (vl == NULL) + return (NULL); + memcpy (vl, vl_orig, sizeof (*vl)); + + vl->values = calloc (vl_orig->values_len, sizeof (*vl->values)); + if (vl->values == NULL) + { + plugin_value_list_free (vl); + return (NULL); + } + memcpy (vl->values, vl_orig->values, + vl_orig->values_len * sizeof (*vl->values)); + + vl->meta = meta_data_clone (vl->meta); + if ((vl_orig->meta != NULL) && (vl->meta == NULL)) + { + plugin_value_list_free (vl); + return (NULL); + } + + if (vl->time == 0) + vl->time = cdtime (); + + /* Fill in the interval from the thread context, if it is zero. */ + if (vl->interval == 0) + { + plugin_ctx_t ctx = plugin_get_ctx (); + + if (ctx.interval != 0) + vl->interval = ctx.interval; + else + { + char name[6 * DATA_MAX_NAME_LEN]; + FORMAT_VL (name, sizeof (name), vl); + ERROR ("plugin_value_list_clone: Unable to determine " + "interval from context for " + "value list \"%s\". " + "This indicates a broken plugin. " + "Please report this problem to the " + "collectd mailing list or at " + ".", name); + vl->interval = cf_get_default_interval (); + } + } + + return (vl); +} /* }}} value_list_t *plugin_value_list_clone */ + +static int plugin_write_enqueue (value_list_t const *vl) /* {{{ */ +{ + write_queue_t *q; + + q = malloc (sizeof (*q)); + if (q == NULL) + return (ENOMEM); + q->next = NULL; + + q->vl = plugin_value_list_clone (vl); + if (q->vl == NULL) + { + sfree (q); + return (ENOMEM); + } + + /* Store context of caller (read plugin); otherwise, it would not be + * available to the write plugins when actually dispatching the + * value-list later on. */ + q->ctx = plugin_get_ctx (); + + pthread_mutex_lock (&write_lock); + + if (write_queue_tail == NULL) + { + write_queue_head = q; + write_queue_tail = q; + write_queue_length = 1; + } + else + { + write_queue_tail->next = q; + write_queue_tail = q; + write_queue_length += 1; + } + + pthread_cond_signal (&write_cond); + pthread_mutex_unlock (&write_lock); + + return (0); +} /* }}} int plugin_write_enqueue */ + +static value_list_t *plugin_write_dequeue (void) /* {{{ */ +{ + write_queue_t *q; + value_list_t *vl; + + pthread_mutex_lock (&write_lock); + + while (write_loop && (write_queue_head == NULL)) + pthread_cond_wait (&write_cond, &write_lock); + + if (write_queue_head == NULL) + { + pthread_mutex_unlock (&write_lock); + return (NULL); + } + + q = write_queue_head; + write_queue_head = q->next; + write_queue_length -= 1; + if (write_queue_head == NULL) { + write_queue_tail = NULL; + assert(0 == write_queue_length); + } + + pthread_mutex_unlock (&write_lock); + + (void) plugin_set_ctx (q->ctx); + + vl = q->vl; + sfree (q); + return (vl); +} /* }}} value_list_t *plugin_write_dequeue */ + +static void *plugin_write_thread (void __attribute__((unused)) *args) /* {{{ */ +{ + while (write_loop) + { + value_list_t *vl = plugin_write_dequeue (); + if (vl == NULL) + continue; + + plugin_dispatch_values_internal (vl); + + plugin_value_list_free (vl); + } + + pthread_exit (NULL); + return ((void *) 0); +} /* }}} void *plugin_write_thread */ + +static void start_write_threads (size_t num) /* {{{ */ +{ + size_t i; + + if (write_threads != NULL) + return; + + write_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); + if (write_threads == NULL) + { + ERROR ("plugin: start_write_threads: calloc failed."); + return; + } + + write_threads_num = 0; + for (i = 0; i < num; i++) + { + int status; + + status = pthread_create (write_threads + write_threads_num, + /* attr = */ NULL, + plugin_write_thread, + /* arg = */ NULL); + if (status != 0) + { + char errbuf[1024]; + ERROR ("plugin: start_write_threads: pthread_create failed " + "with status %i (%s).", status, + sstrerror (status, errbuf, sizeof (errbuf))); + return; + } + + write_threads_num++; + } /* for (i) */ +} /* }}} void start_write_threads */ + +static void stop_write_threads (void) /* {{{ */ +{ + write_queue_t *q; + int i; + + if (write_threads == NULL) + return; + + INFO ("collectd: Stopping %zu write threads.", write_threads_num); + + pthread_mutex_lock (&write_lock); + write_loop = 0; + DEBUG ("plugin: stop_write_threads: Signalling `write_cond'"); + pthread_cond_broadcast (&write_cond); + pthread_mutex_unlock (&write_lock); + + for (i = 0; i < write_threads_num; i++) + { + if (pthread_join (write_threads[i], NULL) != 0) + { + ERROR ("plugin: stop_write_threads: pthread_join failed."); + } + write_threads[i] = (pthread_t) 0; + } + sfree (write_threads); + write_threads_num = 0; + + pthread_mutex_lock (&write_lock); + i = 0; + for (q = write_queue_head; q != NULL; ) + { + write_queue_t *q1 = q; + plugin_value_list_free (q->vl); + q = q->next; + sfree (q1); + i++; + } + write_queue_head = NULL; + write_queue_tail = NULL; + write_queue_length = 0; + pthread_mutex_unlock (&write_lock); + + if (i > 0) + { + WARNING ("plugin: %i value list%s left after shutting down " + "the write threads.", + i, (i == 1) ? " was" : "s were"); + } +} /* }}} void stop_write_threads */ + +/* + * Public functions + */ +void plugin_set_dir (const char *dir) +{ + if (plugindir != NULL) + free (plugindir); + + if (dir == NULL) + plugindir = NULL; + else if ((plugindir = strdup (dir)) == NULL) + { + char errbuf[1024]; + ERROR ("strdup failed: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + } +} + +static _Bool plugin_is_loaded (char const *name) +{ + int status; + + if (plugins_loaded == NULL) + plugins_loaded = c_avl_create ((void *) strcasecmp); + assert (plugins_loaded != NULL); + + status = c_avl_get (plugins_loaded, name, /* ret_value = */ NULL); + return (status == 0); +} + +static int plugin_mark_loaded (char const *name) +{ + char *name_copy; + int status; + + name_copy = strdup (name); + if (name_copy == NULL) + return (ENOMEM); + + status = c_avl_insert (plugins_loaded, + /* key = */ name_copy, /* value = */ NULL); + return (status); +} + +static void plugin_free_loaded () +{ + void *key; + void *value; + + if (plugins_loaded == NULL) + return; + + while (c_avl_pick (plugins_loaded, &key, &value) == 0) + { + sfree (key); + assert (value == NULL); + } + + c_avl_destroy (plugins_loaded); + plugins_loaded = NULL; +} + +#define BUFSIZE 512 +int plugin_load (char const *plugin_name, uint32_t flags) +{ + DIR *dh; + const char *dir; + char filename[BUFSIZE] = ""; + char typename[BUFSIZE]; + int ret; + struct stat statbuf; + struct dirent *de; + int status; + + if (plugin_name == NULL) + return (EINVAL); + + /* Check if plugin is already loaded and don't do anything in this + * case. */ + if (plugin_is_loaded (plugin_name)) + return (0); + + dir = plugin_get_dir (); + ret = 1; + + /* + * XXX: Magic at work: + * + * Some of the language bindings, for example the Python and Perl + * plugins, need to be able to export symbols to the scripts they run. + * For this to happen, the "Globals" flag needs to be set. + * Unfortunately, this technical detail is hard to explain to the + * average user and she shouldn't have to worry about this, ideally. + * So in order to save everyone's sanity use a different default for a + * handful of special plugins. --octo + */ + if ((strcasecmp ("perl", plugin_name) == 0) + || (strcasecmp ("python", plugin_name) == 0)) + flags |= PLUGIN_FLAGS_GLOBAL; + + /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the + * type when matching the filename */ + status = ssnprintf (typename, sizeof (typename), "%s.so", plugin_name); + if ((status < 0) || ((size_t) status >= sizeof (typename))) + { + WARNING ("plugin_load: Filename too long: \"%s.so\"", plugin_name); + return (-1); + } + + if ((dh = opendir (dir)) == NULL) + { + char errbuf[1024]; + ERROR ("plugin_load: opendir (%s) failed: %s", dir, + sstrerror (errno, errbuf, sizeof (errbuf))); + return (-1); + } + + while ((de = readdir (dh)) != NULL) + { + if (strcasecmp (de->d_name, typename)) + continue; + + status = ssnprintf (filename, sizeof (filename), + "%s/%s", dir, de->d_name); + if ((status < 0) || ((size_t) status >= sizeof (filename))) + { + WARNING ("plugin_load: Filename too long: \"%s/%s\"", + dir, de->d_name); + continue; + } + + if (lstat (filename, &statbuf) == -1) + { + char errbuf[1024]; + WARNING ("plugin_load: stat (\"%s\") failed: %s", + filename, + sstrerror (errno, errbuf, sizeof (errbuf))); + continue; + } + else if (!S_ISREG (statbuf.st_mode)) + { + /* don't follow symlinks */ + WARNING ("plugin_load: %s is not a regular file.", + filename); + continue; + } + + status = plugin_load_file (filename, flags); + if (status == 0) + { + /* success */ + plugin_mark_loaded (plugin_name); + ret = 0; + break; + } + else + { + ERROR ("plugin_load: Load plugin \"%s\" failed with " + "status %i.", plugin_name, status); + } + } + + closedir (dh); + + if (filename[0] == 0) + ERROR ("plugin_load: Could not find plugin \"%s\" in %s", + plugin_name, dir); + + return (ret); +} + +/* + * The `register_*' functions follow + */ +int plugin_register_config (const char *name, + int (*callback) (const char *key, const char *val), + const char **keys, int keys_num) +{ + cf_register (name, callback, keys, keys_num); + return (0); +} /* int plugin_register_config */ + +int plugin_register_complex_config (const char *type, + int (*callback) (oconfig_item_t *)) +{ + return (cf_register_complex (type, callback)); +} /* int plugin_register_complex_config */ + +int plugin_register_init (const char *name, + int (*callback) (void)) +{ + return (create_register_callback (&list_init, name, (void *) callback, + /* user_data = */ NULL)); +} /* plugin_register_init */ + +static int plugin_compare_read_func (const void *arg0, const void *arg1) +{ + const read_func_t *rf0; + const read_func_t *rf1; + + rf0 = arg0; + rf1 = arg1; + + if (rf0->rf_next_read < rf1->rf_next_read) + return (-1); + else if (rf0->rf_next_read > rf1->rf_next_read) + return (1); + else + return (0); +} /* int plugin_compare_read_func */ + +/* Add a read function to both, the heap and a linked list. The linked list if + * used to look-up read functions, especially for the remove function. The heap + * is used to determine which plugin to read next. */ +static int plugin_insert_read (read_func_t *rf) +{ + int status; + llentry_t *le; + + rf->rf_next_read = cdtime (); + rf->rf_effective_interval = rf->rf_interval; + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + read_list = llist_create (); + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: read_list failed."); + return (-1); + } + } + + if (read_heap == NULL) + { + read_heap = c_heap_create (plugin_compare_read_func); + if (read_heap == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_create failed."); + return (-1); + } + } + + le = llist_search (read_list, rf->rf_name); + if (le != NULL) + { + pthread_mutex_unlock (&read_lock); + WARNING ("The read function \"%s\" is already registered. " + "Check for duplicate \"LoadPlugin\" lines " + "in your configuration!", + rf->rf_name); + return (EINVAL); + } + + le = llentry_create (rf->rf_name, rf); + if (le == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: llentry_create failed."); + return (-1); + } + + status = c_heap_insert (read_heap, rf); + if (status != 0) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_insert failed."); + llentry_destroy (le); + return (-1); + } + + /* This does not fail. */ + llist_append (read_list, le); + + /* Wake up all the read threads. */ + pthread_cond_broadcast (&read_cond); + pthread_mutex_unlock (&read_lock); + return (0); +} /* int plugin_insert_read */ + +int plugin_register_read (const char *name, + int (*callback) (void)) +{ + read_func_t *rf; + int status; + + rf = malloc (sizeof (*rf)); + if (rf == NULL) + { + ERROR ("plugin_register_read: malloc failed."); + return (ENOMEM); + } + + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + rf->rf_udata.data = NULL; + rf->rf_udata.free_func = NULL; + rf->rf_ctx = plugin_get_ctx (); + rf->rf_group[0] = '\0'; + rf->rf_name = strdup (name); + rf->rf_type = RF_SIMPLE; + rf->rf_interval = plugin_get_interval (); + + status = plugin_insert_read (rf); - if (status != 0) ++ if (status != 0) { ++ sfree (rf->rf_name); + sfree (rf); ++ } + + return (status); +} /* int plugin_register_read */ + +int plugin_register_complex_read (const char *group, const char *name, + plugin_read_cb callback, + const struct timespec *interval, + user_data_t *user_data) +{ + read_func_t *rf; + int status; + + rf = malloc (sizeof (*rf)); + if (rf == NULL) + { + ERROR ("plugin_register_complex_read: malloc failed."); + return (ENOMEM); + } + + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + if (group != NULL) + sstrncpy (rf->rf_group, group, sizeof (rf->rf_group)); + else + rf->rf_group[0] = '\0'; + rf->rf_name = strdup (name); + rf->rf_type = RF_COMPLEX; + if (interval != NULL) + rf->rf_interval = TIMESPEC_TO_CDTIME_T (interval); + else + rf->rf_interval = plugin_get_interval (); + + /* Set user data */ + if (user_data == NULL) + { + rf->rf_udata.data = NULL; + rf->rf_udata.free_func = NULL; + } + else + { + rf->rf_udata = *user_data; + } + + rf->rf_ctx = plugin_get_ctx (); + + status = plugin_insert_read (rf); - if (status != 0) ++ if (status != 0) { ++ sfree (rf->rf_name); + sfree (rf); ++ } + + return (status); +} /* int plugin_register_complex_read */ + +int plugin_register_write (const char *name, + plugin_write_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_write, name, + (void *) callback, ud)); +} /* int plugin_register_write */ + +int plugin_register_flush (const char *name, + plugin_flush_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_flush, name, + (void *) callback, ud)); +} /* int plugin_register_flush */ + +int plugin_register_missing (const char *name, + plugin_missing_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_missing, name, + (void *) callback, ud)); +} /* int plugin_register_missing */ + +int plugin_register_shutdown (const char *name, + int (*callback) (void)) +{ + return (create_register_callback (&list_shutdown, name, + (void *) callback, /* user_data = */ NULL)); +} /* int plugin_register_shutdown */ + +static void plugin_free_data_sets (void) +{ + void *key; + void *value; + + if (data_sets == NULL) + return; + + while (c_avl_pick (data_sets, &key, &value) == 0) + { + data_set_t *ds = value; + /* key is a pointer to ds->type */ + + sfree (ds->ds); + sfree (ds); + } + + c_avl_destroy (data_sets); + data_sets = NULL; +} /* void plugin_free_data_sets */ + +int plugin_register_data_set (const data_set_t *ds) +{ + data_set_t *ds_copy; + int i; + + if ((data_sets != NULL) + && (c_avl_get (data_sets, ds->type, NULL) == 0)) + { + NOTICE ("Replacing DS `%s' with another version.", ds->type); + plugin_unregister_data_set (ds->type); + } + else if (data_sets == NULL) + { + data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp); + if (data_sets == NULL) + return (-1); + } + + ds_copy = (data_set_t *) malloc (sizeof (data_set_t)); + if (ds_copy == NULL) + return (-1); + memcpy(ds_copy, ds, sizeof (data_set_t)); + + ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t) + * ds->ds_num); + if (ds_copy->ds == NULL) + { + free (ds_copy); + return (-1); + } + + for (i = 0; i < ds->ds_num; i++) + memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t)); + + return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy)); +} /* int plugin_register_data_set */ + +int plugin_register_log (const char *name, + plugin_log_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_log, name, + (void *) callback, ud)); +} /* int plugin_register_log */ + +int plugin_register_notification (const char *name, + plugin_notification_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_notification, name, + (void *) callback, ud)); +} /* int plugin_register_log */ + +int plugin_unregister_config (const char *name) +{ + cf_unregister (name); + return (0); +} /* int plugin_unregister_config */ + +int plugin_unregister_complex_config (const char *name) +{ + cf_unregister_complex (name); + return (0); +} /* int plugin_unregister_complex_config */ + +int plugin_unregister_init (const char *name) +{ + return (plugin_unregister (list_init, name)); +} + +int plugin_unregister_read (const char *name) /* {{{ */ +{ + llentry_t *le; + read_func_t *rf; + + if (name == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + return (-ENOENT); + } + + le = llist_search (read_list, name); + if (le == NULL) + { + pthread_mutex_unlock (&read_lock); + WARNING ("plugin_unregister_read: No such read function: %s", + name); + return (-ENOENT); + } + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + pthread_mutex_unlock (&read_lock); + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read: Marked `%s' for removal.", name); + + return (0); +} /* }}} int plugin_unregister_read */ + +void plugin_log_available_writers (void) +{ + log_list_callbacks (&list_write, "Available write targets:"); +} + +static int compare_read_func_group (llentry_t *e, void *ud) /* {{{ */ +{ + read_func_t *rf = e->value; + char *group = ud; + + return strcmp (rf->rf_group, (const char *)group); +} /* }}} int compare_read_func_group */ + +int plugin_unregister_read_group (const char *group) /* {{{ */ +{ + llentry_t *le; + read_func_t *rf; + + int found = 0; + + if (group == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + return (-ENOENT); + } + + while (42) + { + le = llist_search_custom (read_list, + compare_read_func_group, (void *)group); + + if (le == NULL) + break; + + ++found; + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read_group: " + "Marked `%s' (group `%s') for removal.", + rf->rf_name, group); + } + + pthread_mutex_unlock (&read_lock); + + if (found == 0) + { + WARNING ("plugin_unregister_read_group: No such " + "group of read function: %s", group); + return (-ENOENT); + } + + return (0); +} /* }}} int plugin_unregister_read_group */ + +int plugin_unregister_write (const char *name) +{ + return (plugin_unregister (list_write, name)); +} + +int plugin_unregister_flush (const char *name) +{ + return (plugin_unregister (list_flush, name)); +} + +int plugin_unregister_missing (const char *name) +{ + return (plugin_unregister (list_missing, name)); +} + +int plugin_unregister_shutdown (const char *name) +{ + return (plugin_unregister (list_shutdown, name)); +} + +int plugin_unregister_data_set (const char *name) +{ + data_set_t *ds; + + if (data_sets == NULL) + return (-1); + + if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0) + return (-1); + + sfree (ds->ds); + sfree (ds); + + return (0); +} /* int plugin_unregister_data_set */ + +int plugin_unregister_log (const char *name) +{ + return (plugin_unregister (list_log, name)); +} + +int plugin_unregister_notification (const char *name) +{ + return (plugin_unregister (list_notification, name)); +} + +void plugin_init_all (void) +{ + char const *chain_name; + long write_threads_num; + llentry_t *le; + int status; + + /* Init the value cache */ + uc_init (); + + if (IS_TRUE (global_option_get ("CollectInternalStats"))) + record_statistics = 1; + + chain_name = global_option_get ("PreCacheChain"); + pre_cache_chain = fc_chain_get_by_name (chain_name); + + chain_name = global_option_get ("PostCacheChain"); + post_cache_chain = fc_chain_get_by_name (chain_name); + + write_limit_high = global_option_get_long ("WriteQueueLimitHigh", + /* default = */ 0); + if (write_limit_high < 0) + { + ERROR ("WriteQueueLimitHigh must be positive or zero."); + write_limit_high = 0; + } + + write_limit_low = global_option_get_long ("WriteQueueLimitLow", + /* default = */ write_limit_high / 2); + if (write_limit_low < 0) + { + ERROR ("WriteQueueLimitLow must be positive or zero."); + write_limit_low = write_limit_high / 2; + } + else if (write_limit_low > write_limit_high) + { + ERROR ("WriteQueueLimitLow must not be larger than " + "WriteQueueLimitHigh."); + write_limit_low = write_limit_high; + } + + write_threads_num = global_option_get_long ("WriteThreads", + /* default = */ 5); + if (write_threads_num < 1) + { + ERROR ("WriteThreads must be positive."); + write_threads_num = 5; + } + + start_write_threads ((size_t) write_threads_num); + + if ((list_init == NULL) && (read_heap == NULL)) + return; + + /* Calling all init callbacks before checking if read callbacks + * are available allows the init callbacks to register the read + * callback. */ + le = llist_head (list_init); + while (le != NULL) + { + callback_func_t *cf; + plugin_init_cb callback; + plugin_ctx_t old_ctx; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + status = (*callback) (); + plugin_set_ctx (old_ctx); + + if (status != 0) + { + ERROR ("Initialization of plugin `%s' " + "failed with status %i. " + "Plugin will be unloaded.", + le->key, status); + /* Plugins that register read callbacks from the init + * callback should take care of appropriate error + * handling themselves. */ + /* FIXME: Unload _all_ functions */ + plugin_unregister_read (le->key); + } + + le = le->next; + } + + max_read_interval = global_option_get_time ("MaxReadInterval", + DEFAULT_MAX_READ_INTERVAL); + + /* Start read-threads */ + if (read_heap != NULL) + { + const char *rt; + int num; + + rt = global_option_get ("ReadThreads"); + num = atoi (rt); + if (num != -1) + start_read_threads ((num > 0) ? num : 5); + } +} /* void plugin_init_all */ + +/* TODO: Rename this function. */ +void plugin_read_all (void) +{ + if(record_statistics) { + plugin_update_internal_statistics (); + } + uc_check_timeout (); + + return; +} /* void plugin_read_all */ + +/* Read function called when the `-T' command line argument is given. */ +int plugin_read_all_once (void) +{ + int status; + int return_status = 0; + + if (read_heap == NULL) + { + NOTICE ("No read-functions are registered."); + return (0); + } + + while (42) + { + read_func_t *rf; + plugin_ctx_t old_ctx; + + rf = c_heap_get_root (read_heap); + if (rf == NULL) + break; + + old_ctx = plugin_set_ctx (rf->rf_ctx); + + if (rf->rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + plugin_set_ctx (old_ctx); + + if (status != 0) + { + NOTICE ("read-function of plugin `%s' failed.", + rf->rf_name); + return_status = -1; + } + ++ sfree (rf->rf_name); + destroy_callback ((void *) rf); + } + + return (return_status); +} /* int plugin_read_all_once */ + +int plugin_write (const char *plugin, /* {{{ */ + const data_set_t *ds, const value_list_t *vl) +{ + llentry_t *le; + int status; + + if (vl == NULL) + return (EINVAL); + + if (list_write == NULL) + return (ENOENT); + + if (ds == NULL) + { + ds = plugin_get_ds (vl->type); + if (ds == NULL) + { + ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type); + return (ENOENT); + } + } + + if (plugin == NULL) + { + int success = 0; + int failure = 0; + + le = llist_head (list_write); + while (le != NULL) + { + callback_func_t *cf = le->value; + plugin_write_cb callback; + + /* do not switch plugin context; rather keep the context (interval) + * information of the calling read plugin */ + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); + if (status != 0) + failure++; + else + success++; + + le = le->next; + } + + if ((success == 0) && (failure != 0)) + status = -1; + else + status = 0; + } + else /* plugin != NULL */ + { + callback_func_t *cf; + plugin_write_cb callback; + + le = llist_head (list_write); + while (le != NULL) + { + if (strcasecmp (plugin, le->key) == 0) + break; + + le = le->next; + } + + if (le == NULL) + return (ENOENT); + + cf = le->value; + + /* do not switch plugin context; rather keep the context (interval) + * information of the calling read plugin */ + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); + } + + return (status); +} /* }}} int plugin_write */ + +int plugin_flush (const char *plugin, cdtime_t timeout, const char *identifier) +{ + llentry_t *le; + + if (list_flush == NULL) + return (0); + + le = llist_head (list_flush); + while (le != NULL) + { + callback_func_t *cf; + plugin_flush_cb callback; + plugin_ctx_t old_ctx; + + if ((plugin != NULL) + && (strcmp (plugin, le->key) != 0)) + { + le = le->next; + continue; + } + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + (*callback) (timeout, identifier, &cf->cf_udata); + + plugin_set_ctx (old_ctx); + + le = le->next; + } + return (0); +} /* int plugin_flush */ + +void plugin_shutdown_all (void) +{ + llentry_t *le; + + stop_read_threads (); + + destroy_all_callbacks (&list_init); + + pthread_mutex_lock (&read_lock); + llist_destroy (read_list); + read_list = NULL; + pthread_mutex_unlock (&read_lock); + + destroy_read_heap (); + + plugin_flush (/* plugin = */ NULL, + /* timeout = */ 0, + /* identifier = */ NULL); + + le = NULL; + if (list_shutdown != NULL) + le = llist_head (list_shutdown); + + while (le != NULL) + { + callback_func_t *cf; + plugin_shutdown_cb callback; + plugin_ctx_t old_ctx; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + /* Advance the pointer before calling the callback allows + * shutdown functions to unregister themselves. If done the + * other way around the memory `le' points to will be freed + * after callback returns. */ + le = le->next; + + (*callback) (); + + plugin_set_ctx (old_ctx); + } + + stop_write_threads (); + + /* Write plugins which use the `user_data' pointer usually need the + * same data available to the flush callback. If this is the case, set + * the free_function to NULL when registering the flush callback and to + * the real free function when registering the write callback. This way + * the data isn't freed twice. */ + destroy_all_callbacks (&list_flush); + destroy_all_callbacks (&list_missing); + destroy_all_callbacks (&list_write); + + destroy_all_callbacks (&list_notification); + destroy_all_callbacks (&list_shutdown); + destroy_all_callbacks (&list_log); + + plugin_free_loaded (); + plugin_free_data_sets (); +} /* void plugin_shutdown_all */ + +int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */ +{ + llentry_t *le; + + if (list_missing == NULL) + return (0); + + le = llist_head (list_missing); + while (le != NULL) + { + callback_func_t *cf; + plugin_missing_cb callback; + plugin_ctx_t old_ctx; + int status; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + status = (*callback) (vl, &cf->cf_udata); + plugin_set_ctx (old_ctx); + if (status != 0) + { + if (status < 0) + { + ERROR ("plugin_dispatch_missing: Callback function \"%s\" " + "failed with status %i.", + le->key, status); + return (status); + } + else + { + return (0); + } + } + + le = le->next; + } + return (0); +} /* int }}} plugin_dispatch_missing */ + +static int plugin_dispatch_values_internal (value_list_t *vl) +{ + int status; + static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC; + + value_t *saved_values; + int saved_values_len; + + data_set_t *ds; + + int free_meta_data = 0; + + if ((vl == NULL) || (vl->type[0] == 0) + || (vl->values == NULL) || (vl->values_len < 1)) + { + ERROR ("plugin_dispatch_values: Invalid value list " + "from plugin %s.", vl->plugin); + return (-1); + } + + /* Free meta data only if the calling function didn't specify any. In + * this case matches and targets may add some and the calling function + * may not expect (and therefore free) that data. */ + if (vl->meta == NULL) + free_meta_data = 1; + + if (list_write == NULL) + c_complain_once (LOG_WARNING, &no_write_complaint, + "plugin_dispatch_values: No write callback has been " + "registered. Please load at least one output plugin, " + "if you want the collected data to be stored."); + + if (data_sets == NULL) + { + ERROR ("plugin_dispatch_values: No data sets registered. " + "Could the types database be read? Check " + "your `TypesDB' setting!"); + return (-1); + } + + if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0) + { + char ident[6 * DATA_MAX_NAME_LEN]; + + FORMAT_VL (ident, sizeof (ident), vl); + INFO ("plugin_dispatch_values: Dataset not found: %s " + "(from \"%s\"), check your types.db!", + vl->type, ident); + return (-1); + } + + /* Assured by plugin_value_list_clone(). The time is determined at + * _enqueue_ time. */ + assert (vl->time != 0); + assert (vl->interval != 0); + + DEBUG ("plugin_dispatch_values: time = %.3f; interval = %.3f; " + "host = %s; " + "plugin = %s; plugin_instance = %s; " + "type = %s; type_instance = %s;", + CDTIME_T_TO_DOUBLE (vl->time), + CDTIME_T_TO_DOUBLE (vl->interval), + vl->host, + vl->plugin, vl->plugin_instance, + vl->type, vl->type_instance); + +#if COLLECT_DEBUG + assert (0 == strcmp (ds->type, vl->type)); +#else + if (0 != strcmp (ds->type, vl->type)) + WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)", + ds->type, vl->type); +#endif + +#if COLLECT_DEBUG + assert (ds->ds_num == vl->values_len); +#else + if (ds->ds_num != vl->values_len) + { + ERROR ("plugin_dispatch_values: ds->type = %s: " + "(ds->ds_num = %i) != " + "(vl->values_len = %i)", + ds->type, ds->ds_num, vl->values_len); + return (-1); + } +#endif + + escape_slashes (vl->host, sizeof (vl->host)); + escape_slashes (vl->plugin, sizeof (vl->plugin)); + escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance)); + escape_slashes (vl->type, sizeof (vl->type)); + escape_slashes (vl->type_instance, sizeof (vl->type_instance)); + + /* Copy the values. This way, we can assure `targets' that they get + * dynamically allocated values, which they can free and replace if + * they like. */ + if ((pre_cache_chain != NULL) || (post_cache_chain != NULL)) + { + saved_values = vl->values; + saved_values_len = vl->values_len; + + vl->values = (value_t *) calloc (vl->values_len, + sizeof (*vl->values)); + if (vl->values == NULL) + { + ERROR ("plugin_dispatch_values: calloc failed."); + vl->values = saved_values; + return (-1); + } + memcpy (vl->values, saved_values, + vl->values_len * sizeof (*vl->values)); + } + else /* if ((pre == NULL) && (post == NULL)) */ + { + saved_values = NULL; + saved_values_len = 0; + } + + if (pre_cache_chain != NULL) + { + status = fc_process_chain (ds, vl, pre_cache_chain); + if (status < 0) + { + WARNING ("plugin_dispatch_values: Running the " + "pre-cache chain failed with " + "status %i (%#x).", + status, status); + } + else if (status == FC_TARGET_STOP) + { + /* Restore the state of the value_list so that plugins + * don't get confused.. */ + if (saved_values != NULL) + { + free (vl->values); + vl->values = saved_values; + vl->values_len = saved_values_len; + } + return (0); + } + } + + /* Update the value cache */ + uc_update (ds, vl); + + if (post_cache_chain != NULL) + { + status = fc_process_chain (ds, vl, post_cache_chain); + if (status < 0) + { + WARNING ("plugin_dispatch_values: Running the " + "post-cache chain failed with " + "status %i (%#x).", + status, status); + } + } + else + fc_default_action (ds, vl); + + /* Restore the state of the value_list so that plugins don't get + * confused.. */ + if (saved_values != NULL) + { + free (vl->values); + vl->values = saved_values; + vl->values_len = saved_values_len; + } + + if ((free_meta_data != 0) && (vl->meta != NULL)) + { + meta_data_destroy (vl->meta); + vl->meta = NULL; + } + + return (0); +} /* int plugin_dispatch_values_internal */ + +static double get_drop_probability (void) /* {{{ */ +{ + long pos; + long size; + long wql; + + pthread_mutex_lock (&write_lock); + wql = write_queue_length; + pthread_mutex_unlock (&write_lock); + + if (wql < write_limit_low) + return (0.0); + if (wql >= write_limit_high) + return (1.0); + + pos = 1 + wql - write_limit_low; + size = 1 + write_limit_high - write_limit_low; + + return (((double) pos) / ((double) size)); +} /* }}} double get_drop_probability */ + +static _Bool check_drop_value (void) /* {{{ */ +{ + static cdtime_t last_message_time = 0; + static pthread_mutex_t last_message_lock = PTHREAD_MUTEX_INITIALIZER; + + double p; + double q; + int status; + + if (write_limit_high == 0) + return (0); + + p = get_drop_probability (); + if (p == 0.0) + return (0); + + status = pthread_mutex_trylock (&last_message_lock); + if (status == 0) + { + cdtime_t now; + + now = cdtime (); + if ((now - last_message_time) > TIME_T_TO_CDTIME_T (1)) + { + last_message_time = now; + ERROR ("plugin_dispatch_values: Low water mark " + "reached. Dropping %.0f%% of metrics.", + 100.0 * p); + } + pthread_mutex_unlock (&last_message_lock); + } + + if (p == 1.0) + return (1); + + q = cdrand_d (); + if (q > p) + return (1); + else + return (0); +} /* }}} _Bool check_drop_value */ + +int plugin_dispatch_values (value_list_t const *vl) +{ + int status; + static pthread_mutex_t statistics_lock = PTHREAD_MUTEX_INITIALIZER; + + if (check_drop_value ()) { + if(record_statistics) { + pthread_mutex_lock(&statistics_lock); + stats_values_dropped++; + pthread_mutex_unlock(&statistics_lock); + } + return (0); + } + + status = plugin_write_enqueue (vl); + if (status != 0) + { + char errbuf[1024]; + ERROR ("plugin_dispatch_values: plugin_write_enqueue failed " + "with status %i (%s).", status, + sstrerror (status, errbuf, sizeof (errbuf))); + return (status); + } + + return (0); +} + +__attribute__((sentinel)) +int plugin_dispatch_multivalue (value_list_t const *template, /* {{{ */ + _Bool store_percentage, int store_type, ...) +{ + value_list_t *vl; + int failed = 0; + gauge_t sum = 0.0; + va_list ap; + + assert (template->values_len == 1); + + /* Calculate sum for Gauge to calculate percent if needed */ + if (DS_TYPE_GAUGE == store_type) { + va_start (ap, store_type); + while (42) + { + char const *name; + gauge_t value; + + name = va_arg (ap, char const *); + if (name == NULL) + break; + + value = va_arg (ap, gauge_t); + if (!isnan (value)) + sum += value; + } + va_end (ap); + } + + + vl = plugin_value_list_clone (template); + /* plugin_value_list_clone makes sure vl->time is set to non-zero. */ + if (store_percentage) + sstrncpy (vl->type, "percent", sizeof (vl->type)); + + va_start (ap, store_type); + while (42) + { + char const *name; + int status; + + /* Set the type instance. */ + name = va_arg (ap, char const *); + if (name == NULL) + break; + sstrncpy (vl->type_instance, name, sizeof (vl->type_instance)); + + /* Set the value. */ + switch (store_type) + { + case DS_TYPE_GAUGE: + vl->values[0].gauge = va_arg (ap, gauge_t); + if (store_percentage) + vl->values[0].gauge *= 100.0 / sum; + break; + case DS_TYPE_ABSOLUTE: + vl->values[0].absolute = va_arg (ap, absolute_t); + break; + case DS_TYPE_COUNTER: + vl->values[0].counter = va_arg (ap, counter_t); + break; + case DS_TYPE_DERIVE: + vl->values[0].derive = va_arg (ap, derive_t); + break; + default: + ERROR ("plugin_dispatch_multivalue: given store_type is incorrect."); + failed++; + } + + + status = plugin_write_enqueue (vl); + if (status != 0) + failed++; + } + va_end (ap); + + plugin_value_list_free (vl); + return (failed); +} /* }}} int plugin_dispatch_multivalue */ + +int plugin_dispatch_notification (const notification_t *notif) +{ + llentry_t *le; + /* Possible TODO: Add flap detection here */ + + DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; " + "time = %.3f; host = %s;", + notif->severity, notif->message, + CDTIME_T_TO_DOUBLE (notif->time), notif->host); + + /* Nobody cares for notifications */ + if (list_notification == NULL) + return (-1); + + le = llist_head (list_notification); + while (le != NULL) + { + callback_func_t *cf; + plugin_notification_cb callback; + int status; + + /* do not switch plugin context; rather keep the context + * (interval) information of the calling plugin */ + + cf = le->value; + callback = cf->cf_callback; + status = (*callback) (notif, &cf->cf_udata); + if (status != 0) + { + WARNING ("plugin_dispatch_notification: Notification " + "callback %s returned %i.", + le->key, status); + } + + le = le->next; + } + + return (0); +} /* int plugin_dispatch_notification */ + +void plugin_log (int level, const char *format, ...) +{ + char msg[1024]; + va_list ap; + llentry_t *le; + +#if !COLLECT_DEBUG + if (level >= LOG_DEBUG) + return; +#endif + + va_start (ap, format); + vsnprintf (msg, sizeof (msg), format, ap); + msg[sizeof (msg) - 1] = '\0'; + va_end (ap); + + if (list_log == NULL) + { + fprintf (stderr, "%s\n", msg); + return; + } + + le = llist_head (list_log); + while (le != NULL) + { + callback_func_t *cf; + plugin_log_cb callback; + + cf = le->value; + callback = cf->cf_callback; + + /* do not switch plugin context; rather keep the context + * (interval) information of the calling plugin */ + + (*callback) (level, msg, &cf->cf_udata); + + le = le->next; + } +} /* void plugin_log */ + +int parse_log_severity (const char *severity) +{ + int log_level = -1; + + if ((0 == strcasecmp (severity, "emerg")) + || (0 == strcasecmp (severity, "alert")) + || (0 == strcasecmp (severity, "crit")) + || (0 == strcasecmp (severity, "err"))) + log_level = LOG_ERR; + else if (0 == strcasecmp (severity, "warning")) + log_level = LOG_WARNING; + else if (0 == strcasecmp (severity, "notice")) + log_level = LOG_NOTICE; + else if (0 == strcasecmp (severity, "info")) + log_level = LOG_INFO; +#if COLLECT_DEBUG + else if (0 == strcasecmp (severity, "debug")) + log_level = LOG_DEBUG; +#endif /* COLLECT_DEBUG */ + + return (log_level); +} /* int parse_log_severity */ + +int parse_notif_severity (const char *severity) +{ + int notif_severity = -1; + + if (strcasecmp (severity, "FAILURE") == 0) + notif_severity = NOTIF_FAILURE; + else if (strcmp (severity, "OKAY") == 0) + notif_severity = NOTIF_OKAY; + else if ((strcmp (severity, "WARNING") == 0) + || (strcmp (severity, "WARN") == 0)) + notif_severity = NOTIF_WARNING; + + return (notif_severity); +} /* int parse_notif_severity */ + +const data_set_t *plugin_get_ds (const char *name) +{ + data_set_t *ds; + + if (data_sets == NULL) + { + ERROR ("plugin_get_ds: No data sets are defined yet."); + return (NULL); + } + + if (c_avl_get (data_sets, name, (void *) &ds) != 0) + { + DEBUG ("No such dataset registered: %s", name); + return (NULL); + } + + return (ds); +} /* data_set_t *plugin_get_ds */ + +static int plugin_notification_meta_add (notification_t *n, + const char *name, + enum notification_meta_type_e type, + const void *value) +{ + notification_meta_t *meta; + notification_meta_t *tail; + + if ((n == NULL) || (name == NULL) || (value == NULL)) + { + ERROR ("plugin_notification_meta_add: A pointer is NULL!"); + return (-1); + } + + meta = (notification_meta_t *) malloc (sizeof (notification_meta_t)); + if (meta == NULL) + { + ERROR ("plugin_notification_meta_add: malloc failed."); + return (-1); + } + memset (meta, 0, sizeof (notification_meta_t)); + + sstrncpy (meta->name, name, sizeof (meta->name)); + meta->type = type; + + switch (type) + { + case NM_TYPE_STRING: + { + meta->nm_value.nm_string = strdup ((const char *) value); + if (meta->nm_value.nm_string == NULL) + { + ERROR ("plugin_notification_meta_add: strdup failed."); + sfree (meta); + return (-1); + } + break; + } + case NM_TYPE_SIGNED_INT: + { + meta->nm_value.nm_signed_int = *((int64_t *) value); + break; + } + case NM_TYPE_UNSIGNED_INT: + { + meta->nm_value.nm_unsigned_int = *((uint64_t *) value); + break; + } + case NM_TYPE_DOUBLE: + { + meta->nm_value.nm_double = *((double *) value); + break; + } + case NM_TYPE_BOOLEAN: + { + meta->nm_value.nm_boolean = *((_Bool *) value); + break; + } + default: + { + ERROR ("plugin_notification_meta_add: Unknown type: %i", type); + sfree (meta); + return (-1); + } + } /* switch (type) */ + + meta->next = NULL; + tail = n->meta; + while ((tail != NULL) && (tail->next != NULL)) + tail = tail->next; + + if (tail == NULL) + n->meta = meta; + else + tail->next = meta; + + return (0); +} /* int plugin_notification_meta_add */ + +int plugin_notification_meta_add_string (notification_t *n, + const char *name, + const char *value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value)); +} + +int plugin_notification_meta_add_signed_int (notification_t *n, + const char *name, + int64_t value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value)); +} + +int plugin_notification_meta_add_unsigned_int (notification_t *n, + const char *name, + uint64_t value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value)); +} + +int plugin_notification_meta_add_double (notification_t *n, + const char *name, + double value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value)); +} + +int plugin_notification_meta_add_boolean (notification_t *n, + const char *name, + _Bool value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value)); +} + +int plugin_notification_meta_copy (notification_t *dst, + const notification_t *src) +{ + notification_meta_t *meta; + + assert (dst != NULL); + assert (src != NULL); + assert (dst != src); + assert ((src->meta == NULL) || (src->meta != dst->meta)); + + for (meta = src->meta; meta != NULL; meta = meta->next) + { + if (meta->type == NM_TYPE_STRING) + plugin_notification_meta_add_string (dst, meta->name, + meta->nm_value.nm_string); + else if (meta->type == NM_TYPE_SIGNED_INT) + plugin_notification_meta_add_signed_int (dst, meta->name, + meta->nm_value.nm_signed_int); + else if (meta->type == NM_TYPE_UNSIGNED_INT) + plugin_notification_meta_add_unsigned_int (dst, meta->name, + meta->nm_value.nm_unsigned_int); + else if (meta->type == NM_TYPE_DOUBLE) + plugin_notification_meta_add_double (dst, meta->name, + meta->nm_value.nm_double); + else if (meta->type == NM_TYPE_BOOLEAN) + plugin_notification_meta_add_boolean (dst, meta->name, + meta->nm_value.nm_boolean); + } + + return (0); +} /* int plugin_notification_meta_copy */ + +int plugin_notification_meta_free (notification_meta_t *n) +{ + notification_meta_t *this; + notification_meta_t *next; + + if (n == NULL) + { + ERROR ("plugin_notification_meta_free: n == NULL!"); + return (-1); + } + + this = n; + while (this != NULL) + { + next = this->next; + + if (this->type == NM_TYPE_STRING) + { + free ((char *)this->nm_value.nm_string); + this->nm_value.nm_string = NULL; + } + sfree (this); + + this = next; + } + + return (0); +} /* int plugin_notification_meta_free */ + +static void plugin_ctx_destructor (void *ctx) +{ + sfree (ctx); +} /* void plugin_ctx_destructor */ + +static plugin_ctx_t ctx_init = { /* interval = */ 0 }; + +static plugin_ctx_t *plugin_ctx_create (void) +{ + plugin_ctx_t *ctx; + + ctx = malloc (sizeof (*ctx)); + if (ctx == NULL) { + char errbuf[1024]; + ERROR ("Failed to allocate plugin context: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return NULL; + } + + *ctx = ctx_init; + assert (plugin_ctx_key_initialized); + pthread_setspecific (plugin_ctx_key, ctx); + DEBUG("Created new plugin context."); + return (ctx); +} /* int plugin_ctx_create */ + +void plugin_init_ctx (void) +{ + pthread_key_create (&plugin_ctx_key, plugin_ctx_destructor); + plugin_ctx_key_initialized = 1; +} /* void plugin_init_ctx */ + +plugin_ctx_t plugin_get_ctx (void) +{ + plugin_ctx_t *ctx; + + assert (plugin_ctx_key_initialized); + ctx = pthread_getspecific (plugin_ctx_key); + + if (ctx == NULL) { + ctx = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (ctx == NULL) + return ctx_init; + } + + return (*ctx); +} /* plugin_ctx_t plugin_get_ctx */ + +plugin_ctx_t plugin_set_ctx (plugin_ctx_t ctx) +{ + plugin_ctx_t *c; + plugin_ctx_t old; + + assert (plugin_ctx_key_initialized); + c = pthread_getspecific (plugin_ctx_key); + + if (c == NULL) { + c = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (c == NULL) + return ctx_init; + } + + old = *c; + *c = ctx; + + return (old); +} /* void plugin_set_ctx */ + +cdtime_t plugin_get_interval (void) +{ + cdtime_t interval; + + interval = plugin_get_ctx().interval; + if (interval > 0) + return interval; + + return cf_get_default_interval (); +} /* cdtime_t plugin_get_interval */ + +typedef struct { + plugin_ctx_t ctx; + void *(*start_routine) (void *); + void *arg; +} plugin_thread_t; + +static void *plugin_thread_start (void *arg) +{ + plugin_thread_t *plugin_thread = arg; + + void *(*start_routine) (void *) = plugin_thread->start_routine; + void *plugin_arg = plugin_thread->arg; + + plugin_set_ctx (plugin_thread->ctx); + + free (plugin_thread); + + return start_routine (plugin_arg); +} /* void *plugin_thread_start */ + +int plugin_thread_create (pthread_t *thread, const pthread_attr_t *attr, + void *(*start_routine) (void *), void *arg) +{ + plugin_thread_t *plugin_thread; + + plugin_thread = malloc (sizeof (*plugin_thread)); + if (plugin_thread == NULL) + return -1; + + plugin_thread->ctx = plugin_get_ctx (); + plugin_thread->start_routine = start_routine; + plugin_thread->arg = arg; + + return pthread_create (thread, attr, + plugin_thread_start, plugin_thread); +} /* int plugin_thread_create */ + +/* vim: set sw=8 ts=8 noet fdm=marker : */ diff --cc src/daemon/utils_cache.c index 7b2f9583,00000000..76213955 mode 100644,000000..100644 --- a/src/daemon/utils_cache.c +++ b/src/daemon/utils_cache.c @@@ -1,996 -1,0 +1,997 @@@ +/** + * collectd - src/utils_cache.c + * Copyright (C) 2007-2010 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + **/ + +#include "collectd.h" +#include "common.h" +#include "plugin.h" +#include "utils_avltree.h" +#include "utils_cache.h" +#include "meta_data.h" + +#include +#include + +typedef struct cache_entry_s +{ + char name[6 * DATA_MAX_NAME_LEN]; + int values_num; + gauge_t *values_gauge; + value_t *values_raw; + /* Time contained in the package + * (for calculating rates) */ + cdtime_t last_time; + /* Time according to the local clock + * (for purging old entries) */ + cdtime_t last_update; + /* Interval in which the data is collected + * (for purding old entries) */ + cdtime_t interval; + int state; + int hits; + + /* + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! 0 ! 1 ! 2 ! 3 ! 4 ! 5 ! 6 ! 7 ! 8 ! ... + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ... + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! t = 0 ! t = 1 ! t = 2 ! ... + * +-----------------+-----------------+-----------------+---- + */ + gauge_t *history; + size_t history_index; /* points to the next position to write to. */ + size_t history_length; + + meta_data_t *meta; +} cache_entry_t; + +static c_avl_tree_t *cache_tree = NULL; +static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; + +static int cache_compare (const cache_entry_t *a, const cache_entry_t *b) +{ +#if COLLECT_DEBUG + assert ((a != NULL) && (b != NULL)); +#endif + return (strcmp (a->name, b->name)); +} /* int cache_compare */ + +static cache_entry_t *cache_alloc (int values_num) +{ + cache_entry_t *ce; + + ce = (cache_entry_t *) malloc (sizeof (cache_entry_t)); + if (ce == NULL) + { + ERROR ("utils_cache: cache_alloc: malloc failed."); + return (NULL); + } + memset (ce, '\0', sizeof (cache_entry_t)); + ce->values_num = values_num; + + ce->values_gauge = calloc (values_num, sizeof (*ce->values_gauge)); + ce->values_raw = calloc (values_num, sizeof (*ce->values_raw)); + if ((ce->values_gauge == NULL) || (ce->values_raw == NULL)) + { + sfree (ce->values_gauge); + sfree (ce->values_raw); + sfree (ce); + ERROR ("utils_cache: cache_alloc: calloc failed."); + return (NULL); + } + + ce->history = NULL; + ce->history_length = 0; + ce->meta = NULL; + + return (ce); +} /* cache_entry_t *cache_alloc */ + +static void cache_free (cache_entry_t *ce) +{ + if (ce == NULL) + return; + + sfree (ce->values_gauge); + sfree (ce->values_raw); + sfree (ce->history); + if (ce->meta != NULL) + { + meta_data_destroy (ce->meta); + ce->meta = NULL; + } + sfree (ce); +} /* void cache_free */ + +static void uc_check_range (const data_set_t *ds, cache_entry_t *ce) +{ + int i; + + for (i = 0; i < ds->ds_num; i++) + { + if (isnan (ce->values_gauge[i])) + continue; + else if (ce->values_gauge[i] < ds->ds[i].min) + ce->values_gauge[i] = NAN; + else if (ce->values_gauge[i] > ds->ds[i].max) + ce->values_gauge[i] = NAN; + } +} /* void uc_check_range */ + +static int uc_insert (const data_set_t *ds, const value_list_t *vl, + const char *key) +{ + int i; + char *key_copy; + cache_entry_t *ce; + + /* `cache_lock' has been locked by `uc_update' */ + + key_copy = strdup (key); + if (key_copy == NULL) + { + ERROR ("uc_insert: strdup failed."); + return (-1); + } + + ce = cache_alloc (ds->ds_num); + if (ce == NULL) + { + sfree (key_copy); + ERROR ("uc_insert: cache_alloc (%i) failed.", ds->ds_num); + return (-1); + } + + sstrncpy (ce->name, key, sizeof (ce->name)); + + for (i = 0; i < ds->ds_num; i++) + { + switch (ds->ds[i].type) + { + case DS_TYPE_COUNTER: + ce->values_gauge[i] = NAN; + ce->values_raw[i].counter = vl->values[i].counter; + break; + + case DS_TYPE_GAUGE: + ce->values_gauge[i] = vl->values[i].gauge; + ce->values_raw[i].gauge = vl->values[i].gauge; + break; + + case DS_TYPE_DERIVE: + ce->values_gauge[i] = NAN; + ce->values_raw[i].derive = vl->values[i].derive; + break; + + case DS_TYPE_ABSOLUTE: + ce->values_gauge[i] = NAN; + if (vl->interval > 0) + ce->values_gauge[i] = ((double) vl->values[i].absolute) + / CDTIME_T_TO_DOUBLE (vl->interval); + ce->values_raw[i].absolute = vl->values[i].absolute; + break; + + default: + /* This shouldn't happen. */ + ERROR ("uc_insert: Don't know how to handle data source type %i.", + ds->ds[i].type); ++ sfree (key_copy); + return (-1); + } /* switch (ds->ds[i].type) */ + } /* for (i) */ + + /* Prune invalid gauge data */ + uc_check_range (ds, ce); + + ce->last_time = vl->time; + ce->last_update = cdtime (); + ce->interval = vl->interval; + ce->state = STATE_OKAY; + + if (c_avl_insert (cache_tree, key_copy, ce) != 0) + { + sfree (key_copy); + ERROR ("uc_insert: c_avl_insert failed."); + return (-1); + } + + DEBUG ("uc_insert: Added %s to the cache.", key); + return (0); +} /* int uc_insert */ + +int uc_init (void) +{ + if (cache_tree == NULL) + cache_tree = c_avl_create ((int (*) (const void *, const void *)) + cache_compare); + + return (0); +} /* int uc_init */ + +int uc_check_timeout (void) +{ + cdtime_t now; + cache_entry_t *ce; + + char **keys = NULL; + cdtime_t *keys_time = NULL; + cdtime_t *keys_interval = NULL; + int keys_len = 0; + + char *key; + c_avl_iterator_t *iter; + + int status; + int i; + + pthread_mutex_lock (&cache_lock); + + now = cdtime (); + + /* Build a list of entries to be flushed */ + iter = c_avl_get_iterator (cache_tree); + while (c_avl_iterator_next (iter, (void *) &key, (void *) &ce) == 0) + { + char **tmp; + cdtime_t *tmp_time; + + /* If the entry is fresh enough, continue. */ + if ((now - ce->last_update) < (ce->interval * timeout_g)) + continue; + + /* If entry has not been updated, add to `keys' array */ + tmp = (char **) realloc ((void *) keys, + (keys_len + 1) * sizeof (char *)); + if (tmp == NULL) + { + ERROR ("uc_check_timeout: realloc failed."); + continue; + } + keys = tmp; + + tmp_time = realloc (keys_time, (keys_len + 1) * sizeof (*keys_time)); + if (tmp_time == NULL) + { + ERROR ("uc_check_timeout: realloc failed."); + continue; + } + keys_time = tmp_time; + + tmp_time = realloc (keys_interval, (keys_len + 1) * sizeof (*keys_interval)); + if (tmp_time == NULL) + { + ERROR ("uc_check_timeout: realloc failed."); + continue; + } + keys_interval = tmp_time; + + keys[keys_len] = strdup (key); + if (keys[keys_len] == NULL) + { + ERROR ("uc_check_timeout: strdup failed."); + continue; + } + keys_time[keys_len] = ce->last_time; + keys_interval[keys_len] = ce->interval; + + keys_len++; + } /* while (c_avl_iterator_next) */ + + c_avl_iterator_destroy (iter); + pthread_mutex_unlock (&cache_lock); + + if (keys_len == 0) + { + /* realloc() may have been called for these. */ + sfree (keys); + sfree (keys_time); + sfree (keys_interval); + return (0); + } + + /* Call the "missing" callback for each value. Do this before removing the + * value from the cache, so that callbacks can still access the data stored, + * including plugin specific meta data, rates, history, …. This must be done + * without holding the lock, otherwise we will run into a deadlock if a + * plugin calls the cache interface. */ + for (i = 0; i < keys_len; i++) + { + value_list_t vl = VALUE_LIST_INIT; + + vl.values = NULL; + vl.values_len = 0; + vl.meta = NULL; + + status = parse_identifier_vl (keys[i], &vl); + if (status != 0) + { + ERROR ("uc_check_timeout: parse_identifier_vl (\"%s\") failed.", keys[i]); + continue; + } + + vl.time = keys_time[i]; + vl.interval = keys_interval[i]; + + plugin_dispatch_missing (&vl); + } /* for (i = 0; i < keys_len; i++) */ + + /* Now actually remove all the values from the cache. We don't re-evaluate + * the timestamp again, so in theory it is possible we remove a value after + * it is updated here. */ + pthread_mutex_lock (&cache_lock); + for (i = 0; i < keys_len; i++) + { + key = NULL; + ce = NULL; + + status = c_avl_remove (cache_tree, keys[i], + (void *) &key, (void *) &ce); + if (status != 0) + { + ERROR ("uc_check_timeout: c_avl_remove (\"%s\") failed.", keys[i]); + sfree (keys[i]); + continue; + } + + sfree (keys[i]); + sfree (key); + cache_free (ce); + } /* for (i = 0; i < keys_len; i++) */ + pthread_mutex_unlock (&cache_lock); + + sfree (keys); + sfree (keys_time); + sfree (keys_interval); + + return (0); +} /* int uc_check_timeout */ + +int uc_update (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int status; + int i; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_update: FORMAT_VL failed."); + return (-1); + } + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) /* entry does not yet exist */ + { + status = uc_insert (ds, vl, name); + pthread_mutex_unlock (&cache_lock); + return (status); + } + + assert (ce != NULL); + assert (ce->values_num == ds->ds_num); + + if (ce->last_time >= vl->time) + { + pthread_mutex_unlock (&cache_lock); + NOTICE ("uc_update: Value too old: name = %s; value time = %.3f; " + "last cache update = %.3f;", + name, + CDTIME_T_TO_DOUBLE (vl->time), + CDTIME_T_TO_DOUBLE (ce->last_time)); + return (-1); + } + + for (i = 0; i < ds->ds_num; i++) + { + switch (ds->ds[i].type) + { + case DS_TYPE_COUNTER: + { + counter_t diff; + + /* check if the counter has wrapped around */ + if (vl->values[i].counter < ce->values_raw[i].counter) + { + if (ce->values_raw[i].counter <= 4294967295U) + diff = (4294967295U - ce->values_raw[i].counter) + + vl->values[i].counter; + else + diff = (18446744073709551615ULL - ce->values_raw[i].counter) + + vl->values[i].counter; + } + else /* counter has NOT wrapped around */ + { + diff = vl->values[i].counter - ce->values_raw[i].counter; + } + + ce->values_gauge[i] = ((double) diff) + / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time)); + ce->values_raw[i].counter = vl->values[i].counter; + } + break; + + case DS_TYPE_GAUGE: + ce->values_raw[i].gauge = vl->values[i].gauge; + ce->values_gauge[i] = vl->values[i].gauge; + break; + + case DS_TYPE_DERIVE: + { + derive_t diff; + + diff = vl->values[i].derive - ce->values_raw[i].derive; + + ce->values_gauge[i] = ((double) diff) + / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time)); + ce->values_raw[i].derive = vl->values[i].derive; + } + break; + + case DS_TYPE_ABSOLUTE: + ce->values_gauge[i] = ((double) vl->values[i].absolute) + / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time)); + ce->values_raw[i].absolute = vl->values[i].absolute; + break; + + default: + /* This shouldn't happen. */ + pthread_mutex_unlock (&cache_lock); + ERROR ("uc_update: Don't know how to handle data source type %i.", + ds->ds[i].type); + return (-1); + } /* switch (ds->ds[i].type) */ + + DEBUG ("uc_update: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]); + } /* for (i) */ + + /* Update the history if it exists. */ + if (ce->history != NULL) + { + assert (ce->history_index < ce->history_length); + for (i = 0; i < ce->values_num; i++) + { + size_t hist_idx = (ce->values_num * ce->history_index) + i; + ce->history[hist_idx] = ce->values_gauge[i]; + } + + assert (ce->history_length > 0); + ce->history_index = (ce->history_index + 1) % ce->history_length; + } + + /* Prune invalid gauge data */ + uc_check_range (ds, ce); + + ce->last_time = vl->time; + ce->last_update = cdtime (); + ce->interval = vl->interval; + + pthread_mutex_unlock (&cache_lock); + + return (0); +} /* int uc_update */ + +int uc_get_rate_by_name (const char *name, gauge_t **ret_values, size_t *ret_values_num) +{ + gauge_t *ret = NULL; + size_t ret_num = 0; + cache_entry_t *ce = NULL; + int status = 0; + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + + /* remove missing values from getval */ + if (ce->state == STATE_MISSING) + { + status = -1; + } + else + { + ret_num = ce->values_num; + ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t)); + if (ret == NULL) + { + ERROR ("utils_cache: uc_get_rate_by_name: malloc failed."); + status = -1; + } + else + { + memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t)); + } + } + } + else + { + DEBUG ("utils_cache: uc_get_rate_by_name: No such value: %s", name); + status = -1; + } + + pthread_mutex_unlock (&cache_lock); + + if (status == 0) + { + *ret_values = ret; + *ret_values_num = ret_num; + } + + return (status); +} /* gauge_t *uc_get_rate_by_name */ + +gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + gauge_t *ret = NULL; + size_t ret_num = 0; + int status; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("utils_cache: uc_get_rate: FORMAT_VL failed."); + return (NULL); + } + + status = uc_get_rate_by_name (name, &ret, &ret_num); + if (status != 0) + return (NULL); + + /* This is important - the caller has no other way of knowing how many + * values are returned. */ + if (ret_num != (size_t) ds->ds_num) + { + ERROR ("utils_cache: uc_get_rate: ds[%s] has %i values, " + "but uc_get_rate_by_name returned %zu.", + ds->type, ds->ds_num, ret_num); + sfree (ret); + return (NULL); + } + + return (ret); +} /* gauge_t *uc_get_rate */ + +size_t uc_get_size() { + size_t size_arrays = 0; + + pthread_mutex_lock (&cache_lock); + size_arrays = (size_t) c_avl_size (cache_tree); + pthread_mutex_unlock (&cache_lock); + + return (size_arrays); +} + +int uc_get_names (char ***ret_names, cdtime_t **ret_times, size_t *ret_number) +{ + c_avl_iterator_t *iter; + char *key; + cache_entry_t *value; + + char **names = NULL; + cdtime_t *times = NULL; + size_t number = 0; + size_t size_arrays = 0; + + int status = 0; + + if ((ret_names == NULL) || (ret_number == NULL)) + return (-1); + + pthread_mutex_lock (&cache_lock); + + size_arrays = (size_t) c_avl_size (cache_tree); + if (size_arrays < 1) + { + /* Handle the "no values" case here, to avoid the error message when + * calloc() returns NULL. */ + pthread_mutex_unlock (&cache_lock); + return (0); + } + + names = calloc (size_arrays, sizeof (*names)); + times = calloc (size_arrays, sizeof (*times)); + if ((names == NULL) || (times == NULL)) + { + ERROR ("uc_get_names: calloc failed."); + sfree (names); + sfree (times); + pthread_mutex_unlock (&cache_lock); + return (ENOMEM); + } + + iter = c_avl_get_iterator (cache_tree); + while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0) + { + /* remove missing values when list values */ + if (value->state == STATE_MISSING) + continue; + + /* c_avl_size does not return a number smaller than the number of elements + * returned by c_avl_iterator_next. */ + assert (number < size_arrays); + + if (ret_times != NULL) + times[number] = value->last_time; + + names[number] = strdup (key); + if (names[number] == NULL) + { + status = -1; + break; + } + + number++; + } /* while (c_avl_iterator_next) */ + + c_avl_iterator_destroy (iter); + pthread_mutex_unlock (&cache_lock); + + if (status != 0) + { + size_t i; + + for (i = 0; i < number; i++) + { + sfree (names[i]); + } + sfree (names); + sfree (times); + + return (-1); + } + + *ret_names = names; + if (ret_times != NULL) + *ret_times = times; + else + sfree (times); + *ret_number = number; + + return (0); +} /* int uc_get_names */ + +int uc_get_state (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = STATE_ERROR; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->state; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_get_state */ + +int uc_set_state (const data_set_t *ds, const value_list_t *vl, int state) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->state; + ce->state = state; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_set_state */ + +int uc_get_history_by_name (const char *name, + gauge_t *ret_history, size_t num_steps, size_t num_ds) +{ + cache_entry_t *ce = NULL; + size_t i; + int status = 0; + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + return (-ENOENT); + } + + if (((size_t) ce->values_num) != num_ds) + { + pthread_mutex_unlock (&cache_lock); + return (-EINVAL); + } + + /* Check if there are enough values available. If not, increase the buffer + * size. */ + if (ce->history_length < num_steps) + { + gauge_t *tmp; + size_t i; + + tmp = realloc (ce->history, sizeof (*ce->history) + * num_steps * ce->values_num); + if (tmp == NULL) + { + pthread_mutex_unlock (&cache_lock); + return (-ENOMEM); + } + + for (i = ce->history_length * ce->values_num; + i < (num_steps * ce->values_num); + i++) + tmp[i] = NAN; + + ce->history = tmp; + ce->history_length = num_steps; + } /* if (ce->history_length < num_steps) */ + + /* Copy the values to the output buffer. */ + for (i = 0; i < num_steps; i++) + { + size_t src_index; + size_t dst_index; + + if (i < ce->history_index) + src_index = ce->history_index - (i + 1); + else + src_index = ce->history_length + ce->history_index - (i + 1); + src_index = src_index * num_ds; + + dst_index = i * num_ds; + + memcpy (ret_history + dst_index, ce->history + src_index, + sizeof (*ret_history) * num_ds); + } + + pthread_mutex_unlock (&cache_lock); + + return (0); +} /* int uc_get_history_by_name */ + +int uc_get_history (const data_set_t *ds, const value_list_t *vl, + gauge_t *ret_history, size_t num_steps, size_t num_ds) +{ + char name[6 * DATA_MAX_NAME_LEN]; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("utils_cache: uc_get_history: FORMAT_VL failed."); + return (-1); + } + + return (uc_get_history_by_name (name, ret_history, num_steps, num_ds)); +} /* int uc_get_history */ + +int uc_get_hits (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = STATE_ERROR; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_get_hits */ + +int uc_set_hits (const data_set_t *ds, const value_list_t *vl, int hits) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + ce->hits = hits; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_set_hits */ + +int uc_inc_hits (const data_set_t *ds, const value_list_t *vl, int step) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + ce->hits = ret + step; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_inc_hits */ + +/* + * Meta data interface + */ +/* XXX: This function will acquire `cache_lock' but will not free it! */ +static meta_data_t *uc_get_meta (const value_list_t *vl) /* {{{ */ +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int status; + + status = FORMAT_VL (name, sizeof (name), vl); + if (status != 0) + { + ERROR ("utils_cache: uc_get_meta: FORMAT_VL failed."); + return (NULL); + } + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + return (NULL); + } + assert (ce != NULL); + + if (ce->meta == NULL) + ce->meta = meta_data_create (); + + if (ce->meta == NULL) + pthread_mutex_unlock (&cache_lock); + + return (ce->meta); +} /* }}} meta_data_t *uc_get_meta */ + +/* Sorry about this preprocessor magic, but it really makes this file much + * shorter.. */ +#define UC_WRAP(wrap_function) { \ + meta_data_t *meta; \ + int status; \ + meta = uc_get_meta (vl); \ + if (meta == NULL) return (-1); \ + status = wrap_function (meta, key); \ + pthread_mutex_unlock (&cache_lock); \ + return (status); \ +} +int uc_meta_data_exists (const value_list_t *vl, const char *key) + UC_WRAP (meta_data_exists) + +int uc_meta_data_delete (const value_list_t *vl, const char *key) + UC_WRAP (meta_data_delete) +#undef UC_WRAP + +/* We need a new version of this macro because the following functions take + * two argumetns. */ +#define UC_WRAP(wrap_function) { \ + meta_data_t *meta; \ + int status; \ + meta = uc_get_meta (vl); \ + if (meta == NULL) return (-1); \ + status = wrap_function (meta, key, value); \ + pthread_mutex_unlock (&cache_lock); \ + return (status); \ +} +int uc_meta_data_add_string (const value_list_t *vl, + const char *key, + const char *value) + UC_WRAP(meta_data_add_string) +int uc_meta_data_add_signed_int (const value_list_t *vl, + const char *key, + int64_t value) + UC_WRAP(meta_data_add_signed_int) +int uc_meta_data_add_unsigned_int (const value_list_t *vl, + const char *key, + uint64_t value) + UC_WRAP(meta_data_add_unsigned_int) +int uc_meta_data_add_double (const value_list_t *vl, + const char *key, + double value) + UC_WRAP(meta_data_add_double) +int uc_meta_data_add_boolean (const value_list_t *vl, + const char *key, + _Bool value) + UC_WRAP(meta_data_add_boolean) + +int uc_meta_data_get_string (const value_list_t *vl, + const char *key, + char **value) + UC_WRAP(meta_data_get_string) +int uc_meta_data_get_signed_int (const value_list_t *vl, + const char *key, + int64_t *value) + UC_WRAP(meta_data_get_signed_int) +int uc_meta_data_get_unsigned_int (const value_list_t *vl, + const char *key, + uint64_t *value) + UC_WRAP(meta_data_get_unsigned_int) +int uc_meta_data_get_double (const value_list_t *vl, + const char *key, + double *value) + UC_WRAP(meta_data_get_double) +int uc_meta_data_get_boolean (const value_list_t *vl, + const char *key, + _Bool *value) + UC_WRAP(meta_data_get_boolean) +#undef UC_WRAP + +/* vim: set sw=2 ts=8 sts=2 tw=78 : */