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