From: Florian Forster Date: Sat, 15 Dec 2007 11:11:06 +0000 (+0100) Subject: src/utils_cache.[ch]: Added the `uc_check_timeout' function. X-Git-Tag: collectd-4.3.0beta0~63 X-Git-Url: https://git.verplant.org/?a=commitdiff_plain;h=fbcda5ac431bbb293157bf00e204310dcb65afd3;p=collectd.git src/utils_cache.[ch]: Added the `uc_check_timeout' function. This function is called before the read plugins. It checks if values are missing, i. e. have not been reported for longer than twice their ``interval''. In this case a notification is created, though this is probably not the final behavior. This code is highly experimental. --- diff --git a/src/plugin.c b/src/plugin.c index ec9ede25..b24707a0 100644 --- a/src/plugin.c +++ b/src/plugin.c @@ -605,6 +605,8 @@ void plugin_read_all (const int *loop) llentry_t *le; read_func_t *rf; + uc_check_timeout (); + if (list_read == NULL) return; diff --git a/src/utils_cache.c b/src/utils_cache.c index 9349a1f4..20d4deb6 100644 --- a/src/utils_cache.c +++ b/src/utils_cache.c @@ -33,7 +33,15 @@ typedef struct cache_entry_s int values_num; gauge_t *values_gauge; counter_t *values_counter; + /* Time contained in the package + * (for calculating rates) */ + time_t last_time; + /* Time according to the local clock + * (for purging old entries) */ time_t last_update; + /* Interval in which the data is collected + * (for purding old entries) */ + int interval; } cache_entry_t; static avl_tree_t *cache_tree = NULL; @@ -45,6 +53,77 @@ static int cache_compare (const cache_entry_t *a, const cache_entry_t *b) return (strcmp (a->name, b->name)); } /* int cache_compare */ +static int uc_send_notification (const char *name) +{ + cache_entry_t *ce = NULL; + int status; + + char *name_copy; + char *host; + char *plugin; + char *plugin_instance; + char *type; + char *type_instance; + + notification_t n; + + memset (&n, '\0', sizeof (n)); + + name_copy = strdup (name); + if (name_copy == NULL) + { + ERROR ("uc_send_notification: strdup failed."); + return (-1); + } + + status = parse_identifier (name_copy, &host, + &plugin, &plugin_instance, + &type, &type_instance); + if (status != 0) + { + ERROR ("uc_send_notification: Cannot parse name `%s'", name); + return (-1); + } + + n.severity = NOTIF_FAILURE; + strncpy (n.host, host, sizeof (n.host)); + n.host[sizeof (n.host) - 1] = '\0'; + + sfree (name_copy); + name_copy = host = plugin = plugin_instance = type = type_instance = NULL; + + pthread_mutex_lock (&cache_lock); + + n.time = time (NULL); + + status = avl_get (cache_tree, name, (void *) &ce); + if (status != 0) + { + pthread_mutex_unlock (&cache_lock); + sfree (name_copy); + return (-1); + } + + /* Check if the entry has been updated in the meantime */ + if ((n.time - ce->last_update) <= (2 * ce->interval)) + { + pthread_mutex_unlock (&cache_lock); + sfree (name_copy); + return (-1); + } + + snprintf (n.message, sizeof (n.message), + "%s has not been updated for %i seconds.", name, + (int) (ce->last_update - n.time)); + + pthread_mutex_unlock (&cache_lock); + + n.message[sizeof (n.message) - 1] = '\0'; + plugin_dispatch_notification (&n); + + return (0); +} /* int uc_send_notification */ + int uc_init (void) { if (cache_tree == NULL) @@ -54,6 +133,86 @@ int uc_init (void) return (0); } /* int uc_init */ +int uc_check_timeout (void) +{ + time_t now; + cache_entry_t *ce; + + char **keys = NULL; + int keys_len = 0; + + char *key; + avl_iterator_t *iter; + int i; + + pthread_mutex_lock (&cache_lock); + + now = time (NULL); + + /* Build a list of entries to be flushed */ + iter = avl_get_iterator (cache_tree); + while (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)) + { + char **tmp; + + tmp = (char **) realloc ((void *) keys, + (keys_len + 1) * sizeof (char *)); + if (tmp == NULL) + { + ERROR ("uc_purge: realloc failed."); + avl_iterator_destroy (iter); + return (-1); + } + + keys = tmp; + keys[keys_len] = strdup (key); + if (keys[keys_len] == NULL) + { + ERROR ("uc_check_timeout: strdup failed."); + continue; + } + keys_len++; + } + } /* while (avl_iterator_next) */ + + for (i = 0; i < keys_len; i++) + { + int status; + + /* TODO: Check if value interesting: + * - Not interesting: Remove value from cache and shut up + * - Interesting: Don't remove value from cache but send a + * notification. + */ + status = avl_remove (cache_tree, keys[i], (void *) &key, (void *) &ce); + if (status != 0) + { + ERROR ("uc_check_timeout: avl_remove (%s) failed.", keys[i]); + continue; + } + + sfree (key); + sfree (ce); + } + + avl_iterator_destroy (iter); + + pthread_mutex_unlock (&cache_lock); + + for (i = 0; i < keys_len; i++) + { + uc_send_notification (keys[i]); + sfree (keys[i]); + } + + sfree (keys); + + return (0); +} /* int uc_check_timeout */ + int uc_update (const data_set_t *ds, const value_list_t *vl) { char name[6 * DATA_MAX_NAME_LEN]; @@ -74,15 +233,22 @@ int uc_update (const data_set_t *ds, const value_list_t *vl) assert (ce != NULL); assert (ce->values_num == ds->ds_num); - if (ce->last_update >= vl->time) + if (ce->last_time >= vl->time) { pthread_mutex_unlock (&cache_lock); NOTICE ("uc_insert: Value too old: name = %s; value time = %u; " "last cache update = %u;", - name, (unsigned int) vl->time, (unsigned int) ce->last_update); + name, (unsigned int) vl->time, (unsigned int) ce->last_time); return (-1); } + if ((ce->last_time + ce->interval) < vl->time) + { + /* TODO: Implement a `real' okay notification. Watch out for locking + * issues, though! */ + NOTICE ("uc_insert: Okay notification for %s", name); + } + for (i = 0; i < ds->ds_num; i++) { if (ds->ds[i].type == DS_TYPE_COUNTER) @@ -105,7 +271,7 @@ int uc_update (const data_set_t *ds, const value_list_t *vl) } ce->values_gauge[i] = ((double) diff) - / ((double) (vl->time - ce->last_update)); + / ((double) (vl->time - ce->last_time)); ce->values_counter[i] = vl->values[i].counter; } else /* if (ds->ds[i].type == DS_TYPE_GAUGE) */ @@ -115,7 +281,9 @@ int uc_update (const data_set_t *ds, const value_list_t *vl) DEBUG ("uc_insert: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]); } /* for (i) */ - ce->last_update = vl->time; + ce->last_time = vl->time; + ce->last_update = time (NULL); + ce->interval = vl->interval; } else /* key is not found */ { @@ -162,7 +330,7 @@ int uc_update (const data_set_t *ds, const value_list_t *vl) } } /* for (i) */ - ce->last_update = vl->time; + ce->last_time = vl->time; if (avl_insert (cache_tree, key, ce) != 0) { diff --git a/src/utils_cache.h b/src/utils_cache.h index 8d1c7549..2983d9ac 100644 --- a/src/utils_cache.h +++ b/src/utils_cache.h @@ -25,6 +25,7 @@ #include "plugin.h" int uc_init (void); +int uc_check_timeout (void); int uc_update (const data_set_t *ds, const value_list_t *vl); gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl);