2 * collectd - src/mqtt.c
3 * Copyright (C) 2014 Marc Falzon
4 * Copyright (C) 2014,2015 Florian octo Forster
6 * Permission is hereby granted, free of charge, to any person obtaining a
7 * copy of this software and associated documentation files (the "Software"),
8 * to deal in the Software without restriction, including without limitation
9 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
10 * and/or sell copies of the Software, and to permit persons to whom the
11 * Software is furnished to do so, subject to the following conditions:
13 * The above copyright notice and this permission notice shall be included in
14 * all copies or substantial portions of the Software.
16 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
22 * DEALINGS IN THE SOFTWARE.
25 * Marc Falzon <marc at baha dot mu>
26 * Florian octo Forster <octo at collectd.org>
27 * Jan-Piet Mens <jpmens at gmail.com>
30 // Reference: http://mosquitto.org/api/files/mosquitto-h.html
36 #include "utils_complain.h"
38 #include <mosquitto.h>
40 #define MQTT_MAX_TOPIC_SIZE 1024
41 #define MQTT_MAX_MESSAGE_SIZE MQTT_MAX_TOPIC_SIZE + 1024
42 #define MQTT_DEFAULT_HOST "localhost"
43 #define MQTT_DEFAULT_PORT 1883
44 #define MQTT_DEFAULT_TOPIC_PREFIX "collectd"
45 #define MQTT_DEFAULT_TOPIC "collectd/#"
46 #ifndef MQTT_KEEPALIVE
47 # define MQTT_KEEPALIVE 60
49 #ifndef SSL_VERIFY_PEER
50 # define SSL_VERIFY_PEER 1
57 struct mqtt_client_conf
62 struct mosquitto *mosq;
71 char *cacertificatefile;
72 char *certificatefile;
73 char *certificatekeyfile;
88 c_complain_t complaint_cantpublish;
91 typedef struct mqtt_client_conf mqtt_client_conf_t;
93 static mqtt_client_conf_t **subscribers = NULL;
94 static size_t subscribers_num = 0;
99 #if LIBMOSQUITTO_MAJOR == 0
100 static char const *mosquitto_strerror (int code)
104 case MOSQ_ERR_SUCCESS: return "MOSQ_ERR_SUCCESS";
105 case MOSQ_ERR_NOMEM: return "MOSQ_ERR_NOMEM";
106 case MOSQ_ERR_PROTOCOL: return "MOSQ_ERR_PROTOCOL";
107 case MOSQ_ERR_INVAL: return "MOSQ_ERR_INVAL";
108 case MOSQ_ERR_NO_CONN: return "MOSQ_ERR_NO_CONN";
109 case MOSQ_ERR_CONN_REFUSED: return "MOSQ_ERR_CONN_REFUSED";
110 case MOSQ_ERR_NOT_FOUND: return "MOSQ_ERR_NOT_FOUND";
111 case MOSQ_ERR_CONN_LOST: return "MOSQ_ERR_CONN_LOST";
112 case MOSQ_ERR_SSL: return "MOSQ_ERR_SSL";
113 case MOSQ_ERR_PAYLOAD_SIZE: return "MOSQ_ERR_PAYLOAD_SIZE";
114 case MOSQ_ERR_NOT_SUPPORTED: return "MOSQ_ERR_NOT_SUPPORTED";
115 case MOSQ_ERR_AUTH: return "MOSQ_ERR_AUTH";
116 case MOSQ_ERR_ACL_DENIED: return "MOSQ_ERR_ACL_DENIED";
117 case MOSQ_ERR_UNKNOWN: return "MOSQ_ERR_UNKNOWN";
118 case MOSQ_ERR_ERRNO: return "MOSQ_ERR_ERRNO";
121 return "UNKNOWN ERROR CODE";
124 /* provided by libmosquitto */
127 static void mqtt_free (mqtt_client_conf_t *conf)
133 (void) mosquitto_disconnect (conf->mosq);
135 (void) mosquitto_destroy (conf->mosq);
138 sfree (conf->username);
139 sfree (conf->password);
140 sfree (conf->client_id);
141 sfree (conf->topic_prefix);
145 static char *strip_prefix (char *topic)
151 for (i = 0; topic[i] != 0; i++)
160 char *tmp = strchr (topic, '/');
170 static void on_message (
171 #if LIBMOSQUITTO_MAJOR == 0
173 __attribute__((unused)) struct mosquitto *m,
175 __attribute__((unused)) void *arg,
176 const struct mosquitto_message *msg)
178 value_list_t vl = VALUE_LIST_INIT;
179 data_set_t const *ds;
185 if (msg->payloadlen <= 0) {
186 DEBUG ("mqtt plugin: message has empty payload");
190 topic = strdup (msg->topic);
191 name = strip_prefix (topic);
193 status = parse_identifier_vl (name, &vl);
196 ERROR ("mqtt plugin: Unable to parse topic \"%s\".", topic);
202 ds = plugin_get_ds (vl.type);
205 ERROR ("mqtt plugin: Unknown type: \"%s\".", vl.type);
209 vl.values = calloc (ds->ds_num, sizeof (*vl.values));
210 if (vl.values == NULL)
212 ERROR ("mqtt plugin: calloc failed.");
215 vl.values_len = ds->ds_num;
217 payload = malloc (msg->payloadlen+1);
220 ERROR ("mqtt plugin: malloc for payload buffer failed.");
224 memmove (payload, msg->payload, msg->payloadlen);
225 payload[msg->payloadlen] = 0;
227 DEBUG ("mqtt plugin: payload = \"%s\"", payload);
228 status = parse_values (payload, &vl, ds);
231 ERROR ("mqtt plugin: Unable to parse payload \"%s\".", payload);
238 plugin_dispatch_values (&vl);
240 } /* void on_message */
242 /* must hold conf->lock when calling. */
243 static int mqtt_reconnect (mqtt_client_conf_t *conf)
250 status = mosquitto_reconnect (conf->mosq);
251 if (status != MOSQ_ERR_SUCCESS)
254 ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
255 (status == MOSQ_ERR_ERRNO)
256 ? sstrerror(errno, errbuf, sizeof (errbuf))
257 : mosquitto_strerror (status));
264 &conf->complaint_cantpublish,
265 "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
266 conf->host, conf->port);
269 } /* mqtt_reconnect */
271 /* must hold conf->lock when calling. */
272 static int mqtt_connect (mqtt_client_conf_t *conf)
274 char const *client_id;
277 if (conf->mosq != NULL)
278 return mqtt_reconnect (conf);
281 client_id = conf->client_id;
283 client_id = hostname_g;
285 #if LIBMOSQUITTO_MAJOR == 0
286 conf->mosq = mosquitto_new (client_id, /* user data = */ conf);
288 conf->mosq = mosquitto_new (client_id, conf->clean_session, /* user data = */ conf);
290 if (conf->mosq == NULL)
292 ERROR ("mqtt plugin: mosquitto_new failed");
296 #if LIBMOSQUITTO_MAJOR != 0
297 if (conf->cacertificatefile) {
298 status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL,
299 conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL);
300 if (status != MOSQ_ERR_SUCCESS) {
301 ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status));
302 mosquitto_destroy (conf->mosq);
307 status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite);
308 if (status != MOSQ_ERR_SUCCESS) {
309 ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status));
310 mosquitto_destroy (conf->mosq);
315 status = mosquitto_tls_insecure_set(conf->mosq, false);
316 if (status != MOSQ_ERR_SUCCESS) {
317 ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status));
318 mosquitto_destroy (conf->mosq);
325 if (conf->username && conf->password)
327 status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password);
328 if (status != MOSQ_ERR_SUCCESS)
331 ERROR ("mqtt plugin: mosquitto_username_pw_set failed: %s",
332 (status == MOSQ_ERR_ERRNO)
333 ? sstrerror (errno, errbuf, sizeof (errbuf))
334 : mosquitto_strerror (status));
336 mosquitto_destroy (conf->mosq);
342 #if LIBMOSQUITTO_MAJOR == 0
343 status = mosquitto_connect (conf->mosq, conf->host, conf->port,
344 /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session);
346 status = mosquitto_connect (conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE);
348 if (status != MOSQ_ERR_SUCCESS)
351 ERROR ("mqtt plugin: mosquitto_connect failed: %s",
352 (status == MOSQ_ERR_ERRNO)
353 ? sstrerror (errno, errbuf, sizeof (errbuf))
354 : mosquitto_strerror (status));
356 mosquitto_destroy (conf->mosq);
363 mosquitto_message_callback_set (conf->mosq, on_message);
365 status = mosquitto_subscribe (conf->mosq,
366 /* message_id = */ NULL,
367 conf->topic, conf->qos);
368 if (status != MOSQ_ERR_SUCCESS)
370 ERROR ("mqtt plugin: Subscribing to \"%s\" failed: %s",
371 conf->topic, mosquitto_strerror (status));
373 mosquitto_disconnect (conf->mosq);
374 mosquitto_destroy (conf->mosq);
384 static void *subscribers_thread (void *arg)
386 mqtt_client_conf_t *conf = arg;
393 status = mqtt_connect (conf);
400 /* The documentation says "0" would map to the default (1000ms), but
401 * that does not work on some versions. */
402 #if LIBMOSQUITTO_MAJOR == 0
403 status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */);
405 status = mosquitto_loop (conf->mosq,
406 /* timeout[ms] = */ 1000,
407 /* max_packets = */ 100);
409 if (status == MOSQ_ERR_CONN_LOST)
414 else if (status != MOSQ_ERR_SUCCESS)
416 ERROR ("mqtt plugin: mosquitto_loop failed: %s",
417 mosquitto_strerror (status));
418 mosquitto_destroy (conf->mosq);
424 DEBUG ("mqtt plugin: mosquitto_loop succeeded.");
425 } /* while (conf->loop) */
428 } /* void *subscribers_thread */
430 static int publish (mqtt_client_conf_t *conf, char const *topic,
431 void const *payload, size_t payload_len)
435 pthread_mutex_lock (&conf->lock);
437 status = mqtt_connect (conf);
439 pthread_mutex_unlock (&conf->lock);
440 ERROR ("mqtt plugin: unable to reconnect to broker");
444 status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic,
445 #if LIBMOSQUITTO_MAJOR == 0
446 (uint32_t) payload_len, payload,
448 (int) payload_len, payload,
450 conf->qos, conf->retain);
451 if (status != MOSQ_ERR_SUCCESS)
455 &conf->complaint_cantpublish,
456 "mqtt plugin: mosquitto_publish failed: %s",
457 (status == MOSQ_ERR_ERRNO)
458 ? sstrerror(errno, errbuf, sizeof (errbuf))
459 : mosquitto_strerror(status));
460 /* Mark our connection "down" regardless of the error as a safety
461 * measure; we will try to reconnect the next time we have to publish a
465 pthread_mutex_unlock (&conf->lock);
469 pthread_mutex_unlock (&conf->lock);
473 static int format_topic (char *buf, size_t buf_len,
474 data_set_t const *ds, value_list_t const *vl,
475 mqtt_client_conf_t *conf)
477 char name[MQTT_MAX_TOPIC_SIZE];
480 if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
481 return (FORMAT_VL (buf, buf_len, vl));
483 status = FORMAT_VL (name, sizeof (name), vl);
487 status = ssnprintf (buf, buf_len, "%s/%s", conf->topic_prefix, name);
488 if ((status < 0) || (((size_t) status) >= buf_len))
492 } /* int format_topic */
494 static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
495 user_data_t *user_data)
497 mqtt_client_conf_t *conf;
498 char topic[MQTT_MAX_TOPIC_SIZE];
499 char payload[MQTT_MAX_MESSAGE_SIZE];
502 if ((user_data == NULL) || (user_data->data == NULL))
504 conf = user_data->data;
506 status = format_topic (topic, sizeof (topic), ds, vl, conf);
509 ERROR ("mqtt plugin: format_topic failed with status %d.", status);
513 status = format_values (payload, sizeof (payload),
514 ds, vl, conf->store_rates);
517 ERROR ("mqtt plugin: format_values failed with status %d.", status);
521 status = publish (conf, topic, payload, strlen (payload) + 1);
524 ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
535 * ClientId "collectd"
542 * CACert "ca.pem" Enables TLS if set
543 * CertificateFile "client-cert.pem" optional
544 * CertificateKeyFile "client-key.pem" optional
545 * TLSProtocol "tlsv1.2" optional
548 static int mqtt_config_publisher (oconfig_item_t *ci)
550 mqtt_client_conf_t *conf;
552 user_data_t user_data;
556 conf = calloc (1, sizeof (*conf));
559 ERROR ("mqtt plugin: calloc failed.");
565 status = cf_util_get_string (ci, &conf->name);
572 conf->host = strdup (MQTT_DEFAULT_HOST);
573 conf->port = MQTT_DEFAULT_PORT;
574 conf->client_id = NULL;
576 conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
577 conf->store_rates = 1;
579 status = pthread_mutex_init (&conf->lock, NULL);
586 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
588 for (i = 0; i < ci->children_num; i++)
590 oconfig_item_t *child = ci->children + i;
591 if (strcasecmp ("Host", child->key) == 0)
592 cf_util_get_string (child, &conf->host);
593 else if (strcasecmp ("Port", child->key) == 0)
595 int tmp = cf_util_get_port_number (child);
597 ERROR ("mqtt plugin: Invalid port number.");
601 else if (strcasecmp ("ClientId", child->key) == 0)
602 cf_util_get_string (child, &conf->client_id);
603 else if (strcasecmp ("User", child->key) == 0)
604 cf_util_get_string (child, &conf->username);
605 else if (strcasecmp ("Password", child->key) == 0)
606 cf_util_get_string (child, &conf->password);
607 else if (strcasecmp ("QoS", child->key) == 0)
610 status = cf_util_get_int (child, &tmp);
611 if ((status != 0) || (tmp < 0) || (tmp > 2))
612 ERROR ("mqtt plugin: Not a valid QoS setting.");
616 else if (strcasecmp ("Prefix", child->key) == 0)
617 cf_util_get_string (child, &conf->topic_prefix);
618 else if (strcasecmp ("StoreRates", child->key) == 0)
619 cf_util_get_boolean (child, &conf->store_rates);
620 else if (strcasecmp ("Retain", child->key) == 0)
621 cf_util_get_boolean (child, &conf->retain);
622 else if (strcasecmp ("CACert", child->key) == 0)
623 cf_util_get_string (child, &conf->cacertificatefile);
624 else if (strcasecmp ("CertificateFile", child->key) == 0)
625 cf_util_get_string (child, &conf->certificatefile);
626 else if (strcasecmp ("CertificateKeyFile", child->key) == 0)
627 cf_util_get_string (child, &conf->certificatekeyfile);
628 else if (strcasecmp ("TLSProtocol", child->key) == 0)
629 cf_util_get_string (child, &conf->tlsprotocol);
630 else if (strcasecmp ("CipherSuite", child->key) == 0)
631 cf_util_get_string (child, &conf->ciphersuite);
633 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
636 ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name);
637 memset (&user_data, 0, sizeof (user_data));
638 user_data.data = conf;
640 plugin_register_write (cb_name, mqtt_write, &user_data);
642 } /* mqtt_config_publisher */
648 * ClientId "collectd"
654 static int mqtt_config_subscriber (oconfig_item_t *ci)
656 mqtt_client_conf_t **tmp;
657 mqtt_client_conf_t *conf;
661 conf = calloc (1, sizeof (*conf));
664 ERROR ("mqtt plugin: calloc failed.");
670 status = cf_util_get_string (ci, &conf->name);
677 conf->host = strdup (MQTT_DEFAULT_HOST);
678 conf->port = MQTT_DEFAULT_PORT;
679 conf->client_id = NULL;
681 conf->topic = strdup (MQTT_DEFAULT_TOPIC);
682 conf->clean_session = 1;
684 status = pthread_mutex_init (&conf->lock, NULL);
691 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
693 for (i = 0; i < ci->children_num; i++)
695 oconfig_item_t *child = ci->children + i;
696 if (strcasecmp ("Host", child->key) == 0)
697 cf_util_get_string (child, &conf->host);
698 else if (strcasecmp ("Port", child->key) == 0)
700 status = cf_util_get_port_number (child);
702 ERROR ("mqtt plugin: Invalid port number.");
706 else if (strcasecmp ("ClientId", child->key) == 0)
707 cf_util_get_string (child, &conf->client_id);
708 else if (strcasecmp ("User", child->key) == 0)
709 cf_util_get_string (child, &conf->username);
710 else if (strcasecmp ("Password", child->key) == 0)
711 cf_util_get_string (child, &conf->password);
712 else if (strcasecmp ("QoS", child->key) == 0)
715 status = cf_util_get_int (child, &qos);
716 if ((status != 0) || (qos < 0) || (qos > 2))
717 ERROR ("mqtt plugin: Not a valid QoS setting.");
721 else if (strcasecmp ("Topic", child->key) == 0)
722 cf_util_get_string (child, &conf->topic);
723 else if (strcasecmp ("CleanSession", child->key) == 0)
724 cf_util_get_boolean (child, &conf->clean_session);
726 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
729 tmp = realloc (subscribers, sizeof (*subscribers) * (subscribers_num + 1) );
732 ERROR ("mqtt plugin: realloc failed.");
737 subscribers[subscribers_num] = conf;
741 } /* mqtt_config_subscriber */
753 static int mqtt_config (oconfig_item_t *ci)
757 for (i = 0; i < ci->children_num; i++)
759 oconfig_item_t *child = ci->children + i;
761 if (strcasecmp ("Publish", child->key) == 0)
762 mqtt_config_publisher (child);
763 else if (strcasecmp ("Subscribe", child->key) == 0)
764 mqtt_config_subscriber (child);
766 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
770 } /* int mqtt_config */
772 static int mqtt_init (void)
776 mosquitto_lib_init ();
778 for (i = 0; i < subscribers_num; i++)
782 if (subscribers[i]->loop)
785 status = plugin_thread_create (&subscribers[i]->thread,
787 /* func = */ subscribers_thread,
788 /* args = */ subscribers[i]);
792 ERROR ("mqtt plugin: pthread_create failed: %s",
793 sstrerror (errno, errbuf, sizeof (errbuf)));
801 void module_register (void)
803 plugin_register_complex_config ("mqtt", mqtt_config);
804 plugin_register_init ("mqtt", mqtt_init);
805 } /* void module_register */
807 /* vim: set sw=4 sts=4 et fdm=marker : */