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