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 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  */
  44
  45 #include "collectd.h"
  46
  47 #include "common.h"
  48 #include "plugin.h"
  49
  50 #include "utils_complain.h"
  51 #include "utils_format_graphite.h"
  52
  53 #include <netdb.h>
  54
  55 #ifndef WG_DEFAULT_NODE
  56 #define WG_DEFAULT_NODE "localhost"
  57 #endif
  58
  59 #ifndef WG_DEFAULT_SERVICE
  60 #define WG_DEFAULT_SERVICE "2003"
  61 #endif
  62
  63 #ifndef WG_DEFAULT_PROTOCOL
  64 #define WG_DEFAULT_PROTOCOL "tcp"
  65 #endif
  66
  67 #ifndef WG_DEFAULT_LOG_SEND_ERRORS
  68 #define WG_DEFAULT_LOG_SEND_ERRORS true
  69 #endif
  70
  71 #ifndef WG_DEFAULT_ESCAPE
  72 #define WG_DEFAULT_ESCAPE '_'
  73 #endif
  74
  75 /* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */
  76 #ifndef WG_SEND_BUF_SIZE
  77 #define WG_SEND_BUF_SIZE 1428
  78 #endif
  79
  80 #ifndef WG_MIN_RECONNECT_INTERVAL
  81 #define WG_MIN_RECONNECT_INTERVAL TIME_T_TO_CDTIME_T(1)
  82 #endif
  83
  84 /*
  85  * Private variables
  86  */
  87 struct wg_callback {
  88   int sock_fd;
  89
  90   char *name;
  91
  92   char *node;
  93   char *service;
  94   char *protocol;
  95   bool log_send_errors;
  96   char *prefix;
  97   char *postfix;
  98   char escape_char;
  99
 100   unsigned int format_flags;
 101
 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;
 106
 107   pthread_mutex_t send_lock;
 108   c_complain_t init_complaint;
 109   cdtime_t last_connect_time;
 110
 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 };
 116
 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;
 121
 122   if (cb->reconnect_interval == 0)
 123     return;
 124
 125   /* check if address changes if addr_timeout */
 126   now = cdtime();
 127   if ((now - cb->last_reconnect_time) < cb->reconnect_interval)
 128     return;
 129
 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 = true;
 135
 136   INFO("write_graphite plugin: Connection closed after %.3f seconds.",
 137        CDTIME_T_TO_DOUBLE(now - cb->last_reconnect_time));
 138 }
 139
 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 }
 149
 150 static int wg_send_buffer(struct wg_callback *cb) {
 151   ssize_t status;
 152
 153   if (cb->sock_fd < 0)
 154     return -1;
 155
 156   status = swrite(cb->sock_fd, cb->send_buf, strlen(cb->send_buf));
 157   if (status != 0) {
 158     if (cb->log_send_errors) {
 159       ERROR("write_graphite plugin: send to %s:%s (%s) failed with status %zi "
 160             "(%s)",
 161             cb->node, cb->service, cb->protocol, status, STRERRNO);
 162     }
 163
 164     close(cb->sock_fd);
 165     cb->sock_fd = -1;
 166
 167     return -1;
 168   }
 169
 170   return 0;
 171 }
 172
 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) {
 175   int status;
 176
 177   DEBUG("write_graphite plugin: wg_flush_nolock: timeout = %.3f; "
 178         "send_buf_fill = %" PRIsz ";",
 179         (double)timeout, cb->send_buf_fill);
 180
 181   /* timeout == 0  => flush unconditionally */
 182   if (timeout > 0) {
 183     cdtime_t now;
 184
 185     now = cdtime();
 186     if ((cb->send_buf_init_time + timeout) > now)
 187       return 0;
 188   }
 189
 190   if (cb->send_buf_fill == 0) {
 191     cb->send_buf_init_time = cdtime();
 192     return 0;
 193   }
 194
 195   status = wg_send_buffer(cb);
 196   wg_reset_buffer(cb);
 197
 198   return status;
 199 }
 200
 201 static int wg_callback_init(struct wg_callback *cb) {
 202   struct addrinfo *ai_list;
 203   cdtime_t now;
 204   int status;
 205
 206   char connerr[1024] = "";
 207
 208   if (cb->sock_fd > 0)
 209     return 0;
 210
 211   /* Don't try to reconnect too often. By default, one reconnection attempt
 212    * is made per second. */
 213   now = cdtime();
 214   if ((now - cb->last_connect_time) < WG_MIN_RECONNECT_INTERVAL)
 215     return EAGAIN;
 216   cb->last_connect_time = now;
 217
 218   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
 219                               .ai_flags = AI_ADDRCONFIG};
 220
 221   if (0 == strcasecmp("tcp", cb->protocol))
 222     ai_hints.ai_socktype = SOCK_STREAM;
 223   else
 224     ai_hints.ai_socktype = SOCK_DGRAM;
 225
 226   status = getaddrinfo(cb->node, cb->service, &ai_hints, &ai_list);
 227   if (status != 0) {
 228     ERROR("write_graphite plugin: getaddrinfo (%s, %s, %s) failed: %s",
 229           cb->node, cb->service, cb->protocol, gai_strerror(status));
 230     return -1;
 231   }
 232
 233   assert(ai_list != NULL);
 234   for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
 235        ai_ptr = ai_ptr->ai_next) {
 236     cb->sock_fd =
 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);
 240       continue;
 241     }
 242
 243     set_sock_opts(cb->sock_fd);
 244
 245     status = connect(cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 246     if (status != 0) {
 247       snprintf(connerr, sizeof(connerr), "failed to connect to remote host: %s",
 248                STRERRNO);
 249       close(cb->sock_fd);
 250       cb->sock_fd = -1;
 251       continue;
 252     }
 253
 254     break;
 255   }
 256
 257   freeaddrinfo(ai_list);
 258
 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);
 264     return -1;
 265   } else {
 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);
 269   }
 270
 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))
 275     wg_reset_buffer(cb);
 276   else
 277     cb->reconnect_interval_reached = false;
 278
 279   return 0;
 280 }
 281
 282 static void wg_callback_free(void *data) {
 283   struct wg_callback *cb;
 284
 285   if (data == NULL)
 286     return;
 287
 288   cb = data;
 289
 290   pthread_mutex_lock(&cb->send_lock);
 291
 292   wg_flush_nolock(/* timeout = */ 0, cb);
 293
 294   if (cb->sock_fd >= 0) {
 295     close(cb->sock_fd);
 296     cb->sock_fd = -1;
 297   }
 298
 299   sfree(cb->name);
 300   sfree(cb->node);
 301   sfree(cb->protocol);
 302   sfree(cb->service);
 303   sfree(cb->prefix);
 304   sfree(cb->postfix);
 305
 306   pthread_mutex_unlock(&cb->send_lock);
 307   pthread_mutex_destroy(&cb->send_lock);
 308
 309   sfree(cb);
 310 }
 311
 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;
 316   int status;
 317
 318   if (user_data == NULL)
 319     return -EINVAL;
 320
 321   cb = user_data->data;
 322
 323   pthread_mutex_lock(&cb->send_lock);
 324
 325   if (cb->sock_fd < 0) {
 326     status = wg_callback_init(cb);
 327     if (status != 0) {
 328       /* An error message has already been printed. */
 329       pthread_mutex_unlock(&cb->send_lock);
 330       return -1;
 331     }
 332   }
 333
 334   status = wg_flush_nolock(timeout, cb);
 335   pthread_mutex_unlock(&cb->send_lock);
 336
 337   return status;
 338 }
 339
 340 static int wg_send_message(char const *message, struct wg_callback *cb) {
 341   int status;
 342   size_t message_len;
 343
 344   message_len = strlen(message);
 345
 346   pthread_mutex_lock(&cb->send_lock);
 347
 348   wg_force_reconnect_check(cb);
 349
 350   if (cb->sock_fd < 0) {
 351     status = wg_callback_init(cb);
 352     if (status != 0) {
 353       /* An error message has already been printed. */
 354       pthread_mutex_unlock(&cb->send_lock);
 355       return -1;
 356     }
 357   }
 358
 359   if (message_len >= cb->send_buf_free) {
 360     status = wg_flush_nolock(/* timeout = */ 0, cb);
 361     if (status != 0) {
 362       pthread_mutex_unlock(&cb->send_lock);
 363       return status;
 364     }
 365   }
 366
 367   /* Assert that we have enough space for this message. */
 368   assert(message_len < cb->send_buf_free);
 369
 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;
 375
 376   DEBUG("write_graphite plugin: [%s]:%s (%s) buf %" PRIsz "/%" PRIsz
 377         " (%.1f %%) \"%s\"",
 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)),
 381         message);
 382
 383   pthread_mutex_unlock(&cb->send_lock);
 384
 385   return 0;
 386 }
 387
 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};
 391   int status;
 392
 393   if (0 != strcmp(ds->type, vl->type)) {
 394     ERROR("write_graphite plugin: DS type does not match "
 395           "value list type");
 396     return -1;
 397   }
 398
 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. */
 402     return status;
 403
 404   /* Send the message to graphite */
 405   status = wg_send_message(buffer, cb);
 406   if (status != 0) /* error message has been printed already. */
 407     return status;
 408
 409   return 0;
 410 } /* int wg_write_messages */
 411
 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;
 415   int status;
 416
 417   if (user_data == NULL)
 418     return EINVAL;
 419
 420   cb = user_data->data;
 421
 422   status = wg_write_messages(ds, vl, cb);
 423
 424   return status;
 425 }
 426
 427 static int config_set_char(char *dest, oconfig_item_t *ci) {
 428   char buffer[4] = {0};
 429   int status;
 430
 431   status = cf_util_get_string_buffer(ci, buffer, sizeof(buffer));
 432   if (status != 0)
 433     return status;
 434
 435   if (buffer[0] == 0) {
 436     ERROR("write_graphite plugin: Cannot use an empty string for the "
 437           "\"EscapeCharacter\" option.");
 438     return -1;
 439   }
 440
 441   if (buffer[1] != 0) {
 442     WARNING("write_graphite plugin: Only the first character of the "
 443             "\"EscapeCharacter\" option ('%c') will be used.",
 444             (int)buffer[0]);
 445   }
 446
 447   *dest = buffer[0];
 448
 449   return 0;
 450 }
 451
 452 static int wg_config_node(oconfig_item_t *ci) {
 453   struct wg_callback *cb;
 454   char callback_name[DATA_MAX_NAME_LEN];
 455   int status = 0;
 456
 457   cb = calloc(1, sizeof(*cb));
 458   if (cb == NULL) {
 459     ERROR("write_graphite plugin: calloc failed.");
 460     return -1;
 461   }
 462   cb->sock_fd = -1;
 463   cb->name = NULL;
 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 = false;
 470   cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
 471   cb->prefix = NULL;
 472   cb->postfix = NULL;
 473   cb->escape_char = WG_DEFAULT_ESCAPE;
 474   cb->format_flags = GRAPHITE_STORE_RATES;
 475
 476   /* FIXME: Legacy configuration syntax. */
 477   if (strcasecmp("Carbon", ci->key) != 0) {
 478     status = cf_util_get_string(ci, &cb->name);
 479     if (status != 0) {
 480       wg_callback_free(cb);
 481       return status;
 482     }
 483   }
 484
 485   pthread_mutex_init(&cb->send_lock, /* attr = */ NULL);
 486   C_COMPLAIN_INIT(&cb->init_complaint);
 487
 488   for (int i = 0; i < ci->children_num; i++) {
 489     oconfig_item_t *child = ci->children + i;
 490
 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);
 497
 498       if (strcasecmp("UDP", cb->protocol) != 0 &&
 499           strcasecmp("TCP", cb->protocol) != 0) {
 500         ERROR("write_graphite plugin: Unknown protocol (%s)", cb->protocol);
 501         status = -1;
 502       }
 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);
 523     else {
 524       ERROR("write_graphite plugin: Invalid configuration "
 525             "option: %s.",
 526             child->key);
 527       status = -1;
 528     }
 529
 530     if (status != 0)
 531       break;
 532   }
 533
 534   if (status != 0) {
 535     wg_callback_free(cb);
 536     return status;
 537   }
 538
 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);
 543   else
 544     snprintf(callback_name, sizeof(callback_name), "write_graphite/%s",
 545              cb->name);
 546
 547   plugin_register_write(callback_name, wg_write,
 548                         &(user_data_t){
 549                             .data = cb, .free_func = wg_callback_free,
 550                         });
 551
 552   plugin_register_flush(callback_name, wg_flush, &(user_data_t){.data = cb});
 553
 554   return 0;
 555 }
 556
 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;
 560
 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);
 566     else {
 567       ERROR("write_graphite plugin: Invalid configuration "
 568             "option: %s.",
 569             child->key);
 570     }
 571   }
 572
 573   return 0;
 574 }
 575
 576 void module_register(void) {
 577   plugin_register_complex_config("write_graphite", wg_config);
 578 }