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