X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fmqtt.c;h=8bc412c9497213074a738ca7dfa355209c1872d1;hb=0a95b83f89a553be1ac2beae2003927e74230d23;hp=98b1751db7fafcc9b537aeca7ac2efa95df21465;hpb=40b0765c2180eba24d1bb4740b5a08754052ab89;p=collectd.git diff --git a/src/mqtt.c b/src/mqtt.c index 98b1751d..c08fd29e 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -24,19 +24,18 @@ * Authors: * Marc Falzon * Florian octo Forster + * Jan-Piet Mens **/ // Reference: http://mosquitto.org/api/files/mosquitto-h.html #include "collectd.h" + #include "common.h" #include "plugin.h" -#include "utils_cache.h" #include "utils_complain.h" -#include - #include #define MQTT_MAX_TOPIC_SIZE 1024 @@ -48,6 +47,9 @@ #ifndef MQTT_KEEPALIVE # define MQTT_KEEPALIVE 60 #endif +#ifndef SSL_VERIFY_PEER +# define SSL_VERIFY_PEER 1 +#endif /* @@ -67,6 +69,11 @@ struct mqtt_client_conf char *username; char *password; int qos; + char *cacertificatefile; + char *certificatefile; + char *certificatekeyfile; + char *tlsprotocol; + char *ciphersuite; /* For publishing */ char *topic_prefix; @@ -90,6 +97,7 @@ static size_t subscribers_num = 0; /* * Functions */ +#if LIBMOSQUITTO_MAJOR == 0 static char const *mosquitto_strerror (int code) { switch (code) @@ -113,6 +121,9 @@ static char const *mosquitto_strerror (int code) return "UNKNOWN ERROR CODE"; } +#else +/* provided by libmosquitto */ +#endif static void mqtt_free (mqtt_client_conf_t *conf) { @@ -134,11 +145,9 @@ static void mqtt_free (mqtt_client_conf_t *conf) static char *strip_prefix (char *topic) { - size_t num; - size_t i; + size_t num = 0; - num = 0; - for (i = 0; topic[i] != 0; i++) + for (size_t i = 0; topic[i] != 0; i++) if (topic[i] == '/') num++; @@ -157,7 +166,12 @@ static char *strip_prefix (char *topic) return (topic); } -static void on_message (__attribute__((unused)) void *arg, +static void on_message ( +#if LIBMOSQUITTO_MAJOR == 0 +#else + __attribute__((unused)) struct mosquitto *m, +#endif + __attribute__((unused)) void *arg, const struct mosquitto_message *msg) { value_list_t vl = VALUE_LIST_INIT; @@ -167,8 +181,10 @@ static void on_message (__attribute__((unused)) void *arg, char *payload; int status; - if ((msg->payloadlen <= 0) || (msg->payload[msg->payloadlen - 1] != 0)) + if (msg->payloadlen <= 0) { + DEBUG ("mqtt plugin: message has empty payload"); return; + } topic = strdup (msg->topic); name = strip_prefix (topic); @@ -197,7 +213,16 @@ static void on_message (__attribute__((unused)) void *arg, } vl.values_len = ds->ds_num; - payload = strdup ((void *) msg->payload); + payload = malloc (msg->payloadlen+1); + if (payload == NULL) + { + ERROR ("mqtt plugin: malloc for payload buffer failed."); + sfree (vl.values); + return; + } + memmove (payload, msg->payload, msg->payloadlen); + payload[msg->payloadlen] = 0; + DEBUG ("mqtt plugin: payload = \"%s\"", payload); status = parse_values (payload, &vl, ds); if (status != 0) @@ -256,13 +281,46 @@ static int mqtt_connect (mqtt_client_conf_t *conf) else client_id = hostname_g; +#if LIBMOSQUITTO_MAJOR == 0 conf->mosq = mosquitto_new (client_id, /* user data = */ conf); +#else + conf->mosq = mosquitto_new (client_id, conf->clean_session, /* user data = */ conf); +#endif if (conf->mosq == NULL) { ERROR ("mqtt plugin: mosquitto_new failed"); return (-1); } +#if LIBMOSQUITTO_MAJOR != 0 + if (conf->cacertificatefile) { + status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL, + conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL); + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status)); + mosquitto_destroy (conf->mosq); + conf->mosq = NULL; + return (-1); + } + + status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite); + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status)); + mosquitto_destroy (conf->mosq); + conf->mosq = NULL; + return (-1); + } + + status = mosquitto_tls_insecure_set(conf->mosq, false); + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status)); + mosquitto_destroy (conf->mosq); + conf->mosq = NULL; + return (-1); + } + } +#endif + if (conf->username && conf->password) { status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password); @@ -280,8 +338,12 @@ static int mqtt_connect (mqtt_client_conf_t *conf) } } +#if LIBMOSQUITTO_MAJOR == 0 status = mosquitto_connect (conf->mosq, conf->host, conf->port, /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session); +#else + status = mosquitto_connect (conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE); +#endif if (status != MOSQ_ERR_SUCCESS) { char errbuf[1024]; @@ -299,7 +361,8 @@ static int mqtt_connect (mqtt_client_conf_t *conf) { mosquitto_message_callback_set (conf->mosq, on_message); - status = mosquitto_subscribe (conf->mosq, /* mid = */ NULL, + status = mosquitto_subscribe (conf->mosq, + /* message_id = */ NULL, conf->topic, conf->qos); if (status != MOSQ_ERR_SUCCESS) { @@ -335,7 +398,13 @@ static void *subscribers_thread (void *arg) /* The documentation says "0" would map to the default (1000ms), but * that does not work on some versions. */ +#if LIBMOSQUITTO_MAJOR == 0 status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */); +#else + status = mosquitto_loop (conf->mosq, + /* timeout[ms] = */ 1000, + /* max_packets = */ 100); +#endif if (status == MOSQ_ERR_CONN_LOST) { conf->connected = 0; @@ -371,21 +440,22 @@ static int publish (mqtt_client_conf_t *conf, char const *topic, return (status); } - status = mosquitto_publish(conf->mosq, - /* message id */ NULL, - topic, + status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic, +#if LIBMOSQUITTO_MAJOR == 0 (uint32_t) payload_len, payload, - /* qos */ conf->qos, - /* retain */ conf->retain); +#else + (int) payload_len, payload, +#endif + conf->qos, conf->retain); if (status != MOSQ_ERR_SUCCESS) { char errbuf[1024]; c_complain (LOG_ERR, - &conf->complaint_cantpublish, - "plugin mqtt: mosquitto_publish failed: %s", - status == MOSQ_ERR_ERRNO ? - sstrerror(errno, errbuf, sizeof (errbuf)) : - mosquitto_strerror(status)); + &conf->complaint_cantpublish, + "mqtt plugin: mosquitto_publish failed: %s", + (status == MOSQ_ERR_ERRNO) + ? sstrerror(errno, errbuf, sizeof (errbuf)) + : mosquitto_strerror(status)); /* Mark our connection "down" regardless of the error as a safety * measure; we will try to reconnect the next time we have to publish a * message */ @@ -468,19 +538,22 @@ static int mqtt_write (const data_set_t *ds, const value_list_t *vl, * StoreRates true * Retain false * QoS 0 + * CACert "ca.pem" Enables TLS if set + * CertificateFile "client-cert.pem" optional + * CertificateKeyFile "client-key.pem" optional + * TLSProtocol "tlsv1.2" optional * */ static int mqtt_config_publisher (oconfig_item_t *ci) { mqtt_client_conf_t *conf; - user_data_t user_data; + char cb_name[1024]; int status; - int i; conf = calloc (1, sizeof (*conf)); if (conf == NULL) { - ERROR ("mqtt plugin: malloc failed."); + ERROR ("mqtt plugin: calloc failed."); return (-1); } conf->publish = 1; @@ -496,11 +569,20 @@ static int mqtt_config_publisher (oconfig_item_t *ci) conf->host = strdup (MQTT_DEFAULT_HOST); conf->port = MQTT_DEFAULT_PORT; conf->client_id = NULL; + conf->qos = 0; conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX); + conf->store_rates = 1; + + status = pthread_mutex_init (&conf->lock, NULL); + if (status != 0) + { + mqtt_free (conf); + return (status); + } C_COMPLAIN_INIT (&conf->complaint_cantpublish); - for (i = 0; i < ci->children_num; i++) + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; if (strcasecmp ("Host", child->key) == 0) @@ -534,14 +616,24 @@ static int mqtt_config_publisher (oconfig_item_t *ci) cf_util_get_boolean (child, &conf->store_rates); else if (strcasecmp ("Retain", child->key) == 0) cf_util_get_boolean (child, &conf->retain); + else if (strcasecmp ("CACert", child->key) == 0) + cf_util_get_string (child, &conf->cacertificatefile); + else if (strcasecmp ("CertificateFile", child->key) == 0) + cf_util_get_string (child, &conf->certificatefile); + else if (strcasecmp ("CertificateKeyFile", child->key) == 0) + cf_util_get_string (child, &conf->certificatekeyfile); + else if (strcasecmp ("TLSProtocol", child->key) == 0) + cf_util_get_string (child, &conf->tlsprotocol); + else if (strcasecmp ("CipherSuite", child->key) == 0) + cf_util_get_string (child, &conf->ciphersuite); else ERROR ("mqtt plugin: Unknown config option: %s", child->key); } - memset (&user_data, 0, sizeof (user_data)); - user_data.data = conf; - - plugin_register_write ("mqtt", mqtt_write, &user_data); + ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name); + plugin_register_write (cb_name, mqtt_write, &(user_data_t) { + .data = conf, + }); return (0); } /* mqtt_config_publisher */ @@ -553,19 +645,18 @@ static int mqtt_config_publisher (oconfig_item_t *ci) * User "guest" * Password "secret" * Topic "collectd/#" - * + * */ static int mqtt_config_subscriber (oconfig_item_t *ci) { mqtt_client_conf_t **tmp; mqtt_client_conf_t *conf; int status; - int i; conf = calloc (1, sizeof (*conf)); if (conf == NULL) { - ERROR ("mqtt plugin: malloc failed."); + ERROR ("mqtt plugin: calloc failed."); return (-1); } conf->publish = 0; @@ -581,22 +672,31 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) conf->host = strdup (MQTT_DEFAULT_HOST); conf->port = MQTT_DEFAULT_PORT; conf->client_id = NULL; + conf->qos = 2; conf->topic = strdup (MQTT_DEFAULT_TOPIC); + conf->clean_session = 1; + + status = pthread_mutex_init (&conf->lock, NULL); + if (status != 0) + { + mqtt_free (conf); + return (status); + } C_COMPLAIN_INIT (&conf->complaint_cantpublish); - for (i = 0; i < ci->children_num; i++) + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; if (strcasecmp ("Host", child->key) == 0) cf_util_get_string (child, &conf->host); else if (strcasecmp ("Port", child->key) == 0) { - int tmp = cf_util_get_port_number (child); - if (tmp < 0) + status = cf_util_get_port_number (child); + if (status < 0) ERROR ("mqtt plugin: Invalid port number."); else - conf->port = tmp; + conf->port = status; } else if (strcasecmp ("ClientId", child->key) == 0) cf_util_get_string (child, &conf->client_id); @@ -606,12 +706,12 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) cf_util_get_string (child, &conf->password); else if (strcasecmp ("QoS", child->key) == 0) { - int tmp = -1; - status = cf_util_get_int (child, &tmp); - if ((status != 0) || (tmp < 0) || (tmp > 2)) + int qos = -1; + status = cf_util_get_int (child, &qos); + if ((status != 0) || (qos < 0) || (qos > 2)) ERROR ("mqtt plugin: Not a valid QoS setting."); else - conf->qos = tmp; + conf->qos = qos; } else if (strcasecmp ("Topic", child->key) == 0) cf_util_get_string (child, &conf->topic); @@ -621,7 +721,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) ERROR ("mqtt plugin: Unknown config option: %s", child->key); } - tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num); + tmp = realloc (subscribers, sizeof (*subscribers) * (subscribers_num + 1) ); if (tmp == NULL) { ERROR ("mqtt plugin: realloc failed."); @@ -647,9 +747,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) */ static int mqtt_config (oconfig_item_t *ci) { - int i; - - for (i = 0; i < ci->children_num; i++) + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; @@ -666,11 +764,9 @@ static int mqtt_config (oconfig_item_t *ci) static int mqtt_init (void) { - size_t i; - mosquitto_lib_init (); - for (i = 0; i < subscribers_num; i++) + for (size_t i = 0; i < subscribers_num; i++) { int status;