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