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