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)
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 = %" PRIsz ";",
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_unlock(&cb->send_lock);
307 pthread_mutex_destroy(&cb->send_lock);
312 static int wg_flush(cdtime_t timeout,
313 const char *identifier __attribute__((unused)),
314 user_data_t *user_data) {
315 struct wg_callback *cb;
318 if (user_data == NULL)
321 cb = user_data->data;
323 pthread_mutex_lock(&cb->send_lock);
325 if (cb->sock_fd < 0) {
326 status = wg_callback_init(cb);
328 /* An error message has already been printed. */
329 pthread_mutex_unlock(&cb->send_lock);
334 status = wg_flush_nolock(timeout, cb);
335 pthread_mutex_unlock(&cb->send_lock);
340 static int wg_send_message(char const *message, struct wg_callback *cb) {
344 message_len = strlen(message);
346 pthread_mutex_lock(&cb->send_lock);
348 wg_force_reconnect_check(cb);
350 if (cb->sock_fd < 0) {
351 status = wg_callback_init(cb);
353 /* An error message has already been printed. */
354 pthread_mutex_unlock(&cb->send_lock);
359 if (message_len >= cb->send_buf_free) {
360 status = wg_flush_nolock(/* timeout = */ 0, cb);
362 pthread_mutex_unlock(&cb->send_lock);
367 /* Assert that we have enough space for this message. */
368 assert(message_len < cb->send_buf_free);
370 /* `message_len + 1' because `message_len' does not include the
371 * trailing null byte. Neither does `send_buffer_fill'. */
372 memcpy(cb->send_buf + cb->send_buf_fill, message, message_len + 1);
373 cb->send_buf_fill += message_len;
374 cb->send_buf_free -= message_len;
376 DEBUG("write_graphite plugin: [%s]:%s (%s) buf %" PRIsz "/%" PRIsz
378 cb->node, cb->service, cb->protocol, cb->send_buf_fill,
379 sizeof(cb->send_buf),
380 100.0 * ((double)cb->send_buf_fill) / ((double)sizeof(cb->send_buf)),
383 pthread_mutex_unlock(&cb->send_lock);
388 static int wg_write_messages(const data_set_t *ds, const value_list_t *vl,
389 struct wg_callback *cb) {
390 char buffer[WG_SEND_BUF_SIZE] = {0};
393 if (0 != strcmp(ds->type, vl->type)) {
394 ERROR("write_graphite plugin: DS type does not match "
399 status = format_graphite(buffer, sizeof(buffer), ds, vl, cb->prefix,
400 cb->postfix, cb->escape_char, cb->format_flags);
401 if (status != 0) /* error message has been printed already. */
404 /* Send the message to graphite */
405 status = wg_send_message(buffer, cb);
406 if (status != 0) /* error message has been printed already. */
410 } /* int wg_write_messages */
412 static int wg_write(const data_set_t *ds, const value_list_t *vl,
413 user_data_t *user_data) {
414 struct wg_callback *cb;
417 if (user_data == NULL)
420 cb = user_data->data;
422 status = wg_write_messages(ds, vl, cb);
427 static int config_set_char(char *dest, oconfig_item_t *ci) {
428 char buffer[4] = {0};
431 status = cf_util_get_string_buffer(ci, buffer, sizeof(buffer));
435 if (buffer[0] == 0) {
436 ERROR("write_graphite plugin: Cannot use an empty string for the "
437 "\"EscapeCharacter\" option.");
441 if (buffer[1] != 0) {
442 WARNING("write_graphite plugin: Only the first character of the "
443 "\"EscapeCharacter\" option ('%c') will be used.",
452 static int wg_config_node(oconfig_item_t *ci) {
453 struct wg_callback *cb;
454 char callback_name[DATA_MAX_NAME_LEN];
457 cb = calloc(1, sizeof(*cb));
459 ERROR("write_graphite plugin: calloc failed.");
464 cb->node = strdup(WG_DEFAULT_NODE);
465 cb->service = strdup(WG_DEFAULT_SERVICE);
466 cb->protocol = strdup(WG_DEFAULT_PROTOCOL);
467 cb->last_reconnect_time = cdtime();
468 cb->reconnect_interval = 0;
469 cb->reconnect_interval_reached = 0;
470 cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
473 cb->escape_char = WG_DEFAULT_ESCAPE;
474 cb->format_flags = GRAPHITE_STORE_RATES;
476 /* FIXME: Legacy configuration syntax. */
477 if (strcasecmp("Carbon", ci->key) != 0) {
478 status = cf_util_get_string(ci, &cb->name);
480 wg_callback_free(cb);
485 pthread_mutex_init(&cb->send_lock, /* attr = */ NULL);
486 C_COMPLAIN_INIT(&cb->init_complaint);
488 for (int i = 0; i < ci->children_num; i++) {
489 oconfig_item_t *child = ci->children + i;
491 if (strcasecmp("Host", child->key) == 0)
492 cf_util_get_string(child, &cb->node);
493 else if (strcasecmp("Port", child->key) == 0)
494 cf_util_get_service(child, &cb->service);
495 else if (strcasecmp("Protocol", child->key) == 0) {
496 cf_util_get_string(child, &cb->protocol);
498 if (strcasecmp("UDP", cb->protocol) != 0 &&
499 strcasecmp("TCP", cb->protocol) != 0) {
500 ERROR("write_graphite plugin: Unknown protocol (%s)", cb->protocol);
503 } else if (strcasecmp("ReconnectInterval", child->key) == 0)
504 cf_util_get_cdtime(child, &cb->reconnect_interval);
505 else if (strcasecmp("LogSendErrors", child->key) == 0)
506 cf_util_get_boolean(child, &cb->log_send_errors);
507 else if (strcasecmp("Prefix", child->key) == 0)
508 cf_util_get_string(child, &cb->prefix);
509 else if (strcasecmp("Postfix", child->key) == 0)
510 cf_util_get_string(child, &cb->postfix);
511 else if (strcasecmp("StoreRates", child->key) == 0)
512 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_STORE_RATES);
513 else if (strcasecmp("SeparateInstances", child->key) == 0)
514 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_SEPARATE_INSTANCES);
515 else if (strcasecmp("AlwaysAppendDS", child->key) == 0)
516 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_ALWAYS_APPEND_DS);
517 else if (strcasecmp("PreserveSeparator", child->key) == 0)
518 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_PRESERVE_SEPARATOR);
519 else if (strcasecmp("DropDuplicateFields", child->key) == 0)
520 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_DROP_DUPE_FIELDS);
521 else if (strcasecmp("EscapeCharacter", child->key) == 0)
522 config_set_char(&cb->escape_char, child);
524 ERROR("write_graphite plugin: Invalid configuration "
535 wg_callback_free(cb);
539 /* FIXME: Legacy configuration syntax. */
540 if (cb->name == NULL)
541 snprintf(callback_name, sizeof(callback_name), "write_graphite/%s/%s/%s",
542 cb->node, cb->service, cb->protocol);
544 snprintf(callback_name, sizeof(callback_name), "write_graphite/%s",
547 plugin_register_write(callback_name, wg_write,
549 .data = cb, .free_func = wg_callback_free,
552 plugin_register_flush(callback_name, wg_flush, &(user_data_t){.data = cb});
557 static int wg_config(oconfig_item_t *ci) {
558 for (int i = 0; i < ci->children_num; i++) {
559 oconfig_item_t *child = ci->children + i;
561 if (strcasecmp("Node", child->key) == 0)
562 wg_config_node(child);
563 /* FIXME: Remove this legacy mode in version 6. */
564 else if (strcasecmp("Carbon", child->key) == 0)
565 wg_config_node(child);
567 ERROR("write_graphite plugin: Invalid configuration "
576 void module_register(void) {
577 plugin_register_complex_config("write_graphite", wg_config);