src/aggregation.c

   1 /**
   2  * collectd - src/aggregation.c
   3  * Copyright (C) 2012       Florian 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 Forster <octo at collectd.org>
  25  **/
  26
  27 #include "collectd.h"
  28
  29 #include <pthread.h>
  30
  31 #include "plugin.h"
  32 #include "common.h"
  33 #include "configfile.h"
  34 #include "meta_data.h"
  35 #include "utils_cache.h" /* for uc_get_rate() */
  36 #include "utils_subst.h"
  37 #include "utils_vl_lookup.h"
  38
  39 #define AGG_MATCHES_ALL(str) (strcmp ("/.*/", str) == 0)
  40 #define AGG_FUNC_PLACEHOLDER "%{aggregation}"
  41
  42 struct aggregation_s /* {{{ */
  43 {
  44   identifier_t ident;
  45   unsigned int group_by;
  46
  47   unsigned int regex_fields;
  48
  49   char *set_host;
  50   char *set_plugin;
  51   char *set_plugin_instance;
  52   char *set_type_instance;
  53
  54   _Bool calc_num;
  55   _Bool calc_sum;
  56   _Bool calc_average;
  57   _Bool calc_min;
  58   _Bool calc_max;
  59   _Bool calc_stddev;
  60 }; /* }}} */
  61 typedef struct aggregation_s aggregation_t;
  62
  63 struct agg_instance_s;
  64 typedef struct agg_instance_s agg_instance_t;
  65 struct agg_instance_s /* {{{ */
  66 {
  67   pthread_mutex_t lock;
  68   identifier_t ident;
  69
  70   int ds_type;
  71
  72   derive_t num;
  73   gauge_t sum;
  74   gauge_t squares_sum;
  75
  76   gauge_t min;
  77   gauge_t max;
  78
  79   rate_to_value_state_t *state_num;
  80   rate_to_value_state_t *state_sum;
  81   rate_to_value_state_t *state_average;
  82   rate_to_value_state_t *state_min;
  83   rate_to_value_state_t *state_max;
  84   rate_to_value_state_t *state_stddev;
  85
  86   agg_instance_t *next;
  87 }; /* }}} */
  88
  89 static lookup_t *lookup = NULL;
  90
  91 static pthread_mutex_t agg_instance_list_lock = PTHREAD_MUTEX_INITIALIZER;
  92 static agg_instance_t *agg_instance_list_head = NULL;
  93
  94 static _Bool agg_is_regex (char const *str) /* {{{ */
  95 {
  96   size_t len;
  97
  98   if (str == NULL)
  99     return (0);
 100
 101   len = strlen (str);
 102   if (len < 3)
 103     return (0);
 104
 105   if ((str[0] == '/') && (str[len - 1] == '/'))
 106     return (1);
 107   else
 108     return (0);
 109 } /* }}} _Bool agg_is_regex */
 110
 111 static void agg_destroy (aggregation_t *agg) /* {{{ */
 112 {
 113   sfree (agg);
 114 } /* }}} void agg_destroy */
 115
 116 /* Frees all dynamically allocated memory within the instance. */
 117 static void agg_instance_destroy (agg_instance_t *inst) /* {{{ */
 118 {
 119   if (inst == NULL)
 120     return;
 121
 122   /* Remove this instance from the global list of instances. */
 123   pthread_mutex_lock (&agg_instance_list_lock);
 124   if (agg_instance_list_head == inst)
 125     agg_instance_list_head = inst->next;
 126   else if (agg_instance_list_head != NULL)
 127   {
 128     agg_instance_t *prev = agg_instance_list_head;
 129     while ((prev != NULL) && (prev->next != inst))
 130       prev = prev->next;
 131     if (prev != NULL)
 132       prev->next = inst->next;
 133   }
 134   pthread_mutex_unlock (&agg_instance_list_lock);
 135
 136   sfree (inst->state_num);
 137   sfree (inst->state_sum);
 138   sfree (inst->state_average);
 139   sfree (inst->state_min);
 140   sfree (inst->state_max);
 141   sfree (inst->state_stddev);
 142
 143   memset (inst, 0, sizeof (*inst));
 144   inst->ds_type = -1;
 145   inst->min = NAN;
 146   inst->max = NAN;
 147 } /* }}} void agg_instance_destroy */
 148
 149 static int agg_instance_create_name (agg_instance_t *inst, /* {{{ */
 150     value_list_t const *vl, aggregation_t const *agg)
 151 {
 152 #define COPY_FIELD(buffer, buffer_size, field, group_mask, all_value) do { \
 153   if (agg->set_ ## field != NULL) \
 154     sstrncpy (buffer, agg->set_ ## field, buffer_size); \
 155   else if ((agg->regex_fields & group_mask) \
 156       && (agg->group_by & group_mask)) \
 157     sstrncpy (buffer, vl->field, buffer_size); \
 158   else if ((agg->regex_fields & group_mask) \
 159       && (AGG_MATCHES_ALL (agg->ident.field))) \
 160     sstrncpy (buffer, all_value, buffer_size); \
 161   else \
 162     sstrncpy (buffer, agg->ident.field, buffer_size); \
 163 } while (0)
 164
 165   /* Host */
 166   COPY_FIELD (inst->ident.host, sizeof (inst->ident.host),
 167       host, LU_GROUP_BY_HOST, "global");
 168
 169   /* Plugin */
 170   if (agg->set_plugin != NULL)
 171     sstrncpy (inst->ident.plugin, agg->set_plugin,
 172         sizeof (inst->ident.plugin));
 173   else
 174     sstrncpy (inst->ident.plugin, "aggregation", sizeof (inst->ident.plugin));
 175
 176   /* Plugin instance */
 177   if (agg->set_plugin_instance != NULL)
 178     sstrncpy (inst->ident.plugin_instance, agg->set_plugin_instance,
 179         sizeof (inst->ident.plugin_instance));
 180   else
 181   {
 182     char tmp_plugin[DATA_MAX_NAME_LEN];
 183     char tmp_plugin_instance[DATA_MAX_NAME_LEN] = "";
 184
 185     if ((agg->regex_fields & LU_GROUP_BY_PLUGIN)
 186         && (agg->group_by & LU_GROUP_BY_PLUGIN))
 187       sstrncpy (tmp_plugin, vl->plugin, sizeof (tmp_plugin));
 188     else if ((agg->regex_fields & LU_GROUP_BY_PLUGIN)
 189         && (AGG_MATCHES_ALL (agg->ident.plugin)))
 190       sstrncpy (tmp_plugin, "", sizeof (tmp_plugin));
 191     else
 192       sstrncpy (tmp_plugin, agg->ident.plugin, sizeof (tmp_plugin));
 193
 194     if ((agg->regex_fields & LU_GROUP_BY_PLUGIN_INSTANCE)
 195         && (agg->group_by & LU_GROUP_BY_PLUGIN_INSTANCE))
 196       sstrncpy (tmp_plugin_instance, vl->plugin_instance,
 197           sizeof (tmp_plugin_instance));
 198     else if ((agg->regex_fields & LU_GROUP_BY_PLUGIN_INSTANCE)
 199         && (AGG_MATCHES_ALL (agg->ident.plugin_instance)))
 200       sstrncpy (tmp_plugin_instance, "", sizeof (tmp_plugin_instance));
 201     else
 202       sstrncpy (tmp_plugin_instance, agg->ident.plugin_instance,
 203           sizeof (tmp_plugin_instance));
 204
 205     if ((strcmp ("", tmp_plugin) == 0)
 206         && (strcmp ("", tmp_plugin_instance) == 0))
 207       sstrncpy (inst->ident.plugin_instance, AGG_FUNC_PLACEHOLDER,
 208           sizeof (inst->ident.plugin_instance));
 209     else if (strcmp ("", tmp_plugin) != 0)
 210       ssnprintf (inst->ident.plugin_instance,
 211           sizeof (inst->ident.plugin_instance),
 212           "%s-%s", tmp_plugin, AGG_FUNC_PLACEHOLDER);
 213     else if (strcmp ("", tmp_plugin_instance) != 0)
 214       ssnprintf (inst->ident.plugin_instance,
 215           sizeof (inst->ident.plugin_instance),
 216           "%s-%s", tmp_plugin_instance, AGG_FUNC_PLACEHOLDER);
 217     else
 218       ssnprintf (inst->ident.plugin_instance,
 219           sizeof (inst->ident.plugin_instance),
 220           "%s-%s-%s", tmp_plugin, tmp_plugin_instance, AGG_FUNC_PLACEHOLDER);
 221   }
 222
 223   /* Type */
 224   sstrncpy (inst->ident.type, agg->ident.type, sizeof (inst->ident.type));
 225
 226   /* Type instance */
 227   COPY_FIELD (inst->ident.type_instance, sizeof (inst->ident.type_instance),
 228       type_instance, LU_GROUP_BY_TYPE_INSTANCE, "");
 229
 230 #undef COPY_FIELD
 231
 232   return (0);
 233 } /* }}} int agg_instance_create_name */
 234
 235 /* Create a new aggregation instance. */
 236 static agg_instance_t *agg_instance_create (data_set_t const *ds, /* {{{ */
 237     value_list_t const *vl, aggregation_t *agg)
 238 {
 239   agg_instance_t *inst;
 240
 241   DEBUG ("aggregation plugin: Creating new instance.");
 242
 243   inst = malloc (sizeof (*inst));
 244   if (inst == NULL)
 245   {
 246     ERROR ("aggregation plugin: malloc() failed.");
 247     return (NULL);
 248   }
 249   memset (inst, 0, sizeof (*inst));
 250   pthread_mutex_init (&inst->lock, /* attr = */ NULL);
 251
 252   inst->ds_type = ds->ds[0].type;
 253
 254   agg_instance_create_name (inst, vl, agg);
 255
 256   inst->min = NAN;
 257   inst->max = NAN;
 258
 259 #define INIT_STATE(field) do { \
 260   inst->state_ ## field = NULL; \
 261   if (agg->calc_ ## field) { \
 262     inst->state_ ## field = malloc (sizeof (*inst->state_ ## field)); \
 263     if (inst->state_ ## field == NULL) { \
 264       agg_instance_destroy (inst); \
 265       ERROR ("aggregation plugin: malloc() failed."); \
 266       return (NULL); \
 267     } \
 268     memset (inst->state_ ## field, 0, sizeof (*inst->state_ ## field)); \
 269   } \
 270 } while (0)
 271
 272   INIT_STATE (num);
 273   INIT_STATE (sum);
 274   INIT_STATE (average);
 275   INIT_STATE (min);
 276   INIT_STATE (max);
 277   INIT_STATE (stddev);
 278
 279 #undef INIT_STATE
 280
 281   pthread_mutex_lock (&agg_instance_list_lock);
 282   inst->next = agg_instance_list_head;
 283   agg_instance_list_head = inst;
 284   pthread_mutex_unlock (&agg_instance_list_lock);
 285
 286   return (inst);
 287 } /* }}} agg_instance_t *agg_instance_create */
 288
 289 /* Update the num, sum, min, max, ... fields of the aggregation instance, if
 290  * the rate of the value list is available. Value lists with more than one data
 291  * source are not supported and will return an error. Returns zero on success
 292  * and non-zero otherwise. */
 293 static int agg_instance_update (agg_instance_t *inst, /* {{{ */
 294     data_set_t const *ds, value_list_t const *vl)
 295 {
 296   gauge_t *rate;
 297
 298   if (ds->ds_num != 1)
 299   {
 300     ERROR ("aggregation plugin: The \"%s\" type (data set) has more than one "
 301         "data source. This is currently not supported by this plugin. "
 302         "Sorry.", ds->type);
 303     return (EINVAL);
 304   }
 305
 306   rate = uc_get_rate (ds, vl);
 307   if (rate == NULL)
 308   {
 309     char ident[6 * DATA_MAX_NAME_LEN];
 310     FORMAT_VL (ident, sizeof (ident), vl);
 311     ERROR ("aggregation plugin: Unable to read the current rate of \"%s\".",
 312         ident);
 313     return (ENOENT);
 314   }
 315
 316   if (isnan (rate[0]))
 317   {
 318     sfree (rate);
 319     return (0);
 320   }
 321
 322   pthread_mutex_lock (&inst->lock);
 323
 324   inst->num++;
 325   inst->sum += rate[0];
 326   inst->squares_sum += (rate[0] * rate[0]);
 327
 328   if (isnan (inst->min) || (inst->min > rate[0]))
 329     inst->min = rate[0];
 330   if (isnan (inst->max) || (inst->max < rate[0]))
 331     inst->max = rate[0];
 332
 333   pthread_mutex_unlock (&inst->lock);
 334
 335   sfree (rate);
 336   return (0);
 337 } /* }}} int agg_instance_update */
 338
 339 static int agg_instance_read_func (agg_instance_t *inst, /* {{{ */
 340   char const *func, gauge_t rate, rate_to_value_state_t *state,
 341   value_list_t *vl, char const *pi_prefix, cdtime_t t)
 342 {
 343   value_t v;
 344   int status;
 345
 346   if (pi_prefix[0] != 0)
 347     subst_string (vl->plugin_instance, sizeof (vl->plugin_instance),
 348         pi_prefix, AGG_FUNC_PLACEHOLDER, func);
 349   else
 350     sstrncpy (vl->plugin_instance, func, sizeof (vl->plugin_instance));
 351
 352   memset (&v, 0, sizeof (v));
 353   status = rate_to_value (&v, rate, state, inst->ds_type, t);
 354   if (status != 0)
 355   {
 356     /* If this is the first iteration and rate_to_value() was asked to return a
 357      * COUNTER or a DERIVE, it will return EAGAIN. Catch this and handle
 358      * gracefully. */
 359     if (status == EAGAIN)
 360       return (0);
 361
 362     WARNING ("aggregation plugin: rate_to_value failed with status %i.",
 363         status);
 364     return (-1);
 365   }
 366
 367   vl->values = &v;
 368   vl->values_len = 1;
 369
 370   plugin_dispatch_values (vl);
 371
 372   vl->values = NULL;
 373   vl->values_len = 0;
 374
 375   return (0);
 376 } /* }}} int agg_instance_read_func */
 377
 378 static int agg_instance_read (agg_instance_t *inst, cdtime_t t) /* {{{ */
 379 {
 380   value_list_t vl = VALUE_LIST_INIT;
 381
 382   /* Pre-set all the fields in the value list that will not change per
 383    * aggregation type (sum, average, ...). The struct will be re-used and must
 384    * therefore be dispatched using the "secure" function. */
 385
 386   vl.time = t;
 387   vl.interval = 0;
 388
 389   vl.meta = meta_data_create ();
 390   if (vl.meta == NULL)
 391   {
 392     ERROR ("aggregation plugin: meta_data_create failed.");
 393     return (-1);
 394   }
 395   meta_data_add_boolean (vl.meta, "aggregation:created", 1);
 396
 397   sstrncpy (vl.host, inst->ident.host, sizeof (vl.host));
 398   sstrncpy (vl.plugin, inst->ident.plugin, sizeof (vl.plugin));
 399   sstrncpy (vl.type, inst->ident.type, sizeof (vl.type));
 400   sstrncpy (vl.type_instance, inst->ident.type_instance,
 401       sizeof (vl.type_instance));
 402
 403 #define READ_FUNC(func, rate) do { \
 404   if (inst->state_ ## func != NULL) { \
 405     agg_instance_read_func (inst, #func, rate, \
 406         inst->state_ ## func, &vl, inst->ident.plugin_instance, t); \
 407   } \
 408 } while (0)
 409
 410   pthread_mutex_lock (&inst->lock);
 411
 412   READ_FUNC (num, (gauge_t) inst->num);
 413
 414   /* All other aggregations are only defined when there have been any values
 415    * at all. */
 416   if (inst->num > 0)
 417   {
 418     READ_FUNC (sum, inst->sum);
 419     READ_FUNC (average, (inst->sum / ((gauge_t) inst->num)));
 420     READ_FUNC (min, inst->min);
 421     READ_FUNC (max, inst->max);
 422     READ_FUNC (stddev, sqrt((((gauge_t) inst->num) * inst->squares_sum)
 423           - (inst->sum * inst->sum)) / ((gauge_t) inst->num));
 424   }
 425
 426   /* Reset internal state. */
 427   inst->num = 0;
 428   inst->sum = 0.0;
 429   inst->squares_sum = 0.0;
 430   inst->min = NAN;
 431   inst->max = NAN;
 432
 433   pthread_mutex_unlock (&inst->lock);
 434
 435   meta_data_destroy (vl.meta);
 436   vl.meta = NULL;
 437
 438   return (0);
 439 } /* }}} int agg_instance_read */
 440
 441 /* lookup_class_callback_t for utils_vl_lookup */
 442 static void *agg_lookup_class_callback ( /* {{{ */
 443     __attribute__((unused)) data_set_t const *ds,
 444     value_list_t const *vl, void *user_class)
 445 {
 446   return (agg_instance_create (ds, vl, (aggregation_t *) user_class));
 447 } /* }}} void *agg_class_callback */
 448
 449 /* lookup_obj_callback_t for utils_vl_lookup */
 450 static int agg_lookup_obj_callback (data_set_t const *ds, /* {{{ */
 451     value_list_t const *vl,
 452     __attribute__((unused)) void *user_class,
 453     void *user_obj)
 454 {
 455   return (agg_instance_update ((agg_instance_t *) user_obj, ds, vl));
 456 } /* }}} int agg_lookup_obj_callback */
 457
 458 /* lookup_free_class_callback_t for utils_vl_lookup */
 459 static void agg_lookup_free_class_callback (void *user_class) /* {{{ */
 460 {
 461   agg_destroy ((aggregation_t *) user_class);
 462 } /* }}} void agg_lookup_free_class_callback */
 463
 464 /* lookup_free_obj_callback_t for utils_vl_lookup */
 465 static void agg_lookup_free_obj_callback (void *user_obj) /* {{{ */
 466 {
 467   agg_instance_destroy ((agg_instance_t *) user_obj);
 468 } /* }}} void agg_lookup_free_obj_callback */
 469
 470 /*
 471  * <Plugin "aggregation">
 472  *   <Aggregation>
 473  *     Plugin "cpu"
 474  *     Type "cpu"
 475  *
 476  *     GroupBy Host
 477  *     GroupBy TypeInstance
 478  *
 479  *     CalculateNum true
 480  *     CalculateSum true
 481  *     CalculateAverage true
 482  *     CalculateMinimum true
 483  *     CalculateMaximum true
 484  *     CalculateStddev true
 485  *   </Aggregation>
 486  * </Plugin>
 487  */
 488 static int agg_config_handle_group_by (oconfig_item_t const *ci, /* {{{ */
 489     aggregation_t *agg)
 490 {
 491   int i;
 492
 493   for (i = 0; i < ci->values_num; i++)
 494   {
 495     char const *value;
 496
 497     if (ci->values[i].type != OCONFIG_TYPE_STRING)
 498     {
 499       ERROR ("aggregation plugin: Argument %i of the \"GroupBy\" option "
 500           "is not a string.", i + 1);
 501       continue;
 502     }
 503
 504     value = ci->values[i].value.string;
 505
 506     if (strcasecmp ("Host", value) == 0)
 507       agg->group_by |= LU_GROUP_BY_HOST;
 508     else if (strcasecmp ("Plugin", value) == 0)
 509       agg->group_by |= LU_GROUP_BY_PLUGIN;
 510     else if (strcasecmp ("PluginInstance", value) == 0)
 511       agg->group_by |= LU_GROUP_BY_PLUGIN_INSTANCE;
 512     else if (strcasecmp ("TypeInstance", value) == 0)
 513       agg->group_by |= LU_GROUP_BY_TYPE_INSTANCE;
 514     else if (strcasecmp ("Type", value) == 0)
 515       ERROR ("aggregation plugin: Grouping by type is not supported.");
 516     else
 517       WARNING ("aggregation plugin: The \"%s\" argument to the \"GroupBy\" "
 518           "option is invalid and will be ignored.", value);
 519   } /* for (ci->values) */
 520
 521   return (0);
 522 } /* }}} int agg_config_handle_group_by */
 523
 524 static int agg_config_aggregation (oconfig_item_t *ci) /* {{{ */
 525 {
 526   aggregation_t *agg;
 527   _Bool is_valid;
 528   int status;
 529   int i;
 530
 531   agg = malloc (sizeof (*agg));
 532   if (agg == NULL)
 533   {
 534     ERROR ("aggregation plugin: malloc failed.");
 535     return (-1);
 536   }
 537   memset (agg, 0, sizeof (*agg));
 538
 539   sstrncpy (agg->ident.host, "/.*/", sizeof (agg->ident.host));
 540   sstrncpy (agg->ident.plugin, "/.*/", sizeof (agg->ident.plugin));
 541   sstrncpy (agg->ident.plugin_instance, "/.*/",
 542       sizeof (agg->ident.plugin_instance));
 543   sstrncpy (agg->ident.type, "/.*/", sizeof (agg->ident.type));
 544   sstrncpy (agg->ident.type_instance, "/.*/",
 545       sizeof (agg->ident.type_instance));
 546
 547   for (i = 0; i < ci->children_num; i++)
 548   {
 549     oconfig_item_t *child = ci->children + i;
 550
 551     if (strcasecmp ("Host", child->key) == 0)
 552       cf_util_get_string_buffer (child, agg->ident.host,
 553           sizeof (agg->ident.host));
 554     else if (strcasecmp ("Plugin", child->key) == 0)
 555       cf_util_get_string_buffer (child, agg->ident.plugin,
 556           sizeof (agg->ident.plugin));
 557     else if (strcasecmp ("PluginInstance", child->key) == 0)
 558       cf_util_get_string_buffer (child, agg->ident.plugin_instance,
 559           sizeof (agg->ident.plugin_instance));
 560     else if (strcasecmp ("Type", child->key) == 0)
 561       cf_util_get_string_buffer (child, agg->ident.type,
 562           sizeof (agg->ident.type));
 563     else if (strcasecmp ("TypeInstance", child->key) == 0)
 564       cf_util_get_string_buffer (child, agg->ident.type_instance,
 565           sizeof (agg->ident.type_instance));
 566     else if (strcasecmp ("SetHost", child->key) == 0)
 567       cf_util_get_string (child, &agg->set_host);
 568     else if (strcasecmp ("SetPlugin", child->key) == 0)
 569       cf_util_get_string (child, &agg->set_plugin);
 570     else if (strcasecmp ("SetPluginInstance", child->key) == 0)
 571       cf_util_get_string (child, &agg->set_plugin_instance);
 572     else if (strcasecmp ("SetTypeInstance", child->key) == 0)
 573       cf_util_get_string (child, &agg->set_type_instance);
 574     else if (strcasecmp ("GroupBy", child->key) == 0)
 575       agg_config_handle_group_by (child, agg);
 576     else if (strcasecmp ("CalculateNum", child->key) == 0)
 577       cf_util_get_boolean (child, &agg->calc_num);
 578     else if (strcasecmp ("CalculateSum", child->key) == 0)
 579       cf_util_get_boolean (child, &agg->calc_sum);
 580     else if (strcasecmp ("CalculateAverage", child->key) == 0)
 581       cf_util_get_boolean (child, &agg->calc_average);
 582     else if (strcasecmp ("CalculateMinimum", child->key) == 0)
 583       cf_util_get_boolean (child, &agg->calc_min);
 584     else if (strcasecmp ("CalculateMaximum", child->key) == 0)
 585       cf_util_get_boolean (child, &agg->calc_max);
 586     else if (strcasecmp ("CalculateStddev", child->key) == 0)
 587       cf_util_get_boolean (child, &agg->calc_stddev);
 588     else
 589       WARNING ("aggregation plugin: The \"%s\" key is not allowed inside "
 590           "<Aggregation /> blocks and will be ignored.", child->key);
 591   }
 592
 593   if (agg_is_regex (agg->ident.host))
 594     agg->regex_fields |= LU_GROUP_BY_HOST;
 595   if (agg_is_regex (agg->ident.plugin))
 596     agg->regex_fields |= LU_GROUP_BY_PLUGIN;
 597   if (agg_is_regex (agg->ident.plugin_instance))
 598     agg->regex_fields |= LU_GROUP_BY_PLUGIN_INSTANCE;
 599   if (agg_is_regex (agg->ident.type_instance))
 600     agg->regex_fields |= LU_GROUP_BY_TYPE_INSTANCE;
 601
 602   /* Sanity checking */
 603   is_valid = 1;
 604   if (strcmp ("/.*/", agg->ident.type) == 0) /* {{{ */
 605   {
 606     ERROR ("aggregation plugin: It appears you did not specify the required "
 607         "\"Type\" option in this aggregation. "
 608         "(Host \"%s\", Plugin \"%s\", PluginInstance \"%s\", "
 609         "Type \"%s\", TypeInstance \"%s\")",
 610         agg->ident.host, agg->ident.plugin, agg->ident.plugin_instance,
 611         agg->ident.type, agg->ident.type_instance);
 612     is_valid = 0;
 613   }
 614   else if (strchr (agg->ident.type, '/') != NULL)
 615   {
 616     ERROR ("aggregation plugin: The \"Type\" may not contain the '/' "
 617         "character. Especially, it may not be a regex. The current "
 618         "value is \"%s\".", agg->ident.type);
 619     is_valid = 0;
 620   } /* }}} */
 621
 622   /* Check that there is at least one regex field without a grouping. {{{ */
 623   if ((agg->regex_fields & ~agg->group_by) == 0)
 624   {
 625     ERROR ("aggregation plugin: An aggregation must contain at least one "
 626         "wildcard. This is achieved by leaving at least one of the \"Host\", "
 627         "\"Plugin\", \"PluginInstance\" and \"TypeInstance\" options blank "
 628         "or using a regular expression and not grouping by that field. "
 629         "(Host \"%s\", Plugin \"%s\", PluginInstance \"%s\", "
 630         "Type \"%s\", TypeInstance \"%s\")",
 631         agg->ident.host, agg->ident.plugin, agg->ident.plugin_instance,
 632         agg->ident.type, agg->ident.type_instance);
 633     is_valid = 0;
 634   } /* }}} */
 635
 636   /* Check that all grouping fields are regular expressions. {{{ */
 637   if (agg->group_by & ~agg->regex_fields)
 638   {
 639     ERROR ("aggregation plugin: Only wildcard fields (fields for which a "
 640         "regular expression is configured or which are left blank) can be "
 641         "specified in the \"GroupBy\" option. "
 642         "(Host \"%s\", Plugin \"%s\", PluginInstance \"%s\", "
 643         "Type \"%s\", TypeInstance \"%s\")",
 644         agg->ident.host, agg->ident.plugin, agg->ident.plugin_instance,
 645         agg->ident.type, agg->ident.type_instance);
 646     is_valid = 0;
 647   } /* }}} */
 648
 649   if (!agg->calc_num && !agg->calc_sum && !agg->calc_average /* {{{ */
 650       && !agg->calc_min && !agg->calc_max && !agg->calc_stddev)
 651   {
 652     ERROR ("aggregation plugin: No aggregation function has been specified. "
 653         "Without this, I don't know what I should be calculating. "
 654         "(Host \"%s\", Plugin \"%s\", PluginInstance \"%s\", "
 655         "Type \"%s\", TypeInstance \"%s\")",
 656         agg->ident.host, agg->ident.plugin, agg->ident.plugin_instance,
 657         agg->ident.type, agg->ident.type_instance);
 658     is_valid = 0;
 659   } /* }}} */
 660
 661   if (!is_valid) /* {{{ */
 662   {
 663     sfree (agg);
 664     return (-1);
 665   } /* }}} */
 666
 667   status = lookup_add (lookup, &agg->ident, agg->group_by, agg);
 668   if (status != 0)
 669   {
 670     ERROR ("aggregation plugin: lookup_add failed with status %i.", status);
 671     sfree (agg);
 672     return (-1);
 673   }
 674
 675   DEBUG ("aggregation plugin: Successfully added aggregation: "
 676       "(Host \"%s\", Plugin \"%s\", PluginInstance \"%s\", "
 677       "Type \"%s\", TypeInstance \"%s\")",
 678       agg->ident.host, agg->ident.plugin, agg->ident.plugin_instance,
 679       agg->ident.type, agg->ident.type_instance);
 680   return (0);
 681 } /* }}} int agg_config_aggregation */
 682
 683 static int agg_config (oconfig_item_t *ci) /* {{{ */
 684 {
 685   int i;
 686
 687   pthread_mutex_lock (&agg_instance_list_lock);
 688
 689   if (lookup == NULL)
 690   {
 691     lookup = lookup_create (agg_lookup_class_callback,
 692         agg_lookup_obj_callback,
 693         agg_lookup_free_class_callback,
 694         agg_lookup_free_obj_callback);
 695     if (lookup == NULL)
 696     {
 697       pthread_mutex_unlock (&agg_instance_list_lock);
 698       ERROR ("aggregation plugin: lookup_create failed.");
 699       return (-1);
 700     }
 701   }
 702
 703   for (i = 0; i < ci->children_num; i++)
 704   {
 705     oconfig_item_t *child = ci->children + i;
 706
 707     if (strcasecmp ("Aggregation", child->key) == 0)
 708       agg_config_aggregation (child);
 709     else
 710       WARNING ("aggregation plugin: The \"%s\" key is not allowed inside "
 711           "<Plugin aggregation /> blocks and will be ignored.", child->key);
 712   }
 713
 714   pthread_mutex_unlock (&agg_instance_list_lock);
 715
 716   return (0);
 717 } /* }}} int agg_config */
 718
 719 static int agg_read (void) /* {{{ */
 720 {
 721   agg_instance_t *this;
 722   cdtime_t t;
 723   int success;
 724
 725   t = cdtime ();
 726   success = 0;
 727
 728   pthread_mutex_lock (&agg_instance_list_lock);
 729
 730   /* agg_instance_list_head only holds data, after the "write" callback has
 731    * been called with a matching value list at least once. So on startup,
 732    * there's a race between the aggregations read() and write() callback. If
 733    * the read() callback is called first, agg_instance_list_head is NULL and
 734    * "success" may be zero. This is expected and should not result in an error.
 735    * Therefore we need to handle this case separately. */
 736   if (agg_instance_list_head == NULL)
 737   {
 738     pthread_mutex_unlock (&agg_instance_list_lock);
 739     return (0);
 740   }
 741
 742   for (this = agg_instance_list_head; this != NULL; this = this->next)
 743   {
 744     int status;
 745
 746     status = agg_instance_read (this, t);
 747     if (status != 0)
 748       WARNING ("aggregation plugin: Reading an aggregation instance "
 749           "failed with status %i.", status);
 750     else
 751       success++;
 752   }
 753
 754   pthread_mutex_unlock (&agg_instance_list_lock);
 755
 756   return ((success > 0) ? 0 : -1);
 757 } /* }}} int agg_read */
 758
 759 static int agg_write (data_set_t const *ds, value_list_t const *vl, /* {{{ */
 760     __attribute__((unused)) user_data_t *user_data)
 761 {
 762   _Bool created_by_aggregation = 0;
 763   int status;
 764
 765   /* Ignore values that were created by the aggregation plugin to avoid weird
 766    * effects. */
 767   (void) meta_data_get_boolean (vl->meta, "aggregation:created",
 768       &created_by_aggregation);
 769   if (created_by_aggregation)
 770     return (0);
 771
 772   if (lookup == NULL)
 773     status = ENOENT;
 774   else
 775   {
 776     status = lookup_search (lookup, ds, vl);
 777     if (status > 0)
 778       status = 0;
 779   }
 780
 781   return (status);
 782 } /* }}} int agg_write */
 783
 784 void module_register (void)
 785 {
 786   plugin_register_complex_config ("aggregation", agg_config);
 787   plugin_register_read ("aggregation", agg_read);
 788   plugin_register_write ("aggregation", agg_write, /* user_data = */ NULL);
 789 }
 790
 791 /* vim: set sw=2 sts=2 tw=78 et fdm=marker : */