src/utils_cache.c

   1 /**
   2  * collectd - src/utils_cache.c
   3  * Copyright (C) 2007,2008  Florian octo Forster
   4  * Copyright (C) 2010       Andrés J. Díaz
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License as published by the
   8  * Free Software Foundation; only version 2 of the License is applicable.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18  *
  19  * Author:
  20  *   Florian octo Forster <octo at verplant.org>
  21  *   Andrés J. Díaz <ajdiaz at connectical.com>
  22  **/
  23
  24 #include "collectd.h"
  25 #include "common.h"
  26 #include "plugin.h"
  27 #include "utils_avltree.h"
  28 #include "utils_cache.h"
  29 #include "utils_threshold.h"
  30 #include "utils_subst.h"
  31 #include "meta_data.h"
  32
  33 #include <assert.h>
  34 #include <pthread.h>
  35
  36 typedef struct cache_entry_s
  37 {
  38         char name[6 * DATA_MAX_NAME_LEN];
  39         int        values_num;
  40         gauge_t   *values_gauge;
  41         value_t   *values_raw;
  42         /* Time contained in the package
  43          * (for calculating rates) */
  44         time_t last_time;
  45         /* Time according to the local clock
  46          * (for purging old entries) */
  47         time_t last_update;
  48         /* Interval in which the data is collected
  49          * (for purding old entries) */
  50         int interval;
  51         int state;
  52         int hits;
  53
  54         /*
  55          * +-----+-----+-----+-----+-----+-----+-----+-----+-----+----
  56          * !  0  !  1  !  2  !  3  !  4  !  5  !  6  !  7  !  8  ! ...
  57          * +-----+-----+-----+-----+-----+-----+-----+-----+-----+----
  58          * ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ...
  59          * +-----+-----+-----+-----+-----+-----+-----+-----+-----+----
  60          * !      t = 0      !      t = 1      !      t = 2      ! ...
  61          * +-----------------+-----------------+-----------------+----
  62          */
  63         gauge_t *history;
  64         size_t   history_index; /* points to the next position to write to. */
  65         size_t   history_length;
  66
  67         meta_data_t *meta;
  68 } cache_entry_t;
  69
  70 static c_avl_tree_t   *cache_tree = NULL;
  71 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
  72
  73 static int cache_compare (const cache_entry_t *a, const cache_entry_t *b)
  74 {
  75   assert ((a != NULL) && (b != NULL));
  76   return (strcmp (a->name, b->name));
  77 } /* int cache_compare */
  78
  79 static cache_entry_t *cache_alloc (int values_num)
  80 {
  81   cache_entry_t *ce;
  82
  83   ce = (cache_entry_t *) malloc (sizeof (cache_entry_t));
  84   if (ce == NULL)
  85   {
  86     ERROR ("utils_cache: cache_alloc: malloc failed.");
  87     return (NULL);
  88   }
  89   memset (ce, '\0', sizeof (cache_entry_t));
  90   ce->values_num = values_num;
  91
  92   ce->values_gauge = calloc (values_num, sizeof (*ce->values_gauge));
  93   ce->values_raw   = calloc (values_num, sizeof (*ce->values_raw));
  94   if ((ce->values_gauge == NULL) || (ce->values_raw == NULL))
  95   {
  96     sfree (ce->values_gauge);
  97     sfree (ce->values_raw);
  98     sfree (ce);
  99     ERROR ("utils_cache: cache_alloc: calloc failed.");
 100     return (NULL);
 101   }
 102
 103   ce->history = NULL;
 104   ce->history_length = 0;
 105   ce->meta = NULL;
 106
 107   return (ce);
 108 } /* cache_entry_t *cache_alloc */
 109
 110 static void cache_free (cache_entry_t *ce)
 111 {
 112   if (ce == NULL)
 113     return;
 114
 115   sfree (ce->values_gauge);
 116   sfree (ce->values_raw);
 117   sfree (ce->history);
 118   if (ce->meta != NULL)
 119   {
 120     meta_data_destroy (ce->meta);
 121     ce->meta = NULL;
 122   }
 123   sfree (ce);
 124 } /* void cache_free */
 125
 126 static int uc_send_notification (const char *name)
 127 {
 128   cache_entry_t *ce = NULL;
 129   int status;
 130
 131   char *name_copy;
 132   char *host;
 133   char *plugin;
 134   char *plugin_instance;
 135   char *type;
 136   char *type_instance;
 137   threshold_t th;
 138
 139   notification_t n;
 140
 141   data_set_t ds;
 142   value_list_t vl;
 143
 144   name_copy = strdup (name);
 145   if (name_copy == NULL)
 146   {
 147     ERROR ("uc_send_notification: strdup failed.");
 148     return (-1);
 149   }
 150
 151   status = parse_identifier (name_copy, &host,
 152       &plugin, &plugin_instance,
 153       &type, &type_instance);
 154   if (status != 0)
 155   {
 156     ERROR ("uc_send_notification: Cannot parse name `%s'", name);
 157     return (-1);
 158   }
 159
 160   memset (&ds, '\0', sizeof (ds));
 161   memset (&vl, '\0', sizeof (vl));
 162   sstrncpy (vl.host, host, sizeof (vl.host));
 163   sstrncpy (vl.plugin, plugin, sizeof (vl.plugin));
 164   if (plugin_instance != NULL)
 165     sstrncpy (vl.plugin_instance, plugin_instance, sizeof (vl.plugin_instance));
 166   sstrncpy (ds.type, type, sizeof (ds.type));
 167   sstrncpy (vl.type, type, sizeof (vl.type));
 168   if (type_instance != NULL)
 169     sstrncpy (vl.type_instance, type_instance, sizeof (vl.type_instance));
 170
 171
 172   /* Copy the associative members */
 173   notification_init (&n, NOTIF_FAILURE, /* host = */ NULL,
 174       host, plugin, plugin_instance, type, type_instance);
 175
 176   sfree (name_copy);
 177   name_copy = host = plugin = plugin_instance = type = type_instance = NULL;
 178
 179
 180   pthread_mutex_lock (&cache_lock);
 181
 182   /*
 183    * Set the time _after_ getting the lock because we don't know how long
 184    * acquiring the lock takes and we will use this time later to decide
 185    * whether or not the state is OKAY.
 186    */
 187   n.time = time (NULL);
 188
 189   status = c_avl_get (cache_tree, name, (void *) &ce);
 190   if (status != 0)
 191   {
 192     pthread_mutex_unlock (&cache_lock);
 193     sfree (name_copy);
 194     return (-1);
 195   }
 196
 197   /* Check if the entry has been updated in the meantime */
 198   if ((n.time - ce->last_update) < (2 * ce->interval))
 199   {
 200     ce->state = STATE_OKAY;
 201     pthread_mutex_unlock (&cache_lock);
 202     sfree (name_copy);
 203     return (-1);
 204   }
 205
 206   /* if the associated threshold has a missing message, then use custom
 207    * message. FIXME: we do a threshold_search here and in uc_check_timeout
 208    * (calling ut_check_interesting, but we really need to do this once */
 209   if ( !ut_search_threshold(&vl, &th) &&
 210       (th.missing_message != NULL) )
 211   {
 212     char msg[NOTIF_MAX_MSG_LEN];
 213     char temp[NOTIF_MAX_MSG_LEN];
 214
 215     sstrncpy (msg, th.missing_message, sizeof (msg));
 216     (void) ut_build_message (msg, NOTIF_MAX_MSG_LEN, th.missing_message,
 217         &ds, 0, &vl, ce->values_gauge,
 218         &n, &th);
 219
 220 #define REPLACE_FIELD(t,v) \
 221     if (subst_string (temp, sizeof (temp), msg, t, v) != NULL) \
 222     sstrncpy (msg, temp, sizeof (msg));
 223
 224     char itoa_temp[NOTIF_MAX_MSG_LEN];
 225 #define ITOA(string,i) \
 226     memset(string,0x00,sizeof(string)); \
 227     snprintf(string, sizeof(string), "%i", i);
 228
 229     ITOA(itoa_temp, (int)(n.time - ce->last_update))
 230       REPLACE_FIELD("%{missing}", itoa_temp)
 231
 232     (void) ssnprintf (n.message, sizeof (n.message),
 233         "%s", msg);
 234   }
 235   else
 236   {
 237     ssnprintf (n.message, sizeof (n.message),
 238         "%s has not been updated for %i seconds.", name,
 239         (int) (n.time - ce->last_update));
 240   }
 241
 242   pthread_mutex_unlock (&cache_lock);
 243
 244   plugin_dispatch_notification (&n);
 245
 246   return (0);
 247 } /* int uc_send_notification */
 248
 249 static void uc_check_range (const data_set_t *ds, cache_entry_t *ce)
 250 {
 251   int i;
 252
 253   for (i = 0; i < ds->ds_num; i++)
 254   {
 255     if (isnan (ce->values_gauge[i]))
 256       continue;
 257     else if (ce->values_gauge[i] < ds->ds[i].min)
 258       ce->values_gauge[i] = NAN;
 259     else if (ce->values_gauge[i] > ds->ds[i].max)
 260       ce->values_gauge[i] = NAN;
 261   }
 262 } /* void uc_check_range */
 263
 264 static int uc_insert (const data_set_t *ds, const value_list_t *vl,
 265     const char *key)
 266 {
 267   int i;
 268   char *key_copy;
 269   cache_entry_t *ce;
 270
 271   /* `cache_lock' has been locked by `uc_update' */
 272
 273   key_copy = strdup (key);
 274   if (key_copy == NULL)
 275   {
 276     ERROR ("uc_insert: strdup failed.");
 277     return (-1);
 278   }
 279
 280   ce = cache_alloc (ds->ds_num);
 281   if (ce == NULL)
 282   {
 283     sfree (key_copy);
 284     ERROR ("uc_insert: cache_alloc (%i) failed.", ds->ds_num);
 285     return (-1);
 286   }
 287
 288   sstrncpy (ce->name, key, sizeof (ce->name));
 289
 290   for (i = 0; i < ds->ds_num; i++)
 291   {
 292     switch (ds->ds[i].type)
 293     {
 294       case DS_TYPE_COUNTER:
 295         ce->values_gauge[i] = NAN;
 296         ce->values_raw[i].counter = vl->values[i].counter;
 297         break;
 298
 299       case DS_TYPE_GAUGE:
 300         ce->values_gauge[i] = vl->values[i].gauge;
 301         ce->values_raw[i].gauge = vl->values[i].gauge;
 302         break;
 303
 304       case DS_TYPE_DERIVE:
 305         ce->values_gauge[i] = NAN;
 306         ce->values_raw[i].derive = vl->values[i].derive;
 307         break;
 308
 309       case DS_TYPE_ABSOLUTE:
 310         ce->values_gauge[i] = NAN;
 311         if (vl->interval > 0)
 312           ce->values_gauge[i] = ((double) vl->values[i].absolute)
 313             / ((double) vl->interval);
 314         ce->values_raw[i].absolute = vl->values[i].absolute;
 315         break;
 316
 317       default:
 318         /* This shouldn't happen. */
 319         ERROR ("uc_insert: Don't know how to handle data source type %i.",
 320             ds->ds[i].type);
 321         return (-1);
 322     } /* switch (ds->ds[i].type) */
 323   } /* for (i) */
 324
 325   /* Prune invalid gauge data */
 326   uc_check_range (ds, ce);
 327
 328   ce->last_time = vl->time;
 329   ce->last_update = time (NULL);
 330   ce->interval = vl->interval;
 331   ce->state = STATE_OKAY;
 332
 333   if (c_avl_insert (cache_tree, key_copy, ce) != 0)
 334   {
 335     sfree (key_copy);
 336     ERROR ("uc_insert: c_avl_insert failed.");
 337     return (-1);
 338   }
 339
 340   DEBUG ("uc_insert: Added %s to the cache.", key);
 341   return (0);
 342 } /* int uc_insert */
 343
 344 int uc_init (void)
 345 {
 346   if (cache_tree == NULL)
 347     cache_tree = c_avl_create ((int (*) (const void *, const void *))
 348         cache_compare);
 349
 350   return (0);
 351 } /* int uc_init */
 352
 353 int uc_check_timeout (void)
 354 {
 355   time_t now;
 356   cache_entry_t *ce;
 357
 358   char **keys = NULL;
 359   int keys_len = 0;
 360
 361   char *key;
 362   c_avl_iterator_t *iter;
 363   int i;
 364
 365   pthread_mutex_lock (&cache_lock);
 366
 367   now = time (NULL);
 368
 369   /* Build a list of entries to be flushed */
 370   iter = c_avl_get_iterator (cache_tree);
 371   while (c_avl_iterator_next (iter, (void *) &key, (void *) &ce) == 0)
 372   {
 373     /* If entry has not been updated, add to `keys' array */
 374     if ((now - ce->last_update) >= (timeout_g * ce->interval))
 375     {
 376       char **tmp;
 377
 378       tmp = (char **) realloc ((void *) keys,
 379           (keys_len + 1) * sizeof (char *));
 380       if (tmp == NULL)
 381       {
 382         ERROR ("uc_check_timeout: realloc failed.");
 383         c_avl_iterator_destroy (iter);
 384         sfree (keys);
 385         pthread_mutex_unlock (&cache_lock);
 386         return (-1);
 387       }
 388
 389       keys = tmp;
 390       keys[keys_len] = strdup (key);
 391       if (keys[keys_len] == NULL)
 392       {
 393         ERROR ("uc_check_timeout: strdup failed.");
 394         continue;
 395       }
 396       keys_len++;
 397     }
 398   } /* while (c_avl_iterator_next) */
 399
 400   ce = NULL;
 401
 402   for (i = 0; i < keys_len; i++)
 403   {
 404     int status;
 405
 406     status = ut_check_interesting (keys[i]);
 407
 408     if (status < 0)
 409     {
 410       ERROR ("uc_check_timeout: ut_check_interesting failed.");
 411       sfree (keys[i]);
 412       continue;
 413     }
 414     else if (status == 0) /* ``service'' is uninteresting */
 415     {
 416       DEBUG ("uc_check_timeout: %s is missing but ``uninteresting''",
 417           keys[i]);
 418       ce = NULL;
 419       status = c_avl_remove (cache_tree, keys[i],
 420           (void *) &key, (void *) &ce);
 421       if (status != 0)
 422       {
 423         ERROR ("uc_check_timeout: c_avl_remove (%s) failed.", keys[i]);
 424       }
 425       sfree (keys[i]);
 426       sfree (key);
 427       if (ce != NULL)
 428         cache_free (ce);
 429       continue;
 430     }
 431
 432     /* If we get here, the value is ``interesting''. Query the record from the
 433      * cache and update the state field. */
 434     if (c_avl_get (cache_tree, keys[i], (void *) &ce) != 0)
 435     {
 436       ERROR ("uc_check_timeout: cannot get data for %s from cache", keys[i]);
 437       /* Do not free `keys[i]' so a notification is sent further down. */
 438       continue;
 439     }
 440     assert (ce != NULL);
 441
 442     if (status == 2) /* persist */
 443     {
 444       DEBUG ("uc_check_timeout: %s is missing, sending notification.",
 445           keys[i]);
 446       ce->state = STATE_MISSING;
 447       /* Do not free `keys[i]' so a notification is sent further down. */
 448     }
 449     else if (status == 1) /* do not persist */
 450     {
 451       if (ce->state == STATE_MISSING)
 452       {
 453         DEBUG ("uc_check_timeout: %s is missing but "
 454             "notification has already been sent.",
 455             keys[i]);
 456         /* Set `keys[i]' to NULL to no notification is sent. */
 457         sfree (keys[i]);
 458       }
 459       else /* (ce->state != STATE_MISSING) */
 460       {
 461         DEBUG ("uc_check_timeout: %s is missing, sending one notification.",
 462             keys[i]);
 463         ce->state = STATE_MISSING;
 464         /* Do not free `keys[i]' so a notification is sent further down. */
 465       }
 466     }
 467     else
 468     {
 469       WARNING ("uc_check_timeout: ut_check_interesting (%s) returned "
 470           "invalid status %i.",
 471           keys[i], status);
 472       sfree (keys[i]);
 473     }
 474
 475     /* Make really sure the next iteration doesn't work with this pointer.
 476      * There have been too many bugs in the past.. :/  -- octo */
 477     ce = NULL;
 478   } /* for (keys[i]) */
 479
 480   c_avl_iterator_destroy (iter);
 481
 482   pthread_mutex_unlock (&cache_lock);
 483
 484   for (i = 0; i < keys_len; i++)
 485   {
 486     if (keys[i] == NULL)
 487       continue;
 488
 489     uc_send_notification (keys[i]);
 490     sfree (keys[i]);
 491   }
 492
 493   sfree (keys);
 494
 495   return (0);
 496 } /* int uc_check_timeout */
 497
 498 int uc_update (const data_set_t *ds, const value_list_t *vl)
 499 {
 500   char name[6 * DATA_MAX_NAME_LEN];
 501   cache_entry_t *ce = NULL;
 502   int send_okay_notification = 0;
 503   time_t update_delay = 0;
 504   notification_t n;
 505   int status;
 506   int i;
 507
 508   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 509   {
 510     ERROR ("uc_update: FORMAT_VL failed.");
 511     return (-1);
 512   }
 513
 514   pthread_mutex_lock (&cache_lock);
 515
 516   status = c_avl_get (cache_tree, name, (void *) &ce);
 517   if (status != 0) /* entry does not yet exist */
 518   {
 519     status = uc_insert (ds, vl, name);
 520     pthread_mutex_unlock (&cache_lock);
 521     return (status);
 522   }
 523
 524   assert (ce != NULL);
 525   assert (ce->values_num == ds->ds_num);
 526
 527   if (ce->last_time >= vl->time)
 528   {
 529     pthread_mutex_unlock (&cache_lock);
 530     NOTICE ("uc_update: Value too old: name = %s; value time = %u; "
 531         "last cache update = %u;",
 532         name, (unsigned int) vl->time, (unsigned int) ce->last_time);
 533     return (-1);
 534   }
 535
 536   /* Send a notification (after the lock has been released) if we switch the
 537    * state from something else to `okay'. */
 538   if (ce->state == STATE_MISSING)
 539   {
 540     send_okay_notification = 1;
 541     ce->state = STATE_OKAY;
 542     update_delay = time (NULL) - ce->last_update;
 543   }
 544
 545   for (i = 0; i < ds->ds_num; i++)
 546   {
 547     switch (ds->ds[i].type)
 548     {
 549       case DS_TYPE_COUNTER:
 550         {
 551           counter_t diff;
 552
 553           /* check if the counter has wrapped around */
 554           if (vl->values[i].counter < ce->values_raw[i].counter)
 555           {
 556             if (ce->values_raw[i].counter <= 4294967295U)
 557               diff = (4294967295U - ce->values_raw[i].counter)
 558                 + vl->values[i].counter;
 559             else
 560               diff = (18446744073709551615ULL - ce->values_raw[i].counter)
 561                 + vl->values[i].counter;
 562           }
 563           else /* counter has NOT wrapped around */
 564           {
 565             diff = vl->values[i].counter - ce->values_raw[i].counter;
 566           }
 567
 568           ce->values_gauge[i] = ((double) diff)
 569             / ((double) (vl->time - ce->last_time));
 570           ce->values_raw[i].counter = vl->values[i].counter;
 571         }
 572         break;
 573
 574       case DS_TYPE_GAUGE:
 575         ce->values_raw[i].gauge = vl->values[i].gauge;
 576         ce->values_gauge[i] = vl->values[i].gauge;
 577         break;
 578
 579       case DS_TYPE_DERIVE:
 580         {
 581           derive_t diff;
 582
 583           diff = vl->values[i].derive - ce->values_raw[i].derive;
 584
 585           ce->values_gauge[i] = ((double) diff)
 586             / ((double) (vl->time - ce->last_time));
 587           ce->values_raw[i].derive = vl->values[i].derive;
 588         }
 589         break;
 590
 591       case DS_TYPE_ABSOLUTE:
 592         ce->values_gauge[i] = ((double) vl->values[i].absolute)
 593           / ((double) (vl->time - ce->last_time));
 594         ce->values_raw[i].absolute = vl->values[i].absolute;
 595         break;
 596
 597       default:
 598         /* This shouldn't happen. */
 599         pthread_mutex_unlock (&cache_lock);
 600         ERROR ("uc_update: Don't know how to handle data source type %i.",
 601             ds->ds[i].type);
 602         return (-1);
 603     } /* switch (ds->ds[i].type) */
 604
 605     DEBUG ("uc_update: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]);
 606   } /* for (i) */
 607
 608   /* Update the history if it exists. */
 609   if (ce->history != NULL)
 610   {
 611     assert (ce->history_index < ce->history_length);
 612     for (i = 0; i < ce->values_num; i++)
 613     {
 614       size_t hist_idx = (ce->values_num * ce->history_index) + i;
 615       ce->history[hist_idx] = ce->values_gauge[i];
 616     }
 617
 618     assert (ce->history_length > 0);
 619     ce->history_index = (ce->history_index + 1) % ce->history_length;
 620   }
 621
 622   /* Prune invalid gauge data */
 623   uc_check_range (ds, ce);
 624
 625   ce->last_time = vl->time;
 626   ce->last_update = time (NULL);
 627   ce->interval = vl->interval;
 628
 629   pthread_mutex_unlock (&cache_lock);
 630
 631   if (send_okay_notification == 0)
 632     return (0);
 633
 634   /* Do not send okay notifications for uninteresting values, i. e. values for
 635    * which no threshold is configured. */
 636   status = ut_check_interesting (name);
 637   if (status <= 0)
 638     return (0);
 639
 640   /* Initialize the notification */
 641   memset (&n, '\0', sizeof (n));
 642   NOTIFICATION_INIT_VL (&n, vl, ds);
 643
 644   n.severity = NOTIF_OKAY;
 645   n.time = vl->time;
 646
 647   ssnprintf (n.message, sizeof (n.message),
 648       "Received a value for %s. It was missing for %u seconds.",
 649       name, (unsigned int) update_delay);
 650
 651   plugin_dispatch_notification (&n);
 652
 653   return (0);
 654 } /* int uc_update */
 655
 656 int uc_get_rate_by_name (const char *name, gauge_t **ret_values, size_t *ret_values_num)
 657 {
 658   gauge_t *ret = NULL;
 659   size_t ret_num = 0;
 660   cache_entry_t *ce = NULL;
 661   int status = 0;
 662
 663   pthread_mutex_lock (&cache_lock);
 664
 665   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
 666   {
 667     assert (ce != NULL);
 668
 669     /* remove missing values from getval */
 670     if (ce->state == STATE_MISSING)
 671     {
 672       status = -1;
 673     }
 674     else
 675     {
 676       ret_num = ce->values_num;
 677       ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t));
 678       if (ret == NULL)
 679       {
 680         ERROR ("utils_cache: uc_get_rate_by_name: malloc failed.");
 681         status = -1;
 682       }
 683       else
 684       {
 685         memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t));
 686       }
 687     }
 688   }
 689   else
 690   {
 691     DEBUG ("utils_cache: uc_get_rate_by_name: No such value: %s", name);
 692     status = -1;
 693   }
 694
 695   pthread_mutex_unlock (&cache_lock);
 696
 697   if (status == 0)
 698   {
 699     *ret_values = ret;
 700     *ret_values_num = ret_num;
 701   }
 702
 703   return (status);
 704 } /* gauge_t *uc_get_rate_by_name */
 705
 706 gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl)
 707 {
 708   char name[6 * DATA_MAX_NAME_LEN];
 709   gauge_t *ret = NULL;
 710   size_t ret_num = 0;
 711   int status;
 712
 713   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 714   {
 715     ERROR ("utils_cache: uc_get_rate: FORMAT_VL failed.");
 716     return (NULL);
 717   }
 718
 719   status = uc_get_rate_by_name (name, &ret, &ret_num);
 720   if (status != 0)
 721     return (NULL);
 722
 723   /* This is important - the caller has no other way of knowing how many
 724    * values are returned. */
 725   if (ret_num != (size_t) ds->ds_num)
 726   {
 727     ERROR ("utils_cache: uc_get_rate: ds[%s] has %i values, "
 728         "but uc_get_rate_by_name returned %zu.",
 729         ds->type, ds->ds_num, ret_num);
 730     sfree (ret);
 731     return (NULL);
 732   }
 733
 734   return (ret);
 735 } /* gauge_t *uc_get_rate */
 736
 737 int uc_get_names (char ***ret_names, time_t **ret_times, size_t *ret_number)
 738 {
 739   c_avl_iterator_t *iter;
 740   char *key;
 741   cache_entry_t *value;
 742
 743   char **names = NULL;
 744   time_t *times = NULL;
 745   size_t number = 0;
 746
 747   int status = 0;
 748
 749   if ((ret_names == NULL) || (ret_number == NULL))
 750     return (-1);
 751
 752   pthread_mutex_lock (&cache_lock);
 753
 754   iter = c_avl_get_iterator (cache_tree);
 755   while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
 756   {
 757     char **temp;
 758
 759     /* remove missing values when list values */
 760     if (value->state == STATE_MISSING)
 761       continue;
 762
 763     if (ret_times != NULL)
 764     {
 765       time_t *tmp_times;
 766
 767       tmp_times = (time_t *) realloc (times, sizeof (time_t) * (number + 1));
 768       if (tmp_times == NULL)
 769       {
 770         status = -1;
 771         break;
 772       }
 773       times = tmp_times;
 774       times[number] = value->last_time;
 775     }
 776
 777     temp = (char **) realloc (names, sizeof (char *) * (number + 1));
 778     if (temp == NULL)
 779     {
 780       status = -1;
 781       break;
 782     }
 783     names = temp;
 784     names[number] = strdup (key);
 785     if (names[number] == NULL)
 786     {
 787       status = -1;
 788       break;
 789     }
 790     number++;
 791   } /* while (c_avl_iterator_next) */
 792
 793   c_avl_iterator_destroy (iter);
 794   pthread_mutex_unlock (&cache_lock);
 795
 796   if (status != 0)
 797   {
 798     size_t i;
 799
 800     for (i = 0; i < number; i++)
 801     {
 802       sfree (names[i]);
 803     }
 804     sfree (names);
 805
 806     return (-1);
 807   }
 808
 809   *ret_names = names;
 810   if (ret_times != NULL)
 811     *ret_times = times;
 812   *ret_number = number;
 813
 814   return (0);
 815 } /* int uc_get_names */
 816
 817 int uc_get_state (const data_set_t *ds, const value_list_t *vl)
 818 {
 819   char name[6 * DATA_MAX_NAME_LEN];
 820   cache_entry_t *ce = NULL;
 821   int ret = STATE_ERROR;
 822
 823   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 824   {
 825     ERROR ("uc_get_state: FORMAT_VL failed.");
 826     return (STATE_ERROR);
 827   }
 828
 829   pthread_mutex_lock (&cache_lock);
 830
 831   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
 832   {
 833     assert (ce != NULL);
 834     ret = ce->state;
 835   }
 836
 837   pthread_mutex_unlock (&cache_lock);
 838
 839   return (ret);
 840 } /* int uc_get_state */
 841
 842 int uc_set_state (const data_set_t *ds, const value_list_t *vl, int state)
 843 {
 844   char name[6 * DATA_MAX_NAME_LEN];
 845   cache_entry_t *ce = NULL;
 846   int ret = -1;
 847
 848   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 849   {
 850     ERROR ("uc_get_state: FORMAT_VL failed.");
 851     return (STATE_ERROR);
 852   }
 853
 854   pthread_mutex_lock (&cache_lock);
 855
 856   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
 857   {
 858     assert (ce != NULL);
 859     ret = ce->state;
 860     ce->state = state;
 861   }
 862
 863   pthread_mutex_unlock (&cache_lock);
 864
 865   return (ret);
 866 } /* int uc_set_state */
 867
 868 int uc_get_history_by_name (const char *name,
 869     gauge_t *ret_history, size_t num_steps, size_t num_ds)
 870 {
 871   cache_entry_t *ce = NULL;
 872   size_t i;
 873   int status = 0;
 874
 875   pthread_mutex_lock (&cache_lock);
 876
 877   status = c_avl_get (cache_tree, name, (void *) &ce);
 878   if (status != 0)
 879   {
 880     pthread_mutex_unlock (&cache_lock);
 881     return (-ENOENT);
 882   }
 883
 884   if (((size_t) ce->values_num) != num_ds)
 885   {
 886     pthread_mutex_unlock (&cache_lock);
 887     return (-EINVAL);
 888   }
 889
 890   /* Check if there are enough values available. If not, increase the buffer
 891    * size. */
 892   if (ce->history_length < num_steps)
 893   {
 894     gauge_t *tmp;
 895     size_t i;
 896
 897     tmp = realloc (ce->history, sizeof (*ce->history)
 898         * num_steps * ce->values_num);
 899     if (tmp == NULL)
 900     {
 901       pthread_mutex_unlock (&cache_lock);
 902       return (-ENOMEM);
 903     }
 904
 905     for (i = ce->history_length * ce->values_num;
 906         i < (num_steps * ce->values_num);
 907         i++)
 908       tmp[i] = NAN;
 909
 910     ce->history = tmp;
 911     ce->history_length = num_steps;
 912   } /* if (ce->history_length < num_steps) */
 913
 914   /* Copy the values to the output buffer. */
 915   for (i = 0; i < num_steps; i++)
 916   {
 917     size_t src_index;
 918     size_t dst_index;
 919
 920     if (i < ce->history_index)
 921       src_index = ce->history_index - (i + 1);
 922     else
 923       src_index = ce->history_length + ce->history_index - (i + 1);
 924     src_index = src_index * num_ds;
 925
 926     dst_index = i * num_ds;
 927
 928     memcpy (ret_history + dst_index, ce->history + src_index,
 929         sizeof (*ret_history) * num_ds);
 930   }
 931
 932   pthread_mutex_unlock (&cache_lock);
 933
 934   return (0);
 935 } /* int uc_get_history_by_name */
 936
 937 int uc_get_history (const data_set_t *ds, const value_list_t *vl,
 938     gauge_t *ret_history, size_t num_steps, size_t num_ds)
 939 {
 940   char name[6 * DATA_MAX_NAME_LEN];
 941
 942   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 943   {
 944     ERROR ("utils_cache: uc_get_history: FORMAT_VL failed.");
 945     return (-1);
 946   }
 947
 948   return (uc_get_history_by_name (name, ret_history, num_steps, num_ds));
 949 } /* int uc_get_history */
 950
 951 int uc_get_hits (const data_set_t *ds, const value_list_t *vl)
 952 {
 953   char name[6 * DATA_MAX_NAME_LEN];
 954   cache_entry_t *ce = NULL;
 955   int ret = STATE_ERROR;
 956
 957   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 958   {
 959     ERROR ("uc_get_state: FORMAT_VL failed.");
 960     return (STATE_ERROR);
 961   }
 962
 963   pthread_mutex_lock (&cache_lock);
 964
 965   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
 966   {
 967     assert (ce != NULL);
 968     ret = ce->hits;
 969   }
 970
 971   pthread_mutex_unlock (&cache_lock);
 972
 973   return (ret);
 974 } /* int uc_get_hits */
 975
 976 int uc_set_hits (const data_set_t *ds, const value_list_t *vl, int hits)
 977 {
 978   char name[6 * DATA_MAX_NAME_LEN];
 979   cache_entry_t *ce = NULL;
 980   int ret = -1;
 981
 982   if (FORMAT_VL (name, sizeof (name), vl) != 0)
 983   {
 984     ERROR ("uc_get_state: FORMAT_VL failed.");
 985     return (STATE_ERROR);
 986   }
 987
 988   pthread_mutex_lock (&cache_lock);
 989
 990   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
 991   {
 992     assert (ce != NULL);
 993     ret = ce->hits;
 994     ce->hits = hits;
 995   }
 996
 997   pthread_mutex_unlock (&cache_lock);
 998
 999   return (ret);
1000 } /* int uc_set_hits */
1001
1002 int uc_inc_hits (const data_set_t *ds, const value_list_t *vl, int step)
1003 {
1004   char name[6 * DATA_MAX_NAME_LEN];
1005   cache_entry_t *ce = NULL;
1006   int ret = -1;
1007
1008   if (FORMAT_VL (name, sizeof (name), vl) != 0)
1009   {
1010     ERROR ("uc_get_state: FORMAT_VL failed.");
1011     return (STATE_ERROR);
1012   }
1013
1014   pthread_mutex_lock (&cache_lock);
1015
1016   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
1017   {
1018     assert (ce != NULL);
1019     ret = ce->hits;
1020     ce->hits = ret + step;
1021   }
1022
1023   pthread_mutex_unlock (&cache_lock);
1024
1025   return (ret);
1026 } /* int uc_inc_hits */
1027
1028 /*
1029  * Meta data interface
1030  */
1031 /* XXX: This function will acquire `cache_lock' but will not free it! */
1032 static meta_data_t *uc_get_meta (const value_list_t *vl) /* {{{ */
1033 {
1034   char name[6 * DATA_MAX_NAME_LEN];
1035   cache_entry_t *ce = NULL;
1036   int status;
1037
1038   status = FORMAT_VL (name, sizeof (name), vl);
1039   if (status != 0)
1040   {
1041     ERROR ("utils_cache: uc_get_meta: FORMAT_VL failed.");
1042     return (NULL);
1043   }
1044
1045   pthread_mutex_lock (&cache_lock);
1046
1047   status = c_avl_get (cache_tree, name, (void *) &ce);
1048   if (status != 0)
1049   {
1050     pthread_mutex_unlock (&cache_lock);
1051     return (NULL);
1052   }
1053   assert (ce != NULL);
1054
1055   if (ce->meta == NULL)
1056     ce->meta = meta_data_create ();
1057
1058   if (ce->meta == NULL)
1059     pthread_mutex_unlock (&cache_lock);
1060
1061   return (ce->meta);
1062 } /* }}} meta_data_t *uc_get_meta */
1063
1064 /* Sorry about this preprocessor magic, but it really makes this file much
1065  * shorter.. */
1066 #define UC_WRAP(wrap_function) { \
1067   meta_data_t *meta; \
1068   int status; \
1069   meta = uc_get_meta (vl); \
1070   if (meta == NULL) return (-1); \
1071   status = wrap_function (meta, key); \
1072   pthread_mutex_unlock (&cache_lock); \
1073   return (status); \
1074 }
1075 int uc_meta_data_exists (const value_list_t *vl, const char *key)
1076   UC_WRAP (meta_data_exists)
1077
1078 int uc_meta_data_delete (const value_list_t *vl, const char *key)
1079   UC_WRAP (meta_data_delete)
1080 #undef UC_WRAP
1081
1082 /* We need a new version of this macro because the following functions take
1083  * two argumetns. */
1084 #define UC_WRAP(wrap_function) { \
1085   meta_data_t *meta; \
1086   int status; \
1087   meta = uc_get_meta (vl); \
1088   if (meta == NULL) return (-1); \
1089   status = wrap_function (meta, key, value); \
1090   pthread_mutex_unlock (&cache_lock); \
1091   return (status); \
1092 }
1093 int uc_meta_data_add_string (const value_list_t *vl,
1094     const char *key,
1095     const char *value)
1096   UC_WRAP(meta_data_add_string)
1097 int uc_meta_data_add_signed_int (const value_list_t *vl,
1098     const char *key,
1099     int64_t value)
1100   UC_WRAP(meta_data_add_signed_int)
1101 int uc_meta_data_add_unsigned_int (const value_list_t *vl,
1102     const char *key,
1103     uint64_t value)
1104   UC_WRAP(meta_data_add_unsigned_int)
1105 int uc_meta_data_add_double (const value_list_t *vl,
1106     const char *key,
1107     double value)
1108   UC_WRAP(meta_data_add_double)
1109 int uc_meta_data_add_boolean (const value_list_t *vl,
1110     const char *key,
1111     _Bool value)
1112   UC_WRAP(meta_data_add_boolean)
1113
1114 int uc_meta_data_get_string (const value_list_t *vl,
1115     const char *key,
1116     char **value)
1117   UC_WRAP(meta_data_get_string)
1118 int uc_meta_data_get_signed_int (const value_list_t *vl,
1119     const char *key,
1120     int64_t *value)
1121   UC_WRAP(meta_data_get_signed_int)
1122 int uc_meta_data_get_unsigned_int (const value_list_t *vl,
1123     const char *key,
1124     uint64_t *value)
1125   UC_WRAP(meta_data_get_unsigned_int)
1126 int uc_meta_data_get_double (const value_list_t *vl,
1127     const char *key,
1128     double *value)
1129   UC_WRAP(meta_data_get_double)
1130 int uc_meta_data_get_boolean (const value_list_t *vl,
1131     const char *key,
1132     _Bool *value)
1133   UC_WRAP(meta_data_get_boolean)
1134 #undef UC_WRAP
1135
1136 /* vim: set sw=2 ts=8 sts=2 tw=78 : */