src/utils_cache.c

   1 /**
   2  * collectd - src/utils_cache.c
   3  * Copyright (C) 2007  Florian octo Forster
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Author:
  19  *   Florian octo Forster <octo at verplant.org>
  20  **/
  21
  22 #include "collectd.h"
  23 #include "common.h"
  24 #include "plugin.h"
  25 #include "utils_avltree.h"
  26 #include "utils_threshold.h"
  27
  28 #include <assert.h>
  29 #include <pthread.h>
  30
  31 typedef struct cache_entry_s
  32 {
  33         char name[6 * DATA_MAX_NAME_LEN];
  34         int        values_num;
  35         gauge_t   *values_gauge;
  36         counter_t *values_counter;
  37         /* Time contained in the package
  38          * (for calculating rates) */
  39         time_t last_time;
  40         /* Time according to the local clock
  41          * (for purging old entries) */
  42         time_t last_update;
  43         /* Interval in which the data is collected
  44          * (for purding old entries) */
  45         int interval;
  46 } cache_entry_t;
  47
  48 static avl_tree_t     *cache_tree = NULL;
  49 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
  50
  51 static int cache_compare (const cache_entry_t *a, const cache_entry_t *b)
  52 {
  53   assert ((a != NULL) && (b != NULL));
  54   return (strcmp (a->name, b->name));
  55 } /* int cache_compare */
  56
  57 static int uc_send_notification (const char *name)
  58 {
  59   cache_entry_t *ce = NULL;
  60   int status;
  61
  62   char *name_copy;
  63   char *host;
  64   char *plugin;
  65   char *plugin_instance;
  66   char *type;
  67   char *type_instance;
  68
  69   notification_t n;
  70
  71   memset (&n, '\0', sizeof (n));
  72
  73   name_copy = strdup (name);
  74   if (name_copy == NULL)
  75   {
  76     ERROR ("uc_send_notification: strdup failed.");
  77     return (-1);
  78   }
  79
  80   status = parse_identifier (name_copy, &host,
  81       &plugin, &plugin_instance,
  82       &type, &type_instance);
  83   if (status != 0)
  84   {
  85     ERROR ("uc_send_notification: Cannot parse name `%s'", name);
  86     return (-1);
  87   }
  88
  89   n.severity = NOTIF_FAILURE;
  90   strncpy (n.host, host, sizeof (n.host));
  91   n.host[sizeof (n.host) - 1] = '\0';
  92
  93   sfree (name_copy);
  94   name_copy = host = plugin = plugin_instance = type = type_instance = NULL;
  95
  96   pthread_mutex_lock (&cache_lock);
  97
  98   n.time = time (NULL);
  99
 100   status = avl_get (cache_tree, name, (void *) &ce);
 101   if (status != 0)
 102   {
 103     pthread_mutex_unlock (&cache_lock);
 104     sfree (name_copy);
 105     return (-1);
 106   }
 107
 108   /* Check if the entry has been updated in the meantime */
 109   if ((n.time - ce->last_update) < (2 * ce->interval))
 110   {
 111     pthread_mutex_unlock (&cache_lock);
 112     sfree (name_copy);
 113     return (-1);
 114   }
 115
 116   snprintf (n.message, sizeof (n.message),
 117       "%s has not been updated for %i seconds.", name,
 118       (int) (n.time - ce->last_update));
 119
 120   pthread_mutex_unlock (&cache_lock);
 121
 122   n.message[sizeof (n.message) - 1] = '\0';
 123   plugin_dispatch_notification (&n);
 124
 125   return (0);
 126 } /* int uc_send_notification */
 127
 128 int uc_init (void)
 129 {
 130   if (cache_tree == NULL)
 131     cache_tree = avl_create ((int (*) (const void *, const void *))
 132         cache_compare);
 133
 134   return (0);
 135 } /* int uc_init */
 136
 137 int uc_check_timeout (void)
 138 {
 139   time_t now;
 140   cache_entry_t *ce;
 141
 142   char **keys = NULL;
 143   int keys_len = 0;
 144
 145   char *key;
 146   avl_iterator_t *iter;
 147   int i;
 148
 149   pthread_mutex_lock (&cache_lock);
 150
 151   now = time (NULL);
 152
 153   /* Build a list of entries to be flushed */
 154   iter = avl_get_iterator (cache_tree);
 155   while (avl_iterator_next (iter, (void *) &key, (void *) &ce) == 0)
 156   {
 157     /* If entry has not been updated, add to `keys' array */
 158     if ((now - ce->last_update) >= (2 * ce->interval))
 159     {
 160       char **tmp;
 161
 162       tmp = (char **) realloc ((void *) keys,
 163           (keys_len + 1) * sizeof (char *));
 164       if (tmp == NULL)
 165       {
 166         ERROR ("uc_purge: realloc failed.");
 167         avl_iterator_destroy (iter);
 168         return (-1);
 169       }
 170
 171       keys = tmp;
 172       keys[keys_len] = strdup (key);
 173       if (keys[keys_len] == NULL)
 174       {
 175         ERROR ("uc_check_timeout: strdup failed.");
 176         continue;
 177       }
 178       keys_len++;
 179     }
 180   } /* while (avl_iterator_next) */
 181
 182   for (i = 0; i < keys_len; i++)
 183   {
 184     int status;
 185
 186     status = ut_check_interesting (keys[i]);
 187
 188     if (status < 0)
 189     {
 190       ERROR ("uc_check_timeout: ut_check_interesting failed.");
 191       sfree (keys[i]);
 192     }
 193     else if (status == 0) /* ``service'' is uninteresting */
 194     {
 195       ce = NULL;
 196       DEBUG ("uc_check_timeout: %s is missing but ``uninteresting''", keys[i]);
 197       status = avl_remove (cache_tree, keys[i], (void *) &key, (void *) &ce);
 198       if (status != 0)
 199       {
 200         ERROR ("uc_check_timeout: avl_remove (%s) failed.", keys[i]);
 201       }
 202       sfree (keys[i]);
 203       sfree (ce);
 204     }
 205     else /* (status > 0); ``service'' is interesting */
 206     {
 207       /*
 208        * `keys[i]' is not freed and set to NULL, so that the for-loop below
 209        * will send out notifications. There's nothing else to do here.
 210        */
 211       DEBUG ("uc_check_timeout: %s is missing and ``interesting''", keys[i]);
 212     }
 213   } /* for (keys[i]) */
 214
 215   avl_iterator_destroy (iter);
 216
 217   pthread_mutex_unlock (&cache_lock);
 218
 219   for (i = 0; i < keys_len; i++)
 220   {
 221     if (keys[i] == NULL)
 222       continue;
 223
 224     uc_send_notification (keys[i]);
 225     sfree (keys[i]);
 226   }
 227
 228   sfree (keys);
 229
 230   return (0);
 231 } /* int uc_check_timeout */
 232
 233 int uc_update (const data_set_t *ds, const value_list_t *vl)
 234 {
 235   char name[6 * DATA_MAX_NAME_LEN];
 236   cache_entry_t *ce = NULL;
 237
 238   if (FORMAT_VL (name, sizeof (name), vl, ds) != 0)
 239   {
 240     ERROR ("uc_insert: FORMAT_VL failed.");
 241     return (-1);
 242   }
 243
 244   pthread_mutex_lock (&cache_lock);
 245
 246   if (avl_get (cache_tree, name, (void *) &ce) == 0)
 247   {
 248     int i;
 249
 250     assert (ce != NULL);
 251     assert (ce->values_num == ds->ds_num);
 252
 253     if (ce->last_time >= vl->time)
 254     {
 255       pthread_mutex_unlock (&cache_lock);
 256       NOTICE ("uc_insert: Value too old: name = %s; value time = %u; "
 257           "last cache update = %u;",
 258           name, (unsigned int) vl->time, (unsigned int) ce->last_time);
 259       return (-1);
 260     }
 261
 262     if ((ce->last_time + ce->interval) < vl->time)
 263     {
 264       /* TODO: Implement a `real' okay notification. Watch out for locking
 265        * issues, though! */
 266       NOTICE ("uc_insert: Okay notification for %s", name);
 267     }
 268
 269     for (i = 0; i < ds->ds_num; i++)
 270     {
 271       if (ds->ds[i].type == DS_TYPE_COUNTER)
 272       {
 273         counter_t diff;
 274
 275         /* check if the counter has wrapped around */
 276         if (vl->values[i].counter < ce->values_counter[i])
 277         {
 278           if (ce->values_counter[i] <= 4294967295U)
 279             diff = (4294967295U - ce->values_counter[i])
 280               + vl->values[i].counter;
 281           else
 282             diff = (18446744073709551615ULL - ce->values_counter[i])
 283               + vl->values[i].counter;
 284         }
 285         else /* counter has NOT wrapped around */
 286         {
 287           diff = vl->values[i].counter - ce->values_counter[i];
 288         }
 289
 290         ce->values_gauge[i] = ((double) diff)
 291           / ((double) (vl->time - ce->last_time));
 292         ce->values_counter[i] = vl->values[i].counter;
 293       }
 294       else /* if (ds->ds[i].type == DS_TYPE_GAUGE) */
 295       {
 296         ce->values_gauge[i] = vl->values[i].gauge;
 297       }
 298       DEBUG ("uc_insert: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]);
 299     } /* for (i) */
 300
 301     ce->last_time = vl->time;
 302     ce->last_update = time (NULL);
 303     ce->interval = vl->interval;
 304   }
 305   else /* key is not found */
 306   {
 307     int i;
 308     size_t ce_size = sizeof (cache_entry_t)
 309       + ds->ds_num * (sizeof (counter_t) + sizeof (gauge_t));
 310     char *key;
 311
 312     key = strdup (name);
 313     if (key == NULL)
 314     {
 315       pthread_mutex_unlock (&cache_lock);
 316       ERROR ("uc_insert: strdup failed.");
 317       return (-1);
 318     }
 319
 320     ce = (cache_entry_t *) malloc (ce_size);
 321     if (ce == NULL)
 322     {
 323       pthread_mutex_unlock (&cache_lock);
 324       ERROR ("uc_insert: malloc (%u) failed.", (unsigned int) ce_size);
 325       return (-1);
 326     }
 327
 328     memset (ce, '\0', ce_size);
 329
 330     strncpy (ce->name, name, sizeof (ce->name));
 331     ce->name[sizeof (ce->name) - 1] = '\0';
 332
 333     ce->values_num = ds->ds_num;
 334     ce->values_gauge = (gauge_t *) (ce + 1);
 335     ce->values_counter = (counter_t *) (ce->values_gauge + ce->values_num);
 336
 337     for (i = 0; i < ds->ds_num; i++)
 338     {
 339       if (ds->ds[i].type == DS_TYPE_COUNTER)
 340       {
 341         ce->values_gauge[i] = NAN;
 342         ce->values_counter[i] = vl->values[i].counter;
 343       }
 344       else /* if (ds->ds[i].type == DS_TYPE_GAUGE) */
 345       {
 346         ce->values_gauge[i] = vl->values[i].gauge;
 347       }
 348     } /* for (i) */
 349
 350     ce->last_time = vl->time;
 351     ce->last_update = time (NULL);
 352     ce->interval = vl->interval;
 353
 354     if (avl_insert (cache_tree, key, ce) != 0)
 355     {
 356       pthread_mutex_unlock (&cache_lock);
 357       ERROR ("uc_insert: avl_insert failed.");
 358       return (-1);
 359     }
 360
 361     DEBUG ("uc_insert: Added %s to the cache.", name);
 362   } /* if (key is not found) */
 363
 364   pthread_mutex_unlock (&cache_lock);
 365
 366   return (0);
 367 } /* int uc_insert */
 368
 369 gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl)
 370 {
 371   char name[6 * DATA_MAX_NAME_LEN];
 372   gauge_t *ret = NULL;
 373   cache_entry_t *ce = NULL;
 374
 375   if (FORMAT_VL (name, sizeof (name), vl, ds) != 0)
 376   {
 377     ERROR ("uc_insert: FORMAT_VL failed.");
 378     return (NULL);
 379   }
 380
 381   pthread_mutex_lock (&cache_lock);
 382
 383   if (avl_get (cache_tree, name, (void *) &ce) == 0)
 384   {
 385     assert (ce != NULL);
 386     assert (ce->values_num == ds->ds_num);
 387
 388     ret = (gauge_t *) malloc (ce->values_num * sizeof (gauge_t));
 389     if (ret == NULL)
 390     {
 391       ERROR ("uc_get_rate: malloc failed.");
 392     }
 393     else
 394     {
 395       memcpy (ret, ce->values_gauge, ce->values_num * sizeof (gauge_t));
 396     }
 397   }
 398
 399   pthread_mutex_unlock (&cache_lock);
 400
 401   return (ret);
 402 } /* gauge_t *uc_get_rate */
 403
 404 /* vim: set sw=2 ts=8 sts=2 tw=78 : */