src/utils_cache.[ch]: Added a global cache for all values that are dispatched.

The cache translates counter values to gauge values using the last counter
value available. This can then be used to check against threshold values, as
we'll need to do for monitoring functionality.

Right now the cache doesn't do much: It sits there and caches the values, but
is not yet ever queried. The already implemented (but so far unused) function
`uc_get_rate' returns an array of gauge_t values.

The longterm goal is to have the network, rrdtool and unixsock plugins use this
cache, too. This will require some `plugin specific' data with appropriate
control structures and, which is likely the hardest part, some clever locking
for all that.

diff --git a/src/Makefile.am b/src/Makefile.am
index d229407..9ddd43a 100644 (file)
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -14,13 +14,14 @@ sbin_PROGRAMS = collectd
 bin_PROGRAMS = collectd-nagios
 
 collectd_SOURCES = collectd.c collectd.h \
-                  utils_avltree.c utils_avltree.h \
-                  utils_mount.c utils_mount.h \
-                  utils_llist.c utils_llist.h \
-                  utils_ignorelist.c utils_ignorelist.h \
                   common.c common.h \
-                  plugin.c plugin.h \
                   configfile.c configfile.h \
+                  plugin.c plugin.h \
+                  utils_avltree.c utils_avltree.h \
+                  utils_cache.c utils_cache.h \
+                  utils_ignorelist.c utils_ignorelist.h \
+                  utils_llist.c utils_llist.h \
+                  utils_mount.c utils_mount.h \
                   types_list.c types_list.h
 collectd_CPPFLAGS = $(LTDLINCL)
 collectd_CPPFLAGS += -DPREFIX='"${prefix}"'

diff --git a/src/plugin.c b/src/plugin.c
index 23ea901..37c91a9 100644 (file)
--- a/src/plugin.c
+++ b/src/plugin.c
@@ -31,6 +31,7 @@
 #include "plugin.h"
 #include "configfile.h"
 #include "utils_llist.h"
+#include "utils_cache.h"
 
 /*
  * Private structures
@@ -549,6 +550,9 @@ void plugin_init_all (void)
                start_threads ((num > 0) ? num : 5);
        }
 
+       /* Init the value cache */
+       uc_init ();
+
        if (list_init == NULL)
                return;
 
@@ -678,6 +682,9 @@ int plugin_dispatch_values (const char *name, value_list_t *vl)
        escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
        escape_slashes (vl->type_instance, sizeof (vl->type_instance));
 
+       /* Update the value cache */
+       uc_update (ds, vl);
+
        le = llist_head (list_write);
        while (le != NULL)
        {

diff --git a/src/utils_cache.c b/src/utils_cache.c
new file mode 100644 (file)
index 0000000..5ecb83c
--- /dev/null
+++ b/src/utils_cache.c
@@ -0,0 +1,201 @@
+/**
+ * collectd - src/utils_cache.c
+ * Copyright (C) 2007  Florian octo Forster
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the
+ * Free Software Foundation; only version 2 of the License is applicable.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
+ *
+ * Author:
+ *   Florian octo Forster <octo at verplant.org>
+ **/
+
+#include "collectd.h"
+#include "common.h"
+#include "plugin.h"
+#include "utils_avltree.h"
+
+#include <assert.h>
+#include <pthread.h>
+
+typedef struct cache_entry_s
+{
+       char name[6 * DATA_MAX_NAME_LEN];
+       int        values_num;
+       gauge_t   *values_gauge;
+       counter_t *values_counter;
+       time_t last_update;
+} cache_entry_t;
+
+static avl_tree_t     *cache_tree = NULL;
+static pthread_mutex_t cache_lock;
+
+static int cache_compare (const cache_entry_t *a, const cache_entry_t *b)
+{
+  assert ((a != NULL) && (b != NULL));
+  return (strcmp (a->name, b->name));
+} /* int cache_compare */
+
+int uc_init (void)
+{
+  if (cache_tree == NULL)
+    cache_tree = avl_create ((int (*) (const void *, const void *))
+       cache_compare);
+
+  return (0);
+} /* int uc_init */
+
+int uc_update (const data_set_t *ds, const value_list_t *vl)
+{
+  char name[6 * DATA_MAX_NAME_LEN];
+  cache_entry_t *ce = NULL;
+
+  if (FORMAT_VL (name, sizeof (name), vl, ds) != 0)
+  {
+    ERROR ("uc_insert: FORMAT_VL failed.");
+    return (-1);
+  }
+
+  pthread_mutex_lock (&cache_lock);
+
+  if (avl_get (cache_tree, name, (void *) &ce) == 0)
+  {
+    int i;
+
+    assert (ce != NULL);
+    assert (ce->values_num == ds->ds_num);
+
+    if (ce->last_update >= 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);
+      return (-1);
+    }
+
+    for (i = 0; i < ds->ds_num; i++)
+    {
+      if (ds->ds[i].type == DS_TYPE_COUNTER)
+      {
+       ce->values_gauge[i] = ((double) (vl->values[i].counter
+           - ce->values_counter[i]))
+         / ((double) (vl->time - ce->last_update));
+       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_insert: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]);
+    } /* for (i) */
+
+    ce->last_update = vl->time;
+  }
+  else /* key is not found */
+  {
+    int i;
+    size_t ce_size = sizeof (cache_entry_t)
+      + ds->ds_num * (sizeof (counter_t) + sizeof (gauge_t));
+    char *key;
+    
+    key = strdup (name);
+    if (key == NULL)
+    {
+      pthread_mutex_unlock (&cache_lock);
+      ERROR ("uc_insert: strdup failed.");
+      return (-1);
+    }
+
+    ce = (cache_entry_t *) malloc (ce_size);
+    if (ce == NULL)
+    {
+      pthread_mutex_unlock (&cache_lock);
+      ERROR ("uc_insert: malloc (%u) failed.", (unsigned int) ce_size);
+      return (-1);
+    }
+
+    memset (ce, '\0', ce_size);
+
+    strncpy (ce->name, name, sizeof (ce->name));
+    ce->name[sizeof (ce->name) - 1] = '\0';
+
+    ce->values_num = ds->ds_num;
+    ce->values_gauge = (gauge_t *) (ce + 1);
+    ce->values_counter = (counter_t *) (ce->values_gauge + ce->values_num);
+
+    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) */
+      {
+       ce->values_gauge[i] = vl->values[i].gauge;
+      }
+    } /* for (i) */
+
+    ce->last_update = vl->time;
+
+    if (avl_insert (cache_tree, key, ce) != 0)
+    {
+      pthread_mutex_unlock (&cache_lock);
+      ERROR ("uc_insert: avl_insert failed.");
+      return (-1);
+    }
+
+    DEBUG ("uc_insert: Added %s to the cache.", name);
+  } /* if (key is not found) */
+
+  pthread_mutex_unlock (&cache_lock);
+
+  return (0);
+} /* int uc_insert */
+
+gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl)
+{
+  char name[6 * DATA_MAX_NAME_LEN];
+  gauge_t *ret = NULL;
+  cache_entry_t *ce = NULL;
+
+  if (FORMAT_VL (name, sizeof (name), vl, ds) != 0)
+  {
+    ERROR ("uc_insert: FORMAT_VL failed.");
+    return (NULL);
+  }
+
+  pthread_mutex_lock (&cache_lock);
+
+  if (avl_get (cache_tree, name, (void *) &ce) == 0)
+  {
+    assert (ce != NULL);
+    assert (ce->values_num == ds->ds_num);
+
+    ret = (gauge_t *) malloc (ce->values_num * sizeof (gauge_t));
+    if (ret == NULL)
+    {
+      ERROR ("uc_get_rate: malloc failed.");
+    }
+    else
+    {
+      memcpy (ret, ce->values_gauge, ce->values_num * sizeof (gauge_t));
+    }
+  }
+
+  pthread_mutex_unlock (&cache_lock);
+
+  return (ret);
+} /* gauge_t *uc_get_rate */
+
+/* vim: set sw=2 ts=8 sts=2 tw=78 : */

diff --git a/src/utils_cache.h b/src/utils_cache.h
new file mode 100644 (file)
index 0000000..9145405
--- /dev/null
+++ b/src/utils_cache.h
@@ -0,0 +1,31 @@
+/**
+ * collectd - src/utils_cache.c
+ * Copyright (C) 2007  Florian octo Forster
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the
+ * Free Software Foundation; only version 2 of the License is applicable.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
+ *
+ * Author:
+ *   Florian octo Forster <octo at verplant.org>
+ **/
+
+#ifndef UTILS_CACHE_H
+#define UTILS_CACHE_H 1
+
+#include "plugin.h"
+
+int uc_init (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);
+
+#endif /* !UTILS_CACHE_H */