src/rrdcached.c

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