src/write_tsdb.c

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