2 * A gateway between MQTT and InfluxDB
4 * (c) 2018--2020 Martin Mares <mj@ucw.cz>
8 #include <ucw/fastbuf.h>
11 #include <ucw/string.h>
21 #include <curl/curl.h>
22 #include <mosquitto.h>
24 #define MEASUREMENT_TIMEOUT 120
26 #define INFLUX_URL "http://localhost:8086/write?db=burrow&precision=s"
27 #define INFLUX_TIMEOUT 30
28 #define INFLUX_VERBOSE 0
29 #define INFLUX_INTERVAL 60
31 static struct mosquitto *mosq;
35 const char *value_name;
41 static const struct attr attr_table[] = {
43 .metric = "temp,where=loft",
45 .topic = "burrow/temp/loft",
48 .metric = "temp,where=ursarium",
50 .topic = "burrow/temp/ursarium"
53 .metric = "temp,where=catarium",
55 .topic = "burrow/temp/catarium"
58 .metric = "temp,where=garage",
60 .topic = "burrow/temp/garage"
63 .metric = "temp,where=terarium",
65 .topic = "burrow/temp/terarium"
68 .metric = "rh,where=ursarium",
70 .topic = "burrow/temp/ursarium-rh"
73 .metric = "temp,where=catarium_clock",
75 .topic = "burrow/temp/clock"
78 .metric = "pressure,where=catarium_clock",
80 .topic = "burrow/pressure/clock"
83 .metric = "air,where=inside_intake",
85 .topic = "burrow/air/inside-intake",
88 .metric = "air,where=inside_exhaust",
90 .topic = "burrow/air/inside-exhaust",
93 .metric = "air,where=outside_intake",
95 .topic = "burrow/air/outside-intake",
98 .metric = "air,where=outside_exhaust",
100 .topic = "burrow/air/outside-exhaust",
103 .metric = "air,where=mixed",
105 .topic = "burrow/air/mixed",
108 .metric = "air,where=inside_intake_avg",
110 .topic = "burrow/avg/air/inside-intake",
113 .metric = "air,where=outside_intake_avg",
115 .topic = "burrow/avg/air/outside-intake",
118 .metric = "air_bypass",
120 .topic = "burrow/air/bypass"
123 .metric = "air_fan_pwm",
125 .topic = "burrow/air/exchanger-fan"
129 .metric = "loft_fan",
130 .topic = "burrow/loft/fan"
134 .metric = "water_circ",
136 .topic = "burrow/loft/circulation"
139 .metric = "pm_voltage,phase=L1N",
141 .topic = "burrow/power/voltage/l1n",
144 .metric = "pm_voltage,phase=L2N",
146 .topic = "burrow/power/voltage/l2n",
149 .metric = "pm_voltage,phase=L3N",
151 .topic = "burrow/power/voltage/l3n",
154 .metric = "pm_current,phase=L1",
156 .topic = "burrow/power/current/l1",
159 .metric = "pm_current,phase=L2",
161 .topic = "burrow/power/current/l2",
164 .metric = "pm_current,phase=L3",
166 .topic = "burrow/power/current/l3",
169 .metric = "pm_power",
171 .topic = "burrow/power/power",
174 .metric = "pm_energy",
176 .topic = "burrow/power/energy",
179 .metric = "pm_reactive_power",
181 .topic = "burrow/power/reactive/power",
184 .metric = "pm_reactive_energy",
185 .value_name = "kVArh",
186 .topic = "burrow/power/reactive/energy",
189 .metric = "heating_water_pressure",
191 .topic = "burrow/heating/water-pressure",
195 .metric = "heating_outside_temp",
197 .topic = "burrow/heating/outside-temp",
201 .metric = "heating_room_temp,circuit=1",
203 .topic = "burrow/heating/circuit1/room-temp",
207 .metric = "heating_mix-temp,circuit=1",
209 .topic = "burrow/heating/circuit1/mix-temp",
213 .metric = "heating_mix-valve,circuit=1",
215 .topic = "burrow/heating/circuit1/mix-valve",
219 .metric = "heating_active,circuit=1",
221 .topic = "burrow/heating/circuit1/active",
225 .metric = "heating_pump_active,circuit=1",
227 .topic = "burrow/heating/circuit1/pump",
231 .metric = "heating_room_temp,circuit=2",
233 .topic = "burrow/heating/circuit2/room-temp",
237 .metric = "heating_active,circuit=2",
239 .topic = "burrow/heating/circuit2/active",
243 .metric = "heating_active,circuit=water",
245 .topic = "burrow/heating/water/active",
249 .metric = "heating_error",
251 .topic = "burrow/heating/error",
255 .metric = "heating_clock",
257 .topic = "burrow/heating/clock",
265 static char *attr_values[ARRAY_SIZE(attr_table)];
266 static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER;
268 static void mqtt_publish(const char *topic, const char *fmt, ...)
273 int l = vsnprintf(m, sizeof(m), fmt, args);
274 if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
275 msg(L_ERROR, "Mosquitto: publish failed");
279 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
282 msg(L_DEBUG, "MQTT: Connection established, subscribing");
283 if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS)
284 die("Mosquitto: subscribe failed");
286 mqtt_publish("status/influx", "ok");
290 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
292 // msg(L_INFO, "MQTT(%d): %s", level, message);
295 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
298 if (m->payloadlen >= sizeof(val) - 1) {
299 msg(L_ERROR, "Invalid value for topic %s", m->topic);
302 memcpy(val, m->payload, m->payloadlen);
303 val[m->payloadlen] = 0;
304 msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
306 for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
307 if (attr_table[i].topic && !strcmp(attr_table[i].topic, m->topic)) {
308 pthread_mutex_lock(&attr_mutex);
309 if (attr_values[i]) {
310 xfree(attr_values[i]);
311 attr_values[i] = NULL;
314 attr_values[i] = xstrdup(val);
315 pthread_mutex_unlock(&attr_mutex);
324 static struct fastbuf *influx_fb;
325 static struct fastbuf *influx_err_fb;
327 static size_t influx_write_callback(char *ptr, size_t size, size_t nmemb, void *userdata UNUSED)
330 for (size_t i=0; i<nmemb; i++) {
331 int c = *(byte *)ptr;
337 else if (c < 0x20 || c > 0x7e)
339 bputc(influx_err_fb, c);
344 static void influx_init(void)
346 curl = curl_easy_init();
348 die("libcurl init failed");
350 curl_easy_setopt(curl, CURLOPT_URL, INFLUX_URL);
351 curl_easy_setopt(curl, CURLOPT_VERBOSE, (long) INFLUX_VERBOSE);
352 curl_easy_setopt(curl, CURLOPT_TIMEOUT, (long) INFLUX_TIMEOUT);
353 curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, influx_write_callback);
355 influx_fb = fbgrow_create(4096);
356 influx_err_fb = fbgrow_create(256);
359 static struct fastbuf *influx_write_start(void)
361 fbgrow_reset(influx_fb);
365 static void influx_write_flush(void)
369 fbgrow_get_buf(influx_fb, &buf);
370 curl_easy_setopt(curl, CURLOPT_POSTFIELDS, buf);
372 msg(L_DEBUG, "InfluxDB: Sending data (%u bytes)", strlen(buf));
374 fprintf(stderr, "%s", buf);
376 fbgrow_reset(influx_err_fb);
378 CURLcode code = curl_easy_perform(curl);
380 if (code != CURLE_OK) {
381 msg(L_ERROR, "Write to InfluxDB failed: %s", curl_easy_strerror(code));
386 curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &rc);
387 if (rc != 200 && rc != 204) {
388 bputc(influx_err_fb, 0);
390 fbgrow_get_buf(influx_err_fb, &err);
391 msg(L_DEBUG, "InfluxDB HTTP error %ld: %s [...]", rc, err);
397 static int use_daemon;
398 static int use_debug;
400 static struct opt_section options = {
402 OPT_HELP("A daemon for transferring MQTT data to InfluxDB"),
404 OPT_HELP("Options:"),
405 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
406 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
413 int main(int argc UNUSED, char **argv)
416 opt_parse(&options, argv+1);
419 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
420 log_set_default_stream(ls);
423 log_default_stream()->levels &= ~(1U << L_DEBUG);
425 mosquitto_lib_init();
426 mosq = mosquitto_new("influx", 1, NULL);
428 die("Mosquitto: initialization failed");
430 mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
431 mosquitto_log_callback_set(mosq, mqtt_log_callback);
432 mosquitto_message_callback_set(mosq, mqtt_msg_callback);
434 if (mosquitto_will_set(mosq, "status/influx", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
435 die("Mosquitto: unable to set will");
437 if (mosquitto_tls_set(mosq, "/etc/burrow-mqtt/ca.crt", NULL, "/etc/burrow-mqtt/client.crt", "/etc/burrow-mqtt/client.key", NULL) != MOSQ_ERR_SUCCESS)
438 die("Mosquitto: unable to set TLS parameters");
440 if (mosquitto_connect_async(mosq, "burrow-mqtt", 8883, 60) != MOSQ_ERR_SUCCESS)
441 die("Mosquitto: connect failed");
443 if (mosquitto_loop_start(mosq))
444 die("Mosquitto: cannot start service thread");
449 struct fastbuf *f = influx_write_start();
450 char val[256], *w[2];
451 time_t now = time(NULL);
453 for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
454 const struct attr *a = &attr_table[i];
456 pthread_mutex_lock(&attr_mutex);
457 snprintf(val, sizeof(val), "%s", attr_values[i] ? : "");
458 pthread_mutex_unlock(&attr_mutex);
462 int fields = str_wordsplit(val, w, ARRAY_SIZE(w));
466 time_t t = atoll(w[1]);
467 uint timeout = a->timeout ? : MEASUREMENT_TIMEOUT;
468 if (t < now - timeout)
473 bprintf(f, "%s %s=%s\n", a->metric, a->value_name, val);
475 bprintf(f, "%s %s=\"%s\"\n", a->metric, a->value_name, val);
477 influx_write_flush();
478 sleep(INFLUX_INTERVAL);