2 * A gateway between MQTT and Prometheus
4 * (c) 2018 Martin Mares <mj@ucw.cz>
8 #include <ucw/fastbuf.h>
10 #include <ucw/mempool.h>
12 #include <ucw/string.h>
14 #include <netinet/in.h>
21 #include <sys/socket.h>
26 #include <mosquitto.h>
28 #define MEASUREMENT_TIMEOUT 120
30 static struct mosquitto *mosq;
39 static const struct attr attr_table[] = {
41 .metric = "temp_loft",
42 .help = "Temperature in the loft [degC]",
44 .topic = "burrow/temp/loft",
48 .help = "Fan speed in the loft (0-3)",
50 .topic = "burrow/loft/fan"
53 .metric = "loft_circ",
54 .help = "Warm water circulation (0-1)",
56 .topic = "burrow/loft/circulation"
59 .metric = "temp_ursarium",
60 .help = "Temperature in the Ursarium [degC]",
62 .topic = "burrow/temp/ursarium"
65 .metric = "temp_catarium",
66 .help = "Temperature in the Catarium [degC]",
68 .topic = "burrow/temp/catarium"
71 .metric = "temp_garage",
72 .help = "Temperature in the garage [degC]",
74 .topic = "burrow/temp/garage"
77 .metric = "temp_kitchen",
78 .help = "Temperature in the kitchen [degC]",
80 .topic = "burrow/temp/kitchen"
83 .metric = "rh_ursarium",
84 .help = "Relative humidity in the Ursarium [%]",
86 .topic = "burrow/temp/ursarium-rh"
89 .metric = "temp_catarium_clock",
90 .help = "Temperature on Catarium clock [degC]",
92 .topic = "burrow/temp/clock"
95 .metric = "press_catarium_clock",
96 .help = "Pressure on Catarium clock [Pa]",
98 .topic = "burrow/pressure/clock"
101 .metric = "air_inside_intake",
102 .help = "Temperature of air intake from inside [degC]",
104 .topic = "burrow/air/inside-intake",
107 .metric = "air_inside_exhaust",
108 .help = "Temperature of air exhaust to inside [degC]",
110 .topic = "burrow/air/inside-exhaust",
113 .metric = "air_outside_intake",
114 .help = "Temperature of air intake from outside [degC]",
116 .topic = "burrow/air/outside-intake",
119 .metric = "air_outside_exhaust",
120 .help = "Temperature of air exhaust to outside [degC]",
122 .topic = "burrow/air/outside-exhaust",
125 .metric = "air_chilled",
126 .help = "Temperature of chilled air [degC]",
128 .topic = "burrow/air/chilled",
131 .metric = "air_bypass",
132 .help = "Heat exchanger bypass (0-1)",
134 .topic = "burrow/air/bypass"
138 static char *attr_values[ARRAY_SIZE(attr_table)];
139 static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER;
141 static void mqtt_publish(const char *topic, const char *fmt, ...)
146 int l = vsnprintf(m, sizeof(m), fmt, args);
147 if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
148 msg(L_ERROR, "Mosquitto: publish failed");
152 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
155 msg(L_DEBUG, "MQTT: Connection established, subscribing");
156 if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS)
157 die("Mosquitto: subscribe failed");
159 mqtt_publish("status/prometheus", "ok");
163 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
165 // msg(L_INFO, "MQTT(%d): %s", level, message);
168 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
171 if (m->payloadlen >= sizeof(val) - 1) {
172 msg(L_ERROR, "Invalid value for topic %s", m->topic);
175 memcpy(val, m->payload, m->payloadlen);
176 val[m->payloadlen] = 0;
177 msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
179 for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
180 if (attr_table[i].topic && !strcmp(attr_table[i].topic, m->topic)) {
181 pthread_mutex_lock(&attr_mutex);
182 if (attr_values[i]) {
183 xfree(attr_values[i]);
184 attr_values[i] = NULL;
187 attr_values[i] = xstrdup(val);
188 pthread_mutex_unlock(&attr_mutex);
197 uint iobuf_pos, iobuf_max;
199 struct fbpool fbpool;
202 #define IO_TIMEOUT 60000 // in ms
203 #define IOBUF_SIZE 1024
204 #define LINEBUF_SIZE 1024
206 static void http_send(struct http *http)
208 byte *buf = fbpool_end(&http->fbpool);
209 uint len = mp_size(http->mp, buf);
212 struct pollfd pfd = { .fd = http->sk, .events = POLLOUT, .revents = 0 };
213 int res = poll(&pfd, 1, IO_TIMEOUT);
215 die("Poll failed: %m");
217 msg(L_ERROR, "HTTP write timed out");
221 res = write(http->sk, buf, len);
223 msg(L_ERROR, "HTTP write failed: %m");
231 static void http_error(struct http *http, const char *err)
233 msg(L_INFO, "HTTP error: %s", err);
234 fbpool_start(&http->fbpool, http->mp, 0);
235 bprintf(&http->fbpool.fb, "HTTP/1.1 %s\r\n", err);
239 static int http_get_char(struct http *http) {
240 if (http->iobuf_pos >= http->iobuf_max) {
241 struct pollfd pfd = { .fd = http->sk, .events = POLLIN, .revents = 0 };
242 int res = poll(&pfd, 1, IO_TIMEOUT);
244 die("Poll failed: %m");
246 msg(L_ERROR, "HTTP read timed out");
249 int len = read(http->sk, http->iobuf, IOBUF_SIZE);
251 msg(L_ERROR, "HTTP read error: %m");
255 msg(L_ERROR, "HTTP connection closed");
259 http->iobuf_max = len;
261 return http->iobuf[http->iobuf_pos++];
264 static bool http_get_line(struct http *http)
268 int c = http_get_char(http);
272 if (i > 0 && http->linebuf[i-1] == '\r')
274 http->linebuf[i] = 0;
277 if (i >= IOBUF_SIZE-1) {
278 http_error(http, "400 Line too long");
281 http->linebuf[i++] = c;
285 static void http_answer(struct fastbuf *fb)
287 char val[256], *w[2];
288 time_t now = time(NULL);
290 for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
291 const struct attr *a = &attr_table[i];
293 pthread_mutex_lock(&attr_mutex);
294 snprintf(val, sizeof(val), "%s", attr_values[i] ? : "");
295 pthread_mutex_unlock(&attr_mutex);
299 int fields = str_wordsplit(val, w, ARRAY_SIZE(w));
303 time_t t = atoll(w[1]);
304 if (t < now - MEASUREMENT_TIMEOUT)
309 bprintf(fb, "# HELP %s %s\n", a->metric, a->help);
311 bprintf(fb, "# TYPE %s %s\n", a->metric, a->type);
312 bprintf(fb, "%s %s", a->metric, val);
314 // Prometheus does not like our timestamps -- why?
316 bprintf(fb, " %s", w[1]);
322 static void http_connection(struct http *http)
324 http->iobuf = mp_alloc(http->mp, IOBUF_SIZE);
325 http->linebuf = mp_alloc(http->mp, LINEBUF_SIZE);
326 fbpool_init(&http->fbpool);
328 if (!http_get_line(http))
330 // msg(L_DEBUG, "Request line: <%s>", http->linebuf);
332 if (str_wordsplit(http->linebuf, words, 3) != 3) {
333 http_error(http, "400 Invalid request line");
336 if (strcmp(words[0], "GET")) {
337 http_error(http, "501 Method not implemented");
342 if (!http_get_line(http))
344 if (!http->linebuf[0])
346 // msg(L_DEBUG, "Header line: <%s>", http->linebuf);
349 fbpool_start(&http->fbpool, http->mp, 0);
350 struct fastbuf *fb = &http->fbpool.fb;
351 bprintf(fb, "HTTP/1.1 200 OK\r\n");
352 bprintf(fb, "Content-type: text/plain; version=0.0.4\r\n");
358 static int use_daemon;
359 static int use_debug;
361 static struct opt_section options = {
363 OPT_HELP("A daemon for controlling the solid state relay module via MQTT"),
365 OPT_HELP("Options:"),
366 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
367 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
374 int main(int argc UNUSED, char **argv)
377 opt_parse(&options, argv+1);
380 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
381 log_set_default_stream(ls);
384 log_default_stream()->levels &= ~(1U << L_DEBUG);
386 mosquitto_lib_init();
387 mosq = mosquitto_new("prometheus", 1, NULL);
389 die("Mosquitto: initialization failed");
391 mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
392 mosquitto_log_callback_set(mosq, mqtt_log_callback);
393 mosquitto_message_callback_set(mosq, mqtt_msg_callback);
395 if (mosquitto_will_set(mosq, "status/prometheus", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
396 die("Mosquitto: unable to set will");
398 if (mosquitto_connect_async(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS)
399 die("Mosquitto: connect failed");
401 if (mosquitto_loop_start(mosq))
402 die("Mosquitto: cannot start service thread");
404 int listen_sk = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
406 die("Cannot create listening socket: %m");
409 if (setsockopt(listen_sk, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0)
410 die("Cannot set SO_REUSEADDR: %m");
412 struct sockaddr_in sin = { .sin_family = AF_INET, .sin_port = htons(1422), .sin_addr = INADDR_ANY };
413 if (bind(listen_sk, (const struct sockaddr *) &sin, sizeof(sin)) < 0)
414 die("Cannot bind listening socket: %m");
416 if (listen(listen_sk, 64) < 0)
417 die("Cannot listen: %m");
420 int sk = accept(listen_sk, NULL, NULL);
422 msg(L_ERROR, "HTTP accept failed: %m");
425 msg(L_DEBUG, "HTTP accepted connection");
426 struct mempool *mp = mp_new(4096);
427 struct http *http = mp_alloc_zero(mp, sizeof(*http));
430 http_connection(http);