X-Git-Url: https://git.octo.it/?p=collectd.git;a=blobdiff_plain;f=src%2Fstatsd.c;h=72b8e2b6342b2beb92c71e1322a0e836649521ad;hp=0c6fc8768e8acd853c8b5d540702e250f5592460;hb=633c3966f770e4d46651a2fe219a18d8a9907a9f;hpb=2dd0f9076e04a3d28793f8f45d098ea0a743ab3b diff --git a/src/statsd.c b/src/statsd.c index 0c6fc876..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,14 +56,17 @@ 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; + double value; + latency_counter_t *latency; + c_avl_tree_t *set; unsigned long updates_num; }; typedef struct statsd_metric_s statsd_metric_t; @@ -70,90 +84,128 @@ 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) { + char key[DATA_MAX_NAME_LEN + 2]; + char *key_copy; statsd_metric_t *metric; - char *key; int status; - status = c_avl_get (metrics_tree, name, (void *) &metric); - if (status == 0) + switch (type) { - metric->value = value; - metric->updates_num++; + 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); + } - return (0); + 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); } - DEBUG ("stats plugin: Adding new metric \"%s\".", name); - /* FIXME: The keys should have a prefix so counter, gauge and timer with the - * same name can exist. */ - 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->updates_num = 1; + 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) { statsd_metric_t *metric; - int status; 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->updates_num++; - 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) @@ -168,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); @@ -176,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 */ @@ -195,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 */ @@ -276,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 */ @@ -461,6 +589,37 @@ 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; @@ -479,6 +638,18 @@ static int statsd_config (oconfig_item_t *ci) /* {{{ */ 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); @@ -491,7 +662,7 @@ 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) { @@ -517,24 +688,33 @@ static int statsd_init (void) /* {{{ */ return (0); } /* }}} int statsd_init */ -static int statsd_metric_submit (char const *name, /* {{{ */ - statsd_metric_t const *metric) +/* Must hold metrics_lock when calling this function. */ +static int statsd_metric_clear_set_unsafe (statsd_metric_t *metric) /* {{{ */ { - value_t values[1]; - value_list_t vl = VALUE_LIST_INIT; + void *key; + void *value; - if (metric->type == STATSD_GAUGE) - values[0].gauge = (gauge_t) metric->value; - else if (metric->type == STATSD_TIMER) + if ((metric == NULL) || (metric->type != STATSD_SET)) + return (EINVAL); + + if (metric->set == NULL) + return (0); + + while (c_avl_pick (metric->set, &key, &value) == 0) { - if (metric->updates_num == 0) - values[0].gauge = NAN; - else - values[0].gauge = - ((gauge_t) metric->value) / ((gauge_t) metric->updates_num); + sfree (key); + sfree (value); } - else - values[0].derive = (derive_t) metric->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; @@ -545,13 +725,89 @@ static int statsd_metric_submit (char const *name, /* {{{ */ 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); + + 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) /* {{{ */ { @@ -577,15 +833,22 @@ static int statsd_read (void) /* {{{ */ 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_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; } - statsd_metric_submit (name, metric); + /* 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);