From d5ebfb09e1cb043d92e4c35ddf767749c64c7b60 Mon Sep 17 00:00:00 2001 From: Martin Mares Date: Sat, 11 Aug 2018 23:39:23 +0200 Subject: [PATCH] Prometheus: First version of MQTT <-> Prometheus gateway --- prometheus/burrow-prometheus.c | 227 ++++++++++++++++++++++++--------- 1 file changed, 168 insertions(+), 59 deletions(-) diff --git a/prometheus/burrow-prometheus.c b/prometheus/burrow-prometheus.c index 8f0f560..272deae 100644 --- a/prometheus/burrow-prometheus.c +++ b/prometheus/burrow-prometheus.c @@ -5,16 +5,23 @@ */ #include +#include #include -#include +#include #include +#include +#include +#include #include #include #include #include +#include +#include #include #include +#include #include @@ -26,16 +33,16 @@ struct attr { }; static const struct attr attr_table[] = { - { "# HELP loft_temp Temperature in the loft", NULL }, + { "# HELP loft_temp Temperature in the loft [degC]", NULL }, { "# TYPE loft_temp gauge", NULL }, { "loft_temp", "burrow/loft/temperature" }, { "# HELP loft_fan Fan speed in the loft", NULL }, { "# TYPE loft_fan gauge", NULL }, { "loft_fan", "burrow/loft/fan" }, - { NULL, NULL }, }; static char *attr_values[ARRAY_SIZE(attr_table)]; +static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER; static void mqtt_publish(const char *topic, const char *fmt, ...) { @@ -48,12 +55,15 @@ static void mqtt_publish(const char *topic, const char *fmt, ...) va_end(args); } -static void mqtt_setup(void) +static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status) { - if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS) - die("Mosquitto: subscribe failed"); + if (!status) { + msg(L_DEBUG, "MQTT: Connection established, subscribing"); + if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS) + die("Mosquitto: subscribe failed"); - // mqtt_publish("burrow/loft/status", "ok"); + // mqtt_publish("burrow/loft/status", "ok"); + } } static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message) @@ -74,69 +84,152 @@ static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, c for (uint i=0; i < ARRAY_SIZE(attr_table); i++) { if (attr_table[i].topic && !strcmp(attr_table[i].topic, m->topic)) { + pthread_mutex_lock(&attr_mutex); if (attr_values[i]) { xfree(attr_values[i]); attr_values[i] = NULL; } if (val[0]) attr_values[i] = xstrdup(val); + pthread_mutex_unlock(&attr_mutex); } } } -static struct main_file mqtt_file; -static struct main_timer mqtt_timer; -#define MQTT_TIMER_FREQ 5000 +struct http { + struct mempool *mp; + int sk; + char *iobuf; + uint iobuf_pos, iobuf_max; + char *linebuf; + struct fbpool fbpool; +}; + +#define IO_TIMEOUT 60000 // in ms +#define IOBUF_SIZE 1024 +#define LINEBUF_SIZE 1024 -static void mqtt_recalc_main(void) +static void http_send(struct http *http) { - int fd = mosquitto_socket(mosq); - if (fd != mqtt_file.fd) { - msg(L_DEBUG, "MQTT: Changing socket fd to %d", fd); - if (fd < 0) { - file_del(&mqtt_file); - mqtt_file.fd = -1; - } else { - mqtt_file.fd = fd; - file_add(&mqtt_file); + byte *buf = fbpool_end(&http->fbpool); + uint len = mp_size(http->mp, buf); + + while (len) { + struct pollfd pfd = { .fd = http->sk, .events = POLLOUT, .revents = 0 }; + int res = poll(&pfd, 1, IO_TIMEOUT); + if (res < 0) + die("Poll failed: %m"); + if (!res) { + msg(L_ERROR, "HTTP write timed out"); + return; } + + res = write(http->sk, buf, len); + if (res < 0) { + msg(L_ERROR, "HTTP write failed: %m"); + return; + } + buf += res; + len -= res; } } -static void mqtt_main_timer(struct main_timer *tm) +static void http_error(struct http *http, const char *err) { - msg(L_DEBUG, "MQTT: Timer handler"); - mosquitto_loop_misc(mosq); - mqtt_recalc_main(); - timer_add_rel(&mqtt_timer, MQTT_TIMER_FREQ); + msg(L_INFO, "HTTP error: %s", err); + fbpool_start(&http->fbpool, http->mp, 0); + bprintf(&http->fbpool.fb, "HTTP/1.1 %s\r\n", err); + http_send(http); } -static int mqtt_main_read(struct main_file *mf UNUSED) -{ - msg(L_DEBUG, "MQTT: Read handler"); - mosquitto_loop_read(mosq, 1); - mqtt_recalc_main(); - return HOOK_IDLE; +static int http_get_char(struct http *http) { + if (http->iobuf_pos >= http->iobuf_max) { + struct pollfd pfd = { .fd = http->sk, .events = POLLIN, .revents = 0 }; + int res = poll(&pfd, 1, IO_TIMEOUT); + if (res < 0) + die("Poll failed: %m"); + if (!res) { + msg(L_ERROR, "HTTP read timed out"); + return -1; + } + int len = read(http->sk, http->iobuf, IOBUF_SIZE); + if (len < 0) { + msg(L_ERROR, "HTTP read error: %m"); + return -1; + } + if (!len) { + msg(L_ERROR, "HTTP connection closed"); + return -1; + } + http->iobuf_pos = 0; + http->iobuf_max = len; + } + return http->iobuf[http->iobuf_pos++]; } -static int mqtt_main_write(struct main_file *mf UNUSED) +static bool http_get_line(struct http *http) { - msg(L_DEBUG, "MQTT: Write handler"); - mosquitto_loop_write(mosq, 1); - mqtt_recalc_main(); - return HOOK_IDLE; + uint i = 0; + for (;;) { + int c = http_get_char(http); + if (c < 0) + return false; + if (c == '\n') { + if (i > 0 && http->linebuf[i-1] == '\r') + i--; + http->linebuf[i] = 0; + return true; + } + if (i >= IOBUF_SIZE-1) { + http_error(http, "400 Line too long"); + return false; + } + http->linebuf[i++] = c; + } } -static void mqtt_main_init(void) +static void http_connection(struct http *http) { - mqtt_file.fd = -1; - mqtt_file.read_handler = mqtt_main_read; - mqtt_file.write_handler = mqtt_main_write; + http->iobuf = mp_alloc(http->mp, IOBUF_SIZE); + http->linebuf = mp_alloc(http->mp, LINEBUF_SIZE); + fbpool_init(&http->fbpool); - mqtt_timer.handler = mqtt_main_timer; - timer_add_rel(&mqtt_timer, MQTT_TIMER_FREQ); + if (!http_get_line(http)) + return; + // msg(L_DEBUG, "Request line: <%s>", http->linebuf); + char *words[3]; + if (str_wordsplit(http->linebuf, words, 3) != 3) { + http_error(http, "400 Invalid request line"); + return; + } + if (strcmp(words[0], "GET")) { + http_error(http, "501 Method not implemented"); + return; + } - mqtt_recalc_main(); + for (;;) { + if (!http_get_line(http)) + return; + if (!http->linebuf[0]) + break; + // msg(L_DEBUG, "Header line: <%s>", http->linebuf); + } + + fbpool_start(&http->fbpool, http->mp, 0); + struct fastbuf *fb = &http->fbpool.fb; + bprintf(fb, "HTTP/1.1 200 OK\r\n"); + bprintf(fb, "Content-type: text/plain; version=0.0.4\r\n"); + bprintf(fb, "\r\n"); + for (uint i=0; i < ARRAY_SIZE(attr_table); i++) { + if (attr_table[i].topic) { + pthread_mutex_lock(&attr_mutex); + bprintf(fb, "%s %s\n", attr_table[i].pm, attr_values[i] ? : ""); + pthread_mutex_unlock(&attr_mutex); + } else { + bprintf(fb, "%s\n", attr_table[i].pm); + } + } + http_send(http); } static int use_daemon; @@ -159,7 +252,6 @@ int main(int argc UNUSED, char **argv) { log_init(argv[0]); opt_parse(&options, argv+1); - main_init(); if (use_daemon) { struct log_stream *ls = log_new_syslog("daemon", LOG_PID); @@ -173,9 +265,9 @@ int main(int argc UNUSED, char **argv) if (!mosq) die("Mosquitto: initialization failed"); + mosquitto_connect_callback_set(mosq, mqtt_conn_callback); mosquitto_log_callback_set(mosq, mqtt_log_callback); mosquitto_message_callback_set(mosq, mqtt_msg_callback); - mqtt_main_init(); #if 0 // FIXME: Publish online/offline status @@ -183,23 +275,40 @@ int main(int argc UNUSED, char **argv) die("Mosquitto: unable to set will"); #endif - if (mosquitto_connect(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS) + if (mosquitto_connect_async(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS) die("Mosquitto: connect failed"); - mqtt_setup(); + if (mosquitto_loop_start(mosq)) + die("Mosquitto: cannot start service thread"); + + int listen_sk = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); + if (listen_sk < 0) + die("Cannot create listening socket: %m"); + + int one = 1; + if (setsockopt(listen_sk, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) + die("Cannot set SO_REUSEADDR: %m"); + + struct sockaddr_in sin = { .sin_family = AF_INET, .sin_port = htons(1422), .sin_addr = INADDR_ANY }; + if (bind(listen_sk, (const struct sockaddr *) &sin, sizeof(sin)) < 0) + die("Cannot bind listening socket: %m"); + + if (listen(listen_sk, 64) < 0) + die("Cannot listen: %m"); - main_loop(); -#if 0 for (;;) { - int err = mosquitto_loop(mosq, 5000, 1); - if (err == MOSQ_ERR_NO_CONN) { - err = mosquitto_reconnect(mosq); - if (err == MOSQ_ERR_SUCCESS) - mqtt_setup(); - else - msg(L_ERROR, "Mosquitto: cannot reconnect, error %d", err); - } else if (err != MOSQ_ERR_SUCCESS) - msg(L_ERROR, "Mosquitto: loop returned error %d", err); + int sk = accept(listen_sk, NULL, NULL); + if (sk < 0) { + msg(L_ERROR, "HTTP accept failed: %m"); + continue; + } + msg(L_DEBUG, "HTTP accepted connection"); + struct mempool *mp = mp_new(4096); + struct http *http = mp_alloc_zero(mp, sizeof(*http)); + http->mp = mp; + http->sk = sk; + http_connection(http); + mp_delete(mp); + close(sk); } -#endif } -- 2.39.5