Code

timeseries: Add support for fetching specific data-sources.
[sysdb.git] / src / plugins / timeseries / rrdtool.c
index db24903f5513785852a22748f05de88b1a4b0f91..b70aa8c9edab94c1f19121195ba8f843d71ae557 100644 (file)
@@ -1,6 +1,6 @@
 /*
  * SysDB - src/plugins/timeseries/rrdtool.c
- * Copyright (C) 2014 Sebastian 'tokkee' Harl <sh@tokkee.org>
+ * Copyright (C) 2014-2016 Sebastian 'tokkee' Harl <sh@tokkee.org>
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -37,6 +37,8 @@
 
 #include <errno.h>
 #include <stdlib.h>
+#include <string.h>
+#include <strings.h>
 #include <rrd.h>
 #ifdef HAVE_RRD_CLIENT_H
 #      include <rrd_client.h>
@@ -47,12 +49,171 @@ SDB_PLUGIN_MAGIC;
 /* Current versions of RRDtool do not support multiple RRDCacheD client
  * connections. Use this to guard against multiple configured RRDCacheD
  * instances. */
-static _Bool rrdcached_in_use = 0;
+static bool rrdcached_in_use = 0;
+
+static bool
+rrdcached_connect(char *addr)
+{
+#ifdef HAVE_RRD_CLIENT_H
+       rrd_clear_error();
+       if (! rrdc_is_connected(addr)) {
+               if (rrdc_connect(addr)) {
+                       sdb_log(SDB_LOG_ERR, "Failed to connectd to RRDCacheD at %s: %s",
+                                       addr, rrd_get_error());
+                       return 0;
+               }
+       }
+#else
+       sdb_log(SDB_LOG_ERR, "Callback called with RRDCacheD address "
+                       "but your build of SysDB does not support that");
+       return 0;
+#endif
+       return 1;
+} /* rrdcached_connect */
 
 /*
  * plugin API
  */
 
+static sdb_timeseries_info_t *
+sdb_rrd_describe(const char *id, sdb_object_t *user_data)
+{
+       rrd_info_t *info, *iter;
+       char filename[strlen(id) + 1];
+       sdb_timeseries_info_t *ts_info;
+
+       strncpy(filename, id, sizeof(filename));
+
+       if (user_data) {
+               /* -> use RRDCacheD */
+               char *addr = SDB_OBJ_WRAPPER(user_data)->data;
+
+               if (! rrdcached_connect(addr))
+                       return NULL;
+
+#ifdef HAVE_RRD_CLIENT_H
+               /* TODO: detect and use rrdc_info if possible */
+               sdb_log(SDB_LOG_ERR, "DESCRIBE not yet supported via RRDCacheD");
+               return NULL;
+#endif
+       }
+       else {
+               rrd_clear_error();
+               info = rrd_info_r(filename);
+       }
+       if (! info) {
+               sdb_log(SDB_LOG_ERR, "Failed to extract header information from '%s': %s",
+                               filename, rrd_get_error());
+               return NULL;
+       }
+
+       ts_info = calloc(1, sizeof(*ts_info));
+       if (! ts_info) {
+               sdb_log(SDB_LOG_ERR, "Failed to allocate memory");
+               rrd_info_free(info);
+               return NULL;
+       }
+
+       for (iter = info; iter != NULL; iter = iter->next) {
+               size_t len, n, m;
+               char *ds_name;
+               char **tmp;
+
+               /* Parse the DS name. The raw value is not exposed via the rrd_info
+                * interface. */
+               n = strlen("ds[");
+               if (strncmp(iter->key, "ds[", n))
+                       continue;
+
+               len = strlen(iter->key);
+               m = strlen("].index");
+               if ((len < m) || strcmp(iter->key + len - m, "].index"))
+                       continue;
+
+               ds_name = iter->key + n;
+               len -= n;
+               ds_name[len - m] = '\0';
+
+               /* Append the new datum. */
+               tmp = realloc(ts_info->data_names,
+                               (ts_info->data_names_len + 1) * sizeof(*ts_info->data_names));
+               if (! tmp) {
+                       sdb_log(SDB_LOG_ERR, "Failed to allocate memory");
+                       sdb_timeseries_info_destroy(ts_info);
+                       rrd_info_free(info);
+                       return NULL;
+               }
+
+               ts_info->data_names = tmp;
+               ts_info->data_names[ts_info->data_names_len] = strdup(ds_name);
+               if (! ts_info->data_names[ts_info->data_names_len]) {
+                       sdb_log(SDB_LOG_ERR, "Failed to allocate memory");
+                       sdb_timeseries_info_destroy(ts_info);
+                       rrd_info_free(info);
+                       return NULL;
+               }
+               ts_info->data_names_len++;
+       }
+       rrd_info_free(info);
+
+       return ts_info;
+} /* sdb_rrd_describe */
+
+static int
+copy_data(sdb_timeseries_t *ts, rrd_value_t *data, time_t step,
+               size_t ds_cnt, char **ds_names)
+{
+       time_t start = SDB_TIME_TO_SECS(ts->start);
+       time_t end = SDB_TIME_TO_SECS(ts->end);
+       time_t t;
+
+       ssize_t ds_target[ds_cnt];
+       size_t i, j;
+
+       /* determine the target index of each data-source,
+        * or -1 if the data-source isn't wanted */
+       for (i = 0; i < ds_cnt; i++) {
+               ds_target[i] = -1;
+               for (j = 0; j < ts->data_names_len; j++) {
+                       if (!strcmp(ds_names[i], ts->data_names[j])) {
+                               ds_target[i] = j;
+                               break;
+                       }
+               }
+       }
+
+       /* check if any wanted data-sources is missing */
+       for (i = 0; i < ts->data_names_len; i++) {
+               bool found = false;
+
+               for (j = 0; j < ds_cnt; j++) {
+                       if (!strcmp(ts->data_names[i], ds_names[j])) {
+                               found = true;
+                               break;
+                       }
+               }
+               if (!found) {
+                       sdb_log(SDB_LOG_ERR, "Requested data-source '%s' not found",
+                                       ts->data_names[i]);
+                       return -1;
+               }
+       }
+
+       for (t = start; t <= end; t += step) {
+               i = (size_t)(t - start) / (size_t)step;
+
+               for (j = 0; j < ds_cnt; ++j) {
+                       if (ds_target[j] >= 0) {
+                               size_t x = (size_t)ds_target[j];
+                               ts->data[x][i].timestamp = SECS_TO_SDB_TIME(t);
+                               ts->data[x][i].value = *data;
+                       }
+                       ++data;
+               }
+       }
+       return 0;
+}
+
 static sdb_timeseries_t *
 sdb_rrd_fetch(const char *id, sdb_timeseries_opts_t *opts,
                sdb_object_t *user_data)
@@ -61,39 +222,26 @@ sdb_rrd_fetch(const char *id, sdb_timeseries_opts_t *opts,
 
        time_t start = (time_t)SDB_TIME_TO_SECS(opts->start);
        time_t end = (time_t)SDB_TIME_TO_SECS(opts->end);
-       time_t t;
 
        unsigned long step = 0;
        unsigned long ds_cnt = 0;
        unsigned long val_cnt = 0;
        char **ds_namv = NULL;
-       rrd_value_t *data = NULL, *data_ptr;
+       rrd_value_t *data = NULL;
 
        if (user_data) {
-#ifdef HAVE_RRD_CLIENT_H
                /* -> use RRDCacheD */
                char *addr = SDB_OBJ_WRAPPER(user_data)->data;
 
-               rrd_clear_error();
-               if (! rrdc_is_connected(addr)) {
-                       if (rrdc_connect(addr)) {
-                               sdb_log(SDB_LOG_ERR, "timeseries::rrdtool: Failed to "
-                                               "connectd to RRDCacheD at %s: %s",
-                                               addr, rrd_get_error());
-                               return NULL;
-                       }
-               }
+               if (! rrdcached_connect(addr))
+                       return NULL;
 
+#ifdef HAVE_RRD_CLIENT_H
                if (rrdc_flush(id)) {
-                       sdb_log(SDB_LOG_ERR, "timeseries::rrdtool: Failed to flush "
-                                       "'%s' through RRDCacheD: %s", id, rrd_get_error());
+                       sdb_log(SDB_LOG_ERR, "Failed to flush '%s' through RRDCacheD: %s",
+                                       id, rrd_get_error());
                        return NULL;
                }
-#else
-               sdb_log(SDB_LOG_ERR, "timeseries::rrdtool: Callback called with "
-                               "RRDCacheD address but your build of SysDB does not support "
-                               "that");
-               return NULL;
 #endif
        }
 
@@ -106,23 +254,31 @@ sdb_rrd_fetch(const char *id, sdb_timeseries_opts_t *opts,
                rrd_freemem(data); \
        } while (0)
 
+       /* limit to about 1000 data-points for now
+        * TODO: make this configurable */
+       step = (end - start) / 1000;
+
        if (rrd_fetch_r(id, "AVERAGE", &start, &end, &step,
                                &ds_cnt, &ds_namv, &data)) {
                char errbuf[1024];
                sdb_strerror(errno, errbuf, sizeof(errbuf));
-               sdb_log(SDB_LOG_ERR, "rrdtool plugin: Failed to fetch data "
-                               "from %s: %s", id, errbuf);
+               sdb_log(SDB_LOG_ERR, "Failed to fetch data from %s: %s", id, errbuf);
                return NULL;
        }
 
        val_cnt = (unsigned long)(end - start) / step;
 
-       ts = sdb_timeseries_create(ds_cnt, (const char * const *)ds_namv, val_cnt);
+       /* RRDtool does not support fetching specific data-sources, so we'll have
+        * to filter the requested ones after fetching them all */
+       if (opts->data_names && opts->data_names_len)
+               ts = sdb_timeseries_create(opts->data_names_len,
+                               (const char * const *)opts->data_names, val_cnt);
+       else
+               ts = sdb_timeseries_create(ds_cnt, (const char * const *)ds_namv, val_cnt);
        if (! ts) {
                char errbuf[1024];
                sdb_strerror(errno, errbuf, sizeof(errbuf));
-               sdb_log(SDB_LOG_ERR, "rrdtool plugin: Failed to allocate "
-                               "time-series object: %s", errbuf);
+               sdb_log(SDB_LOG_ERR, "Failed to allocate time-series object: %s", errbuf);
                FREE_RRD_DATA();
                return NULL;
        }
@@ -130,23 +286,20 @@ sdb_rrd_fetch(const char *id, sdb_timeseries_opts_t *opts,
        ts->start = SECS_TO_SDB_TIME(start + (time_t)step);
        ts->end = SECS_TO_SDB_TIME(end);
 
-       data_ptr = data;
-       for (t = start + (time_t)step; t <= end; t += (time_t)step) {
-               unsigned long i, j;
-
-               i = (unsigned long)(t - start) / step - 1;
-
-               for (j = 0; j < ds_cnt; ++j) {
-                       ts->data[j][i].timestamp = SECS_TO_SDB_TIME(t);
-                       ts->data[j][i].value = *data_ptr;
-                       ++data_ptr;
-               }
+       if (copy_data(ts, data, (time_t)step, (size_t)ds_cnt, ds_namv) < 0) {
+               FREE_RRD_DATA();
+               sdb_timeseries_destroy(ts);
+               return NULL;
        }
 
        FREE_RRD_DATA();
        return ts;
 } /* sdb_rrd_fetch */
 
+static sdb_timeseries_fetcher_t fetcher_impl = {
+       sdb_rrd_describe, sdb_rrd_fetch,
+};
+
 static int
 sdb_rrdcached_shutdown(sdb_object_t __attribute__((unused)) *user_data)
 {
@@ -163,50 +316,45 @@ sdb_rrd_config_rrdcached(oconfig_item_t *ci)
        char *addr = NULL;
 
        if (rrdcached_in_use) {
-               sdb_log(SDB_LOG_ERR, "timeseries::rrdtool: RRDCacheD does "
-                               "not support multiple connections");
+               sdb_log(SDB_LOG_ERR, "RRDCacheD does not support multiple connections");
                return -1;
        }
 
 #ifndef HAVE_RRD_CLIENT_H
-       sdb_log(SDB_LOG_ERR, "timeseries::rrdtool: RRDCacheD client "
-                       "support not available in your SysDB build");
+       sdb_log(SDB_LOG_ERR, "RRDCacheD client support not available in your SysDB build");
        return -1;
 #else
        if (oconfig_get_string(ci, &addr)) {
-               sdb_log(SDB_LOG_ERR, "timeseries::unixsock: RRDCacheD requires "
-                               "a single string argument\n\tUsage <RRDCacheD ADDR>");
+               sdb_log(SDB_LOG_ERR, "RRDCacheD requires a single string argument\n"
+                               "\tUsage <RRDCacheD ADDR>");
                return -1;
        }
        if ((*addr != '/') && strncmp(addr, "unix:", strlen("unix:"))) {
                /* XXX: add (optional) support for rrdc_fetch if available */
-               sdb_log(SDB_LOG_ERR, "timeseries::unixsock: RRDCacheD only "
-                               "supports local (UNIX socket) addresses");
+               sdb_log(SDB_LOG_ERR, "RRDCacheD only supports local (UNIX socket) addresses");
                return -1;
        }
 
        addr = strdup(addr);
        if (! addr) {
                char errbuf[1024];
-               sdb_log(SDB_LOG_ERR, "timeseries::unixsock: Failed to duplicate "
-                               "string: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
+               sdb_log(SDB_LOG_ERR, "Failed to duplicate string: %s",
+                               sdb_strerror(errno, errbuf, sizeof(errbuf)));
                return -1;
        }
        if (ci->children_num)
-               sdb_log(SDB_LOG_WARNING, "timeseries::unixsock: RRDCacheD does "
-                               "not support any child config options");
+               sdb_log(SDB_LOG_WARNING, "RRDCacheD does not support any child config options");
 
        ud = sdb_object_create_wrapper("rrdcached-addr", addr, free);
        if (! ud) {
                char errbuf[1024];
-               sdb_log(SDB_LOG_ERR, "timeseries::unixsock: Failed to create "
-                               "user-data object: %s",
+               sdb_log(SDB_LOG_ERR, "Failed to create user-data object: %s",
                                sdb_strerror(errno, errbuf, sizeof(errbuf)));
                free(addr);
                return -1;
        }
 
-       sdb_plugin_register_ts_fetcher("rrdcached", sdb_rrd_fetch, ud);
+       sdb_plugin_register_timeseries_fetcher("rrdcached", &fetcher_impl, ud);
        sdb_plugin_register_shutdown("rrdcached", sdb_rrdcached_shutdown, NULL);
        sdb_object_deref(ud);
        rrdcached_in_use = 1;
@@ -230,8 +378,7 @@ sdb_rrd_config(oconfig_item_t *ci)
                if (! strcasecmp(child->key, "RRDCacheD"))
                        sdb_rrd_config_rrdcached(child);
                else
-                       sdb_log(SDB_LOG_WARNING, "timeseries::rrdtool: Ignoring "
-                                       "unknown config option '%s'.", child->key);
+                       sdb_log(SDB_LOG_WARNING, "Ignoring unknown config option '%s'.", child->key);
        }
        return 0;
 } /* sdb_rrd_config */
@@ -247,7 +394,7 @@ sdb_module_init(sdb_plugin_info_t *info)
        sdb_plugin_set_info(info, SDB_PLUGIN_INFO_VERSION, SDB_VERSION);
        sdb_plugin_set_info(info, SDB_PLUGIN_INFO_PLUGIN_VERSION, SDB_VERSION);
 
-       sdb_plugin_register_ts_fetcher("rrdtool", sdb_rrd_fetch, NULL);
+       sdb_plugin_register_timeseries_fetcher("rrdtool", &fetcher_impl, NULL);
        sdb_plugin_register_config(sdb_rrd_config);
        return 0;
 } /* sdb_module_init */