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, int buffer_len,
 116     const data_set_t *ds, const value_list_t *vl)
 117 {
 118   int offset = 0;
 119   int status;
 120
 121   assert (0 == strcmp (ds->type, vl->type));
 122
 123   if (datadir != NULL)
 124   {
 125     status = ssnprintf (buffer + offset, buffer_len - offset,
 126         "%s/", datadir);
 127     if ((status < 1) || (status >= buffer_len - offset))
 128       return (-1);
 129     offset += status;
 130   }
 131
 132   status = ssnprintf (buffer + offset, buffer_len - offset,
 133       "%s/", vl->host);
 134   if ((status < 1) || (status >= buffer_len - offset))
 135     return (-1);
 136   offset += status;
 137
 138   if (strlen (vl->plugin_instance) > 0)
 139     status = ssnprintf (buffer + offset, buffer_len - offset,
 140         "%s-%s/", vl->plugin, vl->plugin_instance);
 141   else
 142     status = ssnprintf (buffer + offset, buffer_len - offset,
 143         "%s/", vl->plugin);
 144   if ((status < 1) || (status >= buffer_len - offset))
 145     return (-1);
 146   offset += status;
 147
 148   if (strlen (vl->type_instance) > 0)
 149     status = ssnprintf (buffer + offset, buffer_len - offset,
 150         "%s-%s", vl->type, vl->type_instance);
 151   else
 152     status = ssnprintf (buffer + offset, buffer_len - offset,
 153         "%s", vl->type);
 154   if ((status < 1) || (status >= buffer_len - offset))
 155     return (-1);
 156   offset += status;
 157
 158   strncpy (buffer + offset, ".rrd", buffer_len - offset);
 159   buffer[buffer_len - 1] = 0;
 160
 161   return (0);
 162 } /* int value_list_to_filename */
 163
 164 static int rc_config_get_int_positive (oconfig_item_t const *ci, int *ret)
 165 {
 166   int status;
 167   int tmp = 0;
 168
 169   status = cf_util_get_int (ci, &tmp);
 170   if (status != 0)
 171     return (status);
 172   if (tmp < 0)
 173     return (EINVAL);
 174
 175   *ret = tmp;
 176   return (0);
 177 } /* int rc_config_get_int_positive */
 178
 179 static int rc_config_get_xff (oconfig_item_t const *ci, double *ret)
 180 {
 181   double value;
 182
 183   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_NUMBER))
 184   {
 185     ERROR ("rrdcached plugin: The \"%s\" needs exactly one numeric argument "
 186         "in the range [0.0, 1.0)", ci->key);
 187     return (EINVAL);
 188   }
 189
 190   value = ci->values[0].value.number;
 191   if ((value >= 0.0) && (value < 1.0))
 192   {
 193     *ret = value;
 194     return (0);
 195   }
 196
 197   ERROR ("rrdcached plugin: The \"%s\" needs exactly one numeric argument "
 198       "in the range [0.0, 1.0)", ci->key);
 199   return (EINVAL);
 200 } /* int rc_config_get_xff */
 201
 202 static int rc_config_add_timespan (int timespan)
 203 {
 204   int *tmp;
 205
 206   if (timespan <= 0)
 207     return (EINVAL);
 208
 209   tmp = realloc (rrdcreate_config.timespans,
 210       sizeof (*rrdcreate_config.timespans)
 211       * (rrdcreate_config.timespans_num + 1));
 212   if (tmp == NULL)
 213     return (ENOMEM);
 214   rrdcreate_config.timespans = tmp;
 215
 216   rrdcreate_config.timespans[rrdcreate_config.timespans_num] = timespan;
 217   rrdcreate_config.timespans_num++;
 218
 219   return (0);
 220 } /* int rc_config_add_timespan */
 221
 222 static int rc_config (oconfig_item_t *ci)
 223 {
 224   int i;
 225
 226   for (i = 0; i < ci->children_num; i++)
 227   {
 228     oconfig_item_t const *child = ci->children + i;
 229     const char *key = child->key;
 230     int status = 0;
 231
 232     if (strcasecmp ("DataDir", key) == 0)
 233     {
 234       status = cf_util_get_string (child, &datadir);
 235       if (status == 0)
 236       {
 237         int len = strlen (datadir);
 238
 239         while ((len > 0) && (datadir[len - 1] == '/'))
 240         {
 241           len--;
 242           datadir[len] = 0;
 243         }
 244
 245         if (len <= 0)
 246           sfree (datadir);
 247       }
 248     }
 249     else if (strcasecmp ("DaemonAddress", key) == 0)
 250       status = cf_util_get_string (child, &daemon_address);
 251     else if (strcasecmp ("CreateFiles", key) == 0)
 252       status = cf_util_get_boolean (child, &config_create_files);
 253     else if (strcasecmp ("CollectStatistics", key) == 0)
 254       status = cf_util_get_boolean (child, &config_collect_stats);
 255     else if (strcasecmp ("StepSize", key) == 0)
 256     {
 257       int tmp = -1;
 258
 259       status = rc_config_get_int_positive (child, &tmp);
 260       if (status == 0)
 261         rrdcreate_config.stepsize = (unsigned long) tmp;
 262     }
 263     else if (strcasecmp ("HeartBeat", key) == 0)
 264       status = rc_config_get_int_positive (child, &rrdcreate_config.heartbeat);
 265     else if (strcasecmp ("RRARows", key) == 0)
 266       status = rc_config_get_int_positive (child, &rrdcreate_config.rrarows);
 267     else if (strcasecmp ("RRATimespan", key) == 0)
 268     {
 269       int tmp = -1;
 270       status = rc_config_get_int_positive (child, &tmp);
 271       if (status == 0)
 272         status = rc_config_add_timespan (tmp);
 273     }
 274     else if (strcasecmp ("XFF", key) == 0)
 275       status = rc_config_get_xff (child, &rrdcreate_config.xff);
 276     else
 277     {
 278       WARNING ("rrdcached plugin: Ignoring invalid option %s.", key);
 279       continue;
 280     }
 281
 282     if (status != 0)
 283       WARNING ("rrdcached plugin: Handling the \"%s\" option failed.", key);
 284   }
 285
 286   if (daemon_address != NULL)
 287   {
 288     plugin_register_write ("rrdcached", rc_write, /* user_data = */ NULL);
 289     plugin_register_flush ("rrdcached", rc_flush, /* user_data = */ NULL);
 290   }
 291   return (0);
 292 } /* int rc_config */
 293
 294 static int rc_read (void)
 295 {
 296   int status;
 297   rrdc_stats_t *head;
 298   rrdc_stats_t *ptr;
 299
 300   value_t values[1];
 301   value_list_t vl = VALUE_LIST_INIT;
 302
 303   if (daemon_address == NULL)
 304     return (-1);
 305
 306   if (!config_collect_stats)
 307     return (-1);
 308
 309   vl.values = values;
 310   vl.values_len = 1;
 311
 312   if ((strncmp ("unix:", daemon_address, strlen ("unix:")) == 0)
 313       || (daemon_address[0] == '/'))
 314     sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 315   else
 316     sstrncpy (vl.host, daemon_address, sizeof (vl.host));
 317   sstrncpy (vl.plugin, "rrdcached", sizeof (vl.plugin));
 318
 319   status = rrdc_connect (daemon_address);
 320   if (status != 0)
 321   {
 322     ERROR ("rrdcached plugin: rrdc_connect (%s) failed with status %i.",
 323         daemon_address, status);
 324     return (-1);
 325   }
 326
 327   head = NULL;
 328   status = rrdc_stats_get (&head);
 329   if (status != 0)
 330   {
 331     ERROR ("rrdcached plugin: rrdc_stats_get failed with status %i.", status);
 332     return (-1);
 333   }
 334
 335   for (ptr = head; ptr != NULL; ptr = ptr->next)
 336   {
 337     if (ptr->type == RRDC_STATS_TYPE_GAUGE)
 338       values[0].gauge = (gauge_t) ptr->value.gauge;
 339     else if (ptr->type == RRDC_STATS_TYPE_COUNTER)
 340       values[0].counter = (counter_t) ptr->value.counter;
 341     else
 342       continue;
 343
 344     if (strcasecmp ("QueueLength", ptr->name) == 0)
 345     {
 346       sstrncpy (vl.type, "queue_length", sizeof (vl.type));
 347       sstrncpy (vl.type_instance, "", sizeof (vl.type_instance));
 348     }
 349     else if (strcasecmp ("UpdatesWritten", ptr->name) == 0)
 350     {
 351       sstrncpy (vl.type, "operations", sizeof (vl.type));
 352       sstrncpy (vl.type_instance, "write-updates", sizeof (vl.type_instance));
 353     }
 354     else if (strcasecmp ("DataSetsWritten", ptr->name) == 0)
 355     {
 356       sstrncpy (vl.type, "operations", sizeof (vl.type));
 357       sstrncpy (vl.type_instance, "write-data_sets",
 358           sizeof (vl.type_instance));
 359     }
 360     else if (strcasecmp ("TreeNodesNumber", ptr->name) == 0)
 361     {
 362       sstrncpy (vl.type, "gauge", sizeof (vl.type));
 363       sstrncpy (vl.type_instance, "tree_nodes", sizeof (vl.type_instance));
 364     }
 365     else if (strcasecmp ("TreeDepth", ptr->name) == 0)
 366     {
 367       sstrncpy (vl.type, "gauge", sizeof (vl.type));
 368       sstrncpy (vl.type_instance, "tree_depth", sizeof (vl.type_instance));
 369     }
 370     else if (strcasecmp ("FlushesReceived", ptr->name) == 0)
 371     {
 372       sstrncpy (vl.type, "operations", sizeof (vl.type));
 373       sstrncpy (vl.type_instance, "receive-flush", sizeof (vl.type_instance));
 374     }
 375     else if (strcasecmp ("JournalBytes", ptr->name) == 0)
 376     {
 377       sstrncpy (vl.type, "counter", sizeof (vl.type));
 378       sstrncpy (vl.type_instance, "journal-bytes", sizeof (vl.type_instance));
 379     }
 380     else if (strcasecmp ("JournalRotate", ptr->name) == 0)
 381     {
 382       sstrncpy (vl.type, "counter", sizeof (vl.type));
 383       sstrncpy (vl.type_instance, "journal-rotates", sizeof (vl.type_instance));
 384     }
 385     else if (strcasecmp ("UpdatesReceived", ptr->name) == 0)
 386     {
 387       sstrncpy (vl.type, "operations", sizeof (vl.type));
 388       sstrncpy (vl.type_instance, "receive-update", sizeof (vl.type_instance));
 389     }
 390     else
 391     {
 392       DEBUG ("rrdcached plugin: rc_read: Unknown statistic `%s'.", ptr->name);
 393       continue;
 394     }
 395
 396     plugin_dispatch_values (&vl);
 397   } /* for (ptr = head; ptr != NULL; ptr = ptr->next) */
 398
 399   rrdc_stats_free (head);
 400
 401   return (0);
 402 } /* int rc_read */
 403
 404 static int rc_init (void)
 405 {
 406   if (config_collect_stats)
 407     plugin_register_read ("rrdcached", rc_read);
 408
 409   return (0);
 410 } /* int rc_init */
 411
 412 static int rc_write (const data_set_t *ds, const value_list_t *vl,
 413     user_data_t __attribute__((unused)) *user_data)
 414 {
 415   char filename[PATH_MAX];
 416   char values[512];
 417   char *values_array[2];
 418   int status;
 419
 420   if (daemon_address == NULL)
 421   {
 422     ERROR ("rrdcached plugin: daemon_address == NULL.");
 423     plugin_unregister_write ("rrdcached");
 424     return (-1);
 425   }
 426
 427   if (strcmp (ds->type, vl->type) != 0)
 428   {
 429     ERROR ("rrdcached plugin: DS type does not match value list type");
 430     return (-1);
 431   }
 432
 433   if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
 434   {
 435     ERROR ("rrdcached plugin: value_list_to_filename failed.");
 436     return (-1);
 437   }
 438
 439   if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
 440   {
 441     ERROR ("rrdcached plugin: value_list_to_string failed.");
 442     return (-1);
 443   }
 444
 445   values_array[0] = values;
 446   values_array[1] = NULL;
 447
 448   if (config_create_files)
 449   {
 450     struct stat statbuf;
 451
 452     status = stat (filename, &statbuf);
 453     if (status != 0)
 454     {
 455       if (errno != ENOENT)
 456       {
 457         char errbuf[1024];
 458         ERROR ("rrdcached plugin: stat (%s) failed: %s",
 459             filename, sstrerror (errno, errbuf, sizeof (errbuf)));
 460         return (-1);
 461       }
 462
 463       status = cu_rrd_create_file (filename, ds, vl, &rrdcreate_config);
 464       if (status != 0)
 465       {
 466         ERROR ("rrdcached plugin: cu_rrd_create_file (%s) failed.",
 467             filename);
 468         return (-1);
 469       }
 470     }
 471   }
 472
 473   status = rrdc_connect (daemon_address);
 474   if (status != 0)
 475   {
 476     ERROR ("rrdcached plugin: rrdc_connect (%s) failed with status %i.",
 477         daemon_address, status);
 478     return (-1);
 479   }
 480
 481   status = rrdc_update (filename, /* values_num = */ 1, (void *) values_array);
 482   if (status != 0)
 483   {
 484     ERROR ("rrdcached plugin: rrdc_update (%s, [%s], 1) failed with "
 485         "status %i.",
 486         filename, values_array[0], status);
 487     return (-1);
 488   }
 489
 490   return (0);
 491 } /* int rc_write */
 492
 493 static int rc_flush (__attribute__((unused)) cdtime_t timeout, /* {{{ */
 494     const char *identifier,
 495     __attribute__((unused)) user_data_t *ud)
 496 {
 497   char filename[PATH_MAX + 1];
 498   int status;
 499
 500   if (identifier == NULL)
 501     return (EINVAL);
 502
 503   if (datadir != NULL)
 504     ssnprintf (filename, sizeof (filename), "%s/%s.rrd", datadir, identifier);
 505   else
 506     ssnprintf (filename, sizeof (filename), "%s.rrd", identifier);
 507
 508   status = rrdc_connect (daemon_address);
 509   if (status != 0)
 510   {
 511     ERROR ("rrdcached plugin: rrdc_connect (%s) failed with status %i.",
 512         daemon_address, status);
 513     return (-1);
 514   }
 515
 516   status = rrdc_flush (filename);
 517   if (status != 0)
 518   {
 519     ERROR ("rrdcached plugin: rrdc_flush (%s) failed with status %i.",
 520         filename, status);
 521     return (-1);
 522   }
 523   DEBUG ("rrdcached plugin: rrdc_flush (%s): Success.", filename);
 524
 525   return (0);
 526 } /* }}} int rc_flush */
 527
 528 static int rc_shutdown (void)
 529 {
 530   rrdc_disconnect ();
 531   return (0);
 532 } /* int rc_shutdown */
 533
 534 void module_register (void)
 535 {
 536   plugin_register_complex_config ("rrdcached", rc_config);
 537   plugin_register_init ("rrdcached", rc_init);
 538   plugin_register_shutdown ("rrdcached", rc_shutdown);
 539 } /* void module_register */
 540
 541 /*
 542  * vim: set sw=2 sts=2 et :
 543  */