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