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) {
159 ERROR("write_graphite plugin: send to %s:%s (%s) failed with status %zi "
161 cb->node, cb->service, cb->protocol, status, STRERRNO);
173 /* NOTE: You must hold cb->send_lock when calling this function! */
174 static int wg_flush_nolock(cdtime_t timeout, struct wg_callback *cb) {
177 DEBUG("write_graphite plugin: wg_flush_nolock: timeout = %.3f; "
178 "send_buf_fill = %zu;",
179 (double)timeout, cb->send_buf_fill);
181 /* timeout == 0 => flush unconditionally */
186 if ((cb->send_buf_init_time + timeout) > now)
190 if (cb->send_buf_fill == 0) {
191 cb->send_buf_init_time = cdtime();
195 status = wg_send_buffer(cb);
201 static int wg_callback_init(struct wg_callback *cb) {
202 struct addrinfo *ai_list;
206 char connerr[1024] = "";
211 /* Don't try to reconnect too often. By default, one reconnection attempt
212 * is made per second. */
214 if ((now - cb->last_connect_time) < WG_MIN_RECONNECT_INTERVAL)
216 cb->last_connect_time = now;
218 struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
219 .ai_flags = AI_ADDRCONFIG};
221 if (0 == strcasecmp("tcp", cb->protocol))
222 ai_hints.ai_socktype = SOCK_STREAM;
224 ai_hints.ai_socktype = SOCK_DGRAM;
226 status = getaddrinfo(cb->node, cb->service, &ai_hints, &ai_list);
228 ERROR("write_graphite plugin: getaddrinfo (%s, %s, %s) failed: %s",
229 cb->node, cb->service, cb->protocol, gai_strerror(status));
233 assert(ai_list != NULL);
234 for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
235 ai_ptr = ai_ptr->ai_next) {
237 socket(ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
238 if (cb->sock_fd < 0) {
239 snprintf(connerr, sizeof(connerr), "failed to open socket: %s", STRERRNO);
243 set_sock_opts(cb->sock_fd);
245 status = connect(cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
247 snprintf(connerr, sizeof(connerr), "failed to connect to remote host: %s",
257 freeaddrinfo(ai_list);
259 if (cb->sock_fd < 0) {
260 c_complain(LOG_ERR, &cb->init_complaint,
261 "write_graphite plugin: Connecting to %s:%s via %s failed. "
262 "The last error was: %s",
263 cb->node, cb->service, cb->protocol, connerr);
266 c_release(LOG_INFO, &cb->init_complaint,
267 "write_graphite plugin: Successfully connected to %s:%s via %s.",
268 cb->node, cb->service, cb->protocol);
271 /* wg_force_reconnect_check does not flush the buffer before closing a
272 * sending socket, so only call wg_reset_buffer() if the socket was closed
273 * for a different reason (tracked in cb->reconnect_interval_reached). */
274 if (!cb->reconnect_interval_reached || (cb->send_buf_free == 0))
277 cb->reconnect_interval_reached = 0;
282 static void wg_callback_free(void *data) {
283 struct wg_callback *cb;
290 pthread_mutex_lock(&cb->send_lock);
292 wg_flush_nolock(/* timeout = */ 0, cb);
294 if (cb->sock_fd >= 0) {
306 pthread_mutex_destroy(&cb->send_lock);
311 static int wg_flush(cdtime_t timeout,
312 const char *identifier __attribute__((unused)),
313 user_data_t *user_data) {
314 struct wg_callback *cb;
317 if (user_data == NULL)
320 cb = user_data->data;
322 pthread_mutex_lock(&cb->send_lock);
324 if (cb->sock_fd < 0) {
325 status = wg_callback_init(cb);
327 /* An error message has already been printed. */
328 pthread_mutex_unlock(&cb->send_lock);
333 status = wg_flush_nolock(timeout, cb);
334 pthread_mutex_unlock(&cb->send_lock);
339 static int wg_send_message(char const *message, struct wg_callback *cb) {
343 message_len = strlen(message);
345 pthread_mutex_lock(&cb->send_lock);
347 wg_force_reconnect_check(cb);
349 if (cb->sock_fd < 0) {
350 status = wg_callback_init(cb);
352 /* An error message has already been printed. */
353 pthread_mutex_unlock(&cb->send_lock);
358 if (message_len >= cb->send_buf_free) {
359 status = wg_flush_nolock(/* timeout = */ 0, cb);
361 pthread_mutex_unlock(&cb->send_lock);
366 /* Assert that we have enough space for this message. */
367 assert(message_len < cb->send_buf_free);
369 /* `message_len + 1' because `message_len' does not include the
370 * trailing null byte. Neither does `send_buffer_fill'. */
371 memcpy(cb->send_buf + cb->send_buf_fill, message, message_len + 1);
372 cb->send_buf_fill += message_len;
373 cb->send_buf_free -= message_len;
375 DEBUG("write_graphite plugin: [%s]:%s (%s) buf %zu/%zu (%.1f %%) \"%s\"",
376 cb->node, cb->service, cb->protocol, cb->send_buf_fill,
377 sizeof(cb->send_buf),
378 100.0 * ((double)cb->send_buf_fill) / ((double)sizeof(cb->send_buf)),
381 pthread_mutex_unlock(&cb->send_lock);
386 static int wg_write_messages(const data_set_t *ds, const value_list_t *vl,
387 struct wg_callback *cb) {
388 char buffer[WG_SEND_BUF_SIZE] = {0};
391 if (0 != strcmp(ds->type, vl->type)) {
392 ERROR("write_graphite plugin: DS type does not match "
397 status = format_graphite(buffer, sizeof(buffer), ds, vl, cb->prefix,
398 cb->postfix, cb->escape_char, cb->format_flags);
399 if (status != 0) /* error message has been printed already. */
402 /* Send the message to graphite */
403 status = wg_send_message(buffer, cb);
404 if (status != 0) /* error message has been printed already. */
408 } /* int wg_write_messages */
410 static int wg_write(const data_set_t *ds, const value_list_t *vl,
411 user_data_t *user_data) {
412 struct wg_callback *cb;
415 if (user_data == NULL)
418 cb = user_data->data;
420 status = wg_write_messages(ds, vl, cb);
425 static int config_set_char(char *dest, oconfig_item_t *ci) {
426 char buffer[4] = {0};
429 status = cf_util_get_string_buffer(ci, buffer, sizeof(buffer));
433 if (buffer[0] == 0) {
434 ERROR("write_graphite plugin: Cannot use an empty string for the "
435 "\"EscapeCharacter\" option.");
439 if (buffer[1] != 0) {
440 WARNING("write_graphite plugin: Only the first character of the "
441 "\"EscapeCharacter\" option ('%c') will be used.",
450 static int wg_config_node(oconfig_item_t *ci) {
451 struct wg_callback *cb;
452 char callback_name[DATA_MAX_NAME_LEN];
455 cb = calloc(1, sizeof(*cb));
457 ERROR("write_graphite plugin: calloc failed.");
462 cb->node = strdup(WG_DEFAULT_NODE);
463 cb->service = strdup(WG_DEFAULT_SERVICE);
464 cb->protocol = strdup(WG_DEFAULT_PROTOCOL);
465 cb->last_reconnect_time = cdtime();
466 cb->reconnect_interval = 0;
467 cb->reconnect_interval_reached = 0;
468 cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
471 cb->escape_char = WG_DEFAULT_ESCAPE;
472 cb->format_flags = GRAPHITE_STORE_RATES;
474 /* FIXME: Legacy configuration syntax. */
475 if (strcasecmp("Carbon", ci->key) != 0) {
476 status = cf_util_get_string(ci, &cb->name);
478 wg_callback_free(cb);
483 pthread_mutex_init(&cb->send_lock, /* attr = */ NULL);
484 C_COMPLAIN_INIT(&cb->init_complaint);
486 for (int i = 0; i < ci->children_num; i++) {
487 oconfig_item_t *child = ci->children + i;
489 if (strcasecmp("Host", child->key) == 0)
490 cf_util_get_string(child, &cb->node);
491 else if (strcasecmp("Port", child->key) == 0)
492 cf_util_get_service(child, &cb->service);
493 else if (strcasecmp("Protocol", child->key) == 0) {
494 cf_util_get_string(child, &cb->protocol);
496 if (strcasecmp("UDP", cb->protocol) != 0 &&
497 strcasecmp("TCP", cb->protocol) != 0) {
498 ERROR("write_graphite plugin: Unknown protocol (%s)", cb->protocol);
501 } else if (strcasecmp("ReconnectInterval", child->key) == 0)
502 cf_util_get_cdtime(child, &cb->reconnect_interval);
503 else if (strcasecmp("LogSendErrors", child->key) == 0)
504 cf_util_get_boolean(child, &cb->log_send_errors);
505 else if (strcasecmp("Prefix", child->key) == 0)
506 cf_util_get_string(child, &cb->prefix);
507 else if (strcasecmp("Postfix", child->key) == 0)
508 cf_util_get_string(child, &cb->postfix);
509 else if (strcasecmp("StoreRates", child->key) == 0)
510 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_STORE_RATES);
511 else if (strcasecmp("SeparateInstances", child->key) == 0)
512 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_SEPARATE_INSTANCES);
513 else if (strcasecmp("AlwaysAppendDS", child->key) == 0)
514 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_ALWAYS_APPEND_DS);
515 else if (strcasecmp("PreserveSeparator", child->key) == 0)
516 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_PRESERVE_SEPARATOR);
517 else if (strcasecmp("DropDuplicateFields", child->key) == 0)
518 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_DROP_DUPE_FIELDS);
519 else if (strcasecmp("EscapeCharacter", child->key) == 0)
520 config_set_char(&cb->escape_char, child);
522 ERROR("write_graphite plugin: Invalid configuration "
533 wg_callback_free(cb);
537 /* FIXME: Legacy configuration syntax. */
538 if (cb->name == NULL)
539 snprintf(callback_name, sizeof(callback_name), "write_graphite/%s/%s/%s",
540 cb->node, cb->service, cb->protocol);
542 snprintf(callback_name, sizeof(callback_name), "write_graphite/%s",
545 plugin_register_write(callback_name, wg_write,
547 .data = cb, .free_func = wg_callback_free,
550 plugin_register_flush(callback_name, wg_flush, &(user_data_t){.data = cb});
555 static int wg_config(oconfig_item_t *ci) {
556 for (int i = 0; i < ci->children_num; i++) {
557 oconfig_item_t *child = ci->children + i;
559 if (strcasecmp("Node", child->key) == 0)
560 wg_config_node(child);
561 /* FIXME: Remove this legacy mode in version 6. */
562 else if (strcasecmp("Carbon", child->key) == 0)
563 wg_config_node(child);
565 ERROR("write_graphite plugin: Invalid configuration "
574 void module_register(void) {
575 plugin_register_complex_config("write_graphite", wg_config);