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