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_cache.h"
37 #include "utils_complain.h"
41 #include <mosquitto.h>
43 #define MQTT_MAX_TOPIC_SIZE 1024
44 #define MQTT_MAX_MESSAGE_SIZE MQTT_MAX_TOPIC_SIZE + 1024
45 #define MQTT_DEFAULT_HOST "localhost"
46 #define MQTT_DEFAULT_PORT 1883
47 #define MQTT_DEFAULT_TOPIC_PREFIX "collectd"
48 #define MQTT_DEFAULT_TOPIC "collectd/#"
49 #ifndef MQTT_KEEPALIVE
50 # define MQTT_KEEPALIVE 60
52 #ifndef SSL_VERIFY_PEER
53 # define SSL_VERIFY_PEER 1
60 struct mqtt_client_conf
65 struct mosquitto *mosq;
74 char *cacertificatefile;
75 char *certificatefile;
76 char *certificatekeyfile;
91 c_complain_t complaint_cantpublish;
94 typedef struct mqtt_client_conf mqtt_client_conf_t;
96 static mqtt_client_conf_t **subscribers = NULL;
97 static size_t subscribers_num = 0;
102 #if LIBMOSQUITTO_MAJOR == 0
103 static char const *mosquitto_strerror (int code)
107 case MOSQ_ERR_SUCCESS: return "MOSQ_ERR_SUCCESS";
108 case MOSQ_ERR_NOMEM: return "MOSQ_ERR_NOMEM";
109 case MOSQ_ERR_PROTOCOL: return "MOSQ_ERR_PROTOCOL";
110 case MOSQ_ERR_INVAL: return "MOSQ_ERR_INVAL";
111 case MOSQ_ERR_NO_CONN: return "MOSQ_ERR_NO_CONN";
112 case MOSQ_ERR_CONN_REFUSED: return "MOSQ_ERR_CONN_REFUSED";
113 case MOSQ_ERR_NOT_FOUND: return "MOSQ_ERR_NOT_FOUND";
114 case MOSQ_ERR_CONN_LOST: return "MOSQ_ERR_CONN_LOST";
115 case MOSQ_ERR_SSL: return "MOSQ_ERR_SSL";
116 case MOSQ_ERR_PAYLOAD_SIZE: return "MOSQ_ERR_PAYLOAD_SIZE";
117 case MOSQ_ERR_NOT_SUPPORTED: return "MOSQ_ERR_NOT_SUPPORTED";
118 case MOSQ_ERR_AUTH: return "MOSQ_ERR_AUTH";
119 case MOSQ_ERR_ACL_DENIED: return "MOSQ_ERR_ACL_DENIED";
120 case MOSQ_ERR_UNKNOWN: return "MOSQ_ERR_UNKNOWN";
121 case MOSQ_ERR_ERRNO: return "MOSQ_ERR_ERRNO";
124 return "UNKNOWN ERROR CODE";
127 /* provided by libmosquitto */
130 static void mqtt_free (mqtt_client_conf_t *conf)
136 (void) mosquitto_disconnect (conf->mosq);
138 (void) mosquitto_destroy (conf->mosq);
141 sfree (conf->username);
142 sfree (conf->password);
143 sfree (conf->client_id);
144 sfree (conf->topic_prefix);
148 static char *strip_prefix (char *topic)
154 for (i = 0; topic[i] != 0; i++)
163 char *tmp = strchr (topic, '/');
173 static void on_message (
174 #if LIBMOSQUITTO_MAJOR == 0
176 __attribute__((unused)) struct mosquitto *m,
178 __attribute__((unused)) void *arg,
179 const struct mosquitto_message *msg)
181 value_list_t vl = VALUE_LIST_INIT;
182 data_set_t const *ds;
188 if ((msg->payloadlen <= 0)
189 || (((uint8_t *) msg->payload)[msg->payloadlen - 1] != 0))
192 topic = strdup (msg->topic);
193 name = strip_prefix (topic);
195 status = parse_identifier_vl (name, &vl);
198 ERROR ("mqtt plugin: Unable to parse topic \"%s\".", topic);
204 ds = plugin_get_ds (vl.type);
207 ERROR ("mqtt plugin: Unknown type: \"%s\".", vl.type);
211 vl.values = calloc (ds->ds_num, sizeof (*vl.values));
212 if (vl.values == NULL)
214 ERROR ("mqtt plugin: calloc failed.");
217 vl.values_len = ds->ds_num;
219 payload = strdup ((void *) msg->payload);
220 DEBUG ("mqtt plugin: payload = \"%s\"", payload);
221 status = parse_values (payload, &vl, ds);
224 ERROR ("mqtt plugin: Unable to parse payload \"%s\".", payload);
231 plugin_dispatch_values (&vl);
233 } /* void on_message */
235 /* must hold conf->lock when calling. */
236 static int mqtt_reconnect (mqtt_client_conf_t *conf)
243 status = mosquitto_reconnect (conf->mosq);
244 if (status != MOSQ_ERR_SUCCESS)
247 ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
248 (status == MOSQ_ERR_ERRNO)
249 ? sstrerror(errno, errbuf, sizeof (errbuf))
250 : mosquitto_strerror (status));
257 &conf->complaint_cantpublish,
258 "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
259 conf->host, conf->port);
262 } /* mqtt_reconnect */
264 /* must hold conf->lock when calling. */
265 static int mqtt_connect (mqtt_client_conf_t *conf)
267 char const *client_id;
270 if (conf->mosq != NULL)
271 return mqtt_reconnect (conf);
274 client_id = conf->client_id;
276 client_id = hostname_g;
278 #if LIBMOSQUITTO_MAJOR == 0
279 conf->mosq = mosquitto_new (client_id, /* user data = */ conf);
281 conf->mosq = mosquitto_new (client_id, conf->clean_session, /* user data = */ conf);
283 if (conf->mosq == NULL)
285 ERROR ("mqtt plugin: mosquitto_new failed");
289 #if LIBMOSQUITTO_MAJOR != 0
290 if (conf->cacertificatefile) {
291 status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL,
292 conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL);
293 if (status != MOSQ_ERR_SUCCESS) {
294 ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status));
295 mosquitto_destroy (conf->mosq);
300 status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite);
301 if (status != MOSQ_ERR_SUCCESS) {
302 ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status));
303 mosquitto_destroy (conf->mosq);
308 status = mosquitto_tls_insecure_set(conf->mosq, false);
309 if (status != MOSQ_ERR_SUCCESS) {
310 ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status));
311 mosquitto_destroy (conf->mosq);
318 if (conf->username && conf->password)
320 status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password);
321 if (status != MOSQ_ERR_SUCCESS)
324 ERROR ("mqtt plugin: mosquitto_username_pw_set failed: %s",
325 (status == MOSQ_ERR_ERRNO)
326 ? sstrerror (errno, errbuf, sizeof (errbuf))
327 : mosquitto_strerror (status));
329 mosquitto_destroy (conf->mosq);
335 #if LIBMOSQUITTO_MAJOR == 0
336 status = mosquitto_connect (conf->mosq, conf->host, conf->port,
337 /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session);
339 status = mosquitto_connect (conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE);
341 if (status != MOSQ_ERR_SUCCESS)
344 ERROR ("mqtt plugin: mosquitto_connect failed: %s",
345 (status == MOSQ_ERR_ERRNO)
346 ? sstrerror (errno, errbuf, sizeof (errbuf))
347 : mosquitto_strerror (status));
349 mosquitto_destroy (conf->mosq);
356 mosquitto_message_callback_set (conf->mosq, on_message);
358 status = mosquitto_subscribe (conf->mosq,
359 /* message_id = */ NULL,
360 conf->topic, conf->qos);
361 if (status != MOSQ_ERR_SUCCESS)
363 ERROR ("mqtt plugin: Subscribing to \"%s\" failed: %s",
364 conf->topic, mosquitto_strerror (status));
366 mosquitto_disconnect (conf->mosq);
367 mosquitto_destroy (conf->mosq);
377 static void *subscribers_thread (void *arg)
379 mqtt_client_conf_t *conf = arg;
386 status = mqtt_connect (conf);
393 /* The documentation says "0" would map to the default (1000ms), but
394 * that does not work on some versions. */
395 #if LIBMOSQUITTO_MAJOR == 0
396 status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */);
398 status = mosquitto_loop (conf->mosq,
399 /* timeout[ms] = */ 1000,
400 /* max_packets = */ 100);
402 if (status == MOSQ_ERR_CONN_LOST)
407 else if (status != MOSQ_ERR_SUCCESS)
409 ERROR ("mqtt plugin: mosquitto_loop failed: %s",
410 mosquitto_strerror (status));
411 mosquitto_destroy (conf->mosq);
417 DEBUG ("mqtt plugin: mosquitto_loop succeeded.");
418 } /* while (conf->loop) */
421 } /* void *subscribers_thread */
423 static int publish (mqtt_client_conf_t *conf, char const *topic,
424 void const *payload, size_t payload_len)
428 pthread_mutex_lock (&conf->lock);
430 status = mqtt_connect (conf);
432 pthread_mutex_unlock (&conf->lock);
433 ERROR ("mqtt plugin: unable to reconnect to broker");
437 status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic,
438 #if LIBMOSQUITTO_MAJOR == 0
439 (uint32_t) payload_len, payload,
441 (int) payload_len, payload,
443 conf->qos, conf->retain);
444 if (status != MOSQ_ERR_SUCCESS)
448 &conf->complaint_cantpublish,
449 "mqtt plugin: mosquitto_publish failed: %s",
450 (status == MOSQ_ERR_ERRNO)
451 ? sstrerror(errno, errbuf, sizeof (errbuf))
452 : mosquitto_strerror(status));
453 /* Mark our connection "down" regardless of the error as a safety
454 * measure; we will try to reconnect the next time we have to publish a
458 pthread_mutex_unlock (&conf->lock);
462 pthread_mutex_unlock (&conf->lock);
466 static int format_topic (char *buf, size_t buf_len,
467 data_set_t const *ds, value_list_t const *vl,
468 mqtt_client_conf_t *conf)
470 char name[MQTT_MAX_TOPIC_SIZE];
473 if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
474 return (FORMAT_VL (buf, buf_len, vl));
476 status = FORMAT_VL (name, sizeof (name), vl);
480 status = ssnprintf (buf, buf_len, "%s/%s", conf->topic_prefix, name);
481 if ((status < 0) || (((size_t) status) >= buf_len))
485 } /* int format_topic */
487 static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
488 user_data_t *user_data)
490 mqtt_client_conf_t *conf;
491 char topic[MQTT_MAX_TOPIC_SIZE];
492 char payload[MQTT_MAX_MESSAGE_SIZE];
495 if ((user_data == NULL) || (user_data->data == NULL))
497 conf = user_data->data;
499 status = format_topic (topic, sizeof (topic), ds, vl, conf);
502 ERROR ("mqtt plugin: format_topic failed with status %d.", status);
506 status = format_values (payload, sizeof (payload),
507 ds, vl, conf->store_rates);
510 ERROR ("mqtt plugin: format_values failed with status %d.", status);
514 status = publish (conf, topic, payload, strlen (payload) + 1);
517 ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
528 * ClientId "collectd"
535 * CACert "ca.pem" Enables TLS if set
536 * CertificateFile "client-cert.pem" optional
537 * CertificateKeyFile "client-key.pem" optional
538 * TLSProtocol "tlsv1.2" optional
541 static int mqtt_config_publisher (oconfig_item_t *ci)
543 mqtt_client_conf_t *conf;
545 user_data_t user_data;
549 conf = calloc (1, sizeof (*conf));
552 ERROR ("mqtt plugin: malloc failed.");
558 status = cf_util_get_string (ci, &conf->name);
565 conf->host = strdup (MQTT_DEFAULT_HOST);
566 conf->port = MQTT_DEFAULT_PORT;
567 conf->client_id = NULL;
569 conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
570 conf->store_rates = 1;
572 status = pthread_mutex_init (&conf->lock, NULL);
579 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
581 for (i = 0; i < ci->children_num; i++)
583 oconfig_item_t *child = ci->children + i;
584 if (strcasecmp ("Host", child->key) == 0)
585 cf_util_get_string (child, &conf->host);
586 else if (strcasecmp ("Port", child->key) == 0)
588 int tmp = cf_util_get_port_number (child);
590 ERROR ("mqtt plugin: Invalid port number.");
594 else if (strcasecmp ("ClientId", child->key) == 0)
595 cf_util_get_string (child, &conf->client_id);
596 else if (strcasecmp ("User", child->key) == 0)
597 cf_util_get_string (child, &conf->username);
598 else if (strcasecmp ("Password", child->key) == 0)
599 cf_util_get_string (child, &conf->password);
600 else if (strcasecmp ("QoS", child->key) == 0)
603 status = cf_util_get_int (child, &tmp);
604 if ((status != 0) || (tmp < 0) || (tmp > 2))
605 ERROR ("mqtt plugin: Not a valid QoS setting.");
609 else if (strcasecmp ("Prefix", child->key) == 0)
610 cf_util_get_string (child, &conf->topic_prefix);
611 else if (strcasecmp ("StoreRates", child->key) == 0)
612 cf_util_get_boolean (child, &conf->store_rates);
613 else if (strcasecmp ("Retain", child->key) == 0)
614 cf_util_get_boolean (child, &conf->retain);
615 else if (strcasecmp ("CACert", child->key) == 0)
616 cf_util_get_string (child, &conf->cacertificatefile);
617 else if (strcasecmp ("CertificateFile", child->key) == 0)
618 cf_util_get_string (child, &conf->certificatefile);
619 else if (strcasecmp ("CertificateKeyFile", child->key) == 0)
620 cf_util_get_string (child, &conf->certificatekeyfile);
621 else if (strcasecmp ("TLSProtocol", child->key) == 0)
622 cf_util_get_string (child, &conf->tlsprotocol);
623 else if (strcasecmp ("CipherSuite", child->key) == 0)
624 cf_util_get_string (child, &conf->ciphersuite);
626 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
629 ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name);
630 memset (&user_data, 0, sizeof (user_data));
631 user_data.data = conf;
633 plugin_register_write (cb_name, mqtt_write, &user_data);
635 } /* mqtt_config_publisher */
641 * ClientId "collectd"
647 static int mqtt_config_subscriber (oconfig_item_t *ci)
649 mqtt_client_conf_t **tmp;
650 mqtt_client_conf_t *conf;
654 conf = calloc (1, sizeof (*conf));
657 ERROR ("mqtt plugin: malloc failed.");
663 status = cf_util_get_string (ci, &conf->name);
670 conf->host = strdup (MQTT_DEFAULT_HOST);
671 conf->port = MQTT_DEFAULT_PORT;
672 conf->client_id = NULL;
674 conf->topic = strdup (MQTT_DEFAULT_TOPIC);
675 conf->clean_session = 1;
677 status = pthread_mutex_init (&conf->lock, NULL);
684 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
686 for (i = 0; i < ci->children_num; i++)
688 oconfig_item_t *child = ci->children + i;
689 if (strcasecmp ("Host", child->key) == 0)
690 cf_util_get_string (child, &conf->host);
691 else if (strcasecmp ("Port", child->key) == 0)
693 int tmp = cf_util_get_port_number (child);
695 ERROR ("mqtt plugin: Invalid port number.");
699 else if (strcasecmp ("ClientId", child->key) == 0)
700 cf_util_get_string (child, &conf->client_id);
701 else if (strcasecmp ("User", child->key) == 0)
702 cf_util_get_string (child, &conf->username);
703 else if (strcasecmp ("Password", child->key) == 0)
704 cf_util_get_string (child, &conf->password);
705 else if (strcasecmp ("QoS", child->key) == 0)
708 status = cf_util_get_int (child, &tmp);
709 if ((status != 0) || (tmp < 0) || (tmp > 2))
710 ERROR ("mqtt plugin: Not a valid QoS setting.");
714 else if (strcasecmp ("Topic", child->key) == 0)
715 cf_util_get_string (child, &conf->topic);
716 else if (strcasecmp ("CleanSession", child->key) == 0)
717 cf_util_get_boolean (child, &conf->clean_session);
719 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
722 tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num);
725 ERROR ("mqtt plugin: realloc failed.");
730 subscribers[subscribers_num] = conf;
734 } /* mqtt_config_subscriber */
746 static int mqtt_config (oconfig_item_t *ci)
750 for (i = 0; i < ci->children_num; i++)
752 oconfig_item_t *child = ci->children + i;
754 if (strcasecmp ("Publish", child->key) == 0)
755 mqtt_config_publisher (child);
756 else if (strcasecmp ("Subscribe", child->key) == 0)
757 mqtt_config_subscriber (child);
759 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
763 } /* int mqtt_config */
765 static int mqtt_init (void)
769 mosquitto_lib_init ();
771 for (i = 0; i < subscribers_num; i++)
775 if (subscribers[i]->loop)
778 status = plugin_thread_create (&subscribers[i]->thread,
780 /* func = */ subscribers_thread,
781 /* args = */ subscribers[i]);
785 ERROR ("mqtt plugin: pthread_create failed: %s",
786 sstrerror (errno, errbuf, sizeof (errbuf)));
794 void module_register (void)
796 plugin_register_complex_config ("mqtt", mqtt_config);
797 plugin_register_init ("mqtt", mqtt_init);
798 } /* void module_register */
800 /* vim: set sw=4 sts=4 et fdm=marker : */