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 }