src/rrdcached.c

   1 /**
   2  * collectd - src/rrdcached.c
   3  * Copyright (C) 2008-2013  Florian octo Forster
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a
   6  * copy of this software and associated documentation files (the "Software"),
   7  * to deal in the Software without restriction, including without limitation
   8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9  * and/or sell copies of the Software, and to permit persons to whom the
  10  * Software is furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  21  * DEALINGS IN THE SOFTWARE.
  22  *
  23  * Authors:
  24  *   Florian octo Forster <octo at collectd.org>
  25  **/
  26
  27 #include "collectd.h"
  28
  29 #include "common.h"
  30 #include "plugin.h"
  31 #include "utils_rrdcreate.h"
  32
  33 #undef HAVE_CONFIG_H
  34 #include <rrd.h>
  35 #include <rrd_client.h>
  36
  37 /*
  38  * Private variables
  39  */
  40 static char *datadir = NULL;
  41 static char *daemon_address = NULL;
  42 static _Bool config_create_files = 1;
  43 static _Bool config_collect_stats = 1;
  44 static rrdcreate_config_t rrdcreate_config = {
  45     /* stepsize = */ 0,
  46     /* heartbeat = */ 0,
  47     /* rrarows = */ 1200,
  48     /* xff = */ 0.1,
  49
  50     /* timespans = */ NULL,
  51     /* timespans_num = */ 0,
  52
  53     /* consolidation_functions = */ NULL,
  54     /* consolidation_functions_num = */ 0,
  55
  56     /* async = */ 0};
  57
  58 /*
  59  * Prototypes.
  60  */
  61 static int rc_write(const data_set_t *ds, const value_list_t *vl,
  62                     user_data_t __attribute__((unused)) * user_data);
  63 static int rc_flush(__attribute__((unused)) cdtime_t timeout,
  64                     const char *identifier,
  65                     __attribute__((unused)) user_data_t *ud);
  66
  67 static int value_list_to_string(char *buffer, int buffer_len,
  68                                 const data_set_t *ds, const value_list_t *vl) {
  69   int offset;
  70   int status;
  71   time_t t;
  72
  73   assert(0 == strcmp(ds->type, vl->type));
  74
  75   memset(buffer, '\0', buffer_len);
  76
  77   t = CDTIME_T_TO_TIME_T(vl->time);
  78   status = snprintf(buffer, buffer_len, "%lu", (unsigned long)t);
  79   if ((status < 1) || (status >= buffer_len))
  80     return -1;
  81   offset = status;
  82
  83   for (size_t i = 0; i < ds->ds_num; i++) {
  84     if ((ds->ds[i].type != DS_TYPE_COUNTER) &&
  85         (ds->ds[i].type != DS_TYPE_GAUGE) &&
  86         (ds->ds[i].type != DS_TYPE_DERIVE) &&
  87         (ds->ds[i].type != DS_TYPE_ABSOLUTE))
  88       return -1;
  89
  90     if (ds->ds[i].type == DS_TYPE_COUNTER) {
  91       status = snprintf(buffer + offset, buffer_len - offset, ":%llu",
  92                         vl->values[i].counter);
  93     } else if (ds->ds[i].type == DS_TYPE_GAUGE) {
  94       status = snprintf(buffer + offset, buffer_len - offset, ":%f",
  95                         vl->values[i].gauge);
  96     } else if (ds->ds[i].type == DS_TYPE_DERIVE) {
  97       status = snprintf(buffer + offset, buffer_len - offset, ":%" PRIi64,
  98                         vl->values[i].derive);
  99     } else /* if (ds->ds[i].type == DS_TYPE_ABSOLUTE) */ {
 100       status = snprintf(buffer + offset, buffer_len - offset, ":%" PRIu64,
 101                         vl->values[i].absolute);
 102     }
 103
 104     if ((status < 1) || (status >= (buffer_len - offset)))
 105       return -1;
 106
 107     offset += status;
 108   } /* for ds->ds_num */
 109
 110   return 0;
 111 } /* int value_list_to_string */
 112
 113 static int value_list_to_filename(char *buffer, size_t buffer_size,
 114                                   value_list_t const *vl) {
 115   char const suffix[] = ".rrd";
 116   int status;
 117   size_t len;
 118
 119   if (datadir != NULL) {
 120     size_t datadir_len = strlen(datadir) + 1;
 121
 122     if (datadir_len >= buffer_size)
 123       return ENOMEM;
 124
 125     sstrncpy(buffer, datadir, buffer_size);
 126     buffer[datadir_len - 1] = '/';
 127     buffer[datadir_len] = 0;
 128
 129     buffer += datadir_len;
 130     buffer_size -= datadir_len;
 131   }
 132
 133   status = FORMAT_VL(buffer, buffer_size, vl);
 134   if (status != 0)
 135     return status;
 136
 137   len = strlen(buffer);
 138   assert(len < buffer_size);
 139   buffer += len;
 140   buffer_size -= len;
 141
 142   if (buffer_size <= sizeof(suffix))
 143     return ENOMEM;
 144
 145   memcpy(buffer, suffix, sizeof(suffix));
 146   return 0;
 147 } /* int value_list_to_filename */
 148
 149 static int rc_config_get_int_positive(oconfig_item_t const *ci, int *ret) {
 150   int status;
 151   int tmp = 0;
 152
 153   status = cf_util_get_int(ci, &tmp);
 154   if (status != 0)
 155     return status;
 156   if (tmp < 0)
 157     return EINVAL;
 158
 159   *ret = tmp;
 160   return 0;
 161 } /* int rc_config_get_int_positive */
 162
 163 static int rc_config_get_xff(oconfig_item_t const *ci, double *ret) {
 164   double value;
 165
 166   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_NUMBER)) {
 167     ERROR("rrdcached plugin: The \"%s\" needs exactly one numeric argument "
 168           "in the range [0.0, 1.0)",
 169           ci->key);
 170     return EINVAL;
 171   }
 172
 173   value = ci->values[0].value.number;
 174   if ((value >= 0.0) && (value < 1.0)) {
 175     *ret = value;
 176     return 0;
 177   }
 178
 179   ERROR("rrdcached plugin: The \"%s\" needs exactly one numeric argument "
 180         "in the range [0.0, 1.0)",
 181         ci->key);
 182   return EINVAL;
 183 } /* int rc_config_get_xff */
 184
 185 static int rc_config_add_timespan(int timespan) {
 186   int *tmp;
 187
 188   if (timespan <= 0)
 189     return EINVAL;
 190
 191   tmp = realloc(rrdcreate_config.timespans,
 192                 sizeof(*rrdcreate_config.timespans) *
 193                     (rrdcreate_config.timespans_num + 1));
 194   if (tmp == NULL)
 195     return ENOMEM;
 196   rrdcreate_config.timespans = tmp;
 197
 198   rrdcreate_config.timespans[rrdcreate_config.timespans_num] = timespan;
 199   rrdcreate_config.timespans_num++;
 200
 201   return 0;
 202 } /* int rc_config_add_timespan */
 203
 204 static int rc_config(oconfig_item_t *ci) {
 205   for (int i = 0; i < ci->children_num; i++) {
 206     oconfig_item_t const *child = ci->children + i;
 207     const char *key = child->key;
 208     int status = 0;
 209
 210     if (strcasecmp("DataDir", key) == 0) {
 211       status = cf_util_get_string(child, &datadir);
 212       if (status == 0) {
 213         int len = strlen(datadir);
 214
 215         while ((len > 0) && (datadir[len - 1] == '/')) {
 216           len--;
 217           datadir[len] = 0;
 218         }
 219
 220         if (len <= 0)
 221           sfree(datadir);
 222       }
 223     } else if (strcasecmp("DaemonAddress", key) == 0)
 224       status = cf_util_get_string(child, &daemon_address);
 225     else if (strcasecmp("CreateFiles", key) == 0)
 226       status = cf_util_get_boolean(child, &config_create_files);
 227     else if (strcasecmp("CreateFilesAsync", key) == 0)
 228       status = cf_util_get_boolean(child, &rrdcreate_config.async);
 229     else if (strcasecmp("CollectStatistics", key) == 0)
 230       status = cf_util_get_boolean(child, &config_collect_stats);
 231     else if (strcasecmp("StepSize", key) == 0) {
 232       int tmp = -1;
 233
 234       status = rc_config_get_int_positive(child, &tmp);
 235       if (status == 0)
 236         rrdcreate_config.stepsize = (unsigned long)tmp;
 237     } else if (strcasecmp("HeartBeat", key) == 0)
 238       status = rc_config_get_int_positive(child, &rrdcreate_config.heartbeat);
 239     else if (strcasecmp("RRARows", key) == 0)
 240       status = rc_config_get_int_positive(child, &rrdcreate_config.rrarows);
 241     else if (strcasecmp("RRATimespan", key) == 0) {
 242       int tmp = -1;
 243       status = rc_config_get_int_positive(child, &tmp);
 244       if (status == 0)
 245         status = rc_config_add_timespan(tmp);
 246     } else if (strcasecmp("XFF", key) == 0)
 247       status = rc_config_get_xff(child, &rrdcreate_config.xff);
 248     else {
 249       WARNING("rrdcached plugin: Ignoring invalid option %s.", key);
 250       continue;
 251     }
 252
 253     if (status != 0)
 254       WARNING("rrdcached plugin: Handling the \"%s\" option failed.", key);
 255   }
 256
 257   if (daemon_address != NULL) {
 258     plugin_register_write("rrdcached", rc_write, /* user_data = */ NULL);
 259     plugin_register_flush("rrdcached", rc_flush, /* user_data = */ NULL);
 260   }
 261   return 0;
 262 } /* int rc_config */
 263
 264 static int try_reconnect(void) {
 265   int status;
 266
 267   rrdc_disconnect();
 268
 269   rrd_clear_error();
 270   status = rrdc_connect(daemon_address);
 271   if (status != 0) {
 272     ERROR("rrdcached plugin: Failed to reconnect to RRDCacheD "
 273           "at %s: %s (status=%d)",
 274           daemon_address, rrd_get_error(), status);
 275     return -1;
 276   }
 277
 278   INFO("rrdcached plugin: Successfully reconnected to RRDCacheD "
 279        "at %s",
 280        daemon_address);
 281   return 0;
 282 } /* int try_reconnect */
 283
 284 static int rc_read(void) {
 285   int status;
 286   rrdc_stats_t *head;
 287   _Bool retried = 0;
 288
 289   value_list_t vl = VALUE_LIST_INIT;
 290   vl.values = &(value_t){.gauge = NAN};
 291   vl.values_len = 1;
 292
 293   if (daemon_address == NULL)
 294     return -1;
 295
 296   if (!config_collect_stats)
 297     return -1;
 298
 299   if ((strncmp("unix:", daemon_address, strlen("unix:")) != 0) &&
 300       (daemon_address[0] != '/'))
 301     sstrncpy(vl.host, daemon_address, sizeof(vl.host));
 302   sstrncpy(vl.plugin, "rrdcached", sizeof(vl.plugin));
 303
 304   rrd_clear_error();
 305   status = rrdc_connect(daemon_address);
 306   if (status != 0) {
 307     ERROR("rrdcached plugin: Failed to connect to RRDCacheD "
 308           "at %s: %s (status=%d)",
 309           daemon_address, rrd_get_error(), status);
 310     return -1;
 311   }
 312
 313   while (42) {
 314     /* The RRD client lib does not provide any means for checking a
 315      * connection, hence we'll have to retry upon failed operations. */
 316     head = NULL;
 317     rrd_clear_error();
 318     status = rrdc_stats_get(&head);
 319     if (status == 0)
 320       break;
 321
 322     if (!retried) {
 323       retried = 1;
 324       if (try_reconnect() == 0)
 325         continue;
 326       /* else: report the error and fail */
 327     }
 328
 329     ERROR("rrdcached plugin: rrdc_stats_get failed: %s (status=%i).",
 330           rrd_get_error(), status);
 331     return -1;
 332   }
 333
 334   for (rrdc_stats_t *ptr = head; ptr != NULL; ptr = ptr->next) {
 335     if (ptr->type == RRDC_STATS_TYPE_GAUGE)
 336       vl.values[0].gauge = (gauge_t)ptr->value.gauge;
 337     else if (ptr->type == RRDC_STATS_TYPE_COUNTER)
 338       vl.values[0].counter = (counter_t)ptr->value.counter;
 339     else
 340       continue;
 341
 342     if (strcasecmp("QueueLength", ptr->name) == 0) {
 343       sstrncpy(vl.type, "queue_length", sizeof(vl.type));
 344       sstrncpy(vl.type_instance, "", sizeof(vl.type_instance));
 345     } else if (strcasecmp("UpdatesWritten", ptr->name) == 0) {
 346       sstrncpy(vl.type, "operations", sizeof(vl.type));
 347       sstrncpy(vl.type_instance, "write-updates", sizeof(vl.type_instance));
 348     } else if (strcasecmp("DataSetsWritten", ptr->name) == 0) {
 349       sstrncpy(vl.type, "operations", sizeof(vl.type));
 350       sstrncpy(vl.type_instance, "write-data_sets", sizeof(vl.type_instance));
 351     } else if (strcasecmp("TreeNodesNumber", ptr->name) == 0) {
 352       sstrncpy(vl.type, "gauge", sizeof(vl.type));
 353       sstrncpy(vl.type_instance, "tree_nodes", sizeof(vl.type_instance));
 354     } else if (strcasecmp("TreeDepth", ptr->name) == 0) {
 355       sstrncpy(vl.type, "gauge", sizeof(vl.type));
 356       sstrncpy(vl.type_instance, "tree_depth", sizeof(vl.type_instance));
 357     } else if (strcasecmp("FlushesReceived", ptr->name) == 0) {
 358       sstrncpy(vl.type, "operations", sizeof(vl.type));
 359       sstrncpy(vl.type_instance, "receive-flush", sizeof(vl.type_instance));
 360     } else if (strcasecmp("JournalBytes", ptr->name) == 0) {
 361       sstrncpy(vl.type, "counter", sizeof(vl.type));
 362       sstrncpy(vl.type_instance, "journal-bytes", sizeof(vl.type_instance));
 363     } else if (strcasecmp("JournalRotate", ptr->name) == 0) {
 364       sstrncpy(vl.type, "counter", sizeof(vl.type));
 365       sstrncpy(vl.type_instance, "journal-rotates", sizeof(vl.type_instance));
 366     } else if (strcasecmp("UpdatesReceived", ptr->name) == 0) {
 367       sstrncpy(vl.type, "operations", sizeof(vl.type));
 368       sstrncpy(vl.type_instance, "receive-update", sizeof(vl.type_instance));
 369     } else {
 370       DEBUG("rrdcached plugin: rc_read: Unknown statistic `%s'.", ptr->name);
 371       continue;
 372     }
 373
 374     plugin_dispatch_values(&vl);
 375   } /* for (ptr = head; ptr != NULL; ptr = ptr->next) */
 376
 377   rrdc_stats_free(head);
 378
 379   return 0;
 380 } /* int rc_read */
 381
 382 static int rc_init(void) {
 383   if (config_collect_stats)
 384     plugin_register_read("rrdcached", rc_read);
 385
 386   return 0;
 387 } /* int rc_init */
 388
 389 static int rc_write(const data_set_t *ds, const value_list_t *vl,
 390                     user_data_t __attribute__((unused)) * user_data) {
 391   char filename[PATH_MAX];
 392   char values[512];
 393   char *values_array[2];
 394   int status;
 395   _Bool retried = 0;
 396
 397   if (daemon_address == NULL) {
 398     ERROR("rrdcached plugin: daemon_address == NULL.");
 399     plugin_unregister_write("rrdcached");
 400     return -1;
 401   }
 402
 403   if (strcmp(ds->type, vl->type) != 0) {
 404     ERROR("rrdcached plugin: DS type does not match value list type");
 405     return -1;
 406   }
 407
 408   if (value_list_to_filename(filename, sizeof(filename), vl) != 0) {
 409     ERROR("rrdcached plugin: value_list_to_filename failed.");
 410     return -1;
 411   }
 412
 413   if (value_list_to_string(values, sizeof(values), ds, vl) != 0) {
 414     ERROR("rrdcached plugin: value_list_to_string failed.");
 415     return -1;
 416   }
 417
 418   values_array[0] = values;
 419   values_array[1] = NULL;
 420
 421   if (config_create_files) {
 422     struct stat statbuf;
 423
 424     status = stat(filename, &statbuf);
 425     if (status != 0) {
 426       if (errno != ENOENT) {
 427         char errbuf[1024];
 428         ERROR("rrdcached plugin: stat (%s) failed: %s", filename,
 429               sstrerror(errno, errbuf, sizeof(errbuf)));
 430         return -1;
 431       }
 432
 433       status = cu_rrd_create_file(filename, ds, vl, &rrdcreate_config);
 434       if (status != 0) {
 435         ERROR("rrdcached plugin: cu_rrd_create_file (%s) failed.", filename);
 436         return -1;
 437       } else if (rrdcreate_config.async)
 438         return 0;
 439     }
 440   }
 441
 442   rrd_clear_error();
 443   status = rrdc_connect(daemon_address);
 444   if (status != 0) {
 445     ERROR("rrdcached plugin: Failed to connect to RRDCacheD "
 446           "at %s: %s (status=%d)",
 447           daemon_address, rrd_get_error(), status);
 448     return -1;
 449   }
 450
 451   while (42) {
 452     /* The RRD client lib does not provide any means for checking a
 453      * connection, hence we'll have to retry upon failed operations. */
 454     rrd_clear_error();
 455     status = rrdc_update(filename, /* values_num = */ 1, (void *)values_array);
 456     if (status == 0)
 457       break;
 458
 459     if (!retried) {
 460       retried = 1;
 461       if (try_reconnect() == 0)
 462         continue;
 463       /* else: report the error and fail */
 464     }
 465
 466     ERROR("rrdcached plugin: rrdc_update (%s, [%s], 1) failed: %s (status=%i)",
 467           filename, values_array[0], rrd_get_error(), status);
 468     return -1;
 469   }
 470
 471   return 0;
 472 } /* int rc_write */
 473
 474 static int rc_flush(__attribute__((unused)) cdtime_t timeout, /* {{{ */
 475                     const char *identifier,
 476                     __attribute__((unused)) user_data_t *ud) {
 477   char filename[PATH_MAX + 1];
 478   int status;
 479   _Bool retried = 0;
 480
 481   if (identifier == NULL)
 482     return EINVAL;
 483
 484   if (datadir != NULL)
 485     snprintf(filename, sizeof(filename), "%s/%s.rrd", datadir, identifier);
 486   else
 487     snprintf(filename, sizeof(filename), "%s.rrd", identifier);
 488
 489   rrd_clear_error();
 490   status = rrdc_connect(daemon_address);
 491   if (status != 0) {
 492     ERROR("rrdcached plugin: Failed to connect to RRDCacheD "
 493           "at %s: %s (status=%d)",
 494           daemon_address, rrd_get_error(), status);
 495     return -1;
 496   }
 497
 498   while (42) {
 499     /* The RRD client lib does not provide any means for checking a
 500      * connection, hence we'll have to retry upon failed operations. */
 501     rrd_clear_error();
 502     status = rrdc_flush(filename);
 503     if (status == 0)
 504       break;
 505
 506     if (!retried) {
 507       retried = 1;
 508       if (try_reconnect() == 0)
 509         continue;
 510       /* else: report the error and fail */
 511     }
 512
 513     ERROR("rrdcached plugin: rrdc_flush (%s) failed: %s (status=%i).", filename,
 514           rrd_get_error(), status);
 515     return -1;
 516   }
 517   DEBUG("rrdcached plugin: rrdc_flush (%s): Success.", filename);
 518
 519   return 0;
 520 } /* }}} int rc_flush */
 521
 522 static int rc_shutdown(void) {
 523   rrdc_disconnect();
 524   return 0;
 525 } /* int rc_shutdown */
 526
 527 void module_register(void) {
 528   plugin_register_complex_config("rrdcached", rc_config);
 529   plugin_register_init("rrdcached", rc_init);
 530   plugin_register_shutdown("rrdcached", rc_shutdown);
 531 } /* void module_register */