X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fwrite_graphite.c;h=d6583a75f04ebb63901649a6f6f763ae3a9aa473;hb=b61192846083d41c7bbad8195fe268193b2fb022;hp=783b1f5b9f933951bf37b8b68ffcb3685c11ef57;hpb=3506f7a5735c6ad217323c0ad3cb3ba24bc8a6f9;p=collectd.git diff --git a/src/write_graphite.c b/src/write_graphite.c index 783b1f5b..d6583a75 100644 --- a/src/write_graphite.c +++ b/src/write_graphite.c @@ -1,6 +1,10 @@ /** * collectd - src/write_graphite.c - * Copyright (C) 2011 Scott Sanders + * Copyright (C) 2012 Pierre-Yves Ritschard + * Copyright (C) 2011 Scott Sanders + * Copyright (C) 2009 Paul Sadauskas + * Copyright (C) 2009 Doug MacEachern + * Copyright (C) 2007-2012 Florian octo Forster * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the @@ -15,10 +19,14 @@ * with this program; if not, write to the Free Software Foundation, Inc., * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA * - * Author: - * Scott Sanders + * Authors: + * Florian octo Forster + * Doug MacEachern + * Paul Sadauskas + * Scott Sanders + * Pierre-Yves Ritschard * - * based on the excellent write_http plugin + * Based on the write_http plugin. **/ /* write_graphite plugin configuation example @@ -46,15 +54,6 @@ #include #include -#ifndef WG_FORMAT_NAME -#define WG_FORMAT_NAME(ret, ret_len, vl, cb, ds_name) \ - wg_format_name (ret, ret_len, (vl)->host, \ - (vl)->plugin, (vl)->plugin_instance, \ - (vl)->type, (vl)->type_instance, \ - (cb)->prefix, (cb)->postfix, \ - ds_name, (cb)->escape_char) -#endif - #ifndef WG_DEFAULT_NODE # define WG_DEFAULT_NODE "localhost" #endif @@ -63,8 +62,13 @@ # define WG_DEFAULT_SERVICE "2003" #endif +#ifndef WG_DEFAULT_ESCAPE +# define WG_DEFAULT_ESCAPE '_' +#endif + +/* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */ #ifndef WG_SEND_BUF_SIZE -# define WG_SEND_BUF_SIZE 4096 +# define WG_SEND_BUF_SIZE 1428 #endif /* @@ -73,7 +77,6 @@ struct wg_callback { int sock_fd; - struct hostent *server; char *node; char *service; @@ -81,6 +84,10 @@ struct wg_callback char *postfix; char escape_char; + _Bool store_rates; + _Bool separate_instances; + _Bool always_append_ds; + char send_buf[WG_SEND_BUF_SIZE]; size_t send_buf_free; size_t send_buf_fill; @@ -103,30 +110,26 @@ static void wg_reset_buffer (struct wg_callback *cb) static int wg_send_buffer (struct wg_callback *cb) { - int status = 0; + ssize_t status = 0; - status = write (cb->sock_fd, cb->send_buf, strlen (cb->send_buf)); + status = swrite (cb->sock_fd, cb->send_buf, strlen (cb->send_buf)); if (status < 0) { - ERROR ("write_graphite plugin: send failed with " - "status %i (%s)", - status, - strerror (errno)); + char errbuf[1024]; + ERROR ("write_graphite plugin: send failed with status %zi (%s)", + status, sstrerror (errno, errbuf, sizeof (errbuf))); - pthread_mutex_trylock (&cb->send_lock); - DEBUG ("write_graphite plugin: closing socket and restting fd " - "so reinit will occur"); close (cb->sock_fd); cb->sock_fd = -1; - pthread_mutex_unlock (&cb->send_lock); - return (-1); } + return (0); } +/* NOTE: You must hold cb->send_lock when calling this function! */ static int wg_flush_nolock (cdtime_t timeout, struct wg_callback *cb) { int status; @@ -233,14 +236,20 @@ static void wg_callback_free (void *data) cb = data; + pthread_mutex_lock (&cb->send_lock); + wg_flush_nolock (/* timeout = */ 0, cb); close(cb->sock_fd); + cb->sock_fd = -1; + sfree(cb->node); sfree(cb->service); sfree(cb->prefix); sfree(cb->postfix); + pthread_mutex_destroy (&cb->send_lock); + sfree(cb); } @@ -350,86 +359,79 @@ static void wg_copy_escape_part (char *dst, const char *src, size_t dst_len, for (i = 0; i < dst_len; i++) { + if (src[i] == 0) + { + dst[i] = 0; + break; + } + if ((src[i] == '.') || isspace ((int) src[i]) || iscntrl ((int) src[i])) dst[i] = escape_char; else dst[i] = src[i]; - - if (src[i] == 0) - break; } } static int wg_format_name (char *ret, int ret_len, - const char *hostname, - const char *plugin, const char *plugin_instance, - const char *type, const char *type_instance, - const char *prefix, const char *postfix, - const char *ds_name, char escape_char) + const value_list_t *vl, + const struct wg_callback *cb, + const char *ds_name) { - char n_hostname[DATA_MAX_NAME_LEN]; + char n_host[DATA_MAX_NAME_LEN]; char n_plugin[DATA_MAX_NAME_LEN]; char n_plugin_instance[DATA_MAX_NAME_LEN]; char n_type[DATA_MAX_NAME_LEN]; char n_type_instance[DATA_MAX_NAME_LEN]; - int status; - assert (hostname != NULL); - assert (plugin != NULL); - assert (type != NULL); - assert (ds_name != NULL); + char *prefix; + char *postfix; + + char tmp_plugin[2 * DATA_MAX_NAME_LEN + 1]; + char tmp_type[2 * DATA_MAX_NAME_LEN + 1]; + prefix = cb->prefix; if (prefix == NULL) prefix = ""; + postfix = cb->postfix; if (postfix == NULL) postfix = ""; - wg_copy_escape_part (n_hostname, hostname, - sizeof (n_hostname), escape_char); - wg_copy_escape_part (n_plugin, plugin, - sizeof (n_plugin), escape_char); - wg_copy_escape_part (n_plugin_instance, plugin_instance, - sizeof (n_plugin_instance), escape_char); - wg_copy_escape_part (n_type, type, - sizeof (n_type), escape_char); - wg_copy_escape_part (n_type_instance, type_instance, - sizeof (n_type_instance), escape_char); - - if (n_plugin_instance[0] == '\0') - { - if (n_type_instance[0] == '\0') - { - status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s", - prefix, n_hostname, postfix, n_plugin, n_type, ds_name); - } - else - { - status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s-%s.%s", - prefix, n_hostname, postfix, n_plugin, n_type, - n_type_instance, ds_name); - } - } + wg_copy_escape_part (n_host, vl->host, + sizeof (n_host), cb->escape_char); + wg_copy_escape_part (n_plugin, vl->plugin, + sizeof (n_plugin), cb->escape_char); + wg_copy_escape_part (n_plugin_instance, vl->plugin_instance, + sizeof (n_plugin_instance), cb->escape_char); + wg_copy_escape_part (n_type, vl->type, + sizeof (n_type), cb->escape_char); + wg_copy_escape_part (n_type_instance, vl->type_instance, + sizeof (n_type_instance), cb->escape_char); + + if (n_plugin_instance[0] != '\0') + ssnprintf (tmp_plugin, sizeof (tmp_plugin), "%s%c%s", + n_plugin, + cb->separate_instances ? '.' : '-', + n_plugin_instance); else - { - if (n_type_instance[0] == '\0') - { - status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s.%s", - prefix, n_hostname, postfix, n_plugin, - n_plugin_instance, n_type, ds_name); - } - else - { - status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s-%s.%s", - prefix, n_hostname, postfix, n_plugin, - n_plugin_instance, n_type, n_type_instance, ds_name); - } - } + sstrncpy (tmp_plugin, n_plugin, sizeof (tmp_plugin)); - if ((status < 1) || (status >= ret_len)) - return (-1); + if (n_type_instance[0] != '\0') + ssnprintf (tmp_type, sizeof (tmp_type), "%s%c%s", + n_type, + cb->separate_instances ? '.' : '-', + n_type_instance); + else + sstrncpy (tmp_type, n_type, sizeof (tmp_type)); + + if (ds_name != NULL) + ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s", + prefix, n_host, postfix, tmp_plugin, tmp_type, ds_name); + else + ssnprintf (ret, ret_len, "%s%s%s.%s.%s", + prefix, n_host, postfix, tmp_plugin, tmp_type); return (0); } @@ -442,17 +444,16 @@ static int wg_send_message (const char* key, const char* value, char message[1024]; message_len = (size_t) ssnprintf (message, sizeof (message), - "%s %s %.0f\n", + "%s %s %u\r\n", key, value, - CDTIME_T_TO_DOUBLE(time)); + (unsigned int) CDTIME_T_TO_TIME_T (time)); if (message_len >= sizeof (message)) { ERROR ("write_graphite plugin: message buffer too small: " "Need %zu bytes.", message_len + 1); return (-1); } - pthread_mutex_lock (&cb->send_lock); if (cb->sock_fd < 0) @@ -475,6 +476,8 @@ static int wg_send_message (const char* key, const char* value, return (status); } } + + /* Assert that we have enough space for this message. */ assert (message_len < cb->send_buf_free); /* `message_len + 1' because `message_len' does not include the @@ -484,14 +487,13 @@ static int wg_send_message (const char* key, const char* value, cb->send_buf_fill += message_len; cb->send_buf_free -= message_len; - DEBUG ("write_graphite plugin: <%s:%s> buf %zu/%zu (%g%%) \"%s\"", + DEBUG ("write_graphite plugin: [%s]:%s buf %zu/%zu (%.1f %%) \"%s\"", cb->node, cb->service, cb->send_buf_fill, sizeof (cb->send_buf), 100.0 * ((double) cb->send_buf_fill) / ((double) sizeof (cb->send_buf)), message); - /* Check if we have enough space for this message. */ pthread_mutex_unlock (&cb->send_lock); return (0); @@ -512,43 +514,15 @@ static int wg_write_messages (const data_set_t *ds, const value_list_t *vl, return -1; } - if (ds->ds_num > 1) - { - for (i = 0; i < ds->ds_num; i++) - { - /* Copy the identifier to `key' and escape it. */ - status = WG_FORMAT_NAME (key, sizeof (key), vl, cb, ds->ds[i].name); - if (status != 0) - { - ERROR ("write_graphite plugin: error with format_name"); - return (status); - } - - escape_string (key, sizeof (key)); - /* Convert the values to an ASCII representation and put that - * into `values'. */ - status = wg_format_values (values, sizeof (values), i, ds, vl, 0); - if (status != 0) - { - ERROR ("write_graphite plugin: error with " - "wg_format_values"); - return (status); - } - - /* Send the message to graphite */ - status = wg_send_message (key, values, vl->time, cb); - if (status != 0) - { - ERROR ("write_graphite plugin: error with " - "wg_send_message"); - return (status); - } - } - } - else + for (i = 0; i < ds->ds_num; i++) { + const char *ds_name = NULL; + + if (cb->always_append_ds || (ds->ds_num > 1)) + ds_name = ds->ds[i].name; + /* Copy the identifier to `key' and escape it. */ - status = WG_FORMAT_NAME (key, sizeof (key), vl, cb, NULL); + status = wg_format_name (key, sizeof (key), vl, cb, ds_name); if (status != 0) { ERROR ("write_graphite plugin: error with format_name"); @@ -558,7 +532,8 @@ static int wg_write_messages (const data_set_t *ds, const value_list_t *vl, escape_string (key, sizeof (key)); /* Convert the values to an ASCII representation and put that into * `values'. */ - status = wg_format_values (values, sizeof (values), 0, ds, vl, 0); + status = wg_format_values (values, sizeof (values), i, ds, vl, + cb->store_rates); if (status != 0) { ERROR ("write_graphite plugin: error with " @@ -586,7 +561,7 @@ static int wg_write (const data_set_t *ds, const value_list_t *vl, int status; if (user_data == NULL) - return (-EINVAL); + return (EINVAL); cb = user_data->data; @@ -598,14 +573,30 @@ static int wg_write (const data_set_t *ds, const value_list_t *vl, static int config_set_char (char *dest, oconfig_item_t *ci) { - if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) + char buffer[4]; + int status; + + memset (buffer, 0, sizeof (buffer)); + + status = cf_util_get_string_buffer (ci, buffer, sizeof (buffer)); + if (status != 0) + return (status); + + if (buffer[0] == 0) { - WARNING ("write_graphite plugin: The `%s' config option " - "needs exactly one string argument.", ci->key); + ERROR ("write_graphite plugin: Cannot use an empty string for the " + "\"EscapeCharacter\" option."); return (-1); } - *dest = ci->values[0].value.string[0]; + if (buffer[1] != 0) + { + WARNING ("write_graphite plugin: Only the first character of the " + "\"EscapeCharacter\" option ('%c') will be used.", + (int) buffer[0]); + } + + *dest = buffer[0]; return (0); } @@ -614,6 +605,7 @@ static int wg_config_carbon (oconfig_item_t *ci) { struct wg_callback *cb; user_data_t user_data; + char callback_name[DATA_MAX_NAME_LEN]; int i; cb = malloc (sizeof (*cb)); @@ -628,8 +620,8 @@ static int wg_config_carbon (oconfig_item_t *ci) cb->service = NULL; cb->prefix = NULL; cb->postfix = NULL; - cb->server = NULL; - cb->escape_char = '_'; + cb->escape_char = WG_DEFAULT_ESCAPE; + cb->store_rates = 1; pthread_mutex_init (&cb->send_lock, /* attr = */ NULL); @@ -640,11 +632,17 @@ static int wg_config_carbon (oconfig_item_t *ci) if (strcasecmp ("Host", child->key) == 0) cf_util_get_string (child, &cb->node); else if (strcasecmp ("Port", child->key) == 0) - cf_util_get_string (child, &cb->service); + cf_util_get_service (child, &cb->service); else if (strcasecmp ("Prefix", child->key) == 0) cf_util_get_string (child, &cb->prefix); else if (strcasecmp ("Postfix", child->key) == 0) cf_util_get_string (child, &cb->postfix); + else if (strcasecmp ("StoreRates", child->key) == 0) + cf_util_get_boolean (child, &cb->store_rates); + else if (strcasecmp ("SeparateInstances", child->key) == 0) + cf_util_get_boolean (child, &cb->separate_instances); + else if (strcasecmp ("AlwaysAppendDS", child->key) == 0) + cf_util_get_boolean (child, &cb->always_append_ds); else if (strcasecmp ("EscapeCharacter", child->key) == 0) config_set_char (&cb->escape_char, child); else @@ -654,17 +652,17 @@ static int wg_config_carbon (oconfig_item_t *ci) } } - DEBUG ("write_graphite: Registering write callback to carbon agent %s:%s", - cb->node ? cb->node : WG_DEFAULT_NODE, - cb->service ? cb->service : WG_DEFAULT_SERVICE); + ssnprintf (callback_name, sizeof (callback_name), "write_graphite/%s/%s", + cb->node != NULL ? cb->node : WG_DEFAULT_NODE, + cb->service != NULL ? cb->service : WG_DEFAULT_SERVICE); memset (&user_data, 0, sizeof (user_data)); user_data.data = cb; - user_data.free_func = NULL; - plugin_register_flush ("write_graphite", wg_flush, &user_data); - user_data.free_func = wg_callback_free; - plugin_register_write ("write_graphite", wg_write, &user_data); + plugin_register_write (callback_name, wg_write, &user_data); + + user_data.free_func = NULL; + plugin_register_flush (callback_name, wg_flush, &user_data); return (0); }