Remove grpc from Mac builds for now
[collectd.git] / src / write_graphite.c
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  **/
31
32 /* write_graphite plugin configuration example
33  *
34  * <Plugin write_graphite>
35  *   <Carbon>
36  *     Host "localhost"
37  *     Port "2003"
38  *     Protocol "udp"
39  *     LogSendErrors true
40  *     Prefix "collectd"
41  *     UseTags true
42  *     ReverseHost false
43  *   </Carbon>
44  * </Plugin>
45  */
46
47 #include "collectd.h"
48
49 #include "plugin.h"
50 #include "utils/common/common.h"
51
52 #include "utils/format_graphite/format_graphite.h"
53 #include "utils_complain.h"
54
55 #include <netdb.h>
56
57 #ifndef WG_DEFAULT_NODE
58 #define WG_DEFAULT_NODE "localhost"
59 #endif
60
61 #ifndef WG_DEFAULT_SERVICE
62 #define WG_DEFAULT_SERVICE "2003"
63 #endif
64
65 #ifndef WG_DEFAULT_PROTOCOL
66 #define WG_DEFAULT_PROTOCOL "tcp"
67 #endif
68
69 #ifndef WG_DEFAULT_LOG_SEND_ERRORS
70 #define WG_DEFAULT_LOG_SEND_ERRORS true
71 #endif
72
73 #ifndef WG_DEFAULT_ESCAPE
74 #define WG_DEFAULT_ESCAPE '_'
75 #endif
76
77 /* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */
78 #ifndef WG_SEND_BUF_SIZE
79 #define WG_SEND_BUF_SIZE 1428
80 #endif
81
82 #ifndef WG_MIN_RECONNECT_INTERVAL
83 #define WG_MIN_RECONNECT_INTERVAL TIME_T_TO_CDTIME_T(1)
84 #endif
85
86 /*
87  * Private variables
88  */
89 struct wg_callback {
90   int sock_fd;
91
92   char *name;
93
94   char *node;
95   char *service;
96   char *protocol;
97   bool log_send_errors;
98   char *prefix;
99   char *postfix;
100   char escape_char;
101
102   unsigned int format_flags;
103
104   char send_buf[WG_SEND_BUF_SIZE];
105   size_t send_buf_free;
106   size_t send_buf_fill;
107   cdtime_t send_buf_init_time;
108
109   pthread_mutex_t send_lock;
110   c_complain_t init_complaint;
111   cdtime_t last_connect_time;
112
113   /* Force reconnect useful for load balanced environments */
114   cdtime_t last_reconnect_time;
115   cdtime_t reconnect_interval;
116   bool reconnect_interval_reached;
117 };
118
119 /* wg_force_reconnect_check closes cb->sock_fd when it was open for longer
120  * than cb->reconnect_interval. Must hold cb->send_lock when calling. */
121 static void wg_force_reconnect_check(struct wg_callback *cb) {
122   cdtime_t now;
123
124   if (cb->reconnect_interval == 0)
125     return;
126
127   /* check if address changes if addr_timeout */
128   now = cdtime();
129   if ((now - cb->last_reconnect_time) < cb->reconnect_interval)
130     return;
131
132   /* here we should close connection on next */
133   close(cb->sock_fd);
134   cb->sock_fd = -1;
135   cb->last_reconnect_time = now;
136   cb->reconnect_interval_reached = true;
137
138   INFO("write_graphite plugin: Connection closed after %.3f seconds.",
139        CDTIME_T_TO_DOUBLE(now - cb->last_reconnect_time));
140 }
141
142 /*
143  * Functions
144  */
145 static void wg_reset_buffer(struct wg_callback *cb) {
146   memset(cb->send_buf, 0, sizeof(cb->send_buf));
147   cb->send_buf_free = sizeof(cb->send_buf);
148   cb->send_buf_fill = 0;
149   cb->send_buf_init_time = cdtime();
150 }
151
152 static int wg_send_buffer(struct wg_callback *cb) {
153   ssize_t status;
154
155   if (cb->sock_fd < 0)
156     return -1;
157
158   status = swrite(cb->sock_fd, cb->send_buf, strlen(cb->send_buf));
159   if (status != 0) {
160     if (cb->log_send_errors) {
161       ERROR("write_graphite plugin: send to %s:%s (%s) failed with status %zi "
162             "(%s)",
163             cb->node, cb->service, cb->protocol, status, STRERRNO);
164     }
165
166     close(cb->sock_fd);
167     cb->sock_fd = -1;
168
169     return -1;
170   }
171
172   return 0;
173 }
174
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;
178
179   DEBUG("write_graphite plugin: wg_flush_nolock: timeout = %.3f; "
180         "send_buf_fill = %" PRIsz ";",
181         (double)timeout, cb->send_buf_fill);
182
183   /* timeout == 0  => flush unconditionally */
184   if (timeout > 0) {
185     cdtime_t now;
186
187     now = cdtime();
188     if ((cb->send_buf_init_time + timeout) > now)
189       return 0;
190   }
191
192   if (cb->send_buf_fill == 0) {
193     cb->send_buf_init_time = cdtime();
194     return 0;
195   }
196
197   status = wg_send_buffer(cb);
198   wg_reset_buffer(cb);
199
200   return status;
201 }
202
203 static int wg_callback_init(struct wg_callback *cb) {
204   struct addrinfo *ai_list;
205   cdtime_t now;
206   int status;
207
208   char connerr[1024] = "";
209
210   if (cb->sock_fd > 0)
211     return 0;
212
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;
219
220   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
221                               .ai_flags = AI_ADDRCONFIG};
222
223   if (0 == strcasecmp("tcp", cb->protocol))
224     ai_hints.ai_socktype = SOCK_STREAM;
225   else
226     ai_hints.ai_socktype = SOCK_DGRAM;
227
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   }
234
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       snprintf(connerr, sizeof(connerr), "failed to open socket: %s", STRERRNO);
242       continue;
243     }
244
245     set_sock_opts(cb->sock_fd);
246
247     status = connect(cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
248     if (status != 0) {
249       snprintf(connerr, sizeof(connerr), "failed to connect to remote host: %s",
250                STRERRNO);
251       close(cb->sock_fd);
252       cb->sock_fd = -1;
253       continue;
254     }
255
256     break;
257   }
258
259   freeaddrinfo(ai_list);
260
261   if (cb->sock_fd < 0) {
262     c_complain(LOG_ERR, &cb->init_complaint,
263                "write_graphite plugin: Connecting to %s:%s via %s failed. "
264                "The last error was: %s",
265                cb->node, cb->service, cb->protocol, connerr);
266     return -1;
267   } else {
268     c_release(LOG_INFO, &cb->init_complaint,
269               "write_graphite plugin: Successfully connected to %s:%s via %s.",
270               cb->node, cb->service, cb->protocol);
271   }
272
273   /* wg_force_reconnect_check does not flush the buffer before closing a
274    * sending socket, so only call wg_reset_buffer() if the socket was closed
275    * for a different reason (tracked in cb->reconnect_interval_reached). */
276   if (!cb->reconnect_interval_reached || (cb->send_buf_free == 0))
277     wg_reset_buffer(cb);
278   else
279     cb->reconnect_interval_reached = false;
280
281   return 0;
282 }
283
284 static void wg_callback_free(void *data) {
285   struct wg_callback *cb;
286
287   if (data == NULL)
288     return;
289
290   cb = data;
291
292   pthread_mutex_lock(&cb->send_lock);
293
294   wg_flush_nolock(/* timeout = */ 0, cb);
295
296   if (cb->sock_fd >= 0) {
297     close(cb->sock_fd);
298     cb->sock_fd = -1;
299   }
300
301   sfree(cb->name);
302   sfree(cb->node);
303   sfree(cb->protocol);
304   sfree(cb->service);
305   sfree(cb->prefix);
306   sfree(cb->postfix);
307
308   pthread_mutex_unlock(&cb->send_lock);
309   pthread_mutex_destroy(&cb->send_lock);
310
311   sfree(cb);
312 }
313
314 static int wg_flush(cdtime_t timeout,
315                     const char *identifier __attribute__((unused)),
316                     user_data_t *user_data) {
317   struct wg_callback *cb;
318   int status;
319
320   if (user_data == NULL)
321     return -EINVAL;
322
323   cb = user_data->data;
324
325   pthread_mutex_lock(&cb->send_lock);
326
327   if (cb->sock_fd < 0) {
328     status = wg_callback_init(cb);
329     if (status != 0) {
330       /* An error message has already been printed. */
331       pthread_mutex_unlock(&cb->send_lock);
332       return -1;
333     }
334   }
335
336   status = wg_flush_nolock(timeout, cb);
337   pthread_mutex_unlock(&cb->send_lock);
338
339   return status;
340 }
341
342 static int wg_send_message(char const *message, struct wg_callback *cb) {
343   int status;
344   size_t message_len;
345
346   message_len = strlen(message);
347
348   pthread_mutex_lock(&cb->send_lock);
349
350   wg_force_reconnect_check(cb);
351
352   if (cb->sock_fd < 0) {
353     status = wg_callback_init(cb);
354     if (status != 0) {
355       /* An error message has already been printed. */
356       pthread_mutex_unlock(&cb->send_lock);
357       return -1;
358     }
359   }
360
361   if (message_len >= cb->send_buf_free) {
362     status = wg_flush_nolock(/* timeout = */ 0, cb);
363     if (status != 0) {
364       pthread_mutex_unlock(&cb->send_lock);
365       return status;
366     }
367   }
368
369   /* Assert that we have enough space for this message. */
370   assert(message_len < cb->send_buf_free);
371
372   /* `message_len + 1' because `message_len' does not include the
373    * trailing null byte. Neither does `send_buffer_fill'. */
374   memcpy(cb->send_buf + cb->send_buf_fill, message, message_len + 1);
375   cb->send_buf_fill += message_len;
376   cb->send_buf_free -= message_len;
377
378   DEBUG("write_graphite plugin: [%s]:%s (%s) buf %" PRIsz "/%" PRIsz
379         " (%.1f %%) \"%s\"",
380         cb->node, cb->service, cb->protocol, cb->send_buf_fill,
381         sizeof(cb->send_buf),
382         100.0 * ((double)cb->send_buf_fill) / ((double)sizeof(cb->send_buf)),
383         message);
384
385   pthread_mutex_unlock(&cb->send_lock);
386
387   return 0;
388 }
389
390 static int wg_write_messages(const data_set_t *ds, const value_list_t *vl,
391                              struct wg_callback *cb) {
392   char buffer[WG_SEND_BUF_SIZE] = {0};
393   int status;
394
395   if (0 != strcmp(ds->type, vl->type)) {
396     ERROR("write_graphite plugin: DS type does not match "
397           "value list type");
398     return -1;
399   }
400
401   status = format_graphite(buffer, sizeof(buffer), ds, vl, cb->prefix,
402                            cb->postfix, cb->escape_char, cb->format_flags);
403   if (status != 0) /* error message has been printed already. */
404     return status;
405
406   /* Send the message to graphite */
407   status = wg_send_message(buffer, cb);
408   if (status != 0) /* error message has been printed already. */
409     return status;
410
411   return 0;
412 } /* int wg_write_messages */
413
414 static int wg_write(const data_set_t *ds, const value_list_t *vl,
415                     user_data_t *user_data) {
416   struct wg_callback *cb;
417   int status;
418
419   if (user_data == NULL)
420     return EINVAL;
421
422   cb = user_data->data;
423
424   status = wg_write_messages(ds, vl, cb);
425
426   return status;
427 }
428
429 static int config_set_char(char *dest, oconfig_item_t *ci) {
430   char buffer[4] = {0};
431   int status;
432
433   status = cf_util_get_string_buffer(ci, buffer, sizeof(buffer));
434   if (status != 0)
435     return status;
436
437   if (buffer[0] == 0) {
438     ERROR("write_graphite plugin: Cannot use an empty string for the "
439           "\"EscapeCharacter\" option.");
440     return -1;
441   }
442
443   if (buffer[1] != 0) {
444     WARNING("write_graphite plugin: Only the first character of the "
445             "\"EscapeCharacter\" option ('%c') will be used.",
446             (int)buffer[0]);
447   }
448
449   *dest = buffer[0];
450
451   return 0;
452 }
453
454 static int wg_config_node(oconfig_item_t *ci) {
455   struct wg_callback *cb;
456   char callback_name[DATA_MAX_NAME_LEN];
457   int status = 0;
458
459   cb = calloc(1, sizeof(*cb));
460   if (cb == NULL) {
461     ERROR("write_graphite plugin: calloc failed.");
462     return -1;
463   }
464   cb->sock_fd = -1;
465   cb->name = NULL;
466   cb->node = strdup(WG_DEFAULT_NODE);
467   cb->service = strdup(WG_DEFAULT_SERVICE);
468   cb->protocol = strdup(WG_DEFAULT_PROTOCOL);
469   cb->last_reconnect_time = cdtime();
470   cb->reconnect_interval = 0;
471   cb->reconnect_interval_reached = false;
472   cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
473   cb->prefix = NULL;
474   cb->postfix = NULL;
475   cb->escape_char = WG_DEFAULT_ESCAPE;
476   cb->format_flags = GRAPHITE_STORE_RATES;
477
478   /* FIXME: Legacy configuration syntax. */
479   if (strcasecmp("Carbon", ci->key) != 0) {
480     status = cf_util_get_string(ci, &cb->name);
481     if (status != 0) {
482       wg_callback_free(cb);
483       return status;
484     }
485   }
486
487   pthread_mutex_init(&cb->send_lock, /* attr = */ NULL);
488   C_COMPLAIN_INIT(&cb->init_complaint);
489
490   for (int i = 0; i < ci->children_num; i++) {
491     oconfig_item_t *child = ci->children + i;
492
493     if (strcasecmp("Host", child->key) == 0)
494       cf_util_get_string(child, &cb->node);
495     else if (strcasecmp("Port", child->key) == 0)
496       cf_util_get_service(child, &cb->service);
497     else if (strcasecmp("Protocol", child->key) == 0) {
498       cf_util_get_string(child, &cb->protocol);
499
500       if (strcasecmp("UDP", cb->protocol) != 0 &&
501           strcasecmp("TCP", cb->protocol) != 0) {
502         ERROR("write_graphite plugin: Unknown protocol (%s)", cb->protocol);
503         status = -1;
504       }
505     } else if (strcasecmp("ReconnectInterval", child->key) == 0)
506       cf_util_get_cdtime(child, &cb->reconnect_interval);
507     else if (strcasecmp("LogSendErrors", child->key) == 0)
508       cf_util_get_boolean(child, &cb->log_send_errors);
509     else if (strcasecmp("Prefix", child->key) == 0)
510       cf_util_get_string(child, &cb->prefix);
511     else if (strcasecmp("Postfix", child->key) == 0)
512       cf_util_get_string(child, &cb->postfix);
513     else if (strcasecmp("StoreRates", child->key) == 0)
514       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_STORE_RATES);
515     else if (strcasecmp("SeparateInstances", child->key) == 0)
516       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_SEPARATE_INSTANCES);
517     else if (strcasecmp("AlwaysAppendDS", child->key) == 0)
518       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_ALWAYS_APPEND_DS);
519     else if (strcasecmp("PreserveSeparator", child->key) == 0)
520       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_PRESERVE_SEPARATOR);
521     else if (strcasecmp("DropDuplicateFields", child->key) == 0)
522       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_DROP_DUPE_FIELDS);
523     else if (strcasecmp("UseTags", child->key) == 0)
524       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_USE_TAGS);
525     else if (strcasecmp("ReverseHost", child->key) == 0)
526       cf_util_get_flag(child, &cb->format_flags, GRAPHITE_REVERSE_HOST);
527     else if (strcasecmp("EscapeCharacter", child->key) == 0)
528       config_set_char(&cb->escape_char, child);
529     else {
530       ERROR("write_graphite plugin: Invalid configuration "
531             "option: %s.",
532             child->key);
533       status = -1;
534     }
535
536     if (status != 0)
537       break;
538   }
539
540   if (status != 0) {
541     wg_callback_free(cb);
542     return status;
543   }
544
545   /* FIXME: Legacy configuration syntax. */
546   if (cb->name == NULL)
547     snprintf(callback_name, sizeof(callback_name), "write_graphite/%s/%s/%s",
548              cb->node, cb->service, cb->protocol);
549   else
550     snprintf(callback_name, sizeof(callback_name), "write_graphite/%s",
551              cb->name);
552
553   plugin_register_write(callback_name, wg_write,
554                         &(user_data_t){
555                             .data = cb,
556                             .free_func = wg_callback_free,
557                         });
558
559   plugin_register_flush(callback_name, wg_flush, &(user_data_t){.data = cb});
560
561   return 0;
562 }
563
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;
567
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   }
579
580   return 0;
581 }
582
583 void module_register(void) {
584   plugin_register_complex_config("write_graphite", wg_config);
585 }