src/core/plugin.c

   1 /*
   2  * SysDB - src/core/plugin.c
   3  * Copyright (C) 2012 Sebastian 'tokkee' Harl <sh@tokkee.org>
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions
   8  * are met:
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  *
  15  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  16  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
  17  * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  18  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR
  19  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
  20  * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  21  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  22  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  23  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  25  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27
  28 #if HAVE_CONFIG_H
  29 #       include "config.h"
  30 #endif /* HAVE_CONFIG_H */
  31
  32 #include "sysdb.h"
  33 #include "core/plugin.h"
  34 #include "core/time.h"
  35 #include "utils/error.h"
  36 #include "utils/llist.h"
  37 #include "utils/strbuf.h"
  38
  39 #include <assert.h>
  40
  41 #include <errno.h>
  42
  43 #include <stdarg.h>
  44 #include <stdio.h>
  45 #include <string.h>
  46 #include <strings.h>
  47 #include <unistd.h>
  48
  49 #include <ltdl.h>
  50
  51 #include <pthread.h>
  52
  53 /* helper to access info attributes */
  54 #define INFO_GET(i, attr) \
  55         ((i)->attr ? (i)->attr : #attr" not set")
  56
  57 /*
  58  * private data types
  59  */
  60
  61 typedef struct {
  62         sdb_object_t super;
  63         sdb_plugin_ctx_t public;
  64
  65         sdb_plugin_info_t info;
  66         lt_dlhandle handle;
  67
  68         /* The usage count differs from the object's ref count
  69          * in that it provides higher level information about how
  70          * the plugin is in use. */
  71         size_t use_cnt;
  72 } ctx_t;
  73 #define CTX_INIT { SDB_OBJECT_INIT, \
  74         SDB_PLUGIN_CTX_INIT, SDB_PLUGIN_INFO_INIT, NULL, 0 }
  75
  76 #define CTX(obj) ((ctx_t *)(obj))
  77
  78 typedef struct {
  79         sdb_object_t super;
  80         void *cb_callback;
  81         sdb_object_t *cb_user_data;
  82         ctx_t *cb_ctx;
  83 } callback_t;
  84 #define CB_INIT { SDB_OBJECT_INIT, \
  85         /* callback = */ NULL, /* user_data = */ NULL, \
  86         SDB_PLUGIN_CTX_INIT }
  87 #define CB(obj) ((callback_t *)(obj))
  88 #define CONST_CB(obj) ((const callback_t *)(obj))
  89
  90 typedef struct {
  91         callback_t super;
  92 #define ccb_callback super.cb_callback
  93 #define ccb_user_data super.cb_user_data
  94 #define ccb_ctx super.cb_ctx
  95         sdb_time_t ccb_interval;
  96         sdb_time_t ccb_next_update;
  97 } collector_t;
  98 #define CCB(obj) ((collector_t *)(obj))
  99 #define CONST_CCB(obj) ((const collector_t *)(obj))
 100
 101 typedef struct {
 102         callback_t super; /* cb_callback will always be NULL */
 103 #define w_user_data super.cb_user_data
 104 #define w_ctx super.cb_ctx
 105         sdb_store_writer_t impl;
 106 } writer_t;
 107 #define WRITER(obj) ((writer_t *)(obj))
 108
 109 typedef struct {
 110         callback_t super; /* cb_callback will always be NULL */
 111 #define r_user_data super.cb_user_data
 112 #define r_ctx super.cb_ctx
 113         sdb_store_reader_t impl;
 114 } reader_t;
 115 #define READER(obj) ((reader_t *)(obj))
 116
 117 /*
 118  * private variables
 119  */
 120
 121 static sdb_plugin_ctx_t  plugin_default_ctx  = SDB_PLUGIN_CTX_INIT;
 122 static sdb_plugin_info_t plugin_default_info = SDB_PLUGIN_INFO_INIT;
 123
 124 static pthread_key_t     plugin_ctx_key;
 125 static bool              plugin_ctx_key_initialized = 0;
 126
 127 /* a list of the plugin contexts of all registered plugins */
 128 static sdb_llist_t      *all_plugins = NULL;
 129
 130 static sdb_llist_t      *config_list = NULL;
 131 static sdb_llist_t      *init_list = NULL;
 132 static sdb_llist_t      *collector_list = NULL;
 133 static sdb_llist_t      *cname_list = NULL;
 134 static sdb_llist_t      *shutdown_list = NULL;
 135 static sdb_llist_t      *log_list = NULL;
 136 static sdb_llist_t      *ts_fetcher_list = NULL;
 137 static sdb_llist_t      *writer_list = NULL;
 138 static sdb_llist_t      *reader_list = NULL;
 139
 140 static struct {
 141         const char   *type;
 142         sdb_llist_t **list;
 143 } all_lists[] = {
 144         { "config",             &config_list },
 145         { "init",               &init_list },
 146         { "collector",          &collector_list },
 147         { "cname",              &cname_list },
 148         { "shutdown",           &shutdown_list },
 149         { "log",                &log_list },
 150         { "timeseries fetcher", &ts_fetcher_list },
 151         { "store writer",       &writer_list },
 152         { "store reader",       &reader_list },
 153 };
 154
 155 /*
 156  * private helper functions
 157  */
 158
 159 static void
 160 plugin_info_clear(sdb_plugin_info_t *info)
 161 {
 162         sdb_plugin_info_t empty_info = SDB_PLUGIN_INFO_INIT;
 163         if (! info)
 164                 return;
 165
 166         if (info->plugin_name)
 167                 free(info->plugin_name);
 168         if (info->filename)
 169                 free(info->filename);
 170
 171         if (info->description)
 172                 free(info->description);
 173         if (info->copyright)
 174                 free(info->copyright);
 175         if (info->license)
 176                 free(info->license);
 177
 178         *info = empty_info;
 179 } /* plugin_info_clear */
 180
 181 static void
 182 ctx_key_init(void)
 183 {
 184         if (plugin_ctx_key_initialized)
 185                 return;
 186
 187         pthread_key_create(&plugin_ctx_key, /* destructor */ NULL);
 188         plugin_ctx_key_initialized = 1;
 189 } /* ctx_key_init */
 190
 191 static int
 192 plugin_cmp_next_update(const sdb_object_t *a, const sdb_object_t *b)
 193 {
 194         const collector_t *ccb1 = (const collector_t *)a;
 195         const collector_t *ccb2 = (const collector_t *)b;
 196
 197         assert(ccb1 && ccb2);
 198
 199         return (ccb1->ccb_next_update > ccb2->ccb_next_update)
 200                 ? 1 : (ccb1->ccb_next_update < ccb2->ccb_next_update)
 201                 ? -1 : 0;
 202 } /* plugin_cmp_next_update */
 203
 204 static int
 205 plugin_lookup_by_name(const sdb_object_t *obj, const void *id)
 206 {
 207         const callback_t *cb = CONST_CB(obj);
 208         const char *name = id;
 209
 210         assert(cb && id);
 211
 212         /* when a plugin was registered from outside a plugin (e.g. the core),
 213          * we don't have a plugin context */
 214         if (! cb->cb_ctx)
 215                 return 1;
 216
 217         if (!strcasecmp(cb->cb_ctx->info.plugin_name, name))
 218                 return 0;
 219         return 1;
 220 } /* plugin_lookup_by_name */
 221
 222 /* since this function is called from sdb_plugin_reconfigure_finish()
 223  * when iterating through all_plugins, we may not do any additional
 224  * modifications to all_plugins except for the optional removal */
 225 static void
 226 plugin_unregister_by_name(const char *plugin_name)
 227 {
 228         sdb_object_t *obj;
 229         size_t i;
 230
 231         for (i = 0; i < SDB_STATIC_ARRAY_LEN(all_lists); ++i) {
 232                 const char  *type =  all_lists[i].type;
 233                 sdb_llist_t *list = *all_lists[i].list;
 234
 235                 while (1) {
 236                         callback_t *cb;
 237
 238                         cb = CB(sdb_llist_remove(list,
 239                                                 plugin_lookup_by_name, plugin_name));
 240                         if (! cb)
 241                                 break;
 242
 243                         assert(cb->cb_ctx);
 244
 245                         sdb_log(SDB_LOG_INFO, "core: Unregistering "
 246                                         "%s callback '%s' (module %s)", type, cb->super.name,
 247                                         cb->cb_ctx->info.plugin_name);
 248                         sdb_object_deref(SDB_OBJ(cb));
 249                 }
 250         }
 251
 252         obj = sdb_llist_search_by_name(all_plugins, plugin_name);
 253         /* when called from sdb_plugin_reconfigure_finish, the object has already
 254          * been removed from the list */
 255         if (obj && (obj->ref_cnt <= 1)) {
 256                 sdb_llist_remove_by_name(all_plugins, plugin_name);
 257                 sdb_object_deref(obj);
 258         }
 259         /* else: other callbacks still reference it */
 260 } /* plugin_unregister_by_name */
 261
 262 /*
 263  * private types
 264  */
 265
 266 static int
 267 ctx_init(sdb_object_t *obj, va_list __attribute__((unused)) ap)
 268 {
 269         ctx_t *ctx = CTX(obj);
 270
 271         assert(ctx);
 272
 273         ctx->public = plugin_default_ctx;
 274         ctx->info = plugin_default_info;
 275         ctx->handle = NULL;
 276         ctx->use_cnt = 1;
 277         return 0;
 278 } /* ctx_init */
 279
 280 static void
 281 ctx_destroy(sdb_object_t *obj)
 282 {
 283         ctx_t *ctx = CTX(obj);
 284
 285         if (ctx->handle) {
 286                 const char *err;
 287
 288                 sdb_log(SDB_LOG_INFO, "core: Unloading module %s",
 289                                 ctx->info.plugin_name);
 290
 291                 lt_dlerror();
 292                 lt_dlclose(ctx->handle);
 293                 if ((err = lt_dlerror()))
 294                         sdb_log(SDB_LOG_WARNING, "core: Failed to unload module %s: %s",
 295                                         ctx->info.plugin_name, err);
 296         }
 297
 298         plugin_info_clear(&ctx->info);
 299 } /* ctx_destroy */
 300
 301 static sdb_type_t ctx_type = {
 302         sizeof(ctx_t),
 303
 304         ctx_init,
 305         ctx_destroy
 306 };
 307
 308 static ctx_t *
 309 ctx_get(void)
 310 {
 311         if (! plugin_ctx_key_initialized)
 312                 ctx_key_init();
 313         return pthread_getspecific(plugin_ctx_key);
 314 } /* ctx_get */
 315
 316 static ctx_t *
 317 ctx_set(ctx_t *new)
 318 {
 319         ctx_t *old;
 320
 321         if (! plugin_ctx_key_initialized)
 322                 ctx_key_init();
 323
 324         old = pthread_getspecific(plugin_ctx_key);
 325         if (old)
 326                 sdb_object_deref(SDB_OBJ(old));
 327         if (new)
 328                 sdb_object_ref(SDB_OBJ(new));
 329         pthread_setspecific(plugin_ctx_key, new);
 330         return old;
 331 } /* ctx_set */
 332
 333 static ctx_t *
 334 ctx_create(const char *name)
 335 {
 336         ctx_t *ctx;
 337
 338         ctx = CTX(sdb_object_create(name, ctx_type));
 339         if (! ctx)
 340                 return NULL;
 341
 342         if (! plugin_ctx_key_initialized)
 343                 ctx_key_init();
 344         ctx_set(ctx);
 345         return ctx;
 346 } /* ctx_create */
 347
 348 static int
 349 plugin_cb_init(sdb_object_t *obj, va_list ap)
 350 {
 351         sdb_llist_t **list = va_arg(ap, sdb_llist_t **);
 352         const char   *type = va_arg(ap, const char *);
 353         void     *callback = va_arg(ap, void *);
 354         sdb_object_t   *ud = va_arg(ap, sdb_object_t *);
 355
 356         assert(list);
 357         assert(type);
 358         assert(obj);
 359
 360         if (sdb_llist_search_by_name(*list, obj->name)) {
 361                 sdb_log(SDB_LOG_WARNING, "core: %s callback '%s' "
 362                                 "has already been registered. Ignoring newly "
 363                                 "registered version.", type, obj->name);
 364                 return -1;
 365         }
 366
 367         /* cb_ctx may be NULL if the plugin was not registered by a plugin */
 368
 369         CB(obj)->cb_callback = callback;
 370         CB(obj)->cb_ctx      = ctx_get();
 371         sdb_object_ref(SDB_OBJ(CB(obj)->cb_ctx));
 372
 373         sdb_object_ref(ud);
 374         CB(obj)->cb_user_data = ud;
 375         return 0;
 376 } /* plugin_cb_init */
 377
 378 static void
 379 plugin_cb_destroy(sdb_object_t *obj)
 380 {
 381         assert(obj);
 382         sdb_object_deref(CB(obj)->cb_user_data);
 383         sdb_object_deref(SDB_OBJ(CB(obj)->cb_ctx));
 384 } /* plugin_cb_destroy */
 385
 386 static sdb_type_t callback_type = {
 387         sizeof(callback_t),
 388
 389         plugin_cb_init,
 390         plugin_cb_destroy
 391 };
 392
 393 static sdb_type_t collector_type = {
 394         sizeof(collector_t),
 395
 396         plugin_cb_init,
 397         plugin_cb_destroy
 398 };
 399
 400 static int
 401 plugin_writer_init(sdb_object_t *obj, va_list ap)
 402 {
 403         sdb_store_writer_t *impl = va_arg(ap, sdb_store_writer_t *);
 404         sdb_object_t       *ud   = va_arg(ap, sdb_object_t *);
 405
 406         assert(impl);
 407
 408         if ((! impl->store_host) || (! impl->store_service)
 409                         || (! impl->store_metric) || (! impl->store_attribute)) {
 410                 sdb_log(SDB_LOG_ERR, "core: store writer callback '%s' "
 411                                 "does not fully implement the writer interface.",
 412                                 obj->name);
 413                 return -1;
 414         }
 415         if (sdb_llist_search_by_name(writer_list, obj->name)) {
 416                 sdb_log(SDB_LOG_WARNING, "core: store writer callback '%s' "
 417                                 "has already been registered. Ignoring newly "
 418                                 "registered version.", obj->name);
 419                 return -1;
 420         }
 421
 422         /* ctx may be NULL if the callback was not registered by a plugin */
 423
 424         WRITER(obj)->impl = *impl;
 425         WRITER(obj)->w_ctx  = ctx_get();
 426         sdb_object_ref(SDB_OBJ(WRITER(obj)->w_ctx));
 427
 428         sdb_object_ref(ud);
 429         WRITER(obj)->w_user_data = ud;
 430         return 0;
 431 } /* plugin_writer_init */
 432
 433 static void
 434 plugin_writer_destroy(sdb_object_t *obj)
 435 {
 436         assert(obj);
 437         sdb_object_deref(WRITER(obj)->w_user_data);
 438         sdb_object_deref(SDB_OBJ(WRITER(obj)->w_ctx));
 439 } /* plugin_writer_destroy */
 440
 441 static sdb_type_t writer_type = {
 442         sizeof(writer_t),
 443
 444         plugin_writer_init,
 445         plugin_writer_destroy
 446 };
 447
 448 static int
 449 plugin_reader_init(sdb_object_t *obj, va_list ap)
 450 {
 451         sdb_store_reader_t *impl = va_arg(ap, sdb_store_reader_t *);
 452         sdb_object_t       *ud   = va_arg(ap, sdb_object_t *);
 453
 454         assert(impl);
 455
 456         if ((! impl->prepare_query) || (! impl->execute_query)) {
 457                 sdb_log(SDB_LOG_ERR, "core: store reader callback '%s' "
 458                                 "does not fully implement the reader interface.",
 459                                 obj->name);
 460                 return -1;
 461         }
 462         if (sdb_llist_search_by_name(reader_list, obj->name)) {
 463                 sdb_log(SDB_LOG_WARNING, "core: store reader callback '%s' "
 464                                 "has already been registered. Ignoring newly "
 465                                 "registered version.", obj->name);
 466                 return -1;
 467         }
 468
 469         /* ctx may be NULL if the callback was not registered by a plugin */
 470
 471         READER(obj)->impl = *impl;
 472         READER(obj)->r_ctx  = ctx_get();
 473         sdb_object_ref(SDB_OBJ(READER(obj)->r_ctx));
 474
 475         sdb_object_ref(ud);
 476         READER(obj)->r_user_data = ud;
 477         return 0;
 478 } /* plugin_reader_init */
 479
 480 static void
 481 plugin_reader_destroy(sdb_object_t *obj)
 482 {
 483         assert(obj);
 484         sdb_object_deref(READER(obj)->r_user_data);
 485         sdb_object_deref(SDB_OBJ(READER(obj)->r_ctx));
 486 } /* plugin_reader_destroy */
 487
 488 static sdb_type_t reader_type = {
 489         sizeof(reader_t),
 490
 491         plugin_reader_init,
 492         plugin_reader_destroy
 493 };
 494
 495 static int
 496 module_init(const char *name, lt_dlhandle lh, sdb_plugin_info_t *info)
 497 {
 498         int (*mod_init)(sdb_plugin_info_t *);
 499         int status;
 500
 501         mod_init = (int (*)(sdb_plugin_info_t *))lt_dlsym(lh, "sdb_module_init");
 502         if (! mod_init) {
 503                 sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s': "
 504                                 "could not find symbol 'sdb_module_init'", name);
 505                 return -1;
 506         }
 507
 508         status = mod_init(info);
 509         if (status) {
 510                 sdb_log(SDB_LOG_ERR, "core: Failed to initialize "
 511                                 "module '%s'", name);
 512                 plugin_unregister_by_name(name);
 513                 return -1;
 514         }
 515         return 0;
 516 } /* module_init */
 517
 518 static int
 519 module_load(const char *basedir, const char *name,
 520                 const sdb_plugin_ctx_t *plugin_ctx)
 521 {
 522         char  base_name[name ? strlen(name) + 1 : 1];
 523         const char *name_ptr;
 524         char *tmp;
 525
 526         char filename[1024];
 527         lt_dlhandle lh;
 528
 529         ctx_t *ctx;
 530
 531         int status;
 532
 533         assert(name);
 534
 535         base_name[0] = '\0';
 536         name_ptr = name;
 537
 538         while ((tmp = strstr(name_ptr, "::"))) {
 539                 strncat(base_name, name_ptr, (size_t)(tmp - name_ptr));
 540                 strcat(base_name, "/");
 541                 name_ptr = tmp + strlen("::");
 542         }
 543         strcat(base_name, name_ptr);
 544
 545         if (! basedir)
 546                 basedir = PKGLIBDIR;
 547
 548         snprintf(filename, sizeof(filename), "%s/%s.so", basedir, base_name);
 549         filename[sizeof(filename) - 1] = '\0';
 550
 551         if (access(filename, R_OK)) {
 552                 char errbuf[1024];
 553                 sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s' (%s): %s",
 554                                 name, filename, sdb_strerror(errno, errbuf, sizeof(errbuf)));
 555                 return -1;
 556         }
 557
 558         lt_dlinit();
 559         lt_dlerror();
 560
 561         lh = lt_dlopen(filename);
 562         if (! lh) {
 563                 sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s': %s"
 564                                 "The most common cause for this problem are missing "
 565                                 "dependencies.\n", name, lt_dlerror());
 566                 return -1;
 567         }
 568
 569         if (ctx_get())
 570                 sdb_log(SDB_LOG_WARNING, "core: Discarding old plugin context");
 571
 572         ctx = ctx_create(name);
 573         if (! ctx) {
 574                 sdb_log(SDB_LOG_ERR, "core: Failed to initialize plugin context");
 575                 return -1;
 576         }
 577
 578         ctx->info.plugin_name = strdup(name);
 579         ctx->info.filename = strdup(filename);
 580         ctx->handle = lh;
 581
 582         if (plugin_ctx)
 583                 ctx->public = *plugin_ctx;
 584
 585         if ((status = module_init(name, lh, &ctx->info))) {
 586                 sdb_object_deref(SDB_OBJ(ctx));
 587                 return status;
 588         }
 589
 590         /* compare minor version */
 591         if ((ctx->info.version < 0)
 592                         || ((int)(ctx->info.version / 100) != (int)(SDB_VERSION / 100)))
 593                 sdb_log(SDB_LOG_WARNING, "core: WARNING: version of "
 594                                 "plugin '%s' (%i.%i.%i) does not match our version "
 595                                 "(%i.%i.%i); this might cause problems",
 596                                 name, SDB_VERSION_DECODE(ctx->info.version),
 597                                 SDB_VERSION_DECODE(SDB_VERSION));
 598
 599         sdb_llist_append(all_plugins, SDB_OBJ(ctx));
 600
 601         sdb_log(SDB_LOG_INFO, "core: Successfully loaded "
 602                         "plugin %s v%i (%s)", ctx->info.plugin_name,
 603                         ctx->info.plugin_version,
 604                         INFO_GET(&ctx->info, description));
 605         sdb_log(SDB_LOG_INFO, "core: Plugin %s: %s, License: %s",
 606                         ctx->info.plugin_name,
 607                         INFO_GET(&ctx->info, copyright),
 608                         INFO_GET(&ctx->info, license));
 609
 610         /* any registered callbacks took ownership of the context */
 611         sdb_object_deref(SDB_OBJ(ctx));
 612
 613         /* reset */
 614         ctx_set(NULL);
 615         return 0;
 616 } /* module_load */
 617
 618 static char *
 619 plugin_get_name(const char *name, char *buf, size_t bufsize)
 620 {
 621         ctx_t *ctx = ctx_get();
 622
 623         if (ctx)
 624                 snprintf(buf, bufsize, "%s::%s", ctx->info.plugin_name, name);
 625         else
 626                 snprintf(buf, bufsize, "core::%s", name);
 627         return buf;
 628 } /* plugin_get_name */
 629
 630 static int
 631 plugin_add_callback(sdb_llist_t **list, const char *type,
 632                 const char *name, void *callback, sdb_object_t *user_data)
 633 {
 634         sdb_object_t *obj;
 635
 636         if ((! name) || (! callback))
 637                 return -1;
 638
 639         assert(list);
 640
 641         if (! *list)
 642                 *list = sdb_llist_create();
 643         if (! *list)
 644                 return -1;
 645
 646         obj = sdb_object_create(name, callback_type,
 647                         list, type, callback, user_data);
 648         if (! obj)
 649                 return -1;
 650
 651         if (sdb_llist_append(*list, obj)) {
 652                 sdb_object_deref(obj);
 653                 return -1;
 654         }
 655
 656         /* pass control to the list */
 657         sdb_object_deref(obj);
 658
 659         sdb_log(SDB_LOG_INFO, "core: Registered %s callback '%s'.",
 660                         type, name);
 661         return 0;
 662 } /* plugin_add_callback */
 663
 664 /*
 665  * object meta-data
 666  */
 667
 668 typedef struct {
 669         int obj_type;
 670         sdb_time_t last_update;
 671         sdb_time_t interval;
 672 } interval_fetcher_t;
 673
 674 static int
 675 interval_fetcher_host(sdb_store_host_t *host, sdb_object_t *user_data)
 676 {
 677         interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data;
 678         lu->obj_type = SDB_HOST;
 679         lu->last_update = host->last_update;
 680         return 0;
 681 } /* interval_fetcher_host */
 682
 683 static int
 684 interval_fetcher_service(sdb_store_service_t *svc, sdb_object_t *user_data)
 685 {
 686         interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data;
 687         lu->obj_type = SDB_SERVICE;
 688         lu->last_update = svc->last_update;
 689         return 0;
 690 } /* interval_fetcher_service */
 691
 692 static int
 693 interval_fetcher_metric(sdb_store_metric_t *metric, sdb_object_t *user_data)
 694 {
 695         interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data;
 696         lu->obj_type = SDB_METRIC;
 697         lu->last_update = metric->last_update;
 698         return 0;
 699 } /* interval_fetcher_metric */
 700
 701 static int
 702 interval_fetcher_attr(sdb_store_attribute_t *attr, sdb_object_t *user_data)
 703 {
 704         interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data;
 705         lu->obj_type = SDB_ATTRIBUTE;
 706         lu->last_update = attr->last_update;
 707         return 0;
 708 } /* interval_fetcher_attr */
 709
 710 static sdb_store_writer_t interval_fetcher = {
 711         interval_fetcher_host, interval_fetcher_service,
 712         interval_fetcher_metric, interval_fetcher_attr,
 713 };
 714
 715 static int
 716 get_interval(int obj_type, const char *hostname,
 717                 int parent_type, const char *parent, const char *name,
 718                 sdb_time_t last_update, sdb_time_t *interval_out)
 719 {
 720         sdb_ast_fetch_t fetch = SDB_AST_FETCH_INIT;
 721         char hn[hostname ? strlen(hostname) + 1 : 1];
 722         char pn[parent ? strlen(parent) + 1 : 1];
 723         char n[strlen(name) + 1];
 724         int status;
 725
 726         interval_fetcher_t lu = { 0, 0, 0 };
 727         sdb_object_wrapper_t obj = SDB_OBJECT_WRAPPER_STATIC(&lu);
 728         sdb_time_t interval;
 729
 730         assert(name);
 731
 732         if (hostname)
 733                 strncpy(hn, hostname, sizeof(hn));
 734         if (parent)
 735                 strncpy(pn, parent, sizeof(pn));
 736         strncpy(n, name, sizeof(n));
 737
 738         fetch.obj_type = obj_type;
 739         fetch.hostname = hostname ? hn : NULL;
 740         fetch.parent_type = parent_type;
 741         fetch.parent = parent ? pn : NULL;
 742         fetch.name = n;
 743
 744         status = sdb_plugin_query(SDB_AST_NODE(&fetch),
 745                         &interval_fetcher, SDB_OBJ(&obj), NULL);
 746         if ((status < 0) || (lu.obj_type != obj_type) || (lu.last_update == 0)) {
 747                 *interval_out = 0;
 748                 return 0;
 749         }
 750
 751         if (lu.last_update >= last_update) {
 752                 if (lu.last_update > last_update)
 753                         sdb_log(SDB_LOG_DEBUG, "memstore: Cannot update %s '%s' - "
 754                                         "value too old (%"PRIsdbTIME" < %"PRIsdbTIME")",
 755                                         SDB_STORE_TYPE_TO_NAME(obj_type), name,
 756                                         lu.last_update, last_update);
 757                 *interval_out = lu.interval;
 758                 return 1;
 759         }
 760
 761         interval = last_update - lu.last_update;
 762         if (lu.interval && interval)
 763                 interval = (sdb_time_t)((0.9 * (double)lu.interval)
 764                                 + (0.1 * (double)interval));
 765         *interval_out = interval;
 766         return 0;
 767 } /* get_interval */
 768
 769 static void
 770 get_backend(char **backends, size_t *backends_num)
 771 {
 772         const sdb_plugin_info_t *info;
 773
 774         info = sdb_plugin_current();
 775         if ((! info) || (! info->plugin_name) || (! *info->plugin_name)) {
 776                 *backends_num = 0;
 777                 return;
 778         }
 779
 780         backends[0] = info->plugin_name;
 781         *backends_num = 1;
 782 } /* get_backend */
 783
 784 /*
 785  * public API
 786  */
 787
 788 int
 789 sdb_plugin_load(const char *basedir, const char *name,
 790                 const sdb_plugin_ctx_t *plugin_ctx)
 791 {
 792         ctx_t *ctx;
 793
 794         int status;
 795
 796         if ((! name) || (! *name))
 797                 return -1;
 798
 799         if (! all_plugins) {
 800                 if (! (all_plugins = sdb_llist_create())) {
 801                         sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s': "
 802                                         "internal error while creating linked list", name);
 803                         return -1;
 804                 }
 805         }
 806
 807         ctx = CTX(sdb_llist_search_by_name(all_plugins, name));
 808         if (ctx) {
 809                 /* plugin already loaded */
 810                 if (! ctx->use_cnt) {
 811                         /* reloading plugin */
 812                         ctx_t *old_ctx = ctx_set(ctx);
 813
 814                         status = module_init(ctx->info.plugin_name, ctx->handle, NULL);
 815                         if (status)
 816                                 return status;
 817
 818                         sdb_log(SDB_LOG_INFO, "core: Successfully reloaded plugin "
 819                                         "'%s' (%s)", ctx->info.plugin_name,
 820                                         INFO_GET(&ctx->info, description));
 821                         ctx_set(old_ctx);
 822                 }
 823                 ++ctx->use_cnt;
 824                 return 0;
 825         }
 826
 827         return module_load(basedir, name, plugin_ctx);
 828 } /* sdb_plugin_load */
 829
 830 int
 831 sdb_plugin_set_info(sdb_plugin_info_t *info, int type, ...)
 832 {
 833         va_list ap;
 834
 835         if (! info)
 836                 return -1;
 837
 838         va_start(ap, type);
 839
 840         switch (type) {
 841                 case SDB_PLUGIN_INFO_DESC:
 842                         {
 843                                 char *desc = va_arg(ap, char *);
 844                                 if (desc) {
 845                                         if (info->description)
 846                                                 free(info->description);
 847                                         info->description = strdup(desc);
 848                                 }
 849                         }
 850                         break;
 851                 case SDB_PLUGIN_INFO_COPYRIGHT:
 852                         {
 853                                 char *copyright = va_arg(ap, char *);
 854                                 if (copyright)
 855                                         info->copyright = strdup(copyright);
 856                         }
 857                         break;
 858                 case SDB_PLUGIN_INFO_LICENSE:
 859                         {
 860                                 char *license = va_arg(ap, char *);
 861                                 if (license) {
 862                                         if (info->license)
 863                                                 free(info->license);
 864                                         info->license = strdup(license);
 865                                 }
 866                         }
 867                         break;
 868                 case SDB_PLUGIN_INFO_VERSION:
 869                         {
 870                                 int version = va_arg(ap, int);
 871                                 info->version = version;
 872                         }
 873                         break;
 874                 case SDB_PLUGIN_INFO_PLUGIN_VERSION:
 875                         {
 876                                 int version = va_arg(ap, int);
 877                                 info->plugin_version = version;
 878                         }
 879                         break;
 880                 default:
 881                         va_end(ap);
 882                         return -1;
 883         }
 884
 885         va_end(ap);
 886         return 0;
 887 } /* sdb_plugin_set_info */
 888
 889 int
 890 sdb_plugin_register_config(sdb_plugin_config_cb callback)
 891 {
 892         ctx_t *ctx = ctx_get();
 893
 894         if (! ctx) {
 895                 sdb_log(SDB_LOG_ERR, "core: Invalid attempt to register a "
 896                                 "config callback from outside a plugin");
 897                 return -1;
 898         }
 899         return plugin_add_callback(&config_list, "config", ctx->info.plugin_name,
 900                         (void *)callback, NULL);
 901 } /* sdb_plugin_register_config */
 902
 903 int
 904 sdb_plugin_register_init(const char *name, sdb_plugin_init_cb callback,
 905                 sdb_object_t *user_data)
 906 {
 907         char cb_name[1024];
 908         return plugin_add_callback(&init_list, "init",
 909                         plugin_get_name(name, cb_name, sizeof(cb_name)),
 910                         (void *)callback, user_data);
 911 } /* sdb_plugin_register_init */
 912
 913 int
 914 sdb_plugin_register_shutdown(const char *name, sdb_plugin_shutdown_cb callback,
 915                 sdb_object_t *user_data)
 916 {
 917         char cb_name[1024];
 918         return plugin_add_callback(&shutdown_list, "shutdown",
 919                         plugin_get_name(name, cb_name, sizeof(cb_name)),
 920                         (void *)callback, user_data);
 921 } /* sdb_plugin_register_shutdown */
 922
 923 int
 924 sdb_plugin_register_log(const char *name, sdb_plugin_log_cb callback,
 925                 sdb_object_t *user_data)
 926 {
 927         char cb_name[1024];
 928         return plugin_add_callback(&log_list, "log",
 929                         plugin_get_name(name, cb_name, sizeof(cb_name)),
 930                         callback, user_data);
 931 } /* sdb_plugin_register_log */
 932
 933 int
 934 sdb_plugin_register_cname(const char *name, sdb_plugin_cname_cb callback,
 935                 sdb_object_t *user_data)
 936 {
 937         char cb_name[1024];
 938         return plugin_add_callback(&cname_list, "cname",
 939                         plugin_get_name(name, cb_name, sizeof(cb_name)),
 940                         callback, user_data);
 941 } /* sdb_plugin_register_cname */
 942
 943 int
 944 sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback,
 945                 const sdb_time_t *interval, sdb_object_t *user_data)
 946 {
 947         char cb_name[1024];
 948         sdb_object_t *obj;
 949
 950         if ((! name) || (! callback))
 951                 return -1;
 952
 953         if (! collector_list)
 954                 collector_list = sdb_llist_create();
 955         if (! collector_list)
 956                 return -1;
 957
 958         plugin_get_name(name, cb_name, sizeof(cb_name));
 959
 960         obj = sdb_object_create(cb_name, collector_type,
 961                         &collector_list, "collector", callback, user_data);
 962         if (! obj)
 963                 return -1;
 964
 965         if (interval)
 966                 CCB(obj)->ccb_interval = *interval;
 967         else {
 968                 ctx_t *ctx = ctx_get();
 969
 970                 if (! ctx) {
 971                         sdb_log(SDB_LOG_ERR, "core: Cannot determine interval "
 972                                         "for collector %s; none specified and no plugin "
 973                                         "context found", cb_name);
 974                         return -1;
 975                 }
 976
 977                 CCB(obj)->ccb_interval = ctx->public.interval;
 978         }
 979
 980         if (! (CCB(obj)->ccb_next_update = sdb_gettime())) {
 981                 char errbuf[1024];
 982                 sdb_log(SDB_LOG_ERR, "core: Failed to determine current "
 983                                 "time: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
 984                 sdb_object_deref(obj);
 985                 return -1;
 986         }
 987
 988         if (sdb_llist_insert_sorted(collector_list, obj,
 989                                 plugin_cmp_next_update)) {
 990                 sdb_object_deref(obj);
 991                 return -1;
 992         }
 993
 994         /* pass control to the list */
 995         sdb_object_deref(obj);
 996
 997         sdb_log(SDB_LOG_INFO, "core: Registered collector callback '%s' "
 998                         "(interval = %.3fs).", cb_name,
 999                         SDB_TIME_TO_DOUBLE(CCB(obj)->ccb_interval));
1000         return 0;
1001 } /* sdb_plugin_register_collector */
1002
1003 int
1004 sdb_plugin_register_ts_fetcher(const char *name,
1005                 sdb_plugin_fetch_ts_cb callback, sdb_object_t *user_data)
1006 {
1007         return plugin_add_callback(&ts_fetcher_list, "time-series fetcher",
1008                         name, callback, user_data);
1009 } /* sdb_plugin_register_ts_fetcher */
1010
1011 int
1012 sdb_plugin_register_writer(const char *name,
1013                 sdb_store_writer_t *writer, sdb_object_t *user_data)
1014 {
1015         char cb_name[1024];
1016         sdb_object_t *obj;
1017
1018         if ((! name) || (! writer))
1019                 return -1;
1020
1021         if (! writer_list)
1022                 writer_list = sdb_llist_create();
1023         if (! writer_list)
1024                 return -1;
1025
1026         plugin_get_name(name, cb_name, sizeof(cb_name));
1027
1028         obj = sdb_object_create(cb_name, writer_type,
1029                         writer, user_data);
1030         if (! obj)
1031                 return -1;
1032
1033         if (sdb_llist_append(writer_list, obj)) {
1034                 sdb_object_deref(obj);
1035                 return -1;
1036         }
1037
1038         /* pass control to the list */
1039         sdb_object_deref(obj);
1040
1041         sdb_log(SDB_LOG_INFO, "core: Registered store writer callback '%s'.",
1042                         cb_name);
1043         return 0;
1044 } /* sdb_store_register_writer */
1045
1046 int
1047 sdb_plugin_register_reader(const char *name,
1048                 sdb_store_reader_t *reader, sdb_object_t *user_data)
1049 {
1050         char cb_name[1024];
1051         sdb_object_t *obj;
1052
1053         if ((! name) || (! reader))
1054                 return -1;
1055
1056         if (! reader_list)
1057                 reader_list = sdb_llist_create();
1058         if (! reader_list)
1059                 return -1;
1060
1061         plugin_get_name(name, cb_name, sizeof(cb_name));
1062
1063         obj = sdb_object_create(cb_name, reader_type,
1064                         reader, user_data);
1065         if (! obj)
1066                 return -1;
1067
1068         if (sdb_llist_append(reader_list, obj)) {
1069                 sdb_object_deref(obj);
1070                 return -1;
1071         }
1072
1073         /* pass control to the list */
1074         sdb_object_deref(obj);
1075
1076         sdb_log(SDB_LOG_INFO, "core: Registered store reader callback '%s'.",
1077                         cb_name);
1078         return 0;
1079 } /* sdb_plugin_register_reader */
1080
1081 void
1082 sdb_plugin_unregister_all(void)
1083 {
1084         size_t i;
1085
1086         for (i = 0; i < SDB_STATIC_ARRAY_LEN(all_lists); ++i) {
1087                 const char  *type =  all_lists[i].type;
1088                 sdb_llist_t *list = *all_lists[i].list;
1089
1090                 size_t len = sdb_llist_len(list);
1091
1092                 if (! len)
1093                         continue;
1094
1095                 sdb_llist_clear(list);
1096                 sdb_log(SDB_LOG_INFO, "core: Unregistered %zu %s callback%s",
1097                                 len, type, len == 1 ? "" : "s");
1098         }
1099 } /* sdb_plugin_unregister_all */
1100
1101 sdb_plugin_ctx_t
1102 sdb_plugin_get_ctx(void)
1103 {
1104         ctx_t *c;
1105
1106         c = ctx_get();
1107         if (! c) {
1108                 sdb_plugin_log(SDB_LOG_ERR, "core: Invalid read access to plugin "
1109                                 "context outside a plugin");
1110                 return plugin_default_ctx;
1111         }
1112         return c->public;
1113 } /* sdb_plugin_get_ctx */
1114
1115 int
1116 sdb_plugin_set_ctx(sdb_plugin_ctx_t ctx, sdb_plugin_ctx_t *old)
1117 {
1118         ctx_t *c;
1119
1120         c = ctx_get();
1121         if (! c) {
1122                 sdb_plugin_log(SDB_LOG_ERR, "core: Invalid write access to plugin "
1123                                 "context outside a plugin");
1124                 return -1;
1125         }
1126
1127         if (old)
1128                 *old = c->public;
1129         c->public = ctx;
1130         return 0;
1131 } /* sdb_plugin_set_ctx */
1132
1133 const sdb_plugin_info_t *
1134 sdb_plugin_current(void)
1135 {
1136         ctx_t *ctx = ctx_get();
1137
1138         if (! ctx)
1139                 return NULL;
1140         return &ctx->info;
1141 } /* sdb_plugin_current */
1142
1143 int
1144 sdb_plugin_configure(const char *name, oconfig_item_t *ci)
1145 {
1146         callback_t *plugin;
1147         sdb_plugin_config_cb callback;
1148
1149         ctx_t *old_ctx;
1150
1151         int status;
1152
1153         if ((! name) || (! ci))
1154                 return -1;
1155
1156         plugin = CB(sdb_llist_search_by_name(config_list, name));
1157         if (! plugin) {
1158                 ctx_t *ctx = CTX(sdb_llist_search_by_name(all_plugins, name));
1159                 if (! ctx)
1160                         sdb_log(SDB_LOG_ERR, "core: Cannot configure unknown "
1161                                         "plugin '%s'. Missing 'LoadPlugin \"%s\"'?",
1162                                         name, name);
1163                 else
1164                         sdb_log(SDB_LOG_ERR, "core: Plugin '%s' did not register "
1165                                         "a config callback.", name);
1166                 errno = ENOENT;
1167                 return -1;
1168         }
1169
1170         old_ctx = ctx_set(plugin->cb_ctx);
1171         callback = (sdb_plugin_config_cb)plugin->cb_callback;
1172         status = callback(ci);
1173         ctx_set(old_ctx);
1174         return status;
1175 } /* sdb_plugin_configure */
1176
1177 int
1178 sdb_plugin_reconfigure_init(void)
1179 {
1180         sdb_llist_iter_t *iter;
1181
1182         iter = sdb_llist_get_iter(config_list);
1183         if (config_list && (! iter))
1184                 return -1;
1185
1186         /* deconfigure all plugins */
1187         while (sdb_llist_iter_has_next(iter)) {
1188                 callback_t *plugin;
1189                 sdb_plugin_config_cb callback;
1190                 ctx_t *old_ctx;
1191
1192                 plugin = CB(sdb_llist_iter_get_next(iter));
1193                 old_ctx = ctx_set(plugin->cb_ctx);
1194                 callback = (sdb_plugin_config_cb)plugin->cb_callback;
1195                 callback(NULL);
1196                 ctx_set(old_ctx);
1197         }
1198         sdb_llist_iter_destroy(iter);
1199
1200         iter = sdb_llist_get_iter(all_plugins);
1201         if (all_plugins && (! iter))
1202                 return -1;
1203
1204         /* record all plugins as being unused */
1205         while (sdb_llist_iter_has_next(iter))
1206                 CTX(sdb_llist_iter_get_next(iter))->use_cnt = 0;
1207         sdb_llist_iter_destroy(iter);
1208
1209         sdb_plugin_unregister_all();
1210         return 0;
1211 } /* sdb_plugin_reconfigure_init */
1212
1213 int
1214 sdb_plugin_reconfigure_finish(void)
1215 {
1216         sdb_llist_iter_t *iter;
1217
1218         iter = sdb_llist_get_iter(all_plugins);
1219         if (all_plugins && (! iter))
1220                 return -1;
1221
1222         while (sdb_llist_iter_has_next(iter)) {
1223                 ctx_t *ctx = CTX(sdb_llist_iter_get_next(iter));
1224                 if (ctx->use_cnt)
1225                         continue;
1226
1227                 sdb_log(SDB_LOG_INFO, "core: Module %s no longer in use",
1228                                 ctx->info.plugin_name);
1229                 sdb_llist_iter_remove_current(iter);
1230                 plugin_unregister_by_name(ctx->info.plugin_name);
1231                 sdb_object_deref(SDB_OBJ(ctx));
1232         }
1233         sdb_llist_iter_destroy(iter);
1234         return 0;
1235 } /* sdb_plugin_reconfigure_finish */
1236
1237 int
1238 sdb_plugin_init_all(void)
1239 {
1240         sdb_llist_iter_t *iter;
1241         int ret = 0;
1242
1243         iter = sdb_llist_get_iter(init_list);
1244         while (sdb_llist_iter_has_next(iter)) {
1245                 callback_t *cb;
1246                 sdb_plugin_init_cb callback;
1247                 ctx_t *old_ctx;
1248
1249                 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
1250                 assert(obj);
1251                 cb = CB(obj);
1252
1253                 callback = (sdb_plugin_init_cb)cb->cb_callback;
1254
1255                 old_ctx = ctx_set(cb->cb_ctx);
1256                 if (callback(cb->cb_user_data)) {
1257                         sdb_log(SDB_LOG_ERR, "core: Failed to initialize plugin "
1258                                         "'%s'. Unregistering all callbacks.", obj->name);
1259                         ctx_set(old_ctx);
1260                         plugin_unregister_by_name(cb->cb_ctx->info.plugin_name);
1261                         ++ret;
1262                 }
1263                 else
1264                         ctx_set(old_ctx);
1265         }
1266         sdb_llist_iter_destroy(iter);
1267         return ret;
1268 } /* sdb_plugin_init_all */
1269
1270 int
1271 sdb_plugin_shutdown_all(void)
1272 {
1273         sdb_llist_iter_t *iter;
1274         int ret = 0;
1275
1276         iter = sdb_llist_get_iter(shutdown_list);
1277         while (sdb_llist_iter_has_next(iter)) {
1278                 callback_t *cb;
1279                 sdb_plugin_shutdown_cb callback;
1280                 ctx_t *old_ctx;
1281
1282                 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
1283                 assert(obj);
1284                 cb = CB(obj);
1285
1286                 callback = (sdb_plugin_shutdown_cb)cb->cb_callback;
1287
1288                 old_ctx = ctx_set(cb->cb_ctx);
1289                 if (callback(cb->cb_user_data)) {
1290                         sdb_log(SDB_LOG_ERR, "core: Failed to shutdown plugin '%s'.",
1291                                         obj->name);
1292                         ++ret;
1293                 }
1294                 ctx_set(old_ctx);
1295         }
1296         sdb_llist_iter_destroy(iter);
1297         return ret;
1298 } /* sdb_plugin_shutdown_all */
1299
1300 int
1301 sdb_plugin_collector_loop(sdb_plugin_loop_t *loop)
1302 {
1303         if (! collector_list) {
1304                 sdb_log(SDB_LOG_WARNING, "core: No collectors registered. "
1305                                 "Quiting main loop.");
1306                 return -1;
1307         }
1308
1309         if (! loop)
1310                 return -1;
1311
1312         while (loop->do_loop) {
1313                 sdb_plugin_collector_cb callback;
1314                 ctx_t *old_ctx;
1315
1316                 sdb_time_t interval, now;
1317
1318                 sdb_object_t *obj = sdb_llist_shift(collector_list);
1319                 if (! obj)
1320                         return -1;
1321
1322                 callback = (sdb_plugin_collector_cb)CCB(obj)->ccb_callback;
1323
1324                 if (! (now = sdb_gettime())) {
1325                         char errbuf[1024];
1326                         sdb_log(SDB_LOG_ERR, "core: Failed to determine current "
1327                                         "time in collector main loop: %s",
1328                                         sdb_strerror(errno, errbuf, sizeof(errbuf)));
1329                         now = CCB(obj)->ccb_next_update;
1330                 }
1331
1332                 if (now < CCB(obj)->ccb_next_update) {
1333                         interval = CCB(obj)->ccb_next_update - now;
1334
1335                         errno = 0;
1336                         while (loop->do_loop && sdb_sleep(interval, &interval)) {
1337                                 if (errno != EINTR) {
1338                                         char errbuf[1024];
1339                                         sdb_log(SDB_LOG_ERR, "core: Failed to sleep "
1340                                                         "in collector main loop: %s",
1341                                                         sdb_strerror(errno, errbuf, sizeof(errbuf)));
1342                                         sdb_llist_insert_sorted(collector_list, obj,
1343                                                         plugin_cmp_next_update);
1344                                         sdb_object_deref(obj);
1345                                         return -1;
1346                                 }
1347                                 errno = 0;
1348                         }
1349
1350                         if (! loop->do_loop) {
1351                                 /* put back; don't worry about errors */
1352                                 sdb_llist_insert_sorted(collector_list, obj,
1353                                                 plugin_cmp_next_update);
1354                                 sdb_object_deref(obj);
1355                                 return 0;
1356                         }
1357                 }
1358
1359                 old_ctx = ctx_set(CCB(obj)->ccb_ctx);
1360                 if (callback(CCB(obj)->ccb_user_data)) {
1361                         /* XXX */
1362                 }
1363                 ctx_set(old_ctx);
1364
1365                 interval = CCB(obj)->ccb_interval;
1366                 if (! interval)
1367                         interval = loop->default_interval;
1368                 if (! interval) {
1369                         sdb_log(SDB_LOG_WARNING, "core: No interval configured "
1370                                         "for plugin '%s'; skipping any further "
1371                                         "iterations.", obj->name);
1372                         sdb_object_deref(obj);
1373                         continue;
1374                 }
1375
1376                 CCB(obj)->ccb_next_update += interval;
1377
1378                 if (! (now = sdb_gettime())) {
1379                         char errbuf[1024];
1380                         sdb_log(SDB_LOG_ERR, "core: Failed to determine current "
1381                                         "time in collector main loop: %s",
1382                                         sdb_strerror(errno, errbuf, sizeof(errbuf)));
1383                         now = CCB(obj)->ccb_next_update;
1384                 }
1385
1386                 if (now > CCB(obj)->ccb_next_update) {
1387                         sdb_log(SDB_LOG_WARNING, "core: Plugin '%s' took too "
1388                                         "long; skipping iterations to keep up.",
1389                                         obj->name);
1390                         CCB(obj)->ccb_next_update = now;
1391                 }
1392
1393                 if (sdb_llist_insert_sorted(collector_list, obj,
1394                                         plugin_cmp_next_update)) {
1395                         sdb_log(SDB_LOG_ERR, "core: Failed to re-insert "
1396                                         "plugin '%s' into collector list. Unable to further "
1397                                         "use the plugin.",
1398                                         obj->name);
1399                         sdb_object_deref(obj);
1400                         return -1;
1401                 }
1402
1403                 /* pass control back to the list */
1404                 sdb_object_deref(obj);
1405         }
1406         return 0;
1407 } /* sdb_plugin_read_loop */
1408
1409 char *
1410 sdb_plugin_cname(char *hostname)
1411 {
1412         sdb_llist_iter_t *iter;
1413
1414         if (! hostname)
1415                 return NULL;
1416
1417         if (! cname_list)
1418                 return hostname;
1419
1420         iter = sdb_llist_get_iter(cname_list);
1421         while (sdb_llist_iter_has_next(iter)) {
1422                 sdb_plugin_cname_cb callback;
1423                 char *cname;
1424
1425                 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
1426                 assert(obj);
1427
1428                 callback = (sdb_plugin_cname_cb)CB(obj)->cb_callback;
1429                 cname = callback(hostname, CB(obj)->cb_user_data);
1430                 if (cname) {
1431                         free(hostname);
1432                         hostname = cname;
1433                 }
1434                 /* else: don't change hostname */
1435         }
1436         sdb_llist_iter_destroy(iter);
1437         return hostname;
1438 } /* sdb_plugin_cname */
1439
1440 int
1441 sdb_plugin_log(int prio, const char *msg)
1442 {
1443         sdb_llist_iter_t *iter;
1444         int ret = -1;
1445
1446         bool logged = 0;
1447
1448         if (! msg)
1449                 return 0;
1450
1451         iter = sdb_llist_get_iter(log_list);
1452         while (sdb_llist_iter_has_next(iter)) {
1453                 sdb_plugin_log_cb callback;
1454                 int tmp;
1455
1456                 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
1457                 assert(obj);
1458
1459                 callback = (sdb_plugin_log_cb)CB(obj)->cb_callback;
1460                 tmp = callback(prio, msg, CB(obj)->cb_user_data);
1461                 if (tmp > ret)
1462                         ret = tmp;
1463
1464                 if (CB(obj)->cb_ctx)
1465                         logged = 1;
1466                 /* else: this is an internally registered callback */
1467         }
1468         sdb_llist_iter_destroy(iter);
1469
1470         if (! logged)
1471                 return fprintf(stderr, "[%s] %s\n", SDB_LOG_PRIO_TO_STRING(prio), msg);
1472         return ret;
1473 } /* sdb_plugin_log */
1474
1475 int
1476 sdb_plugin_vlogf(int prio, const char *fmt, va_list ap)
1477 {
1478         sdb_strbuf_t *buf;
1479         int ret;
1480
1481         if (! fmt)
1482                 return 0;
1483
1484         buf = sdb_strbuf_create(64);
1485         if (! buf) {
1486                 ret = fprintf(stderr, "[%s] ", SDB_LOG_PRIO_TO_STRING(prio));
1487                 ret += vfprintf(stderr, fmt, ap);
1488                 return ret;
1489         }
1490
1491         if (sdb_strbuf_vsprintf(buf, fmt, ap) < 0) {
1492                 sdb_strbuf_destroy(buf);
1493                 return -1;
1494         }
1495
1496         ret = sdb_plugin_log(prio, sdb_strbuf_string(buf));
1497         sdb_strbuf_destroy(buf);
1498         return ret;
1499 } /* sdb_plugin_vlogf */
1500
1501 int
1502 sdb_plugin_logf(int prio, const char *fmt, ...)
1503 {
1504         va_list ap;
1505         int ret;
1506
1507         if (! fmt)
1508                 return 0;
1509
1510         va_start(ap, fmt);
1511         ret = sdb_plugin_vlogf(prio, fmt, ap);
1512         va_end(ap);
1513         return ret;
1514 } /* sdb_plugin_logf */
1515
1516 sdb_timeseries_t *
1517 sdb_plugin_fetch_timeseries(const char *type, const char *id,
1518                 sdb_timeseries_opts_t *opts)
1519 {
1520         callback_t *plugin;
1521         sdb_plugin_fetch_ts_cb callback;
1522         sdb_timeseries_t *ts;
1523
1524         ctx_t *old_ctx;
1525
1526         if ((! type) || (! id) || (! opts))
1527                 return NULL;
1528
1529         plugin = CB(sdb_llist_search_by_name(ts_fetcher_list, type));
1530         if (! plugin) {
1531                 sdb_log(SDB_LOG_ERR, "core: Cannot fetch time-series of type %s: "
1532                                 "no such plugin loaded", type);
1533                 errno = ENOENT;
1534                 return NULL;
1535         }
1536
1537         old_ctx = ctx_set(plugin->cb_ctx);
1538         callback = (sdb_plugin_fetch_ts_cb)plugin->cb_callback;
1539         ts = callback(id, opts, plugin->cb_user_data);
1540         ctx_set(old_ctx);
1541         return ts;
1542 } /* sdb_plugin_fetch_timeseries */
1543
1544 int
1545 sdb_plugin_query(sdb_ast_node_t *ast,
1546                 sdb_store_writer_t *w, sdb_object_t *wd, sdb_strbuf_t *errbuf)
1547 {
1548         size_t n = sdb_llist_len(reader_list);
1549         reader_t *reader;
1550         sdb_object_t *q;
1551         int status = 0;
1552
1553         if (! ast)
1554                 return 0;
1555
1556         if ((ast->type != SDB_AST_TYPE_FETCH)
1557                         && (ast->type != SDB_AST_TYPE_LIST)
1558                         && (ast->type != SDB_AST_TYPE_LOOKUP)) {
1559                 sdb_log(SDB_LOG_ERR, "core: Cannot execute query of type %s",
1560                                 SDB_AST_TYPE_TO_STRING(ast));
1561                 sdb_strbuf_sprintf(errbuf, "Cannot execute query of type %s",
1562                                 SDB_AST_TYPE_TO_STRING(ast));
1563                 return -1;
1564         }
1565
1566         if (n != 1) {
1567                 char *msg = (n > 0)
1568                         ? "Cannot execute query: multiple readers not supported"
1569                         : "Cannot execute query: no readers registered";
1570                 sdb_strbuf_sprintf(errbuf, "%s", msg);
1571                 sdb_log(SDB_LOG_ERR, "core: %s", msg);
1572                 return -1;
1573         }
1574
1575         reader = READER(sdb_llist_get(reader_list, 0));
1576         assert(reader);
1577
1578         q = reader->impl.prepare_query(ast, errbuf, reader->r_user_data);
1579         if (q)
1580                 status = reader->impl.execute_query(q, w, SDB_OBJ(wd),
1581                                 errbuf, reader->r_user_data);
1582         else
1583                 status = -1;
1584
1585         sdb_object_deref(SDB_OBJ(q));
1586         sdb_object_deref(SDB_OBJ(reader));
1587         return status;
1588 } /* sdb_plugin_query */
1589
1590 int
1591 sdb_plugin_store_host(const char *name, sdb_time_t last_update)
1592 {
1593         sdb_store_host_t host = SDB_STORE_HOST_INIT;
1594         char *backends[1];
1595         char *cname;
1596
1597         sdb_llist_iter_t *iter;
1598         int status = 0;
1599
1600         if (! name)
1601                 return -1;
1602
1603         if (! sdb_llist_len(writer_list)) {
1604                 sdb_log(SDB_LOG_ERR, "core: Cannot store host: "
1605                                 "no writers registered");
1606                 return -1;
1607         }
1608
1609         cname = sdb_plugin_cname(strdup(name));
1610         if (! cname) {
1611                 sdb_log(SDB_LOG_ERR, "core: strdup failed");
1612                 return -1;
1613         }
1614
1615         host.name = cname;
1616         host.last_update = last_update ? last_update : sdb_gettime();
1617         if (get_interval(SDB_HOST, NULL, -1, NULL, cname,
1618                                 host.last_update, &host.interval)) {
1619                 free(cname);
1620                 return 1;
1621         }
1622         host.backends = (const char * const *)backends;
1623         get_backend(backends, &host.backends_num);
1624
1625         iter = sdb_llist_get_iter(writer_list);
1626         while (sdb_llist_iter_has_next(iter)) {
1627                 writer_t *writer = WRITER(sdb_llist_iter_get_next(iter));
1628                 int s;
1629                 assert(writer);
1630                 s = writer->impl.store_host(&host, writer->w_user_data);
1631                 if (((s > 0) && (status >= 0)) || (s < 0))
1632                         status = s;
1633         }
1634         sdb_llist_iter_destroy(iter);
1635         free(cname);
1636         return status;
1637 } /* sdb_plugin_store_host */
1638
1639 int
1640 sdb_plugin_store_service(const char *hostname, const char *name,
1641                 sdb_time_t last_update)
1642 {
1643         sdb_store_service_t service = SDB_STORE_SERVICE_INIT;
1644         char *backends[1];
1645         char *cname;
1646
1647         sdb_llist_iter_t *iter;
1648         sdb_data_t d;
1649
1650         int status = 0;
1651
1652         if ((! hostname) || (! name))
1653                 return -1;
1654
1655         if (! sdb_llist_len(writer_list)) {
1656                 sdb_log(SDB_LOG_ERR, "core: Cannot store service: "
1657                                 "no writers registered");
1658                 return -1;
1659         }
1660
1661         cname = sdb_plugin_cname(strdup(hostname));
1662         if (! cname) {
1663                 sdb_log(SDB_LOG_ERR, "core: strdup failed");
1664                 return -1;
1665         }
1666
1667         service.hostname = cname;
1668         service.name = name;
1669         service.last_update = last_update ? last_update : sdb_gettime();
1670         if (get_interval(SDB_SERVICE, cname, -1, NULL, name,
1671                                 service.last_update, &service.interval)) {
1672                 free(cname);
1673                 return 1;
1674         }
1675         service.backends = (const char * const *)backends;
1676         get_backend(backends, &service.backends_num);
1677
1678         iter = sdb_llist_get_iter(writer_list);
1679         while (sdb_llist_iter_has_next(iter)) {
1680                 writer_t *writer = WRITER(sdb_llist_iter_get_next(iter));
1681                 int s;
1682                 assert(writer);
1683                 s = writer->impl.store_service(&service, writer->w_user_data);
1684                 if (((s > 0) && (status >= 0)) || (s < 0))
1685                         status = s;
1686         }
1687         sdb_llist_iter_destroy(iter);
1688
1689         if (! status) {
1690                 /* record the hostname as an attribute */
1691                 d.type = SDB_TYPE_STRING;
1692                 d.data.string = cname;
1693                 if (sdb_plugin_store_service_attribute(cname, name,
1694                                         "hostname", &d, service.last_update))
1695                         status = -1;
1696         }
1697
1698         free(cname);
1699         return status;
1700 } /* sdb_plugin_store_service */
1701
1702 int
1703 sdb_plugin_store_metric(const char *hostname, const char *name,
1704                 sdb_metric_store_t *store, sdb_time_t last_update)
1705 {
1706         sdb_store_metric_t metric = SDB_STORE_METRIC_INIT;
1707         char *backends[1];
1708         char *cname;
1709
1710         sdb_llist_iter_t *iter;
1711         sdb_data_t d;
1712
1713         int status = 0;
1714
1715         if ((! hostname) || (! name))
1716                 return -1;
1717
1718         if (! sdb_llist_len(writer_list)) {
1719                 sdb_log(SDB_LOG_ERR, "core: Cannot store metric: "
1720                                 "no writers registered");
1721                 return -1;
1722         }
1723
1724         cname = sdb_plugin_cname(strdup(hostname));
1725         if (! cname) {
1726                 sdb_log(SDB_LOG_ERR, "core: strdup failed");
1727                 return -1;
1728         }
1729
1730         if (store && ((! store->type) || (! store->id)))
1731                 store = NULL;
1732
1733         metric.hostname = cname;
1734         metric.name = name;
1735         if (store) {
1736                 metric.store.type = store->type;
1737                 metric.store.id = store->id;
1738         }
1739         metric.last_update = last_update ? last_update : sdb_gettime();
1740         if (get_interval(SDB_METRIC, cname, -1, NULL, name,
1741                                 metric.last_update, &metric.interval)) {
1742                 free(cname);
1743                 return 1;
1744         }
1745         metric.backends = (const char * const *)backends;
1746         get_backend(backends, &metric.backends_num);
1747
1748         iter = sdb_llist_get_iter(writer_list);
1749         while (sdb_llist_iter_has_next(iter)) {
1750                 writer_t *writer = WRITER(sdb_llist_iter_get_next(iter));
1751                 int s;
1752                 assert(writer);
1753                 s = writer->impl.store_metric(&metric, writer->w_user_data);
1754                 if (((s > 0) && (status >= 0)) || (s < 0))
1755                         status = s;
1756         }
1757         sdb_llist_iter_destroy(iter);
1758
1759         if (! status) {
1760                 /* record the hostname as an attribute */
1761                 d.type = SDB_TYPE_STRING;
1762                 d.data.string = cname;
1763                 if (sdb_plugin_store_metric_attribute(cname, name,
1764                                         "hostname", &d, metric.last_update))
1765                         status = -1;
1766         }
1767
1768         free(cname);
1769         return status;
1770 } /* sdb_plugin_store_metric */
1771
1772 int
1773 sdb_plugin_store_attribute(const char *hostname, const char *key,
1774                 const sdb_data_t *value, sdb_time_t last_update)
1775 {
1776         sdb_store_attribute_t attr = SDB_STORE_ATTRIBUTE_INIT;
1777         char *backends[1];
1778         char *cname;
1779
1780         sdb_llist_iter_t *iter;
1781         int status = 0;
1782
1783         if ((! hostname) || (! key) || (! value))
1784                 return -1;
1785
1786         if (! sdb_llist_len(writer_list)) {
1787                 sdb_log(SDB_LOG_ERR, "core: Cannot store attribute: "
1788                                 "no writers registered");
1789                 return -1;
1790         }
1791
1792         cname = sdb_plugin_cname(strdup(hostname));
1793         if (! cname) {
1794                 sdb_log(SDB_LOG_ERR, "core: strdup failed");
1795                 return -1;
1796         }
1797
1798         attr.parent_type = SDB_HOST;
1799         attr.parent = cname;
1800         attr.key = key;
1801         attr.value = *value;
1802         attr.last_update = last_update ? last_update : sdb_gettime();
1803         if (get_interval(SDB_ATTRIBUTE, cname, -1, NULL, key,
1804                                 attr.last_update, &attr.interval)) {
1805                 free(cname);
1806                 return 1;
1807         }
1808         attr.backends = (const char * const *)backends;
1809         get_backend(backends, &attr.backends_num);
1810
1811         iter = sdb_llist_get_iter(writer_list);
1812         while (sdb_llist_iter_has_next(iter)) {
1813                 writer_t *writer = WRITER(sdb_llist_iter_get_next(iter));
1814                 int s;
1815                 assert(writer);
1816                 s = writer->impl.store_attribute(&attr, writer->w_user_data);
1817                 if (((s > 0) && (status >= 0)) || (s < 0))
1818                         status = s;
1819         }
1820         sdb_llist_iter_destroy(iter);
1821         free(cname);
1822         return status;
1823 } /* sdb_plugin_store_attribute */
1824
1825 int
1826 sdb_plugin_store_service_attribute(const char *hostname, const char *service,
1827                 const char *key, const sdb_data_t *value, sdb_time_t last_update)
1828 {
1829         sdb_store_attribute_t attr = SDB_STORE_ATTRIBUTE_INIT;
1830         char *backends[1];
1831         char *cname;
1832
1833         sdb_llist_iter_t *iter;
1834         int status = 0;
1835
1836         if ((! hostname) || (! service) || (! key) || (! value))
1837                 return -1;
1838
1839         if (! sdb_llist_len(writer_list)) {
1840                 sdb_log(SDB_LOG_ERR, "core: Cannot store service attribute: "
1841                                 "no writers registered");
1842                 return -1;
1843         }
1844
1845         cname = sdb_plugin_cname(strdup(hostname));
1846         if (! cname) {
1847                 sdb_log(SDB_LOG_ERR, "core: strdup failed");
1848                 return -1;
1849         }
1850
1851         attr.hostname = cname;
1852         attr.parent_type = SDB_SERVICE;
1853         attr.parent = service;
1854         attr.key = key;
1855         attr.value = *value;
1856         attr.last_update = last_update ? last_update : sdb_gettime();
1857         if (get_interval(SDB_ATTRIBUTE, cname, SDB_SERVICE, service, key,
1858                                 attr.last_update, &attr.interval)) {
1859                 free(cname);
1860                 return 1;
1861         }
1862         attr.backends = (const char * const *)backends;
1863         get_backend(backends, &attr.backends_num);
1864
1865         iter = sdb_llist_get_iter(writer_list);
1866         while (sdb_llist_iter_has_next(iter)) {
1867                 writer_t *writer = WRITER(sdb_llist_iter_get_next(iter));
1868                 int s;
1869                 assert(writer);
1870                 s = writer->impl.store_attribute(&attr, writer->w_user_data);
1871                 if (((s > 0) && (status >= 0)) || (s < 0))
1872                         status = s;
1873         }
1874         sdb_llist_iter_destroy(iter);
1875         free(cname);
1876         return status;
1877 } /* sdb_plugin_store_service_attribute */
1878
1879 int
1880 sdb_plugin_store_metric_attribute(const char *hostname, const char *metric,
1881                 const char *key, const sdb_data_t *value, sdb_time_t last_update)
1882 {
1883         sdb_store_attribute_t attr = SDB_STORE_ATTRIBUTE_INIT;
1884         char *backends[1];
1885         char *cname;
1886
1887         sdb_llist_iter_t *iter;
1888         int status = 0;
1889
1890         if ((! hostname) || (! metric) || (! key) || (! value))
1891                 return -1;
1892
1893         if (! sdb_llist_len(writer_list)) {
1894                 sdb_log(SDB_LOG_ERR, "core: Cannot store metric attribute: "
1895                                 "no writers registered");
1896                 return -1;
1897         }
1898
1899         cname = sdb_plugin_cname(strdup(hostname));
1900         if (! cname) {
1901                 sdb_log(SDB_LOG_ERR, "core: strdup failed");
1902                 return -1;
1903         }
1904
1905         attr.hostname = cname;
1906         attr.parent_type = SDB_METRIC;
1907         attr.parent = metric;
1908         attr.key = key;
1909         attr.value = *value;
1910         attr.last_update = last_update ? last_update : sdb_gettime();
1911         if (get_interval(SDB_ATTRIBUTE, cname, SDB_METRIC, metric, key,
1912                                 attr.last_update, &attr.interval)) {
1913                 free(cname);
1914                 return 1;
1915         }
1916         attr.backends = (const char * const *)backends;
1917         get_backend(backends, &attr.backends_num);
1918
1919         iter = sdb_llist_get_iter(writer_list);
1920         while (sdb_llist_iter_has_next(iter)) {
1921                 writer_t *writer = WRITER(sdb_llist_iter_get_next(iter));
1922                 int s;
1923                 assert(writer);
1924                 s = writer->impl.store_attribute(&attr, writer->w_user_data);
1925                 if (((s > 0) && (status >= 0)) || (s < 0))
1926                         status = s;
1927         }
1928         sdb_llist_iter_destroy(iter);
1929         free(cname);
1930         return status;
1931 } /* sdb_plugin_store_metric_attribute */
1932
1933 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */
1934