2017-10-15 22:02:43 +02:00
|
|
|
/** Node-type for InfluxDB.
|
|
|
|
*
|
|
|
|
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
|
2020-01-20 17:17:00 +01:00
|
|
|
* @copyright 2014-2020, Institute for Automation of Complex Power Systems, EONERC
|
2017-10-15 22:02:43 +02:00
|
|
|
* @license GNU General Public License (version 3)
|
|
|
|
*
|
|
|
|
* VILLASnode
|
|
|
|
*
|
|
|
|
* This program is free software: you can redistribute it and/or modify
|
|
|
|
* it under the terms of the GNU General Public License as published by
|
|
|
|
* the Free Software Foundation, either version 3 of the License, or
|
|
|
|
* any later version.
|
|
|
|
*
|
|
|
|
* This program is distributed in the hope that it will be useful,
|
|
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
|
|
* GNU General Public License for more details.
|
|
|
|
*
|
|
|
|
* You should have received a copy of the GNU General Public License
|
|
|
|
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
*********************************************************************************/
|
|
|
|
|
2019-06-23 16:57:00 +02:00
|
|
|
#include <cstring>
|
|
|
|
#include <cinttypes>
|
2018-08-06 11:24:49 +02:00
|
|
|
#include <sys/types.h>
|
|
|
|
#include <sys/socket.h>
|
|
|
|
#include <netdb.h>
|
2017-11-02 13:08:12 +01:00
|
|
|
|
2017-12-09 02:19:28 +08:00
|
|
|
#include <villas/node.h>
|
|
|
|
#include <villas/plugin.h>
|
2018-05-24 09:23:17 +02:00
|
|
|
#include <villas/signal.h>
|
2018-08-23 17:31:01 +02:00
|
|
|
#include <villas/node/config.h>
|
2019-04-23 00:12:31 +02:00
|
|
|
#include <villas/nodes/influxdb.hpp>
|
2017-12-09 02:19:28 +08:00
|
|
|
#include <villas/memory.h>
|
2017-10-15 22:02:43 +02:00
|
|
|
|
2019-06-04 16:55:38 +02:00
|
|
|
using namespace villas::utils;
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int influxdb_parse(struct vnode *n, json_t *json)
|
2017-10-15 22:02:43 +02:00
|
|
|
{
|
2017-10-18 15:39:53 +02:00
|
|
|
struct influxdb *i = (struct influxdb *) n->_vd;
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
json_error_t err;
|
|
|
|
int ret;
|
|
|
|
|
2019-02-18 01:09:33 +01:00
|
|
|
char *tmp, *host, *port, *lasts;
|
2017-10-15 22:02:43 +02:00
|
|
|
const char *server, *key;
|
|
|
|
|
2019-03-31 22:59:38 +02:00
|
|
|
ret = json_unpack_ex(json, &err, 0, "{ s: s, s: s }",
|
2017-10-15 22:02:43 +02:00
|
|
|
"server", &server,
|
2018-05-24 09:23:17 +02:00
|
|
|
"key", &key
|
2017-10-15 22:02:43 +02:00
|
|
|
);
|
|
|
|
if (ret)
|
|
|
|
jerror(&err, "Failed to parse configuration of node %s", node_name(n));
|
|
|
|
|
|
|
|
tmp = strdup(server);
|
|
|
|
|
2019-02-18 01:09:33 +01:00
|
|
|
host = strtok_r(tmp, ":", &lasts);
|
2019-04-22 23:45:38 +02:00
|
|
|
port = strtok_r(nullptr, "", &lasts);
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
i->key = strdup(key);
|
|
|
|
i->host = strdup(host);
|
|
|
|
i->port = strdup(port ? port : "8089");
|
|
|
|
|
|
|
|
free(tmp);
|
|
|
|
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int influxdb_open(struct vnode *n)
|
2017-10-15 22:02:43 +02:00
|
|
|
{
|
|
|
|
int ret;
|
2017-10-18 15:39:53 +02:00
|
|
|
struct influxdb *i = (struct influxdb *) n->_vd;
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
struct addrinfo hints, *servinfo, *p;
|
|
|
|
|
|
|
|
memset(&hints, 0, sizeof hints);
|
|
|
|
hints.ai_family = AF_UNSPEC;
|
|
|
|
hints.ai_socktype = SOCK_DGRAM;
|
|
|
|
|
|
|
|
ret = getaddrinfo(i->host, i->port, &hints, &servinfo);
|
|
|
|
if (ret)
|
|
|
|
error("Failed to lookup server: %s", gai_strerror(ret));
|
|
|
|
|
|
|
|
/* Loop through all the results and connect to the first we can */
|
2019-04-22 23:45:38 +02:00
|
|
|
for (p = servinfo; p != nullptr; p = p->ai_next) {
|
2017-10-15 22:02:43 +02:00
|
|
|
i->sd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
|
|
|
|
if (i->sd == -1) {
|
|
|
|
serror("socket");
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
|
|
|
|
ret = connect(i->sd, p->ai_addr, p->ai_addrlen);
|
|
|
|
if (ret == -1) {
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("connect");
|
2017-10-15 22:02:43 +02:00
|
|
|
close(i->sd);
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
|
|
|
|
/* If we get here, we must have connected successfully */
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
return p ? 0 : -1;
|
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int influxdb_close(struct vnode *n)
|
2017-10-15 22:02:43 +02:00
|
|
|
{
|
2017-10-18 15:39:53 +02:00
|
|
|
struct influxdb *i = (struct influxdb *) n->_vd;
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
close(i->sd);
|
|
|
|
|
2018-05-24 09:23:17 +02:00
|
|
|
if (i->host)
|
|
|
|
free(i->host);
|
|
|
|
if (i->port)
|
|
|
|
free(i->port);
|
|
|
|
if (i->key)
|
|
|
|
free(i->key);
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
int influxdb_write(struct vnode *n, struct sample *smps[], unsigned cnt, unsigned *release)
|
2017-10-15 22:02:43 +02:00
|
|
|
{
|
2017-10-18 15:39:53 +02:00
|
|
|
struct influxdb *i = (struct influxdb *) n->_vd;
|
2017-10-15 22:02:43 +02:00
|
|
|
|
2019-04-22 23:45:38 +02:00
|
|
|
char *buf = nullptr;
|
2017-10-15 22:02:43 +02:00
|
|
|
ssize_t sentlen, buflen;
|
|
|
|
|
2019-04-22 23:43:46 +02:00
|
|
|
for (unsigned k = 0; k < cnt; k++) {
|
2018-08-20 18:27:23 +02:00
|
|
|
struct sample *smp = smps[k];
|
|
|
|
|
2017-10-15 22:02:43 +02:00
|
|
|
/* Key */
|
|
|
|
strcatf(&buf, "%s", i->key);
|
|
|
|
|
|
|
|
/* Fields */
|
2019-04-22 23:43:46 +02:00
|
|
|
for (unsigned j = 0; j < smp->length; j++) {
|
2019-01-07 10:28:55 +01:00
|
|
|
struct signal *sig = (struct signal *) vlist_at(smp->signals, j);
|
2020-03-10 22:25:07 +01:00
|
|
|
union signal_data *data = &smp->data[j];
|
2018-08-20 18:27:23 +02:00
|
|
|
|
2019-03-09 13:34:51 +01:00
|
|
|
if (
|
2019-06-23 16:13:23 +02:00
|
|
|
sig->type != SignalType::BOOLEAN &&
|
2020-01-07 10:51:59 +01:00
|
|
|
sig->type != SignalType::FLOAT &&
|
2020-01-09 10:35:06 +01:00
|
|
|
sig->type != SignalType::INTEGER &&
|
|
|
|
sig->type != SignalType::COMPLEX
|
2019-03-09 13:34:51 +01:00
|
|
|
) {
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Unsupported signal format for node %s. Skipping", node_name(n));
|
2018-08-20 18:27:23 +02:00
|
|
|
continue;
|
|
|
|
}
|
|
|
|
|
2017-10-15 22:02:43 +02:00
|
|
|
strcatf(&buf, "%c", j == 0 ? ' ' : ',');
|
|
|
|
|
2019-01-07 10:28:55 +01:00
|
|
|
sig = (struct signal *) vlist_at(smp->signals, j);
|
2018-08-20 18:27:23 +02:00
|
|
|
if (!sig)
|
|
|
|
return -1;
|
2017-10-15 22:02:43 +02:00
|
|
|
|
2020-01-09 10:35:06 +01:00
|
|
|
char name[32];
|
2018-08-20 18:27:23 +02:00
|
|
|
if (sig->name)
|
2020-09-30 10:56:58 +02:00
|
|
|
strncpy(name, sig->name, sizeof(name)-1);
|
2017-10-15 22:02:43 +02:00
|
|
|
else
|
2020-09-11 14:57:05 +02:00
|
|
|
snprintf(name, sizeof(name), "value%u", j);
|
2017-10-15 22:02:43 +02:00
|
|
|
|
2020-01-09 10:35:06 +01:00
|
|
|
if (sig->type == SignalType::COMPLEX) {
|
|
|
|
strcatf(&buf, "%s_re=%f, %s_im=%f",
|
|
|
|
name, std::real(data->z),
|
|
|
|
name, std::imag(data->z)
|
|
|
|
);
|
|
|
|
}
|
|
|
|
else {
|
|
|
|
strcatf(&buf, "%s=", name);
|
|
|
|
|
|
|
|
switch (sig->type) {
|
|
|
|
case SignalType::BOOLEAN:
|
|
|
|
strcatf(&buf, "%s", data->b ? "true" : "false");
|
|
|
|
break;
|
2018-08-20 18:27:23 +02:00
|
|
|
|
2020-01-09 10:35:06 +01:00
|
|
|
case SignalType::FLOAT:
|
|
|
|
strcatf(&buf, "%f", data->f);
|
|
|
|
break;
|
2018-08-20 18:27:23 +02:00
|
|
|
|
2020-01-09 10:35:06 +01:00
|
|
|
case SignalType::INTEGER:
|
|
|
|
strcatf(&buf, "%" PRIi64, data->i);
|
|
|
|
break;
|
2018-08-20 18:27:23 +02:00
|
|
|
|
2020-01-09 10:35:06 +01:00
|
|
|
default: { }
|
|
|
|
}
|
2017-10-15 22:02:43 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Timestamp */
|
2021-01-06 15:11:26 +01:00
|
|
|
strcatf(&buf, " %lld%09lld\n", (long long) smp->ts.origin.tv_sec,
|
|
|
|
(long long) smp->ts.origin.tv_nsec);
|
2017-10-15 22:02:43 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
buflen = strlen(buf) + 1;
|
|
|
|
sentlen = send(i->sd, buf, buflen, 0);
|
|
|
|
if (sentlen < 0)
|
|
|
|
return -1;
|
|
|
|
else if (sentlen < buflen)
|
2018-10-21 21:36:08 +01:00
|
|
|
warning("Partial sent");
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
free(buf);
|
|
|
|
|
2018-07-11 18:14:29 +02:00
|
|
|
return cnt;
|
2017-10-15 22:02:43 +02:00
|
|
|
}
|
|
|
|
|
2020-08-25 21:00:52 +02:00
|
|
|
char * influxdb_print(struct vnode *n)
|
2017-10-15 22:02:43 +02:00
|
|
|
{
|
2017-10-18 15:39:53 +02:00
|
|
|
struct influxdb *i = (struct influxdb *) n->_vd;
|
2019-04-22 23:45:38 +02:00
|
|
|
char *buf = nullptr;
|
2017-10-15 22:02:43 +02:00
|
|
|
|
2018-05-24 09:23:17 +02:00
|
|
|
strcatf(&buf, "host=%s, port=%s, key=%s", i->host, i->port, i->key);
|
2017-10-15 22:02:43 +02:00
|
|
|
|
|
|
|
return buf;
|
|
|
|
}
|
|
|
|
|
2019-04-23 00:36:06 +02:00
|
|
|
static struct plugin p;
|
|
|
|
|
|
|
|
__attribute__((constructor(110)))
|
|
|
|
static void register_plugin() {
|
|
|
|
p.name = "influxdb";
|
|
|
|
p.description = "Write results to InfluxDB";
|
2019-06-23 16:13:23 +02:00
|
|
|
p.type = PluginType::NODE;
|
|
|
|
p.node.instances.state = State::DESTROYED;
|
2019-04-23 00:36:06 +02:00
|
|
|
p.node.vectorize = 0;
|
|
|
|
p.node.size = sizeof(struct influxdb);
|
|
|
|
p.node.parse = influxdb_parse;
|
|
|
|
p.node.print = influxdb_print;
|
|
|
|
p.node.start = influxdb_open;
|
|
|
|
p.node.stop = influxdb_close;
|
|
|
|
p.node.write = influxdb_write;
|
|
|
|
|
2020-09-10 11:11:42 +02:00
|
|
|
int ret = vlist_init(&p.node.instances);
|
|
|
|
if (!ret)
|
|
|
|
vlist_init_and_push(&plugins, &p);
|
2019-04-23 00:36:06 +02:00
|
|
|
}
|
2017-10-15 22:02:43 +02:00
|
|
|
|
2019-04-23 00:36:06 +02:00
|
|
|
__attribute__((destructor(110)))
|
|
|
|
static void deregister_plugin() {
|
2020-06-16 02:35:34 +02:00
|
|
|
vlist_remove_all(&plugins, &p);
|
2019-06-04 16:55:38 +02:00
|
|
|
}
|