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=kitchen",
  64                 .value_name = "t",
  65                 .topic = "burrow/temp/kitchen"
  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 #if 0
 128         {
 129                 .metric = "loft_fan",
 130                 .topic = "burrow/loft/fan"
 131         },
 132 #endif
 133         {
 134                 .metric = "water_circ",
 135                 .value_name = "on",
 136                 .topic = "burrow/loft/circulation"
 137         },
 138         {
 139                 .metric = "pm_voltage,phase=L1N",
 140                 .value_name = "V",
 141                 .topic = "burrow/power/voltage/l1n",
 142         },
 143         {
 144                 .metric = "pm_voltage,phase=L2N",
 145                 .value_name = "V",
 146                 .topic = "burrow/power/voltage/l2n",
 147         },
 148         {
 149                 .metric = "pm_voltage,phase=L3N",
 150                 .value_name = "V",
 151                 .topic = "burrow/power/voltage/l3n",
 152         },
 153         {
 154                 .metric = "pm_current,phase=L1",
 155                 .value_name = "A",
 156                 .topic = "burrow/power/current/l1",
 157         },
 158         {
 159                 .metric = "pm_current,phase=L2",
 160                 .value_name = "A",
 161                 .topic = "burrow/power/current/l2",
 162         },
 163         {
 164                 .metric = "pm_current,phase=L3",
 165                 .value_name = "A",
 166                 .topic = "burrow/power/current/l3",
 167         },
 168         {
 169                 .metric = "pm_power",
 170                 .value_name = "W",
 171                 .topic = "burrow/power/power",
 172         },
 173         {
 174                 .metric = "pm_energy",
 175                 .value_name = "kWh",
 176                 .topic = "burrow/power/energy",
 177         },
 178         {
 179                 .metric = "pm_reactive_power",
 180                 .value_name = "VAr",
 181                 .topic = "burrow/power/reactive/power",
 182         },
 183         {
 184                 .metric = "pm_reactive_energy",
 185                 .value_name = "kVArh",
 186                 .topic = "burrow/power/reactive/energy",
 187         },
 188         {
 189                 .metric = "heating_water_pressure",
 190                 .value_name = "p",
 191                 .topic = "burrow/heating/water-pressure",
 192                 .timeout = 300,
 193         },
 194         {
 195                 .metric = "heating_outside_temp",
 196                 .value_name = "t",
 197                 .topic = "burrow/heating/outside-temp",
 198                 .timeout = 660,
 199         },
 200         {
 201                 .metric = "heating_room_temp,circuit=1",
 202                 .value_name = "t",
 203                 .topic = "burrow/heating/circuit1/room-temp",
 204                 .timeout = 300,
 205         },
 206         {
 207                 .metric = "heating_mix-temp,circuit=1",
 208                 .value_name = "t",
 209                 .topic = "burrow/heating/circuit1/mix-temp",
 210                 .timeout = 300,
 211         },
 212         {
 213                 .metric = "heating_mix-valve,circuit=1",
 214                 .value_name = "x",
 215                 .topic = "burrow/heating/circuit1/mix-valve",
 216                 .timeout = 300,
 217         },
 218         {
 219                 .metric = "heating_active,circuit=1",
 220                 .value_name = "x",
 221                 .topic = "burrow/heating/circuit1/active",
 222                 .timeout = 660,
 223         },
 224         {
 225                 .metric = "heating_pump_active,circuit=1",
 226                 .value_name = "x",
 227                 .topic = "burrow/heating/circuit1/pump",
 228                 .timeout = 300,
 229         },
 230         {
 231                 .metric = "heating_room_temp,circuit=2",
 232                 .value_name = "t",
 233                 .topic = "burrow/heating/circuit2/room-temp",
 234                 .timeout = 300,
 235         },
 236         {
 237                 .metric = "heating_active,circuit=2",
 238                 .value_name = "x",
 239                 .topic = "burrow/heating/circuit2/active",
 240                 .timeout = 660,
 241         },
 242         {
 243                 .metric = "heating_active,circuit=water",
 244                 .value_name = "x",
 245                 .topic = "burrow/heating/water/active",
 246                 .timeout = 660,
 247         },
 248         {
 249                 .metric = "heating_error",
 250                 .value_name = "err",
 251                 .topic = "burrow/heating/error",
 252                 .timeout = 300,
 253         },
 254         {
 255                 .metric = "heating_clock",
 256                 .value_name = "t",
 257                 .topic = "burrow/heating/clock",
 258                 .timeout = 660,
 259                 .is_string = true,
 260         },
 261 };
 262
 263 /*** MQTT ***/
 264
 265 static char *attr_values[ARRAY_SIZE(attr_table)];
 266 static pthread_mutex_t attr_mutex = PTHREAD_MUTEX_INITIALIZER;
 267
 268 static void mqtt_publish(const char *topic, const char *fmt, ...)
 269 {
 270         va_list args;
 271         va_start(args, fmt);
 272         char m[256];
 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");
 276         va_end(args);
 277 }
 278
 279 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
 280 {
 281         if (!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");
 285
 286                 mqtt_publish("status/influx", "ok");
 287         }
 288 }
 289
 290 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
 291 {
 292         // msg(L_INFO, "MQTT(%d): %s", level, message);
 293 }
 294
 295 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
 296 {
 297         char val[256];
 298         if (m->payloadlen >= sizeof(val) - 1) {
 299                 msg(L_ERROR, "Invalid value for topic %s", m->topic);
 300                 return;
 301         }
 302         memcpy(val, m->payload, m->payloadlen);
 303         val[m->payloadlen] = 0;
 304         msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
 305
 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;
 312                         }
 313                         if (val[0])
 314                                 attr_values[i] = xstrdup(val);
 315                         pthread_mutex_unlock(&attr_mutex);
 316                 }
 317         }
 318 }
 319
 320 /*** InfluxDB ***/
 321
 322 static CURL *curl;
 323
 324 static struct fastbuf *influx_fb;
 325 static struct fastbuf *influx_err_fb;
 326
 327 static size_t influx_write_callback(char *ptr, size_t size, size_t nmemb, void *userdata UNUSED)
 328 {
 329         ASSERT(size == 1);
 330         for (size_t i=0; i<nmemb; i++) {
 331                 int c = *(byte *)ptr;
 332                 ptr++;
 333                 if (c == '\r')
 334                         continue;
 335                 if (c == '\n')
 336                         c = ' ';
 337                 else if (c < 0x20 || c > 0x7e)
 338                         c = '?';
 339                 bputc(influx_err_fb, c);
 340         }
 341         return nmemb;
 342 }
 343
 344 static void influx_init(void)
 345 {
 346         curl = curl_easy_init();
 347         if (!curl)
 348                 die("libcurl init failed");
 349
 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);
 354
 355         influx_fb = fbgrow_create(4096);
 356         influx_err_fb = fbgrow_create(256);
 357 }
 358
 359 static struct fastbuf *influx_write_start(void)
 360 {
 361         fbgrow_reset(influx_fb);
 362         return influx_fb;
 363 }
 364
 365 static void influx_write_flush(void)
 366 {
 367         bputc(influx_fb, 0);
 368         byte *buf;
 369         fbgrow_get_buf(influx_fb, &buf);
 370         curl_easy_setopt(curl, CURLOPT_POSTFIELDS, buf);
 371
 372         msg(L_DEBUG, "InfluxDB: Sending data (%u bytes)", strlen(buf));
 373         if (INFLUX_VERBOSE)
 374                 fprintf(stderr, "%s", buf);
 375
 376         fbgrow_reset(influx_err_fb);
 377
 378         CURLcode code = curl_easy_perform(curl);
 379
 380         if (code != CURLE_OK) {
 381                 msg(L_ERROR, "Write to InfluxDB failed: %s", curl_easy_strerror(code));
 382                 return;
 383         }
 384
 385         long rc = 0;
 386         curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &rc);
 387         if (rc != 200 && rc != 204) {
 388                 bputc(influx_err_fb, 0);
 389                 byte *err;
 390                 fbgrow_get_buf(influx_err_fb, &err);
 391                 msg(L_DEBUG, "InfluxDB HTTP error %ld: %s [...]", rc, err);
 392         }
 393 }
 394
 395 /*** Main ***/
 396
 397 static int use_daemon;
 398 static int use_debug;
 399
 400 static struct opt_section options = {
 401         OPT_ITEMS {
 402                 OPT_HELP("A daemon for transferring MQTT data to InfluxDB"),
 403                 OPT_HELP(""),
 404                 OPT_HELP("Options:"),
 405                 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
 406                 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
 407                 OPT_HELP_OPTION,
 408                 OPT_CONF_OPTIONS,
 409                 OPT_END
 410         }
 411 };
 412
 413 int main(int argc UNUSED, char **argv)
 414 {
 415         log_init(argv[0]);
 416         opt_parse(&options, argv+1);
 417
 418         if (use_daemon) {
 419                 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
 420                 log_set_default_stream(ls);
 421         }
 422         if (!use_debug)
 423                 log_default_stream()->levels &= ~(1U << L_DEBUG);
 424
 425         mosquitto_lib_init();
 426         mosq = mosquitto_new("influx", 1, NULL);
 427         if (!mosq)
 428                 die("Mosquitto: initialization failed");
 429
 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);
 433
 434         if (mosquitto_will_set(mosq, "status/influx", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
 435                 die("Mosquitto: unable to set will");
 436
 437         if (mosquitto_connect_async(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS)
 438                 die("Mosquitto: connect failed");
 439
 440         if (mosquitto_loop_start(mosq))
 441                 die("Mosquitto: cannot start service thread");
 442
 443         influx_init();
 444
 445         for (;;) {
 446                 struct fastbuf *f = influx_write_start();
 447                 char val[256], *w[2];
 448                 time_t now = time(NULL);
 449
 450                 for (uint i=0; i < ARRAY_SIZE(attr_table); i++) {
 451                         const struct attr *a = &attr_table[i];
 452
 453                         pthread_mutex_lock(&attr_mutex);
 454                         snprintf(val, sizeof(val), "%s", attr_values[i] ? : "");
 455                         pthread_mutex_unlock(&attr_mutex);
 456
 457                         if (!val[0])
 458                                 continue;
 459                         int fields = str_wordsplit(val, w, ARRAY_SIZE(w));
 460                         if (fields < 1)
 461                                 continue;
 462                         if (fields >= 2) {
 463                                 time_t t = atoll(w[1]);
 464                                 uint timeout = a->timeout ? : MEASUREMENT_TIMEOUT;
 465                                 if (t < now - timeout)
 466                                         continue;
 467                         }
 468
 469                         if (!a->is_string)
 470                                 bprintf(f, "%s %s=%s\n", a->metric, a->value_name, val);
 471                         else
 472                                 bprintf(f, "%s %s=\"%s\"\n", a->metric, a->value_name, val);
 473                 }
 474                 influx_write_flush();
 475                 sleep(INFLUX_INTERVAL);
 476         }
 477 }