X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Futils_cache.c;h=aeb662d55bc09e63c18ae856ac182aa04fc6b369;hb=d7ca6a64067a4e01a5e47044a6a80323d0138bee;hp=d0e77ef351b840a79fcd91123264c3b43670dbb2;hpb=39ba035b60b509311874ccf89d11d240c0271a90;p=collectd.git diff --git a/src/utils_cache.c b/src/utils_cache.c index d0e77ef3..aeb662d5 100644 --- a/src/utils_cache.c +++ b/src/utils_cache.c @@ -25,6 +25,7 @@ #include "utils_avltree.h" #include "utils_cache.h" #include "utils_threshold.h" +#include "meta_data.h" #include #include @@ -34,7 +35,7 @@ typedef struct cache_entry_s char name[6 * DATA_MAX_NAME_LEN]; int values_num; gauge_t *values_gauge; - counter_t *values_counter; + value_t *values_raw; /* Time contained in the package * (for calculating rates) */ time_t last_time; @@ -45,6 +46,22 @@ typedef struct cache_entry_s * (for purding old entries) */ int interval; int state; + int hits; + + /* + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! 0 ! 1 ! 2 ! 3 ! 4 ! 5 ! 6 ! 7 ! 8 ! ... + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ... + * +-----+-----+-----+-----+-----+-----+-----+-----+-----+---- + * ! t = 0 ! t = 1 ! t = 2 ! ... + * +-----------------+-----------------+-----------------+---- + */ + gauge_t *history; + size_t history_index; /* points to the next position to write to. */ + size_t history_length; + + meta_data_t *meta; } cache_entry_t; static c_avl_tree_t *cache_tree = NULL; @@ -69,17 +86,21 @@ static cache_entry_t *cache_alloc (int values_num) memset (ce, '\0', sizeof (cache_entry_t)); ce->values_num = values_num; - ce->values_gauge = (gauge_t *) calloc (values_num, sizeof (gauge_t)); - ce->values_counter = (counter_t *) calloc (values_num, sizeof (counter_t)); - if ((ce->values_gauge == NULL) || (ce->values_counter == NULL)) + ce->values_gauge = calloc (values_num, sizeof (*ce->values_gauge)); + ce->values_raw = calloc (values_num, sizeof (*ce->values_raw)); + if ((ce->values_gauge == NULL) || (ce->values_raw == NULL)) { sfree (ce->values_gauge); - sfree (ce->values_counter); + sfree (ce->values_raw); sfree (ce); ERROR ("utils_cache: cache_alloc: calloc failed."); return (NULL); } + ce->history = NULL; + ce->history_length = 0; + ce->meta = NULL; + return (ce); } /* cache_entry_t *cache_alloc */ @@ -89,7 +110,13 @@ static void cache_free (cache_entry_t *ce) return; sfree (ce->values_gauge); - sfree (ce->values_counter); + sfree (ce->values_raw); + sfree (ce->history); + if (ce->meta != NULL) + { + meta_data_destroy (ce->meta); + ce->meta = NULL; + } sfree (ce); } /* void cache_free */ @@ -148,7 +175,7 @@ static int uc_send_notification (const char *name) } /* Check if the entry has been updated in the meantime */ - if ((n.time - ce->last_update) < (2 * ce->interval)) + if ((n.time - ce->last_update) < (timeout_g * ce->interval)) { ce->state = STATE_OKAY; pthread_mutex_unlock (&cache_lock); @@ -167,6 +194,21 @@ static int uc_send_notification (const char *name) return (0); } /* int uc_send_notification */ +static void uc_check_range (const data_set_t *ds, cache_entry_t *ce) +{ + int i; + + for (i = 0; i < ds->ds_num; i++) + { + if (isnan (ce->values_gauge[i])) + continue; + else if (ce->values_gauge[i] < ds->ds[i].min) + ce->values_gauge[i] = NAN; + else if (ce->values_gauge[i] > ds->ds[i].max) + ce->values_gauge[i] = NAN; + } +} /* void uc_check_range */ + static int uc_insert (const data_set_t *ds, const value_list_t *vl, const char *key) { @@ -186,6 +228,7 @@ static int uc_insert (const data_set_t *ds, const value_list_t *vl, ce = cache_alloc (ds->ds_num); if (ce == NULL) { + sfree (key_copy); ERROR ("uc_insert: cache_alloc (%i) failed.", ds->ds_num); return (-1); } @@ -194,17 +237,42 @@ static int uc_insert (const data_set_t *ds, const value_list_t *vl, for (i = 0; i < ds->ds_num; i++) { - if (ds->ds[i].type == DS_TYPE_COUNTER) - { - ce->values_gauge[i] = NAN; - ce->values_counter[i] = vl->values[i].counter; - } - else /* if (ds->ds[i].type == DS_TYPE_GAUGE) */ + switch (ds->ds[i].type) { - ce->values_gauge[i] = vl->values[i].gauge; - } + case DS_TYPE_COUNTER: + ce->values_gauge[i] = NAN; + ce->values_raw[i].counter = vl->values[i].counter; + break; + + case DS_TYPE_GAUGE: + ce->values_gauge[i] = vl->values[i].gauge; + ce->values_raw[i].gauge = vl->values[i].gauge; + break; + + case DS_TYPE_DERIVE: + ce->values_gauge[i] = NAN; + ce->values_raw[i].derive = vl->values[i].derive; + break; + + case DS_TYPE_ABSOLUTE: + ce->values_gauge[i] = NAN; + if (vl->interval > 0) + ce->values_gauge[i] = ((double) vl->values[i].absolute) + / ((double) vl->interval); + ce->values_raw[i].absolute = vl->values[i].absolute; + break; + + default: + /* This shouldn't happen. */ + ERROR ("uc_insert: Don't know how to handle data source type %i.", + ds->ds[i].type); + return (-1); + } /* switch (ds->ds[i].type) */ } /* for (i) */ + /* Prune invalid gauge data */ + uc_check_range (ds, ce); + ce->last_time = vl->time; ce->last_update = time (NULL); ce->interval = vl->interval; @@ -251,7 +319,7 @@ int uc_check_timeout (void) while (c_avl_iterator_next (iter, (void *) &key, (void *) &ce) == 0) { /* If entry has not been updated, add to `keys' array */ - if ((now - ce->last_update) >= (2 * ce->interval)) + if ((now - ce->last_update) >= (timeout_g * ce->interval)) { char **tmp; @@ -259,8 +327,10 @@ int uc_check_timeout (void) (keys_len + 1) * sizeof (char *)); if (tmp == NULL) { - ERROR ("uc_purge: realloc failed."); + ERROR ("uc_check_timeout: realloc failed."); c_avl_iterator_destroy (iter); + sfree (keys); + pthread_mutex_unlock (&cache_lock); return (-1); } @@ -275,6 +345,8 @@ int uc_check_timeout (void) } } /* while (c_avl_iterator_next) */ + ce = NULL; + for (i = 0; i < keys_len; i++) { int status; @@ -285,12 +357,13 @@ int uc_check_timeout (void) { ERROR ("uc_check_timeout: ut_check_interesting failed."); sfree (keys[i]); + continue; } else if (status == 0) /* ``service'' is uninteresting */ { - ce = NULL; DEBUG ("uc_check_timeout: %s is missing but ``uninteresting''", keys[i]); + ce = NULL; status = c_avl_remove (cache_tree, keys[i], (void *) &key, (void *) &ce); if (status != 0) @@ -298,21 +371,37 @@ int uc_check_timeout (void) ERROR ("uc_check_timeout: c_avl_remove (%s) failed.", keys[i]); } sfree (keys[i]); - cache_free (ce); + sfree (key); + if (ce != NULL) + cache_free (ce); + continue; } - else if (status == 1) /* persist */ + + /* If we get here, the value is ``interesting''. Query the record from the + * cache and update the state field. */ + if (c_avl_get (cache_tree, keys[i], (void *) &ce) != 0) + { + ERROR ("uc_check_timeout: cannot get data for %s from cache", keys[i]); + /* Do not free `keys[i]' so a notification is sent further down. */ + continue; + } + assert (ce != NULL); + + if (status == 2) /* persist */ { DEBUG ("uc_check_timeout: %s is missing, sending notification.", keys[i]); ce->state = STATE_MISSING; + /* Do not free `keys[i]' so a notification is sent further down. */ } - else if (status == 2) /* do not persist */ + else if (status == 1) /* do not persist */ { if (ce->state == STATE_MISSING) { DEBUG ("uc_check_timeout: %s is missing but " "notification has already been sent.", keys[i]); + /* Set `keys[i]' to NULL to no notification is sent. */ sfree (keys[i]); } else /* (ce->state != STATE_MISSING) */ @@ -320,6 +409,7 @@ int uc_check_timeout (void) DEBUG ("uc_check_timeout: %s is missing, sending one notification.", keys[i]); ce->state = STATE_MISSING; + /* Do not free `keys[i]' so a notification is sent further down. */ } } else @@ -327,7 +417,12 @@ int uc_check_timeout (void) WARNING ("uc_check_timeout: ut_check_interesting (%s) returned " "invalid status %i.", keys[i], status); + sfree (keys[i]); } + + /* Make really sure the next iteration doesn't work with this pointer. + * There have been too many bugs in the past.. :/ -- octo */ + ce = NULL; } /* for (keys[i]) */ c_avl_iterator_destroy (iter); @@ -358,7 +453,7 @@ int uc_update (const data_set_t *ds, const value_list_t *vl) int status; int i; - if (FORMAT_VL (name, sizeof (name), vl, ds) != 0) + if (FORMAT_VL (name, sizeof (name), vl) != 0) { ERROR ("uc_update: FORMAT_VL failed."); return (-1); @@ -397,36 +492,84 @@ int uc_update (const data_set_t *ds, const value_list_t *vl) for (i = 0; i < ds->ds_num; i++) { - if (ds->ds[i].type == DS_TYPE_COUNTER) + switch (ds->ds[i].type) { - counter_t diff; + case DS_TYPE_COUNTER: + { + counter_t diff; + + /* check if the counter has wrapped around */ + if (vl->values[i].counter < ce->values_raw[i].counter) + { + if (ce->values_raw[i].counter <= 4294967295U) + diff = (4294967295U - ce->values_raw[i].counter) + + vl->values[i].counter; + else + diff = (18446744073709551615ULL - ce->values_raw[i].counter) + + vl->values[i].counter; + } + else /* counter has NOT wrapped around */ + { + diff = vl->values[i].counter - ce->values_raw[i].counter; + } + + ce->values_gauge[i] = ((double) diff) + / ((double) (vl->time - ce->last_time)); + ce->values_raw[i].counter = vl->values[i].counter; + } + break; - /* check if the counter has wrapped around */ - if (vl->values[i].counter < ce->values_counter[i]) - { - if (ce->values_counter[i] <= 4294967295U) - diff = (4294967295U - ce->values_counter[i]) - + vl->values[i].counter; - else - diff = (18446744073709551615ULL - ce->values_counter[i]) - + vl->values[i].counter; - } - else /* counter has NOT wrapped around */ - { - diff = vl->values[i].counter - ce->values_counter[i]; - } + case DS_TYPE_GAUGE: + ce->values_raw[i].gauge = vl->values[i].gauge; + ce->values_gauge[i] = vl->values[i].gauge; + break; + + case DS_TYPE_DERIVE: + { + derive_t diff; + + diff = vl->values[i].derive - ce->values_raw[i].derive; + + ce->values_gauge[i] = ((double) diff) + / ((double) (vl->time - ce->last_time)); + ce->values_raw[i].derive = vl->values[i].derive; + } + break; + + case DS_TYPE_ABSOLUTE: + ce->values_gauge[i] = ((double) vl->values[i].absolute) + / ((double) (vl->time - ce->last_time)); + ce->values_raw[i].absolute = vl->values[i].absolute; + break; + + default: + /* This shouldn't happen. */ + pthread_mutex_unlock (&cache_lock); + ERROR ("uc_update: Don't know how to handle data source type %i.", + ds->ds[i].type); + return (-1); + } /* switch (ds->ds[i].type) */ - ce->values_gauge[i] = ((double) diff) - / ((double) (vl->time - ce->last_time)); - ce->values_counter[i] = vl->values[i].counter; - } - else /* if (ds->ds[i].type == DS_TYPE_GAUGE) */ - { - ce->values_gauge[i] = vl->values[i].gauge; - } DEBUG ("uc_update: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]); } /* for (i) */ + /* Update the history if it exists. */ + if (ce->history != NULL) + { + assert (ce->history_index < ce->history_length); + for (i = 0; i < ce->values_num; i++) + { + size_t hist_idx = (ce->values_num * ce->history_index) + i; + ce->history[hist_idx] = ce->values_gauge[i]; + } + + assert (ce->history_length > 0); + ce->history_index = (ce->history_index + 1) % ce->history_length; + } + + /* Prune invalid gauge data */ + uc_check_range (ds, ce); + ce->last_time = vl->time; ce->last_update = time (NULL); ce->interval = vl->interval; @@ -471,16 +614,24 @@ int uc_get_rate_by_name (const char *name, gauge_t **ret_values, size_t *ret_val { assert (ce != NULL); - ret_num = ce->values_num; - ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t)); - if (ret == NULL) + /* remove missing values from getval */ + if (ce->state == STATE_MISSING) { - ERROR ("utils_cache: uc_get_rate_by_name: malloc failed."); status = -1; } else { - memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t)); + ret_num = ce->values_num; + ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t)); + if (ret == NULL) + { + ERROR ("utils_cache: uc_get_rate_by_name: malloc failed."); + status = -1; + } + else + { + memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t)); + } } } else @@ -507,9 +658,9 @@ gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl) size_t ret_num = 0; int status; - if (FORMAT_VL (name, sizeof (name), vl, ds) != 0) + if (FORMAT_VL (name, sizeof (name), vl) != 0) { - ERROR ("uc_insert: FORMAT_VL failed."); + ERROR ("utils_cache: uc_get_rate: FORMAT_VL failed."); return (NULL); } @@ -519,7 +670,7 @@ gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl) /* This is important - the caller has no other way of knowing how many * values are returned. */ - if (ret_num != ds->ds_num) + if (ret_num != (size_t) ds->ds_num) { ERROR ("utils_cache: uc_get_rate: ds[%s] has %i values, " "but uc_get_rate_by_name returned %zu.", @@ -553,6 +704,10 @@ int uc_get_names (char ***ret_names, time_t **ret_times, size_t *ret_number) { char **temp; + /* remove missing values when list values */ + if (value->state == STATE_MISSING) + continue; + if (ret_times != NULL) { time_t *tmp_times; @@ -613,7 +768,7 @@ int uc_get_state (const data_set_t *ds, const value_list_t *vl) cache_entry_t *ce = NULL; int ret = STATE_ERROR; - if (FORMAT_VL (name, sizeof (name), vl, ds) != 0) + if (FORMAT_VL (name, sizeof (name), vl) != 0) { ERROR ("uc_get_state: FORMAT_VL failed."); return (STATE_ERROR); @@ -638,7 +793,7 @@ int uc_set_state (const data_set_t *ds, const value_list_t *vl, int state) cache_entry_t *ce = NULL; int ret = -1; - if (FORMAT_VL (name, sizeof (name), vl, ds) != 0) + if (FORMAT_VL (name, sizeof (name), vl) != 0) { ERROR ("uc_get_state: FORMAT_VL failed."); return (STATE_ERROR); @@ -657,4 +812,273 @@ int uc_set_state (const data_set_t *ds, const value_list_t *vl, int state) return (ret); } /* int uc_set_state */ + +int uc_get_history_by_name (const char *name, + gauge_t *ret_history, size_t num_steps, size_t num_ds) +{ + cache_entry_t *ce = NULL; + size_t i; + int status = 0; + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + return (-ENOENT); + } + + if (((size_t) ce->values_num) != num_ds) + { + pthread_mutex_unlock (&cache_lock); + return (-EINVAL); + } + + /* Check if there are enough values available. If not, increase the buffer + * size. */ + if (ce->history_length < num_steps) + { + gauge_t *tmp; + size_t i; + + tmp = realloc (ce->history, sizeof (*ce->history) + * num_steps * ce->values_num); + if (tmp == NULL) + { + pthread_mutex_unlock (&cache_lock); + return (-ENOMEM); + } + + for (i = ce->history_length * ce->values_num; + i < (num_steps * ce->values_num); + i++) + tmp[i] = NAN; + + ce->history = tmp; + ce->history_length = num_steps; + } /* if (ce->history_length < num_steps) */ + + /* Copy the values to the output buffer. */ + for (i = 0; i < num_steps; i++) + { + size_t src_index; + size_t dst_index; + + if (i < ce->history_index) + src_index = ce->history_index - (i + 1); + else + src_index = ce->history_length + ce->history_index - (i + 1); + src_index = src_index * num_ds; + + dst_index = i * num_ds; + + memcpy (ret_history + dst_index, ce->history + src_index, + sizeof (*ret_history) * num_ds); + } + + pthread_mutex_unlock (&cache_lock); + + return (0); +} /* int uc_get_history_by_name */ + +int uc_get_history (const data_set_t *ds, const value_list_t *vl, + gauge_t *ret_history, size_t num_steps, size_t num_ds) +{ + char name[6 * DATA_MAX_NAME_LEN]; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("utils_cache: uc_get_history: FORMAT_VL failed."); + return (-1); + } + + return (uc_get_history_by_name (name, ret_history, num_steps, num_ds)); +} /* int uc_get_history */ + +int uc_get_hits (const data_set_t *ds, const value_list_t *vl) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = STATE_ERROR; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_get_hits */ + +int uc_set_hits (const data_set_t *ds, const value_list_t *vl, int hits) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + ce->hits = hits; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_set_hits */ + +int uc_inc_hits (const data_set_t *ds, const value_list_t *vl, int step) +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int ret = -1; + + if (FORMAT_VL (name, sizeof (name), vl) != 0) + { + ERROR ("uc_get_state: FORMAT_VL failed."); + return (STATE_ERROR); + } + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, name, (void *) &ce) == 0) + { + assert (ce != NULL); + ret = ce->hits; + ce->hits = ret + step; + } + + pthread_mutex_unlock (&cache_lock); + + return (ret); +} /* int uc_inc_hits */ + +/* + * Meta data interface + */ +/* XXX: This function will acquire `cache_lock' but will not free it! */ +static meta_data_t *uc_get_meta (const value_list_t *vl) /* {{{ */ +{ + char name[6 * DATA_MAX_NAME_LEN]; + cache_entry_t *ce = NULL; + int status; + + status = FORMAT_VL (name, sizeof (name), vl); + if (status != 0) + { + ERROR ("utils_cache: uc_get_meta: FORMAT_VL failed."); + return (NULL); + } + + pthread_mutex_lock (&cache_lock); + + status = c_avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + return (NULL); + } + assert (ce != NULL); + + if (ce->meta == NULL) + ce->meta = meta_data_create (); + + if (ce->meta == NULL) + pthread_mutex_unlock (&cache_lock); + + return (ce->meta); +} /* }}} meta_data_t *uc_get_meta */ + +/* Sorry about this preprocessor magic, but it really makes this file much + * shorter.. */ +#define UC_WRAP(wrap_function) { \ + meta_data_t *meta; \ + int status; \ + meta = uc_get_meta (vl); \ + if (meta == NULL) return (-1); \ + status = wrap_function (meta, key); \ + pthread_mutex_unlock (&cache_lock); \ + return (status); \ +} +int uc_meta_data_exists (const value_list_t *vl, const char *key) + UC_WRAP (meta_data_exists) + +int uc_meta_data_delete (const value_list_t *vl, const char *key) + UC_WRAP (meta_data_delete) +#undef UC_WRAP + +/* We need a new version of this macro because the following functions take + * two argumetns. */ +#define UC_WRAP(wrap_function) { \ + meta_data_t *meta; \ + int status; \ + meta = uc_get_meta (vl); \ + if (meta == NULL) return (-1); \ + status = wrap_function (meta, key, value); \ + pthread_mutex_unlock (&cache_lock); \ + return (status); \ +} +int uc_meta_data_add_string (const value_list_t *vl, + const char *key, + const char *value) + UC_WRAP(meta_data_add_string) +int uc_meta_data_add_signed_int (const value_list_t *vl, + const char *key, + int64_t value) + UC_WRAP(meta_data_add_signed_int) +int uc_meta_data_add_unsigned_int (const value_list_t *vl, + const char *key, + uint64_t value) + UC_WRAP(meta_data_add_unsigned_int) +int uc_meta_data_add_double (const value_list_t *vl, + const char *key, + double value) + UC_WRAP(meta_data_add_double) +int uc_meta_data_add_boolean (const value_list_t *vl, + const char *key, + _Bool value) + UC_WRAP(meta_data_add_boolean) + +int uc_meta_data_get_string (const value_list_t *vl, + const char *key, + char **value) + UC_WRAP(meta_data_get_string) +int uc_meta_data_get_signed_int (const value_list_t *vl, + const char *key, + int64_t *value) + UC_WRAP(meta_data_get_signed_int) +int uc_meta_data_get_unsigned_int (const value_list_t *vl, + const char *key, + uint64_t *value) + UC_WRAP(meta_data_get_unsigned_int) +int uc_meta_data_get_double (const value_list_t *vl, + const char *key, + double *value) + UC_WRAP(meta_data_get_double) +int uc_meta_data_get_boolean (const value_list_t *vl, + const char *key, + _Bool *value) + UC_WRAP(meta_data_get_boolean) +#undef UC_WRAP + /* vim: set sw=2 ts=8 sts=2 tw=78 : */