]> mj.ucw.cz Git - home-hw.git/blob - prometheus/burrow-prometheus.c
Prometheus: Power meter delivered via MQTT
[home-hw.git] / prometheus / burrow-prometheus.c
1 /*
2  *      A gateway between MQTT and Prometheus
3  *
4  *      (c) 2018--2019 Martin Mares <mj@ucw.cz>
5  */
6
7 #include <ucw/lib.h>
8 #include <ucw/fastbuf.h>
9 #include <ucw/log.h>
10 #include <ucw/mempool.h>
11 #include <ucw/opt.h>
12 #include <ucw/string.h>
13
14 #include <netinet/in.h>
15 #include <pthread.h>
16 #include <stdarg.h>
17 #include <stdio.h>
18 #include <stdlib.h>
19 #include <string.h>
20 #include <sys/poll.h>
21 #include <sys/socket.h>
22 #include <syslog.h>
23 #include <time.h>
24 #include <unistd.h>
25
26 #include <mosquitto.h>
27
28 #define MEASUREMENT_TIMEOUT 120
29
30 static struct mosquitto *mosq;
31
32 struct attr {
33         const char *metric;
34         const char *help;
35         const char *type;
36         const char *topic;
37 };
38
39 static const struct attr attr_table[] = {
40         {
41                 .metric = "temp_loft",
42                 .help = "Temperature in the loft [degC]",
43                 .type = "gauge",
44                 .topic = "burrow/temp/loft",
45         },
46         {
47                 .metric = "loft_fan",
48                 .help = "Fan speed in the loft (0-3)",
49                 .type = "gauge",
50                 .topic = "burrow/loft/fan"
51         },
52         {
53                 .metric = "loft_circ",
54                 .help = "Warm water circulation (0-1)",
55                 .type = "gauge",
56                 .topic = "burrow/loft/circulation"
57         },
58         {
59                 .metric = "temp_ursarium",
60                 .help = "Temperature in the Ursarium [degC]",
61                 .type = "gauge",
62                 .topic = "burrow/temp/ursarium"
63         },
64         {
65                 .metric = "temp_catarium",
66                 .help = "Temperature in the Catarium [degC]",
67                 .type = "gauge",
68                 .topic = "burrow/temp/catarium"
69         },
70         {
71                 .metric = "temp_garage",
72                 .help = "Temperature in the garage [degC]",
73                 .type = "gauge",
74                 .topic = "burrow/temp/garage"
75         },
76         {
77                 .metric = "temp_kitchen",
78                 .help = "Temperature in the kitchen [degC]",
79                 .type = "gauge",
80                 .topic = "burrow/temp/kitchen"
81         },
82         {
83                 .metric = "rh_ursarium",
84                 .help = "Relative humidity in the Ursarium [%]",
85                 .type = "gauge",
86                 .topic = "burrow/temp/ursarium-rh"
87         },
88         {
89                 .metric = "temp_catarium_clock",
90                 .help = "Temperature on Catarium clock [degC]",
91                 .type = "gauge",
92                 .topic = "burrow/temp/clock"
93         },
94         {
95                 .metric = "press_catarium_clock",
96                 .help = "Pressure on Catarium clock [Pa]",
97                 .type = "gauge",
98                 .topic = "burrow/pressure/clock"
99         },
100         {
101                 .metric = "air_inside_intake",
102                 .help = "Temperature of air intake from inside [degC]",
103                 .type = "gauge",
104                 .topic = "burrow/air/inside-intake",
105         },
106         {
107                 .metric = "air_inside_exhaust",
108                 .help = "Temperature of air exhaust to inside [degC]",
109                 .type = "gauge",
110                 .topic = "burrow/air/inside-exhaust",
111         },
112         {
113                 .metric = "air_outside_intake",
114                 .help = "Temperature of air intake from outside [degC]",
115                 .type = "gauge",
116                 .topic = "burrow/air/outside-intake",
117         },
118         {
119                 .metric = "air_outside_exhaust",
120                 .help = "Temperature of air exhaust to outside [degC]",
121                 .type = "gauge",
122                 .topic = "burrow/air/outside-exhaust",
123         },
124         {
125                 .metric = "air_mixed",
126                 .help = "Temperature of mixed air [degC]",
127                 .type = "gauge",
128                 .topic = "burrow/air/mixed",
129         },
130         {
131                 .metric = "air_bypass",
132                 .help = "Heat exchanger bypass (0-1)",
133                 .type = "gauge",
134                 .topic = "burrow/air/bypass"
135         },
136         {
137                 .metric = "air_fan_pwm",
138                 .help = "Heat exchanger fan PWM (0-255)",
139                 .type = "gauge",
140                 .topic = "burrow/air/fan-pwm"
141         },
142         {
143                 // Common heading for all voltages
144                 .metric = "pm_voltage",
145                 .help = "Voltage between phases and neutral [V]",
146                 .type = "gauge",
147         },
148         {
149                 .metric = "pm_voltage{phase=\"L1N\"}",
150                 .topic = "burrow/power/voltage/l1n",
151         },
152         {
153                 .metric = "pm_voltage{phase=\"L2N\"}",
154                 .topic = "burrow/power/voltage/l2n",
155         },
156         {
157                 .metric = "pm_voltage{phase=\"L3N\"}",
158                 .topic = "burrow/power/voltage/l3n",
159         },
160         {
161                 // Common heading for all currents
162                 .metric = "pm_current",
163                 .help = "Current through phases [A]",
164                 .type = "gauge",
165         },
166         {
167                 .metric = "pm_current{phase=\"L1\"}",
168                 .topic = "burrow/power/current/l1",
169         },
170         {
171                 .metric = "pm_current{phase=\"L2\"}",
172                 .topic = "burrow/power/current/l2",
173         },
174         {
175                 .metric = "pm_current{phase=\"L3\"}",
176                 .topic = "burrow/power/current/l3",
177         },
178         {
179                 .metric = "pm_power",
180                 .help = "Total power [W]",
181                 .type = "gauge",
182                 .topic = "burrow/power/power",
183         },
184         {
185                 .metric = "pm_energy",
186                 .help = "Total energy [kWh]",
187                 .type = "gauge",
188                 .topic = "burrow/power/energy",
189         },
190         {
191                 .metric = "pm_reactive_power",
192                 .help = "Total reactive power [VAr]",
193                 .type = "gauge",
194                 .topic = "burrow/power/reactive/power",
195         },
196         {
197                 .metric = "pm_reactive_energy",
198                 .help = "Total reactive energy [kVArh]",
199                 .type = "gauge",
200                 .topic = "burrow/power/reactive/energy",
201         },
202 };
203
204 static char *attr_values[ARRAY_SIZE(attr_table)];
205 static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER;
206
207 static void mqtt_publish(const char *topic, const char *fmt, ...)
208 {
209         va_list args;
210         va_start(args, fmt);
211         char m[256];
212         int l = vsnprintf(m, sizeof(m), fmt, args);
213         if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
214                 msg(L_ERROR, "Mosquitto: publish failed");
215         va_end(args);
216 }
217
218 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
219 {
220         if (!status) {
221                 msg(L_DEBUG, "MQTT: Connection established, subscribing");
222                 if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS)
223                         die("Mosquitto: subscribe failed");
224
225                 mqtt_publish("status/prometheus", "ok");
226         }
227 }
228
229 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
230 {
231         // msg(L_INFO, "MQTT(%d): %s", level, message);
232 }
233
234 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
235 {
236         char val[256];
237         if (m->payloadlen >= sizeof(val) - 1) {
238                 msg(L_ERROR, "Invalid value for topic %s", m->topic);
239                 return;
240         }
241         memcpy(val, m->payload, m->payloadlen);
242         val[m->payloadlen] = 0;
243         msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
244
245         for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
246                 if (attr_table[i].topic && !strcmp(attr_table[i].topic, m->topic)) {
247                         pthread_mutex_lock(&attr_mutex);
248                         if (attr_values[i]) {
249                                 xfree(attr_values[i]);
250                                 attr_values[i] = NULL;
251                         }
252                         if (val[0])
253                                 attr_values[i] = xstrdup(val);
254                         pthread_mutex_unlock(&attr_mutex);
255                 }
256         }
257 }
258
259 struct http {
260         struct mempool *mp;
261         int sk;
262         char *iobuf;
263         uint iobuf_pos, iobuf_max;
264         char *linebuf;
265         struct fbpool fbpool;
266 };
267
268 #define IO_TIMEOUT 60000        // in ms
269 #define IOBUF_SIZE 1024
270 #define LINEBUF_SIZE 1024
271
272 static void http_send(struct http *http)
273 {
274         byte *buf = fbpool_end(&http->fbpool);
275         uint len = mp_size(http->mp, buf);
276
277         while (len) {
278                 struct pollfd pfd = { .fd = http->sk, .events = POLLOUT, .revents = 0 };
279                 int res = poll(&pfd, 1, IO_TIMEOUT);
280                 if (res < 0)
281                         die("Poll failed: %m");
282                 if (!res) {
283                         msg(L_ERROR, "HTTP write timed out");
284                         return;
285                 }
286
287                 res = write(http->sk, buf, len);
288                 if (res < 0) {
289                         msg(L_ERROR, "HTTP write failed: %m");
290                         return;
291                 }
292                 buf += res;
293                 len -= res;
294         }
295 }
296
297 static void http_error(struct http *http, const char *err)
298 {
299         msg(L_INFO, "HTTP error: %s", err);
300         fbpool_start(&http->fbpool, http->mp, 0);
301         bprintf(&http->fbpool.fb, "HTTP/1.1 %s\r\n", err);
302         http_send(http);
303 }
304
305 static int http_get_char(struct http *http) {
306         if (http->iobuf_pos >= http->iobuf_max) {
307                 struct pollfd pfd = { .fd = http->sk, .events = POLLIN, .revents = 0 };
308                 int res = poll(&pfd, 1, IO_TIMEOUT);
309                 if (res < 0)
310                         die("Poll failed: %m");
311                 if (!res) {
312                         msg(L_ERROR, "HTTP read timed out");
313                         return -1;
314                 }
315                 int len = read(http->sk, http->iobuf, IOBUF_SIZE);
316                 if (len < 0) {
317                         msg(L_ERROR, "HTTP read error: %m");
318                         return -1;
319                 }
320                 if (!len) {
321                         msg(L_ERROR, "HTTP connection closed");
322                         return -1;
323                 }
324                 http->iobuf_pos = 0;
325                 http->iobuf_max = len;
326         }
327         return http->iobuf[http->iobuf_pos++];
328 }
329
330 static bool http_get_line(struct http *http)
331 {
332         uint i = 0;
333         for (;;) {
334                 int c = http_get_char(http);
335                 if (c < 0)
336                         return false;
337                 if (c == '\n') {
338                         if (i > 0 && http->linebuf[i-1] == '\r')
339                                 i--;
340                         http->linebuf[i] = 0;
341                         return true;
342                 }
343                 if (i >= IOBUF_SIZE-1) {
344                         http_error(http, "400 Line too long");
345                         return false;
346                 }
347                 http->linebuf[i++] = c;
348         }
349 }
350
351 static void http_answer(struct fastbuf *fb)
352 {
353         char val[256], *w[2];
354         time_t now = time(NULL);
355
356         for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
357                 const struct attr *a = &attr_table[i];
358
359                 pthread_mutex_lock(&attr_mutex);
360                 snprintf(val, sizeof(val), "%s", attr_values[i] ? : "");
361                 pthread_mutex_unlock(&attr_mutex);
362
363                 if (a->help)
364                         bprintf(fb, "# HELP %s %s\n", a->metric, a->help);
365                 if (a->type)
366                         bprintf(fb, "# TYPE %s %s\n", a->metric, a->type);
367
368                 if (!val[0])
369                         continue;
370                 int fields = str_wordsplit(val, w, ARRAY_SIZE(w));
371                 if (fields < 1)
372                         continue;
373                 if (fields >= 2) {
374                         time_t t = atoll(w[1]);
375                         if (t < now - MEASUREMENT_TIMEOUT)
376                                 continue;
377                 }
378
379                 if (a->topic) {
380                         bprintf(fb, "%s %s", a->metric, val);
381 #if 0
382                         // Prometheus does not like our timestamps -- why?
383                         if (fields >= 2)
384                                 bprintf(fb, " %s", w[1]);
385 #endif
386                         bputc(fb, '\n');
387                 }
388         }
389 }
390
391 static void http_connection(struct http *http)
392 {
393         http->iobuf = mp_alloc(http->mp, IOBUF_SIZE);
394         http->linebuf = mp_alloc(http->mp, LINEBUF_SIZE);
395         fbpool_init(&http->fbpool);
396
397         if (!http_get_line(http))
398                 return;
399         // msg(L_DEBUG, "Request line: <%s>", http->linebuf);
400         char *words[3];
401         if (str_wordsplit(http->linebuf, words, 3) != 3) {
402                 http_error(http, "400 Invalid request line");
403                 return;
404         }
405         if (strcmp(words[0], "GET")) {
406                 http_error(http, "501 Method not implemented");
407                 return;
408         }
409
410         for (;;) {
411                 if (!http_get_line(http))
412                         return;
413                 if (!http->linebuf[0])
414                         break;
415                 // msg(L_DEBUG, "Header line: <%s>", http->linebuf);
416         }
417
418         fbpool_start(&http->fbpool, http->mp, 0);
419         struct fastbuf *fb = &http->fbpool.fb;
420         bprintf(fb, "HTTP/1.1 200 OK\r\n");
421         bprintf(fb, "Content-type: text/plain; version=0.0.4\r\n");
422         bprintf(fb, "\r\n");
423         http_answer(fb);
424         http_send(http);
425 }
426
427 static int use_daemon;
428 static int use_debug;
429
430 static struct opt_section options = {
431         OPT_ITEMS {
432                 OPT_HELP("A daemon for controlling the solid state relay module via MQTT"),
433                 OPT_HELP(""),
434                 OPT_HELP("Options:"),
435                 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
436                 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
437                 OPT_HELP_OPTION,
438                 OPT_CONF_OPTIONS,
439                 OPT_END
440         }
441 };
442
443 int main(int argc UNUSED, char **argv)
444 {
445         log_init(argv[0]);
446         opt_parse(&options, argv+1);
447
448         if (use_daemon) {
449                 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
450                 log_set_default_stream(ls);
451         }
452         if (!use_debug)
453                 log_default_stream()->levels &= ~(1U << L_DEBUG);
454
455         mosquitto_lib_init();
456         mosq = mosquitto_new("prometheus", 1, NULL);
457         if (!mosq)
458                 die("Mosquitto: initialization failed");
459
460         mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
461         mosquitto_log_callback_set(mosq, mqtt_log_callback);
462         mosquitto_message_callback_set(mosq, mqtt_msg_callback);
463
464         if (mosquitto_will_set(mosq, "status/prometheus", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
465                 die("Mosquitto: unable to set will");
466
467         if (mosquitto_connect_async(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS)
468                 die("Mosquitto: connect failed");
469
470         if (mosquitto_loop_start(mosq))
471                 die("Mosquitto: cannot start service thread");
472
473         int listen_sk = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
474         if (listen_sk < 0)
475                 die("Cannot create listening socket: %m");
476
477         int one = 1;
478         if (setsockopt(listen_sk, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0)
479                 die("Cannot set SO_REUSEADDR: %m");
480
481         struct sockaddr_in sin = { .sin_family = AF_INET, .sin_port = htons(1422), .sin_addr = INADDR_ANY };
482         if (bind(listen_sk, (const struct sockaddr *) &sin, sizeof(sin)) < 0)
483                 die("Cannot bind listening socket: %m");
484
485         if (listen(listen_sk, 64) < 0)
486                 die("Cannot listen: %m");
487
488         for (;;) {
489                 int sk = accept(listen_sk, NULL, NULL);
490                 if (sk < 0) {
491                         msg(L_ERROR, "HTTP accept failed: %m");
492                         continue;
493                 }
494                 msg(L_DEBUG, "HTTP accepted connection");
495                 struct mempool *mp = mp_new(4096);
496                 struct http *http = mp_alloc_zero(mp, sizeof(*http));
497                 http->mp = mp;
498                 http->sk = sk;
499                 http_connection(http);
500                 mp_delete(mp);
501                 close(sk);
502         }
503 }