1 /**
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
8 *
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.
12 *
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.
17 *
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
21 *
22 * Authors:
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>
28 *
29 * Based on the write_http plugin.
30 **/
32 /* write_graphite plugin configuation example
33 *
34 * <Plugin write_graphite>
35 * <Carbon>
36 * Host "localhost"
37 * Port "2003"
38 * Protocol "udp"
39 * LogSendErrors true
40 * Prefix "collectd"
41 * </Carbon>
42 * </Plugin>
43 */
45 #include "collectd.h"
47 #include "common.h"
48 #include "plugin.h"
50 #include "utils_complain.h"
51 #include "utils_format_graphite.h"
53 #include <netdb.h>
55 #ifndef WG_DEFAULT_NODE
56 #define WG_DEFAULT_NODE "localhost"
57 #endif
59 #ifndef WG_DEFAULT_SERVICE
60 #define WG_DEFAULT_SERVICE "2003"
61 #endif
63 #ifndef WG_DEFAULT_PROTOCOL
64 #define WG_DEFAULT_PROTOCOL "tcp"
65 #endif
67 #ifndef WG_DEFAULT_LOG_SEND_ERRORS
68 #define WG_DEFAULT_LOG_SEND_ERRORS 1
69 #endif
71 #ifndef WG_DEFAULT_ESCAPE
72 #define WG_DEFAULT_ESCAPE '_'
73 #endif
75 /* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */
76 #ifndef WG_SEND_BUF_SIZE
77 #define WG_SEND_BUF_SIZE 1428
78 #endif
80 #ifndef WG_MIN_RECONNECT_INTERVAL
81 #define WG_MIN_RECONNECT_INTERVAL TIME_T_TO_CDTIME_T(1)
82 #endif
84 /*
85 * Private variables
86 */
87 struct wg_callback {
88 int sock_fd;
90 char *name;
92 char *node;
93 char *service;
94 char *protocol;
95 _Bool log_send_errors;
96 char *prefix;
97 char *postfix;
98 char escape_char;
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;
115 };
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) {
120 cdtime_t now;
122 if (cb->reconnect_interval == 0)
123 return;
125 /* check if address changes if addr_timeout */
126 now = cdtime();
127 if ((now - cb->last_reconnect_time) < cb->reconnect_interval)
128 return;
130 /* here we should close connection on next */
131 close(cb->sock_fd);
132 cb->sock_fd = -1;
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));
138 }
140 /*
141 * Functions
142 */
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();
148 }
150 static int wg_send_buffer(struct wg_callback *cb) {
151 ssize_t status;
153 if (cb->sock_fd < 0)
154 return (-1);
156 status = swrite(cb->sock_fd, cb->send_buf, strlen(cb->send_buf));
157 if (status != 0) {
158 if (cb->log_send_errors) {
159 char errbuf[1024];
160 ERROR("write_graphite plugin: send to %s:%s (%s) failed with status %zi "
161 "(%s)",
162 cb->node, cb->service, cb->protocol, status,
163 sstrerror(errno, errbuf, sizeof(errbuf)));
164 }
166 close(cb->sock_fd);
167 cb->sock_fd = -1;
169 return (-1);
170 }
172 return (0);
173 }
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) {
177 int status;
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 */
184 if (timeout > 0) {
185 cdtime_t now;
187 now = cdtime();
188 if ((cb->send_buf_init_time + timeout) > now)
189 return (0);
190 }
192 if (cb->send_buf_fill == 0) {
193 cb->send_buf_init_time = cdtime();
194 return (0);
195 }
197 status = wg_send_buffer(cb);
198 wg_reset_buffer(cb);
200 return (status);
201 }
203 static int wg_callback_init(struct wg_callback *cb) {
204 struct addrinfo *ai_list;
205 cdtime_t now;
206 int status;
208 char connerr[1024] = "";
210 if (cb->sock_fd > 0)
211 return (0);
213 /* Don't try to reconnect too often. By default, one reconnection attempt
214 * is made per second. */
215 now = cdtime();
216 if ((now - cb->last_connect_time) < WG_MIN_RECONNECT_INTERVAL)
217 return (EAGAIN);
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;
225 else
226 ai_hints.ai_socktype = SOCK_DGRAM;
228 status = getaddrinfo(cb->node, cb->service, &ai_hints, &ai_list);
229 if (status != 0) {
230 ERROR("write_graphite plugin: getaddrinfo (%s, %s, %s) failed: %s",
231 cb->node, cb->service, cb->protocol, gai_strerror(status));
232 return (-1);
233 }
235 assert(ai_list != NULL);
236 for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
237 ai_ptr = ai_ptr->ai_next) {
238 cb->sock_fd =
239 socket(ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
240 if (cb->sock_fd < 0) {
241 char errbuf[1024];
242 snprintf(connerr, sizeof(connerr), "failed to open socket: %s",
243 sstrerror(errno, errbuf, sizeof(errbuf)));
244 continue;
245 }
247 set_sock_opts(cb->sock_fd);
249 status = connect(cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
250 if (status != 0) {
251 char errbuf[1024];
252 snprintf(connerr, sizeof(connerr), "failed to connect to remote "
253 "host: %s",
254 sstrerror(errno, errbuf, sizeof(errbuf)));
255 close(cb->sock_fd);
256 cb->sock_fd = -1;
257 continue;
258 }
260 break;
261 }
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);
273 return (-1);
274 } else {
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);
278 }
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))
284 wg_reset_buffer(cb);
285 else
286 cb->reconnect_interval_reached = 0;
288 return (0);
289 }
291 static void wg_callback_free(void *data) {
292 struct wg_callback *cb;
294 if (data == NULL)
295 return;
297 cb = data;
299 pthread_mutex_lock(&cb->send_lock);
301 wg_flush_nolock(/* timeout = */ 0, cb);
303 if (cb->sock_fd >= 0) {
304 close(cb->sock_fd);
305 cb->sock_fd = -1;
306 }
308 sfree(cb->name);
309 sfree(cb->node);
310 sfree(cb->protocol);
311 sfree(cb->service);
312 sfree(cb->prefix);
313 sfree(cb->postfix);
315 pthread_mutex_destroy(&cb->send_lock);
317 sfree(cb);
318 }
320 static int wg_flush(cdtime_t timeout,
321 const char *identifier __attribute__((unused)),
322 user_data_t *user_data) {
323 struct wg_callback *cb;
324 int status;
326 if (user_data == NULL)
327 return (-EINVAL);
329 cb = user_data->data;
331 pthread_mutex_lock(&cb->send_lock);
333 if (cb->sock_fd < 0) {
334 status = wg_callback_init(cb);
335 if (status != 0) {
336 /* An error message has already been printed. */
337 pthread_mutex_unlock(&cb->send_lock);
338 return (-1);
339 }
340 }
342 status = wg_flush_nolock(timeout, cb);
343 pthread_mutex_unlock(&cb->send_lock);
345 return (status);
346 }
348 static int wg_send_message(char const *message, struct wg_callback *cb) {
349 int status;
350 size_t message_len;
352 message_len = strlen(message);
354 pthread_mutex_lock(&cb->send_lock);
356 wg_force_reconnect_check(cb);
358 if (cb->sock_fd < 0) {
359 status = wg_callback_init(cb);
360 if (status != 0) {
361 /* An error message has already been printed. */
362 pthread_mutex_unlock(&cb->send_lock);
363 return (-1);
364 }
365 }
367 if (message_len >= cb->send_buf_free) {
368 status = wg_flush_nolock(/* timeout = */ 0, cb);
369 if (status != 0) {
370 pthread_mutex_unlock(&cb->send_lock);
371 return (status);
372 }
373 }
375 /* Assert that we have enough space for this message. */
376 assert(message_len < cb->send_buf_free);
378 /* `message_len + 1' because `message_len' does not include the
379 * trailing null byte. Neither does `send_buffer_fill'. */
380 memcpy(cb->send_buf + cb->send_buf_fill, message, message_len + 1);
381 cb->send_buf_fill += message_len;
382 cb->send_buf_free -= message_len;
384 DEBUG("write_graphite plugin: [%s]:%s (%s) buf %zu/%zu (%.1f %%) \"%s\"",
385 cb->node, cb->service, cb->protocol, cb->send_buf_fill,
386 sizeof(cb->send_buf),
387 100.0 * ((double)cb->send_buf_fill) / ((double)sizeof(cb->send_buf)),
388 message);
390 pthread_mutex_unlock(&cb->send_lock);
392 return (0);
393 }
395 static int wg_write_messages(const data_set_t *ds, const value_list_t *vl,
396 struct wg_callback *cb) {
397 char buffer[WG_SEND_BUF_SIZE] = {0};
398 int status;
400 if (0 != strcmp(ds->type, vl->type)) {
401 ERROR("write_graphite plugin: DS type does not match "
402 "value list type");
403 return -1;
404 }
406 status = format_graphite(buffer, sizeof(buffer), ds, vl, cb->prefix,
407 cb->postfix, cb->escape_char, cb->format_flags);
408 if (status != 0) /* error message has been printed already. */
409 return (status);
411 /* Send the message to graphite */
412 status = wg_send_message(buffer, cb);
413 if (status != 0) /* error message has been printed already. */
414 return (status);
416 return (0);
417 } /* int wg_write_messages */
419 static int wg_write(const data_set_t *ds, const value_list_t *vl,
420 user_data_t *user_data) {
421 struct wg_callback *cb;
422 int status;
424 if (user_data == NULL)
425 return (EINVAL);
427 cb = user_data->data;
429 status = wg_write_messages(ds, vl, cb);
431 return (status);
432 }
434 static int config_set_char(char *dest, oconfig_item_t *ci) {
435 char buffer[4] = {0};
436 int status;
438 status = cf_util_get_string_buffer(ci, buffer, sizeof(buffer));
439 if (status != 0)
440 return (status);
442 if (buffer[0] == 0) {
443 ERROR("write_graphite plugin: Cannot use an empty string for the "
444 "\"EscapeCharacter\" option.");
445 return (-1);
446 }
448 if (buffer[1] != 0) {
449 WARNING("write_graphite plugin: Only the first character of the "
450 "\"EscapeCharacter\" option ('%c') will be used.",
451 (int)buffer[0]);
452 }
454 *dest = buffer[0];
456 return (0);
457 }
459 static int wg_config_node(oconfig_item_t *ci) {
460 struct wg_callback *cb;
461 char callback_name[DATA_MAX_NAME_LEN];
462 int status = 0;
464 cb = calloc(1, sizeof(*cb));
465 if (cb == NULL) {
466 ERROR("write_graphite plugin: calloc failed.");
467 return (-1);
468 }
469 cb->sock_fd = -1;
470 cb->name = NULL;
471 cb->node = strdup(WG_DEFAULT_NODE);
472 cb->service = strdup(WG_DEFAULT_SERVICE);
473 cb->protocol = strdup(WG_DEFAULT_PROTOCOL);
474 cb->last_reconnect_time = cdtime();
475 cb->reconnect_interval = 0;
476 cb->reconnect_interval_reached = 0;
477 cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
478 cb->prefix = NULL;
479 cb->postfix = NULL;
480 cb->escape_char = WG_DEFAULT_ESCAPE;
481 cb->format_flags = GRAPHITE_STORE_RATES;
483 /* FIXME: Legacy configuration syntax. */
484 if (strcasecmp("Carbon", ci->key) != 0) {
485 status = cf_util_get_string(ci, &cb->name);
486 if (status != 0) {
487 wg_callback_free(cb);
488 return (status);
489 }
490 }
492 pthread_mutex_init(&cb->send_lock, /* attr = */ NULL);
493 C_COMPLAIN_INIT(&cb->init_complaint);
495 for (int i = 0; i < ci->children_num; i++) {
496 oconfig_item_t *child = ci->children + i;
498 if (strcasecmp("Host", child->key) == 0)
499 cf_util_get_string(child, &cb->node);
500 else if (strcasecmp("Port", child->key) == 0)
501 cf_util_get_service(child, &cb->service);
502 else if (strcasecmp("Protocol", child->key) == 0) {
503 cf_util_get_string(child, &cb->protocol);
505 if (strcasecmp("UDP", cb->protocol) != 0 &&
506 strcasecmp("TCP", cb->protocol) != 0) {
507 ERROR("write_graphite plugin: Unknown protocol (%s)", cb->protocol);
508 status = -1;
509 }
510 } else if (strcasecmp("ReconnectInterval", child->key) == 0)
511 cf_util_get_cdtime(child, &cb->reconnect_interval);
512 else if (strcasecmp("LogSendErrors", child->key) == 0)
513 cf_util_get_boolean(child, &cb->log_send_errors);
514 else if (strcasecmp("Prefix", child->key) == 0)
515 cf_util_get_string(child, &cb->prefix);
516 else if (strcasecmp("Postfix", child->key) == 0)
517 cf_util_get_string(child, &cb->postfix);
518 else if (strcasecmp("StoreRates", child->key) == 0)
519 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_STORE_RATES);
520 else if (strcasecmp("SeparateInstances", child->key) == 0)
521 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_SEPARATE_INSTANCES);
522 else if (strcasecmp("AlwaysAppendDS", child->key) == 0)
523 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_ALWAYS_APPEND_DS);
524 else if (strcasecmp("PreserveSeparator", child->key) == 0)
525 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_PRESERVE_SEPARATOR);
526 else if (strcasecmp("DropDuplicateFields", child->key) == 0)
527 cf_util_get_flag(child, &cb->format_flags, GRAPHITE_DROP_DUPE_FIELDS);
528 else if (strcasecmp("EscapeCharacter", child->key) == 0)
529 config_set_char(&cb->escape_char, child);
530 else {
531 ERROR("write_graphite plugin: Invalid configuration "
532 "option: %s.",
533 child->key);
534 status = -1;
535 }
537 if (status != 0)
538 break;
539 }
541 if (status != 0) {
542 wg_callback_free(cb);
543 return (status);
544 }
546 /* FIXME: Legacy configuration syntax. */
547 if (cb->name == NULL)
548 ssnprintf(callback_name, sizeof(callback_name), "write_graphite/%s/%s/%s",
549 cb->node, cb->service, cb->protocol);
550 else
551 ssnprintf(callback_name, sizeof(callback_name), "write_graphite/%s",
552 cb->name);
554 plugin_register_write(callback_name, wg_write,
555 &(user_data_t){
556 .data = cb, .free_func = wg_callback_free,
557 });
559 plugin_register_flush(callback_name, wg_flush, &(user_data_t){.data = cb});
561 return (0);
562 }
564 static int wg_config(oconfig_item_t *ci) {
565 for (int i = 0; i < ci->children_num; i++) {
566 oconfig_item_t *child = ci->children + i;
568 if (strcasecmp("Node", child->key) == 0)
569 wg_config_node(child);
570 /* FIXME: Remove this legacy mode in version 6. */
571 else if (strcasecmp("Carbon", child->key) == 0)
572 wg_config_node(child);
573 else {
574 ERROR("write_graphite plugin: Invalid configuration "
575 "option: %s.",
576 child->key);
577 }
578 }
580 return (0);
581 }
583 void module_register(void) {
584 plugin_register_complex_config("write_graphite", wg_config);
585 }