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