2 * collectd - src/write_graphite.c
3 * Copyright (C) 2012 Pierre-Yves Ritschard
4 * Copyright (C) 2011 Scott Sanders
5 * Copyright (C) 2009 Paul Sadauskas
6 * Copyright (C) 2009 Doug MacEachern
7 * Copyright (C) 2007-2013 Florian octo Forster
9 * This program is free software; you can redistribute it and/or modify it
10 * under the terms of the GNU General Public License as published by the
11 * Free Software Foundation; only version 2 of the License is applicable.
13 * This program is distributed in the hope that it will be useful, but
14 * WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * General Public License for more details.
18 * You should have received a copy of the GNU General Public License along
19 * with this program; if not, write to the Free Software Foundation, Inc.,
20 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
23 * Florian octo Forster <octo at collectd.org>
24 * Doug MacEachern <dougm at hyperic.com>
25 * Paul Sadauskas <psadauskas at gmail.com>
26 * Scott Sanders <scott at jssjr.com>
27 * Pierre-Yves Ritschard <pyr at spootnik.org>
29 * Based on the write_http plugin.
32 /* write_graphite plugin configuation example
34 * <Plugin write_graphite>
50 #include "utils_complain.h"
51 #include "utils_format_graphite.h"
55 #ifndef WG_DEFAULT_NODE
56 #define WG_DEFAULT_NODE "localhost"
59 #ifndef WG_DEFAULT_SERVICE
60 #define WG_DEFAULT_SERVICE "2003"
63 #ifndef WG_DEFAULT_PROTOCOL
64 #define WG_DEFAULT_PROTOCOL "tcp"
67 #ifndef WG_DEFAULT_LOG_SEND_ERRORS
68 #define WG_DEFAULT_LOG_SEND_ERRORS 1
71 #ifndef WG_DEFAULT_ESCAPE
72 #define WG_DEFAULT_ESCAPE '_'
75 /* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */
76 #ifndef WG_SEND_BUF_SIZE
77 #define WG_SEND_BUF_SIZE 1428
80 #ifndef WG_MIN_RECONNECT_INTERVAL
81 #define WG_MIN_RECONNECT_INTERVAL TIME_T_TO_CDTIME_T(1)
95 _Bool log_send_errors;
100 unsigned int format_flags;
102 char send_buf[WG_SEND_BUF_SIZE];
103 size_t send_buf_free;
104 size_t send_buf_fill;
105 cdtime_t send_buf_init_time;
107 pthread_mutex_t send_lock;
108 c_complain_t init_complaint;
109 cdtime_t last_connect_time;
111 /* Force reconnect useful for load balanced environments */
112 cdtime_t last_reconnect_time;
113 cdtime_t reconnect_interval;
114 _Bool reconnect_interval_reached;
117 /* wg_force_reconnect_check closes cb->sock_fd when it was open for longer
118 * than cb->reconnect_interval. Must hold cb->send_lock when calling. */
119 static void wg_force_reconnect_check(struct wg_callback *cb) {
122 if (cb->reconnect_interval == 0)
125 /* check if address changes if addr_timeout */
127 if ((now - cb->last_reconnect_time) < cb->reconnect_interval)
130 /* here we should close connection on next */
133 cb->last_reconnect_time = now;
134 cb->reconnect_interval_reached = 1;
136 INFO("write_graphite plugin: Connection closed after %.3f seconds.",
137 CDTIME_T_TO_DOUBLE(now - cb->last_reconnect_time));
143 static void wg_reset_buffer(struct wg_callback *cb) {
144 memset(cb->send_buf, 0, sizeof(cb->send_buf));
145 cb->send_buf_free = sizeof(cb->send_buf);
146 cb->send_buf_fill = 0;
147 cb->send_buf_init_time = cdtime();
150 static int wg_send_buffer(struct wg_callback *cb) {
156 status = swrite(cb->sock_fd, cb->send_buf, strlen(cb->send_buf));
158 if (cb->log_send_errors) {
160 ERROR("write_graphite plugin: send to %s:%s (%s) failed with status %zi "
162 cb->node, cb->service, cb->protocol, status,
163 sstrerror(errno, errbuf, sizeof(errbuf)));
175 /* NOTE: You must hold cb->send_lock when calling this function! */
176 static int wg_flush_nolock(cdtime_t timeout, struct wg_callback *cb) {
179 DEBUG("write_graphite plugin: wg_flush_nolock: timeout = %.3f; "
180 "send_buf_fill = %zu;",
181 (double)timeout, cb->send_buf_fill);
183 /* timeout == 0 => flush unconditionally */
188 if ((cb->send_buf_init_time + timeout) > now)
192 if (cb->send_buf_fill == 0) {
193 cb->send_buf_init_time = cdtime();
197 status = wg_send_buffer(cb);
203 static int wg_callback_init(struct wg_callback *cb) {
204 struct addrinfo *ai_list;
208 char connerr[1024] = "";
213 /* Don't try to reconnect too often. By default, one reconnection attempt
214 * is made per second. */
216 if ((now - cb->last_connect_time) < WG_MIN_RECONNECT_INTERVAL)
218 cb->last_connect_time = now;
220 struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
221 .ai_flags = AI_ADDRCONFIG};
223 if (0 == strcasecmp("tcp", cb->protocol))
224 ai_hints.ai_socktype = SOCK_STREAM;
226 ai_hints.ai_socktype = SOCK_DGRAM;
228 status = getaddrinfo(cb->node, cb->service, &ai_hints, &ai_list);
230 ERROR("write_graphite plugin: getaddrinfo (%s, %s, %s) failed: %s",
231 cb->node, cb->service, cb->protocol, gai_strerror(status));
235 assert(ai_list != NULL);
236 for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
237 ai_ptr = ai_ptr->ai_next) {
239 socket(ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
240 if (cb->sock_fd < 0) {
242 snprintf(connerr, sizeof(connerr), "failed to open socket: %s",
243 sstrerror(errno, errbuf, sizeof(errbuf)));
247 set_sock_opts(cb->sock_fd);
249 status = connect(cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
252 snprintf(connerr, sizeof(connerr), "failed to connect to remote "
254 sstrerror(errno, errbuf, sizeof(errbuf)));
263 freeaddrinfo(ai_list);
265 if (cb->sock_fd < 0) {
266 if (connerr[0] == '\0')
267 /* this should not happen but try to get a message anyway */
268 sstrerror(errno, connerr, sizeof(connerr));
269 c_complain(LOG_ERR, &cb->init_complaint,
270 "write_graphite plugin: Connecting to %s:%s via %s failed. "
271 "The last error was: %s",
272 cb->node, cb->service, cb->protocol, connerr);
275 c_release(LOG_INFO, &cb->init_complaint,
276 "write_graphite plugin: Successfully connected to %s:%s via %s.",
277 cb->node, cb->service, cb->protocol);
280 /* wg_force_reconnect_check does not flush the buffer before closing a
281 * sending socket, so only call wg_reset_buffer() if the socket was closed
282 * for a different reason (tracked in cb->reconnect_interval_reached). */
283 if (!cb->reconnect_interval_reached || (cb->send_buf_free == 0))
286 cb->reconnect_interval_reached = 0;
291 static void wg_callback_free(void *data) {
292 struct wg_callback *cb;
299 pthread_mutex_lock(&cb->send_lock);
301 wg_flush_nolock(/* timeout = */ 0, cb);
303 if (cb->sock_fd >= 0) {
315 pthread_mutex_unlock(&cb->send_lock);
316 pthread_mutex_destroy(&cb->send_lock);
321 static int wg_flush(cdtime_t timeout,
322 const char *identifier __attribute__((unused)),
323 user_data_t *user_data) {
324 struct wg_callback *cb;
327 if (user_data == NULL)
330 cb = user_data->data;
332 pthread_mutex_lock(&cb->send_lock);
334 if (cb->sock_fd < 0) {
335 status = wg_callback_init(cb);
337 /* An error message has already been printed. */
338 pthread_mutex_unlock(&cb->send_lock);
343 status = wg_flush_nolock(timeout, cb);
344 pthread_mutex_unlock(&cb->send_lock);
349 static int wg_send_message(char const *message, struct wg_callback *cb) {
353 message_len = strlen(message);
355 pthread_mutex_lock(&cb->send_lock);
357 wg_force_reconnect_check(cb);
359 if (cb->sock_fd < 0) {
360 status = wg_callback_init(cb);
362 /* An error message has already been printed. */
363 pthread_mutex_unlock(&cb->send_lock);
368 if (message_len >= cb->send_buf_free) {
369 status = wg_flush_nolock(/* timeout = */ 0, cb);
371 pthread_mutex_unlock(&cb->send_lock);
376 /* Assert that we have enough space for this message. */
377 assert(message_len < cb->send_buf_free);
379 /* `message_len + 1' because `message_len' does not include the
380 * trailing null byte. Neither does `send_buffer_fill'. */
381 memcpy(cb->send_buf + cb->send_buf_fill, message, message_len + 1);
382 cb->send_buf_fill += message_len;
383 cb->send_buf_free -= message_len;
385 DEBUG("write_graphite plugin: [%s]:%s (%s) buf %zu/%zu (%.1f %%) \"%s\"",
386 cb->node, cb->service, cb->protocol, cb->send_buf_fill,
387 sizeof(cb->send_buf),
388 100.0 * ((double)cb->send_buf_fill) / ((double)sizeof(cb->send_buf)),
391 pthread_mutex_unlock(&cb->send_lock);
396 static int wg_write_messages(const data_set_t *ds, const value_list_t *vl,
397 struct wg_callback *cb) {
398 char buffer[WG_SEND_BUF_SIZE] = {0};
401 if (0 != strcmp(ds->type, vl->type)) {
402 ERROR("write_graphite plugin: DS type does not match "
407 status = format_graphite(buffer, sizeof(buffer), ds, vl, cb->prefix,
408 cb->postfix, cb->escape_char, cb->format_flags);
409 if (status != 0) /* error message has been printed already. */
412 /* Send the message to graphite */
413 status = wg_send_message(buffer, cb);
414 if (status != 0) /* error message has been printed already. */
418 } /* int wg_write_messages */
420 static int wg_write(const data_set_t *ds, const value_list_t *vl,
421 user_data_t *user_data) {
422 struct wg_callback *cb;
425 if (user_data == NULL)
428 cb = user_data->data;
430 status = wg_write_messages(ds, vl, cb);
435 static int config_set_char(char *dest, oconfig_item_t *ci) {
436 char buffer[4] = {0};
439 status = cf_util_get_string_buffer(ci, buffer, sizeof(buffer));
443 if (buffer[0] == 0) {
444 ERROR("write_graphite plugin: Cannot use an empty string for the "
445 "\"EscapeCharacter\" option.");
449 if (buffer[1] != 0) {
450 WARNING("write_graphite plugin: Only the first character of the "
451 "\"EscapeCharacter\" option ('%c') will be used.",
460 static int wg_config_node(oconfig_item_t *ci) {
461 struct wg_callback *cb;
462 char callback_name[DATA_MAX_NAME_LEN];
465 cb = calloc(1, sizeof(*cb));
467 ERROR("write_graphite plugin: calloc failed.");
472 cb->node = strdup(WG_DEFAULT_NODE);
473 cb->service = strdup(WG_DEFAULT_SERVICE);
474 cb->protocol = strdup(WG_DEFAULT_PROTOCOL);
475 cb->last_reconnect_time = cdtime();
476 cb->reconnect_interval = 0;
477 cb->reconnect_interval_reached = 0;
478 cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
481 cb->escape_char = WG_DEFAULT_ESCAPE;
482 cb->format_flags = GRAPHITE_STORE_RATES;
484 /* FIXME: Legacy configuration syntax. */
485 if (strcasecmp("Carbon", ci->key) != 0) {
486 status = cf_util_get_string(ci, &cb->name);
488 wg_callback_free(cb);
493 pthread_mutex_init(&cb->send_lock, /* attr = */ NULL);
494 C_COMPLAIN_INIT(&cb->init_complaint);
496 for (int i = 0; i < ci->children_num; i++) {
497 oconfig_item_t *child = ci->children + i;
499 if (strcasecmp("Host", child->key) == 0)
500 cf_util_get_string(child, &cb->node);
501 else if (strcasecmp("Port", child->key) == 0)
502 cf_util_get_service(child, &cb->service);
503 else if (strcasecmp("Protocol", child->key) == 0) {
504 cf_util_get_string(child, &cb->protocol);
506 if (strcasecmp("UDP", cb->protocol) != 0 &&
507 strcasecmp("TCP", cb->protocol) != 0) {
508 ERROR("write_graphite plugin: Unknown protocol (%s)", cb->protocol);
511 } else if (strcasecmp("ReconnectInterval", child->key) == 0)
512 cf_util_get_cdtime(child, &cb->reconnect_interval);
513 else if (strcasecmp("LogSendErrors", child->key) == 0)
514 cf_util_get_boolean(child, &cb->log_send_errors);
515 else if (strcasecmp("Prefix", child->key) == 0)
516 cf_util_get_string(child, &cb->prefix);
517 else if (strcasecmp("Postfix", child->key) == 0)
518 cf_util_get_string(child, &cb->postfix);
519 else if (strcasecmp("StoreRates", child->key) == 0)
520 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_STORE_RATES);
521 else if (strcasecmp("SeparateInstances", child->key) == 0)
522 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_SEPARATE_INSTANCES);
523 else if (strcasecmp("AlwaysAppendDS", child->key) == 0)
524 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_ALWAYS_APPEND_DS);
525 else if (strcasecmp("PreserveSeparator", child->key) == 0)
526 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_PRESERVE_SEPARATOR);
527 else if (strcasecmp("DropDuplicateFields", child->key) == 0)
528 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_DROP_DUPE_FIELDS);
529 else if (strcasecmp("EscapeCharacter", child->key) == 0)
530 config_set_char(&cb->escape_char, child);
532 ERROR("write_graphite plugin: Invalid configuration "
543 wg_callback_free(cb);
547 /* FIXME: Legacy configuration syntax. */
548 if (cb->name == NULL)
549 snprintf(callback_name, sizeof(callback_name), "write_graphite/%s/%s/%s",
550 cb->node, cb->service, cb->protocol);
552 snprintf(callback_name, sizeof(callback_name), "write_graphite/%s",
555 plugin_register_write(callback_name, wg_write,
557 .data = cb, .free_func = wg_callback_free,
560 plugin_register_flush(callback_name, wg_flush, &(user_data_t){.data = cb});
565 static int wg_config(oconfig_item_t *ci) {
566 for (int i = 0; i < ci->children_num; i++) {
567 oconfig_item_t *child = ci->children + i;
569 if (strcasecmp("Node", child->key) == 0)
570 wg_config_node(child);
571 /* FIXME: Remove this legacy mode in version 6. */
572 else if (strcasecmp("Carbon", child->key) == 0)
573 wg_config_node(child);
575 ERROR("write_graphite plugin: Invalid configuration "
584 void module_register(void) {
585 plugin_register_complex_config("write_graphite", wg_config);