src/rrdcached.c

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