X-Git-Url: https://git.tokkee.org/?p=sysdb.git;a=blobdiff_plain;f=src%2Fcore%2Fstore.c;h=19faee43994bdde816a6a84f4c09061bfb25bb4f;hp=db57d2e5ba9506bc7adafba2a48a58263a1edca0;hb=742c60b383c9373b2156c3af5ff9e7605c056e60;hpb=ae3324ade282506976a21dbab487e99e41935b88 diff --git a/src/core/store.c b/src/core/store.c index db57d2e..7671e98 100644 --- a/src/core/store.c +++ b/src/core/store.c @@ -1,6 +1,6 @@ /* * SysDB - src/core/store.c - * Copyright (C) 2012 Sebastian 'tokkee' Harl + * Copyright (C) 2012-2013 Sebastian 'tokkee' Harl * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -25,10 +25,15 @@ * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ +#if HAVE_CONFIG_H +# include "config.h" +#endif /* HAVE_CONFIG_H */ + #include "sysdb.h" -#include "core/store.h" +#include "core/store-private.h" +#include "core/plugin.h" +#include "utils/avltree.h" #include "utils/error.h" -#include "utils/llist.h" #include @@ -38,623 +43,1105 @@ #include #include +#include #include /* - * private data types + * private variables */ -/* type used for looking up named objects */ typedef struct { - sdb_object_t parent; - const char *obj_name; -} sdb_store_lookup_obj_t; -#define SDB_STORE_LOOKUP_OBJ_INIT { SDB_OBJECT_INIT, NULL } + sdb_object_t super; -/* - * private variables - */ + /* hosts are the top-level entries and + * reference everything else */ + sdb_avltree_t *hosts; + pthread_rwlock_t host_lock; +} sdb_store_t; +#define ST(obj) ((sdb_store_t *)(obj)) -static sdb_llist_t *host_list = NULL; -static pthread_rwlock_t host_lock = PTHREAD_RWLOCK_INITIALIZER; +sdb_store_t *global_store = NULL; /* - * private helper functions + * private types */ +static sdb_type_t host_type; +static sdb_type_t service_type; +static sdb_type_t metric_type; +static sdb_type_t attribute_type; + static int -sdb_store_obj_cmp_by_name(const sdb_object_t *a, const sdb_object_t *b) +store_init(sdb_object_t *obj, va_list __attribute__((unused)) ap) { - const sdb_store_obj_t *h1 = (const sdb_store_obj_t *)a; - const sdb_store_obj_t *h2 = (const sdb_store_obj_t *)b; + int err; + if (! (ST(obj)->hosts = sdb_avltree_create())) + return -1; + if ((err = pthread_rwlock_init(&ST(obj)->host_lock, /* attr = */ NULL))) { + char errbuf[128]; + sdb_log(SDB_LOG_ERR, "store: Failed to initialize lock: %s", + sdb_strerror(err, errbuf, sizeof(errbuf))); + return -1; + } + return 0; +} /* store_init */ - assert(h1 && h2); - return strcasecmp(h1->name, h2->name); -} /* sdb_store_obj_cmp_by_name */ +static void +store_destroy(sdb_object_t *obj) +{ + int err; + if ((err = pthread_rwlock_destroy(&ST(obj)->host_lock))) { + char errbuf[128]; + sdb_log(SDB_LOG_ERR, "store: Failed to destroy lock: %s", + sdb_strerror(err, errbuf, sizeof(errbuf))); + return; + } + sdb_avltree_destroy(ST(obj)->hosts); + ST(obj)->hosts = NULL; +} /* store_destroy */ static int -sdb_cmp_store_obj_with_name(const sdb_object_t *a, const sdb_object_t *b) +store_obj_init(sdb_object_t *obj, va_list ap) { - const sdb_store_obj_t *obj = (const sdb_store_obj_t *)a; - const sdb_store_lookup_obj_t *lookup = (const sdb_store_lookup_obj_t *)b; + sdb_store_obj_t *sobj = STORE_OBJ(obj); - assert(obj && lookup); - return strcasecmp(obj->name, lookup->obj_name); -} /* sdb_cmp_store_obj_with_name */ + sobj->type = va_arg(ap, int); -/* - * private types - */ + sobj->last_update = va_arg(ap, sdb_time_t); + sobj->interval = 0; + sobj->backends = NULL; + sobj->backends_num = 0; + sobj->parent = NULL; + return 0; +} /* store_obj_init */ -static int -sdb_host_init(sdb_object_t *obj, va_list ap) +static void +store_obj_destroy(sdb_object_t *obj) { - const char *name = va_arg(ap, const char *); + sdb_store_obj_t *sobj = STORE_OBJ(obj); + size_t i; - SDB_HOST(obj)->_name = strdup(name); - if (! SDB_HOST(obj)->_name) - return -1; + for (i = 0; i < sobj->backends_num; ++i) + free(sobj->backends[i]); + free(sobj->backends); + sobj->backends = NULL; + sobj->backends_num = 0; + + // We don't currently keep an extra reference for parent objects to + // avoid circular self-references which are not handled correctly by + // the ref-count base management layer. + //sdb_object_deref(SDB_OBJ(sobj->parent)); +} /* store_obj_destroy */ + +static int +host_init(sdb_object_t *obj, va_list ap) +{ + sdb_host_t *sobj = HOST(obj); + int ret; - SDB_HOST(obj)->_last_update = sdb_gettime(); - /* ignore errors -> last_update will be updated later */ + /* this will consume the first argument (type) of ap */ + ret = store_obj_init(obj, ap); + if (ret) + return ret; - SDB_HOST(obj)->attributes = sdb_llist_create(); - if (! SDB_HOST(obj)->attributes) + sobj->services = sdb_avltree_create(); + if (! sobj->services) + return -1; + sobj->metrics = sdb_avltree_create(); + if (! sobj->metrics) return -1; - SDB_HOST(obj)->services = sdb_llist_create(); - if (! SDB_HOST(obj)->services) + sobj->attributes = sdb_avltree_create(); + if (! sobj->attributes) return -1; return 0; -} /* sdb_host_init */ +} /* host_init */ static void -sdb_host_destroy(sdb_object_t *obj) +host_destroy(sdb_object_t *obj) { + sdb_host_t *sobj = HOST(obj); assert(obj); - if (SDB_HOST(obj)->_name) - free(SDB_HOST(obj)->_name); + store_obj_destroy(obj); - if (SDB_HOST(obj)->attributes) - sdb_llist_destroy(SDB_HOST(obj)->attributes); - if (SDB_HOST(obj)->services) - sdb_llist_destroy(SDB_HOST(obj)->services); -} /* sdb_host_destroy */ + if (sobj->services) + sdb_avltree_destroy(sobj->services); + if (sobj->metrics) + sdb_avltree_destroy(sobj->metrics); + if (sobj->attributes) + sdb_avltree_destroy(sobj->attributes); +} /* host_destroy */ static int -sdb_attr_init(sdb_object_t *obj, va_list ap) +service_init(sdb_object_t *obj, va_list ap) { - const char *hostname = va_arg(ap, const char *); - const char *name = va_arg(ap, const char *); - const char *value = va_arg(ap, const char *); + sdb_service_t *sobj = SVC(obj); + int ret; - SDB_ATTR(obj)->hostname = strdup(hostname); - SDB_ATTR(obj)->_name = strdup(name); - SDB_ATTR(obj)->attr_value = strdup(value); - if ((! SDB_ATTR(obj)->hostname) - || (! SDB_ATTR(obj)->_name) || (! SDB_ATTR(obj)->attr_value)) - return -1; + /* this will consume the first argument (type) of ap */ + ret = store_obj_init(obj, ap); + if (ret) + return ret; - SDB_ATTR(obj)->_last_update = sdb_gettime(); + sobj->attributes = sdb_avltree_create(); + if (! sobj->attributes) + return -1; return 0; -} /* sdb_attr_init */ +} /* service_init */ static void -sdb_attr_destroy(sdb_object_t *obj) +service_destroy(sdb_object_t *obj) { + sdb_service_t *sobj = SVC(obj); assert(obj); - if (SDB_ATTR(obj)->hostname) - free(SDB_ATTR(obj)->hostname); - if (SDB_ATTR(obj)->_name) - free(SDB_ATTR(obj)->_name); - if (SDB_ATTR(obj)->attr_value) - free(SDB_ATTR(obj)->attr_value); -} /* sdb_attr_destroy */ + store_obj_destroy(obj); + + if (sobj->attributes) + sdb_avltree_destroy(sobj->attributes); +} /* service_destroy */ static int -sdb_svc_init(sdb_object_t *obj, va_list ap) +metric_init(sdb_object_t *obj, va_list ap) { - const char *hostname = va_arg(ap, const char *); - const char *name = va_arg(ap, const char *); + sdb_metric_t *sobj = METRIC(obj); + int ret; - SDB_SVC(obj)->hostname = strdup(hostname); - SDB_SVC(obj)->_name = strdup(name); - if ((! SDB_SVC(obj)->hostname) || (! SDB_SVC(obj)->_name)) + /* this will consume the first argument (type) of ap */ + ret = store_obj_init(obj, ap); + if (ret) + return ret; + + sobj->attributes = sdb_avltree_create(); + if (! sobj->attributes) return -1; - SDB_SVC(obj)->_last_update = sdb_gettime(); - /* ignore errors -> last_update will be updated later */ + sobj->store.type = sobj->store.id = NULL; return 0; -} /* sdb_svc_init */ +} /* metric_init */ static void -sdb_svc_destroy(sdb_object_t *obj) +metric_destroy(sdb_object_t *obj) { + sdb_metric_t *sobj = METRIC(obj); assert(obj); - if (SDB_SVC(obj)->hostname) - free(SDB_SVC(obj)->hostname); - if (SDB_SVC(obj)->_name) - free(SDB_SVC(obj)->_name); -} /* sdb_svc_destroy */ + store_obj_destroy(obj); + + if (sobj->attributes) + sdb_avltree_destroy(sobj->attributes); + + if (sobj->store.type) + free(sobj->store.type); + if (sobj->store.id) + free(sobj->store.id); +} /* metric_destroy */ + +static int +attr_init(sdb_object_t *obj, va_list ap) +{ + const sdb_data_t *value; + int ret; + + /* this will consume the first two arguments + * (type and last_update) of ap */ + ret = store_obj_init(obj, ap); + if (ret) + return ret; + value = va_arg(ap, const sdb_data_t *); + + if (value) + if (sdb_data_copy(&ATTR(obj)->value, value)) + return -1; + return 0; +} /* attr_init */ + +static void +attr_destroy(sdb_object_t *obj) +{ + assert(obj); -static sdb_type_t sdb_host_type = { - sizeof(sdb_host_t), + store_obj_destroy(obj); + sdb_data_free_datum(&ATTR(obj)->value); +} /* attr_destroy */ - sdb_host_init, - sdb_host_destroy +static sdb_type_t store_type = { + /* size = */ sizeof(sdb_store_t), + /* init = */ store_init, + /* destroy = */ store_destroy, }; -static sdb_type_t sdb_attr_type = { - sizeof(sdb_attribute_t), +static sdb_type_t host_type = { + /* size = */ sizeof(sdb_host_t), + /* init = */ host_init, + /* destroy = */ host_destroy +}; - sdb_attr_init, - sdb_attr_destroy +static sdb_type_t service_type = { + /* size = */ sizeof(sdb_service_t), + /* init = */ service_init, + /* destroy = */ service_destroy }; -static sdb_type_t sdb_svc_type = { - sizeof(sdb_service_t), +static sdb_type_t metric_type = { + /* size = */ sizeof(sdb_metric_t), + /* init = */ metric_init, + /* destroy = */ metric_destroy +}; - sdb_svc_init, - sdb_svc_destroy +static sdb_type_t attribute_type = { + /* size = */ sizeof(sdb_attribute_t), + /* init = */ attr_init, + /* destroy = */ attr_destroy }; /* - * public API + * private helper functions */ -sdb_host_t * -sdb_host_create(const char *name) +static sdb_host_t * +lookup_host(sdb_store_t *st, const char *name, bool canonicalize) { - sdb_object_t *obj; + sdb_host_t *host; + char *cname; - if (! name) - return NULL; + assert(name); + if (! canonicalize) + return HOST(sdb_avltree_lookup(st->hosts, name)); - obj = sdb_object_create(sdb_host_type, name); - if (! obj) + cname = strdup(name); + cname = sdb_plugin_cname(cname); + if (! cname) { + sdb_log(SDB_LOG_ERR, "store: strdup failed"); return NULL; - return SDB_HOST(obj); -} /* sdb_host_create */ + } + + host = HOST(sdb_avltree_lookup(st->hosts, cname)); + free(cname); + return host; +} /* lookup_host */ -sdb_host_t * -sdb_host_clone(const sdb_host_t *host) +static int +record_backend(sdb_store_obj_t *obj) { - sdb_host_t *new; + const sdb_plugin_info_t *info; + char **tmp; + size_t i; - new = sdb_host_create(host->_name); - if (! new) - return NULL; + info = sdb_plugin_current(); + if (! info) + return 0; - /* make sure these are initialized; else sdb_object_deref() might access - * arbitrary memory in case of an error */ - new->services = new->attributes = NULL; + for (i = 0; i < obj->backends_num; ++i) + if (!strcasecmp(obj->backends[i], info->plugin_name)) + return 0; - if (host->attributes) { - new->attributes = sdb_llist_clone(host->attributes); - if (! new->attributes) { - sdb_object_deref(SDB_OBJ(new)); - return NULL; - } - } + tmp = realloc(obj->backends, + (obj->backends_num + 1) * sizeof(*obj->backends)); + if (! tmp) + return -1; - new->_last_update = host->_last_update; - if (host->services) { - new->services = sdb_llist_clone(host->services); - if (! new->services) { - sdb_object_deref(SDB_OBJ(new)); - return NULL; - } - } - return new; -} /* sdb_host_clone */ + obj->backends = tmp; + obj->backends[obj->backends_num] = strdup(info->plugin_name); + if (! obj->backends[obj->backends_num]) + return -1; -int -sdb_store_host(const sdb_host_t *host) -{ - sdb_store_lookup_obj_t lookup = SDB_STORE_LOOKUP_OBJ_INIT; - sdb_time_t last_update; + ++obj->backends_num; + return 0; +} /* record_backend */ - sdb_host_t *old; +static int +store_obj(sdb_store_obj_t *parent, sdb_avltree_t *parent_tree, + int type, const char *name, sdb_time_t last_update, + sdb_store_obj_t **updated_obj) +{ + sdb_store_obj_t *old, *new; int status = 0; - if ((! host) || (! host->_name)) - return -1; + assert(parent_tree); - last_update = host->_last_update; if (last_update <= 0) last_update = sdb_gettime(); - pthread_rwlock_wrlock(&host_lock); - - if (! host_list) { - if (! (host_list = sdb_llist_create())) { - pthread_rwlock_unlock(&host_lock); - return -1; - } - } - - lookup.obj_name = host->_name; - old = SDB_HOST(sdb_llist_search(host_list, (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); - + old = STORE_OBJ(sdb_avltree_lookup(parent_tree, name)); if (old) { - if (old->_last_update > last_update) { - sdb_log(SDB_LOG_DEBUG, "store: Cannot update host '%s' - " - "value too old (%"PRIscTIME" < %"PRIscTIME")", - host->_name, last_update, old->_last_update); - /* don't report an error; the host may be updated by multiple + if (old->last_update > last_update) { + sdb_log(SDB_LOG_DEBUG, "store: Cannot update %s '%s' - " + "value too old (%"PRIsdbTIME" < %"PRIsdbTIME")", + SDB_STORE_TYPE_TO_NAME(type), name, + last_update, old->last_update); + /* don't report an error; the object may be updated by multiple * backends */ status = 1; } + else if (old->last_update == last_update) { + /* don't report an error and also don't even log this to avoid + * excessive noise on high sampling frequencies */ + status = 1; + } else { - old->_last_update = last_update; + sdb_time_t interval = last_update - old->last_update; + old->last_update = last_update; + if (interval) { + if (old->interval) + old->interval = (sdb_time_t)((0.9 * (double)old->interval) + + (0.1 * (double)interval)); + else + old->interval = interval; + } } + + new = old; + sdb_object_deref(SDB_OBJ(old)); } else { - sdb_host_t *new = sdb_host_clone(host); - if (! new) { - char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "store: Failed to clone host object: %s", - sdb_strerror(errno, errbuf, sizeof(errbuf))); - pthread_rwlock_unlock(&host_lock); - return -1; + if (type == SDB_ATTRIBUTE) { + /* the value will be updated by the caller */ + new = STORE_OBJ(sdb_object_create(name, attribute_type, + type, last_update, NULL)); } - - if (! new->attributes) { - if (! (new->attributes = sdb_llist_create())) { - char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "store: Failed to initialize " - "host object '%s': %s", host->_name, - sdb_strerror(errno, errbuf, sizeof(errbuf))); - sdb_object_deref(SDB_OBJ(new)); - pthread_rwlock_unlock(&host_lock); - return -1; - } + else { + sdb_type_t t; + t = type == SDB_HOST + ? host_type + : type == SDB_SERVICE + ? service_type + : metric_type; + new = STORE_OBJ(sdb_object_create(name, t, type, last_update)); } - if (! new->services) { - if (! (new->services = sdb_llist_create())) { - char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "store: Failed to initialize " - "host object '%s': %s", host->_name, - sdb_strerror(errno, errbuf, sizeof(errbuf))); - sdb_object_deref(SDB_OBJ(new)); - pthread_rwlock_unlock(&host_lock); - return -1; - } + if (new) { + status = sdb_avltree_insert(parent_tree, SDB_OBJ(new)); + + /* pass control to the tree or destroy in case of an error */ + sdb_object_deref(SDB_OBJ(new)); + } + else { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "store: Failed to create %s '%s': %s", + SDB_STORE_TYPE_TO_NAME(type), name, + sdb_strerror(errno, errbuf, sizeof(errbuf))); + status = -1; } + } - status = sdb_llist_insert_sorted(host_list, SDB_OBJ(new), - sdb_store_obj_cmp_by_name); + if (status < 0) + return status; + assert(new); - /* pass control to the list or destroy in case of an error */ - sdb_object_deref(SDB_OBJ(new)); + if (new->parent != parent) { + // Avoid circular self-references which are not handled + // correctly by the ref-count based management layer. + //sdb_object_deref(SDB_OBJ(new->parent)); + //sdb_object_ref(SDB_OBJ(parent)); + new->parent = parent; } - pthread_rwlock_unlock(&host_lock); + if (updated_obj) + *updated_obj = new; + + if (record_backend(new)) + return -1; return status; -} /* sdb_store_host */ +} /* store_obj */ -const sdb_host_t * -sdb_store_get_host(const char *name) +static int +store_attr(sdb_store_obj_t *parent, sdb_avltree_t *attributes, + const char *key, const sdb_data_t *value, sdb_time_t last_update) { - sdb_store_lookup_obj_t lookup = SDB_STORE_LOOKUP_OBJ_INIT; - sdb_host_t *host; + sdb_store_obj_t *attr = NULL; + int status; - if (! name) - return NULL; + status = store_obj(parent, attributes, SDB_ATTRIBUTE, + key, last_update, &attr); + if (status) + return status; - lookup.obj_name = name; - host = SDB_HOST(sdb_llist_search(host_list, (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); + /* don't update unchanged values */ + if (! sdb_data_cmp(&ATTR(attr)->value, value)) + return status; - if (! host) - return NULL; - return host; -} /* sdb_store_get_host */ + assert(attr); + if (sdb_data_copy(&ATTR(attr)->value, value)) + return -1; + return status; +} /* store_attr */ -sdb_attribute_t * -sdb_attribute_create(const char *hostname, - const char *name, const char *value) +static int +store_metric_store(sdb_metric_t *metric, sdb_metric_store_t *store) { - sdb_object_t *obj; + char *type = metric->store.type; + char *id = metric->store.id; + + if ((! metric->store.type) || strcasecmp(metric->store.type, store->type)) { + if (! (type = strdup(store->type))) + return -1; + } + if ((! metric->store.id) || strcasecmp(metric->store.id, store->id)) { + if (! (id = strdup(store->id))) { + if (type != metric->store.type) + free(type); + return -1; + } + } + + if (type != metric->store.type) { + if (metric->store.type) + free(metric->store.type); + metric->store.type = type; + } + if (id != metric->store.id) { + if (metric->store.id) + free(metric->store.id); + metric->store.id = id; + } + return 0; +} /* store_metric_store */ - if ((! hostname) || (! name) || (! value)) +/* The store's host_lock has to be acquired before calling this function. */ +static sdb_avltree_t * +get_host_children(sdb_host_t *host, int type) +{ + if ((type != SDB_SERVICE) && (type != SDB_METRIC) + && (type != SDB_ATTRIBUTE)) return NULL; - obj = sdb_object_create(sdb_attr_type, hostname, name, value); - if (! obj) + if (! host) return NULL; - return SDB_ATTR(obj); -} /* sdb_attribute_create */ -sdb_attribute_t * -sdb_attribute_clone(const sdb_attribute_t *attr) + if (type == SDB_ATTRIBUTE) + return host->attributes; + else if (type == SDB_METRIC) + return host->metrics; + else + return host->services; +} /* get_host_children */ + +/* + * ts_tojson serializes a time-series to JSON. + * + * The function never returns an error. Rather, an error message will be part + * of the serialized data. + */ +static void +ts_tojson(sdb_timeseries_t *ts, sdb_strbuf_t *buf) { - sdb_attribute_t *new; + char start_str[64]; + char end_str[64]; - new = sdb_attribute_create(attr->hostname, - attr->_name, attr->attr_value); - if (! new) - return NULL; + size_t i; - new->_last_update = attr->_last_update; - return new; -} /* sdb_attribute_clone */ + /* TODO: make time format configurable */ + if (! sdb_strftime(start_str, sizeof(start_str), ts->start)) + snprintf(start_str, sizeof(start_str), ""); + start_str[sizeof(start_str) - 1] = '\0'; + if (! sdb_strftime(end_str, sizeof(end_str), ts->end)) + snprintf(end_str, sizeof(end_str), ""); + end_str[sizeof(end_str) - 1] = '\0'; -int -sdb_store_attribute(const sdb_attribute_t *attr) + sdb_strbuf_append(buf, "{\"start\": \"%s\", \"end\": \"%s\", \"data\": {", + start_str, end_str); + + for (i = 0; i < ts->data_names_len; ++i) { + size_t j; + sdb_strbuf_append(buf, "\"%s\": [", ts->data_names[i]); + + for (j = 0; j < ts->data_len; ++j) { + char time_str[64]; + + if (! sdb_strftime(time_str, sizeof(time_str), ts->data[i][j].timestamp)) + snprintf(time_str, sizeof(time_str), ""); + time_str[sizeof(time_str) - 1] = '\0'; + + /* Some GNU libc versions may print '-nan' which we dont' want */ + if (isnan(ts->data[i][j].value)) + sdb_strbuf_append(buf, "{\"timestamp\": \"%s\", " + "\"value\": \"nan\"}", time_str); + else + sdb_strbuf_append(buf, "{\"timestamp\": \"%s\", " + "\"value\": \"%f\"}", time_str, ts->data[i][j].value); + + if (j < ts->data_len - 1) + sdb_strbuf_append(buf, ","); + } + + if (i < ts->data_names_len - 1) + sdb_strbuf_append(buf, "],"); + else + sdb_strbuf_append(buf, "]"); + } + sdb_strbuf_append(buf, "}}"); +} /* ts_tojson */ + +/* + * store writer API + */ + +static int +store_attribute(const char *hostname, + const char *key, const sdb_data_t *value, + sdb_time_t last_update, sdb_object_t *user_data) { - sdb_store_lookup_obj_t lookup = SDB_STORE_LOOKUP_OBJ_INIT; + sdb_store_t *st = ST(user_data); + sdb_host_t *host; + sdb_avltree_t *attrs; + int status = 0; + + if ((! hostname) || (! key)) + return -1; + + pthread_rwlock_wrlock(&st->host_lock); + host = lookup_host(st, hostname, /* canonicalize = */ 1); + attrs = get_host_children(host, SDB_ATTRIBUTE); + if (! attrs) { + sdb_log(SDB_LOG_ERR, "store: Failed to store attribute '%s' - " + "host '%s' not found", key, hostname); + status = -1; + } - sdb_attribute_t *old; + if (! status) + status = store_attr(STORE_OBJ(host), attrs, key, value, last_update); - sdb_time_t last_update; + sdb_object_deref(SDB_OBJ(host)); + pthread_rwlock_unlock(&st->host_lock); + + return status; +} /* store_attribute */ + +static int +store_host(const char *name, sdb_time_t last_update, sdb_object_t *user_data) +{ + sdb_store_t *st = ST(user_data); + char *cname = NULL; int status = 0; - if (! attr) + if (! name) return -1; - last_update = attr->_last_update; - if (last_update <= 0) - last_update = sdb_gettime(); - - if (! host_list) + cname = sdb_plugin_cname(strdup(name)); + if (! cname) { + sdb_log(SDB_LOG_ERR, "store: strdup failed"); return -1; + } + + pthread_rwlock_wrlock(&st->host_lock); + status = store_obj(NULL, st->hosts, + SDB_HOST, cname, last_update, NULL); + pthread_rwlock_unlock(&st->host_lock); + + free(cname); + return status; +} /* store_host */ + +static int +store_service_attr(const char *hostname, const char *service, + const char *key, const sdb_data_t *value, sdb_time_t last_update, + sdb_object_t *user_data) +{ + sdb_store_t *st = ST(user_data); - pthread_rwlock_wrlock(&host_lock); + sdb_host_t *host; + sdb_service_t *svc; + sdb_avltree_t *services; + int status = 0; - lookup.obj_name = attr->hostname; - host = SDB_HOST(sdb_llist_search(host_list, (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); + if ((! hostname) || (! service) || (! key)) + return -1; - if (! host) { - pthread_rwlock_unlock(&host_lock); + pthread_rwlock_wrlock(&st->host_lock); + host = lookup_host(st, hostname, /* canonicalize = */ 1); + services = get_host_children(host, SDB_SERVICE); + sdb_object_deref(SDB_OBJ(host)); + if (! services) { + sdb_log(SDB_LOG_ERR, "store: Failed to store attribute '%s' " + "for service '%s' - host '%ss' not found", + key, service, hostname); + pthread_rwlock_unlock(&st->host_lock); return -1; } - lookup.obj_name = attr->_name; - old = SDB_ATTR(sdb_llist_search(host->attributes, - (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); - - if (old) { - if (old->_last_update > last_update) { - sdb_log(SDB_LOG_DEBUG, "store: Cannot update attribute " - "'%s/%s' - value too old (%"PRIscTIME" < %"PRIscTIME")", - attr->hostname, attr->_name, last_update, - old->_last_update); - status = 1; - } - else { - old->_last_update = last_update; - } + svc = SVC(sdb_avltree_lookup(services, service)); + if (! svc) { + sdb_log(SDB_LOG_ERR, "store: Failed to store attribute '%s' - " + "service '%s/%s' not found", key, hostname, service); + status = -1; } - else { - sdb_attribute_t *new = sdb_attribute_clone(attr); - if (! new) { - char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "store: Failed to clone attribute " - "object: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); - pthread_rwlock_unlock(&host_lock); - return -1; - } - status = sdb_llist_insert_sorted(host->attributes, SDB_OBJ(new), - sdb_store_obj_cmp_by_name); + if (! status) + status = store_attr(STORE_OBJ(svc), svc->attributes, + key, value, last_update); - /* pass control to the list or destroy in case of an error */ - sdb_object_deref(SDB_OBJ(new)); - } + sdb_object_deref(SDB_OBJ(svc)); + pthread_rwlock_unlock(&st->host_lock); - pthread_rwlock_unlock(&host_lock); return status; -} /* sdb_store_attribute */ +} /* store_service_attr */ -sdb_service_t * -sdb_service_create(const char *hostname, const char *name) +static int +store_service(const char *hostname, const char *name, + sdb_time_t last_update, sdb_object_t *user_data) { - sdb_object_t *obj; + sdb_store_t *st = ST(user_data); + + sdb_host_t *host; + sdb_avltree_t *services; + sdb_data_t d; + + int status = 0; if ((! hostname) || (! name)) - return NULL; + return -1; - obj = sdb_object_create(sdb_svc_type, hostname, name); - if (! obj) - return NULL; - return SDB_SVC(obj); -} /* sdb_service_create */ + pthread_rwlock_wrlock(&st->host_lock); + host = lookup_host(st, hostname, /* canonicalize = */ 1); + services = get_host_children(host, SDB_SERVICE); + if (! services) { + sdb_log(SDB_LOG_ERR, "store: Failed to store service '%s' - " + "host '%s' not found", name, hostname); + status = -1; + } -sdb_service_t * -sdb_service_clone(const sdb_service_t *svc) -{ - sdb_service_t *new; + if (! status) + status = store_obj(STORE_OBJ(host), services, SDB_SERVICE, + name, last_update, NULL); - new = sdb_service_create(svc->hostname, svc->_name); - if (! new) - return NULL; + sdb_object_deref(SDB_OBJ(host)); + pthread_rwlock_unlock(&st->host_lock); - new->_last_update = svc->_last_update; - return new; -} /* sdb_service_clone */ + if (status) + return status; -int -sdb_store_service(const sdb_service_t *svc) + /* record the hostname as an attribute */ + d.type = SDB_TYPE_STRING; + d.data.string = SDB_OBJ(host)->name; + if (store_service_attr(hostname, name, "hostname", &d, last_update, user_data)) + status = -1; + return status; +} /* store_service */ + +static int +store_metric_attr(const char *hostname, const char *metric, + const char *key, const sdb_data_t *value, sdb_time_t last_update, + sdb_object_t *user_data) { - sdb_store_lookup_obj_t lookup = SDB_STORE_LOOKUP_OBJ_INIT; + sdb_store_t *st = ST(user_data); + + sdb_avltree_t *metrics; sdb_host_t *host; + sdb_metric_t *m; + int status = 0; - sdb_service_t *old; + if ((! hostname) || (! metric) || (! key)) + return -1; - sdb_time_t last_update; + pthread_rwlock_wrlock(&st->host_lock); + host = lookup_host(st, hostname, /* canonicalize = */ 1); + metrics = get_host_children(host, SDB_METRIC); + sdb_object_deref(SDB_OBJ(host)); + if (! metrics) { + sdb_log(SDB_LOG_ERR, "store: Failed to store attribute '%s' " + "for metric '%s' - host '%s' not found", + key, metric, hostname); + pthread_rwlock_unlock(&st->host_lock); + return -1; + } - int status = 0; + m = METRIC(sdb_avltree_lookup(metrics, metric)); + if (! m) { + sdb_log(SDB_LOG_ERR, "store: Failed to store attribute '%s' - " + "metric '%s/%s' not found", key, hostname, metric); + status = -1; + } - if (! svc) - return -1; + if (! status) + status = store_attr(STORE_OBJ(m), m->attributes, + key, value, last_update); - last_update = svc->_last_update; - if (last_update <= 0) - last_update = sdb_gettime(); + sdb_object_deref(SDB_OBJ(m)); + pthread_rwlock_unlock(&st->host_lock); - if (! host_list) - return -1; + return status; +} /* store_metric_attr */ - pthread_rwlock_wrlock(&host_lock); +static int +store_metric(const char *hostname, const char *name, + sdb_metric_store_t *store, sdb_time_t last_update, + sdb_object_t *user_data) +{ + sdb_store_t *st = ST(user_data); - lookup.obj_name = svc->hostname; - host = SDB_HOST(sdb_llist_search(host_list, (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); + sdb_store_obj_t *obj = NULL; + sdb_host_t *host; + sdb_metric_t *metric; + sdb_data_t d; - if (! host) { - pthread_rwlock_unlock(&host_lock); + sdb_avltree_t *metrics; + + int status = 0; + + if ((! hostname) || (! name)) return -1; - } - lookup.obj_name = svc->_name; - old = SDB_SVC(sdb_llist_search(host->services, (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); + if (store) { + if ((store->type != NULL) != (store->id != NULL)) + return -1; + else if (! store->type) + store = NULL; + } - if (old) { - if (old->_last_update > last_update) { - sdb_log(SDB_LOG_DEBUG, "store: Cannot update service " - "'%s/%s' - value too old (%"PRIscTIME" < %"PRIscTIME")", - svc->hostname, svc->_name, last_update, - old->_last_update); - status = 1; - } - else { - old->_last_update = last_update; - } + pthread_rwlock_wrlock(&st->host_lock); + host = lookup_host(st, hostname, /* canonicalize = */ 1); + metrics = get_host_children(host, SDB_METRIC); + if (! metrics) { + sdb_log(SDB_LOG_ERR, "store: Failed to store metric '%s' - " + "host '%s' not found", name, hostname); + status = -1; } - else { - sdb_service_t *new = sdb_service_clone(svc); - if (! new) { - char errbuf[1024]; - sdb_log(SDB_LOG_ERR, "store: Failed to clone service " - "object: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); - pthread_rwlock_unlock(&host_lock); - return -1; - } - status = sdb_llist_insert_sorted(host->services, SDB_OBJ(new), - sdb_store_obj_cmp_by_name); + if (! status) + status = store_obj(STORE_OBJ(host), metrics, SDB_METRIC, + name, last_update, &obj); + sdb_object_deref(SDB_OBJ(host)); - /* pass control to the list or destroy in case of an error */ - sdb_object_deref(SDB_OBJ(new)); + if (status) { + pthread_rwlock_unlock(&st->host_lock); + return status; } - pthread_rwlock_unlock(&host_lock); + assert(obj); + metric = METRIC(obj); + + if (store) + if (store_metric_store(metric, store)) + status = -1; + pthread_rwlock_unlock(&st->host_lock); + + /* record the hostname as an attribute */ + d.type = SDB_TYPE_STRING; + d.data.string = SDB_OBJ(host)->name; + if (store_metric_attr(hostname, name, "hostname", &d, last_update, user_data)) + status = -1; return status; -} /* sdb_store_service */ +} /* store_metric */ + +static sdb_store_writer_t store_writer = { + store_host, store_service, store_metric, + store_attribute, store_service_attr, store_metric_attr, +}; + +/* + * TODO: let prepare and execute accept a store object as their user_data + * object + */ -const sdb_service_t * -sdb_store_get_service(const sdb_host_t *host, const char *name) +static sdb_object_t * +prepare_query(sdb_ast_node_t *ast, + sdb_strbuf_t __attribute__((unused)) *errbuf, + sdb_object_t __attribute__((unused)) *user_data) { - sdb_store_lookup_obj_t lookup = SDB_STORE_LOOKUP_OBJ_INIT; - sdb_service_t *svc; + return SDB_OBJ(sdb_store_query_prepare(ast)); +} /* prepare_query */ + +static int +execute_query(sdb_object_t *q, + sdb_strbuf_t *buf, sdb_strbuf_t *errbuf, + sdb_object_t __attribute__((unused)) *user_data) +{ + return sdb_store_query_execute(QUERY(q), buf, errbuf); +} /* execute_query */ + +static sdb_store_reader_t store_reader = { + prepare_query, execute_query, +}; + +/* + * public API + */ + +int +sdb_store_init(void) +{ + if (global_store) + return 0; - if ((! host) || (! name)) + global_store = ST(sdb_object_create("store", store_type)); + if (! global_store) { + sdb_log(SDB_LOG_ERR, "store: Failed to allocate store"); + return -1; + } + if (sdb_plugin_register_writer("memstore", + &store_writer, SDB_OBJ(global_store))) + return -1; + return sdb_plugin_register_reader("memstore", + &store_reader, SDB_OBJ(global_store)); +} /* sdb_store_init */ + +void +sdb_store_clear(void) +{ + if (! global_store) + return; + sdb_avltree_clear(global_store->hosts); +} /* sdb_store_clear */ + +sdb_store_obj_t * +sdb_store_get_host(const char *name) +{ + sdb_host_t *host; + + if ((! global_store) || (! name)) return NULL; - lookup.obj_name = name; - svc = SDB_SVC(sdb_llist_search(host->services, - (const sdb_object_t *)&lookup, - sdb_cmp_store_obj_with_name)); + host = lookup_host(global_store, name, /* canonicalize = */ 0); + if (! host) + return NULL; + + return STORE_OBJ(host); +} /* sdb_store_get_host */ + +sdb_store_obj_t * +sdb_store_get_child(sdb_store_obj_t *host, int type, const char *name) +{ + sdb_avltree_t *children; - if (! svc) + if ((! host) || (host->type != SDB_HOST) || (! name)) return NULL; - return svc; -} /* sdb_store_get_service */ + + children = get_host_children(HOST(host), type); + if (! children) + return NULL; + return STORE_OBJ(sdb_avltree_lookup(children, name)); +} /* sdb_store_get_child */ int -sdb_store_dump(FILE *fh) +sdb_store_get_field(sdb_store_obj_t *obj, int field, sdb_data_t *res) { - sdb_llist_iter_t *host_iter; + sdb_data_t tmp; - if (! fh) + if (! obj) return -1; - pthread_rwlock_rdlock(&host_lock); + switch (field) { + case SDB_FIELD_NAME: + tmp.type = SDB_TYPE_STRING; + tmp.data.string = strdup(SDB_OBJ(obj)->name); + if (! tmp.data.string) + return -1; + break; + case SDB_FIELD_LAST_UPDATE: + tmp.type = SDB_TYPE_DATETIME; + tmp.data.datetime = obj->last_update; + break; + case SDB_FIELD_AGE: + tmp.type = SDB_TYPE_DATETIME; + tmp.data.datetime = sdb_gettime() - obj->last_update; + break; + case SDB_FIELD_INTERVAL: + tmp.type = SDB_TYPE_DATETIME; + tmp.data.datetime = obj->interval; + break; + case SDB_FIELD_BACKEND: + if (! res) + return 0; + tmp.type = SDB_TYPE_ARRAY | SDB_TYPE_STRING; + tmp.data.array.length = obj->backends_num; + tmp.data.array.values = obj->backends; + return sdb_data_copy(res, &tmp); + case SDB_FIELD_VALUE: + if (obj->type != SDB_ATTRIBUTE) + return -1; + if (! res) + return 0; + return sdb_data_copy(res, &ATTR(obj)->value); + case SDB_FIELD_TIMESERIES: + if (obj->type != SDB_METRIC) + return -1; + tmp.type = SDB_TYPE_BOOLEAN; + tmp.data.boolean = METRIC(obj)->store.type != NULL; + default: + return -1; + } + if (res) + *res = tmp; + else + sdb_data_free_datum(&tmp); + return 0; +} /* sdb_store_get_field */ + +int +sdb_store_get_attr(sdb_store_obj_t *obj, const char *name, sdb_data_t *res, + sdb_store_matcher_t *filter) +{ + sdb_avltree_t *tree = NULL; + sdb_store_obj_t *attr; + + if ((! obj) || (! name)) + return -1; + + if (obj->type == SDB_HOST) + tree = HOST(obj)->attributes; + else if (obj->type == SDB_SERVICE) + tree = SVC(obj)->attributes; + else if (obj->type == SDB_METRIC) + tree = METRIC(obj)->attributes; - host_iter = sdb_llist_get_iter(host_list); - if (! host_iter) { - pthread_rwlock_unlock(&host_lock); + if (! tree) + return -1; + + attr = STORE_OBJ(sdb_avltree_lookup(tree, name)); + if (! attr) + return -1; + if (filter && (! sdb_store_matcher_matches(filter, attr, NULL))) { + sdb_object_deref(SDB_OBJ(attr)); return -1; } - while (sdb_llist_iter_has_next(host_iter)) { - sdb_host_t *host = SDB_HOST(sdb_llist_iter_get_next(host_iter)); - sdb_llist_iter_t *svc_iter; - sdb_llist_iter_t *attr_iter; + assert(STORE_OBJ(attr)->type == SDB_ATTRIBUTE); + if (res) + sdb_data_copy(res, &ATTR(attr)->value); + sdb_object_deref(SDB_OBJ(attr)); + return 0; +} /* sdb_store_get_attr */ + +/* TODO: sdb_store_fetch_timeseries should move into the plugin module */ - char time_str[64]; +int +sdb_store_fetch_timeseries(const char *hostname, const char *metric, + sdb_timeseries_opts_t *opts, sdb_strbuf_t *buf) +{ + sdb_avltree_t *metrics; + sdb_host_t *host; + sdb_metric_t *m; - assert(host); + sdb_timeseries_t *ts; - if (! sdb_strftime(time_str, sizeof(time_str), - "%F %T %z", host->_last_update)) - snprintf(time_str, sizeof(time_str), ""); - time_str[sizeof(time_str) - 1] = '\0'; + int status = 0; - fprintf(fh, "Host '%s' (last updated: %s):\n", - host->_name, time_str); + if ((! global_store) || (! hostname) || (! metric) || (! opts) || (! buf)) + return -1; - attr_iter = sdb_llist_get_iter(host->attributes); - if (! attr_iter) { - char errbuf[1024]; - fprintf(fh, "Failed to retrieve attributes: %s\n", - sdb_strerror(errno, errbuf, sizeof(errbuf))); - continue; - } + pthread_rwlock_rdlock(&global_store->host_lock); + host = lookup_host(global_store, hostname, /* canonicalize = */ 1); + metrics = get_host_children(host, SDB_METRIC); + sdb_object_deref(SDB_OBJ(host)); + if (! metrics) { + sdb_log(SDB_LOG_ERR, "store: Failed to fetch time-series '%s/%s' " + "- host '%s' not found", hostname, metric, hostname); + pthread_rwlock_unlock(&global_store->host_lock); + return -1; + } - while (sdb_llist_iter_has_next(attr_iter)) { - sdb_attribute_t *attr = SDB_ATTR(sdb_llist_iter_get_next(attr_iter)); - assert(attr); + m = METRIC(sdb_avltree_lookup(metrics, metric)); + if (! m) { + sdb_log(SDB_LOG_ERR, "store: Failed to fetch time-series '%s/%s' " + "- metric '%s' not found", hostname, metric, metric); + pthread_rwlock_unlock(&global_store->host_lock); + return -1; + } - if (! sdb_strftime(time_str, sizeof(time_str), - "%F %T %z", attr->_last_update)) - snprintf(time_str, sizeof(time_str), ""); - time_str[sizeof(time_str) - 1] = '\0'; + if ((! m->store.type) || (! m->store.id)) { + sdb_log(SDB_LOG_ERR, "store: Failed to fetch time-series '%s/%s' " + "- no data-store configured for the stored metric", + hostname, metric); + sdb_object_deref(SDB_OBJ(m)); + pthread_rwlock_unlock(&global_store->host_lock); + return -1; + } - fprintf(fh, "\tAttribute '%s' -> '%s' (last updated: %s)\n", - attr->_name, attr->attr_value, time_str); - } + { + char type[strlen(m->store.type) + 1]; + char id[strlen(m->store.id) + 1]; - sdb_llist_iter_destroy(attr_iter); + strncpy(type, m->store.type, sizeof(type)); + strncpy(id, m->store.id, sizeof(id)); + pthread_rwlock_unlock(&global_store->host_lock); - svc_iter = sdb_llist_get_iter(host->services); - if (! svc_iter) { - char errbuf[1024]; - fprintf(fh, "Failed to retrieve services: %s\n", - sdb_strerror(errno, errbuf, sizeof(errbuf))); - continue; + ts = sdb_plugin_fetch_timeseries(type, id, opts); + if (! ts) { + sdb_log(SDB_LOG_ERR, "store: Failed to fetch time-series '%s/%s' " + "- %s fetcher callback returned no data for '%s'", + hostname, metric, type, id); + status = -1; } + } - while (sdb_llist_iter_has_next(svc_iter)) { - sdb_service_t *svc = SDB_SVC(sdb_llist_iter_get_next(svc_iter)); - assert(svc); + ts_tojson(ts, buf); + sdb_object_deref(SDB_OBJ(m)); + sdb_timeseries_destroy(ts); + return status; +} /* sdb_store_fetch_timeseries */ - if (! sdb_strftime(time_str, sizeof(time_str), - "%F %T %z", svc->_last_update)) - snprintf(time_str, sizeof(time_str), ""); - time_str[sizeof(time_str) - 1] = '\0'; +int +sdb_store_scan(int type, sdb_store_matcher_t *m, sdb_store_matcher_t *filter, + sdb_store_lookup_cb cb, void *user_data) +{ + sdb_avltree_iter_t *host_iter = NULL; + int status = 0; + + if ((! global_store) || (! cb)) + return -1; + + if ((type != SDB_HOST) && (type != SDB_SERVICE) && (type != SDB_METRIC)) { + sdb_log(SDB_LOG_ERR, "store: Cannot scan objects of type %d", type); + return -1; + } + + pthread_rwlock_rdlock(&global_store->host_lock); + host_iter = sdb_avltree_get_iter(global_store->hosts); + if (! host_iter) + status = -1; - fprintf(fh, "\tService '%s' (last updated: %s)\n", - svc->_name, time_str); + /* has_next returns false if the iterator is NULL */ + while (sdb_avltree_iter_has_next(host_iter)) { + sdb_store_obj_t *host; + sdb_avltree_iter_t *iter = NULL; + + host = STORE_OBJ(sdb_avltree_iter_get_next(host_iter)); + assert(host); + + if (! sdb_store_matcher_matches(filter, host, NULL)) + continue; + + if (type == SDB_SERVICE) + iter = sdb_avltree_get_iter(HOST(host)->services); + else if (type == SDB_METRIC) + iter = sdb_avltree_get_iter(HOST(host)->metrics); + + if (iter) { + while (sdb_avltree_iter_has_next(iter)) { + sdb_store_obj_t *obj; + obj = STORE_OBJ(sdb_avltree_iter_get_next(iter)); + assert(obj); + + if (sdb_store_matcher_matches(m, obj, filter)) { + if (cb(obj, filter, user_data)) { + sdb_log(SDB_LOG_ERR, "store: Callback returned " + "an error while scanning"); + status = -1; + break; + } + } + } + } + else if (sdb_store_matcher_matches(m, host, filter)) { + if (cb(host, filter, user_data)) { + sdb_log(SDB_LOG_ERR, "store: Callback returned " + "an error while scanning"); + status = -1; + } } - sdb_llist_iter_destroy(svc_iter); + sdb_avltree_iter_destroy(iter); + if (status) + break; } - sdb_llist_iter_destroy(host_iter); - pthread_rwlock_unlock(&host_lock); - return 0; -} /* sdb_store_dump */ + sdb_avltree_iter_destroy(host_iter); + pthread_rwlock_unlock(&global_store->host_lock); + return status; +} /* sdb_store_scan */ /* vim: set tw=78 sw=4 ts=4 noexpandtab : */