X-Git-Url: https://git.tokkee.org/?p=sysdb.git;a=blobdiff_plain;f=src%2Fcore%2Fplugin.c;h=ecd84b5f9c5f5f525017872ecdf54fe234a6267f;hp=f581eb1629fbe3f043c02e13f6a4f3799de9b6d4;hb=55f6919736b7574b8fb905eced926d8c003411c6;hpb=e39a0c24a894f9e04cbf7698429dc23293c7ced9 diff --git a/src/core/plugin.c b/src/core/plugin.c index f581eb1..ecd84b5 100644 --- a/src/core/plugin.c +++ b/src/core/plugin.c @@ -80,31 +80,47 @@ typedef struct { void *cb_callback; sdb_object_t *cb_user_data; ctx_t *cb_ctx; -} sdb_plugin_cb_t; -#define SDB_PLUGIN_CB_INIT { SDB_OBJECT_INIT, \ +} callback_t; +#define CB_INIT { SDB_OBJECT_INIT, \ /* callback = */ NULL, /* user_data = */ NULL, \ SDB_PLUGIN_CTX_INIT } -#define SDB_PLUGIN_CB(obj) ((sdb_plugin_cb_t *)(obj)) -#define SDB_CONST_PLUGIN_CB(obj) ((const sdb_plugin_cb_t *)(obj)) +#define CB(obj) ((callback_t *)(obj)) +#define CONST_CB(obj) ((const callback_t *)(obj)) typedef struct { - sdb_plugin_cb_t super; + callback_t super; #define ccb_callback super.cb_callback #define ccb_user_data super.cb_user_data #define ccb_ctx super.cb_ctx sdb_time_t ccb_interval; sdb_time_t ccb_next_update; -} sdb_plugin_collector_cb_t; -#define SDB_PLUGIN_CCB(obj) ((sdb_plugin_collector_cb_t *)(obj)) -#define SDB_CONST_PLUGIN_CCB(obj) ((const sdb_plugin_collector_cb_t *)(obj)) +} collector_t; +#define CCB(obj) ((collector_t *)(obj)) +#define CONST_CCB(obj) ((const collector_t *)(obj)) typedef struct { - sdb_plugin_cb_t super; /* cb_callback will always be NULL */ + callback_t super; /* cb_callback will always be NULL */ #define w_user_data super.cb_user_data #define w_ctx super.cb_ctx sdb_store_writer_t impl; -} sdb_plugin_writer_t; -#define SDB_PLUGIN_WRITER(obj) ((sdb_plugin_writer_t *)(obj)) +} writer_t; +#define WRITER(obj) ((writer_t *)(obj)) + +typedef struct { + callback_t super; /* cb_callback will always be NULL */ +#define r_user_data super.cb_user_data +#define r_ctx super.cb_ctx + sdb_store_reader_t impl; +} 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 @@ -125,8 +141,9 @@ 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; static struct { const char *type; @@ -138,8 +155,9 @@ 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 }, }; /* @@ -181,10 +199,8 @@ ctx_key_init(void) static int plugin_cmp_next_update(const sdb_object_t *a, const sdb_object_t *b) { - const sdb_plugin_collector_cb_t *ccb1 - = (const sdb_plugin_collector_cb_t *)a; - const sdb_plugin_collector_cb_t *ccb2 - = (const sdb_plugin_collector_cb_t *)b; + const collector_t *ccb1 = (const collector_t *)a; + const collector_t *ccb2 = (const collector_t *)b; assert(ccb1 && ccb2); @@ -196,7 +212,7 @@ plugin_cmp_next_update(const sdb_object_t *a, const sdb_object_t *b) static int plugin_lookup_by_name(const sdb_object_t *obj, const void *id) { - const sdb_plugin_cb_t *cb = SDB_CONST_PLUGIN_CB(obj); + const callback_t *cb = CONST_CB(obj); const char *name = id; assert(cb && id); @@ -225,9 +241,9 @@ plugin_unregister_by_name(const char *plugin_name) sdb_llist_t *list = *all_lists[i].list; while (1) { - sdb_plugin_cb_t *cb; + callback_t *cb; - cb = SDB_PLUGIN_CB(sdb_llist_remove(list, + cb = CB(sdb_llist_remove(list, plugin_lookup_by_name, plugin_name)); if (! cb) break; @@ -251,6 +267,73 @@ 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; +} query_writer_t; +#define QUERY_WRITER_INIT(w, ud) { SDB_OBJECT_INIT, (w), (ud) } +#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; + + for (i = 0; i < metric->stores_num; i++) { + /* TODO: Make this optional using query options. */ + 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 */ @@ -337,33 +420,48 @@ 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; + 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 */ - SDB_PLUGIN_CB(obj)->cb_callback = callback; - SDB_PLUGIN_CB(obj)->cb_ctx = ctx_get(); - sdb_object_ref(SDB_OBJ(SDB_PLUGIN_CB(obj)->cb_ctx)); + CB(obj)->cb_callback = callback; + CB(obj)->cb_ctx = ctx_get(); + sdb_object_ref(SDB_OBJ(CB(obj)->cb_ctx)); sdb_object_ref(ud); - SDB_PLUGIN_CB(obj)->cb_user_data = ud; + CB(obj)->cb_user_data = ud; return 0; } /* plugin_cb_init */ @@ -371,19 +469,19 @@ static void plugin_cb_destroy(sdb_object_t *obj) { assert(obj); - sdb_object_deref(SDB_PLUGIN_CB(obj)->cb_user_data); - sdb_object_deref(SDB_OBJ(SDB_PLUGIN_CB(obj)->cb_ctx)); + sdb_object_deref(CB(obj)->cb_user_data); + sdb_object_deref(SDB_OBJ(CB(obj)->cb_ctx)); } /* plugin_cb_destroy */ -static sdb_type_t sdb_plugin_cb_type = { - sizeof(sdb_plugin_cb_t), +static sdb_type_t callback_type = { + sizeof(callback_t), plugin_cb_init, plugin_cb_destroy }; -static sdb_type_t sdb_plugin_collector_cb_type = { - sizeof(sdb_plugin_collector_cb_t), +static sdb_type_t collector_type = { + sizeof(collector_t), plugin_cb_init, plugin_cb_destroy @@ -392,34 +490,32 @@ static sdb_type_t sdb_plugin_collector_cb_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) - || (! impl->store_service_attr) || (! impl->store_metric_attr)) { + || (! 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); - return -1; - } - /* ctx may be NULL if the plugin was not registered by a plugin */ + /* ctx may be NULL if the callback was not registered by a plugin */ - SDB_PLUGIN_WRITER(obj)->impl = *impl; - SDB_PLUGIN_WRITER(obj)->w_ctx = ctx_get(); - sdb_object_ref(SDB_OBJ(SDB_PLUGIN_WRITER(obj)->w_ctx)); + WRITER(obj)->impl = *impl; + WRITER(obj)->w_ctx = ctx_get(); + sdb_object_ref(SDB_OBJ(WRITER(obj)->w_ctx)); sdb_object_ref(ud); - SDB_PLUGIN_WRITER(obj)->w_user_data = ud; + WRITER(obj)->w_user_data = ud; return 0; } /* plugin_writer_init */ @@ -427,17 +523,109 @@ static void plugin_writer_destroy(sdb_object_t *obj) { assert(obj); - sdb_object_deref(SDB_PLUGIN_WRITER(obj)->w_user_data); - sdb_object_deref(SDB_OBJ(SDB_PLUGIN_WRITER(obj)->w_ctx)); + sdb_object_deref(WRITER(obj)->w_user_data); + sdb_object_deref(SDB_OBJ(WRITER(obj)->w_ctx)); } /* plugin_writer_destroy */ -static sdb_type_t sdb_plugin_writer_type = { - sizeof(sdb_plugin_writer_t), +static sdb_type_t writer_type = { + sizeof(writer_t), plugin_writer_init, plugin_writer_destroy }; +static int +plugin_reader_init(sdb_object_t *obj, va_list ap) +{ + 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; + } + + /* ctx may be NULL if the callback was not registered by a plugin */ + + READER(obj)->impl = *impl; + READER(obj)->r_ctx = ctx_get(); + sdb_object_ref(SDB_OBJ(READER(obj)->r_ctx)); + + sdb_object_ref(ud); + READER(obj)->r_user_data = ud; + return 0; +} /* plugin_reader_init */ + +static void +plugin_reader_destroy(sdb_object_t *obj) +{ + assert(obj); + sdb_object_deref(READER(obj)->r_user_data); + sdb_object_deref(SDB_OBJ(READER(obj)->r_ctx)); +} /* plugin_reader_destroy */ + +static sdb_type_t reader_type = { + sizeof(reader_t), + + plugin_reader_init, + 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) { @@ -574,12 +762,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); @@ -589,8 +777,7 @@ plugin_add_callback(sdb_llist_t **list, const char *type, if (! *list) return -1; - obj = sdb_object_create(name, sdb_plugin_cb_type, - list, type, callback, user_data); + obj = sdb_object_create(name, T, list, type, impl, user_data); if (! obj) return -1; @@ -605,7 +792,127 @@ plugin_add_callback(sdb_llist_t **list, const char *type, sdb_log(SDB_LOG_INFO, "core: Registered %s callback '%s'.", type, name); return 0; -} /* plugin_add_callback */ +} /* 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); + 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, "memstore: 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; +} /* get_interval */ + +static void +get_backend(char **backends, size_t *backends_num) +{ + const sdb_plugin_info_t *info; + + info = sdb_plugin_current(); + if ((! info) || (! info->plugin_name) || (! *info->plugin_name)) { + *backends_num = 0; + return; + } + + backends[0] = info->plugin_name; + *backends_num = 1; +} /* get_backend */ /* * public API @@ -722,8 +1029,8 @@ sdb_plugin_register_config(sdb_plugin_config_cb callback) "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 @@ -731,7 +1038,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 */ @@ -741,7 +1048,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 */ @@ -751,7 +1058,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 */ @@ -761,7 +1068,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 */ @@ -783,13 +1090,13 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback plugin_get_name(name, cb_name, sizeof(cb_name)); - obj = sdb_object_create(cb_name, sdb_plugin_collector_cb_type, + obj = sdb_object_create(cb_name, collector_type, &collector_list, "collector", callback, user_data); if (! obj) return -1; if (interval) - SDB_PLUGIN_CCB(obj)->ccb_interval = *interval; + CCB(obj)->ccb_interval = *interval; else { ctx_t *ctx = ctx_get(); @@ -800,10 +1107,10 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback return -1; } - SDB_PLUGIN_CCB(obj)->ccb_interval = ctx->public.interval; + CCB(obj)->ccb_interval = ctx->public.interval; } - if (! (SDB_PLUGIN_CCB(obj)->ccb_next_update = sdb_gettime())) { + 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))); @@ -822,53 +1129,38 @@ sdb_plugin_register_collector(const char *name, sdb_plugin_collector_cb callback sdb_log(SDB_LOG_INFO, "core: Registered collector callback '%s' " "(interval = %.3fs).", cb_name, - SDB_TIME_TO_DOUBLE(SDB_PLUGIN_CCB(obj)->ccb_interval)); + 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, sdb_plugin_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 +sdb_plugin_register_reader(const char *name, + sdb_store_reader_t *reader, sdb_object_t *user_data) +{ + char cb_name[1024]; + return plugin_add_impl(&reader_list, reader_type, "store reader", + plugin_get_name(name, cb_name, sizeof(cb_name)), + reader, user_data); +} /* sdb_plugin_register_reader */ + void sdb_plugin_unregister_all(void) { @@ -934,7 +1226,7 @@ sdb_plugin_current(void) int sdb_plugin_configure(const char *name, oconfig_item_t *ci) { - sdb_plugin_cb_t *plugin; + callback_t *plugin; sdb_plugin_config_cb callback; ctx_t *old_ctx; @@ -944,7 +1236,7 @@ sdb_plugin_configure(const char *name, oconfig_item_t *ci) if ((! name) || (! ci)) return -1; - plugin = SDB_PLUGIN_CB(sdb_llist_search_by_name(config_list, name)); + plugin = CB(sdb_llist_search_by_name(config_list, name)); if (! plugin) { ctx_t *ctx = CTX(sdb_llist_search_by_name(all_plugins, name)); if (! ctx) @@ -976,11 +1268,11 @@ sdb_plugin_reconfigure_init(void) /* deconfigure all plugins */ while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_cb_t *plugin; + callback_t *plugin; sdb_plugin_config_cb callback; ctx_t *old_ctx; - plugin = SDB_PLUGIN_CB(sdb_llist_iter_get_next(iter)); + plugin = CB(sdb_llist_iter_get_next(iter)); old_ctx = ctx_set(plugin->cb_ctx); callback = (sdb_plugin_config_cb)plugin->cb_callback; callback(NULL); @@ -1033,13 +1325,13 @@ sdb_plugin_init_all(void) iter = sdb_llist_get_iter(init_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_cb_t *cb; + callback_t *cb; sdb_plugin_init_cb callback; ctx_t *old_ctx; sdb_object_t *obj = sdb_llist_iter_get_next(iter); assert(obj); - cb = SDB_PLUGIN_CB(obj); + cb = CB(obj); callback = (sdb_plugin_init_cb)cb->cb_callback; @@ -1066,13 +1358,13 @@ sdb_plugin_shutdown_all(void) iter = sdb_llist_get_iter(shutdown_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_cb_t *cb; + callback_t *cb; sdb_plugin_shutdown_cb callback; ctx_t *old_ctx; sdb_object_t *obj = sdb_llist_iter_get_next(iter); assert(obj); - cb = SDB_PLUGIN_CB(obj); + cb = CB(obj); callback = (sdb_plugin_shutdown_cb)cb->cb_callback; @@ -1110,18 +1402,18 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) if (! obj) return -1; - callback = (sdb_plugin_collector_cb)SDB_PLUGIN_CCB(obj)->ccb_callback; + callback = (sdb_plugin_collector_cb)CCB(obj)->ccb_callback; if (! (now = sdb_gettime())) { char errbuf[1024]; sdb_log(SDB_LOG_ERR, "core: Failed to determine current " "time in collector main loop: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); - now = SDB_PLUGIN_CCB(obj)->ccb_next_update; + now = CCB(obj)->ccb_next_update; } - if (now < SDB_PLUGIN_CCB(obj)->ccb_next_update) { - interval = SDB_PLUGIN_CCB(obj)->ccb_next_update - now; + if (now < CCB(obj)->ccb_next_update) { + interval = CCB(obj)->ccb_next_update - now; errno = 0; while (loop->do_loop && sdb_sleep(interval, &interval)) { @@ -1147,13 +1439,13 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) } } - old_ctx = ctx_set(SDB_PLUGIN_CCB(obj)->ccb_ctx); - if (callback(SDB_PLUGIN_CCB(obj)->ccb_user_data)) { + old_ctx = ctx_set(CCB(obj)->ccb_ctx); + if (callback(CCB(obj)->ccb_user_data)) { /* XXX */ } ctx_set(old_ctx); - interval = SDB_PLUGIN_CCB(obj)->ccb_interval; + interval = CCB(obj)->ccb_interval; if (! interval) interval = loop->default_interval; if (! interval) { @@ -1164,21 +1456,21 @@ sdb_plugin_collector_loop(sdb_plugin_loop_t *loop) continue; } - SDB_PLUGIN_CCB(obj)->ccb_next_update += interval; + CCB(obj)->ccb_next_update += interval; if (! (now = sdb_gettime())) { char errbuf[1024]; sdb_log(SDB_LOG_ERR, "core: Failed to determine current " "time in collector main loop: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); - now = SDB_PLUGIN_CCB(obj)->ccb_next_update; + now = CCB(obj)->ccb_next_update; } - if (now > SDB_PLUGIN_CCB(obj)->ccb_next_update) { + if (now > CCB(obj)->ccb_next_update) { sdb_log(SDB_LOG_WARNING, "core: Plugin '%s' took too " "long; skipping iterations to keep up.", obj->name); - SDB_PLUGIN_CCB(obj)->ccb_next_update = now; + CCB(obj)->ccb_next_update = now; } if (sdb_llist_insert_sorted(collector_list, obj, @@ -1216,8 +1508,8 @@ sdb_plugin_cname(char *hostname) sdb_object_t *obj = sdb_llist_iter_get_next(iter); assert(obj); - callback = (sdb_plugin_cname_cb)SDB_PLUGIN_CB(obj)->cb_callback; - cname = callback(hostname, SDB_PLUGIN_CB(obj)->cb_user_data); + callback = (sdb_plugin_cname_cb)CB(obj)->cb_callback; + cname = callback(hostname, CB(obj)->cb_user_data); if (cname) { free(hostname); hostname = cname; @@ -1247,12 +1539,12 @@ sdb_plugin_log(int prio, const char *msg) sdb_object_t *obj = sdb_llist_iter_get_next(iter); assert(obj); - callback = (sdb_plugin_log_cb)SDB_PLUGIN_CB(obj)->cb_callback; - tmp = callback(prio, msg, SDB_PLUGIN_CB(obj)->cb_user_data); + callback = (sdb_plugin_log_cb)CB(obj)->cb_callback; + tmp = callback(prio, msg, CB(obj)->cb_user_data); if (tmp > ret) ret = tmp; - if (SDB_PLUGIN_CB(obj)->cb_ctx) + if (CB(obj)->cb_ctx) logged = 1; /* else: this is an internally registered callback */ } @@ -1308,8 +1600,7 @@ sdb_timeseries_t * sdb_plugin_fetch_timeseries(const char *type, const char *id, sdb_timeseries_opts_t *opts) { - sdb_plugin_cb_t *plugin; - sdb_plugin_fetch_ts_cb callback; + ts_fetcher_t *fetcher; sdb_timeseries_t *ts; ctx_t *old_ctx; @@ -1317,39 +1608,139 @@ sdb_plugin_fetch_timeseries(const char *type, const char *id, if ((! type) || (! id) || (! opts)) return NULL; - plugin = SDB_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) +{ + 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 ((ast->type != SDB_AST_TYPE_FETCH) + && (ast->type != SDB_AST_TYPE_LIST) + && (ast->type != SDB_AST_TYPE_LOOKUP)) { + sdb_log(SDB_LOG_ERR, "core: 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)); + return -1; + } + + if (n != 1) { + char *msg = (n > 0) + ? "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); + return -1; + } + + reader = READER(sdb_llist_get(reader_list, 0)); + assert(reader); + + q = reader->impl.prepare_query(ast, errbuf, reader->r_user_data); + if (q) + status = reader->impl.execute_query(q, &query_writer, SDB_OBJ(&qw), + errbuf, reader->r_user_data); + else + status = -1; + + sdb_object_deref(SDB_OBJ(q)); + sdb_object_deref(SDB_OBJ(reader)); + return status; +} /* sdb_plugin_query */ + int sdb_plugin_store_host(const char *name, sdb_time_t last_update) { + sdb_store_host_t host = SDB_STORE_HOST_INIT; + char *backends[1]; + char *cname; + sdb_llist_iter_t *iter; int status = 0; if (! name) return -1; + if (! sdb_llist_len(writer_list)) { + sdb_log(SDB_LOG_ERR, "core: Cannot store host: " + "no writers registered"); + return -1; + } + + cname = sdb_plugin_cname(strdup(name)); + if (! cname) { + sdb_log(SDB_LOG_ERR, "core: strdup failed"); + return -1; + } + + host.name = cname; + 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); + iter = sdb_llist_get_iter(writer_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_writer_t *writer; - writer = SDB_PLUGIN_WRITER(sdb_llist_iter_get_next(iter)); + writer_t *writer = WRITER(sdb_llist_iter_get_next(iter)); + int s; assert(writer); - if (writer->impl.store_host(name, last_update, writer->w_user_data)) - status = -1; + s = writer->impl.store_host(&host, writer->w_user_data); + if (((s > 0) && (status >= 0)) || (s < 0)) + status = s; } sdb_llist_iter_destroy(iter); + free(cname); return status; } /* sdb_plugin_store_host */ @@ -1357,22 +1748,62 @@ int sdb_plugin_store_service(const char *hostname, const char *name, sdb_time_t last_update) { + sdb_store_service_t service = SDB_STORE_SERVICE_INIT; + char *backends[1]; + char *cname; + sdb_llist_iter_t *iter; + sdb_data_t d; + int status = 0; if ((! hostname) || (! name)) return -1; + if (! sdb_llist_len(writer_list)) { + sdb_log(SDB_LOG_ERR, "core: Cannot store service: " + "no writers registered"); + return -1; + } + + cname = sdb_plugin_cname(strdup(hostname)); + if (! cname) { + sdb_log(SDB_LOG_ERR, "core: strdup failed"); + return -1; + } + + service.hostname = cname; + service.name = name; + 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); + iter = sdb_llist_get_iter(writer_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_writer_t *writer; - writer = SDB_PLUGIN_WRITER(sdb_llist_iter_get_next(iter)); + writer_t *writer = WRITER(sdb_llist_iter_get_next(iter)); + int s; assert(writer); - if (writer->impl.store_service(hostname, name, last_update, - writer->w_user_data)) - status = -1; + s = writer->impl.store_service(&service, writer->w_user_data); + if (((s > 0) && (status >= 0)) || (s < 0)) + status = s; } sdb_llist_iter_destroy(iter); + + if (! status) { + /* record the hostname as an attribute */ + d.type = SDB_TYPE_STRING; + d.data.string = cname; + if (sdb_plugin_store_service_attribute(cname, name, + "hostname", &d, service.last_update)) + status = -1; + } + + free(cname); return status; } /* sdb_plugin_store_service */ @@ -1380,25 +1811,71 @@ 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 = SDB_STORE_METRIC_INIT; + char *backends[1]; + char *cname; + sdb_llist_iter_t *iter; + sdb_data_t d; + int status = 0; if ((! hostname) || (! name)) return -1; + if (! sdb_llist_len(writer_list)) { + sdb_log(SDB_LOG_ERR, "core: Cannot store metric: " + "no writers registered"); + return -1; + } + + cname = sdb_plugin_cname(strdup(hostname)); + if (! cname) { + sdb_log(SDB_LOG_ERR, "core: strdup failed"); + return -1; + } + if (store && ((! store->type) || (! store->id))) store = NULL; + metric.hostname = cname; + metric.name = name; + if (store) { + 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.backends = (const char * const *)backends; + get_backend(backends, &metric.backends_num); + iter = sdb_llist_get_iter(writer_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_writer_t *writer; - writer = SDB_PLUGIN_WRITER(sdb_llist_iter_get_next(iter)); + writer_t *writer = WRITER(sdb_llist_iter_get_next(iter)); + int s; assert(writer); - if (writer->impl.store_metric(hostname, name, store, last_update, - writer->w_user_data)) - status = -1; + s = writer->impl.store_metric(&metric, writer->w_user_data); + if (((s > 0) && (status >= 0)) || (s < 0)) + status = s; } sdb_llist_iter_destroy(iter); + + if (! status) { + /* record the hostname as an attribute */ + d.type = SDB_TYPE_STRING; + d.data.string = cname; + if (sdb_plugin_store_metric_attribute(cname, name, + "hostname", &d, metric.last_update)) + status = -1; + } + + free(cname); return status; } /* sdb_plugin_store_metric */ @@ -1406,22 +1883,52 @@ 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 = SDB_STORE_ATTRIBUTE_INIT; + char *backends[1]; + char *cname; + sdb_llist_iter_t *iter; int status = 0; if ((! hostname) || (! key) || (! value)) return -1; + if (! sdb_llist_len(writer_list)) { + sdb_log(SDB_LOG_ERR, "core: Cannot store attribute: " + "no writers registered"); + return -1; + } + + cname = sdb_plugin_cname(strdup(hostname)); + if (! cname) { + sdb_log(SDB_LOG_ERR, "core: strdup failed"); + return -1; + } + + attr.parent_type = SDB_HOST; + attr.parent = cname; + attr.key = key; + attr.value = *value; + 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); + iter = sdb_llist_get_iter(writer_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_writer_t *writer; - writer = SDB_PLUGIN_WRITER(sdb_llist_iter_get_next(iter)); + writer_t *writer = WRITER(sdb_llist_iter_get_next(iter)); + int s; assert(writer); - if (writer->impl.store_attribute(hostname, key, value, last_update, - writer->w_user_data)) - status = -1; + s = writer->impl.store_attribute(&attr, writer->w_user_data); + if (((s > 0) && (status >= 0)) || (s < 0)) + status = s; } sdb_llist_iter_destroy(iter); + free(cname); return status; } /* sdb_plugin_store_attribute */ @@ -1429,22 +1936,53 @@ 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 = SDB_STORE_ATTRIBUTE_INIT; + char *backends[1]; + char *cname; + sdb_llist_iter_t *iter; int status = 0; if ((! hostname) || (! service) || (! key) || (! value)) return -1; + if (! sdb_llist_len(writer_list)) { + sdb_log(SDB_LOG_ERR, "core: 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"); + return -1; + } + + attr.hostname = cname; + attr.parent_type = SDB_SERVICE; + attr.parent = service; + attr.key = key; + attr.value = *value; + 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); + iter = sdb_llist_get_iter(writer_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_writer_t *writer; - writer = SDB_PLUGIN_WRITER(sdb_llist_iter_get_next(iter)); + writer_t *writer = WRITER(sdb_llist_iter_get_next(iter)); + int s; assert(writer); - if (writer->impl.store_service_attr(hostname, service, - key, value, last_update, writer->w_user_data)) - status = -1; + s = writer->impl.store_attribute(&attr, writer->w_user_data); + if (((s > 0) && (status >= 0)) || (s < 0)) + status = s; } sdb_llist_iter_destroy(iter); + free(cname); return status; } /* sdb_plugin_store_service_attribute */ @@ -1452,22 +1990,53 @@ 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 = SDB_STORE_ATTRIBUTE_INIT; + char *backends[1]; + char *cname; + sdb_llist_iter_t *iter; int status = 0; if ((! hostname) || (! metric) || (! key) || (! value)) return -1; + if (! sdb_llist_len(writer_list)) { + sdb_log(SDB_LOG_ERR, "core: 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"); + return -1; + } + + attr.hostname = cname; + attr.parent_type = SDB_METRIC; + attr.parent = metric; + attr.key = key; + attr.value = *value; + 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); + iter = sdb_llist_get_iter(writer_list); while (sdb_llist_iter_has_next(iter)) { - sdb_plugin_writer_t *writer; - writer = SDB_PLUGIN_WRITER(sdb_llist_iter_get_next(iter)); + writer_t *writer = WRITER(sdb_llist_iter_get_next(iter)); + int s; assert(writer); - if (writer->impl.store_metric_attr(hostname, metric, - key, value, last_update, writer->w_user_data)) - status = -1; + s = writer->impl.store_attribute(&attr, writer->w_user_data); + if (((s > 0) && (status >= 0)) || (s < 0)) + status = s; } sdb_llist_iter_destroy(iter); + free(cname); return status; } /* sdb_plugin_store_metric_attribute */