diff --git a/README.md b/README.md index 670df5a9..9c04f58c 100644 --- a/README.md +++ b/README.md @@ -530,6 +530,7 @@ E.g. -X "n=doorbell,m=OOK_PWM,s=400,l=800,r=7000,g=1000,match={24}0xa9878c,repea Specify InfluxDB 2.0 server with e.g. -F "influx://localhost:9999/api/v2/write?org=&bucket=,token=" Specify InfluxDB 1.x server with e.g. -F "influx://localhost:8086/write?db=&p=&u=" Additional parameter -M time:unix:usec:utc for correct timestamps in InfluxDB recommended + Additional parameter `metric` supports expanding keys (see the MQTT section just above for details). [-F syslog[:[//]host[:port] (default: localhost:514) Specify host/port for syslog with e.g. -F syslog:127.0.0.1:1514 [-F trigger:/path/to/file] diff --git a/conf/rtl_433.example.conf b/conf/rtl_433.example.conf index 75657f16..4c818987 100644 --- a/conf/rtl_433.example.conf +++ b/conf/rtl_433.example.conf @@ -203,6 +203,7 @@ report_meta protocol # Specify InfluxDB 2.0 server with e.g. -F "influx://localhost:9999/api/v2/write?org=&bucket=,token=" # Specify InfluxDB 1.x server with e.g. -F "influx://localhost:8086/write?db=&p=&u=" # Additional parameter -M time:unix:usec:utc for correct timestamps in InfluxDB recommended +# Additional parameter `metric` supports expanding keys (see the MQTT section just above for details). # [-F syslog[:[//]host[:port] (default: localhost:514) # Specify host/port for syslog with e.g. -F syslog:127.0.0.1:1514 # [-F trigger:/path/to/file] diff --git a/include/string_expand.h b/include/string_expand.h new file mode 100644 index 00000000..c3cd888a --- /dev/null +++ b/include/string_expand.h @@ -0,0 +1,16 @@ +/** @file + String formatting a-la MQTT topic + + Copyright (C) 2019 Christian Zuckschwerdt + + 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 2 of the License, or + (at your option) any later version. +*/ + +#include "data.h" + +typedef char *(expand_string_sanitizer)(char *, char *); + +char *expand_topic_string(char *topic, char const *format, data_t *data, char const *hostname, expand_string_sanitizer sanitizer); diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index d0beba82..ac76d04b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -46,6 +46,7 @@ add_library(r_433 STATIC samp_grab.c sdr.c sigmf.c + string_expand.c term_ctl.c write_sigrok.c devices/abmt.c diff --git a/src/output_influx.c b/src/output_influx.c index 816300ed..2f91af5a 100644 --- a/src/output_influx.c +++ b/src/output_influx.c @@ -17,6 +17,7 @@ #include "logger.h" #include "fatal.h" #include "r_util.h" +#include "string_expand.h" #include #include @@ -38,6 +39,7 @@ typedef struct { char hostname[64]; char url[400]; char extra_headers[150]; + char metric_format[100]; tls_opts_t tls_opts; int databufidxfill; struct mbuf databufs[2]; @@ -120,10 +122,11 @@ static void influx_client_timer(struct mg_connection *nc, int ev, void *ev_data) } } -static influx_client_t *influx_client_init(influx_client_t *ctx, char const *url, char const *token) +static influx_client_t *influx_client_init(influx_client_t *ctx, char const *url, char const *token, const char *metric_format) { snprintf(ctx->url, sizeof(ctx->url), "%s", url); snprintf(ctx->extra_headers, sizeof (ctx->extra_headers), "Authorization: Token %s\r\n", token); + snprintf(ctx->metric_format, sizeof(ctx->metric_format), "%s", metric_format); return ctx; } @@ -339,7 +342,12 @@ static void R_API_CALLCONV print_influx_data(data_output_t *output, data_t *data data_time = d; } - if (!data_model) { + if (influx->metric_format[0]) { + char metric_name[1000]; + mbuf_reserve(buf, sizeof (metric_name) + 1000); + expand_topic_string(metric_name, influx->metric_format, data, influx->hostname, influx_sanitize_tag); + mbuf_snprintf(buf, "%s", metric_name); + } else if (!data_model) { // data isn't from device (maybe report for example) // use hostname for measurement @@ -347,7 +355,7 @@ static void R_API_CALLCONV print_influx_data(data_output_t *output, data_t *data mbuf_snprintf(buf, "rtl_433_%s", influx->hostname); } else { - // use model for measurement + // default: use "model" for metric name mbuf_reserve(buf, 1000); str = &buf->buf[buf->len]; @@ -357,8 +365,17 @@ static void R_API_CALLCONV print_influx_data(data_output_t *output, data_t *data // write tags while (data) { - if (!strcmp(data->key, "model") - || !strcmp(data->key, "time")) { + if ((!strcmp(data->key, "model") && strstr(influx->metric_format, "[model")) + || (!strcmp(data->key, "type") && strstr(influx->metric_format, "[type")) + || (!strcmp(data->key, "subtype") && strstr(influx->metric_format, "[subtype")) + || (!strcmp(data->key, "hostname") && strstr(influx->metric_format, "[hostname")) + || (!strcmp(data->key, "channel") && strstr(influx->metric_format, "[channel")) + || (!strcmp(data->key, "id") && strstr(influx->metric_format, "[id")) + || (!strcmp(data->key, "protocol") && strstr(influx->metric_format, "[protocol"))) { + // this field is already encoded in the metric name -> skip + } else if (!strcmp(data->key, "model") && !influx->metric_format[0]) { + // non-configurable metric format uses "model" for metric name -> skip + } else if (!strcmp(data->key, "time")) { // skip } else if (!strcmp(data->key, "type") @@ -486,6 +503,7 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts) influx_sanitize_tag(influx->hostname, NULL); char *token = NULL; + char *metric_format = NULL; // param/opts starts with URL if (!opts) { @@ -527,6 +545,8 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts) continue; else if (!strcasecmp(key, "t") || !strcasecmp(key, "token")) token = val; + else if (!strcasecmp(key, "metric")) + metric_format = val; else if (!tls_param(&influx->tls_opts, key, val)) { // ok } @@ -543,7 +563,9 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts) influx->output.print_int = print_influx_int; influx->output.output_free = data_output_influx_free; - print_logf(LOG_CRITICAL, "InfluxDB", "Publishing data to InfluxDB (%s)", url); + print_logf(LOG_CRITICAL, "InfluxDB", "Publishing data to InfluxDB (%s, %s%s)", url, + metric_format ? "dynamic metric " : "static metric", + metric_format ? metric_format : ""); influx->mgr = mgr; @@ -551,7 +573,7 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts) struct mg_add_sock_opts timer_opts = {.user_data = influx}; influx->timer = mg_add_sock_opt(mgr, INVALID_SOCKET, influx_client_timer, timer_opts); - influx_client_init(influx, url, token); + influx_client_init(influx, url, token, metric_format); return (struct data_output *)influx; } diff --git a/src/output_mqtt.c b/src/output_mqtt.c index d5744f9b..670c9e0c 100644 --- a/src/output_mqtt.c +++ b/src/output_mqtt.c @@ -16,6 +16,7 @@ #include "logger.h" #include "fatal.h" #include "r_util.h" +#include "string_expand.h" #include #include @@ -447,9 +448,9 @@ static void mqtt_client_free(mqtt_client_t *ctx) /* Helper */ /// clean the topic inplace to [-.A-Za-z0-9], esp. not whitespace, +, #, /, $ -static char *mqtt_sanitize_topic(char *topic) +static char *mqtt_sanitize_topic(char *topic, char *end) { - for (char *p = topic; *p; ++p) + for (char *p = topic; p < end && *p; ++p) if (*p != '-' && *p != '.' && (*p < 'A' || *p > 'Z') && (*p < 'a' || *p > 'z') && (*p < '0' || *p > '9')) *p = '_'; @@ -484,122 +485,6 @@ static void R_API_CALLCONV print_mqtt_array(data_output_t *output, data_array_t *orig = '\0'; // restore topic } -static char *append_topic(char *topic, data_t *data) -{ - if (data->type == DATA_STRING) { - strcpy(topic, data->value.v_ptr); // NOLINT - mqtt_sanitize_topic(topic); - topic += strlen(data->value.v_ptr); - } - else if (data->type == DATA_INT) { - topic += sprintf(topic, "%d", data->value.v_int); - } - else { - print_logf(LOG_ERROR, __func__, "Can't append data type %d to topic", data->type); - } - - return topic; -} - -static char *expand_topic(char *topic, char const *format, data_t *data, char const *hostname) -{ - // collect well-known top level keys - data_t *data_type = NULL; - data_t *data_model = NULL; - data_t *data_subtype = NULL; - data_t *data_channel = NULL; - data_t *data_id = NULL; - data_t *data_protocol = NULL; - for (data_t *d = data; d; d = d->next) { - if (!strcmp(d->key, "type")) - data_type = d; - else if (!strcmp(d->key, "model")) - data_model = d; - else if (!strcmp(d->key, "subtype")) - data_subtype = d; - else if (!strcmp(d->key, "channel")) - data_channel = d; - else if (!strcmp(d->key, "id")) - data_id = d; - else if (!strcmp(d->key, "protocol")) // NOTE: needs "-M protocol" - data_protocol = d; - } - - // consume entire format string - while (format && *format) { - data_t *data_token = NULL; - char const *string_token = NULL; - int leading_slash = 0; - char const *t_start = NULL; - char const *t_end = NULL; - char const *d_start = NULL; - char const *d_end = NULL; - // copy until '[' - while (*format && *format != '[') - *topic++ = *format++; - // skip '[' - if (!*format) - break; - ++format; - // read slash - if (!leading_slash && (*format < 'a' || *format > 'z')) { - leading_slash = *format; - format++; - } - // read key until : or ] - t_start = t_end = format; - while (*format && *format != ':' && *format != ']' && *format != '[') - t_end = ++format; - // read default until ] - if (*format == ':') { - d_start = d_end = ++format; - while (*format && *format != ']' && *format != '[') - d_end = ++format; - } - // check for proper closing - if (*format != ']') { - print_log(LOG_FATAL, __func__, "unterminated token"); - exit(1); - } - ++format; - - // resolve token - if (!strncmp(t_start, "hostname", t_end - t_start)) - string_token = hostname; - else if (!strncmp(t_start, "type", t_end - t_start)) - data_token = data_type; - else if (!strncmp(t_start, "model", t_end - t_start)) - data_token = data_model; - else if (!strncmp(t_start, "subtype", t_end - t_start)) - data_token = data_subtype; - else if (!strncmp(t_start, "channel", t_end - t_start)) - data_token = data_channel; - else if (!strncmp(t_start, "id", t_end - t_start)) - data_token = data_id; - else if (!strncmp(t_start, "protocol", t_end - t_start)) - data_token = data_protocol; - else { - print_logf(LOG_FATAL, __func__, "unknown token \"%.*s\"", (int)(t_end - t_start), t_start); - exit(1); - } - - // append token or default - if (!data_token && !string_token && !d_start) - continue; - if (leading_slash) - *topic++ = leading_slash; - if (data_token) - topic = append_topic(topic, data_token); - else if (string_token) - topic += sprintf(topic, "%s", string_token); - else - topic += sprintf(topic, "%.*s", (int)(d_end - d_start), d_start); - } - - *topic = '\0'; - return topic; -} - // [/type][/model][/subtype][/channel][/id]/battery: "OK"|"LOW" static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data, char const *format) { @@ -628,7 +513,7 @@ static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data, return; // NOTE: skip output on alloc failure. } data_print_jsons(data, message, message_size); - expand_topic(mqtt->topic, mqtt->states, data, mqtt->hostname); + expand_topic_string(mqtt->topic, mqtt->states, data, mqtt->hostname, mqtt_sanitize_topic); mqtt_client_publish(mqtt->mqc, mqtt->topic, message); *mqtt->topic = '\0'; // clear topic free(message); @@ -640,7 +525,7 @@ static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data, if (mqtt->events) { char message[2048]; // we expect the biggest strings to be around 500 bytes. data_print_jsons(data, message, sizeof(message)); - expand_topic(mqtt->topic, mqtt->events, data, mqtt->hostname); + expand_topic_string(mqtt->topic, mqtt->events, data, mqtt->hostname, mqtt_sanitize_topic); mqtt_client_publish(mqtt->mqc, mqtt->topic, message); *mqtt->topic = '\0'; // clear topic } @@ -650,7 +535,7 @@ static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data, return; } - end = expand_topic(mqtt->topic, mqtt->devices, data, mqtt->hostname); + end = expand_topic_string(mqtt->topic, mqtt->devices, data, mqtt->hostname, mqtt_sanitize_topic); } while (data) { diff --git a/src/string_expand.c b/src/string_expand.c new file mode 100644 index 00000000..bafb4ee4 --- /dev/null +++ b/src/string_expand.c @@ -0,0 +1,134 @@ +/** @file + String formatting a-la MQTT topic + + Copyright (C) 2019 Christian Zuckschwerdt + + 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 2 of the License, or + (at your option) any later version. +*/ + +#include "string_expand.h" +#include "logger.h" + +#include +#include +#include + +static char *append_topic(char *topic, data_t *data, expand_string_sanitizer sanitizer) +{ + if (data->type == DATA_STRING) { + strcpy(topic, data->value.v_ptr); // NOLINT + size_t len = strlen(topic); + (*sanitizer)(topic, topic + len); + topic += len; + } + else if (data->type == DATA_INT) { + topic += sprintf(topic, "%d", data->value.v_int); + } + else { + print_logf(LOG_ERROR, __func__, "Can't append data type %d to topic", data->type); + } + + return topic; +} + +char *expand_topic_string(char *topic, char const *format, data_t *data, char const *hostname, expand_string_sanitizer sanitizer) +{ + // collect well-known top level keys + data_t *data_type = NULL; + data_t *data_model = NULL; + data_t *data_subtype = NULL; + data_t *data_channel = NULL; + data_t *data_id = NULL; + data_t *data_protocol = NULL; + for (data_t *d = data; d; d = d->next) { + if (!strcmp(d->key, "type")) + data_type = d; + else if (!strcmp(d->key, "model")) + data_model = d; + else if (!strcmp(d->key, "subtype")) + data_subtype = d; + else if (!strcmp(d->key, "channel")) + data_channel = d; + else if (!strcmp(d->key, "id")) + data_id = d; + else if (!strcmp(d->key, "protocol")) // NOTE: needs "-M protocol" + data_protocol = d; + } + + // consume entire format string + while (format && *format) { + data_t *data_token = NULL; + char const *string_token = NULL; + int leading_slash = 0; + char const *t_start = NULL; + char const *t_end = NULL; + char const *d_start = NULL; + char const *d_end = NULL; + // copy until '[' + while (*format && *format != '[') + *topic++ = *format++; + // skip '[' + if (!*format) + break; + ++format; + // read slash + if (!leading_slash && (*format < 'a' || *format > 'z')) { + leading_slash = *format; + format++; + } + // read key until : or ] + t_start = t_end = format; + while (*format && *format != ':' && *format != ']' && *format != '[') + t_end = ++format; + // read default until ] + if (*format == ':') { + d_start = d_end = ++format; + while (*format && *format != ']' && *format != '[') + d_end = ++format; + } + // check for proper closing + if (*format != ']') { + print_log(LOG_FATAL, __func__, "unterminated token"); + exit(1); + } + ++format; + + // resolve token + if (!strncmp(t_start, "hostname", t_end - t_start)) + string_token = hostname; + else if (!strncmp(t_start, "type", t_end - t_start)) + data_token = data_type; + else if (!strncmp(t_start, "model", t_end - t_start)) + data_token = data_model; + else if (!strncmp(t_start, "subtype", t_end - t_start)) + data_token = data_subtype; + else if (!strncmp(t_start, "channel", t_end - t_start)) + data_token = data_channel; + else if (!strncmp(t_start, "id", t_end - t_start)) + data_token = data_id; + else if (!strncmp(t_start, "protocol", t_end - t_start)) + data_token = data_protocol; + else { + print_logf(LOG_FATAL, __func__, "unknown token \"%.*s\"", (int)(t_end - t_start), t_start); + exit(1); + } + + // append token or default + if (!data_token && !string_token && !d_start) + continue; + if (leading_slash) + *topic++ = leading_slash; + if (data_token) + topic = append_topic(topic, data_token, sanitizer); + else if (string_token) + topic += sprintf(topic, "%s", string_token); + else + topic += sprintf(topic, "%.*s", (int)(d_end - d_start), d_start); + } + + *topic = '\0'; + return topic; +}