X-Git-Url: https://git.octo.it/?p=collectd.git;a=blobdiff_plain;f=src%2Fstatsd.c;h=72b8e2b6342b2beb92c71e1322a0e836649521ad;hp=896fb619ffcdcc90095b042595ea8aa53e7477b0;hb=633c3966f770e4d46651a2fe219a18d8a9907a9f;hpb=cc4025b8c5561acb292e5fd025dfd4a2beb68165 diff --git a/src/statsd.c b/src/statsd.c index 896fb619..72b8e2b6 100644 --- a/src/statsd.c +++ b/src/statsd.c @@ -1,19 +1,24 @@ /** * 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. + * 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" 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. + * 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 @@ -25,6 +30,7 @@ #include "configfile.h" #include "utils_avltree.h" #include "utils_complain.h" +#include "utils_latency.h" #include @@ -33,6 +39,11 @@ #include #include +/* AIX doesn't have MSG_DONTWAIT */ +#ifndef MSG_DONTWAIT +# define MSG_DONTWAIT MSG_NONBLOCK +#endif + #ifndef STATSD_DEFAULT_NODE # define STATSD_DEFAULT_NODE NULL #endif @@ -45,15 +56,18 @@ enum metric_type_e { STATSD_COUNTER, STATSD_TIMER, - STATSD_GAUGE + STATSD_GAUGE, + STATSD_SET }; typedef enum metric_type_e metric_type_t; struct statsd_metric_s { metric_type_t type; - int64_t value; - cdtime_t last_update; + double value; + latency_counter_t *latency; + c_avl_tree_t *set; + unsigned long updates_num; }; typedef struct statsd_metric_s statsd_metric_t; @@ -67,91 +81,131 @@ 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; + +static _Bool conf_timer_lower = 0; +static _Bool conf_timer_upper = 0; +static _Bool conf_timer_sum = 0; +static _Bool conf_timer_count = 0; + /* Must hold metrics_lock when calling this function. */ -static int statsd_metric_set_unsafe (char const *name, int64_t value, /* {{{ */ +static statsd_metric_t *statsd_metric_lookup_unsafe (char const *name, /* {{{ */ metric_type_t type) { - cdtime_t now; + char key[DATA_MAX_NAME_LEN + 2]; + char *key_copy; statsd_metric_t *metric; - char *key; int status; - now = cdtime (); + 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, name, (void *) &metric); + status = c_avl_get (metrics_tree, key, (void *) &metric); if (status == 0) - { - metric->value = value; - metric->last_update = now; + return (metric); - return (0); + key_copy = strdup (key); + if (key_copy == NULL) + { + ERROR ("statsd plugin: strdup failed."); + return (NULL); } - key = strdup (name); - metric = calloc (1, sizeof (*metric)); - if ((key == NULL) || (metric == NULL)) + metric = malloc (sizeof (*metric)); + if (metric == NULL) { - sfree (key); - sfree (metric); - return (-1); + ERROR ("statsd plugin: malloc failed."); + sfree (key_copy); + return (NULL); } + memset (metric, 0, sizeof (*metric)); metric->type = type; - metric->value = value; - metric->last_update = now; + metric->latency = NULL; + metric->set = NULL; - status = c_avl_insert (metrics_tree, key, metric); + status = c_avl_insert (metrics_tree, key_copy, metric); if (status != 0) { - sfree (key); + ERROR ("statsd plugin: c_avl_insert failed."); + sfree (key_copy); sfree (metric); - - return (-1); + return (NULL); } - return (0); -} /* }}} int statsd_metric_set_unsafe */ + return (metric); +} /* }}} statsd_metric_lookup_unsafe */ -static int statsd_metric_set (char const *name, int64_t value, /* {{{ */ +static int statsd_metric_set (char const *name, double value, /* {{{ */ metric_type_t type) { - int status; + statsd_metric_t *metric; pthread_mutex_lock (&metrics_lock); - status = statsd_metric_set_unsafe (name, value, type); + + 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 (status); + return (0); } /* }}} int statsd_metric_set */ -static int statsd_metric_add (char const *name, int64_t delta, /* {{{ */ +static int statsd_metric_add (char const *name, double delta, /* {{{ */ metric_type_t type) { - cdtime_t now; statsd_metric_t *metric; - int status; - now = cdtime (); pthread_mutex_lock (&metrics_lock); - status = c_avl_get (metrics_tree, name, (void *) &metric); - if (status == 0) + metric = statsd_metric_lookup_unsafe (name, type); + if (metric == NULL) { - metric->value += delta; - metric->last_update = now; - pthread_mutex_unlock (&metrics_lock); - return (0); + return (-1); } - else /* no such value yet */ - { - status = statsd_metric_set_unsafe (name, delta, type); - pthread_mutex_unlock (&metrics_lock); - return (status); - } + 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) @@ -166,7 +220,7 @@ static int statsd_handle_counter (char const *name, /* {{{ */ scale.gauge = 1.0; if (extra != NULL) { - status = parse_value (extra + 1, &scale, DS_TYPE_GAUGE); + status = statsd_parse_value (extra + 1, &scale); if (status != 0) return (status); @@ -174,16 +228,12 @@ static int statsd_handle_counter (char const *name, /* {{{ */ return (-1); } - value.derive = 1; - status = parse_value (value_str, &value, DS_TYPE_DERIVE); + value.gauge = 1.0; + status = statsd_parse_value (value_str, &value); if (status != 0) return (status); - if (value.derive < 1) - return (-1); - - return (statsd_metric_add (name, - (int64_t) (((gauge_t) value.derive) / scale.gauge), + return (statsd_metric_add (name, (double) (value.gauge / scale.gauge), STATSD_COUNTER)); } /* }}} int statsd_handle_counter */ @@ -193,39 +243,109 @@ static int statsd_handle_gauge (char const *name, /* {{{ */ value_t value; int status; - value.derive = 0; - status = parse_value (value_str, &value, DS_TYPE_DERIVE); + 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, (int64_t) value.derive, STATSD_GAUGE)); + return (statsd_metric_add (name, (double) value.gauge, STATSD_GAUGE)); else - return (statsd_metric_set (name, (int64_t) value.derive, STATSD_GAUGE)); + 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) { - value_t value; + statsd_metric_t *metric; + value_t value_ms; + cdtime_t value; int status; - value.derive = 0; - status = parse_value (value_str, &value, DS_TYPE_DERIVE); + value_ms.derive = 0; + status = statsd_parse_value (value_str, &value_ms); if (status != 0) return (status); - return (statsd_metric_add (name, (int64_t) value.derive, STATSD_TIMER)); + 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 __attribute__((unused)), /* {{{ */ - char const *value_str __attribute__((unused))) +static int statsd_handle_set (char const *name, /* {{{ */ + char const *set_key_orig) { - static c_complain_t c = C_COMPLAIN_INIT_STATIC; + 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); + } - c_complain (LOG_WARNING, &c, - "statsd plugin: Support for sets is not yet implemented."); + /* 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 */ @@ -274,22 +394,32 @@ static int statsd_parse_line (char *buffer) /* {{{ */ static void statsd_parse_buffer (char *buffer) /* {{{ */ { - char *dummy; - char *saveptr = NULL; - char *ptr; - - for (dummy = buffer; - (ptr = strtok_r (dummy, "\r\n", &saveptr)) != NULL; - dummy = NULL) + while (buffer != NULL) { - char *line_orig = sstrdup (ptr); + char orig[64]; + char *next; int status; - status = statsd_parse_line (ptr); + 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\"", line_orig); + ERROR ("statsd plugin: Unable to parse line: \"%s\"", orig); - sfree (line_orig); + buffer = next; } } /* }}} void statsd_parse_buffer */ @@ -459,11 +589,80 @@ static void *statsd_network_thread (void *args) /* {{{ */ 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 ("TimerLower", child->key) == 0) + cf_util_get_boolean (child, &conf_timer_lower); + else if (strcasecmp ("TimerUpper", child->key) == 0) + cf_util_get_boolean (child, &conf_timer_upper); + else if (strcasecmp ("TimerSum", child->key) == 0) + cf_util_get_boolean (child, &conf_timer_sum); + else if (strcasecmp ("TimerCount", child->key) == 0) + cf_util_get_boolean (child, &conf_timer_count); + 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 *) strcasecmp); + metrics_tree = c_avl_create ((void *) strcmp); if (!network_thread_running) { @@ -489,17 +688,34 @@ static int statsd_init (void) /* {{{ */ return (0); } /* }}} int statsd_init */ -static int statsd_metric_submit (char const *name, /* {{{ */ +/* 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; - if (metric->type == STATSD_GAUGE) - values[0].gauge = (gauge_t) metric->value; - else - values[0].derive = (derive_t) metric->value; - vl.values = values; vl.values_len = 1; sstrncpy (vl.host, hostname_g, sizeof (vl.host)); @@ -508,21 +724,101 @@ static int statsd_metric_submit (char const *name, /* {{{ */ if (metric->type == STATSD_GAUGE) sstrncpy (vl.type, "gauge", sizeof (vl.type)); else if (metric->type == STATSD_TIMER) - sstrncpy (vl.type, "total_time_in_ms", sizeof (vl.type)); + 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); + + if (conf_timer_lower) { + ssnprintf (vl.type_instance, sizeof (vl.type_instance), + "%s-lower", name); + values[0].gauge = CDTIME_T_TO_DOUBLE ( + latency_counter_get_min (metric->latency)); + plugin_dispatch_values (&vl); + } + + if (conf_timer_upper) { + ssnprintf (vl.type_instance, sizeof (vl.type_instance), + "%s-upper", name); + values[0].gauge = CDTIME_T_TO_DOUBLE ( + latency_counter_get_max (metric->latency)); + plugin_dispatch_values (&vl); + } + + if (conf_timer_sum) { + ssnprintf (vl.type_instance, sizeof (vl.type_instance), + "%s-sum", name); + values[0].gauge = CDTIME_T_TO_DOUBLE ( + latency_counter_get_sum (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); + } + + /* Keep this at the end, since vl.type is set to "gauge" here. The + * vl.type's above are implicitly set to "latency". */ + if (conf_timer_count) { + sstrncpy (vl.type, "gauge", sizeof (vl.type)); + ssnprintf (vl.type_instance, sizeof (vl.type_instance), + "%s-count", name); + values[0].gauge = latency_counter_get_num (metric->latency); + 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 */ +} /* }}} int statsd_metric_submit_unsafe */ static int statsd_read (void) /* {{{ */ { - c_avl_iterator_t *i; + 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) @@ -531,13 +827,52 @@ static int statsd_read (void) /* {{{ */ return (0); } - i = c_avl_get_iterator (metrics_tree); - while (c_avl_iterator_next (i, (void *) &name, (void *) &metric) == 0) - statsd_metric_submit (name, metric); - c_avl_iterator_destroy (i); + 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 */ @@ -564,6 +899,9 @@ static int statsd_shutdown (void) /* {{{ */ c_avl_destroy (metrics_tree); metrics_tree = NULL; + sfree (conf_node); + sfree (conf_service); + pthread_mutex_unlock (&metrics_lock); return (0); @@ -571,6 +909,7 @@ static int statsd_shutdown (void) /* {{{ */ 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);