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