// SPDX-License-Identifier: GPL-3.0-or-later #include "daemon/common.h" #define STATSD_CHART_PREFIX "statsd" #define PLUGIN_STATSD_NAME "statsd.plugin" #define STATSD_LISTEN_PORT 8125 #define STATSD_LISTEN_BACKLOG 4096 #define WORKER_JOB_TYPE_TCP_CONNECTED 0 #define WORKER_JOB_TYPE_TCP_DISCONNECTED 1 #define WORKER_JOB_TYPE_RCV_DATA 2 #define WORKER_JOB_TYPE_SND_DATA 3 #if WORKER_UTILIZATION_MAX_JOB_TYPES < 4 #error Please increase WORKER_UTILIZATION_MAX_JOB_TYPES to at least 4 #endif // -------------------------------------------------------------------------------------- // #define STATSD_MULTITHREADED 1 #ifdef STATSD_MULTITHREADED // DO NOT ENABLE MULTITHREADING - IT IS NOT WELL TESTED #define STATSD_DICTIONARY_OPTIONS DICTIONARY_FLAG_DONT_OVERWRITE_VALUE|DICTIONARY_FLAG_ADD_IN_FRONT #else #define STATSD_DICTIONARY_OPTIONS DICTIONARY_FLAG_DONT_OVERWRITE_VALUE|DICTIONARY_FLAG_ADD_IN_FRONT|DICTIONARY_FLAG_SINGLE_THREADED #endif #define STATSD_DECIMAL_DETAIL 1000 // floating point values get multiplied by this, with the same divisor // -------------------------------------------------------------------------------------------------------------------- // data specific to each metric type typedef struct statsd_metric_gauge { NETDATA_DOUBLE value; } STATSD_METRIC_GAUGE; typedef struct statsd_metric_counter { // counter and meter long long value; } STATSD_METRIC_COUNTER; typedef struct statsd_histogram_extensions { netdata_mutex_t mutex; // average is stored in metric->last collected_number last_min; collected_number last_max; collected_number last_percentile; collected_number last_median; collected_number last_stddev; collected_number last_sum; int zeroed; RRDDIM *rd_min; RRDDIM *rd_max; RRDDIM *rd_percentile; RRDDIM *rd_median; RRDDIM *rd_stddev; //RRDDIM *rd_sum; size_t size; size_t used; NETDATA_DOUBLE *values; // dynamic array of values collected } STATSD_METRIC_HISTOGRAM_EXTENSIONS; typedef struct statsd_metric_histogram { // histogram and timer STATSD_METRIC_HISTOGRAM_EXTENSIONS *ext; } STATSD_METRIC_HISTOGRAM; typedef struct statsd_metric_set { DICTIONARY *dict; size_t unique; } STATSD_METRIC_SET; typedef struct statsd_metric_dictionary_item { size_t count; RRDDIM *rd; } STATSD_METRIC_DICTIONARY_ITEM; typedef struct statsd_metric_dictionary { DICTIONARY *dict; size_t unique; } STATSD_METRIC_DICTIONARY; // -------------------------------------------------------------------------------------------------------------------- // this is a metric - for all types of metrics typedef enum statsd_metric_options { STATSD_METRIC_OPTION_NONE = 0x00000000, // no options set STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED = 0x00000001, // do not update the chart dimension, when this metric is not collected STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED = 0x00000002, // render a private chart for this metric STATSD_METRIC_OPTION_PRIVATE_CHART_CHECKED = 0x00000004, // the metric has been checked if it should get private chart or not STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT = 0x00000008, // show the count of events for this private chart STATSD_METRIC_OPTION_CHECKED_IN_APPS = 0x00000010, // set when this metric has been checked against apps STATSD_METRIC_OPTION_USED_IN_APPS = 0x00000020, // set when this metric is used in apps STATSD_METRIC_OPTION_CHECKED = 0x00000040, // set when the charting thread checks this metric for use in charts (its usefulness) STATSD_METRIC_OPTION_USEFUL = 0x00000080, // set when the charting thread finds the metric useful (i.e. used in a chart) STATSD_METRIC_OPTION_COLLECTION_FULL_LOGGED = 0x00000100, // set when the collection is full for this metric } STATS_METRIC_OPTIONS; typedef enum statsd_metric_type { STATSD_METRIC_TYPE_GAUGE, STATSD_METRIC_TYPE_COUNTER, STATSD_METRIC_TYPE_METER, STATSD_METRIC_TYPE_TIMER, STATSD_METRIC_TYPE_HISTOGRAM, STATSD_METRIC_TYPE_SET, STATSD_METRIC_TYPE_DICTIONARY } STATSD_METRIC_TYPE; typedef struct statsd_metric { const char *name; // the name of the metric - linked to dictionary name uint32_t hash; // hash of the name STATSD_METRIC_TYPE type; // metadata about data collection collected_number events; // the number of times this metric has been collected (never resets) size_t count; // the number of times this metric has been collected since the last flush // the actual collected data union { STATSD_METRIC_GAUGE gauge; STATSD_METRIC_COUNTER counter; STATSD_METRIC_HISTOGRAM histogram; STATSD_METRIC_SET set; STATSD_METRIC_DICTIONARY dictionary; }; char *units; char *dimname; char *family; // chart related members STATS_METRIC_OPTIONS options; // STATSD_METRIC_OPTION_* (bitfield) char reset; // set to 1 by the charting thread to instruct the collector thread(s) to reset this metric collected_number last; // the last value sent to netdata RRDSET *st; // the private chart of this metric RRDDIM *rd_value; // the dimension of this metric value RRDDIM *rd_count; // the dimension for the number of events received // linking, used for walking through all metrics struct statsd_metric *next_useful; } STATSD_METRIC; // -------------------------------------------------------------------------------------------------------------------- // each type of metric has its own index typedef struct statsd_index { char *name; // the name of the index of metrics size_t events; // the number of events processed for this index size_t metrics; // the number of metrics in this index size_t useful; // the number of useful metrics in this index STATSD_METRIC_TYPE type; // the type of index DICTIONARY *dict; STATSD_METRIC *first_useful; // the linked list of useful metrics (new metrics are added in front) STATS_METRIC_OPTIONS default_options; // default options for all metrics in this index } STATSD_INDEX; // -------------------------------------------------------------------------------------------------------------------- // synthetic charts typedef enum statsd_app_chart_dimension_value_type { STATSD_APP_CHART_DIM_VALUE_TYPE_EVENTS, STATSD_APP_CHART_DIM_VALUE_TYPE_LAST, STATSD_APP_CHART_DIM_VALUE_TYPE_AVERAGE, STATSD_APP_CHART_DIM_VALUE_TYPE_SUM, STATSD_APP_CHART_DIM_VALUE_TYPE_MIN, STATSD_APP_CHART_DIM_VALUE_TYPE_MAX, STATSD_APP_CHART_DIM_VALUE_TYPE_PERCENTILE, STATSD_APP_CHART_DIM_VALUE_TYPE_MEDIAN, STATSD_APP_CHART_DIM_VALUE_TYPE_STDDEV } STATSD_APP_CHART_DIM_VALUE_TYPE; typedef struct statsd_app_chart_dimension { const char *name; // the name of this dimension const char *metric; // the source metric name of this dimension uint32_t metric_hash; // hash for fast string comparisons SIMPLE_PATTERN *metric_pattern; // set when the 'metric' is a simple pattern collected_number multiplier; // the multiplier of the dimension collected_number divisor; // the divisor of the dimension RRDDIM_FLAGS flags; // the RRDDIM flags for this dimension STATSD_APP_CHART_DIM_VALUE_TYPE value_type; // which value to use of the source metric RRDDIM *rd; // a pointer to the RRDDIM that has been created for this dimension collected_number *value_ptr; // a pointer to the source metric value RRD_ALGORITHM algorithm; // the algorithm of this dimension struct statsd_app_chart_dimension *next; // the next dimension for this chart } STATSD_APP_CHART_DIM; typedef struct statsd_app_chart { const char *id; const char *name; const char *title; const char *family; const char *context; const char *units; const char *module; long priority; RRDSET_TYPE chart_type; STATSD_APP_CHART_DIM *dimensions; size_t dimensions_count; size_t dimensions_linked_count; RRDSET *st; struct statsd_app_chart *next; } STATSD_APP_CHART; typedef struct statsd_app { const char *name; SIMPLE_PATTERN *metrics; STATS_METRIC_OPTIONS default_options; RRD_MEMORY_MODE rrd_memory_mode; DICTIONARY *dict; long rrd_history_entries; const char *source; STATSD_APP_CHART *charts; struct statsd_app *next; } STATSD_APP; // -------------------------------------------------------------------------------------------------------------------- // global statsd data struct collection_thread_status { int status; size_t max_sockets; netdata_thread_t thread; }; static struct statsd { STATSD_INDEX gauges; STATSD_INDEX counters; STATSD_INDEX timers; STATSD_INDEX histograms; STATSD_INDEX meters; STATSD_INDEX sets; STATSD_INDEX dictionaries; size_t unknown_types; size_t socket_errors; size_t tcp_socket_connects; size_t tcp_socket_disconnects; size_t tcp_socket_connected; size_t tcp_socket_reads; size_t tcp_packets_received; size_t tcp_bytes_read; size_t udp_socket_reads; size_t udp_packets_received; size_t udp_bytes_read; int enabled; int update_every; SIMPLE_PATTERN *charts_for; size_t tcp_idle_timeout; collected_number decimal_detail; size_t private_charts; size_t max_private_charts_hard; long private_charts_rrd_history_entries; unsigned int private_charts_hidden:1; STATSD_APP *apps; size_t recvmmsg_size; size_t histogram_increase_step; double histogram_percentile; char *histogram_percentile_str; size_t dictionary_max_unique; int threads; struct collection_thread_status *collection_threads_status; LISTEN_SOCKETS sockets; } statsd = { .enabled = 1, .max_private_charts_hard = 1000, .private_charts_hidden = 0, .recvmmsg_size = 10, .decimal_detail = STATSD_DECIMAL_DETAIL, .gauges = { .name = "gauge", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_GAUGE, .default_options = STATSD_METRIC_OPTION_NONE }, .counters = { .name = "counter", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_COUNTER, .default_options = STATSD_METRIC_OPTION_NONE }, .timers = { .name = "timer", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_TIMER, .default_options = STATSD_METRIC_OPTION_NONE }, .histograms = { .name = "histogram", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_HISTOGRAM, .default_options = STATSD_METRIC_OPTION_NONE }, .meters = { .name = "meter", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_METER, .default_options = STATSD_METRIC_OPTION_NONE }, .sets = { .name = "set", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_SET, .default_options = STATSD_METRIC_OPTION_NONE }, .dictionaries = { .name = "dictionary", .events = 0, .metrics = 0, .dict = NULL, .type = STATSD_METRIC_TYPE_DICTIONARY, .default_options = STATSD_METRIC_OPTION_NONE }, .tcp_idle_timeout = 600, .apps = NULL, .histogram_percentile = 95.0, .histogram_increase_step = 10, .dictionary_max_unique = 200, .threads = 0, .collection_threads_status = NULL, .sockets = { .config = &netdata_config, .config_section = CONFIG_SECTION_STATSD, .default_bind_to = "udp:localhost tcp:localhost", .default_port = STATSD_LISTEN_PORT, .backlog = STATSD_LISTEN_BACKLOG }, }; // -------------------------------------------------------------------------------------------------------------------- // statsd index management - add/find metrics static void dictionary_metric_insert_callback(const char *name, void *value, void *data) { STATSD_INDEX *index = (STATSD_INDEX *)data; STATSD_METRIC *m = (STATSD_METRIC *)value; debug(D_STATSD, "Creating new %s metric '%s'", index->name, name); m->name = name; m->hash = simple_hash(name); m->type = index->type; m->options = index->default_options; if (m->type == STATSD_METRIC_TYPE_HISTOGRAM || m->type == STATSD_METRIC_TYPE_TIMER) { m->histogram.ext = callocz(1,sizeof(STATSD_METRIC_HISTOGRAM_EXTENSIONS)); netdata_mutex_init(&m->histogram.ext->mutex); } __atomic_fetch_add(&index->metrics, 1, __ATOMIC_RELAXED); } static void dictionary_metric_delete_callback(const char *name, void *value, void *data) { (void)data; // STATSD_INDEX *index = (STATSD_INDEX *)data; (void)name; STATSD_METRIC *m = (STATSD_METRIC *)value; if(m->type == STATSD_METRIC_TYPE_HISTOGRAM || m->type == STATSD_METRIC_TYPE_TIMER) { freez(m->histogram.ext); m->histogram.ext = NULL; } freez(m->units); freez(m->family); freez(m->dimname); } static inline STATSD_METRIC *statsd_find_or_add_metric(STATSD_INDEX *index, const char *name) { debug(D_STATSD, "searching for metric '%s' under '%s'", name, index->name); #ifdef STATSD_MULTITHREADED // avoid the write lock of dictionary_set() for existing metrics STATSD_METRIC *m = dictionary_get(index->dict, name); if(!m) m = dictionary_set(index->dict, name, NULL, sizeof(STATSD_METRIC)); #else // no locks here, go faster // this will call the dictionary_metric_insert_callback() if an item // is inserted, otherwise it will return the existing one. // We used the flag DICTIONARY_FLAG_DONT_OVERWRITE_VALUE to support this. STATSD_METRIC *m = dictionary_set(index->dict, name, NULL, sizeof(STATSD_METRIC)); #endif index->events++; return m; } // -------------------------------------------------------------------------------------------------------------------- // statsd parsing numbers static inline NETDATA_DOUBLE statsd_parse_float(const char *v, NETDATA_DOUBLE def) { NETDATA_DOUBLE value; if(likely(v && *v)) { char *e = NULL; value = str2ndd(v, &e); if(unlikely(e && *e)) error("STATSD: excess data '%s' after value '%s'", e, v); } else value = def; return value; } static inline NETDATA_DOUBLE statsd_parse_sampling_rate(const char *v) { NETDATA_DOUBLE sampling_rate = statsd_parse_float(v, 1.0); if(unlikely(isless(sampling_rate, 0.001))) sampling_rate = 0.001; if(unlikely(isgreater(sampling_rate, 1.0))) sampling_rate = 1.0; return sampling_rate; } static inline long long statsd_parse_int(const char *v, long long def) { long long value; if(likely(v && *v)) { char *e = NULL; value = str2ll(v, &e); if(unlikely(e && *e)) error("STATSD: excess data '%s' after value '%s'", e, v); } else value = def; return value; } // -------------------------------------------------------------------------------------------------------------------- // statsd processors per metric type static inline void statsd_reset_metric(STATSD_METRIC *m) { m->reset = 0; m->count = 0; } static inline int value_is_zinit(const char *value) { return (value && *value == 'z' && *++value == 'i' && *++value == 'n' && *++value == 'i' && *++value == 't' && *++value == '\0'); } #define is_metric_checked(m) ((m)->options & STATSD_METRIC_OPTION_CHECKED) #define is_metric_useful_for_collection(m) (!is_metric_checked(m) || ((m)->options & STATSD_METRIC_OPTION_USEFUL)) static inline void statsd_process_gauge(STATSD_METRIC *m, const char *value, const char *sampling) { if(!is_metric_useful_for_collection(m)) return; if(unlikely(!value || !*value)) { error("STATSD: metric '%s' of type gauge, with empty value is ignored.", m->name); return; } if(unlikely(m->reset)) { // no need to reset anything specific for gauges statsd_reset_metric(m); } if(unlikely(value_is_zinit(value))) { // magic loading of metric, without affecting anything } else { if (unlikely(*value == '+' || *value == '-')) m->gauge.value += statsd_parse_float(value, 1.0) / statsd_parse_sampling_rate(sampling); else m->gauge.value = statsd_parse_float(value, 1.0); m->events++; m->count++; } } static inline void statsd_process_counter_or_meter(STATSD_METRIC *m, const char *value, const char *sampling) { if(!is_metric_useful_for_collection(m)) return; // we accept empty values for counters if(unlikely(m->reset)) statsd_reset_metric(m); if(unlikely(value_is_zinit(value))) { // magic loading of metric, without affecting anything } else { m->counter.value += llrintndd((NETDATA_DOUBLE) statsd_parse_int(value, 1) / statsd_parse_sampling_rate(sampling)); m->events++; m->count++; } } #define statsd_process_counter(m, value, sampling) statsd_process_counter_or_meter(m, value, sampling) #define statsd_process_meter(m, value, sampling) statsd_process_counter_or_meter(m, value, sampling) static inline void statsd_process_histogram_or_timer(STATSD_METRIC *m, const char *value, const char *sampling, const char *type) { if(!is_metric_useful_for_collection(m)) return; if(unlikely(!value || !*value)) { error("STATSD: metric of type %s, with empty value is ignored.", type); return; } if(unlikely(m->reset)) { m->histogram.ext->used = 0; statsd_reset_metric(m); } if(unlikely(value_is_zinit(value))) { // magic loading of metric, without affecting anything } else { NETDATA_DOUBLE v = statsd_parse_float(value, 1.0); NETDATA_DOUBLE sampling_rate = statsd_parse_sampling_rate(sampling); if(unlikely(isless(sampling_rate, 0.01))) sampling_rate = 0.01; if(unlikely(isgreater(sampling_rate, 1.0))) sampling_rate = 1.0; long long samples = llrintndd(1.0 / sampling_rate); while(samples-- > 0) { if(unlikely(m->histogram.ext->used == m->histogram.ext->size)) { netdata_mutex_lock(&m->histogram.ext->mutex); m->histogram.ext->size += statsd.histogram_increase_step; m->histogram.ext->values = reallocz(m->histogram.ext->values, sizeof(NETDATA_DOUBLE) * m->histogram.ext->size); netdata_mutex_unlock(&m->histogram.ext->mutex); } m->histogram.ext->values[m->histogram.ext->used++] = v; } m->events++; m->count++; } } #define statsd_process_timer(m, value, sampling) statsd_process_histogram_or_timer(m, value, sampling, "timer") #define statsd_process_histogram(m, value, sampling) statsd_process_histogram_or_timer(m, value, sampling, "histogram") static void dictionary_metric_set_value_insert_callback(const char *name, void *value, void *data) { (void)name; (void)value; STATSD_METRIC *m = (STATSD_METRIC *)data; m->set.unique++; } static inline void statsd_process_set(STATSD_METRIC *m, const char *value) { if(!is_metric_useful_for_collection(m)) return; if(unlikely(!value || !*value)) { error("STATSD: metric of type set, with empty value is ignored."); return; } if(unlikely(m->reset)) { if(likely(m->set.dict)) { dictionary_destroy(m->set.dict); m->set.dict = NULL; } statsd_reset_metric(m); } if (unlikely(!m->set.dict)) { m->set.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); dictionary_register_insert_callback(m->set.dict, dictionary_metric_set_value_insert_callback, m); m->set.unique = 0; } if(unlikely(value_is_zinit(value))) { // magic loading of metric, without affecting anything } else { #ifdef STATSD_MULTITHREADED // avoid the write lock to check if something is already there if(!dictionary_get(m->set.dict, value)) dictionary_set(m->set.dict, value, NULL, 0); #else dictionary_set(m->set.dict, value, NULL, 0); #endif m->events++; m->count++; } } static void dictionary_metric_dict_value_insert_callback(const char *name, void *value, void *data) { (void)name; (void)value; STATSD_METRIC *m = (STATSD_METRIC *)data; m->dictionary.unique++; } static inline void statsd_process_dictionary(STATSD_METRIC *m, const char *value) { if(!is_metric_useful_for_collection(m)) return; if(unlikely(!value || !*value)) { error("STATSD: metric of type set, with empty value is ignored."); return; } if(unlikely(m->reset)) statsd_reset_metric(m); if (unlikely(!m->dictionary.dict)) { m->dictionary.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); dictionary_register_insert_callback(m->dictionary.dict, dictionary_metric_dict_value_insert_callback, m); m->dictionary.unique = 0; } if(unlikely(value_is_zinit(value))) { // magic loading of metric, without affecting anything } else { STATSD_METRIC_DICTIONARY_ITEM *t = (STATSD_METRIC_DICTIONARY_ITEM *)dictionary_get(m->dictionary.dict, value); if (unlikely(!t)) { if(!t && m->dictionary.unique >= statsd.dictionary_max_unique) value = "other"; t = (STATSD_METRIC_DICTIONARY_ITEM *)dictionary_set(m->dictionary.dict, value, NULL, sizeof(STATSD_METRIC_DICTIONARY_ITEM)); } t->count++; m->events++; m->count++; } } // -------------------------------------------------------------------------------------------------------------------- // statsd parsing static inline const char *statsd_parse_skip_up_to(const char *s, char d1, char d2, char d3) { char c; for(c = *s; c && c != d1 && c != d2 && c != d3 && c != '\r' && c != '\n'; c = *++s) ; return s; } const char *statsd_parse_skip_spaces(const char *s) { char c; for(c = *s; c && ( c == ' ' || c == '\t' || c == '\r' || c == '\n' ); c = *++s) ; return s; } static inline const char *statsd_parse_field_trim(const char *start, char *end) { if(unlikely(!start || !*start)) { start = end; return start; } while(start <= end && (*start == ' ' || *start == '\t')) start++; *end = '\0'; end--; while(end >= start && (*end == ' ' || *end == '\t')) *end-- = '\0'; return start; } static void statsd_process_metric(const char *name, const char *value, const char *type, const char *sampling, const char *tags) { debug(D_STATSD, "STATSD: raw metric '%s', value '%s', type '%s', sampling '%s', tags '%s'", name?name:"(null)", value?value:"(null)", type?type:"(null)", sampling?sampling:"(null)", tags?tags:"(null)"); if(unlikely(!name || !*name)) return; if(unlikely(!type || !*type)) type = "m"; STATSD_METRIC *m = NULL; char t0 = type[0], t1 = type[1]; if(unlikely(t0 == 'g' && t1 == '\0')) { statsd_process_gauge( m = statsd_find_or_add_metric(&statsd.gauges, name), value, sampling); } else if(unlikely((t0 == 'c' || t0 == 'C') && t1 == '\0')) { // etsy/statsd uses 'c' // brubeck uses 'C' statsd_process_counter( m = statsd_find_or_add_metric(&statsd.counters, name), value, sampling); } else if(unlikely(t0 == 'm' && t1 == '\0')) { statsd_process_meter( m = statsd_find_or_add_metric(&statsd.meters, name), value, sampling); } else if(unlikely(t0 == 'h' && t1 == '\0')) { statsd_process_histogram( m = statsd_find_or_add_metric(&statsd.histograms, name), value, sampling); } else if(unlikely(t0 == 's' && t1 == '\0')) { statsd_process_set( m = statsd_find_or_add_metric(&statsd.sets, name), value); } else if(unlikely(t0 == 'd' && t1 == '\0')) { statsd_process_dictionary( m = statsd_find_or_add_metric(&statsd.dictionaries, name), value); } else if(unlikely(t0 == 'm' && t1 == 's' && type[2] == '\0')) { statsd_process_timer( m = statsd_find_or_add_metric(&statsd.timers, name), value, sampling); } else { statsd.unknown_types++; error("STATSD: metric '%s' with value '%s' is sent with unknown metric type '%s'", name, value?value:"", type); } if(m && tags && *tags) { const char *s = tags; while(*s) { const char *tagkey = NULL, *tagvalue = NULL; char *tagkey_end = NULL, *tagvalue_end = NULL; s = tagkey_end = (char *)statsd_parse_skip_up_to(tagkey = s, ':', '=', ','); if(tagkey == tagkey_end) { if (*s) { s++; s = statsd_parse_skip_spaces(s); } continue; } if(likely(*s == ':' || *s == '=')) s = tagvalue_end = (char *) statsd_parse_skip_up_to(tagvalue = ++s, ',', '\0', '\0'); if(*s == ',') s++; statsd_parse_field_trim(tagkey, tagkey_end); statsd_parse_field_trim(tagvalue, tagvalue_end); if(tagkey && *tagkey && tagvalue && *tagvalue) { if (!m->units && strcmp(tagkey, "units") == 0) m->units = strdupz(tagvalue); if (!m->dimname && strcmp(tagkey, "name") == 0) m->dimname = strdupz(tagvalue); if (!m->family && strcmp(tagkey, "family") == 0) m->family = strdupz(tagvalue); } } } } static inline size_t statsd_process(char *buffer, size_t size, int require_newlines) { buffer[size] = '\0'; debug(D_STATSD, "RECEIVED: %zu bytes: '%s'", size, buffer); const char *s = buffer; while(*s) { const char *name = NULL, *value = NULL, *type = NULL, *sampling = NULL, *tags = NULL; char *name_end = NULL, *value_end = NULL, *type_end = NULL, *sampling_end = NULL, *tags_end = NULL; s = name_end = (char *)statsd_parse_skip_up_to(name = s, ':', '=', '|'); if(name == name_end) { if (*s) { s++; s = statsd_parse_skip_spaces(s); } continue; } if(likely(*s == ':' || *s == '=')) s = value_end = (char *) statsd_parse_skip_up_to(value = ++s, '|', '@', '#'); if(likely(*s == '|')) s = type_end = (char *) statsd_parse_skip_up_to(type = ++s, '|', '@', '#'); while(*s == '|' || *s == '@' || *s == '#') { // parse all the fields that may be appended if ((*s == '|' && s[1] == '@') || *s == '@') { s = sampling_end = (char *)statsd_parse_skip_up_to(sampling = ++s, '|', '@', '#'); if (*sampling == '@') sampling++; } else if ((*s == '|' && s[1] == '#') || *s == '#') { s = tags_end = (char *)statsd_parse_skip_up_to(tags = ++s, '|', '@', '#'); if (*tags == '#') tags++; } else { // unknown field, skip it s = (char *)statsd_parse_skip_up_to(++s, '|', '@', '#'); } } // skip everything until the end of the line while(*s && *s != '\n') s++; if(unlikely(require_newlines && *s != '\n' && s > buffer)) { // move the remaining data to the beginning size -= (name - buffer); memmove(buffer, name, size); return size; } else s = statsd_parse_skip_spaces(s); statsd_process_metric( statsd_parse_field_trim(name, name_end) , statsd_parse_field_trim(value, value_end) , statsd_parse_field_trim(type, type_end) , statsd_parse_field_trim(sampling, sampling_end) , statsd_parse_field_trim(tags, tags_end) ); } return 0; } // -------------------------------------------------------------------------------------------------------------------- // statsd pollfd interface #define STATSD_TCP_BUFFER_SIZE 65536 // minimize tcp reads #define STATSD_UDP_BUFFER_SIZE 9000 // this should be up to MTU typedef enum { STATSD_SOCKET_DATA_TYPE_TCP, STATSD_SOCKET_DATA_TYPE_UDP } STATSD_SOCKET_DATA_TYPE; struct statsd_tcp { STATSD_SOCKET_DATA_TYPE type; size_t size; size_t len; char buffer[]; }; #ifdef HAVE_RECVMMSG struct statsd_udp { int *running; STATSD_SOCKET_DATA_TYPE type; size_t size; struct iovec *iovecs; struct mmsghdr *msgs; }; #else struct statsd_udp { int *running; STATSD_SOCKET_DATA_TYPE type; char buffer[STATSD_UDP_BUFFER_SIZE]; }; #endif // new TCP client connected static void *statsd_add_callback(POLLINFO *pi, short int *events, void *data) { (void)pi; (void)data; worker_is_busy(WORKER_JOB_TYPE_TCP_CONNECTED); *events = POLLIN; struct statsd_tcp *t = (struct statsd_tcp *)callocz(sizeof(struct statsd_tcp) + STATSD_TCP_BUFFER_SIZE, 1); t->type = STATSD_SOCKET_DATA_TYPE_TCP; t->size = STATSD_TCP_BUFFER_SIZE - 1; statsd.tcp_socket_connects++; statsd.tcp_socket_connected++; worker_is_idle(); return t; } // TCP client disconnected static void statsd_del_callback(POLLINFO *pi) { worker_is_busy(WORKER_JOB_TYPE_TCP_DISCONNECTED); struct statsd_tcp *t = pi->data; if(likely(t)) { if(t->type == STATSD_SOCKET_DATA_TYPE_TCP) { if(t->len != 0) { statsd.socket_errors++; error("STATSD: client is probably sending unterminated metrics. Closed socket left with '%s'. Trying to process it.", t->buffer); statsd_process(t->buffer, t->len, 0); } statsd.tcp_socket_disconnects++; statsd.tcp_socket_connected--; } else error("STATSD: internal error: received socket data type is %d, but expected %d", (int)t->type, (int)STATSD_SOCKET_DATA_TYPE_TCP); freez(t); } worker_is_idle(); } // Receive data static int statsd_rcv_callback(POLLINFO *pi, short int *events) { int retval = -1; worker_is_busy(WORKER_JOB_TYPE_RCV_DATA); *events = POLLIN; int fd = pi->fd; switch(pi->socktype) { case SOCK_STREAM: { struct statsd_tcp *d = (struct statsd_tcp *)pi->data; if(unlikely(!d)) { error("STATSD: internal error: expected TCP data pointer is NULL"); statsd.socket_errors++; retval = -1; goto cleanup; } #ifdef NETDATA_INTERNAL_CHECKS if(unlikely(d->type != STATSD_SOCKET_DATA_TYPE_TCP)) { error("STATSD: internal error: socket data type should be %d, but it is %d", (int)STATSD_SOCKET_DATA_TYPE_TCP, (int)d->type); statsd.socket_errors++; retval = -1; goto cleanup; } #endif int ret = 0; ssize_t rc; do { rc = recv(fd, &d->buffer[d->len], d->size - d->len, MSG_DONTWAIT); if (rc < 0) { // read failed if (errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR) { error("STATSD: recv() on TCP socket %d failed.", fd); statsd.socket_errors++; ret = -1; } } else if (!rc) { // connection closed debug(D_STATSD, "STATSD: client disconnected."); ret = -1; } else { // data received d->len += rc; statsd.tcp_socket_reads++; statsd.tcp_bytes_read += rc; } if(likely(d->len > 0)) { statsd.tcp_packets_received++; d->len = statsd_process(d->buffer, d->len, 1); } if(unlikely(ret == -1)) { retval = -1; goto cleanup; } } while (rc != -1); break; } case SOCK_DGRAM: { struct statsd_udp *d = (struct statsd_udp *)pi->data; if(unlikely(!d)) { error("STATSD: internal error: expected UDP data pointer is NULL"); statsd.socket_errors++; retval = -1; goto cleanup; } #ifdef NETDATA_INTERNAL_CHECKS if(unlikely(d->type != STATSD_SOCKET_DATA_TYPE_UDP)) { error("STATSD: internal error: socket data should be %d, but it is %d", (int)d->type, (int)STATSD_SOCKET_DATA_TYPE_UDP); statsd.socket_errors++; retval = -1; goto cleanup; } #endif #ifdef HAVE_RECVMMSG ssize_t rc; do { rc = recvmmsg(fd, d->msgs, (unsigned int)d->size, MSG_DONTWAIT, NULL); if (rc < 0) { // read failed if (errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR) { error("STATSD: recvmmsg() on UDP socket %d failed.", fd); statsd.socket_errors++; retval = -1; goto cleanup; } } else if (rc) { // data received statsd.udp_socket_reads++; statsd.udp_packets_received += rc; size_t i; for (i = 0; i < (size_t)rc; ++i) { size_t len = (size_t)d->msgs[i].msg_len; statsd.udp_bytes_read += len; statsd_process(d->msgs[i].msg_hdr.msg_iov->iov_base, len, 0); } } } while (rc != -1); #else // !HAVE_RECVMMSG ssize_t rc; do { rc = recv(fd, d->buffer, STATSD_UDP_BUFFER_SIZE - 1, MSG_DONTWAIT); if (rc < 0) { // read failed if (errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR) { error("STATSD: recv() on UDP socket %d failed.", fd); statsd.socket_errors++; retval = -1; goto cleanup; } } else if (rc) { // data received statsd.udp_socket_reads++; statsd.udp_packets_received++; statsd.udp_bytes_read += rc; statsd_process(d->buffer, (size_t) rc, 0); } } while (rc != -1); #endif break; } default: { error("STATSD: internal error: unknown socktype %d on socket %d", pi->socktype, fd); statsd.socket_errors++; retval = -1; goto cleanup; } } retval = 0; cleanup: worker_is_idle(); return retval; } static int statsd_snd_callback(POLLINFO *pi, short int *events) { (void)pi; (void)events; worker_is_busy(WORKER_JOB_TYPE_SND_DATA); error("STATSD: snd_callback() called, but we never requested to send data to statsd clients."); worker_is_idle(); return -1; } // -------------------------------------------------------------------------------------------------------------------- // statsd child thread to collect metrics from network void statsd_collector_thread_cleanup(void *data) { struct statsd_udp *d = data; *d->running = 0; info("cleaning up..."); #ifdef HAVE_RECVMMSG size_t i; for (i = 0; i < d->size; i++) freez(d->iovecs[i].iov_base); freez(d->iovecs); freez(d->msgs); #endif freez(d); worker_unregister(); } void *statsd_collector_thread(void *ptr) { struct collection_thread_status *status = ptr; status->status = 1; worker_register("STATSD"); worker_register_job_name(WORKER_JOB_TYPE_TCP_CONNECTED, "tcp connect"); worker_register_job_name(WORKER_JOB_TYPE_TCP_DISCONNECTED, "tcp disconnect"); worker_register_job_name(WORKER_JOB_TYPE_RCV_DATA, "receive"); worker_register_job_name(WORKER_JOB_TYPE_SND_DATA, "send"); info("STATSD collector thread started with taskid %d", gettid()); struct statsd_udp *d = callocz(sizeof(struct statsd_udp), 1); d->running = &status->status; netdata_thread_cleanup_push(statsd_collector_thread_cleanup, d); #ifdef HAVE_RECVMMSG d->type = STATSD_SOCKET_DATA_TYPE_UDP; d->size = statsd.recvmmsg_size; d->iovecs = callocz(sizeof(struct iovec), d->size); d->msgs = callocz(sizeof(struct mmsghdr), d->size); size_t i; for (i = 0; i < d->size; i++) { d->iovecs[i].iov_base = mallocz(STATSD_UDP_BUFFER_SIZE); d->iovecs[i].iov_len = STATSD_UDP_BUFFER_SIZE - 1; d->msgs[i].msg_hdr.msg_iov = &d->iovecs[i]; d->msgs[i].msg_hdr.msg_iovlen = 1; } #endif poll_events(&statsd.sockets , statsd_add_callback , statsd_del_callback , statsd_rcv_callback , statsd_snd_callback , NULL , NULL // No access control pattern , 0 // No dns lookups for access control pattern , (void *)d , 0 // tcp request timeout, 0 = disabled , statsd.tcp_idle_timeout // tcp idle timeout, 0 = disabled , statsd.update_every * 1000 , ptr // timer_data , status->max_sockets ); netdata_thread_cleanup_pop(1); return NULL; } // -------------------------------------------------------------------------------------------------------------------- // statsd applications configuration files parsing #define STATSD_CONF_LINE_MAX 8192 static STATSD_APP_CHART_DIM_VALUE_TYPE string2valuetype(const char *type, size_t line, const char *filename) { if(!type || !*type) type = "last"; if(!strcmp(type, "events")) return STATSD_APP_CHART_DIM_VALUE_TYPE_EVENTS; else if(!strcmp(type, "last")) return STATSD_APP_CHART_DIM_VALUE_TYPE_LAST; else if(!strcmp(type, "min")) return STATSD_APP_CHART_DIM_VALUE_TYPE_MIN; else if(!strcmp(type, "max")) return STATSD_APP_CHART_DIM_VALUE_TYPE_MAX; else if(!strcmp(type, "sum")) return STATSD_APP_CHART_DIM_VALUE_TYPE_SUM; else if(!strcmp(type, "average")) return STATSD_APP_CHART_DIM_VALUE_TYPE_AVERAGE; else if(!strcmp(type, "median")) return STATSD_APP_CHART_DIM_VALUE_TYPE_MEDIAN; else if(!strcmp(type, "stddev")) return STATSD_APP_CHART_DIM_VALUE_TYPE_STDDEV; else if(!strcmp(type, "percentile")) return STATSD_APP_CHART_DIM_VALUE_TYPE_PERCENTILE; error("STATSD: invalid type '%s' at line %zu of file '%s'. Using 'last'.", type, line, filename); return STATSD_APP_CHART_DIM_VALUE_TYPE_LAST; } static const char *valuetype2string(STATSD_APP_CHART_DIM_VALUE_TYPE type) { switch(type) { case STATSD_APP_CHART_DIM_VALUE_TYPE_EVENTS: return "events"; case STATSD_APP_CHART_DIM_VALUE_TYPE_LAST: return "last"; case STATSD_APP_CHART_DIM_VALUE_TYPE_MIN: return "min"; case STATSD_APP_CHART_DIM_VALUE_TYPE_MAX: return "max"; case STATSD_APP_CHART_DIM_VALUE_TYPE_SUM: return "sum"; case STATSD_APP_CHART_DIM_VALUE_TYPE_AVERAGE: return "average"; case STATSD_APP_CHART_DIM_VALUE_TYPE_MEDIAN: return "median"; case STATSD_APP_CHART_DIM_VALUE_TYPE_STDDEV: return "stddev"; case STATSD_APP_CHART_DIM_VALUE_TYPE_PERCENTILE: return "percentile"; } return "unknown"; } static STATSD_APP_CHART_DIM *add_dimension_to_app_chart( STATSD_APP *app __maybe_unused , STATSD_APP_CHART *chart , const char *metric_name , const char *dim_name , collected_number multiplier , collected_number divisor , RRDDIM_FLAGS flags , STATSD_APP_CHART_DIM_VALUE_TYPE value_type ) { STATSD_APP_CHART_DIM *dim = callocz(sizeof(STATSD_APP_CHART_DIM), 1); dim->metric = strdupz(metric_name); dim->metric_hash = simple_hash(dim->metric); dim->name = strdupz((dim_name)?dim_name:""); dim->multiplier = multiplier; dim->divisor = divisor; dim->value_type = value_type; dim->flags = flags; if(!dim->multiplier) dim->multiplier = 1; if(!dim->divisor) dim->divisor = 1; // append it to the list of dimension STATSD_APP_CHART_DIM *tdim; for(tdim = chart->dimensions; tdim && tdim->next ; tdim = tdim->next) ; if(!tdim) { dim->next = chart->dimensions; chart->dimensions = dim; } else { dim->next = tdim->next; tdim->next = dim; } chart->dimensions_count++; debug(D_STATSD, "Added dimension '%s' to chart '%s' of app '%s', for metric '%s', with type %u, multiplier " COLLECTED_NUMBER_FORMAT ", divisor " COLLECTED_NUMBER_FORMAT, dim->name, chart->id, app->name, dim->metric, dim->value_type, dim->multiplier, dim->divisor); return dim; } static int statsd_readfile(const char *filename, STATSD_APP *app, STATSD_APP_CHART *chart, DICTIONARY *dict) { debug(D_STATSD, "STATSD configuration reading file '%s'", filename); char *buffer = mallocz(STATSD_CONF_LINE_MAX + 1); FILE *fp = fopen(filename, "r"); if(!fp) { error("STATSD: cannot open file '%s'.", filename); freez(buffer); return -1; } size_t line = 0; char *s; while(fgets(buffer, STATSD_CONF_LINE_MAX, fp) != NULL) { buffer[STATSD_CONF_LINE_MAX] = '\0'; line++; s = trim(buffer); if (!s || *s == '#') { debug(D_STATSD, "STATSD: ignoring line %zu of file '%s', it is empty.", line, filename); continue; } debug(D_STATSD, "STATSD: processing line %zu of file '%s': %s", line, filename, buffer); if(*s == 'i' && strncmp(s, "include", 7) == 0) { s = trim(&s[7]); if(s && *s) { char *tmp; if(*s == '/') tmp = strdupz(s); else { // the file to be included is relative to current file // find the directory name from the file we already read char *filename2 = strdupz(filename); // copy filename, since dirname() will change it char *dir = dirname(filename2); // find the directory part of the filename tmp = strdupz_path_subpath(dir, s); // compose the new filename to read; freez(filename2); // free the filename we copied } statsd_readfile(tmp, app, chart, dict); freez(tmp); } else error("STATSD: ignoring line %zu of file '%s', include filename is empty", line, filename); continue; } int len = (int) strlen(s); if (*s == '[' && s[len - 1] == ']') { // new section s[len - 1] = '\0'; s++; if (!strcmp(s, "app")) { // a new app app = callocz(sizeof(STATSD_APP), 1); app->name = strdupz("unnamed"); app->rrd_memory_mode = localhost->rrd_memory_mode; app->rrd_history_entries = localhost->rrd_history_entries; app->next = statsd.apps; statsd.apps = app; chart = NULL; dict = NULL; { char lineandfile[FILENAME_MAX + 1]; snprintfz(lineandfile, FILENAME_MAX, "%zu@%s", line, filename); app->source = strdupz(lineandfile); } } else if(app) { if(!strcmp(s, "dictionary")) { if(!app->dict) app->dict = dictionary_create(DICTIONARY_FLAG_SINGLE_THREADED); dict = app->dict; } else { dict = NULL; // a new chart chart = callocz(sizeof(STATSD_APP_CHART), 1); netdata_fix_chart_id(s); chart->id = strdupz(s); chart->name = strdupz(s); chart->title = strdupz("Statsd chart"); chart->context = strdupz(s); chart->family = strdupz("overview"); chart->units = strdupz("value"); chart->priority = NETDATA_CHART_PRIO_STATSD_PRIVATE; chart->chart_type = RRDSET_TYPE_LINE; chart->next = app->charts; app->charts = chart; if (!strncmp( filename, netdata_configured_stock_config_dir, strlen(netdata_configured_stock_config_dir))) { char tmpfilename[FILENAME_MAX + 1]; strncpyz(tmpfilename, filename, FILENAME_MAX); chart->module = strdupz(basename(tmpfilename)); } else { chart->module = strdupz("synthetic_chart"); } } } else error("STATSD: ignoring line %zu ('%s') of file '%s', [app] is not defined.", line, s, filename); continue; } if(!app) { error("STATSD: ignoring line %zu ('%s') of file '%s', it is outside all sections.", line, s, filename); continue; } char *name = s; char *value = strchr(s, '='); if(!value) { error("STATSD: ignoring line %zu ('%s') of file '%s', there is no = in it.", line, s, filename); continue; } *value = '\0'; value++; name = trim(name); value = trim(value); if(!name || *name == '#') { error("STATSD: ignoring line %zu of file '%s', name is empty.", line, filename); continue; } if(!value) { debug(D_CONFIG, "STATSD: ignoring line %zu of file '%s', value is empty.", line, filename); continue; } if(unlikely(dict)) { // parse [dictionary] members dictionary_set(dict, name, value, strlen(value) + 1); } else if(!chart) { // parse [app] members if(!strcmp(name, "name")) { freez((void *)app->name); netdata_fix_chart_name(value); app->name = strdupz(value); } else if (!strcmp(name, "metrics")) { simple_pattern_free(app->metrics); app->metrics = simple_pattern_create(value, NULL, SIMPLE_PATTERN_EXACT); } else if (!strcmp(name, "private charts")) { if (!strcmp(value, "yes") || !strcmp(value, "on")) app->default_options |= STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; else app->default_options &= ~STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; } else if (!strcmp(name, "gaps when not collected")) { if (!strcmp(value, "yes") || !strcmp(value, "on")) app->default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; } else if (!strcmp(name, "memory mode")) { app->rrd_memory_mode = rrd_memory_mode_id(value); } else if (!strcmp(name, "history")) { app->rrd_history_entries = atol(value); if (app->rrd_history_entries < 5) app->rrd_history_entries = 5; } else { error("STATSD: ignoring line %zu ('%s') of file '%s'. Unknown keyword for the [app] section.", line, name, filename); continue; } } else { // parse [chart] members if(!strcmp(name, "name")) { freez((void *)chart->name); netdata_fix_chart_id(value); chart->name = strdupz(value); } else if(!strcmp(name, "title")) { freez((void *)chart->title); chart->title = strdupz(value); } else if (!strcmp(name, "family")) { freez((void *)chart->family); chart->family = strdupz(value); } else if (!strcmp(name, "context")) { freez((void *)chart->context); netdata_fix_chart_id(value); chart->context = strdupz(value); } else if (!strcmp(name, "units")) { freez((void *)chart->units); chart->units = strdupz(value); } else if (!strcmp(name, "priority")) { chart->priority = atol(value); } else if (!strcmp(name, "type")) { chart->chart_type = rrdset_type_id(value); } else if (!strcmp(name, "dimension")) { // metric [name [type [multiplier [divisor]]]] char *words[10]; pluginsd_split_words(value, words, 10, NULL, NULL, 0); int pattern = 0; size_t i = 0; char *metric_name = words[i++]; if(strcmp(metric_name, "pattern") == 0) { metric_name = words[i++]; pattern = 1; } char *dim_name = words[i++]; char *type = words[i++]; char *multiplier = words[i++]; char *divisor = words[i++]; char *options = words[i++]; RRDDIM_FLAGS flags = RRDDIM_FLAG_NONE; if(options && *options) { if(strstr(options, "hidden") != NULL) flags |= RRDDIM_FLAG_HIDDEN; if(strstr(options, "noreset") != NULL) flags |= RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS; if(strstr(options, "nooverflow") != NULL) flags |= RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS; } if(!pattern) { if(app->dict) { if(dim_name && *dim_name) { char *n = dictionary_get(app->dict, dim_name); if(n) dim_name = n; } else { dim_name = dictionary_get(app->dict, metric_name); } } if(!dim_name || !*dim_name) dim_name = metric_name; } STATSD_APP_CHART_DIM *dim = add_dimension_to_app_chart( app , chart , metric_name , dim_name , (multiplier && *multiplier)?str2l(multiplier):1 , (divisor && *divisor)?str2l(divisor):1 , flags , string2valuetype(type, line, filename) ); if(pattern) dim->metric_pattern = simple_pattern_create(dim->metric, NULL, SIMPLE_PATTERN_EXACT); } else { error("STATSD: ignoring line %zu ('%s') of file '%s'. Unknown keyword for the [%s] section.", line, name, filename, chart->id); continue; } } } freez(buffer); fclose(fp); return 0; } static int statsd_file_callback(const char *filename, void *data) { (void)data; return statsd_readfile(filename, NULL, NULL, NULL); } static inline void statsd_readdir(const char *user_path, const char *stock_path, const char *subpath) { recursive_config_double_dir_load(user_path, stock_path, subpath, statsd_file_callback, NULL, 0); } // -------------------------------------------------------------------------------------------------------------------- // send metrics to netdata - in private charts - called from the main thread // extract chart type and chart id from metric name static inline void statsd_get_metric_type_and_id(STATSD_METRIC *m, char *type, char *id, char *context, const char *metrictype, size_t len) { // The full chart type.id looks like this: // ${STATSD_CHART_PREFIX} + "_" + ${METRIC_NAME} + "_" + ${METRIC_TYPE} // // where: // STATSD_CHART_PREFIX = "statsd" as defined above // METRIC_NAME = whatever the user gave to statsd // METRIC_TYPE = "gauge", "counter", "meter", "timer", "histogram", "set", "dictionary" // for chart type, we want: // ${STATSD_CHART_PREFIX} + "_" + the first word of ${METRIC_NAME} // find the first word of ${METRIC_NAME} char firstword[len + 1], *s = ""; strncpyz(firstword, m->name, len); for (s = firstword; *s ; s++) { if (unlikely(*s == '.' || *s == '_')) { *s = '\0'; s++; break; } } // firstword has the first word of ${METRIC_NAME} // s has the remaining, if any // create the chart type: snprintfz(type, len, STATSD_CHART_PREFIX "_%s", firstword); // for chart id, we want: // the remaining of the words of ${METRIC_NAME} + "_" + ${METRIC_TYPE} // or the ${METRIC_NAME} has no remaining words, the ${METRIC_TYPE} alone if(*s) snprintfz(id, len, "%s_%s", s, metrictype); else snprintfz(id, len, "%s", metrictype); // for the context, we want the full of both the above, separated with a dot (type.id): snprintfz(context, RRD_ID_LENGTH_MAX, "%s.%s", type, id); // make sure they don't have illegal characters netdata_fix_chart_id(type); netdata_fix_chart_id(id); netdata_fix_chart_id(context); } static inline RRDSET *statsd_private_rrdset_create( STATSD_METRIC *m __maybe_unused , const char *type , const char *id , const char *name , const char *family , const char *context , const char *title , const char *units , long priority , int update_every , RRDSET_TYPE chart_type ) { statsd.private_charts++; RRDSET *st = rrdset_create_custom( localhost // host , type // type , id // id , name // name , family // family , context // context , title // title , units // units , PLUGIN_STATSD_NAME // plugin , "private_chart" // module , priority // priority , update_every // update every , chart_type // chart type , default_rrd_memory_mode // memory mode , default_rrd_history_entries // history ); rrdset_flag_set(st, RRDSET_FLAG_STORE_FIRST); if(statsd.private_charts_hidden) rrdset_flag_set(st, RRDSET_FLAG_HIDDEN); // rrdset_flag_set(st, RRDSET_FLAG_DEBUG); return st; } static inline void statsd_private_chart_gauge(STATSD_METRIC *m) { debug(D_STATSD, "updating private chart for gauge metric '%s'", m->name); if(unlikely(!m->st)) { char type[RRD_ID_LENGTH_MAX + 1], id[RRD_ID_LENGTH_MAX + 1], context[RRD_ID_LENGTH_MAX + 1]; statsd_get_metric_type_and_id(m, type, id, context, "gauge", RRD_ID_LENGTH_MAX); char title[RRD_ID_LENGTH_MAX + 1]; snprintfz(title, RRD_ID_LENGTH_MAX, "statsd private chart for gauge %s", m->name); m->st = statsd_private_rrdset_create( m , type , id , NULL // name , m->family?m->family:"gauges" // family (submenu) , context // context , title // title , m->units?m->units:"value" // units , NETDATA_CHART_PRIO_STATSD_PRIVATE , statsd.update_every , RRDSET_TYPE_LINE ); m->rd_value = rrddim_add(m->st, "gauge", m->dimname?m->dimname:NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); if(m->options & STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT) m->rd_count = rrddim_add(m->st, "events", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); } else rrdset_next(m->st); rrddim_set_by_pointer(m->st, m->rd_value, m->last); if(m->rd_count) rrddim_set_by_pointer(m->st, m->rd_count, m->events); rrdset_done(m->st); } static inline void statsd_private_chart_counter_or_meter(STATSD_METRIC *m, const char *dim, const char *family) { debug(D_STATSD, "updating private chart for %s metric '%s'", dim, m->name); if(unlikely(!m->st)) { char type[RRD_ID_LENGTH_MAX + 1], id[RRD_ID_LENGTH_MAX + 1], context[RRD_ID_LENGTH_MAX + 1]; statsd_get_metric_type_and_id(m, type, id, context, dim, RRD_ID_LENGTH_MAX); char title[RRD_ID_LENGTH_MAX + 1]; snprintfz(title, RRD_ID_LENGTH_MAX, "statsd private chart for %s %s", dim, m->name); m->st = statsd_private_rrdset_create( m , type , id , NULL // name , m->family?m->family:family // family (submenu) , context // context , title // title , m->units?m->units:"events/s" // units , NETDATA_CHART_PRIO_STATSD_PRIVATE , statsd.update_every , RRDSET_TYPE_AREA ); m->rd_value = rrddim_add(m->st, dim, m->dimname?m->dimname:NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); if(m->options & STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT) m->rd_count = rrddim_add(m->st, "events", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); } else rrdset_next(m->st); rrddim_set_by_pointer(m->st, m->rd_value, m->last); if(m->rd_count) rrddim_set_by_pointer(m->st, m->rd_count, m->events); rrdset_done(m->st); } static inline void statsd_private_chart_set(STATSD_METRIC *m) { debug(D_STATSD, "updating private chart for set metric '%s'", m->name); if(unlikely(!m->st)) { char type[RRD_ID_LENGTH_MAX + 1], id[RRD_ID_LENGTH_MAX + 1], context[RRD_ID_LENGTH_MAX + 1]; statsd_get_metric_type_and_id(m, type, id, context, "set", RRD_ID_LENGTH_MAX); char title[RRD_ID_LENGTH_MAX + 1]; snprintfz(title, RRD_ID_LENGTH_MAX, "statsd private chart for set %s", m->name); m->st = statsd_private_rrdset_create( m , type , id , NULL // name , m->family?m->family:"sets" // family (submenu) , context // context , title // title , m->units?m->units:"entries" // units , NETDATA_CHART_PRIO_STATSD_PRIVATE , statsd.update_every , RRDSET_TYPE_LINE ); m->rd_value = rrddim_add(m->st, "set", m->dimname?m->dimname:"unique", 1, 1, RRD_ALGORITHM_ABSOLUTE); if(m->options & STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT) m->rd_count = rrddim_add(m->st, "events", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); } else rrdset_next(m->st); rrddim_set_by_pointer(m->st, m->rd_value, m->last); if(m->rd_count) rrddim_set_by_pointer(m->st, m->rd_count, m->events); rrdset_done(m->st); } static inline void statsd_private_chart_dictionary(STATSD_METRIC *m) { debug(D_STATSD, "updating private chart for dictionary metric '%s'", m->name); if(unlikely(!m->st)) { char type[RRD_ID_LENGTH_MAX + 1], id[RRD_ID_LENGTH_MAX + 1], context[RRD_ID_LENGTH_MAX + 1]; statsd_get_metric_type_and_id(m, type, id, context, "dictionary", RRD_ID_LENGTH_MAX); char title[RRD_ID_LENGTH_MAX + 1]; snprintfz(title, RRD_ID_LENGTH_MAX, "statsd private chart for dictionary %s", m->name); m->st = statsd_private_rrdset_create( m , type , id , NULL // name , m->family?m->family:"dictionaries" // family (submenu) , context // context , title // title , m->units?m->units:"events/s" // units , NETDATA_CHART_PRIO_STATSD_PRIVATE , statsd.update_every , RRDSET_TYPE_STACKED ); if(m->options & STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT) m->rd_count = rrddim_add(m->st, "events", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); } else rrdset_next(m->st); STATSD_METRIC_DICTIONARY_ITEM *t; dfe_start_read(m->dictionary.dict, t) { if (!t->rd) t->rd = rrddim_add(m->st, t_name, NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); rrddim_set_by_pointer(m->st, t->rd, (collected_number)t->count); } dfe_done(t); if(m->rd_count) rrddim_set_by_pointer(m->st, m->rd_count, m->events); rrdset_done(m->st); } static inline void statsd_private_chart_timer_or_histogram(STATSD_METRIC *m, const char *dim, const char *family, const char *units) { debug(D_STATSD, "updating private chart for %s metric '%s'", dim, m->name); if(unlikely(!m->st)) { char type[RRD_ID_LENGTH_MAX + 1], id[RRD_ID_LENGTH_MAX + 1], context[RRD_ID_LENGTH_MAX + 1]; statsd_get_metric_type_and_id(m, type, id, context, dim, RRD_ID_LENGTH_MAX); char title[RRD_ID_LENGTH_MAX + 1]; snprintfz(title, RRD_ID_LENGTH_MAX, "statsd private chart for %s %s", dim, m->name); m->st = statsd_private_rrdset_create( m , type , id , NULL // name , m->family?m->family:family // family (submenu) , context // context , title // title , m->units?m->units:units // units , NETDATA_CHART_PRIO_STATSD_PRIVATE , statsd.update_every , RRDSET_TYPE_AREA ); m->histogram.ext->rd_min = rrddim_add(m->st, "min", NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); m->histogram.ext->rd_max = rrddim_add(m->st, "max", NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); m->rd_value = rrddim_add(m->st, "average", NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); m->histogram.ext->rd_percentile = rrddim_add(m->st, statsd.histogram_percentile_str, NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); m->histogram.ext->rd_median = rrddim_add(m->st, "median", NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); m->histogram.ext->rd_stddev = rrddim_add(m->st, "stddev", NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); //m->histogram.ext->rd_sum = rrddim_add(m->st, "sum", NULL, 1, statsd.decimal_detail, RRD_ALGORITHM_ABSOLUTE); if(m->options & STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT) m->rd_count = rrddim_add(m->st, "events", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); } else rrdset_next(m->st); rrddim_set_by_pointer(m->st, m->histogram.ext->rd_min, m->histogram.ext->last_min); rrddim_set_by_pointer(m->st, m->histogram.ext->rd_max, m->histogram.ext->last_max); rrddim_set_by_pointer(m->st, m->histogram.ext->rd_percentile, m->histogram.ext->last_percentile); rrddim_set_by_pointer(m->st, m->histogram.ext->rd_median, m->histogram.ext->last_median); rrddim_set_by_pointer(m->st, m->histogram.ext->rd_stddev, m->histogram.ext->last_stddev); //rrddim_set_by_pointer(m->st, m->histogram.ext->rd_sum, m->histogram.ext->last_sum); rrddim_set_by_pointer(m->st, m->rd_value, m->last); if(m->rd_count) rrddim_set_by_pointer(m->st, m->rd_count, m->events); rrdset_done(m->st); } // -------------------------------------------------------------------------------------------------------------------- // statsd flush metrics static inline void statsd_flush_gauge(STATSD_METRIC *m) { debug(D_STATSD, "flushing gauge metric '%s'", m->name); int updated = 0; if(unlikely(!m->reset && m->count)) { m->last = (collected_number) (m->gauge.value * statsd.decimal_detail); m->reset = 1; updated = 1; } if(unlikely(m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED && (updated || !(m->options & STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED)))) statsd_private_chart_gauge(m); } static inline void statsd_flush_counter_or_meter(STATSD_METRIC *m, const char *dim, const char *family) { debug(D_STATSD, "flushing %s metric '%s'", dim, m->name); int updated = 0; if(unlikely(!m->reset && m->count)) { m->last = m->counter.value; m->reset = 1; updated = 1; } if(unlikely(m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED && (updated || !(m->options & STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED)))) statsd_private_chart_counter_or_meter(m, dim, family); } static inline void statsd_flush_counter(STATSD_METRIC *m) { statsd_flush_counter_or_meter(m, "counter", "counters"); } static inline void statsd_flush_meter(STATSD_METRIC *m) { statsd_flush_counter_or_meter(m, "meter", "meters"); } static inline void statsd_flush_set(STATSD_METRIC *m) { debug(D_STATSD, "flushing set metric '%s'", m->name); int updated = 0; if(unlikely(!m->reset && m->count)) { m->last = (collected_number)m->set.unique; m->reset = 1; updated = 1; } else { m->last = 0; } if(unlikely(m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED && (updated || !(m->options & STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED)))) statsd_private_chart_set(m); } static inline void statsd_flush_dictionary(STATSD_METRIC *m) { debug(D_STATSD, "flushing dictionary metric '%s'", m->name); int updated = 0; if(unlikely(!m->reset && m->count)) { m->last = (collected_number)m->dictionary.unique; m->reset = 1; updated = 1; } else { m->last = 0; } if(unlikely(m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED && (updated || !(m->options & STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED)))) statsd_private_chart_dictionary(m); if(m->dictionary.unique >= statsd.dictionary_max_unique) { if(!(m->options & STATSD_METRIC_OPTION_COLLECTION_FULL_LOGGED)) { m->options |= STATSD_METRIC_OPTION_COLLECTION_FULL_LOGGED; info( "STATSD dictionary '%s' reach max of %zu items - try increasing 'dictionaries max unique dimensions' in netdata.conf", m->name, m->dictionary.unique); } } } static inline void statsd_flush_timer_or_histogram(STATSD_METRIC *m, const char *dim, const char *family, const char *units) { debug(D_STATSD, "flushing %s metric '%s'", dim, m->name); int updated = 0; if(unlikely(!m->reset && m->count && m->histogram.ext->used > 0)) { netdata_mutex_lock(&m->histogram.ext->mutex); size_t len = m->histogram.ext->used; NETDATA_DOUBLE *series = m->histogram.ext->values; sort_series(series, len); m->histogram.ext->last_min = (collected_number)roundndd(series[0] * statsd.decimal_detail); m->histogram.ext->last_max = (collected_number)roundndd(series[len - 1] * statsd.decimal_detail); m->last = (collected_number)roundndd(average(series, len) * statsd.decimal_detail); m->histogram.ext->last_median = (collected_number)roundndd(median_on_sorted_series(series, len) * statsd.decimal_detail); m->histogram.ext->last_stddev = (collected_number)roundndd(standard_deviation(series, len) * statsd.decimal_detail); m->histogram.ext->last_sum = (collected_number)roundndd(sum(series, len) * statsd.decimal_detail); size_t pct_len = (size_t)floor((double)len * statsd.histogram_percentile / 100.0); if(pct_len < 1) m->histogram.ext->last_percentile = (collected_number)(series[0] * statsd.decimal_detail); else m->histogram.ext->last_percentile = (collected_number)roundndd(series[pct_len - 1] * statsd.decimal_detail); netdata_mutex_unlock(&m->histogram.ext->mutex); debug(D_STATSD, "STATSD %s metric %s: min " COLLECTED_NUMBER_FORMAT ", max " COLLECTED_NUMBER_FORMAT ", last " COLLECTED_NUMBER_FORMAT ", pcent " COLLECTED_NUMBER_FORMAT ", median " COLLECTED_NUMBER_FORMAT ", stddev " COLLECTED_NUMBER_FORMAT ", sum " COLLECTED_NUMBER_FORMAT, dim, m->name, m->histogram.ext->last_min, m->histogram.ext->last_max, m->last, m->histogram.ext->last_percentile, m->histogram.ext->last_median, m->histogram.ext->last_stddev, m->histogram.ext->last_sum); m->histogram.ext->zeroed = 0; m->reset = 1; updated = 1; } else if(unlikely(!m->histogram.ext->zeroed)) { // reset the metrics // if we collected anything, they will be updated below // this ensures that we report zeros if nothing is collected m->histogram.ext->last_min = 0; m->histogram.ext->last_max = 0; m->last = 0; m->histogram.ext->last_median = 0; m->histogram.ext->last_stddev = 0; m->histogram.ext->last_sum = 0; m->histogram.ext->last_percentile = 0; m->histogram.ext->zeroed = 1; } if(unlikely(m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED && (updated || !(m->options & STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED)))) statsd_private_chart_timer_or_histogram(m, dim, family, units); } static inline void statsd_flush_timer(STATSD_METRIC *m) { statsd_flush_timer_or_histogram(m, "timer", "timers", "milliseconds"); } static inline void statsd_flush_histogram(STATSD_METRIC *m) { statsd_flush_timer_or_histogram(m, "histogram", "histograms", "value"); } static inline RRD_ALGORITHM statsd_algorithm_for_metric(STATSD_METRIC *m) { switch(m->type) { default: case STATSD_METRIC_TYPE_GAUGE: case STATSD_METRIC_TYPE_SET: case STATSD_METRIC_TYPE_TIMER: case STATSD_METRIC_TYPE_HISTOGRAM: return RRD_ALGORITHM_ABSOLUTE; case STATSD_METRIC_TYPE_METER: case STATSD_METRIC_TYPE_COUNTER: case STATSD_METRIC_TYPE_DICTIONARY: return RRD_ALGORITHM_INCREMENTAL; } } static inline void link_metric_to_app_dimension(STATSD_APP *app, STATSD_METRIC *m, STATSD_APP_CHART *chart, STATSD_APP_CHART_DIM *dim) { if(dim->value_type == STATSD_APP_CHART_DIM_VALUE_TYPE_EVENTS) { dim->value_ptr = &m->events; dim->algorithm = RRD_ALGORITHM_INCREMENTAL; } else if(m->type == STATSD_METRIC_TYPE_HISTOGRAM || m->type == STATSD_METRIC_TYPE_TIMER) { dim->algorithm = RRD_ALGORITHM_ABSOLUTE; dim->divisor *= statsd.decimal_detail; switch(dim->value_type) { case STATSD_APP_CHART_DIM_VALUE_TYPE_EVENTS: // will never match - added to avoid warning break; case STATSD_APP_CHART_DIM_VALUE_TYPE_LAST: case STATSD_APP_CHART_DIM_VALUE_TYPE_AVERAGE: dim->value_ptr = &m->last; break; case STATSD_APP_CHART_DIM_VALUE_TYPE_SUM: dim->value_ptr = &m->histogram.ext->last_sum; break; case STATSD_APP_CHART_DIM_VALUE_TYPE_MIN: dim->value_ptr = &m->histogram.ext->last_min; break; case STATSD_APP_CHART_DIM_VALUE_TYPE_MAX: dim->value_ptr = &m->histogram.ext->last_max; break; case STATSD_APP_CHART_DIM_VALUE_TYPE_MEDIAN: dim->value_ptr = &m->histogram.ext->last_median; break; case STATSD_APP_CHART_DIM_VALUE_TYPE_PERCENTILE: dim->value_ptr = &m->histogram.ext->last_percentile; break; case STATSD_APP_CHART_DIM_VALUE_TYPE_STDDEV: dim->value_ptr = &m->histogram.ext->last_stddev; break; } } else { if (dim->value_type != STATSD_APP_CHART_DIM_VALUE_TYPE_LAST) error("STATSD: unsupported value type for dimension '%s' of chart '%s' of app '%s' on metric '%s'", dim->name, chart->id, app->name, m->name); dim->value_ptr = &m->last; dim->algorithm = statsd_algorithm_for_metric(m); if(m->type == STATSD_METRIC_TYPE_GAUGE) dim->divisor *= statsd.decimal_detail; } if(unlikely(chart->st && dim->rd)) { rrddim_set_algorithm(chart->st, dim->rd, dim->algorithm); rrddim_set_multiplier(chart->st, dim->rd, dim->multiplier); rrddim_set_divisor(chart->st, dim->rd, dim->divisor); } chart->dimensions_linked_count++; m->options |= STATSD_METRIC_OPTION_USED_IN_APPS; debug(D_STATSD, "metric '%s' of type %u linked with app '%s', chart '%s', dimension '%s', algorithm '%s'", m->name, m->type, app->name, chart->id, dim->name, rrd_algorithm_name(dim->algorithm)); } static inline void check_if_metric_is_for_app(STATSD_INDEX *index, STATSD_METRIC *m) { (void)index; STATSD_APP *app; for(app = statsd.apps; app ;app = app->next) { if(unlikely(simple_pattern_matches(app->metrics, m->name))) { debug(D_STATSD, "metric '%s' matches app '%s'", m->name, app->name); // the metric should get the options from the app if(app->default_options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED) m->options |= STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; else m->options &= ~STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; if(app->default_options & STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED) m->options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; else m->options &= ~STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; m->options |= STATSD_METRIC_OPTION_PRIVATE_CHART_CHECKED; // check if there is a chart in this app, willing to get this metric STATSD_APP_CHART *chart; for(chart = app->charts; chart; chart = chart->next) { STATSD_APP_CHART_DIM *dim; for(dim = chart->dimensions; dim ; dim = dim->next) { if(unlikely(dim->metric_pattern)) { size_t dim_name_len = strlen(dim->name); size_t wildcarded_len = dim_name_len + strlen(m->name) + 1; char wildcarded[wildcarded_len]; strcpy(wildcarded, dim->name); char *ws = &wildcarded[dim_name_len]; if(simple_pattern_matches_extract(dim->metric_pattern, m->name, ws, wildcarded_len - dim_name_len)) { char *final_name = NULL; if(app->dict) { if(likely(*wildcarded)) { // use the name of the wildcarded string final_name = dictionary_get(app->dict, wildcarded); } if(unlikely(!final_name)) { // use the name of the metric final_name = dictionary_get(app->dict, m->name); } } if(unlikely(!final_name)) final_name = wildcarded; add_dimension_to_app_chart( app , chart , m->name , final_name , dim->multiplier , dim->divisor , dim->flags , dim->value_type ); // the new dimension is appended to the list // so, it will be matched and linked later too } } else if(!dim->value_ptr && dim->metric_hash == m->hash && !strcmp(dim->metric, m->name)) { // we have a match - this metric should be linked to this dimension link_metric_to_app_dimension(app, m, chart, dim); } } } } } } static inline RRDDIM *statsd_add_dim_to_app_chart(STATSD_APP *app, STATSD_APP_CHART *chart, STATSD_APP_CHART_DIM *dim) { (void)app; // allow the same statsd metric to be added multiple times to the same chart STATSD_APP_CHART_DIM *tdim; size_t count_same_metric = 0, count_same_metric_value_type = 0; size_t pos_same_metric_value_type = 0; for (tdim = chart->dimensions; tdim && tdim->next; tdim = tdim->next) { if (dim->metric_hash == tdim->metric_hash && !strcmp(dim->metric, tdim->metric)) { count_same_metric++; if(dim->value_type == tdim->value_type) { count_same_metric_value_type++; if (tdim == dim) pos_same_metric_value_type = count_same_metric_value_type; } } } if(count_same_metric > 1) { // the same metric is found multiple times size_t len = strlen(dim->metric) + 100; char metric[ len + 1 ]; if(count_same_metric_value_type > 1) { // the same metric, with the same value type, is added multiple times snprintfz(metric, len, "%s_%s%zu", dim->metric, valuetype2string(dim->value_type), pos_same_metric_value_type); } else { // the same metric, with different value type is added snprintfz(metric, len, "%s_%s", dim->metric, valuetype2string(dim->value_type)); } dim->rd = rrddim_add(chart->st, metric, dim->name, dim->multiplier, dim->divisor, dim->algorithm); if(dim->flags != RRDDIM_FLAG_NONE) dim->rd->flags |= dim->flags; return dim->rd; } dim->rd = rrddim_add(chart->st, dim->metric, dim->name, dim->multiplier, dim->divisor, dim->algorithm); if(dim->flags != RRDDIM_FLAG_NONE) dim->rd->flags |= dim->flags; return dim->rd; } static inline void statsd_update_app_chart(STATSD_APP *app, STATSD_APP_CHART *chart) { debug(D_STATSD, "updating chart '%s' for app '%s'", chart->id, app->name); if(!chart->st) { chart->st = rrdset_create_custom( localhost // host , app->name // type , chart->id // id , chart->name // name , chart->family // family , chart->context // context , chart->title // title , chart->units // units , PLUGIN_STATSD_NAME // plugin , chart->module // module , chart->priority // priority , statsd.update_every // update every , chart->chart_type // chart type , app->rrd_memory_mode // memory mode , app->rrd_history_entries // history ); rrdset_flag_set(chart->st, RRDSET_FLAG_STORE_FIRST); // rrdset_flag_set(chart->st, RRDSET_FLAG_DEBUG); } else rrdset_next(chart->st); STATSD_APP_CHART_DIM *dim; for(dim = chart->dimensions; dim ;dim = dim->next) { if(likely(!dim->metric_pattern)) { if (unlikely(!dim->rd)) statsd_add_dim_to_app_chart(app, chart, dim); if (unlikely(dim->value_ptr)) { debug(D_STATSD, "updating dimension '%s' (%s) of chart '%s' (%s) for app '%s' with value " COLLECTED_NUMBER_FORMAT, dim->name, dim->rd->id, chart->id, chart->st->id, app->name, *dim->value_ptr); rrddim_set_by_pointer(chart->st, dim->rd, *dim->value_ptr); } } } rrdset_done(chart->st); debug(D_STATSD, "completed update of chart '%s' for app '%s'", chart->id, app->name); } static inline void statsd_update_all_app_charts(void) { // debug(D_STATSD, "updating app charts"); STATSD_APP *app; for(app = statsd.apps; app ;app = app->next) { // debug(D_STATSD, "updating charts for app '%s'", app->name); STATSD_APP_CHART *chart; for(chart = app->charts; chart ;chart = chart->next) { if(unlikely(chart->dimensions_linked_count)) { statsd_update_app_chart(app, chart); } } } // debug(D_STATSD, "completed update of app charts"); } const char *statsd_metric_type_string(STATSD_METRIC_TYPE type) { switch(type) { case STATSD_METRIC_TYPE_COUNTER: return "counter"; case STATSD_METRIC_TYPE_GAUGE: return "gauge"; case STATSD_METRIC_TYPE_HISTOGRAM: return "histogram"; case STATSD_METRIC_TYPE_METER: return "meter"; case STATSD_METRIC_TYPE_SET: return "set"; case STATSD_METRIC_TYPE_DICTIONARY: return "dictionary"; case STATSD_METRIC_TYPE_TIMER: return "timer"; default: return "unknown"; } } static inline void statsd_flush_index_metrics(STATSD_INDEX *index, void (*flush_metric)(STATSD_METRIC *)) { STATSD_METRIC *m; // find the useful metrics (incremental = each time we are called, we check the new metrics only) dfe_start_read(index->dict, m) { // since we add new metrics at the beginning // check for useful charts, until the point we last checked if(unlikely(is_metric_checked(m))) break; if(unlikely(!(m->options & STATSD_METRIC_OPTION_CHECKED_IN_APPS))) { log_access("NEW STATSD METRIC '%s': '%s'", statsd_metric_type_string(m->type), m->name); check_if_metric_is_for_app(index, m); m->options |= STATSD_METRIC_OPTION_CHECKED_IN_APPS; } if(unlikely(!(m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_CHECKED))) { if(unlikely(statsd.private_charts >= statsd.max_private_charts_hard)) { debug(D_STATSD, "STATSD: metric '%s' will not be charted, because the hard limit of the maximum number of charts has been reached.", m->name); info("STATSD: metric '%s' will not be charted, because the hard limit of the maximum number of charts (%zu) has been reached. Increase the number of charts by editing netdata.conf, [statsd] section.", m->name, statsd.max_private_charts_hard); m->options &= ~STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; } else { if (simple_pattern_matches(statsd.charts_for, m->name)) { debug(D_STATSD, "STATSD: metric '%s' will be charted.", m->name); m->options |= STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; } else { debug(D_STATSD, "STATSD: metric '%s' will not be charted.", m->name); m->options &= ~STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED; } } m->options |= STATSD_METRIC_OPTION_PRIVATE_CHART_CHECKED; } // mark it as checked m->options |= STATSD_METRIC_OPTION_CHECKED; // check if it is used in charts if((m->options & (STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED|STATSD_METRIC_OPTION_USED_IN_APPS)) && !(m->options & STATSD_METRIC_OPTION_USEFUL)) { m->options |= STATSD_METRIC_OPTION_USEFUL; index->useful++; m->next_useful = index->first_useful; index->first_useful = m; } } dfe_done(m); // flush all the useful metrics for(m = index->first_useful; m ; m = m->next_useful) { flush_metric(m); } } // -------------------------------------------------------------------------------------- // statsd main thread static int statsd_listen_sockets_setup(void) { return listen_sockets_setup(&statsd.sockets); } static void statsd_main_cleanup(void *data) { struct netdata_static_thread *static_thread = (struct netdata_static_thread *)data; static_thread->enabled = NETDATA_MAIN_THREAD_EXITING; info("cleaning up..."); if (statsd.collection_threads_status) { int i; for (i = 0; i < statsd.threads; i++) { if(statsd.collection_threads_status[i].status) { info("STATSD: stopping data collection thread %d...", i + 1); netdata_thread_cancel(statsd.collection_threads_status[i].thread); } else { info("STATSD: data collection thread %d found stopped.", i + 1); } } } info("STATSD: closing sockets..."); listen_sockets_close(&statsd.sockets); // destroy the dictionaries dictionary_destroy(statsd.gauges.dict); dictionary_destroy(statsd.meters.dict); dictionary_destroy(statsd.counters.dict); dictionary_destroy(statsd.histograms.dict); dictionary_destroy(statsd.dictionaries.dict); dictionary_destroy(statsd.sets.dict); dictionary_destroy(statsd.timers.dict); info("STATSD: cleanup completed."); static_thread->enabled = NETDATA_MAIN_THREAD_EXITED; worker_unregister(); } #define WORKER_STATSD_FLUSH_GAUGES 0 #define WORKER_STATSD_FLUSH_COUNTERS 1 #define WORKER_STATSD_FLUSH_METERS 2 #define WORKER_STATSD_FLUSH_TIMERS 3 #define WORKER_STATSD_FLUSH_HISTOGRAMS 4 #define WORKER_STATSD_FLUSH_SETS 5 #define WORKER_STATSD_FLUSH_DICTIONARIES 6 #define WORKER_STATSD_FLUSH_STATS 7 #if WORKER_UTILIZATION_MAX_JOB_TYPES < 8 #error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 8 #endif void *statsd_main(void *ptr) { worker_register("STATSDFLUSH"); worker_register_job_name(WORKER_STATSD_FLUSH_GAUGES, "gauges"); worker_register_job_name(WORKER_STATSD_FLUSH_COUNTERS, "counters"); worker_register_job_name(WORKER_STATSD_FLUSH_METERS, "meters"); worker_register_job_name(WORKER_STATSD_FLUSH_TIMERS, "timers"); worker_register_job_name(WORKER_STATSD_FLUSH_HISTOGRAMS, "histograms"); worker_register_job_name(WORKER_STATSD_FLUSH_SETS, "sets"); worker_register_job_name(WORKER_STATSD_FLUSH_DICTIONARIES, "dictionaries"); worker_register_job_name(WORKER_STATSD_FLUSH_STATS, "statistics"); netdata_thread_cleanup_push(statsd_main_cleanup, ptr); statsd.gauges.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); statsd.meters.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); statsd.counters.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); statsd.histograms.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); statsd.dictionaries.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); statsd.sets.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); statsd.timers.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS); dictionary_register_insert_callback(statsd.gauges.dict, dictionary_metric_insert_callback, &statsd.gauges); dictionary_register_insert_callback(statsd.meters.dict, dictionary_metric_insert_callback, &statsd.meters); dictionary_register_insert_callback(statsd.counters.dict, dictionary_metric_insert_callback, &statsd.counters); dictionary_register_insert_callback(statsd.histograms.dict, dictionary_metric_insert_callback, &statsd.histograms); dictionary_register_insert_callback(statsd.dictionaries.dict, dictionary_metric_insert_callback, &statsd.dictionaries); dictionary_register_insert_callback(statsd.sets.dict, dictionary_metric_insert_callback, &statsd.sets); dictionary_register_insert_callback(statsd.timers.dict, dictionary_metric_insert_callback, &statsd.timers); dictionary_register_delete_callback(statsd.gauges.dict, dictionary_metric_delete_callback, &statsd.gauges); dictionary_register_delete_callback(statsd.meters.dict, dictionary_metric_delete_callback, &statsd.meters); dictionary_register_delete_callback(statsd.counters.dict, dictionary_metric_delete_callback, &statsd.counters); dictionary_register_delete_callback(statsd.histograms.dict, dictionary_metric_delete_callback, &statsd.histograms); dictionary_register_delete_callback(statsd.dictionaries.dict, dictionary_metric_delete_callback, &statsd.dictionaries); dictionary_register_delete_callback(statsd.sets.dict, dictionary_metric_delete_callback, &statsd.sets); dictionary_register_delete_callback(statsd.timers.dict, dictionary_metric_delete_callback, &statsd.timers); // ---------------------------------------------------------------------------------------------------------------- // statsd configuration statsd.enabled = config_get_boolean(CONFIG_SECTION_PLUGINS, "statsd", statsd.enabled); statsd.update_every = default_rrd_update_every; statsd.update_every = (int)config_get_number(CONFIG_SECTION_STATSD, "update every (flushInterval)", statsd.update_every); if(statsd.update_every < default_rrd_update_every) { error("STATSD: minimum flush interval %d given, but the minimum is the update every of netdata. Using %d", statsd.update_every, default_rrd_update_every); statsd.update_every = default_rrd_update_every; } #ifdef HAVE_RECVMMSG statsd.recvmmsg_size = (size_t)config_get_number(CONFIG_SECTION_STATSD, "udp messages to process at once", (long long)statsd.recvmmsg_size); #endif statsd.charts_for = simple_pattern_create(config_get(CONFIG_SECTION_STATSD, "create private charts for metrics matching", "*"), NULL, SIMPLE_PATTERN_EXACT); statsd.max_private_charts_hard = (size_t)config_get_number(CONFIG_SECTION_STATSD, "max private charts hard limit", (long long)statsd.max_private_charts_hard); statsd.private_charts_rrd_history_entries = (int)config_get_number(CONFIG_SECTION_STATSD, "private charts history", default_rrd_history_entries); statsd.decimal_detail = (collected_number)config_get_number(CONFIG_SECTION_STATSD, "decimal detail", (long long int)statsd.decimal_detail); statsd.tcp_idle_timeout = (size_t) config_get_number(CONFIG_SECTION_STATSD, "disconnect idle tcp clients after seconds", (long long int)statsd.tcp_idle_timeout); statsd.private_charts_hidden = (unsigned int)config_get_boolean(CONFIG_SECTION_STATSD, "private charts hidden", statsd.private_charts_hidden); statsd.histogram_percentile = (double)config_get_float(CONFIG_SECTION_STATSD, "histograms and timers percentile (percentThreshold)", statsd.histogram_percentile); if(isless(statsd.histogram_percentile, 0) || isgreater(statsd.histogram_percentile, 100)) { error("STATSD: invalid histograms and timers percentile %0.5f given", statsd.histogram_percentile); statsd.histogram_percentile = 95.0; } { char buffer[314 + 1]; snprintfz(buffer, 314, "%0.1f%%", statsd.histogram_percentile); statsd.histogram_percentile_str = strdupz(buffer); } statsd.dictionary_max_unique = config_get_number(CONFIG_SECTION_STATSD, "dictionaries max unique dimensions", statsd.dictionary_max_unique); if(config_get_boolean(CONFIG_SECTION_STATSD, "add dimension for number of events received", 0)) { statsd.gauges.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; statsd.counters.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; statsd.meters.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; statsd.sets.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; statsd.histograms.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; statsd.timers.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; statsd.dictionaries.default_options |= STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT; } if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on gauges (deleteGauges)", 0)) statsd.gauges.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on counters (deleteCounters)", 0)) statsd.counters.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on meters (deleteMeters)", 0)) statsd.meters.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on sets (deleteSets)", 0)) statsd.sets.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on histograms (deleteHistograms)", 0)) statsd.histograms.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on timers (deleteTimers)", 0)) statsd.timers.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on dictionaries (deleteDictionaries)", 0)) statsd.dictionaries.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED; size_t max_sockets = (size_t)config_get_number(CONFIG_SECTION_STATSD, "statsd server max TCP sockets", (long long int)(rlimit_nofile.rlim_cur / 4)); #ifdef STATSD_MULTITHREADED statsd.threads = (int)config_get_number(CONFIG_SECTION_STATSD, "threads", processors); if(statsd.threads < 1) { error("STATSD: Invalid number of threads %d, using %d", statsd.threads, processors); statsd.threads = processors; config_set_number(CONFIG_SECTION_STATSD, "collector threads", statsd.threads); } #else statsd.threads = 1; #endif // read custom application definitions statsd_readdir(netdata_configured_user_config_dir, netdata_configured_stock_config_dir, "statsd.d"); // ---------------------------------------------------------------------------------------------------------------- // statsd setup if(!statsd.enabled) goto cleanup; statsd_listen_sockets_setup(); if(!statsd.sockets.opened) { error("STATSD: No statsd sockets to listen to. statsd will be disabled."); goto cleanup; } statsd.collection_threads_status = callocz((size_t)statsd.threads, sizeof(struct collection_thread_status)); int i; for(i = 0; i < statsd.threads ;i++) { statsd.collection_threads_status[i].max_sockets = max_sockets / statsd.threads; char tag[NETDATA_THREAD_TAG_MAX + 1]; snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STATSD_COLLECTOR[%d]", i + 1); netdata_thread_create(&statsd.collection_threads_status[i].thread, tag, NETDATA_THREAD_OPTION_DEFAULT, statsd_collector_thread, &statsd.collection_threads_status[i]); } // ---------------------------------------------------------------------------------------------------------------- // statsd monitoring charts RRDSET *st_metrics = rrdset_create_localhost( "netdata" , "statsd_metrics" , NULL , "statsd" , NULL , "Metrics in the netdata statsd database" , "metrics" , PLUGIN_STATSD_NAME , "stats" , 132010 , statsd.update_every , RRDSET_TYPE_STACKED ); RRDDIM *rd_metrics_gauge = rrddim_add(st_metrics, "gauges", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_metrics_counter = rrddim_add(st_metrics, "counters", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_metrics_timer = rrddim_add(st_metrics, "timers", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_metrics_meter = rrddim_add(st_metrics, "meters", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_metrics_histogram = rrddim_add(st_metrics, "histograms", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_metrics_set = rrddim_add(st_metrics, "sets", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_metrics_dictionary= rrddim_add(st_metrics, "dictionaries", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDSET *st_useful_metrics = rrdset_create_localhost( "netdata" , "statsd_useful_metrics" , NULL , "statsd" , NULL , "Useful metrics in the netdata statsd database" , "metrics" , PLUGIN_STATSD_NAME , "stats" , 132010 , statsd.update_every , RRDSET_TYPE_STACKED ); RRDDIM *rd_useful_metrics_gauge = rrddim_add(st_useful_metrics, "gauges", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_useful_metrics_counter = rrddim_add(st_useful_metrics, "counters", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_useful_metrics_timer = rrddim_add(st_useful_metrics, "timers", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_useful_metrics_meter = rrddim_add(st_useful_metrics, "meters", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_useful_metrics_histogram = rrddim_add(st_useful_metrics, "histograms", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_useful_metrics_set = rrddim_add(st_useful_metrics, "sets", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDDIM *rd_useful_metrics_dictionary= rrddim_add(st_useful_metrics, "dictionaries", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDSET *st_events = rrdset_create_localhost( "netdata" , "statsd_events" , NULL , "statsd" , NULL , "Events processed by the netdata statsd server" , "events/s" , PLUGIN_STATSD_NAME , "stats" , 132011 , statsd.update_every , RRDSET_TYPE_STACKED ); RRDDIM *rd_events_gauge = rrddim_add(st_events, "gauges", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_counter = rrddim_add(st_events, "counters", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_timer = rrddim_add(st_events, "timers", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_meter = rrddim_add(st_events, "meters", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_histogram = rrddim_add(st_events, "histograms", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_set = rrddim_add(st_events, "sets", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_dictionary= rrddim_add(st_events, "dictionaries", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_unknown = rrddim_add(st_events, "unknown", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_events_errors = rrddim_add(st_events, "errors", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDSET *st_reads = rrdset_create_localhost( "netdata" , "statsd_reads" , NULL , "statsd" , NULL , "Read operations made by the netdata statsd server" , "reads/s" , PLUGIN_STATSD_NAME , "stats" , 132012 , statsd.update_every , RRDSET_TYPE_STACKED ); RRDDIM *rd_reads_tcp = rrddim_add(st_reads, "tcp", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_reads_udp = rrddim_add(st_reads, "udp", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDSET *st_bytes = rrdset_create_localhost( "netdata" , "statsd_bytes" , NULL , "statsd" , NULL , "Bytes read by the netdata statsd server" , "kilobits/s" , PLUGIN_STATSD_NAME , "stats" , 132013 , statsd.update_every , RRDSET_TYPE_STACKED ); RRDDIM *rd_bytes_tcp = rrddim_add(st_bytes, "tcp", NULL, 8, BITS_IN_A_KILOBIT, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_bytes_udp = rrddim_add(st_bytes, "udp", NULL, 8, BITS_IN_A_KILOBIT, RRD_ALGORITHM_INCREMENTAL); RRDSET *st_packets = rrdset_create_localhost( "netdata" , "statsd_packets" , NULL , "statsd" , NULL , "Network packets processed by the netdata statsd server" , "packets/s" , PLUGIN_STATSD_NAME , "stats" , 132014 , statsd.update_every , RRDSET_TYPE_STACKED ); RRDDIM *rd_packets_tcp = rrddim_add(st_packets, "tcp", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_packets_udp = rrddim_add(st_packets, "udp", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDSET *st_tcp_connects = rrdset_create_localhost( "netdata" , "tcp_connects" , NULL , "statsd" , NULL , "statsd server TCP connects and disconnects" , "events" , PLUGIN_STATSD_NAME , "stats" , 132015 , statsd.update_every , RRDSET_TYPE_LINE ); RRDDIM *rd_tcp_connects = rrddim_add(st_tcp_connects, "connects", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL); RRDDIM *rd_tcp_disconnects = rrddim_add(st_tcp_connects, "disconnects", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL); RRDSET *st_tcp_connected = rrdset_create_localhost( "netdata" , "tcp_connected" , NULL , "statsd" , NULL , "statsd server TCP connected sockets" , "sockets" , PLUGIN_STATSD_NAME , "stats" , 132016 , statsd.update_every , RRDSET_TYPE_LINE ); RRDDIM *rd_tcp_connected = rrddim_add(st_tcp_connected, "connected", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); RRDSET *st_pcharts = rrdset_create_localhost( "netdata" , "private_charts" , NULL , "statsd" , NULL , "Private metric charts created by the netdata statsd server" , "charts" , PLUGIN_STATSD_NAME , "stats" , 132020 , statsd.update_every , RRDSET_TYPE_AREA ); RRDDIM *rd_pcharts = rrddim_add(st_pcharts, "charts", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE); // ---------------------------------------------------------------------------------------------------------------- // statsd thread to turn metrics into charts usec_t step = statsd.update_every * USEC_PER_SEC; heartbeat_t hb; heartbeat_init(&hb); while(!netdata_exit) { worker_is_idle(); usec_t hb_dt = heartbeat_next(&hb, step); worker_is_busy(WORKER_STATSD_FLUSH_GAUGES); statsd_flush_index_metrics(&statsd.gauges, statsd_flush_gauge); worker_is_busy(WORKER_STATSD_FLUSH_COUNTERS); statsd_flush_index_metrics(&statsd.counters, statsd_flush_counter); worker_is_busy(WORKER_STATSD_FLUSH_METERS); statsd_flush_index_metrics(&statsd.meters, statsd_flush_meter); worker_is_busy(WORKER_STATSD_FLUSH_TIMERS); statsd_flush_index_metrics(&statsd.timers, statsd_flush_timer); worker_is_busy(WORKER_STATSD_FLUSH_HISTOGRAMS); statsd_flush_index_metrics(&statsd.histograms, statsd_flush_histogram); worker_is_busy(WORKER_STATSD_FLUSH_SETS); statsd_flush_index_metrics(&statsd.sets, statsd_flush_set); worker_is_busy(WORKER_STATSD_FLUSH_DICTIONARIES); statsd_flush_index_metrics(&statsd.dictionaries,statsd_flush_dictionary); worker_is_busy(WORKER_STATSD_FLUSH_STATS); statsd_update_all_app_charts(); if(unlikely(netdata_exit)) break; if(likely(hb_dt)) { rrdset_next(st_metrics); rrdset_next(st_useful_metrics); rrdset_next(st_events); rrdset_next(st_reads); rrdset_next(st_bytes); rrdset_next(st_packets); rrdset_next(st_tcp_connects); rrdset_next(st_tcp_connected); rrdset_next(st_pcharts); } rrddim_set_by_pointer(st_metrics, rd_metrics_gauge, (collected_number)statsd.gauges.metrics); rrddim_set_by_pointer(st_metrics, rd_metrics_counter, (collected_number)statsd.counters.metrics); rrddim_set_by_pointer(st_metrics, rd_metrics_timer, (collected_number)statsd.timers.metrics); rrddim_set_by_pointer(st_metrics, rd_metrics_meter, (collected_number)statsd.meters.metrics); rrddim_set_by_pointer(st_metrics, rd_metrics_histogram, (collected_number)statsd.histograms.metrics); rrddim_set_by_pointer(st_metrics, rd_metrics_set, (collected_number)statsd.sets.metrics); rrddim_set_by_pointer(st_metrics, rd_metrics_dictionary, (collected_number)statsd.dictionaries.metrics); rrdset_done(st_metrics); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_gauge, (collected_number)statsd.gauges.useful); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_counter, (collected_number)statsd.counters.useful); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_timer, (collected_number)statsd.timers.useful); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_meter, (collected_number)statsd.meters.useful); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_histogram, (collected_number)statsd.histograms.useful); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_set, (collected_number)statsd.sets.useful); rrddim_set_by_pointer(st_useful_metrics, rd_useful_metrics_dictionary, (collected_number)statsd.dictionaries.useful); rrdset_done(st_useful_metrics); rrddim_set_by_pointer(st_events, rd_events_gauge, (collected_number)statsd.gauges.events); rrddim_set_by_pointer(st_events, rd_events_counter, (collected_number)statsd.counters.events); rrddim_set_by_pointer(st_events, rd_events_timer, (collected_number)statsd.timers.events); rrddim_set_by_pointer(st_events, rd_events_meter, (collected_number)statsd.meters.events); rrddim_set_by_pointer(st_events, rd_events_histogram, (collected_number)statsd.histograms.events); rrddim_set_by_pointer(st_events, rd_events_set, (collected_number)statsd.sets.events); rrddim_set_by_pointer(st_events, rd_events_dictionary, (collected_number)statsd.dictionaries.events); rrddim_set_by_pointer(st_events, rd_events_unknown, (collected_number)statsd.unknown_types); rrddim_set_by_pointer(st_events, rd_events_errors, (collected_number)statsd.socket_errors); rrdset_done(st_events); rrddim_set_by_pointer(st_reads, rd_reads_tcp, (collected_number)statsd.tcp_socket_reads); rrddim_set_by_pointer(st_reads, rd_reads_udp, (collected_number)statsd.udp_socket_reads); rrdset_done(st_reads); rrddim_set_by_pointer(st_bytes, rd_bytes_tcp, (collected_number)statsd.tcp_bytes_read); rrddim_set_by_pointer(st_bytes, rd_bytes_udp, (collected_number)statsd.udp_bytes_read); rrdset_done(st_bytes); rrddim_set_by_pointer(st_packets, rd_packets_tcp, (collected_number)statsd.tcp_packets_received); rrddim_set_by_pointer(st_packets, rd_packets_udp, (collected_number)statsd.udp_packets_received); rrdset_done(st_packets); rrddim_set_by_pointer(st_tcp_connects, rd_tcp_connects, (collected_number)statsd.tcp_socket_connects); rrddim_set_by_pointer(st_tcp_connects, rd_tcp_disconnects, (collected_number)statsd.tcp_socket_disconnects); rrdset_done(st_tcp_connects); rrddim_set_by_pointer(st_tcp_connected, rd_tcp_connected, (collected_number)statsd.tcp_socket_connected); rrdset_done(st_tcp_connected); rrddim_set_by_pointer(st_pcharts, rd_pcharts, (collected_number)statsd.private_charts); rrdset_done(st_pcharts); } cleanup: ; // added semi-colon to prevent older gcc error: label at end of compound statement netdata_thread_cleanup_pop(1); return NULL; }