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