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-2012  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   *     Prefix "collectd"
  39   *   </Carbon>
  40   * </Plugin>
  41   */
  42
  43 #include "collectd.h"
  44 #include "common.h"
  45 #include "plugin.h"
  46 #include "configfile.h"
  47
  48 #include "utils_cache.h"
  49 #include "utils_parse_option.h"
  50
  51 /* Folks without pthread will need to disable this plugin. */
  52 #include <pthread.h>
  53
  54 #include <sys/socket.h>
  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_ESCAPE
  66 # define WG_DEFAULT_ESCAPE '_'
  67 #endif
  68
  69 /* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */
  70 #ifndef WG_SEND_BUF_SIZE
  71 # define WG_SEND_BUF_SIZE 1428
  72 #endif
  73
  74 /*
  75  * Private variables
  76  */
  77 struct wg_callback
  78 {
  79     int      sock_fd;
  80
  81     char    *node;
  82     char    *service;
  83     char    *prefix;
  84     char    *postfix;
  85     char     escape_char;
  86
  87     _Bool    store_rates;
  88     _Bool    separate_instances;
  89     _Bool    always_append_ds;
  90
  91     char     send_buf[WG_SEND_BUF_SIZE];
  92     size_t   send_buf_free;
  93     size_t   send_buf_fill;
  94     cdtime_t send_buf_init_time;
  95
  96     pthread_mutex_t send_lock;
  97 };
  98
  99
 100 /*
 101  * Functions
 102  */
 103 static void wg_reset_buffer (struct wg_callback *cb)
 104 {
 105     memset (cb->send_buf, 0, sizeof (cb->send_buf));
 106     cb->send_buf_free = sizeof (cb->send_buf);
 107     cb->send_buf_fill = 0;
 108     cb->send_buf_init_time = cdtime ();
 109 }
 110
 111 static int wg_send_buffer (struct wg_callback *cb)
 112 {
 113     ssize_t status = 0;
 114
 115     status = swrite (cb->sock_fd, cb->send_buf, strlen (cb->send_buf));
 116     if (status < 0)
 117     {
 118         char errbuf[1024];
 119         ERROR ("write_graphite plugin: send failed with status %zi (%s)",
 120                 status, sstrerror (errno, errbuf, sizeof (errbuf)));
 121
 122
 123         close (cb->sock_fd);
 124         cb->sock_fd = -1;
 125
 126         return (-1);
 127     }
 128
 129     return (0);
 130 }
 131
 132 /* NOTE: You must hold cb->send_lock when calling this function! */
 133 static int wg_flush_nolock (cdtime_t timeout, struct wg_callback *cb)
 134 {
 135     int status;
 136
 137     DEBUG ("write_graphite plugin: wg_flush_nolock: timeout = %.3f; "
 138             "send_buf_fill = %zu;",
 139             (double)timeout,
 140             cb->send_buf_fill);
 141
 142     /* timeout == 0  => flush unconditionally */
 143     if (timeout > 0)
 144     {
 145         cdtime_t now;
 146
 147         now = cdtime ();
 148         if ((cb->send_buf_init_time + timeout) > now)
 149             return (0);
 150     }
 151
 152     if (cb->send_buf_fill <= 0)
 153     {
 154         cb->send_buf_init_time = cdtime ();
 155         return (0);
 156     }
 157
 158     status = wg_send_buffer (cb);
 159     wg_reset_buffer (cb);
 160
 161     return (status);
 162 }
 163
 164 static int wg_callback_init (struct wg_callback *cb)
 165 {
 166     struct addrinfo ai_hints;
 167     struct addrinfo *ai_list;
 168     struct addrinfo *ai_ptr;
 169     int status;
 170
 171     const char *node = cb->node ? cb->node : WG_DEFAULT_NODE;
 172     const char *service = cb->service ? cb->service : WG_DEFAULT_SERVICE;
 173
 174     if (cb->sock_fd > 0)
 175         return (0);
 176
 177     memset (&ai_hints, 0, sizeof (ai_hints));
 178 #ifdef AI_ADDRCONFIG
 179     ai_hints.ai_flags |= AI_ADDRCONFIG;
 180 #endif
 181     ai_hints.ai_family = AF_UNSPEC;
 182     ai_hints.ai_socktype = SOCK_STREAM;
 183
 184     ai_list = NULL;
 185
 186     status = getaddrinfo (node, service, &ai_hints, &ai_list);
 187     if (status != 0)
 188     {
 189         ERROR ("write_graphite plugin: getaddrinfo (%s, %s) failed: %s",
 190                 node, service, gai_strerror (status));
 191         return (-1);
 192     }
 193
 194     assert (ai_list != NULL);
 195     for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 196     {
 197         cb->sock_fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype,
 198                 ai_ptr->ai_protocol);
 199         if (cb->sock_fd < 0)
 200             continue;
 201
 202         status = connect (cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 203         if (status != 0)
 204         {
 205             close (cb->sock_fd);
 206             cb->sock_fd = -1;
 207             continue;
 208         }
 209
 210         break;
 211     }
 212
 213     freeaddrinfo (ai_list);
 214
 215     if (cb->sock_fd < 0)
 216     {
 217         char errbuf[1024];
 218         ERROR ("write_graphite plugin: Connecting to %s:%s failed. "
 219                 "The last error was: %s", node, service,
 220                 sstrerror (errno, errbuf, sizeof (errbuf)));
 221         close (cb->sock_fd);
 222         return (-1);
 223     }
 224
 225     wg_reset_buffer (cb);
 226
 227     return (0);
 228 }
 229
 230 static void wg_callback_free (void *data)
 231 {
 232     struct wg_callback *cb;
 233
 234     if (data == NULL)
 235         return;
 236
 237     cb = data;
 238
 239     pthread_mutex_lock (&cb->send_lock);
 240
 241     wg_flush_nolock (/* timeout = */ 0, cb);
 242
 243     close(cb->sock_fd);
 244     cb->sock_fd = -1;
 245
 246     sfree(cb->node);
 247     sfree(cb->service);
 248     sfree(cb->prefix);
 249     sfree(cb->postfix);
 250
 251     pthread_mutex_destroy (&cb->send_lock);
 252
 253     sfree(cb);
 254 }
 255
 256 static int wg_flush (cdtime_t timeout,
 257         const char *identifier __attribute__((unused)),
 258         user_data_t *user_data)
 259 {
 260     struct wg_callback *cb;
 261     int status;
 262
 263     if (user_data == NULL)
 264         return (-EINVAL);
 265
 266     cb = user_data->data;
 267
 268     pthread_mutex_lock (&cb->send_lock);
 269
 270     if (cb->sock_fd < 0)
 271     {
 272         status = wg_callback_init (cb);
 273         if (status != 0)
 274         {
 275             ERROR ("write_graphite plugin: wg_callback_init failed.");
 276             pthread_mutex_unlock (&cb->send_lock);
 277             return (-1);
 278         }
 279     }
 280
 281     status = wg_flush_nolock (timeout, cb);
 282     pthread_mutex_unlock (&cb->send_lock);
 283
 284     return (status);
 285 }
 286
 287 static int wg_format_values (char *ret, size_t ret_len,
 288         int ds_num, const data_set_t *ds, const value_list_t *vl,
 289         _Bool store_rates)
 290 {
 291     size_t offset = 0;
 292     int status;
 293     gauge_t *rates = NULL;
 294
 295     assert (0 == strcmp (ds->type, vl->type));
 296
 297     memset (ret, 0, ret_len);
 298
 299 #define BUFFER_ADD(...) do { \
 300     status = ssnprintf (ret + offset, ret_len - offset, \
 301             __VA_ARGS__); \
 302     if (status < 1) \
 303     { \
 304         sfree (rates); \
 305         return (-1); \
 306     } \
 307     else if (((size_t) status) >= (ret_len - offset)) \
 308     { \
 309         sfree (rates); \
 310         return (-1); \
 311     } \
 312     else \
 313     offset += ((size_t) status); \
 314 } while (0)
 315
 316     if (ds->ds[ds_num].type == DS_TYPE_GAUGE)
 317         BUFFER_ADD ("%f", vl->values[ds_num].gauge);
 318     else if (store_rates)
 319     {
 320         if (rates == NULL)
 321             rates = uc_get_rate (ds, vl);
 322         if (rates == NULL)
 323         {
 324             WARNING ("format_values: "
 325                     "uc_get_rate failed.");
 326             return (-1);
 327         }
 328         BUFFER_ADD ("%g", rates[ds_num]);
 329     }
 330     else if (ds->ds[ds_num].type == DS_TYPE_COUNTER)
 331         BUFFER_ADD ("%llu", vl->values[ds_num].counter);
 332     else if (ds->ds[ds_num].type == DS_TYPE_DERIVE)
 333         BUFFER_ADD ("%"PRIi64, vl->values[ds_num].derive);
 334     else if (ds->ds[ds_num].type == DS_TYPE_ABSOLUTE)
 335         BUFFER_ADD ("%"PRIu64, vl->values[ds_num].absolute);
 336     else
 337     {
 338         ERROR ("format_values plugin: Unknown data source type: %i",
 339                 ds->ds[ds_num].type);
 340         sfree (rates);
 341         return (-1);
 342     }
 343
 344 #undef BUFFER_ADD
 345
 346     sfree (rates);
 347     return (0);
 348 }
 349
 350 static void wg_copy_escape_part (char *dst, const char *src, size_t dst_len,
 351     char escape_char)
 352 {
 353     size_t i;
 354
 355     memset (dst, 0, dst_len);
 356
 357     if (src == NULL)
 358         return;
 359
 360     for (i = 0; i < dst_len; i++)
 361     {
 362         if (src[i] == 0)
 363         {
 364             dst[i] = 0;
 365             break;
 366         }
 367
 368         if ((src[i] == '.')
 369                 || isspace ((int) src[i])
 370                 || iscntrl ((int) src[i]))
 371             dst[i] = escape_char;
 372         else
 373             dst[i] = src[i];
 374     }
 375 }
 376
 377 static int wg_format_name (char *ret, int ret_len,
 378         const value_list_t *vl,
 379         const struct wg_callback *cb,
 380         const char *ds_name)
 381 {
 382     char n_host[DATA_MAX_NAME_LEN];
 383     char n_plugin[DATA_MAX_NAME_LEN];
 384     char n_plugin_instance[DATA_MAX_NAME_LEN];
 385     char n_type[DATA_MAX_NAME_LEN];
 386     char n_type_instance[DATA_MAX_NAME_LEN];
 387
 388     char *prefix;
 389     char *postfix;
 390
 391     char tmp_plugin[2 * DATA_MAX_NAME_LEN + 1];
 392     char tmp_type[2 * DATA_MAX_NAME_LEN + 1];
 393
 394     prefix = cb->prefix;
 395     if (prefix == NULL)
 396         prefix = "";
 397
 398     postfix = cb->postfix;
 399     if (postfix == NULL)
 400         postfix = "";
 401
 402     wg_copy_escape_part (n_host, vl->host,
 403             sizeof (n_host), cb->escape_char);
 404     wg_copy_escape_part (n_plugin, vl->plugin,
 405             sizeof (n_plugin), cb->escape_char);
 406     wg_copy_escape_part (n_plugin_instance, vl->plugin_instance,
 407             sizeof (n_plugin_instance), cb->escape_char);
 408     wg_copy_escape_part (n_type, vl->type,
 409             sizeof (n_type), cb->escape_char);
 410     wg_copy_escape_part (n_type_instance, vl->type_instance,
 411             sizeof (n_type_instance), cb->escape_char);
 412
 413     if (n_plugin_instance[0] != '\0')
 414         ssnprintf (tmp_plugin, sizeof (tmp_plugin), "%s%c%s",
 415             n_plugin,
 416             cb->separate_instances ? '.' : '-',
 417             n_plugin_instance);
 418     else
 419         sstrncpy (tmp_plugin, n_plugin, sizeof (tmp_plugin));
 420
 421     if (n_type_instance[0] != '\0')
 422         ssnprintf (tmp_type, sizeof (tmp_type), "%s%c%s",
 423             n_type,
 424             cb->separate_instances ? '.' : '-',
 425             n_type_instance);
 426     else
 427         sstrncpy (tmp_type, n_type, sizeof (tmp_type));
 428
 429     if (ds_name != NULL)
 430         ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s",
 431             prefix, n_host, postfix, tmp_plugin, tmp_type, ds_name);
 432     else
 433         ssnprintf (ret, ret_len, "%s%s%s.%s.%s",
 434             prefix, n_host, postfix, tmp_plugin, tmp_type);
 435
 436     return (0);
 437 }
 438
 439 static int wg_send_message (const char* key, const char* value,
 440         cdtime_t time, struct wg_callback *cb)
 441 {
 442     int status;
 443     size_t message_len;
 444     char message[1024];
 445
 446     message_len = (size_t) ssnprintf (message, sizeof (message),
 447             "%s %s %u\r\n",
 448             key,
 449             value,
 450             (unsigned int) CDTIME_T_TO_TIME_T (time));
 451     if (message_len >= sizeof (message)) {
 452         ERROR ("write_graphite plugin: message buffer too small: "
 453                 "Need %zu bytes.", message_len + 1);
 454         return (-1);
 455     }
 456
 457     pthread_mutex_lock (&cb->send_lock);
 458
 459     if (cb->sock_fd < 0)
 460     {
 461         status = wg_callback_init (cb);
 462         if (status != 0)
 463         {
 464             ERROR ("write_graphite plugin: wg_callback_init failed.");
 465             pthread_mutex_unlock (&cb->send_lock);
 466             return (-1);
 467         }
 468     }
 469
 470     if (message_len >= cb->send_buf_free)
 471     {
 472         status = wg_flush_nolock (/* timeout = */ 0, cb);
 473         if (status != 0)
 474         {
 475             pthread_mutex_unlock (&cb->send_lock);
 476             return (status);
 477         }
 478     }
 479
 480     /* Assert that we have enough space for this message. */
 481     assert (message_len < cb->send_buf_free);
 482
 483     /* `message_len + 1' because `message_len' does not include the
 484      * trailing null byte. Neither does `send_buffer_fill'. */
 485     memcpy (cb->send_buf + cb->send_buf_fill,
 486             message, message_len + 1);
 487     cb->send_buf_fill += message_len;
 488     cb->send_buf_free -= message_len;
 489
 490     DEBUG ("write_graphite plugin: [%s]:%s buf %zu/%zu (%.1f %%) \"%s\"",
 491             cb->node,
 492             cb->service,
 493             cb->send_buf_fill, sizeof (cb->send_buf),
 494             100.0 * ((double) cb->send_buf_fill) / ((double) sizeof (cb->send_buf)),
 495             message);
 496
 497     pthread_mutex_unlock (&cb->send_lock);
 498
 499     return (0);
 500 }
 501
 502 static int wg_write_messages (const data_set_t *ds, const value_list_t *vl,
 503         struct wg_callback *cb)
 504 {
 505     char key[10*DATA_MAX_NAME_LEN];
 506     char values[512];
 507
 508     int status, i;
 509
 510     if (0 != strcmp (ds->type, vl->type))
 511     {
 512         ERROR ("write_graphite plugin: DS type does not match "
 513                 "value list type");
 514         return -1;
 515     }
 516
 517     for (i = 0; i < ds->ds_num; i++)
 518     {
 519         const char *ds_name = NULL;
 520
 521         if (cb->always_append_ds || (ds->ds_num > 1))
 522             ds_name = ds->ds[i].name;
 523
 524         /* Copy the identifier to `key' and escape it. */
 525         status = wg_format_name (key, sizeof (key), vl, cb, ds_name);
 526         if (status != 0)
 527         {
 528             ERROR ("write_graphite plugin: error with format_name");
 529             return (status);
 530         }
 531
 532         escape_string (key, sizeof (key));
 533         /* Convert the values to an ASCII representation and put that into
 534          * `values'. */
 535         status = wg_format_values (values, sizeof (values), i, ds, vl,
 536                     cb->store_rates);
 537         if (status != 0)
 538         {
 539             ERROR ("write_graphite plugin: error with "
 540                     "wg_format_values");
 541             return (status);
 542         }
 543
 544         /* Send the message to graphite */
 545         status = wg_send_message (key, values, vl->time, cb);
 546         if (status != 0)
 547         {
 548             ERROR ("write_graphite plugin: error with "
 549                     "wg_send_message");
 550             return (status);
 551         }
 552     }
 553
 554     return (0);
 555 }
 556
 557 static int wg_write (const data_set_t *ds, const value_list_t *vl,
 558         user_data_t *user_data)
 559 {
 560     struct wg_callback *cb;
 561     int status;
 562
 563     if (user_data == NULL)
 564         return (EINVAL);
 565
 566     cb = user_data->data;
 567
 568     status = wg_write_messages (ds, vl, cb);
 569
 570     return (status);
 571 }
 572
 573 static int config_set_char (char *dest,
 574         oconfig_item_t *ci)
 575 {
 576     char buffer[4];
 577     int status;
 578
 579     memset (buffer, 0, sizeof (buffer));
 580
 581     status = cf_util_get_string_buffer (ci, buffer, sizeof (buffer));
 582     if (status != 0)
 583         return (status);
 584
 585     if (buffer[0] == 0)
 586     {
 587         ERROR ("write_graphite plugin: Cannot use an empty string for the "
 588                 "\"EscapeCharacter\" option.");
 589         return (-1);
 590     }
 591
 592     if (buffer[1] != 0)
 593     {
 594         WARNING ("write_graphite plugin: Only the first character of the "
 595                 "\"EscapeCharacter\" option ('%c') will be used.",
 596                 (int) buffer[0]);
 597     }
 598
 599     *dest = buffer[0];
 600
 601     return (0);
 602 }
 603
 604 static int wg_config_carbon (oconfig_item_t *ci)
 605 {
 606     struct wg_callback *cb;
 607     user_data_t user_data;
 608     char callback_name[DATA_MAX_NAME_LEN];
 609     int i;
 610
 611     cb = malloc (sizeof (*cb));
 612     if (cb == NULL)
 613     {
 614         ERROR ("write_graphite plugin: malloc failed.");
 615         return (-1);
 616     }
 617     memset (cb, 0, sizeof (*cb));
 618     cb->sock_fd = -1;
 619     cb->node = NULL;
 620     cb->service = NULL;
 621     cb->prefix = NULL;
 622     cb->postfix = NULL;
 623     cb->escape_char = WG_DEFAULT_ESCAPE;
 624     cb->store_rates = 1;
 625
 626     pthread_mutex_init (&cb->send_lock, /* attr = */ NULL);
 627
 628     for (i = 0; i < ci->children_num; i++)
 629     {
 630         oconfig_item_t *child = ci->children + i;
 631
 632         if (strcasecmp ("Host", child->key) == 0)
 633             cf_util_get_string (child, &cb->node);
 634         else if (strcasecmp ("Port", child->key) == 0)
 635             cf_util_get_service (child, &cb->service);
 636         else if (strcasecmp ("Prefix", child->key) == 0)
 637             cf_util_get_string (child, &cb->prefix);
 638         else if (strcasecmp ("Postfix", child->key) == 0)
 639             cf_util_get_string (child, &cb->postfix);
 640         else if (strcasecmp ("StoreRates", child->key) == 0)
 641             cf_util_get_boolean (child, &cb->store_rates);
 642         else if (strcasecmp ("SeparateInstances", child->key) == 0)
 643             cf_util_get_boolean (child, &cb->separate_instances);
 644         else if (strcasecmp ("AlwaysAppendDS", child->key) == 0)
 645             cf_util_get_boolean (child, &cb->always_append_ds);
 646         else if (strcasecmp ("EscapeCharacter", child->key) == 0)
 647             config_set_char (&cb->escape_char, child);
 648         else
 649         {
 650             ERROR ("write_graphite plugin: Invalid configuration "
 651                         "option: %s.", child->key);
 652         }
 653     }
 654
 655     ssnprintf (callback_name, sizeof (callback_name), "write_graphite/%s/%s",
 656             cb->node != NULL ? cb->node : WG_DEFAULT_NODE,
 657             cb->service != NULL ? cb->service : WG_DEFAULT_SERVICE);
 658
 659     memset (&user_data, 0, sizeof (user_data));
 660     user_data.data = cb;
 661     user_data.free_func = wg_callback_free;
 662     plugin_register_write (callback_name, wg_write, &user_data);
 663
 664     user_data.free_func = NULL;
 665     plugin_register_flush (callback_name, wg_flush, &user_data);
 666
 667     return (0);
 668 }
 669
 670 static int wg_config (oconfig_item_t *ci)
 671 {
 672     int i;
 673
 674     for (i = 0; i < ci->children_num; i++)
 675     {
 676         oconfig_item_t *child = ci->children + i;
 677
 678         if (strcasecmp ("Carbon", child->key) == 0)
 679             wg_config_carbon (child);
 680         else
 681         {
 682             ERROR ("write_graphite plugin: Invalid configuration "
 683                     "option: %s.", child->key);
 684         }
 685     }
 686
 687     return (0);
 688 }
 689
 690 void module_register (void)
 691 {
 692     plugin_register_complex_config ("write_graphite", wg_config);
 693 }
 694
 695 /* vim: set sw=4 ts=4 sts=4 tw=78 et : */