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     struct mosquitto    *mosq;
  50     bool                connected;
  51     char                *host;
  52     int                 port;
  53     char                *client_id;
  54     char                *topic_prefix;
  55     c_complain_t        complaint_cantpublish;
  56     pthread_mutex_t     lock;
  57 };
  58
  59 /*
  60  * Functions
  61  */
  62 static int mqtt_reconnect_broker (struct mqtt_client_conf *conf)
  63 {
  64     int status;
  65
  66     if (conf->connected)
  67         return (0);
  68
  69     pthread_mutex_lock (&conf->lock);
  70
  71     status = mosquitto_reconnect (conf->mosq);
  72
  73     if (status != MOSQ_ERR_SUCCESS) {
  74         ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
  75             (status == MOSQ_ERR_ERRNO ?
  76                 strerror(errno) : mosquitto_strerror (status)));
  77         pthread_mutex_unlock (&conf->lock);
  78         return (-1);
  79     }
  80
  81     conf->connected = true;
  82
  83     c_release (LOG_INFO,
  84         &conf->complaint_cantpublish,
  85         "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
  86         conf->host, conf->port);
  87
  88     pthread_mutex_unlock (&conf->lock);
  89
  90     return (0);
  91 } /* mqtt_reconnect_broker */
  92
  93 static int mqtt_publish_message (struct mqtt_client_conf *conf, char *topic,
  94     char *payload, size_t payload_len)
  95 {
  96     char errbuf[1024];
  97     int status;
  98
  99     status = mosquitto_publish(conf->mosq,
 100         /* message id */ NULL,
 101         topic,
 102         (int) payload_len,
 103         payload,
 104         /* qos */ 0,
 105         /* retain */ false);
 106
 107     if (status != MOSQ_ERR_SUCCESS)
 108     {
 109         c_complain (LOG_ERR,
 110             &conf->complaint_cantpublish,
 111             "plugin mqtt: mosquitto_publish failed: %s",
 112             status == MOSQ_ERR_ERRNO ?
 113             sstrerror(errno, errbuf, sizeof (errbuf)) :
 114                 mosquitto_strerror(status));
 115         /*
 116         Mark our connection "down" regardless of the error as a safety measure;
 117         we will try to reconnect the next time we have to publish a message
 118         */
 119         conf->connected = false;
 120
 121         return (-1);
 122     }
 123
 124     return (0);
 125 } /* mqtt_publish_message */
 126
 127 static int mqtt_format_metric_value (char *buf, size_t buf_len,
 128     const data_set_t *data_set, const value_list_t *vl, int ds_num)
 129 {
 130     gauge_t *rates = NULL;
 131     gauge_t *value = NULL;
 132     size_t metric_value_len;
 133     int status = 0;
 134
 135     memset (buf, 0, buf_len);
 136
 137     if (data_set->ds[ds_num].type == DS_TYPE_GAUGE)
 138         value = &vl->values[ds_num].gauge;
 139     else {
 140         rates = uc_get_rate (data_set, vl);
 141         value = &rates[ds_num];
 142     }
 143
 144     metric_value_len = ssnprintf (buf, buf_len, "%f", *value);
 145
 146     if (metric_value_len >= buf_len)
 147         return (-ENOMEM);
 148
 149     if (rates)
 150         sfree (rates);
 151
 152     return (status);
 153 } /* mqtt_format_metric_value */
 154
 155 static int mqtt_format_message_topic (char *buf, size_t buf_len,
 156     char const *prefix, const value_list_t *vl, const char *ds_name)
 157 {
 158     size_t topic_buf_len;
 159
 160     memset (buf, 0, buf_len);
 161
 162     /*
 163         MQTT message topic format:
 164         [<prefix>/]<hostname>/<plugin>/<plugin instance>/<type>/<type instance>/<ds>/
 165     */
 166     topic_buf_len = (size_t) ssnprintf (buf, buf_len,
 167         "%s/%s/%s/%s/%s/%s/%s",
 168         prefix,
 169         vl->host,
 170         vl->plugin,
 171         vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "(null)",
 172         vl->type,
 173         vl->type_instance[0] != '\0' ? vl->type_instance : "(null)",
 174         ds_name);
 175
 176     if (topic_buf_len >= buf_len)
 177     {
 178         ERROR ("mqtt_format_message_topic: topic buffer too small: "
 179                 "Need %zu bytes.", topic_buf_len + 1);
 180         return (-ENOMEM);
 181     }
 182
 183     return (0);
 184 } /* mqtt_format_message_topic */
 185
 186 static int mqtt_format_payload (char *buf, size_t buf_len,
 187     const data_set_t *data_set, const value_list_t *vl, int ds_num)
 188 {
 189     char metric_path[10 * DATA_MAX_NAME_LEN];
 190     char metric_value[512];
 191     size_t payload_buf_len;
 192     int status = 0;
 193
 194     memset (buf, 0, buf_len);
 195
 196     ssnprintf (metric_path, sizeof (metric_path),
 197         "%s.%s%s%s.%s%s%s%s%s",
 198         vl->host,
 199         vl->plugin,
 200         vl->plugin_instance[0] != '\0' ? "." : "",
 201         vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "",
 202         vl->type,
 203         vl->type_instance[0] != '\0' ? "." : "",
 204         vl->type_instance[0] != '\0' ? vl->type_instance : "",
 205         strcmp(data_set->ds[ds_num].name, "value") != 0 ? "." : "",
 206         strcmp(data_set->ds[ds_num].name, "value") != 0 ?
 207             data_set->ds[ds_num].name : "");
 208
 209     status = mqtt_format_metric_value (metric_value,
 210         sizeof (metric_value),
 211         data_set,
 212         vl,
 213         ds_num);
 214
 215     if (status != 0)
 216     {
 217         ERROR ("mqtt_format_payload: error with mqtt_format_metric_value");
 218         return (status);
 219     }
 220
 221     payload_buf_len = (size_t) ssnprintf (buf, buf_len,
 222         "%s %s %u",
 223         metric_path,
 224         metric_value,
 225         (unsigned int) CDTIME_T_TO_TIME_T (vl->time));
 226
 227     if (payload_buf_len >= buf_len)
 228     {
 229         ERROR ("mqtt_format_payload: payload buffer too small: "
 230                 "Need %zu bytes.", payload_buf_len + 1);
 231         return (-ENOMEM);
 232     }
 233
 234     return (status);
 235 } /* mqtt_format_payload */
 236
 237 static int mqtt_write (const data_set_t *data_set, const value_list_t *vl,
 238     user_data_t *user_data)
 239 {
 240     struct mqtt_client_conf *conf;
 241     char msg_topic[MQTT_MAX_TOPIC_SIZE];
 242     char msg_payload[MQTT_MAX_MESSAGE_SIZE];
 243     int status = 0;
 244     int i;
 245
 246     if (user_data == NULL)
 247         return (EINVAL);
 248
 249     conf = user_data->data;
 250
 251     if (!conf->connected)
 252     {
 253         status = mqtt_reconnect_broker (conf);
 254
 255         if (status != 0) {
 256             ERROR ("plugin mqtt: unable to reconnect to broker");
 257             return (status);
 258         }
 259     }
 260
 261     for (i = 0; i < data_set->ds_num; i++)
 262     {
 263         status = mqtt_format_message_topic (msg_topic, sizeof (msg_topic),
 264             conf->topic_prefix, vl, data_set->ds[i].name);
 265         if (status != 0)
 266         {
 267             ERROR ("plugin mqtt: error with mqtt_format_message_topic");
 268             return (status);
 269         }
 270
 271         status = mqtt_format_payload (msg_payload,
 272             sizeof (msg_payload),
 273             data_set,
 274             vl,
 275             i);
 276
 277         if (status != 0)
 278         {
 279             ERROR ("mqtt_write: error with mqtt_format_payload");
 280             return (status);
 281         }
 282
 283         status = mqtt_publish_message (conf,
 284             msg_topic,
 285             msg_payload,
 286             sizeof (msg_payload));
 287         if (status != 0)
 288         {
 289             ERROR ("plugin mqtt: unable to publish message");
 290             return (status);
 291         }
 292
 293         DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_write[%02X]\x1B[0m "
 294             "published message: topic=%s payload=%s",
 295             (unsigned)pthread_self(),
 296             msg_topic,
 297             msg_payload);
 298     }
 299
 300     return (status);
 301 } /* mqtt_write */
 302
 303 static int mqtt_config (oconfig_item_t *ci)
 304 {
 305     struct mqtt_client_conf *conf;
 306     user_data_t user_data;
 307     char errbuf[1024];
 308     int status;
 309
 310     DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_config[%02X]\x1B[0m ",
 311         (unsigned)pthread_self());
 312
 313     conf = malloc (sizeof (*conf));
 314     if (conf == NULL)
 315     {
 316         ERROR ("write_mqtt plugin: malloc failed.");
 317         return (-1);
 318     }
 319
 320     memset (conf, 0, sizeof (*conf));
 321
 322     conf->connected = false;
 323     conf->host = MQTT_DEFAULT_HOST;
 324     conf->port = MQTT_DEFAULT_PORT;
 325     conf->client_id = MQTT_DEFAULT_CLIENT_ID;
 326     conf->topic_prefix = MQTT_DEFAULT_TOPIC_PREFIX;
 327     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 328
 329     memset (&user_data, 0, sizeof (user_data));
 330     user_data.data = conf;
 331
 332     if ((conf->mosq = mosquitto_new (conf->client_id, true, NULL)) == NULL) {
 333         ERROR ("mqtt_config: mosquitto_new failed");
 334         return (-1);
 335     }
 336
 337     status = mosquitto_connect (conf->mosq, conf->host, conf->port, 10);
 338
 339     if (status != MOSQ_ERR_SUCCESS) {
 340         ERROR ("mqtt_config: mosquitto_connect failed: %s",
 341             (status == MOSQ_ERR_ERRNO ?
 342                 sstrerror(errno, errbuf, sizeof (errbuf)) :
 343                 mosquitto_strerror (status)));
 344         return (-1);
 345     }
 346
 347     DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"",
 348         conf->host, conf->port);
 349
 350     conf->connected = true;
 351
 352     plugin_register_write ("mqtt", mqtt_write, &user_data);
 353
 354     return (0);
 355 } /* mqtt_config */
 356
 357 static int mqtt_init (void)
 358 {
 359     mosquitto_lib_init();
 360
 361     return (0);
 362 } /* mqtt_init */
 363
 364 void module_register (void)
 365 {
 366     plugin_register_complex_config ("mqtt", mqtt_config);
 367     plugin_register_init ("mqtt", mqtt_init);
 368 } /* void module_register */
 369
 370 /* vim: set sw=4 sts=4 et fdm=marker : */