X-Git-Url: https://git.tokkee.org/?p=sysdb.git;a=blobdiff_plain;f=src%2Fcore%2Fplugin.c;h=b3f00bd655504075b97c5f158b8c45a8c0d20d05;hp=897769926584b201fab1657ecd0885719486cce4;hb=dcf7802ad0958d4752b2c13f70e2fdf345d3509f;hpb=0c9e3aa9ba3948e88a0241861b75a002085eeccd diff --git a/src/core/plugin.c b/src/core/plugin.c index 8977699..b3f00bd 100644 --- a/src/core/plugin.c +++ b/src/core/plugin.c @@ -114,6 +114,14 @@ typedef struct { } reader_t; #define READER(obj) ((reader_t *)(obj)) +typedef struct { + callback_t super; /* cb_callback will always be NULL */ +#define ts_user_data super.cb_user_data +#define ts_ctx super.cb_ctx + sdb_timeseries_fetcher_t impl; +} ts_fetcher_t; +#define TS_FETCHER(obj) ((ts_fetcher_t *)(obj)) + /* * private variables */ @@ -133,7 +141,7 @@ static sdb_llist_t *collector_list = NULL; static sdb_llist_t *cname_list = NULL; static sdb_llist_t *shutdown_list = NULL; static sdb_llist_t *log_list = NULL; -static sdb_llist_t *ts_fetcher_list = NULL; +static sdb_llist_t *timeseries_fetcher_list = NULL; static sdb_llist_t *writer_list = NULL; static sdb_llist_t *reader_list = NULL; @@ -147,7 +155,7 @@ static struct { { "cname", &cname_list }, { "shutdown", &shutdown_list }, { "log", &log_list }, - { "timeseries fetcher", &ts_fetcher_list }, + { "timeseries fetcher", ×eries_fetcher_list }, { "store writer", &writer_list }, { "store reader", &reader_list }, }; @@ -242,9 +250,8 @@ plugin_unregister_by_name(const char *plugin_name) assert(cb->cb_ctx); - sdb_log(SDB_LOG_INFO, "core: Unregistering " - "%s callback '%s' (module %s)", type, cb->super.name, - cb->cb_ctx->info.plugin_name); + sdb_log(SDB_LOG_INFO, "Unregistering %s callback '%s' (module %s)", + type, cb->super.name, cb->cb_ctx->info.plugin_name); sdb_object_deref(SDB_OBJ(cb)); } } @@ -259,6 +266,81 @@ plugin_unregister_by_name(const char *plugin_name) /* else: other callbacks still reference it */ } /* plugin_unregister_by_name */ +/* + * store writer wrapper for performing database queries: + * It wraps another store writer, adding extra logic as needed. + */ + +typedef struct { + sdb_object_t super; + sdb_store_writer_t *w; + sdb_object_t *ud; + sdb_query_opts_t opts; +} query_writer_t; +#define QUERY_WRITER_INIT(w, ud) { \ + SDB_OBJECT_INIT, \ + (w), (ud), \ + SDB_DEFAULT_QUERY_OPTS \ +} +#define QUERY_WRITER(obj) ((query_writer_t *)(obj)) + +static int +query_store_host(sdb_store_host_t *host, sdb_object_t *user_data) +{ + query_writer_t *qw = QUERY_WRITER(user_data); + return qw->w->store_host(host, qw->ud); +} /* query_store_host */ + +static int +query_store_service(sdb_store_service_t *service, sdb_object_t *user_data) +{ + query_writer_t *qw = QUERY_WRITER(user_data); + return qw->w->store_service(service, qw->ud); +} /* query_store_service */ + +static int +query_store_metric(sdb_store_metric_t *metric, sdb_object_t *user_data) +{ + query_writer_t *qw = QUERY_WRITER(user_data); + sdb_timeseries_info_t *infos[metric->stores_num]; + sdb_metric_store_t stores[metric->stores_num]; + + const sdb_metric_store_t *orig_stores = metric->stores; + int status; + size_t i; + + if (! qw->opts.describe_timeseries) + /* nothing further to do */ + return qw->w->store_metric(metric, qw->ud); + + for (i = 0; i < metric->stores_num; i++) { + sdb_metric_store_t *s = stores + i; + *s = metric->stores[i]; + infos[i] = sdb_plugin_describe_timeseries(s->type, s->id); + s->info = infos[i]; + } + + metric->stores = stores; + status = qw->w->store_metric(metric, qw->ud); + metric->stores = orig_stores; + + for (i = 0; i < metric->stores_num; i++) + sdb_timeseries_info_destroy(infos[i]); + return status; +} /* query_store_metric */ + +static int +query_store_attribute(sdb_store_attribute_t *attr, sdb_object_t *user_data) +{ + query_writer_t *qw = QUERY_WRITER(user_data); + return qw->w->store_attribute(attr, qw->ud); +} /* query_store_attribute */ + +static sdb_store_writer_t query_writer = { + query_store_host, query_store_service, + query_store_metric, query_store_attribute, +}; + /* * private types */ @@ -285,13 +367,12 @@ ctx_destroy(sdb_object_t *obj) if (ctx->handle) { const char *err; - sdb_log(SDB_LOG_INFO, "core: Unloading module %s", - ctx->info.plugin_name); + sdb_log(SDB_LOG_INFO, "Unloading module %s", ctx->info.plugin_name); lt_dlerror(); lt_dlclose(ctx->handle); if ((err = lt_dlerror())) - sdb_log(SDB_LOG_WARNING, "core: Failed to unload module %s: %s", + sdb_log(SDB_LOG_WARNING, "Failed to unload module %s: %s", ctx->info.plugin_name, err); } @@ -345,24 +426,39 @@ ctx_create(const char *name) return ctx; } /* ctx_create */ -static int -plugin_cb_init(sdb_object_t *obj, va_list ap) +/* + * plugin_init_ok: + * Checks whether the registration of a new plugin identified by 'obj' is + * okay. It consumes the first two arguments of 'ap'. + */ +static bool +plugin_init_ok(sdb_object_t *obj, va_list ap) { sdb_llist_t **list = va_arg(ap, sdb_llist_t **); - const char *type = va_arg(ap, const char *); - void *callback = va_arg(ap, void *); - sdb_object_t *ud = va_arg(ap, sdb_object_t *); + const char *type = va_arg(ap, const char *); - assert(list); - assert(type); - assert(obj); + assert(list); assert(type); if (sdb_llist_search_by_name(*list, obj->name)) { - sdb_log(SDB_LOG_WARNING, "core: %s callback '%s' " - "has already been registered. Ignoring newly " - "registered version.", type, obj->name); - return -1; + sdb_log(SDB_LOG_WARNING, "%s callback '%s' has already been " + "registered. Ignoring newly registered version.", + type, obj->name); + return 0; } + return 1; +} /* plugin_init_ok */ + +static int +plugin_cb_init(sdb_object_t *obj, va_list ap) +{ + void *callback; + sdb_object_t *ud; + + if (! plugin_init_ok(obj, ap)) + return -1; + + callback = va_arg(ap, void *); + ud = va_arg(ap, sdb_object_t *); /* cb_ctx may be NULL if the plugin was not registered by a plugin */ @@ -400,22 +496,20 @@ static sdb_type_t collector_type = { static int plugin_writer_init(sdb_object_t *obj, va_list ap) { - sdb_store_writer_t *impl = va_arg(ap, sdb_store_writer_t *); - sdb_object_t *ud = va_arg(ap, sdb_object_t *); + sdb_store_writer_t *impl; + sdb_object_t *ud; + + if (! plugin_init_ok(obj, ap)) + return -1; + impl = va_arg(ap, sdb_store_writer_t *); + ud = va_arg(ap, sdb_object_t *); assert(impl); if ((! impl->store_host) || (! impl->store_service) || (! impl->store_metric) || (! impl->store_attribute)) { - sdb_log(SDB_LOG_ERR, "core: store writer callback '%s' " - "does not fully implement the writer interface.", - obj->name); - return -1; - } - if (sdb_llist_search_by_name(writer_list, obj->name)) { - sdb_log(SDB_LOG_WARNING, "core: store writer callback '%s' " - "has already been registered. Ignoring newly " - "registered version.", obj->name); + sdb_log(SDB_LOG_ERR, "store writer callback '%s' does not fully " + "implement the writer interface.", obj->name); return -1; } @@ -448,21 +542,19 @@ static sdb_type_t writer_type = { static int plugin_reader_init(sdb_object_t *obj, va_list ap) { - sdb_store_reader_t *impl = va_arg(ap, sdb_store_reader_t *); - sdb_object_t *ud = va_arg(ap, sdb_object_t *); + sdb_store_reader_t *impl; + sdb_object_t *ud; + if (! plugin_init_ok(obj, ap)) + return -1; + + impl = va_arg(ap, sdb_store_reader_t *); + ud = va_arg(ap, sdb_object_t *); assert(impl); if ((! impl->prepare_query) || (! impl->execute_query)) { - sdb_log(SDB_LOG_ERR, "core: store reader callback '%s' " - "does not fully implement the reader interface.", - obj->name); - return -1; - } - if (sdb_llist_search_by_name(reader_list, obj->name)) { - sdb_log(SDB_LOG_WARNING, "core: store reader callback '%s' " - "has already been registered. Ignoring newly " - "registered version.", obj->name); + sdb_log(SDB_LOG_ERR, "store reader callback '%s' does not fully " + "implement the reader interface.", obj->name); return -1; } @@ -492,6 +584,51 @@ static sdb_type_t reader_type = { plugin_reader_destroy }; +static int +plugin_ts_fetcher_init(sdb_object_t *obj, va_list ap) +{ + sdb_timeseries_fetcher_t *impl; + sdb_object_t *ud; + + if (! plugin_init_ok(obj, ap)) + return -1; + + impl = va_arg(ap, sdb_timeseries_fetcher_t *); + ud = va_arg(ap, sdb_object_t *); + assert(impl); + + if ((! impl->describe) || (! impl->fetch)) { + sdb_log(SDB_LOG_ERR, "timeseries fetcher callback '%s' does not fully " + "implement the interface.", obj->name); + return -1; + } + + /* ctx may be NULL if the callback was not registered by a plugin */ + + TS_FETCHER(obj)->impl = *impl; + TS_FETCHER(obj)->ts_ctx = ctx_get(); + sdb_object_ref(SDB_OBJ(TS_FETCHER(obj)->ts_ctx)); + + sdb_object_ref(ud); + TS_FETCHER(obj)->ts_user_data = ud; + return 0; +} /* plugin_ts_fetcher_init */ + +static void +plugin_ts_fetcher_destroy(sdb_object_t *obj) +{ + assert(obj); + sdb_object_deref(TS_FETCHER(obj)->ts_user_data); + sdb_object_deref(SDB_OBJ(TS_FETCHER(obj)->ts_ctx)); +} /* plugin_ts_fetcher_destroy */ + +static sdb_type_t ts_fetcher_type = { + sizeof(ts_fetcher_t), + + plugin_ts_fetcher_init, + plugin_ts_fetcher_destroy +}; + static int module_init(const char *name, lt_dlhandle lh, sdb_plugin_info_t *info) { @@ -500,15 +637,14 @@ module_init(const char *name, lt_dlhandle lh, sdb_plugin_info_t *info) mod_init = (int (*)(sdb_plugin_info_t *))lt_dlsym(lh, "sdb_module_init"); if (! mod_init) { - sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s': " + sdb_log(SDB_LOG_ERR, "Failed to load plugin '%s': " "could not find symbol 'sdb_module_init'", name); return -1; } status = mod_init(info); if (status) { - sdb_log(SDB_LOG_ERR, "core: Failed to initialize " - "module '%s'", name); + sdb_log(SDB_LOG_ERR, "Failed to initialize module '%s'", name); plugin_unregister_by_name(name); return -1; } @@ -550,7 +686,7 @@ module_load(const char *basedir, const char *name, if (access(filename, R_OK)) { char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s' (%s): %s", + sdb_log(SDB_LOG_ERR, "Failed to load plugin '%s' (%s): %s", name, filename, sdb_strerror(errno, errbuf, sizeof(errbuf))); return -1; } @@ -560,18 +696,18 @@ module_load(const char *basedir, const char *name, lh = lt_dlopen(filename); if (! lh) { - sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s': %s" + sdb_log(SDB_LOG_ERR, "Failed to load plugin '%s': %s" "The most common cause for this problem are missing " "dependencies.\n", name, lt_dlerror()); return -1; } if (ctx_get()) - sdb_log(SDB_LOG_WARNING, "core: Discarding old plugin context"); + sdb_log(SDB_LOG_WARNING, "Discarding old plugin context"); ctx = ctx_create(name); if (! ctx) { - sdb_log(SDB_LOG_ERR, "core: Failed to initialize plugin context"); + sdb_log(SDB_LOG_ERR, "Failed to initialize plugin context"); return -1; } @@ -584,26 +720,26 @@ module_load(const char *basedir, const char *name, if ((status = module_init(name, lh, &ctx->info))) { sdb_object_deref(SDB_OBJ(ctx)); + ctx_set(NULL); return status; } /* compare minor version */ if ((ctx->info.version < 0) || ((int)(ctx->info.version / 100) != (int)(SDB_VERSION / 100))) - sdb_log(SDB_LOG_WARNING, "core: WARNING: version of " - "plugin '%s' (%i.%i.%i) does not match our version " - "(%i.%i.%i); this might cause problems", - name, SDB_VERSION_DECODE(ctx->info.version), + sdb_log(SDB_LOG_WARNING, "WARNING: plugin version (%i.%i.%i) " + "does not match core version (%i.%i.%i); " + "this might cause problems", + SDB_VERSION_DECODE(ctx->info.version), SDB_VERSION_DECODE(SDB_VERSION)); sdb_llist_append(all_plugins, SDB_OBJ(ctx)); - sdb_log(SDB_LOG_INFO, "core: Successfully loaded " - "plugin %s v%i (%s)", ctx->info.plugin_name, - ctx->info.plugin_version, + /* log messages will be prefixed by the plugin name */ + sdb_log(SDB_LOG_INFO, "Successfully loaded " + "version %i (%s)", ctx->info.plugin_version, INFO_GET(&ctx->info, description)); - sdb_log(SDB_LOG_INFO, "core: Plugin %s: %s, License: %s", - ctx->info.plugin_name, + sdb_log(SDB_LOG_INFO, "%s, License: %s", INFO_GET(&ctx->info, copyright), INFO_GET(&ctx->info, license)); @@ -628,12 +764,12 @@ plugin_get_name(const char *name, char *buf, size_t bufsize) } /* plugin_get_name */ static int -plugin_add_callback(sdb_llist_t **list, const char *type, - const char *name, void *callback, sdb_object_t *user_data) +plugin_add_impl(sdb_llist_t **list, sdb_type_t T, const char *type, + const char *name, void *impl, sdb_object_t *user_data) { sdb_object_t *obj; - if ((! name) || (! callback)) + if ((! name) || (! impl)) return -1; assert(list); @@ -643,8 +779,7 @@ plugin_add_callback(sdb_llist_t **list, const char *type, if (! *list) return -1; - obj = sdb_object_create(name, callback_type, - list, type, callback, user_data); + obj = sdb_object_create(name, T, list, type, impl, user_data); if (! obj) return -1; @@ -656,10 +791,114 @@ plugin_add_callback(sdb_llist_t **list, const char *type, /* pass control to the list */ sdb_object_deref(obj); - sdb_log(SDB_LOG_INFO, "core: Registered %s callback '%s'.", - type, name); + sdb_log(SDB_LOG_INFO, "Registered %s callback '%s'.", type, name); + return 0; +} /* plugin_add_impl */ + +/* + * object meta-data + */ + +typedef struct { + int obj_type; + sdb_time_t last_update; + sdb_time_t interval; +} interval_fetcher_t; + +static int +interval_fetcher_host(sdb_store_host_t *host, sdb_object_t *user_data) +{ + interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data; + lu->obj_type = SDB_HOST; + lu->last_update = host->last_update; + return 0; +} /* interval_fetcher_host */ + +static int +interval_fetcher_service(sdb_store_service_t *svc, sdb_object_t *user_data) +{ + interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data; + lu->obj_type = SDB_SERVICE; + lu->last_update = svc->last_update; + return 0; +} /* interval_fetcher_service */ + +static int +interval_fetcher_metric(sdb_store_metric_t *metric, sdb_object_t *user_data) +{ + interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data; + lu->obj_type = SDB_METRIC; + lu->last_update = metric->last_update; + return 0; +} /* interval_fetcher_metric */ + +static int +interval_fetcher_attr(sdb_store_attribute_t *attr, sdb_object_t *user_data) +{ + interval_fetcher_t *lu = SDB_OBJ_WRAPPER(user_data)->data; + lu->obj_type = SDB_ATTRIBUTE; + lu->last_update = attr->last_update; + return 0; +} /* interval_fetcher_attr */ + +static sdb_store_writer_t interval_fetcher = { + interval_fetcher_host, interval_fetcher_service, + interval_fetcher_metric, interval_fetcher_attr, +}; + +static int +get_interval(int obj_type, const char *hostname, + int parent_type, const char *parent, const char *name, + sdb_time_t last_update, sdb_time_t *interval_out) +{ + sdb_ast_fetch_t fetch = SDB_AST_FETCH_INIT; + char hn[hostname ? strlen(hostname) + 1 : 1]; + char pn[parent ? strlen(parent) + 1 : 1]; + char n[strlen(name) + 1]; + int status; + + interval_fetcher_t lu = { 0, 0, 0 }; + sdb_object_wrapper_t obj = SDB_OBJECT_WRAPPER_STATIC(&lu); + sdb_time_t interval; + + assert(name); + + if (hostname) + strncpy(hn, hostname, sizeof(hn)); + if (parent) + strncpy(pn, parent, sizeof(pn)); + strncpy(n, name, sizeof(n)); + + fetch.obj_type = obj_type; + fetch.hostname = hostname ? hn : NULL; + fetch.parent_type = parent_type; + fetch.parent = parent ? pn : NULL; + fetch.name = n; + + status = sdb_plugin_query(SDB_AST_NODE(&fetch), + &interval_fetcher, SDB_OBJ(&obj), NULL, NULL); + if ((status < 0) || (lu.obj_type != obj_type) || (lu.last_update == 0)) { + *interval_out = 0; + return 0; + } + + if (lu.last_update >= last_update) { + if (lu.last_update > last_update) + sdb_log(SDB_LOG_DEBUG, "Cannot update %s '%s' - " + "value too old (%"PRIsdbTIME" < %"PRIsdbTIME")", + SDB_STORE_TYPE_TO_NAME(obj_type), name, + lu.last_update, last_update); + *interval_out = lu.interval; + return 1; + } + + interval = last_update - lu.last_update; + if (lu.interval && interval) + interval = (sdb_time_t)((0.9 * (double)lu.interval) + + (0.1 * (double)interval)); + *interval_out = interval; return 0; -} /* plugin_add_callback */ +} /* get_interval */ static void get_backend(char **backends, size_t *backends_num) @@ -693,7 +932,7 @@ sdb_plugin_load(const char *basedir, const char *name, if (! all_plugins) { if (! (all_plugins = sdb_llist_create())) { - sdb_log(SDB_LOG_ERR, "core: Failed to load plugin '%s': " + sdb_log(SDB_LOG_ERR, "Failed to load plugin '%s': " "internal error while creating linked list", name); return -1; } @@ -707,13 +946,14 @@ sdb_plugin_load(const char *basedir, const char *name, ctx_t *old_ctx = ctx_set(ctx); status = module_init(ctx->info.plugin_name, ctx->handle, NULL); + ctx_set(old_ctx); if (status) return status; - sdb_log(SDB_LOG_INFO, "core: Successfully reloaded plugin " + ctx_set(old_ctx); + sdb_log(SDB_LOG_INFO, "Successfully reloaded plugin " "'%s' (%s)", ctx->info.plugin_name, INFO_GET(&ctx->info, description)); - ctx_set(old_ctx); } ++ctx->use_cnt; return 0; @@ -787,12 +1027,12 @@ sdb_plugin_register_config(sdb_plugin_config_cb callback) ctx_t *ctx = ctx_get(); if (! ctx) { - sdb_log(SDB_LOG_ERR, "core: Invalid attempt to register a " + sdb_log(SDB_LOG_ERR, "Invalid attempt to register a " "config callback from outside a plugin"); return -1; } - return plugin_add_callback(&config_list, "config", ctx->info.plugin_name, - (void *)callback, NULL); + return plugin_add_impl(&config_list, callback_type, "config", + ctx->info.plugin_name, (void *)callback, NULL); } /* sdb_plugin_register_config */ int @@ -800,7 +1040,7 @@ sdb_plugin_register_init(const char *name, sdb_plugin_init_cb callback, sdb_object_t *user_data) { char cb_name[1024]; - return plugin_add_callback(&init_list, "init", + return plugin_add_impl(&init_list, callback_type, "init", plugin_get_name(name, cb_name, sizeof(cb_name)), (void *)callback, user_data); } /* sdb_plugin_register_init */ @@ -810,7 +1050,7 @@ sdb_plugin_register_shutdown(const char *name, sdb_plugin_shutdown_cb callback, sdb_object_t *user_data) { char cb_name[1024]; - return plugin_add_callback(&shutdown_list, "shutdown", + return plugin_add_impl(&shutdown_list, callback_type, "shutdown", plugin_get_name(name, cb_name, sizeof(cb_name)), (void *)callback, user_data); } /* sdb_plugin_register_shutdown */ @@ -820,7 +1060,7 @@ sdb_plugin_register_log(const char *name, sdb_plugin_log_cb callback, sdb_object_t *user_data) { char cb_name[1024]; - return plugin_add_callback(&log_list, "log", + return plugin_add_impl(&log_list, callback_type, "log", plugin_get_name(name, cb_name, sizeof(cb_name)), callback, user_data); } /* sdb_plugin_register_log */ @@ -830,7 +1070,7 @@ sdb_plugin_register_cname(const char *name, sdb_plugin_cname_cb callback, sdb_object_t *user_data) { char cb_name[1024]; - return plugin_add_callback(&cname_list, "cname", + return plugin_add_impl(&cname_list, callback_type, "cname", plugin_get_name(name, cb_name, sizeof(cb_name)), callback, user_data); } /* sdb_plugin_register_cname */ @@ -863,9 +1103,8 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback ctx_t *ctx = ctx_get(); if (! ctx) { - sdb_log(SDB_LOG_ERR, "core: Cannot determine interval " - "for collector %s; none specified and no plugin " - "context found", cb_name); + sdb_log(SDB_LOG_ERR, "Cannot determine interval for collector %s; " + "none specified and no plugin context found", cb_name); return -1; } @@ -874,8 +1113,8 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback if (! (CCB(obj)->ccb_next_update = sdb_gettime())) { char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "core: Failed to determine current " - "time: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); + sdb_log(SDB_LOG_ERR, "Failed to determine current time: %s", + sdb_strerror(errno, errbuf, sizeof(errbuf))); sdb_object_deref(obj); return -1; } @@ -889,53 +1128,28 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback /* pass control to the list */ sdb_object_deref(obj); - sdb_log(SDB_LOG_INFO, "core: Registered collector callback '%s' " + sdb_log(SDB_LOG_INFO, "Registered collector callback '%s' " "(interval = %.3fs).", cb_name, SDB_TIME_TO_DOUBLE(CCB(obj)->ccb_interval)); return 0; } /* sdb_plugin_register_collector */ int -sdb_plugin_register_ts_fetcher(const char *name, - sdb_plugin_fetch_ts_cb callback, sdb_object_t *user_data) +sdb_plugin_register_timeseries_fetcher(const char *name, + sdb_timeseries_fetcher_t *fetcher, sdb_object_t *user_data) { - return plugin_add_callback(&ts_fetcher_list, "time-series fetcher", - name, callback, user_data); -} /* sdb_plugin_register_ts_fetcher */ + return plugin_add_impl(×eries_fetcher_list, ts_fetcher_type, "time-series fetcher", + name, fetcher, user_data); +} /* sdb_plugin_register_timeseries_fetcher */ int sdb_plugin_register_writer(const char *name, sdb_store_writer_t *writer, sdb_object_t *user_data) { char cb_name[1024]; - sdb_object_t *obj; - - if ((! name) || (! writer)) - return -1; - - if (! writer_list) - writer_list = sdb_llist_create(); - if (! writer_list) - return -1; - - plugin_get_name(name, cb_name, sizeof(cb_name)); - - obj = sdb_object_create(cb_name, writer_type, + return plugin_add_impl(&writer_list, writer_type, "store writer", + plugin_get_name(name, cb_name, sizeof(cb_name)), writer, user_data); - if (! obj) - return -1; - - if (sdb_llist_append(writer_list, obj)) { - sdb_object_deref(obj); - return -1; - } - - /* pass control to the list */ - sdb_object_deref(obj); - - sdb_log(SDB_LOG_INFO, "core: Registered store writer callback '%s'.", - cb_name); - return 0; } /* sdb_store_register_writer */ int @@ -943,34 +1157,9 @@ sdb_plugin_register_reader(const char *name, sdb_store_reader_t *reader, sdb_object_t *user_data) { char cb_name[1024]; - sdb_object_t *obj; - - if ((! name) || (! reader)) - return -1; - - if (! reader_list) - reader_list = sdb_llist_create(); - if (! reader_list) - return -1; - - plugin_get_name(name, cb_name, sizeof(cb_name)); - - obj = sdb_object_create(cb_name, reader_type, + return plugin_add_impl(&reader_list, reader_type, "store reader", + plugin_get_name(name, cb_name, sizeof(cb_name)), reader, user_data); - if (! obj) - return -1; - - if (sdb_llist_append(reader_list, obj)) { - sdb_object_deref(obj); - return -1; - } - - /* pass control to the list */ - sdb_object_deref(obj); - - sdb_log(SDB_LOG_INFO, "core: Registered store reader callback '%s'.", - cb_name); - return 0; } /* sdb_plugin_register_reader */ void @@ -988,7 +1177,7 @@ sdb_plugin_unregister_all(void) continue; sdb_llist_clear(list); - sdb_log(SDB_LOG_INFO, "core: Unregistered %zu %s callback%s", + sdb_log(SDB_LOG_INFO, "Unregistered %zu %s callback%s", len, type, len == 1 ? "" : "s"); } } /* sdb_plugin_unregister_all */ @@ -1000,7 +1189,7 @@ sdb_plugin_get_ctx(void) c = ctx_get(); if (! c) { - sdb_plugin_log(SDB_LOG_ERR, "core: Invalid read access to plugin " + sdb_log(SDB_LOG_ERR, "Invalid read access to plugin " "context outside a plugin"); return plugin_default_ctx; } @@ -1014,7 +1203,7 @@ sdb_plugin_set_ctx(sdb_plugin_ctx_t ctx, sdb_plugin_ctx_t *old) c = ctx_get(); if (! c) { - sdb_plugin_log(SDB_LOG_ERR, "core: Invalid write access to plugin " + sdb_log(SDB_LOG_ERR, "Invalid write access to plugin " "context outside a plugin"); return -1; } @@ -1052,11 +1241,11 @@ sdb_plugin_configure(const char *name, oconfig_item_t *ci) if (! plugin) { ctx_t *ctx = CTX(sdb_llist_search_by_name(all_plugins, name)); if (! ctx) - sdb_log(SDB_LOG_ERR, "core: Cannot configure unknown " - "plugin '%s'. Missing 'LoadPlugin \"%s\"'?", + sdb_log(SDB_LOG_ERR, "Cannot configure unknown plugin '%s'. " + "Missing 'LoadPlugin \"%s\"'?", name, name); else - sdb_log(SDB_LOG_ERR, "core: Plugin '%s' did not register " + sdb_log(SDB_LOG_ERR, "Plugin '%s' did not register " "a config callback.", name); errno = ENOENT; return -1; @@ -1119,7 +1308,7 @@ sdb_plugin_reconfigure_finish(void) if (ctx->use_cnt) continue; - sdb_log(SDB_LOG_INFO, "core: Module %s no longer in use", + sdb_log(SDB_LOG_INFO, "Module %s no longer in use", ctx->info.plugin_name); sdb_llist_iter_remove_current(iter); plugin_unregister_by_name(ctx->info.plugin_name); @@ -1149,8 +1338,8 @@ sdb_plugin_init_all(void) old_ctx = ctx_set(cb->cb_ctx); if (callback(cb->cb_user_data)) { - sdb_log(SDB_LOG_ERR, "core: Failed to initialize plugin " - "'%s'. Unregistering all callbacks.", obj->name); + sdb_log(SDB_LOG_ERR, "Failed to initialize plugin '%s'. " + "Unregistering all callbacks.", obj->name); ctx_set(old_ctx); plugin_unregister_by_name(cb->cb_ctx->info.plugin_name); ++ret; @@ -1182,7 +1371,7 @@ sdb_plugin_shutdown_all(void) old_ctx = ctx_set(cb->cb_ctx); if (callback(cb->cb_user_data)) { - sdb_log(SDB_LOG_ERR, "core: Failed to shutdown plugin '%s'.", + sdb_log(SDB_LOG_ERR, "Failed to shutdown plugin '%s'.", obj->name); ++ret; } @@ -1196,7 +1385,7 @@ int sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) { if (! collector_list) { - sdb_log(SDB_LOG_WARNING, "core: No collectors registered. " + sdb_log(SDB_LOG_WARNING, "No collectors registered. " "Quiting main loop."); return -1; } @@ -1218,7 +1407,7 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) if (! (now = sdb_gettime())) { char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "core: Failed to determine current " + sdb_log(SDB_LOG_ERR, "Failed to determine current " "time in collector main loop: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); now = CCB(obj)->ccb_next_update; @@ -1231,7 +1420,7 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) while (loop->do_loop && sdb_sleep(interval, &interval)) { if (errno != EINTR) { char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "core: Failed to sleep " + sdb_log(SDB_LOG_ERR, "Failed to sleep " "in collector main loop: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); sdb_llist_insert_sorted(collector_list, obj, @@ -1261,7 +1450,7 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) if (! interval) interval = loop->default_interval; if (! interval) { - sdb_log(SDB_LOG_WARNING, "core: No interval configured " + sdb_log(SDB_LOG_WARNING, "No interval configured " "for plugin '%s'; skipping any further " "iterations.", obj->name); sdb_object_deref(obj); @@ -1272,14 +1461,14 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) if (! (now = sdb_gettime())) { char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "core: Failed to determine current " + sdb_log(SDB_LOG_ERR, "Failed to determine current " "time in collector main loop: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); now = CCB(obj)->ccb_next_update; } if (now > CCB(obj)->ccb_next_update) { - sdb_log(SDB_LOG_WARNING, "core: Plugin '%s' took too " + sdb_log(SDB_LOG_WARNING, "Plugin '%s' took too " "long; skipping iterations to keep up.", obj->name); CCB(obj)->ccb_next_update = now; @@ -1287,9 +1476,8 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) if (sdb_llist_insert_sorted(collector_list, obj, plugin_cmp_next_update)) { - sdb_log(SDB_LOG_ERR, "core: Failed to re-insert " - "plugin '%s' into collector list. Unable to further " - "use the plugin.", + sdb_log(SDB_LOG_ERR, "Failed to re-insert plugin '%s' into " + "collector list. Unable to further use the plugin.", obj->name); sdb_object_deref(obj); return -1; @@ -1335,35 +1523,50 @@ sdb_plugin_cname(char *hostname) int sdb_plugin_log(int prio, const char *msg) { - sdb_llist_iter_t *iter; + const sdb_plugin_info_t *info; int ret = -1; - bool logged = 0; - if (! msg) return 0; - iter = sdb_llist_get_iter(log_list); - while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_log_cb callback; - int tmp; + info = sdb_plugin_current(); - sdb_object_t *obj = sdb_llist_iter_get_next(iter); - assert(obj); + { + const char *plugin_name = info ? info->plugin_name : ""; + char log_msg[strlen(msg) + strlen(plugin_name) + strlen(" plugin: ") + 1]; - callback = (sdb_plugin_log_cb)CB(obj)->cb_callback; - tmp = callback(prio, msg, CB(obj)->cb_user_data); - if (tmp > ret) - ret = tmp; + sdb_llist_iter_t *iter; + bool logged = 0; - if (CB(obj)->cb_ctx) - logged = 1; - /* else: this is an internally registered callback */ - } - sdb_llist_iter_destroy(iter); + if (*plugin_name) + snprintf(log_msg, sizeof(log_msg), "%s plugin: %s", + plugin_name, msg); + else + strncpy(log_msg, msg, sizeof(log_msg)); + + iter = sdb_llist_get_iter(log_list); + while (sdb_llist_iter_has_next(iter)) { + sdb_plugin_log_cb callback; + int tmp; - if (! logged) - return fprintf(stderr, "[%s] %s\n", SDB_LOG_PRIO_TO_STRING(prio), msg); + sdb_object_t *obj = sdb_llist_iter_get_next(iter); + assert(obj); + + callback = (sdb_plugin_log_cb)CB(obj)->cb_callback; + tmp = callback(prio, msg, CB(obj)->cb_user_data); + if (tmp > ret) + ret = tmp; + + if (CB(obj)->cb_ctx) + logged = 1; + /* else: this is an internally registered callback */ + } + sdb_llist_iter_destroy(iter); + + if (! logged) + return fprintf(stderr, "[%s] %s\n", + SDB_LOG_PRIO_TO_STRING(prio), log_msg); + } return ret; } /* sdb_plugin_log */ @@ -1412,8 +1615,7 @@ sdb_timeseries_t * sdb_plugin_fetch_timeseries(const char *type, const char *id, sdb_timeseries_opts_t *opts) { - callback_t *plugin; - sdb_plugin_fetch_ts_cb callback; + ts_fetcher_t *fetcher; sdb_timeseries_t *ts; ctx_t *old_ctx; @@ -1421,37 +1623,67 @@ sdb_plugin_fetch_timeseries(const char *type, const char *id, if ((! type) || (! id) || (! opts)) return NULL; - plugin = CB(sdb_llist_search_by_name(ts_fetcher_list, type)); - if (! plugin) { - sdb_log(SDB_LOG_ERR, "core: Cannot fetch time-series of type %s: " + fetcher = TS_FETCHER(sdb_llist_search_by_name(timeseries_fetcher_list, type)); + if (! fetcher) { + sdb_log(SDB_LOG_ERR, "Cannot fetch time-series of type %s: " "no such plugin loaded", type); errno = ENOENT; return NULL; } - old_ctx = ctx_set(plugin->cb_ctx); - callback = (sdb_plugin_fetch_ts_cb)plugin->cb_callback; - ts = callback(id, opts, plugin->cb_user_data); + old_ctx = ctx_set(fetcher->ts_ctx); + ts = fetcher->impl.fetch(id, opts, fetcher->ts_user_data); ctx_set(old_ctx); return ts; } /* sdb_plugin_fetch_timeseries */ +sdb_timeseries_info_t * +sdb_plugin_describe_timeseries(const char *type, const char *id) +{ + ts_fetcher_t *fetcher; + sdb_timeseries_info_t *ts_info; + + ctx_t *old_ctx; + + if ((! type) || (! id)) + return NULL; + + fetcher = TS_FETCHER(sdb_llist_search_by_name(timeseries_fetcher_list, type)); + if (! fetcher) { + sdb_log(SDB_LOG_ERR, "Cannot describe time-series of type %s: " + "no such plugin loaded", type); + errno = ENOENT; + return NULL; + } + + old_ctx = ctx_set(fetcher->ts_ctx); + ts_info = fetcher->impl.describe(id, fetcher->ts_user_data); + ctx_set(old_ctx); + return ts_info; +} /* sdb_plugin_describe_timeseries */ + int -sdb_plugin_query(sdb_ast_node_t *ast, sdb_strbuf_t *buf, sdb_strbuf_t *errbuf) +sdb_plugin_query(sdb_ast_node_t *ast, + sdb_store_writer_t *w, sdb_object_t *wd, + sdb_query_opts_t *opts, sdb_strbuf_t *errbuf) { - size_t n = sdb_llist_len(reader_list); + query_writer_t qw = QUERY_WRITER_INIT(w, wd); reader_t *reader; sdb_object_t *q; + + size_t n = sdb_llist_len(reader_list); int status = 0; if (! ast) return 0; + if (opts) + qw.opts = *opts; + if ((ast->type != SDB_AST_TYPE_FETCH) && (ast->type != SDB_AST_TYPE_LIST) - && (ast->type != SDB_AST_TYPE_LOOKUP) - && (ast->type != SDB_AST_TYPE_TIMESERIES)) { - sdb_log(SDB_LOG_ERR, "core: Cannot execute query of type %s", + && (ast->type != SDB_AST_TYPE_LOOKUP)) { + sdb_log(SDB_LOG_ERR, "Cannot execute query of type %s", SDB_AST_TYPE_TO_STRING(ast)); sdb_strbuf_sprintf(errbuf, "Cannot execute query of type %s", SDB_AST_TYPE_TO_STRING(ast)); @@ -1463,7 +1695,7 @@ sdb_plugin_query(sdb_ast_node_t *ast, sdb_strbuf_t *buf, sdb_strbuf_t *errbuf) ? "Cannot execute query: multiple readers not supported" : "Cannot execute query: no readers registered"; sdb_strbuf_sprintf(errbuf, "%s", msg); - sdb_log(SDB_LOG_ERR, "core: %s", msg); + sdb_log(SDB_LOG_ERR, "%s", msg); return -1; } @@ -1472,7 +1704,8 @@ sdb_plugin_query(sdb_ast_node_t *ast, sdb_strbuf_t *buf, sdb_strbuf_t *errbuf) q = reader->impl.prepare_query(ast, errbuf, reader->r_user_data); if (q) - status = reader->impl.execute_query(q, buf, errbuf, reader->r_user_data); + status = reader->impl.execute_query(q, &query_writer, SDB_OBJ(&qw), + errbuf, reader->r_user_data); else status = -1; @@ -1484,7 +1717,7 @@ sdb_plugin_query(sdb_ast_node_t *ast, sdb_strbuf_t *buf, sdb_strbuf_t *errbuf) int sdb_plugin_store_host(const char *name, sdb_time_t last_update) { - sdb_store_host_t host = { 0 }; + sdb_store_host_t host = SDB_STORE_HOST_INIT; char *backends[1]; char *cname; @@ -1495,19 +1728,23 @@ sdb_plugin_store_host(const char *name, sdb_time_t last_update) return -1; if (! sdb_llist_len(writer_list)) { - sdb_log(SDB_LOG_ERR, "core: Cannot store host: " - "no writers registered"); + sdb_log(SDB_LOG_ERR, "Cannot store host: no writers registered"); return -1; } cname = sdb_plugin_cname(strdup(name)); if (! cname) { - sdb_log(SDB_LOG_ERR, "core: strdup failed"); + sdb_log(SDB_LOG_ERR, "strdup failed"); return -1; } host.name = cname; - host.last_update = last_update; + host.last_update = last_update ? last_update : sdb_gettime(); + if (get_interval(SDB_HOST, NULL, -1, NULL, cname, + host.last_update, &host.interval)) { + free(cname); + return 1; + } host.backends = (const char * const *)backends; get_backend(backends, &host.backends_num); @@ -1529,7 +1766,7 @@ int sdb_plugin_store_service(const char *hostname, const char *name, sdb_time_t last_update) { - sdb_store_service_t service = { 0 }; + sdb_store_service_t service = SDB_STORE_SERVICE_INIT; char *backends[1]; char *cname; @@ -1542,20 +1779,25 @@ sdb_plugin_store_service(const char *hostname, const char *name, return -1; if (! sdb_llist_len(writer_list)) { - sdb_log(SDB_LOG_ERR, "core: Cannot store service: " + sdb_log(SDB_LOG_ERR, "Cannot store service: " "no writers registered"); return -1; } cname = sdb_plugin_cname(strdup(hostname)); if (! cname) { - sdb_log(SDB_LOG_ERR, "core: strdup failed"); + sdb_log(SDB_LOG_ERR, "strdup failed"); return -1; } service.hostname = cname; service.name = name; - service.last_update = last_update; + service.last_update = last_update ? last_update : sdb_gettime(); + if (get_interval(SDB_SERVICE, cname, -1, NULL, name, + service.last_update, &service.interval)) { + free(cname); + return 1; + } service.backends = (const char * const *)backends; get_backend(backends, &service.backends_num); @@ -1575,7 +1817,7 @@ sdb_plugin_store_service(const char *hostname, const char *name, d.type = SDB_TYPE_STRING; d.data.string = cname; if (sdb_plugin_store_service_attribute(cname, name, - "hostname", &d, last_update)) + "hostname", &d, service.last_update)) status = -1; } @@ -1587,7 +1829,7 @@ int sdb_plugin_store_metric(const char *hostname, const char *name, sdb_metric_store_t *store, sdb_time_t last_update) { - sdb_store_metric_t metric = { 0 }; + sdb_store_metric_t metric = SDB_STORE_METRIC_INIT; char *backends[1]; char *cname; @@ -1600,14 +1842,13 @@ sdb_plugin_store_metric(const char *hostname, const char *name, return -1; if (! sdb_llist_len(writer_list)) { - sdb_log(SDB_LOG_ERR, "core: Cannot store metric: " - "no writers registered"); + sdb_log(SDB_LOG_ERR, "Cannot store metric: no writers registered"); return -1; } cname = sdb_plugin_cname(strdup(hostname)); if (! cname) { - sdb_log(SDB_LOG_ERR, "core: strdup failed"); + sdb_log(SDB_LOG_ERR, "strdup failed"); return -1; } @@ -1617,10 +1858,17 @@ sdb_plugin_store_metric(const char *hostname, const char *name, metric.hostname = cname; metric.name = name; if (store) { - metric.store.type = store->type; - metric.store.id = store->id; + if (store->last_update < last_update) + store->last_update = last_update; + metric.stores = store; + metric.stores_num = 1; + } + metric.last_update = last_update ? last_update : sdb_gettime(); + if (get_interval(SDB_METRIC, cname, -1, NULL, name, + metric.last_update, &metric.interval)) { + free(cname); + return 1; } - metric.last_update = last_update; metric.backends = (const char * const *)backends; get_backend(backends, &metric.backends_num); @@ -1640,7 +1888,7 @@ sdb_plugin_store_metric(const char *hostname, const char *name, d.type = SDB_TYPE_STRING; d.data.string = cname; if (sdb_plugin_store_metric_attribute(cname, name, - "hostname", &d, last_update)) + "hostname", &d, metric.last_update)) status = -1; } @@ -1652,7 +1900,7 @@ int sdb_plugin_store_attribute(const char *hostname, const char *key, const sdb_data_t *value, sdb_time_t last_update) { - sdb_store_attribute_t attr = { 0 }; + sdb_store_attribute_t attr = SDB_STORE_ATTRIBUTE_INIT; char *backends[1]; char *cname; @@ -1663,14 +1911,13 @@ sdb_plugin_store_attribute(const char *hostname, const char *key, return -1; if (! sdb_llist_len(writer_list)) { - sdb_log(SDB_LOG_ERR, "core: Cannot store attribute: " - "no writers registered"); + sdb_log(SDB_LOG_ERR, "Cannot store attribute: no writers registered"); return -1; } cname = sdb_plugin_cname(strdup(hostname)); if (! cname) { - sdb_log(SDB_LOG_ERR, "core: strdup failed"); + sdb_log(SDB_LOG_ERR, "strdup failed"); return -1; } @@ -1678,7 +1925,12 @@ sdb_plugin_store_attribute(const char *hostname, const char *key, attr.parent = cname; attr.key = key; attr.value = *value; - attr.last_update = last_update; + attr.last_update = last_update ? last_update : sdb_gettime(); + if (get_interval(SDB_ATTRIBUTE, cname, -1, NULL, key, + attr.last_update, &attr.interval)) { + free(cname); + return 1; + } attr.backends = (const char * const *)backends; get_backend(backends, &attr.backends_num); @@ -1700,7 +1952,7 @@ int sdb_plugin_store_service_attribute(const char *hostname, const char *service, const char *key, const sdb_data_t *value, sdb_time_t last_update) { - sdb_store_attribute_t attr = { 0 }; + sdb_store_attribute_t attr = SDB_STORE_ATTRIBUTE_INIT; char *backends[1]; char *cname; @@ -1711,14 +1963,14 @@ sdb_plugin_store_service_attribute(const char *hostname, const char *service, return -1; if (! sdb_llist_len(writer_list)) { - sdb_log(SDB_LOG_ERR, "core: Cannot store service attribute: " + sdb_log(SDB_LOG_ERR, "Cannot store service attribute: " "no writers registered"); return -1; } cname = sdb_plugin_cname(strdup(hostname)); if (! cname) { - sdb_log(SDB_LOG_ERR, "core: strdup failed"); + sdb_log(SDB_LOG_ERR, "strdup failed"); return -1; } @@ -1727,7 +1979,12 @@ sdb_plugin_store_service_attribute(const char *hostname, const char *service, attr.parent = service; attr.key = key; attr.value = *value; - attr.last_update = last_update; + attr.last_update = last_update ? last_update : sdb_gettime(); + if (get_interval(SDB_ATTRIBUTE, cname, SDB_SERVICE, service, key, + attr.last_update, &attr.interval)) { + free(cname); + return 1; + } attr.backends = (const char * const *)backends; get_backend(backends, &attr.backends_num); @@ -1749,7 +2006,7 @@ int sdb_plugin_store_metric_attribute(const char *hostname, const char *metric, const char *key, const sdb_data_t *value, sdb_time_t last_update) { - sdb_store_attribute_t attr = { 0 }; + sdb_store_attribute_t attr = SDB_STORE_ATTRIBUTE_INIT; char *backends[1]; char *cname; @@ -1760,14 +2017,14 @@ sdb_plugin_store_metric_attribute(const char *hostname, const char *metric, return -1; if (! sdb_llist_len(writer_list)) { - sdb_log(SDB_LOG_ERR, "core: Cannot store metric attribute: " + sdb_log(SDB_LOG_ERR, "Cannot store metric attribute: " "no writers registered"); return -1; } cname = sdb_plugin_cname(strdup(hostname)); if (! cname) { - sdb_log(SDB_LOG_ERR, "core: strdup failed"); + sdb_log(SDB_LOG_ERR, "strdup failed"); return -1; } @@ -1776,7 +2033,12 @@ sdb_plugin_store_metric_attribute(const char *hostname, const char *metric, attr.parent = metric; attr.key = key; attr.value = *value; - attr.last_update = last_update; + attr.last_update = last_update ? last_update : sdb_gettime(); + if (get_interval(SDB_ATTRIBUTE, cname, SDB_METRIC, metric, key, + attr.last_update, &attr.interval)) { + free(cname); + return 1; + } attr.backends = (const char * const *)backends; get_backend(backends, &attr.backends_num);