X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Fstatsd.c;h=2def1914d69614dda6fb0f13c031abc6345d740d;hb=423cb2489520aaa15f0d77f2699aa0da74903c2f;hp=896fb619ffcdcc90095b042595ea8aa53e7477b0;hpb=cc4025b8c5561acb292e5fd025dfd4a2beb68165;p=collectd.git diff --git a/src/statsd.c b/src/statsd.c index 896fb619..2def1914 100644 --- a/src/statsd.c +++ b/src/statsd.c @@ -25,6 +25,7 @@ #include "configfile.h" #include "utils_avltree.h" #include "utils_complain.h" +#include "utils_latency.h" #include @@ -45,7 +46,8 @@ enum metric_type_e { STATSD_COUNTER, STATSD_TIMER, - STATSD_GAUGE + STATSD_GAUGE, + STATSD_SET }; typedef enum metric_type_e metric_type_t; @@ -53,7 +55,9 @@ struct statsd_metric_s { metric_type_t type; int64_t value; - cdtime_t last_update; + latency_counter_t *latency; + c_avl_tree_t *set; + unsigned long updates_num; }; typedef struct statsd_metric_s statsd_metric_t; @@ -67,95 +71,120 @@ 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 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 const *prefix; + 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: prefix = "c"; break; + case STATSD_TIMER: prefix = "t"; break; + case STATSD_GAUGE: prefix = "g"; break; + case STATSD_SET: prefix = "s"; break; + default: return (NULL); + } + + ssnprintf (key, sizeof (key), "%s:%s", prefix, name); - 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, /* {{{ */ 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, /* {{{ */ 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_handle_counter (char const *name, /* {{{ */ char const *value_str, char const *extra) { + char key[DATA_MAX_NAME_LEN + 2]; value_t value; value_t scale; int status; @@ -182,7 +211,9 @@ static int statsd_handle_counter (char const *name, /* {{{ */ if (value.derive < 1) return (-1); - return (statsd_metric_add (name, + ssnprintf (key, sizeof (key), "c:%s", name); + + return (statsd_metric_add (key, (int64_t) (((gauge_t) value.derive) / scale.gauge), STATSD_COUNTER)); } /* }}} int statsd_handle_counter */ @@ -190,6 +221,7 @@ static int statsd_handle_counter (char const *name, /* {{{ */ static int statsd_handle_gauge (char const *name, /* {{{ */ char const *value_str) { + char key[DATA_MAX_NAME_LEN + 2]; value_t value; int status; @@ -198,34 +230,106 @@ static int statsd_handle_gauge (char const *name, /* {{{ */ if (status != 0) return (status); + ssnprintf (key, sizeof (key), "g:%s", name); + if ((value_str[0] == '+') || (value_str[0] == '-')) - return (statsd_metric_add (name, (int64_t) value.derive, STATSD_GAUGE)); + return (statsd_metric_add (key, (int64_t) value.derive, STATSD_GAUGE)); else - return (statsd_metric_set (name, (int64_t) value.derive, STATSD_GAUGE)); + return (statsd_metric_set (key, (int64_t) value.derive, 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 = parse_value (value_str, &value_ms, DS_TYPE_DERIVE); if (status != 0) return (status); - return (statsd_metric_add (name, (int64_t) value.derive, STATSD_TIMER)); + value = MS_TO_CDTIME_T (value_ms.derive); + + 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; - c_complain (LOG_WARNING, &c, - "statsd plugin: Support for sets is not yet implemented."); + 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 */ @@ -459,6 +563,67 @@ 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 ("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); @@ -489,17 +654,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 +690,67 @@ 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); + + 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 (metrics->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 +759,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 +831,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 +841,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);