]> mj.ucw.cz Git - home-hw.git/blob - prometheus/burrow-prometheus.c
Revert "Prometheus: Circulation and fan"
[home-hw.git] / prometheus / burrow-prometheus.c
1 /*
2  *      A gateway between MQTT and Prometheus
3  *
4  *      (c) 2018 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_machinarium",
78                 .help = "Temperature in the Machinarium [degC]",
79                 .type = "gauge",
80                 .topic = "burrow/temp/machinarium"
81         },
82         {
83                 .metric = "rh_garage",
84                 .help = "Relative humidity in the garage [%]",
85                 .type = "gauge",
86                 .topic = "burrow/temp/garage-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
102 static char *attr_values[ARRAY_SIZE(attr_table)];
103 static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER;
104
105 static void mqtt_publish(const char *topic, const char *fmt, ...)
106 {
107         va_list args;
108         va_start(args, fmt);
109         char m[256];
110         int l = vsnprintf(m, sizeof(m), fmt, args);
111         if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
112                 msg(L_ERROR, "Mosquitto: publish failed");
113         va_end(args);
114 }
115
116 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
117 {
118         if (!status) {
119                 msg(L_DEBUG, "MQTT: Connection established, subscribing");
120                 if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS)
121                         die("Mosquitto: subscribe failed");
122
123                 mqtt_publish("status/prometheus", "ok");
124         }
125 }
126
127 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
128 {
129         // msg(L_INFO, "MQTT(%d): %s", level, message);
130 }
131
132 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
133 {
134         char val[256];
135         if (m->payloadlen >= sizeof(val) - 1) {
136                 msg(L_ERROR, "Invalid value for topic %s", m->topic);
137                 return;
138         }
139         memcpy(val, m->payload, m->payloadlen);
140         val[m->payloadlen] = 0;
141         msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
142
143         for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
144                 if (attr_table[i].topic && !strcmp(attr_table[i].topic, m->topic)) {
145                         pthread_mutex_lock(&attr_mutex);
146                         if (attr_values[i]) {
147                                 xfree(attr_values[i]);
148                                 attr_values[i] = NULL;
149                         }
150                         if (val[0])
151                                 attr_values[i] = xstrdup(val);
152                         pthread_mutex_unlock(&attr_mutex);
153                 }
154         }
155 }
156
157 struct http {
158         struct mempool *mp;
159         int sk;
160         char *iobuf;
161         uint iobuf_pos, iobuf_max;
162         char *linebuf;
163         struct fbpool fbpool;
164 };
165
166 #define IO_TIMEOUT 60000        // in ms
167 #define IOBUF_SIZE 1024
168 #define LINEBUF_SIZE 1024
169
170 static void http_send(struct http *http)
171 {
172         byte *buf = fbpool_end(&http->fbpool);
173         uint len = mp_size(http->mp, buf);
174
175         while (len) {
176                 struct pollfd pfd = { .fd = http->sk, .events = POLLOUT, .revents = 0 };
177                 int res = poll(&pfd, 1, IO_TIMEOUT);
178                 if (res < 0)
179                         die("Poll failed: %m");
180                 if (!res) {
181                         msg(L_ERROR, "HTTP write timed out");
182                         return;
183                 }
184
185                 res = write(http->sk, buf, len);
186                 if (res < 0) {
187                         msg(L_ERROR, "HTTP write failed: %m");
188                         return;
189                 }
190                 buf += res;
191                 len -= res;
192         }
193 }
194
195 static void http_error(struct http *http, const char *err)
196 {
197         msg(L_INFO, "HTTP error: %s", err);
198         fbpool_start(&http->fbpool, http->mp, 0);
199         bprintf(&http->fbpool.fb, "HTTP/1.1 %s\r\n", err);
200         http_send(http);
201 }
202
203 static int http_get_char(struct http *http) {
204         if (http->iobuf_pos >= http->iobuf_max) {
205                 struct pollfd pfd = { .fd = http->sk, .events = POLLIN, .revents = 0 };
206                 int res = poll(&pfd, 1, IO_TIMEOUT);
207                 if (res < 0)
208                         die("Poll failed: %m");
209                 if (!res) {
210                         msg(L_ERROR, "HTTP read timed out");
211                         return -1;
212                 }
213                 int len = read(http->sk, http->iobuf, IOBUF_SIZE);
214                 if (len < 0) {
215                         msg(L_ERROR, "HTTP read error: %m");
216                         return -1;
217                 }
218                 if (!len) {
219                         msg(L_ERROR, "HTTP connection closed");
220                         return -1;
221                 }
222                 http->iobuf_pos = 0;
223                 http->iobuf_max = len;
224         }
225         return http->iobuf[http->iobuf_pos++];
226 }
227
228 static bool http_get_line(struct http *http)
229 {
230         uint i = 0;
231         for (;;) {
232                 int c = http_get_char(http);
233                 if (c < 0)
234                         return false;
235                 if (c == '\n') {
236                         if (i > 0 && http->linebuf[i-1] == '\r')
237                                 i--;
238                         http->linebuf[i] = 0;
239                         return true;
240                 }
241                 if (i >= IOBUF_SIZE-1) {
242                         http_error(http, "400 Line too long");
243                         return false;
244                 }
245                 http->linebuf[i++] = c;
246         }
247 }
248
249 static void http_answer(struct fastbuf *fb)
250 {
251         char val[256], *w[2];
252         time_t now = time(NULL);
253
254         for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
255                 const struct attr *a = &attr_table[i];
256
257                 pthread_mutex_lock(&attr_mutex);
258                 snprintf(val, sizeof(val), "%s", attr_values[i] ? : "");
259                 pthread_mutex_unlock(&attr_mutex);
260
261                 if (!val[0])
262                         continue;
263                 int fields = str_wordsplit(val, w, ARRAY_SIZE(w));
264                 if (fields < 1)
265                         continue;
266                 if (fields >= 2) {
267                         time_t t = atoll(w[1]);
268                         if (t < now - MEASUREMENT_TIMEOUT)
269                                 continue;
270                 }
271
272                 if (a->help)
273                         bprintf(fb, "# HELP %s %s\n", a->metric, a->help);
274                 if (a->type)
275                         bprintf(fb, "# TYPE %s %s\n", a->metric, a->type);
276                 bprintf(fb, "%s %s", a->metric, val);
277 #if 0
278                 // Prometheus does not like our timestamps -- why?
279                 if (fields >= 2)
280                         bprintf(fb, " %s", w[1]);
281 #endif
282                 bputc(fb, '\n');
283         }
284 }
285
286 static void http_connection(struct http *http)
287 {
288         http->iobuf = mp_alloc(http->mp, IOBUF_SIZE);
289         http->linebuf = mp_alloc(http->mp, LINEBUF_SIZE);
290         fbpool_init(&http->fbpool);
291
292         if (!http_get_line(http))
293                 return;
294         // msg(L_DEBUG, "Request line: <%s>", http->linebuf);
295         char *words[3];
296         if (str_wordsplit(http->linebuf, words, 3) != 3) {
297                 http_error(http, "400 Invalid request line");
298                 return;
299         }
300         if (strcmp(words[0], "GET")) {
301                 http_error(http, "501 Method not implemented");
302                 return;
303         }
304
305         for (;;) {
306                 if (!http_get_line(http))
307                         return;
308                 if (!http->linebuf[0])
309                         break;
310                 // msg(L_DEBUG, "Header line: <%s>", http->linebuf);
311         }
312
313         fbpool_start(&http->fbpool, http->mp, 0);
314         struct fastbuf *fb = &http->fbpool.fb;
315         bprintf(fb, "HTTP/1.1 200 OK\r\n");
316         bprintf(fb, "Content-type: text/plain; version=0.0.4\r\n");
317         bprintf(fb, "\r\n");
318         http_answer(fb);
319         http_send(http);
320 }
321
322 static int use_daemon;
323 static int use_debug;
324
325 static struct opt_section options = {
326         OPT_ITEMS {
327                 OPT_HELP("A daemon for controlling the solid state relay module via MQTT"),
328                 OPT_HELP(""),
329                 OPT_HELP("Options:"),
330                 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
331                 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
332                 OPT_HELP_OPTION,
333                 OPT_CONF_OPTIONS,
334                 OPT_END
335         }
336 };
337
338 int main(int argc UNUSED, char **argv)
339 {
340         log_init(argv[0]);
341         opt_parse(&options, argv+1);
342
343         if (use_daemon) {
344                 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
345                 log_set_default_stream(ls);
346         }
347         if (!use_debug)
348                 log_default_stream()->levels &= ~(1U << L_DEBUG);
349
350         mosquitto_lib_init();
351         mosq = mosquitto_new("prometheus", 1, NULL);
352         if (!mosq)
353                 die("Mosquitto: initialization failed");
354
355         mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
356         mosquitto_log_callback_set(mosq, mqtt_log_callback);
357         mosquitto_message_callback_set(mosq, mqtt_msg_callback);
358
359         if (mosquitto_will_set(mosq, "status/prometheus", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
360                 die("Mosquitto: unable to set will");
361
362         if (mosquitto_connect_async(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS)
363                 die("Mosquitto: connect failed");
364
365         if (mosquitto_loop_start(mosq))
366                 die("Mosquitto: cannot start service thread");
367
368         int listen_sk = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
369         if (listen_sk < 0)
370                 die("Cannot create listening socket: %m");
371
372         int one = 1;
373         if (setsockopt(listen_sk, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0)
374                 die("Cannot set SO_REUSEADDR: %m");
375
376         struct sockaddr_in sin = { .sin_family = AF_INET, .sin_port = htons(1422), .sin_addr = INADDR_ANY };
377         if (bind(listen_sk, (const struct sockaddr *) &sin, sizeof(sin)) < 0)
378                 die("Cannot bind listening socket: %m");
379
380         if (listen(listen_sk, 64) < 0)
381                 die("Cannot listen: %m");
382
383         for (;;) {
384                 int sk = accept(listen_sk, NULL, NULL);
385                 if (sk < 0) {
386                         msg(L_ERROR, "HTTP accept failed: %m");
387                         continue;
388                 }
389                 msg(L_DEBUG, "HTTP accepted connection");
390                 struct mempool *mp = mp_new(4096);
391                 struct http *http = mp_alloc_zero(mp, sizeof(*http));
392                 http->mp = mp;
393                 http->sk = sk;
394                 http_connection(http);
395                 mp_delete(mp);
396                 close(sk);
397         }
398 }