src/mqtt.c

   1 /**
   2  * collectd - src/mqtt.c
   3  * Copyright (C) 2014       Marc Falzon <marc at baha dot mu>
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a
   6  * copy of this software and associated documentation files (the "Software"),
   7  * to deal in the Software without restriction, including without limitation
   8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9  * and/or sell copies of the Software, and to permit persons to whom the
  10  * Software is furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  21  * DEALINGS IN THE SOFTWARE.
  22  **/
  23
  24 // Reference: http://mosquitto.org/api/files/mosquitto-h.html
  25
  26
  27 #include "collectd.h"
  28 #include "common.h"
  29 #include "plugin.h"
  30 #include "utils_cache.h"
  31 #include "utils_complain.h"
  32
  33 #include <pthread.h>
  34
  35 #include <mosquitto.h>
  36
  37 #define MQTT_MAX_TOPIC_SIZE         1024
  38 #define MQTT_MAX_MESSAGE_SIZE       MQTT_MAX_TOPIC_SIZE + 1024
  39 #define MQTT_DEFAULT_HOST           "localhost"
  40 #define MQTT_DEFAULT_PORT           1883
  41 #define MQTT_DEFAULT_CLIENT_ID      "collectd"
  42 #define MQTT_DEFAULT_TOPIC_PREFIX   "collectd"
  43
  44 /*
  45  * Data types
  46  */
  47 struct mqtt_client_conf
  48 {
  49     char               *name;
  50
  51     struct mosquitto   *mosq;
  52     _Bool               connected;
  53
  54     char               *host;
  55     int                 port;
  56     char               *username;
  57     char               *password;
  58
  59     char               *client_id;
  60     char               *topic_prefix;
  61     _Bool               store_rates;
  62     _Bool               retain;
  63     int qos;
  64
  65     c_complain_t        complaint_cantpublish;
  66     pthread_mutex_t     lock;
  67 };
  68 typedef struct mqtt_client_conf mqtt_client_conf_t;
  69
  70 static char const *mosquitto_strerror (int code)
  71 {
  72     switch (code)
  73     {
  74         case MOSQ_ERR_SUCCESS: return "MOSQ_ERR_SUCCESS";
  75         case MOSQ_ERR_NOMEM: return "MOSQ_ERR_NOMEM";
  76         case MOSQ_ERR_PROTOCOL: return "MOSQ_ERR_PROTOCOL";
  77         case MOSQ_ERR_INVAL: return "MOSQ_ERR_INVAL";
  78         case MOSQ_ERR_NO_CONN: return "MOSQ_ERR_NO_CONN";
  79         case MOSQ_ERR_CONN_REFUSED: return "MOSQ_ERR_CONN_REFUSED";
  80         case MOSQ_ERR_NOT_FOUND: return "MOSQ_ERR_NOT_FOUND";
  81         case MOSQ_ERR_CONN_LOST: return "MOSQ_ERR_CONN_LOST";
  82         case MOSQ_ERR_SSL: return "MOSQ_ERR_SSL";
  83         case MOSQ_ERR_PAYLOAD_SIZE: return "MOSQ_ERR_PAYLOAD_SIZE";
  84         case MOSQ_ERR_NOT_SUPPORTED: return "MOSQ_ERR_NOT_SUPPORTED";
  85         case MOSQ_ERR_AUTH: return "MOSQ_ERR_AUTH";
  86         case MOSQ_ERR_ACL_DENIED: return "MOSQ_ERR_ACL_DENIED";
  87         case MOSQ_ERR_UNKNOWN: return "MOSQ_ERR_UNKNOWN";
  88         case MOSQ_ERR_ERRNO: return "MOSQ_ERR_ERRNO";
  89     }
  90
  91     return "UNKNOWN ERROR CODE";
  92 }
  93
  94 static void mqtt_free (mqtt_client_conf_t *conf)
  95 {
  96     if (conf == NULL)
  97         return;
  98
  99     if (conf->connected)
 100         (void) mosquitto_disconnect (conf->mosq);
 101     conf->connected = 0;
 102     (void) mosquitto_destroy (conf->mosq);
 103
 104     sfree (conf->host);
 105     sfree (conf->username);
 106     sfree (conf->password);
 107     sfree (conf->client_id);
 108     sfree (conf->topic_prefix);
 109     sfree (conf);
 110 }
 111
 112 /*
 113  * Functions
 114  */
 115 /* must hold conf->lock when calling. */
 116 static int mqtt_reconnect_broker (mqtt_client_conf_t *conf)
 117 {
 118     int status;
 119
 120     if (conf->connected)
 121         return (0);
 122
 123     status = mosquitto_reconnect (conf->mosq);
 124     if (status != MOSQ_ERR_SUCCESS)
 125     {
 126         char errbuf[1024];
 127         ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
 128                 (status == MOSQ_ERR_ERRNO)
 129                 ? sstrerror(errno, errbuf, sizeof (errbuf))
 130                 : mosquitto_strerror (status));
 131         return (-1);
 132     }
 133
 134     conf->connected = 1;
 135
 136     c_release (LOG_INFO,
 137             &conf->complaint_cantpublish,
 138             "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
 139             conf->host, conf->port);
 140
 141     return (0);
 142 } /* mqtt_reconnect_broker */
 143
 144 static int publish (mqtt_client_conf_t *conf, char const *topic,
 145     void const *payload, size_t payload_len)
 146 {
 147     int status;
 148
 149     pthread_mutex_lock (&conf->lock);
 150
 151     status = mqtt_reconnect_broker (conf);
 152     if (status != 0) {
 153         pthread_mutex_unlock (&conf->lock);
 154         ERROR ("mqtt plugin: unable to reconnect to broker");
 155         return (status);
 156     }
 157
 158     status = mosquitto_publish(conf->mosq,
 159             /* message id */ NULL,
 160             topic,
 161             (uint32_t) payload_len, payload,
 162             /* qos */ conf->qos,
 163             /* retain */ conf->retain);
 164     if (status != MOSQ_ERR_SUCCESS)
 165     {
 166         char errbuf[1024];
 167         c_complain (LOG_ERR,
 168                 &conf->complaint_cantpublish,
 169                 "plugin mqtt: mosquitto_publish failed: %s",
 170                 status == MOSQ_ERR_ERRNO ?
 171                 sstrerror(errno, errbuf, sizeof (errbuf)) :
 172                 mosquitto_strerror(status));
 173         /* Mark our connection "down" regardless of the error as a safety
 174          * measure; we will try to reconnect the next time we have to publish a
 175          * message */
 176         conf->connected = 0;
 177
 178         pthread_mutex_unlock (&conf->lock);
 179         return (-1);
 180     }
 181
 182     pthread_mutex_unlock (&conf->lock);
 183     return (0);
 184 } /* int publish */
 185
 186 static int format_topic (char *buf, size_t buf_len,
 187     data_set_t const *ds, value_list_t const *vl,
 188     mqtt_client_conf_t *conf)
 189 {
 190     char name[MQTT_MAX_TOPIC_SIZE];
 191     int status;
 192
 193     if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
 194         return (FORMAT_VL (buf, buf_len, vl));
 195
 196     status = FORMAT_VL (name, sizeof (name), vl);
 197     if (status != 0)
 198         return (status);
 199
 200     status = ssnprintf (buf, buf_len, "%s/%s", conf->topic_prefix, name);
 201     if ((status < 0) || (((size_t) status) >= buf_len))
 202         return (ENOMEM);
 203
 204     return (0);
 205 } /* int format_topic */
 206
 207 static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
 208     user_data_t *user_data)
 209 {
 210     mqtt_client_conf_t *conf;
 211     char topic[MQTT_MAX_TOPIC_SIZE];
 212     char payload[MQTT_MAX_MESSAGE_SIZE];
 213     int status = 0;
 214
 215     if ((user_data == NULL) || (user_data->data == NULL))
 216         return (EINVAL);
 217     conf = user_data->data;
 218
 219     status = format_topic (topic, sizeof (topic), ds, vl, conf);
 220     if (status != 0)
 221     {
 222         ERROR ("mqtt plugin: format_topic failed with status %d.", status);
 223         return (status);
 224     }
 225
 226     status = format_values (payload, sizeof (payload),
 227             ds, vl, conf->store_rates);
 228     if (status != 0)
 229     {
 230         ERROR ("mqtt plugin: format_values failed with status %d.", status);
 231         return (status);
 232     }
 233
 234     status = publish (conf, topic, payload, sizeof (payload));
 235     if (status != 0)
 236     {
 237         ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
 238         return (status);
 239     }
 240
 241     return (status);
 242 } /* mqtt_write */
 243
 244 /*
 245  * <Publish "name">
 246  *   Host "example.com"
 247  *   Port 1883
 248  *   Prefix "collectd"
 249  *   ClientId "collectd"
 250  *   User "guest"
 251  *   Password "secret"
 252  *   StoreRates true
 253  *   Retain false
 254  *   QoS 0
 255  * </Publish>
 256  */
 257 static int mqtt_config_broker (oconfig_item_t *ci)
 258 {
 259     mqtt_client_conf_t *conf;
 260     user_data_t user_data;
 261     int status;
 262     int i;
 263
 264     conf = calloc (1, sizeof (*conf));
 265     if (conf == NULL)
 266     {
 267         ERROR ("mqtt plugin: malloc failed.");
 268         return (-1);
 269     }
 270
 271     conf->name = NULL;
 272     status = cf_util_get_string (ci, &conf->name);
 273     if (status != 0)
 274     {
 275         mqtt_free (conf);
 276         return (status);
 277     }
 278
 279     conf->host = strdup (MQTT_DEFAULT_HOST);
 280     conf->port = MQTT_DEFAULT_PORT;
 281     conf->client_id = strdup (MQTT_DEFAULT_CLIENT_ID);
 282     conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
 283     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 284
 285     for (i = 0; i < ci->children_num; i++)
 286     {
 287         oconfig_item_t *child = ci->children + i;
 288         if (strcasecmp ("Host", child->key) == 0)
 289             cf_util_get_string (child, &conf->host);
 290         else if (strcasecmp ("Port", child->key) == 0)
 291         {
 292             int tmp = cf_util_get_port_number (child);
 293             if (tmp < 0)
 294                 ERROR ("mqtt plugin: Invalid port number.");
 295             else
 296                 conf->port = tmp;
 297         }
 298         else if (strcasecmp ("Prefix", child->key) == 0)
 299             cf_util_get_string (child, &conf->topic_prefix);
 300         else if (strcasecmp ("ClientId", child->key) == 0)
 301             cf_util_get_string (child, &conf->client_id);
 302         else if (strcasecmp ("User", child->key) == 0)
 303             cf_util_get_string (child, &conf->username);
 304         else if (strcasecmp ("Password", child->key) == 0)
 305             cf_util_get_string (child, &conf->password);
 306         else if (strcasecmp ("StoreRates", child->key) == 0)
 307             cf_util_get_boolean (child, &conf->store_rates);
 308         else if (strcasecmp ("Retain", child->key) == 0)
 309             cf_util_get_boolean (child, &conf->retain);
 310         else if (strcasecmp ("QoS", child->key) == 0)
 311         {
 312             int tmp = -1;
 313             status = cf_util_get_int (child, &tmp);
 314             if ((status != 0) || (tmp < 0) || (tmp > 2))
 315                 ERROR ("mqtt plugin: Not a valid QoS setting.");
 316             else
 317                 conf->qos = tmp;
 318         }
 319         else
 320             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 321     }
 322
 323     memset (&user_data, 0, sizeof (user_data));
 324     user_data.data = conf;
 325
 326     conf->mosq = mosquitto_new (conf->client_id, /* user data = */ conf);
 327     if (conf->mosq == NULL)
 328     {
 329         ERROR ("mqtt plugin: mosquitto_new failed");
 330         mqtt_free (conf);
 331         return (-1);
 332     }
 333
 334     if (conf->username && conf->password)
 335     {
 336         status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password);
 337         if (status != MOSQ_ERR_SUCCESS)
 338         {
 339             char errbuf[1024];
 340             ERROR ("mqtt plugin: mosquitto_username_pw_set failed: %s",
 341                     (status == MOSQ_ERR_ERRNO)
 342                     ? sstrerror (errno, errbuf, sizeof (errbuf))
 343                     : mosquitto_strerror (status));
 344             mqtt_free (conf);
 345             return (-1);
 346         }
 347     }
 348
 349     status = mosquitto_connect (conf->mosq, conf->host, conf->port,
 350             /* keepalive = */ 10, /* clean session = */ 1);
 351     if (status != MOSQ_ERR_SUCCESS)
 352     {
 353         char errbuf[1024];
 354         ERROR ("mqtt plugin: mosquitto_connect failed: %s",
 355                 (status == MOSQ_ERR_ERRNO)
 356                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 357                 : mosquitto_strerror (status));
 358         mqtt_free (conf);
 359         return (-1);
 360     }
 361
 362     DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"",
 363         conf->host, conf->port);
 364
 365     conf->connected = 1;
 366
 367     plugin_register_write ("mqtt", mqtt_write, &user_data);
 368
 369     return (0);
 370 } /* mqtt_config_broker */
 371
 372 /*
 373  * <Plugin mqtt>
 374  *   <Publish "name">
 375  *     # ...
 376  *   </Publish>
 377  * </Plugin>
 378  */
 379 static int mqtt_config (oconfig_item_t *ci)
 380 {
 381     int i;
 382
 383     for (i = 0; i < ci->children_num; i++)
 384     {
 385         oconfig_item_t *child = ci->children + i;
 386
 387         if (strcasecmp ("Publish", child->key) == 0)
 388             mqtt_config_broker (child);
 389         else
 390             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 391     }
 392
 393     return (0);
 394 } /* int mqtt_config */
 395
 396 static int mqtt_init (void)
 397 {
 398     mosquitto_lib_init();
 399
 400     return (0);
 401 } /* mqtt_init */
 402
 403 void module_register (void)
 404 {
 405     plugin_register_complex_config ("mqtt", mqtt_config);
 406     plugin_register_init ("mqtt", mqtt_init);
 407 } /* void module_register */
 408
 409 /* vim: set sw=4 sts=4 et fdm=marker : */