Merge branch 'ff/statsd'

diff --git a/configure.in b/configure.in
index 0838de8b914317c42cb35894f8bf6328b5dc41f4..dbdc0589a0fee67ca9566f43c07ecb40012c3ae2 100644 (file)
--- a/configure.in
+++ b/configure.in
@@ -5168,6 +5168,7 @@ AC_PLUGIN([sensors],     [$with_libsensors],   [lm_sensors statistics])
 AC_PLUGIN([serial],      [$plugin_serial],     [serial port traffic])
 AC_PLUGIN([sigrok],      [$with_libsigrok],    [sigrok acquisition sources])
 AC_PLUGIN([snmp],        [$with_libnetsnmp],   [SNMP querying plugin])
+AC_PLUGIN([statsd],      [yes],                [StatsD plugin])
 AC_PLUGIN([swap],        [$plugin_swap],       [Swap usage statistics])
 AC_PLUGIN([syslog],      [$have_syslog],       [Syslog logging plugin])
 AC_PLUGIN([table],       [yes],                [Parsing of tabular data])
@@ -5507,6 +5508,7 @@ Configuration:
     serial  . . . . . . . $enable_serial
     sigrok  . . . . . . . $enable_sigrok
     snmp  . . . . . . . . $enable_snmp
+    statsd  . . . . . . . $enable_statsd
     swap  . . . . . . . . $enable_swap
     syslog  . . . . . . . $enable_syslog
     table . . . . . . . . $enable_table

diff --git a/src/Makefile.am b/src/Makefile.am
index bdde5a3518d3703596898a6c9d212b3a602e8b77..f35b45a35c335cc6302df127f6f87a0891c5cacd 100644 (file)
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1079,6 +1079,16 @@ collectd_LDADD += "-dlopen" snmp.la
 collectd_DEPENDENCIES += snmp.la
 endif
 
+if BUILD_PLUGIN_STATSD
+pkglib_LTLIBRARIES += statsd.la
+statsd_la_SOURCES = statsd.c \
+                    utils_latency.h utils_latency.c
+statsd_la_LDFLAGS = -module -avoid-version
+statsd_la_LIBADD = -lpthread
+collectd_LDADD += "-dlopen" statsd.la
+collectd_DEPENDENCIES += statsd.la
+endif
+
 if BUILD_PLUGIN_SWAP
 pkglib_LTLIBRARIES += swap.la
 swap_la_SOURCES = swap.c

diff --git a/src/collectd.conf.in b/src/collectd.conf.in
index e471a7d9c57f24a21a46a76b2d5edf5f26593bc1..e8d4d28ef026c1c09546cdc0d143de52a19bbed5 100644 (file)
--- a/src/collectd.conf.in
+++ b/src/collectd.conf.in
@@ -156,6 +156,7 @@
 #@BUILD_PLUGIN_SERIAL_TRUE@LoadPlugin serial
 #@BUILD_PLUGIN_SIGROK_TRUE@LoadPlugin sigrok
 #@BUILD_PLUGIN_SNMP_TRUE@LoadPlugin snmp
+#@BUILD_PLUGIN_STATSD_TRUE@LoadPlugin statsd
 #@BUILD_PLUGIN_SWAP_TRUE@LoadPlugin swap
 #@BUILD_PLUGIN_TABLE_TRUE@LoadPlugin table
 #@BUILD_PLUGIN_TAIL_TRUE@LoadPlugin tail
@@ -946,6 +947,16 @@
 #   </Host>
 #</Plugin>
 
+#<Plugin statsd>
+#  Host "::"
+#  Port "8125"
+#  DeleteCounters false
+#  DeleteTimers   false
+#  DeleteGauges   false
+#  DeleteSets     false
+#  TimerPercentile 90.0
+#</Plugin>
+
 #<Plugin "swap">
 #      ReportByDevice false
 #      ReportBytes true

diff --git a/src/collectd.conf.pod b/src/collectd.conf.pod
index 7b9a5598b77c65fae169edc835295c0d2801b15a..1b73c417d9ebfc54211eefc252116d733a0b9c96 100644 (file)
--- a/src/collectd.conf.pod
+++ b/src/collectd.conf.pod
@@ -5222,6 +5222,55 @@ Since the configuration of the C<snmp plugin> is a little more complicated than
 other plugins, its documentation has been moved to an own manpage,
 L<collectd-snmp(5)>. Please see there for details.
 
+=head2 Plugin C<statsd>
+
+The I<statsd plugin> listens to a UDP socket, reads "events" in the statsd
+protocol and dispatches rates or other aggregates of these numbers
+periodically.
+
+The plugin implements the I<Counter>, I<Timer>, I<Gauge> and I<Set> types which
+are dispatched as the I<collectd> types C<derive>, C<latency>, C<gauge> and
+C<objects> respectively.
+
+The following configuration options are valid:
+
+=over 4
+
+=item B<Host> I<Host>
+
+Bind to the hostname / address I<Host>. By default, the plugin will bind to the
+"any" address, i.e. accept packets sent to any of the hosts addresses.
+
+=item B<Port> I<Port>
+
+UDP port to listen to. This can be either a service name or a port number.
+Defaults to C<8125>.
+
+=item B<DeleteCounters> B<false>|B<true>
+
+=item B<DeleteTimers> B<false>|B<true>
+
+=item B<DeleteGauges> B<false>|B<true>
+
+=item B<DeleteSets> B<false>|B<true>
+
+These options control what happens if metrics are not updated in an interval.
+If set to B<False>, the default, metrics are dispatched unchanged, i.e. the
+rate of counters and size of sets will be zero, timers report C<NaN> and gauges
+are unchanged. If set to B<True>, the such metrics are not dispatched and
+removed from the internal cache.
+
+=item B<TimerPercentile> I<Percent>
+
+Calculate and dispatch the configured percentile, i.e. compute the latency, so
+that I<Percent> of all reported timers are smaller than or equal to the
+computed latency. This is useful for cutting off the long tail latency, as it's
+often done in I<Service Level Agreements> (SLAs).
+
+If not specified, no percentile is calculated / dispatched.
+
+=back
+
 =head2 Plugin C<swap>
 
 The I<Swap plugin> collects information about used and available swap space. On

diff --git a/src/statsd.c b/src/statsd.c
new file mode 100644 (file)
index 0000000..5af3483
--- /dev/null
+++ b/src/statsd.c
@@ -0,0 +1,861 @@
+/**
+ * collectd - src/statsd.c
+ *
+ * Copyright (C) 2013       Florian octo Forster
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF MIND, USE, DATA OR PROFITS, WHETHER
+ * IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING
+ * OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ *
+ * Authors:
+ *   Florian octo Forster <octo at collectd.org>
+ */
+
+#include "collectd.h"
+#include "plugin.h"
+#include "common.h"
+#include "configfile.h"
+#include "utils_avltree.h"
+#include "utils_complain.h"
+#include "utils_latency.h"
+
+#include <pthread.h>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <poll.h>
+
+#ifndef STATSD_DEFAULT_NODE
+# define STATSD_DEFAULT_NODE NULL
+#endif
+
+#ifndef STATSD_DEFAULT_SERVICE
+# define STATSD_DEFAULT_SERVICE "8125"
+#endif
+
+enum metric_type_e
+{
+  STATSD_COUNTER,
+  STATSD_TIMER,
+  STATSD_GAUGE,
+  STATSD_SET
+};
+typedef enum metric_type_e metric_type_t;
+
+struct statsd_metric_s
+{
+  metric_type_t type;
+  double value;
+  latency_counter_t *latency;
+  c_avl_tree_t *set;
+  unsigned long updates_num;
+};
+typedef struct statsd_metric_s statsd_metric_t;
+
+static c_avl_tree_t   *metrics_tree = NULL;
+static pthread_mutex_t metrics_lock = PTHREAD_MUTEX_INITIALIZER;
+
+static pthread_t network_thread;
+static _Bool     network_thread_running = 0;
+static _Bool     network_thread_shutdown = 0;
+
+static char *conf_node = NULL;
+static char *conf_service = NULL;
+
+static _Bool conf_delete_counters = 0;
+static _Bool conf_delete_timers   = 0;
+static _Bool conf_delete_gauges   = 0;
+static _Bool conf_delete_sets     = 0;
+
+static double *conf_timer_percentile = NULL;
+static size_t  conf_timer_percentile_num = 0;
+
+/* Must hold metrics_lock when calling this function. */
+static statsd_metric_t *statsd_metric_lookup_unsafe (char const *name, /* {{{ */
+    metric_type_t type)
+{
+  char key[DATA_MAX_NAME_LEN + 2];
+  char *key_copy;
+  statsd_metric_t *metric;
+  int status;
+
+  switch (type)
+  {
+    case STATSD_COUNTER: key[0] = 'c'; break;
+    case STATSD_TIMER:   key[0] = 't'; break;
+    case STATSD_GAUGE:   key[0] = 'g'; break;
+    case STATSD_SET:     key[0] = 's'; break;
+    default: return (NULL);
+  }
+
+  key[1] = ':';
+  sstrncpy (&key[2], name, sizeof (key) - 2);
+
+  status = c_avl_get (metrics_tree, key, (void *) &metric);
+  if (status == 0)
+    return (metric);
+
+  key_copy = strdup (key);
+  if (key_copy == NULL)
+  {
+    ERROR ("statsd plugin: strdup failed.");
+    return (NULL);
+  }
+
+  metric = malloc (sizeof (*metric));
+  if (metric == NULL)
+  {
+    ERROR ("statsd plugin: malloc failed.");
+    sfree (key_copy);
+    return (NULL);
+  }
+  memset (metric, 0, sizeof (*metric));
+
+  metric->type = type;
+  metric->latency = NULL;
+  metric->set = NULL;
+
+  status = c_avl_insert (metrics_tree, key_copy, metric);
+  if (status != 0)
+  {
+    ERROR ("statsd plugin: c_avl_insert failed.");
+    sfree (key_copy);
+    sfree (metric);
+    return (NULL);
+  }
+
+  return (metric);
+} /* }}} statsd_metric_lookup_unsafe */
+
+static int statsd_metric_set (char const *name, double value, /* {{{ */
+    metric_type_t type)
+{
+  statsd_metric_t *metric;
+
+  pthread_mutex_lock (&metrics_lock);
+
+  metric = statsd_metric_lookup_unsafe (name, type);
+  if (metric == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    return (-1);
+  }
+
+  metric->value = value;
+  metric->updates_num++;
+
+  pthread_mutex_unlock (&metrics_lock);
+
+  return (0);
+} /* }}} int statsd_metric_set */
+
+static int statsd_metric_add (char const *name, double delta, /* {{{ */
+    metric_type_t type)
+{
+  statsd_metric_t *metric;
+
+  pthread_mutex_lock (&metrics_lock);
+
+  metric = statsd_metric_lookup_unsafe (name, type);
+  if (metric == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    return (-1);
+  }
+
+  metric->value += delta;
+  metric->updates_num++;
+
+  pthread_mutex_unlock (&metrics_lock);
+
+  return (0);
+} /* }}} int statsd_metric_add */
+
+static int statsd_parse_value (char const *str, value_t *ret_value) /* {{{ */
+{
+  char *endptr = NULL;
+
+  ret_value->gauge = (gauge_t) strtod (str, &endptr);
+  if ((str == endptr) || ((endptr != NULL) && (*endptr != 0)))
+    return (-1);
+
+  return (0);
+} /* }}} int statsd_parse_value */
+
+static int statsd_handle_counter (char const *name, /* {{{ */
+    char const *value_str,
+    char const *extra)
+{
+  value_t value;
+  value_t scale;
+  int status;
+
+  if ((extra != NULL) && (extra[0] != '@'))
+    return (-1);
+
+  scale.gauge = 1.0;
+  if (extra != NULL)
+  {
+    status = statsd_parse_value (extra + 1, &scale);
+    if (status != 0)
+      return (status);
+
+    if (!isfinite (scale.gauge) || (scale.gauge <= 0.0) || (scale.gauge > 1.0))
+      return (-1);
+  }
+
+  value.gauge = 1.0;
+  status = statsd_parse_value (value_str, &value);
+  if (status != 0)
+    return (status);
+
+  return (statsd_metric_add (name, (double) (value.gauge / scale.gauge),
+        STATSD_COUNTER));
+} /* }}} int statsd_handle_counter */
+
+static int statsd_handle_gauge (char const *name, /* {{{ */
+    char const *value_str)
+{
+  value_t value;
+  int status;
+
+  value.gauge = 0;
+  status = statsd_parse_value (value_str, &value);
+  if (status != 0)
+    return (status);
+
+  if ((value_str[0] == '+') || (value_str[0] == '-'))
+    return (statsd_metric_add (name, (double) value.gauge, STATSD_GAUGE));
+  else
+    return (statsd_metric_set (name, (double) value.gauge, STATSD_GAUGE));
+} /* }}} int statsd_handle_gauge */
+
+static int statsd_handle_timer (char const *name, /* {{{ */
+    char const *value_str)
+{
+  statsd_metric_t *metric;
+  value_t value_ms;
+  cdtime_t value;
+  int status;
+
+  value_ms.derive = 0;
+  status = statsd_parse_value (value_str, &value_ms);
+  if (status != 0)
+    return (status);
+
+  value = MS_TO_CDTIME_T (value_ms.gauge);
+
+  pthread_mutex_lock (&metrics_lock);
+
+  metric = statsd_metric_lookup_unsafe (name, STATSD_TIMER);
+  if (metric == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    return (-1);
+  }
+
+  if (metric->latency == NULL)
+    metric->latency = latency_counter_create ();
+  if (metric->latency == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    return (-1);
+  }
+
+  latency_counter_add (metric->latency, value);
+  metric->updates_num++;
+
+  pthread_mutex_unlock (&metrics_lock);
+  return (0);
+} /* }}} int statsd_handle_timer */
+
+static int statsd_handle_set (char const *name, /* {{{ */
+    char const *set_key_orig)
+{
+  statsd_metric_t *metric = NULL;
+  char *set_key;
+  int status;
+
+  pthread_mutex_lock (&metrics_lock);
+
+  metric = statsd_metric_lookup_unsafe (name, STATSD_SET);
+  if (metric == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    return (-1);
+  }
+
+  /* Make sure metric->set exists. */
+  if (metric->set == NULL)
+    metric->set = c_avl_create ((void *) strcmp);
+
+  if (metric->set == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    ERROR ("statsd plugin: c_avl_create failed.");
+    return (-1);
+  }
+
+  set_key = strdup (set_key_orig);
+  if (set_key == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    ERROR ("statsd plugin: strdup failed.");
+    return (-1);
+  }
+
+  status = c_avl_insert (metric->set, set_key, /* value = */ NULL);
+  if (status < 0)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    if (status < 0)
+      ERROR ("statsd plugin: c_avl_insert (\"%s\") failed with status %i.",
+          set_key, status);
+    sfree (set_key);
+    return (-1);
+  }
+  else if (status > 0) /* key already exists */
+  {
+    sfree (set_key);
+  }
+
+  metric->updates_num++;
+
+  pthread_mutex_unlock (&metrics_lock);
+  return (0);
+} /* }}} int statsd_handle_set */
+
+static int statsd_parse_line (char *buffer) /* {{{ */
+{
+  char *name = buffer;
+  char *value;
+  char *type;
+  char *extra;
+
+  type = strchr (name, '|');
+  if (type == NULL)
+    return (-1);
+  *type = 0;
+  type++;
+
+  value = strrchr (name, ':');
+  if (value == NULL)
+    return (-1);
+  *value = 0;
+  value++;
+
+  extra = strchr (type, '|');
+  if (extra != NULL)
+  {
+    *extra = 0;
+    extra++;
+  }
+
+  if (strcmp ("c", type) == 0)
+    return (statsd_handle_counter (name, value, extra));
+
+  /* extra is only valid for counters */
+  if (extra != NULL)
+    return (-1);
+
+  if (strcmp ("g", type) == 0)
+    return (statsd_handle_gauge (name, value));
+  else if (strcmp ("ms", type) == 0)
+    return (statsd_handle_timer (name, value));
+  else if (strcmp ("s", type) == 0)
+    return (statsd_handle_set (name, value));
+  else
+    return (-1);
+} /* }}} void statsd_parse_line */
+
+static void statsd_parse_buffer (char *buffer) /* {{{ */
+{
+  while (buffer != NULL)
+  {
+    char orig[64];
+    char *next;
+    int status;
+
+    next = strchr (buffer, '\n');
+    if (next != NULL)
+    {
+      *next = 0;
+      next++;
+    }
+
+    if (*buffer == 0)
+    {
+      buffer = next;
+      continue;
+    }
+
+    sstrncpy (orig, buffer, sizeof (orig));
+
+    status = statsd_parse_line (buffer);
+    if (status != 0)
+      ERROR ("statsd plugin: Unable to parse line: \"%s\"", orig);
+
+    buffer = next;
+  }
+} /* }}} void statsd_parse_buffer */
+
+static void statsd_network_read (int fd) /* {{{ */
+{
+  char buffer[4096];
+  size_t buffer_size;
+  ssize_t status;
+
+  status = recv (fd, buffer, sizeof (buffer), /* flags = */ MSG_DONTWAIT);
+  if (status < 0)
+  {
+    char errbuf[1024];
+
+    if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
+      return;
+
+    ERROR ("statsd plugin: recv(2) failed: %s",
+        sstrerror (errno, errbuf, sizeof (errbuf)));
+    return;
+  }
+
+  buffer_size = (size_t) status;
+  if (buffer_size >= sizeof (buffer))
+    buffer_size = sizeof (buffer) - 1;
+  buffer[buffer_size] = 0;
+
+  statsd_parse_buffer (buffer);
+} /* }}} void statsd_network_read */
+
+static int statsd_network_init (struct pollfd **ret_fds, /* {{{ */
+    size_t *ret_fds_num)
+{
+  struct pollfd *fds = NULL;
+  size_t fds_num = 0;
+
+  struct addrinfo ai_hints;
+  struct addrinfo *ai_list = NULL;
+  struct addrinfo *ai_ptr;
+  int status;
+
+  char const *node = (conf_node != NULL) ? conf_node : STATSD_DEFAULT_NODE;
+  char const *service = (conf_service != NULL)
+    ? conf_service : STATSD_DEFAULT_SERVICE;
+
+  memset (&ai_hints, 0, sizeof (ai_hints));
+  ai_hints.ai_flags = AI_PASSIVE;
+#ifdef AI_ADDRCONFIG
+  ai_hints.ai_flags |= AI_ADDRCONFIG;
+#endif
+  ai_hints.ai_family = AF_UNSPEC;
+  ai_hints.ai_socktype = SOCK_DGRAM;
+
+  status = getaddrinfo (node, service, &ai_hints, &ai_list);
+  if (status != 0)
+  {
+    ERROR ("statsd plugin: getaddrinfo (\"%s\", \"%s\") failed: %s",
+        node, service, gai_strerror (status));
+    return (status);
+  }
+
+  for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
+  {
+    int fd;
+    struct pollfd *tmp;
+
+    char dbg_node[NI_MAXHOST];
+    char dbg_service[NI_MAXSERV];
+
+    fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
+    if (fd < 0)
+    {
+      char errbuf[1024];
+      ERROR ("statsd plugin: socket(2) failed: %s",
+          sstrerror (errno, errbuf, sizeof (errbuf)));
+      continue;
+    }
+
+    getnameinfo (ai_ptr->ai_addr, ai_ptr->ai_addrlen,
+        dbg_node, sizeof (dbg_node), dbg_service, sizeof (dbg_service),
+        NI_DGRAM | NI_NUMERICHOST | NI_NUMERICSERV);
+    DEBUG ("statsd plugin: Trying to bind to [%s]:%s ...", dbg_node, dbg_service);
+
+    status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
+    if (status != 0)
+    {
+      char errbuf[1024];
+      ERROR ("statsd plugin: bind(2) failed: %s",
+          sstrerror (errno, errbuf, sizeof (errbuf)));
+      close (fd);
+      continue;
+    }
+
+    tmp = realloc (fds, sizeof (*fds) * (fds_num + 1));
+    if (tmp == NULL)
+    {
+      ERROR ("statsd plugin: realloc failed.");
+      continue;
+    }
+    fds = tmp;
+    tmp = fds + fds_num;
+    fds_num++;
+
+    memset (tmp, 0, sizeof (*tmp));
+    tmp->fd = fd;
+    tmp->events = POLLIN | POLLPRI;
+  }
+
+  freeaddrinfo (ai_list);
+
+  if (fds_num == 0)
+  {
+    ERROR ("statsd plugin: Unable to create listening socket for [%s]:%s.",
+        (node != NULL) ? node : "::", service);
+    return (ENOENT);
+  }
+
+  *ret_fds = fds;
+  *ret_fds_num = fds_num;
+  return (0);
+} /* }}} int statsd_network_init */
+
+static void *statsd_network_thread (void *args) /* {{{ */
+{
+  struct pollfd *fds = NULL;
+  size_t fds_num = 0;
+  int status;
+  size_t i;
+
+  status = statsd_network_init (&fds, &fds_num);
+  if (status != 0)
+  {
+    ERROR ("statsd plugin: Unable to open listening sockets.");
+    pthread_exit ((void *) 0);
+  }
+
+  while (!network_thread_shutdown)
+  {
+    status = poll (fds, (nfds_t) fds_num, /* timeout = */ -1);
+    if (status < 0)
+    {
+      char errbuf[1024];
+
+      if ((errno == EINTR) || (errno == EAGAIN))
+        continue;
+
+      ERROR ("statsd plugin: poll(2) failed: %s",
+          sstrerror (errno, errbuf, sizeof (errbuf)));
+      break;
+    }
+
+    for (i = 0; i < fds_num; i++)
+    {
+      if ((fds[i].revents & (POLLIN | POLLPRI)) == 0)
+        continue;
+
+      statsd_network_read (fds[i].fd);
+      fds[i].revents = 0;
+    }
+  } /* while (!network_thread_shutdown) */
+
+  /* Clean up */
+  for (i = 0; i < fds_num; i++)
+    close (fds[i].fd);
+  sfree (fds);
+
+  return ((void *) 0);
+} /* }}} void *statsd_network_thread */
+
+static int statsd_config_timer_percentile (oconfig_item_t *ci) /* {{{ */
+{
+  double percent = NAN;
+  double *tmp;
+  int status;
+
+  status = cf_util_get_double (ci, &percent);
+  if (status != 0)
+    return (status);
+
+  if ((percent <= 0.0) || (percent >= 100))
+  {
+    ERROR ("statsd plugin: The value for \"%s\" must be between 0 and 100, "
+        "exclusively.", ci->key);
+    return (ERANGE);
+  }
+
+  tmp = realloc (conf_timer_percentile,
+      sizeof (*conf_timer_percentile) * (conf_timer_percentile_num + 1));
+  if (tmp == NULL)
+  {
+    ERROR ("statsd plugin: realloc failed.");
+    return (ENOMEM);
+  }
+  conf_timer_percentile = tmp;
+  conf_timer_percentile[conf_timer_percentile_num] = percent;
+  conf_timer_percentile_num++;
+
+  return (0);
+} /* }}} int statsd_config_timer_percentile */
+
+static int statsd_config (oconfig_item_t *ci) /* {{{ */
+{
+  int i;
+
+  for (i = 0; i < ci->children_num; i++)
+  {
+    oconfig_item_t *child = ci->children + i;
+
+    if (strcasecmp ("Host", child->key) == 0)
+      cf_util_get_string (child, &conf_node);
+    else if (strcasecmp ("Port", child->key) == 0)
+      cf_util_get_service (child, &conf_service);
+    else if (strcasecmp ("DeleteCounters", child->key) == 0)
+      cf_util_get_boolean (child, &conf_delete_counters);
+    else if (strcasecmp ("DeleteTimers", child->key) == 0)
+      cf_util_get_boolean (child, &conf_delete_timers);
+    else if (strcasecmp ("DeleteGauges", child->key) == 0)
+      cf_util_get_boolean (child, &conf_delete_gauges);
+    else if (strcasecmp ("DeleteSets", child->key) == 0)
+      cf_util_get_boolean (child, &conf_delete_sets);
+    else if (strcasecmp ("TimerPercentile", child->key) == 0)
+      statsd_config_timer_percentile (child);
+    else
+      ERROR ("statsd plugin: The \"%s\" config option is not valid.",
+          child->key);
+  }
+
+  return (0);
+} /* }}} int statsd_config */
+
+static int statsd_init (void) /* {{{ */
+{
+  pthread_mutex_lock (&metrics_lock);
+  if (metrics_tree == NULL)
+    metrics_tree = c_avl_create ((void *) strcmp);
+
+  if (!network_thread_running)
+  {
+    int status;
+
+    status = pthread_create (&network_thread,
+        /* attr = */ NULL,
+        statsd_network_thread,
+        /* args = */ NULL);
+    if (status != 0)
+    {
+      char errbuf[1024];
+      pthread_mutex_unlock (&metrics_lock);
+      ERROR ("statsd plugin: pthread_create failed: %s",
+          sstrerror (errno, errbuf, sizeof (errbuf)));
+      return (status);
+    }
+  }
+  network_thread_running = 1;
+
+  pthread_mutex_unlock (&metrics_lock);
+
+  return (0);
+} /* }}} int statsd_init */
+
+/* Must hold metrics_lock when calling this function. */
+static int statsd_metric_clear_set_unsafe (statsd_metric_t *metric) /* {{{ */
+{
+  void *key;
+  void *value;
+
+  if ((metric == NULL) || (metric->type != STATSD_SET))
+    return (EINVAL);
+
+  if (metric->set == NULL)
+    return (0);
+
+  while (c_avl_pick (metric->set, &key, &value) == 0)
+  {
+    sfree (key);
+    sfree (value);
+  }
+
+  return (0);
+} /* }}} int statsd_metric_clear_set_unsafe */
+
+/* Must hold metrics_lock when calling this function. */
+static int statsd_metric_submit_unsafe (char const *name, /* {{{ */
+    statsd_metric_t const *metric)
+{
+  value_t values[1];
+  value_list_t vl = VALUE_LIST_INIT;
+
+  vl.values = values;
+  vl.values_len = 1;
+  sstrncpy (vl.host, hostname_g, sizeof (vl.host));
+  sstrncpy (vl.plugin, "statsd", sizeof (vl.plugin));
+
+  if (metric->type == STATSD_GAUGE)
+    sstrncpy (vl.type, "gauge", sizeof (vl.type));
+  else if (metric->type == STATSD_TIMER)
+    sstrncpy (vl.type, "latency", sizeof (vl.type));
+  else if (metric->type == STATSD_SET)
+    sstrncpy (vl.type, "objects", sizeof (vl.type));
+  else /* if (metric->type == STATSD_COUNTER) */
+    sstrncpy (vl.type, "derive", sizeof (vl.type));
+
+  sstrncpy (vl.type_instance, name, sizeof (vl.type_instance));
+
+  if (metric->type == STATSD_GAUGE)
+    values[0].gauge = (gauge_t) metric->value;
+  else if (metric->type == STATSD_TIMER)
+  {
+    size_t i;
+
+    if (metric->updates_num == 0)
+      return (0);
+
+    vl.time = cdtime ();
+
+    ssnprintf (vl.type_instance, sizeof (vl.type_instance),
+        "%s-average", name);
+    values[0].gauge = CDTIME_T_TO_DOUBLE (
+        latency_counter_get_average (metric->latency));
+    plugin_dispatch_values (&vl);
+
+    for (i = 0; i < conf_timer_percentile_num; i++)
+    {
+      ssnprintf (vl.type_instance, sizeof (vl.type_instance),
+          "%s-percentile-%.0f", name, conf_timer_percentile[i]);
+      values[0].gauge = CDTIME_T_TO_DOUBLE (
+          latency_counter_get_percentile (
+            metric->latency, conf_timer_percentile[i]));
+      plugin_dispatch_values (&vl);
+    }
+
+    latency_counter_reset (metric->latency);
+    return (0);
+  }
+  else if (metric->type == STATSD_SET)
+  {
+    if (metric->set == NULL)
+      values[0].gauge = 0.0;
+    else
+      values[0].gauge = (gauge_t) c_avl_size (metric->set);
+  }
+  else
+    values[0].derive = (derive_t) metric->value;
+
+  return (plugin_dispatch_values (&vl));
+} /* }}} int statsd_metric_submit_unsafe */
+
+static int statsd_read (void) /* {{{ */
+{
+  c_avl_iterator_t *iter;
+  char *name;
+  statsd_metric_t *metric;
+
+  char **to_be_deleted = NULL;
+  size_t to_be_deleted_num = 0;
+  size_t i;
+
+  pthread_mutex_lock (&metrics_lock);
+
+  if (metrics_tree == NULL)
+  {
+    pthread_mutex_unlock (&metrics_lock);
+    return (0);
+  }
+
+  iter = c_avl_get_iterator (metrics_tree);
+  while (c_avl_iterator_next (iter, (void *) &name, (void *) &metric) == 0)
+  {
+    if ((metric->updates_num == 0)
+        && ((conf_delete_counters && (metric->type == STATSD_COUNTER))
+          || (conf_delete_timers && (metric->type == STATSD_TIMER))
+          || (conf_delete_gauges && (metric->type == STATSD_GAUGE))
+          || (conf_delete_sets && (metric->type == STATSD_SET))))
+    {
+      DEBUG ("statsd plugin: Deleting metric \"%s\".", name);
+      strarray_add (&to_be_deleted, &to_be_deleted_num, name);
+      continue;
+    }
+
+    /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
+     * Remove this here. */
+    statsd_metric_submit_unsafe (name + 2, metric);
+
+    /* Reset the metric. */
+    metric->updates_num = 0;
+    if (metric->type == STATSD_SET)
+      statsd_metric_clear_set_unsafe (metric);
+  }
+  c_avl_iterator_destroy (iter);
+
+  for (i = 0; i < to_be_deleted_num; i++)
+  {
+    int status;
+
+    status = c_avl_remove (metrics_tree, to_be_deleted[i],
+        (void *) &name, (void *) &metric);
+    if (status != 0)
+    {
+      ERROR ("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
+          to_be_deleted[i], status);
+      continue;
+    }
+
+    sfree (name);
+    sfree (metric);
+  }
+
+  pthread_mutex_unlock (&metrics_lock);
+
+  strarray_free (to_be_deleted, to_be_deleted_num);
+
+  return (0);
+} /* }}} int statsd_read */
+
+static int statsd_shutdown (void) /* {{{ */
+{
+  void *key;
+  void *value;
+
+  pthread_mutex_lock (&metrics_lock);
+
+  if (network_thread_running)
+  {
+    network_thread_shutdown = 1;
+    pthread_kill (network_thread, SIGTERM);
+    pthread_join (network_thread, /* retval = */ NULL);
+  }
+  network_thread_running = 0;
+
+  while (c_avl_pick (metrics_tree, &key, &value) == 0)
+  {
+    sfree (key);
+    sfree (value);
+  }
+  c_avl_destroy (metrics_tree);
+  metrics_tree = NULL;
+
+  sfree (conf_node);
+  sfree (conf_service);
+
+  pthread_mutex_unlock (&metrics_lock);
+
+  return (0);
+} /* }}} int statsd_shutdown */
+
+void module_register (void)
+{
+  plugin_register_complex_config ("statsd", statsd_config);
+  plugin_register_init ("statsd", statsd_init);
+  plugin_register_read ("statsd", statsd_read);
+  plugin_register_shutdown ("statsd", statsd_shutdown);
+}
+
+/* vim: set sw=2 sts=2 et fdm=marker : */

diff --git a/src/utils_latency.c b/src/utils_latency.c
new file mode 100644 (file)
index 0000000..01c8b5c
--- /dev/null
+++ b/src/utils_latency.c
@@ -0,0 +1,173 @@
+/**
+ * collectd - src/utils_latency.c
+ * Copyright (C) 2013  Florian 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 Forster <ff at octo.it>
+ **/
+
+#include "collectd.h"
+#include "utils_latency.h"
+#include "common.h"
+
+#ifndef LATENCY_HISTOGRAM_SIZE
+# define LATENCY_HISTOGRAM_SIZE 1000
+#endif
+
+struct latency_counter_s
+{
+  cdtime_t start_time;
+
+  cdtime_t sum;
+  size_t num;
+
+  cdtime_t min;
+  cdtime_t max;
+
+  int histogram[LATENCY_HISTOGRAM_SIZE];
+};
+
+latency_counter_t *latency_counter_create () /* {{{ */
+{
+  latency_counter_t *lc;
+
+  lc = malloc (sizeof (*lc));
+  if (lc == NULL)
+    return (NULL);
+
+  latency_counter_reset (lc);
+  return (lc);
+} /* }}} latency_counter_t *latency_counter_create */
+
+void latency_counter_destroy (latency_counter_t *lc) /* {{{ */
+{
+  sfree (lc);
+} /* }}} void latency_counter_destroy */
+
+void latency_counter_add (latency_counter_t *lc, cdtime_t latency) /* {{{ */
+{
+  size_t latency_ms;
+
+  if ((lc == NULL) || (latency == 0))
+    return;
+
+  lc->sum += latency;
+  lc->num++;
+
+  if ((lc->min == 0) && (lc->max == 0))
+    lc->min = lc->max = latency;
+  if (lc->min > latency)
+    lc->min = latency;
+  if (lc->max < latency)
+    lc->max = latency;
+
+  /* A latency of _exactly_ 1.0 ms should be stored in the buffer 0, so
+   * subtract one from the cdtime_t value so that exactly 1.0 ms get sorted
+   * accordingly. */
+  latency_ms = (size_t) CDTIME_T_TO_MS (latency - 1);
+  if (latency_ms < STATIC_ARRAY_SIZE (lc->histogram))
+    lc->histogram[latency_ms]++;
+} /* }}} void latency_counter_add */
+
+void latency_counter_reset (latency_counter_t *lc) /* {{{ */
+{
+  if (lc == NULL)
+    return;
+
+  memset (lc, 0, sizeof (*lc));
+  lc->start_time = cdtime ();
+} /* }}} void latency_counter_reset */
+
+cdtime_t latency_counter_get_min (latency_counter_t *lc) /* {{{ */
+{
+  if (lc == NULL)
+    return (0);
+  return (lc->min);
+} /* }}} cdtime_t latency_counter_get_min */
+
+cdtime_t latency_counter_get_max (latency_counter_t *lc) /* {{{ */
+{
+  if (lc == NULL)
+    return (0);
+  return (lc->max);
+} /* }}} cdtime_t latency_counter_get_max */
+
+cdtime_t latency_counter_get_average (latency_counter_t *lc) /* {{{ */
+{
+  double average;
+
+  if (lc == NULL)
+    return (0);
+
+  average = CDTIME_T_TO_DOUBLE (lc->sum) / ((double) lc->num);
+  return (DOUBLE_TO_CDTIME_T (average));
+} /* }}} cdtime_t latency_counter_get_average */
+
+cdtime_t latency_counter_get_percentile (latency_counter_t *lc,
+    double percent)
+{
+  double percent_upper;
+  double percent_lower;
+  double ms_upper;
+  double ms_lower;
+  double ms_interpolated;
+  int sum;
+  size_t i;
+
+  if ((lc == NULL) || !((percent > 0.0) && (percent < 100.0)))
+    return (0);
+
+  /* Find index i so that at least "percent" events are within i+1 ms. */
+  percent_upper = 0.0;
+  percent_lower = 0.0;
+  sum = 0;
+  for (i = 0; i < LATENCY_HISTOGRAM_SIZE; i++)
+  {
+    percent_lower = percent_upper;
+    sum += lc->histogram[i];
+    if (sum == 0)
+      percent_upper = 0.0;
+    else
+      percent_upper = 100.0 * ((double) sum) / ((double) lc->num);
+
+    if (percent_upper >= percent)
+      break;
+  }
+
+  if (i >= LATENCY_HISTOGRAM_SIZE)
+    return (0);
+
+  assert (percent_upper >= percent);
+  assert (percent_lower < percent);
+
+  ms_upper = (double) (i + 1);
+  ms_lower = (double) i;
+  if (i == 0)
+    return (MS_TO_CDTIME_T (ms_upper));
+
+  ms_interpolated = (((percent_upper - percent) * ms_lower)
+      + ((percent - percent_lower) * ms_upper))
+    / (percent_upper - percent_lower);
+
+  return (MS_TO_CDTIME_T (ms_interpolated));
+} /* }}} cdtime_t latency_counter_get_percentile */
+
+/* vim: set sw=2 sts=2 et fdm=marker : */

diff --git a/src/utils_latency.h b/src/utils_latency.h
new file mode 100644 (file)
index 0000000..3da2308
--- /dev/null
+++ b/src/utils_latency.h
@@ -0,0 +1,45 @@
+/**
+ * collectd - src/utils_latency.h
+ * Copyright (C) 2013  Florian 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 Forster <ff at octo.it>
+ **/
+
+#include "collectd.h"
+#include "utils_time.h"
+
+struct latency_counter_s;
+typedef struct latency_counter_s latency_counter_t;
+
+latency_counter_t *latency_counter_create ();
+void latency_counter_destroy (latency_counter_t *lc);
+
+void latency_counter_add (latency_counter_t *lc, cdtime_t latency);
+void latency_counter_reset (latency_counter_t *lc);
+
+cdtime_t latency_counter_get_min (latency_counter_t *lc);
+cdtime_t latency_counter_get_max (latency_counter_t *lc);
+cdtime_t latency_counter_get_average (latency_counter_t *lc);
+cdtime_t latency_counter_get_percentile (latency_counter_t *lc,
+    double percent);
+
+/* vim: set sw=2 sts=2 et : */