X-Git-Url: https://git.tokkee.org/?p=sysdb.git;a=blobdiff_plain;f=src%2Fcore%2Fplugin.c;h=76fc2092d022ff8588a2b5f11739e8653fa5c1c4;hp=44ea48a06b53e38434cd23a6a9a951d689d48cf2;hb=2d39d666abde2d66a55ac4536da3f250d9513f48;hpb=d33a86692200478f29b8f46a63e8ec3b54a35691 diff --git a/src/core/plugin.c b/src/core/plugin.c index 44ea48a..76fc209 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 }, }; @@ -259,6 +267,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 */ @@ -505,6 +588,52 @@ 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, "core: 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) { @@ -597,6 +726,7 @@ 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; } @@ -754,7 +884,7 @@ get_interval(int obj_type, const char *hostname, fetch.name = n; status = sdb_plugin_query(SDB_AST_NODE(&fetch), - &interval_fetcher, SDB_OBJ(&obj), NULL); + &interval_fetcher, SDB_OBJ(&obj), NULL, NULL); if ((status < 0) || (lu.obj_type != obj_type) || (lu.last_update == 0)) { *interval_out = 0; return 0; @@ -824,13 +954,13 @@ 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 " "'%s' (%s)", ctx->info.plugin_name, INFO_GET(&ctx->info, description)); - ctx_set(old_ctx); } ++ctx->use_cnt; return 0; @@ -1013,12 +1143,12 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback } /* 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_impl(&ts_fetcher_list, callback_type, "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, @@ -1067,7 +1197,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, "core: Invalid read access to plugin " "context outside a plugin"); return plugin_default_ctx; } @@ -1081,7 +1211,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, "core: Invalid write access to plugin " "context outside a plugin"); return -1; } @@ -1479,8 +1609,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; @@ -1488,33 +1617,63 @@ 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) { + fetcher = TS_FETCHER(sdb_llist_search_by_name(timeseries_fetcher_list, type)); + if (! fetcher) { sdb_log(SDB_LOG_ERR, "core: 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, "core: 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_store_writer_t *w, sdb_object_t *wd, sdb_strbuf_t *errbuf) + 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)) { @@ -1539,7 +1698,7 @@ sdb_plugin_query(sdb_ast_node_t *ast, q = reader->impl.prepare_query(ast, errbuf, reader->r_user_data); if (q) - status = reader->impl.execute_query(q, w, SDB_OBJ(wd), + status = reader->impl.execute_query(q, &query_writer, SDB_OBJ(&qw), errbuf, reader->r_user_data); else status = -1;