influx/burrow-influx.c

   1 /*
   2  *      A gateway between MQTT and InfluxDB
   3  *
   4  *      (c) 2018--2020 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/opt.h>
  11 #include <ucw/string.h>
  12
  13 #include <pthread.h>
  14 #include <stdio.h>
  15 #include <stdlib.h>
  16 #include <string.h>
  17 #include <syslog.h>
  18 #include <time.h>
  19 #include <unistd.h>
  20
  21 #include <curl/curl.h>
  22 #include <mosquitto.h>
  23
  24 #define MEASUREMENT_TIMEOUT 120
  25
  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
  30
  31 static struct mosquitto *mosq;
  32
  33 struct attr {
  34         const char *metric;
  35         const char *value_name;
  36         const char *topic;
  37         uint timeout;
  38         bool is_string;
  39 };
  40
  41 static const struct attr attr_table[] = {
  42         {
  43                 .metric = "temp,where=loft",
  44                 .value_name = "t",
  45                 .topic = "burrow/temp/loft",
  46         },
  47         {
  48                 .metric = "temp,where=ursarium",
  49                 .value_name = "t",
  50                 .topic = "burrow/temp/ursarium"
  51         },
  52         {
  53                 .metric = "temp,where=catarium",
  54                 .value_name = "t",
  55                 .topic = "burrow/temp/catarium"
  56         },
  57         {
  58                 .metric = "temp,where=garage",
  59                 .value_name = "t",
  60                 .topic = "burrow/temp/garage"
  61         },
  62         {
  63                 .metric = "temp,where=terarium",
  64                 .value_name = "t",
  65                 .topic = "burrow/temp/terarium"
  66         },
  67         {
  68                 .metric = "rh,where=ursarium",
  69                 .value_name = "rh",
  70                 .topic = "burrow/temp/ursarium-rh"
  71         },
  72         {
  73                 .metric = "temp,where=catarium_clock",
  74                 .value_name = "t",
  75                 .topic = "burrow/temp/clock"
  76         },
  77         {
  78                 .metric = "pressure,where=catarium_clock",
  79                 .value_name = "pa",
  80                 .topic = "burrow/pressure/clock"
  81         },
  82         {
  83                 .metric = "air,where=inside_intake",
  84                 .value_name = "t",
  85                 .topic = "burrow/air/inside-intake",
  86         },
  87         {
  88                 .metric = "air,where=inside_exhaust",
  89                 .value_name = "t",
  90                 .topic = "burrow/air/inside-exhaust",
  91         },
  92         {
  93                 .metric = "air,where=outside_intake",
  94                 .value_name = "t",
  95                 .topic = "burrow/air/outside-intake",
  96         },
  97         {
  98                 .metric = "air,where=outside_exhaust",
  99                 .value_name = "t",
 100                 .topic = "burrow/air/outside-exhaust",
 101         },
 102         {
 103                 .metric = "air,where=mixed",
 104                 .value_name = "t",
 105                 .topic = "burrow/air/mixed",
 106         },
 107         {
 108                 .metric = "air,where=inside_intake_avg",
 109                 .value_name = "t",
 110                 .topic = "burrow/avg/air/inside-intake",
 111         },
 112         {
 113                 .metric = "air,where=outside_intake_avg",
 114                 .value_name = "t",
 115                 .topic = "burrow/avg/air/outside-intake",
 116         },
 117         {
 118                 .metric = "air_bypass",
 119                 .value_name = "on",
 120                 .topic = "burrow/air/bypass"
 121         },
 122         {
 123                 .metric = "air_fan_pwm",
 124                 .value_name = "pwm",
 125                 .topic = "burrow/air/exchanger-fan"
 126         },
 127         {
 128                 .metric = "loft_fan",
 129                 .value_name = "on",
 130                 .topic = "burrow/loft/fan"
 131         },
 132         {
 133                 .metric = "water_circ",
 134                 .value_name = "on",
 135                 .topic = "burrow/loft/circulation"
 136         },
 137         {
 138                 .metric = "pm_voltage,phase=L1N",
 139                 .value_name = "V",
 140                 .topic = "burrow/power/voltage/l1n",
 141         },
 142         {
 143                 .metric = "pm_voltage,phase=L2N",
 144                 .value_name = "V",
 145                 .topic = "burrow/power/voltage/l2n",
 146         },
 147         {
 148                 .metric = "pm_voltage,phase=L3N",
 149                 .value_name = "V",
 150                 .topic = "burrow/power/voltage/l3n",
 151         },
 152         {
 153                 .metric = "pm_current,phase=L1",
 154                 .value_name = "A",
 155                 .topic = "burrow/power/current/l1",
 156         },
 157         {
 158                 .metric = "pm_current,phase=L2",
 159                 .value_name = "A",
 160                 .topic = "burrow/power/current/l2",
 161         },
 162         {
 163                 .metric = "pm_current,phase=L3",
 164                 .value_name = "A",
 165                 .topic = "burrow/power/current/l3",
 166         },
 167         {
 168                 .metric = "pm_power",
 169                 .value_name = "W",
 170                 .topic = "burrow/power/power",
 171         },
 172         {
 173                 .metric = "pm_energy",
 174                 .value_name = "kWh",
 175                 .topic = "burrow/power/energy",
 176         },
 177         {
 178                 .metric = "pm_reactive_power",
 179                 .value_name = "VAr",
 180                 .topic = "burrow/power/reactive/power",
 181         },
 182         {
 183                 .metric = "pm_reactive_energy",
 184                 .value_name = "kVArh",
 185                 .topic = "burrow/power/reactive/energy",
 186         },
 187         {
 188                 .metric = "heating_water_pressure",
 189                 .value_name = "p",
 190                 .topic = "burrow/heating/water-pressure",
 191                 .timeout = 300,
 192         },
 193         {
 194                 .metric = "heating_outside_temp",
 195                 .value_name = "t",
 196                 .topic = "burrow/heating/outside-temp",
 197                 .timeout = 660,
 198         },
 199         {
 200                 .metric = "heating_room_temp,circuit=1",
 201                 .value_name = "t",
 202                 .topic = "burrow/heating/circuit1/room-temp",
 203                 .timeout = 300,
 204         },
 205         {
 206                 .metric = "heating_mix-temp,circuit=1",
 207                 .value_name = "t",
 208                 .topic = "burrow/heating/circuit1/mix-temp",
 209                 .timeout = 300,
 210         },
 211         {
 212                 .metric = "heating_mix-valve,circuit=1",
 213                 .value_name = "x",
 214                 .topic = "burrow/heating/circuit1/mix-valve",
 215                 .timeout = 300,
 216         },
 217         {
 218                 .metric = "heating_active,circuit=1",
 219                 .value_name = "x",
 220                 .topic = "burrow/heating/circuit1/active",
 221                 .timeout = 660,
 222         },
 223         {
 224                 .metric = "heating_pump_active,circuit=1",
 225                 .value_name = "x",
 226                 .topic = "burrow/heating/circuit1/pump",
 227                 .timeout = 300,
 228         },
 229         {
 230                 .metric = "heating_room_temp,circuit=2",
 231                 .value_name = "t",
 232                 .topic = "burrow/heating/circuit2/room-temp",
 233                 .timeout = 300,
 234         },
 235         {
 236                 .metric = "heating_active,circuit=2",
 237                 .value_name = "x",
 238                 .topic = "burrow/heating/circuit2/active",
 239                 .timeout = 660,
 240         },
 241         {
 242                 .metric = "heating_active,circuit=water",
 243                 .value_name = "x",
 244                 .topic = "burrow/heating/water/active",
 245                 .timeout = 660,
 246         },
 247         {
 248                 .metric = "heating_error",
 249                 .value_name = "err",
 250                 .topic = "burrow/heating/error",
 251                 .timeout = 300,
 252         },
 253         {
 254                 .metric = "heating_clock",
 255                 .value_name = "t",
 256                 .topic = "burrow/heating/clock",
 257                 .timeout = 660,
 258                 .is_string = true,
 259         },
 260 };
 261
 262 /*** MQTT ***/
 263
 264 static char *attr_values[ARRAY_SIZE(attr_table)];
 265 static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER;
 266
 267 static void mqtt_publish(const char *topic, const char *fmt, ...)
 268 {
 269         va_list args;
 270         va_start(args, fmt);
 271         char m[256];
 272         int l = vsnprintf(m, sizeof(m), fmt, args);
 273         if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
 274                 msg(L_ERROR, "Mosquitto: publish failed");
 275         va_end(args);
 276 }
 277
 278 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
 279 {
 280         if (!status) {
 281                 msg(L_DEBUG, "MQTT: Connection established, subscribing");
 282                 if (mosquitto_subscribe(mosq, NULL, "burrow/#", 1) != MOSQ_ERR_SUCCESS)
 283                         die("Mosquitto: subscribe failed");
 284
 285                 mqtt_publish("status/influx", "ok");
 286         }
 287 }
 288
 289 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
 290 {
 291         // msg(L_INFO, "MQTT(%d): %s", level, message);
 292 }
 293
 294 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
 295 {
 296         char val[256];
 297         if (m->payloadlen >= sizeof(val) - 1) {
 298                 msg(L_ERROR, "Invalid value for topic %s", m->topic);
 299                 return;
 300         }
 301         memcpy(val, m->payload, m->payloadlen);
 302         val[m->payloadlen] = 0;
 303         msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
 304
 305         for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
 306                 if (attr_table[i].topic && !strcmp(attr_table[i].topic, m->topic)) {
 307                         pthread_mutex_lock(&attr_mutex);
 308                         if (attr_values[i]) {
 309                                 xfree(attr_values[i]);
 310                                 attr_values[i] = NULL;
 311                         }
 312                         if (val[0])
 313                                 attr_values[i] = xstrdup(val);
 314                         pthread_mutex_unlock(&attr_mutex);
 315                 }
 316         }
 317 }
 318
 319 /*** InfluxDB ***/
 320
 321 static CURL *curl;
 322
 323 static struct fastbuf *influx_fb;
 324 static struct fastbuf *influx_err_fb;
 325
 326 static size_t influx_write_callback(char *ptr, size_t size, size_t nmemb, void *userdata UNUSED)
 327 {
 328         ASSERT(size == 1);
 329         for (size_t i=0; i<nmemb; i++) {
 330                 int c = *(byte *)ptr;
 331                 ptr++;
 332                 if (c == '\r')
 333                         continue;
 334                 if (c == '\n')
 335                         c = ' ';
 336                 else if (c < 0x20 || c > 0x7e)
 337                         c = '?';
 338                 bputc(influx_err_fb, c);
 339         }
 340         return nmemb;
 341 }
 342
 343 static void influx_init(void)
 344 {
 345         curl = curl_easy_init();
 346         if (!curl)
 347                 die("libcurl init failed");
 348
 349         curl_easy_setopt(curl, CURLOPT_URL, INFLUX_URL);
 350         curl_easy_setopt(curl, CURLOPT_VERBOSE, (long) INFLUX_VERBOSE);
 351         curl_easy_setopt(curl, CURLOPT_TIMEOUT, (long) INFLUX_TIMEOUT);
 352         curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, influx_write_callback);
 353
 354         influx_fb = fbgrow_create(4096);
 355         influx_err_fb = fbgrow_create(256);
 356 }
 357
 358 static struct fastbuf *influx_write_start(void)
 359 {
 360         fbgrow_reset(influx_fb);
 361         return influx_fb;
 362 }
 363
 364 static void influx_write_flush(void)
 365 {
 366         bputc(influx_fb, 0);
 367         byte *buf;
 368         fbgrow_get_buf(influx_fb, &buf);
 369         curl_easy_setopt(curl, CURLOPT_POSTFIELDS, buf);
 370
 371         msg(L_DEBUG, "InfluxDB: Sending data (%u bytes)", strlen(buf));
 372         if (INFLUX_VERBOSE)
 373                 fprintf(stderr, "%s", buf);
 374
 375         fbgrow_reset(influx_err_fb);
 376
 377         CURLcode code = curl_easy_perform(curl);
 378
 379         if (code != CURLE_OK) {
 380                 msg(L_ERROR, "Write to InfluxDB failed: %s", curl_easy_strerror(code));
 381                 return;
 382         }
 383
 384         long rc = 0;
 385         curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &rc);
 386         if (rc != 200 && rc != 204) {
 387                 bputc(influx_err_fb, 0);
 388                 byte *err;
 389                 fbgrow_get_buf(influx_err_fb, &err);
 390                 msg(L_DEBUG, "InfluxDB HTTP error %ld: %s [...]", rc, err);
 391         }
 392 }
 393
 394 /*** Main ***/
 395
 396 static int use_daemon;
 397 static int use_debug;
 398
 399 static struct opt_section options = {
 400         OPT_ITEMS {
 401                 OPT_HELP("A daemon for transferring MQTT data to InfluxDB"),
 402                 OPT_HELP(""),
 403                 OPT_HELP("Options:"),
 404                 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
 405                 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
 406                 OPT_HELP_OPTION,
 407                 OPT_CONF_OPTIONS,
 408                 OPT_END
 409         }
 410 };
 411
 412 int main(int argc UNUSED, char **argv)
 413 {
 414         log_init(argv[0]);
 415         opt_parse(&options, argv+1);
 416
 417         if (use_daemon) {
 418                 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
 419                 log_set_default_stream(ls);
 420         }
 421         if (!use_debug)
 422                 log_default_stream()->levels &= ~(1U << L_DEBUG);
 423
 424         mosquitto_lib_init();
 425         mosq = mosquitto_new("influx", 1, NULL);
 426         if (!mosq)
 427                 die("Mosquitto: initialization failed");
 428
 429         mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
 430         mosquitto_log_callback_set(mosq, mqtt_log_callback);
 431         mosquitto_message_callback_set(mosq, mqtt_msg_callback);
 432
 433         if (mosquitto_will_set(mosq, "status/influx", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
 434                 die("Mosquitto: unable to set will");
 435
 436         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)
 437                 die("Mosquitto: unable to set TLS parameters");
 438
 439         if (mosquitto_connect_async(mosq, "burrow-mqtt", 8883, 60) != MOSQ_ERR_SUCCESS)
 440                 die("Mosquitto: connect failed");
 441
 442         if (mosquitto_loop_start(mosq))
 443                 die("Mosquitto: cannot start service thread");
 444
 445         influx_init();
 446
 447         for (;;) {
 448                 struct fastbuf *f = influx_write_start();
 449                 char val[256], *w[2];
 450                 time_t now = time(NULL);
 451
 452                 for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
 453                         const struct attr *a = &attr_table[i];
 454
 455                         pthread_mutex_lock(&attr_mutex);
 456                         snprintf(val, sizeof(val), "%s", attr_values[i] ? : "");
 457                         pthread_mutex_unlock(&attr_mutex);
 458
 459                         if (!val[0])
 460                                 continue;
 461                         int fields = str_wordsplit(val, w, ARRAY_SIZE(w));
 462                         if (fields < 1)
 463                                 continue;
 464                         if (fields >= 2) {
 465                                 time_t t = atoll(w[1]);
 466                                 uint timeout = a->timeout ? : MEASUREMENT_TIMEOUT;
 467                                 if (t < now - timeout)
 468                                         continue;
 469                         }
 470
 471                         if (!a->is_string)
 472                                 bprintf(f, "%s %s=%s\n", a->metric, a->value_name, val);
 473                         else
 474                                 bprintf(f, "%s %s=\"%s\"\n", a->metric, a->value_name, val);
 475                 }
 476                 influx_write_flush();
 477                 sleep(INFLUX_INTERVAL);
 478         }
 479 }