src/write_graphite.c

   1 /**
   2  * collectd - src/write_graphite.c
   3  * Copyright (C) 2011  Scott Sanders
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Author:
  19  *   Scott Sanders <scott@jssjr.com>
  20  *
  21  *   based on the excellent write_http plugin
  22  **/
  23
  24  /* write_graphite plugin configuation example
  25   *
  26   * <Plugin write_graphite>
  27   *   <Carbon>
  28   *     Host "localhost"
  29   *     Port "2003"
  30   *     Prefix "collectd"
  31   *   </Carbon>
  32   * </Plugin>
  33   */
  34
  35 #include "collectd.h"
  36 #include "common.h"
  37 #include "plugin.h"
  38 #include "configfile.h"
  39
  40 #include "utils_cache.h"
  41 #include "utils_parse_option.h"
  42
  43 /* Folks without pthread will need to disable this plugin. */
  44 #include <pthread.h>
  45
  46 #include <sys/socket.h>
  47 #include <netdb.h>
  48
  49 #ifndef WG_FORMAT_NAME
  50 #define WG_FORMAT_NAME(ret, ret_len, vl, cb, ds_name) \
  51         wg_format_name (ret, ret_len, (vl)->host, \
  52                          (vl)->plugin, (vl)->plugin_instance, \
  53                          (vl)->type, (vl)->type_instance, \
  54                          (cb)->prefix, (cb)->postfix, \
  55                          ds_name, (cb)->escape_char)
  56 #endif
  57
  58 #ifndef WG_DEFAULT_NODE
  59 # define WG_DEFAULT_NODE "localhost"
  60 #endif
  61
  62 #ifndef WG_DEFAULT_SERVICE
  63 # define WG_DEFAULT_SERVICE "2003"
  64 #endif
  65
  66 #ifndef WG_SEND_BUF_SIZE
  67 # define WG_SEND_BUF_SIZE 4096
  68 #endif
  69
  70 /*
  71  * Private variables
  72  */
  73 struct wg_callback
  74 {
  75     int      sock_fd;
  76     struct hostent *server;
  77
  78     char    *node;
  79     char    *service;
  80     char    *prefix;
  81     char    *postfix;
  82     char     escape_char;
  83
  84     char     send_buf[WG_SEND_BUF_SIZE];
  85     size_t   send_buf_free;
  86     size_t   send_buf_fill;
  87     cdtime_t send_buf_init_time;
  88
  89     pthread_mutex_t send_lock;
  90 };
  91
  92
  93 /*
  94  * Functions
  95  */
  96 static void wg_reset_buffer (struct wg_callback *cb)
  97 {
  98     memset (cb->send_buf, 0, sizeof (cb->send_buf));
  99     cb->send_buf_free = sizeof (cb->send_buf);
 100     cb->send_buf_fill = 0;
 101     cb->send_buf_init_time = cdtime ();
 102 }
 103
 104 static int wg_send_buffer (struct wg_callback *cb)
 105 {
 106     int status = 0;
 107
 108     status = write (cb->sock_fd, cb->send_buf, strlen (cb->send_buf));
 109     if (status < 0)
 110     {
 111         ERROR ("write_graphite plugin: send failed with "
 112                 "status %i (%s)",
 113                 status,
 114                 strerror (errno));
 115
 116         pthread_mutex_trylock (&cb->send_lock);
 117
 118         DEBUG ("write_graphite plugin: closing socket and restting fd "
 119                 "so reinit will occur");
 120         close (cb->sock_fd);
 121         cb->sock_fd = -1;
 122
 123         pthread_mutex_unlock (&cb->send_lock);
 124
 125         return (-1);
 126     }
 127     return (0);
 128 }
 129
 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     wg_flush_nolock (/* timeout = */ 0, cb);
 237
 238     close(cb->sock_fd);
 239     sfree(cb->node);
 240     sfree(cb->service);
 241     sfree(cb->prefix);
 242     sfree(cb->postfix);
 243
 244     sfree(cb);
 245 }
 246
 247 static int wg_flush (cdtime_t timeout,
 248         const char *identifier __attribute__((unused)),
 249         user_data_t *user_data)
 250 {
 251     struct wg_callback *cb;
 252     int status;
 253
 254     if (user_data == NULL)
 255         return (-EINVAL);
 256
 257     cb = user_data->data;
 258
 259     pthread_mutex_lock (&cb->send_lock);
 260
 261     if (cb->sock_fd < 0)
 262     {
 263         status = wg_callback_init (cb);
 264         if (status != 0)
 265         {
 266             ERROR ("write_graphite plugin: wg_callback_init failed.");
 267             pthread_mutex_unlock (&cb->send_lock);
 268             return (-1);
 269         }
 270     }
 271
 272     status = wg_flush_nolock (timeout, cb);
 273     pthread_mutex_unlock (&cb->send_lock);
 274
 275     return (status);
 276 }
 277
 278 static int wg_format_values (char *ret, size_t ret_len,
 279         int ds_num, const data_set_t *ds, const value_list_t *vl,
 280         _Bool store_rates)
 281 {
 282     size_t offset = 0;
 283     int status;
 284     gauge_t *rates = NULL;
 285
 286     assert (0 == strcmp (ds->type, vl->type));
 287
 288     memset (ret, 0, ret_len);
 289
 290 #define BUFFER_ADD(...) do { \
 291     status = ssnprintf (ret + offset, ret_len - offset, \
 292             __VA_ARGS__); \
 293     if (status < 1) \
 294     { \
 295         sfree (rates); \
 296         return (-1); \
 297     } \
 298     else if (((size_t) status) >= (ret_len - offset)) \
 299     { \
 300         sfree (rates); \
 301         return (-1); \
 302     } \
 303     else \
 304     offset += ((size_t) status); \
 305 } while (0)
 306
 307     if (ds->ds[ds_num].type == DS_TYPE_GAUGE)
 308         BUFFER_ADD ("%f", vl->values[ds_num].gauge);
 309     else if (store_rates)
 310     {
 311         if (rates == NULL)
 312             rates = uc_get_rate (ds, vl);
 313         if (rates == NULL)
 314         {
 315             WARNING ("format_values: "
 316                     "uc_get_rate failed.");
 317             return (-1);
 318         }
 319         BUFFER_ADD ("%g", rates[ds_num]);
 320     }
 321     else if (ds->ds[ds_num].type == DS_TYPE_COUNTER)
 322         BUFFER_ADD ("%llu", vl->values[ds_num].counter);
 323     else if (ds->ds[ds_num].type == DS_TYPE_DERIVE)
 324         BUFFER_ADD ("%"PRIi64, vl->values[ds_num].derive);
 325     else if (ds->ds[ds_num].type == DS_TYPE_ABSOLUTE)
 326         BUFFER_ADD ("%"PRIu64, vl->values[ds_num].absolute);
 327     else
 328     {
 329         ERROR ("format_values plugin: Unknown data source type: %i",
 330                 ds->ds[ds_num].type);
 331         sfree (rates);
 332         return (-1);
 333     }
 334
 335 #undef BUFFER_ADD
 336
 337     sfree (rates);
 338     return (0);
 339 }
 340
 341 static void wg_copy_escape_part (char *dst, const char *src, size_t dst_len,
 342     char escape_char)
 343 {
 344     size_t i;
 345
 346     memset (dst, 0, dst_len);
 347
 348     if (src == NULL)
 349         return;
 350
 351     for (i = 0; i < dst_len; i++)
 352     {
 353         if ((src[i] == '.')
 354                 || isspace ((int) src[i])
 355                 || iscntrl ((int) src[i]))
 356             dst[i] = escape_char;
 357         else
 358             dst[i] = src[i];
 359
 360         if (src[i] == 0)
 361             break;
 362     }
 363 }
 364
 365 static int wg_format_name (char *ret, int ret_len,
 366         const char *hostname,
 367         const char *plugin, const char *plugin_instance,
 368         const char *type, const char *type_instance,
 369         const char *prefix, const char *postfix,
 370         const char *ds_name, char escape_char)
 371 {
 372     char n_hostname[DATA_MAX_NAME_LEN];
 373     char n_plugin[DATA_MAX_NAME_LEN];
 374     char n_plugin_instance[DATA_MAX_NAME_LEN];
 375     char n_type[DATA_MAX_NAME_LEN];
 376     char n_type_instance[DATA_MAX_NAME_LEN];
 377     int  status;
 378
 379     assert (hostname != NULL);
 380     assert (plugin != NULL);
 381     assert (type != NULL);
 382     assert (ds_name != NULL);
 383
 384     if (prefix == NULL)
 385         prefix = "";
 386
 387     if (postfix == NULL)
 388         postfix = "";
 389
 390     wg_copy_escape_part (n_hostname, hostname,
 391             sizeof (n_hostname), escape_char);
 392     wg_copy_escape_part (n_plugin, plugin,
 393             sizeof (n_plugin), escape_char);
 394     wg_copy_escape_part (n_plugin_instance, plugin_instance,
 395             sizeof (n_plugin_instance), escape_char);
 396     wg_copy_escape_part (n_type, type,
 397             sizeof (n_type), escape_char);
 398     wg_copy_escape_part (n_type_instance, type_instance,
 399             sizeof (n_type_instance), escape_char);
 400
 401     if (n_plugin_instance[0] == '\0')
 402     {
 403         if (n_type_instance[0] == '\0')
 404         {
 405             status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s",
 406                     prefix, n_hostname, postfix, n_plugin, n_type, ds_name);
 407         }
 408         else
 409         {
 410             status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s-%s.%s",
 411                     prefix, n_hostname, postfix, n_plugin, n_type,
 412                     n_type_instance, ds_name);
 413         }
 414     }
 415     else
 416     {
 417         if (n_type_instance[0] == '\0')
 418         {
 419             status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s.%s",
 420                     prefix, n_hostname, postfix, n_plugin,
 421                     n_plugin_instance, n_type, ds_name);
 422         }
 423         else
 424         {
 425             status = ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s-%s.%s",
 426                     prefix, n_hostname, postfix, n_plugin,
 427                     n_plugin_instance, n_type, n_type_instance, ds_name);
 428         }
 429     }
 430
 431     if ((status < 1) || (status >= ret_len))
 432         return (-1);
 433
 434     return (0);
 435 }
 436
 437 static int wg_send_message (const char* key, const char* value,
 438         cdtime_t time, struct wg_callback *cb)
 439 {
 440     int status;
 441     size_t message_len;
 442     char message[1024];
 443
 444     message_len = (size_t) ssnprintf (message, sizeof (message),
 445             "%s %s %.0f\n",
 446             key,
 447             value,
 448             CDTIME_T_TO_DOUBLE(time));
 449     if (message_len >= sizeof (message)) {
 450         ERROR ("write_graphite plugin: message buffer too small: "
 451                 "Need %zu bytes.", message_len + 1);
 452         return (-1);
 453     }
 454
 455
 456     pthread_mutex_lock (&cb->send_lock);
 457
 458     if (cb->sock_fd < 0)
 459     {
 460         status = wg_callback_init (cb);
 461         if (status != 0)
 462         {
 463             ERROR ("write_graphite plugin: wg_callback_init failed.");
 464             pthread_mutex_unlock (&cb->send_lock);
 465             return (-1);
 466         }
 467     }
 468
 469     if (message_len >= cb->send_buf_free)
 470     {
 471         status = wg_flush_nolock (/* timeout = */ 0, cb);
 472         if (status != 0)
 473         {
 474             pthread_mutex_unlock (&cb->send_lock);
 475             return (status);
 476         }
 477     }
 478     assert (message_len < cb->send_buf_free);
 479
 480     /* `message_len + 1' because `message_len' does not include the
 481      * trailing null byte. Neither does `send_buffer_fill'. */
 482     memcpy (cb->send_buf + cb->send_buf_fill,
 483             message, message_len + 1);
 484     cb->send_buf_fill += message_len;
 485     cb->send_buf_free -= message_len;
 486
 487     DEBUG ("write_graphite plugin: <%s:%s> buf %zu/%zu (%g%%) \"%s\"",
 488             cb->node,
 489             cb->service,
 490             cb->send_buf_fill, sizeof (cb->send_buf),
 491             100.0 * ((double) cb->send_buf_fill) / ((double) sizeof (cb->send_buf)),
 492             message);
 493
 494     /* Check if we have enough space for this message. */
 495     pthread_mutex_unlock (&cb->send_lock);
 496
 497     return (0);
 498 }
 499
 500 static int wg_write_messages (const data_set_t *ds, const value_list_t *vl,
 501         struct wg_callback *cb)
 502 {
 503     char key[10*DATA_MAX_NAME_LEN];
 504     char values[512];
 505
 506     int status, i;
 507
 508     if (0 != strcmp (ds->type, vl->type))
 509     {
 510         ERROR ("write_graphite plugin: DS type does not match "
 511                 "value list type");
 512         return -1;
 513     }
 514
 515     if (ds->ds_num > 1)
 516     {
 517         for (i = 0; i < ds->ds_num; i++)
 518         {
 519             /* Copy the identifier to `key' and escape it. */
 520             status = WG_FORMAT_NAME (key, sizeof (key), vl, cb, ds->ds[i].name);
 521             if (status != 0)
 522             {
 523                 ERROR ("write_graphite plugin: error with format_name");
 524                 return (status);
 525             }
 526
 527             escape_string (key, sizeof (key));
 528             /* Convert the values to an ASCII representation and put that
 529              * into `values'. */
 530             status = wg_format_values (values, sizeof (values), i, ds, vl, 0);
 531             if (status != 0)
 532             {
 533                 ERROR ("write_graphite plugin: error with "
 534                         "wg_format_values");
 535                 return (status);
 536             }
 537
 538             /* Send the message to graphite */
 539             status = wg_send_message (key, values, vl->time, cb);
 540             if (status != 0)
 541             {
 542                 ERROR ("write_graphite plugin: error with "
 543                         "wg_send_message");
 544                 return (status);
 545             }
 546         }
 547     }
 548     else
 549     {
 550         /* Copy the identifier to `key' and escape it. */
 551         status = WG_FORMAT_NAME (key, sizeof (key), vl, cb, NULL);
 552         if (status != 0)
 553         {
 554             ERROR ("write_graphite plugin: error with format_name");
 555             return (status);
 556         }
 557
 558         escape_string (key, sizeof (key));
 559         /* Convert the values to an ASCII representation and put that into
 560          * `values'. */
 561         status = wg_format_values (values, sizeof (values), 0, ds, vl, 0);
 562         if (status != 0)
 563         {
 564             ERROR ("write_graphite plugin: error with "
 565                     "wg_format_values");
 566             return (status);
 567         }
 568
 569         /* Send the message to graphite */
 570         status = wg_send_message (key, values, vl->time, cb);
 571         if (status != 0)
 572         {
 573             ERROR ("write_graphite plugin: error with "
 574                     "wg_send_message");
 575             return (status);
 576         }
 577     }
 578
 579     return (0);
 580 }
 581
 582 static int wg_write (const data_set_t *ds, const value_list_t *vl,
 583         user_data_t *user_data)
 584 {
 585     struct wg_callback *cb;
 586     int status;
 587
 588     if (user_data == NULL)
 589         return (-EINVAL);
 590
 591     cb = user_data->data;
 592
 593     status = wg_write_messages (ds, vl, cb);
 594
 595     return (status);
 596 }
 597
 598 static int config_set_char (char *dest,
 599         oconfig_item_t *ci)
 600 {
 601     if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING))
 602     {
 603         WARNING ("write_graphite plugin: The `%s' config option "
 604                 "needs exactly one string argument.", ci->key);
 605         return (-1);
 606     }
 607
 608     *dest = ci->values[0].value.string[0];
 609
 610     return (0);
 611 }
 612
 613 static int wg_config_carbon (oconfig_item_t *ci)
 614 {
 615     struct wg_callback *cb;
 616     user_data_t user_data;
 617     int i;
 618
 619     cb = malloc (sizeof (*cb));
 620     if (cb == NULL)
 621     {
 622         ERROR ("write_graphite plugin: malloc failed.");
 623         return (-1);
 624     }
 625     memset (cb, 0, sizeof (*cb));
 626     cb->sock_fd = -1;
 627     cb->node = NULL;
 628     cb->service = NULL;
 629     cb->prefix = NULL;
 630     cb->postfix = NULL;
 631     cb->server = NULL;
 632     cb->escape_char = '_';
 633
 634     pthread_mutex_init (&cb->send_lock, /* attr = */ NULL);
 635
 636     for (i = 0; i < ci->children_num; i++)
 637     {
 638         oconfig_item_t *child = ci->children + i;
 639
 640         if (strcasecmp ("Host", child->key) == 0)
 641             cf_util_get_string (child, &cb->node);
 642         else if (strcasecmp ("Port", child->key) == 0)
 643             cf_util_get_string (child, &cb->service);
 644         else if (strcasecmp ("Prefix", child->key) == 0)
 645             cf_util_get_string (child, &cb->prefix);
 646         else if (strcasecmp ("Postfix", child->key) == 0)
 647             cf_util_get_string (child, &cb->postfix);
 648         else if (strcasecmp ("EscapeCharacter", child->key) == 0)
 649             config_set_char (&cb->escape_char, child);
 650         else
 651         {
 652             ERROR ("write_graphite plugin: Invalid configuration "
 653                         "option: %s.", child->key);
 654         }
 655     }
 656
 657     DEBUG ("write_graphite: Registering write callback to carbon agent %s:%s",
 658             cb->node ? cb->node : WG_DEFAULT_NODE,
 659             cb->service ? cb->service : WG_DEFAULT_SERVICE);
 660
 661     memset (&user_data, 0, sizeof (user_data));
 662     user_data.data = cb;
 663     user_data.free_func = NULL;
 664     plugin_register_flush ("write_graphite", wg_flush, &user_data);
 665
 666     user_data.free_func = wg_callback_free;
 667     plugin_register_write ("write_graphite", wg_write, &user_data);
 668
 669     return (0);
 670 }
 671
 672 static int wg_config (oconfig_item_t *ci)
 673 {
 674     int i;
 675
 676     for (i = 0; i < ci->children_num; i++)
 677     {
 678         oconfig_item_t *child = ci->children + i;
 679
 680         if (strcasecmp ("Carbon", child->key) == 0)
 681             wg_config_carbon (child);
 682         else
 683         {
 684             ERROR ("write_graphite plugin: Invalid configuration "
 685                     "option: %s.", child->key);
 686         }
 687     }
 688
 689     return (0);
 690 }
 691
 692 void module_register (void)
 693 {
 694     plugin_register_complex_config ("write_graphite", wg_config);
 695 }
 696
 697 /* vim: set sw=4 ts=4 sts=4 tw=78 et : */