src/core/plugin.c

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