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