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