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 **/
24 // Reference: http://mosquitto.org/api/files/mosquitto-h.html
27 #include "collectd.h"
28 #include "common.h"
29 #include "plugin.h"
30 #include "utils_cache.h"
31 #include "utils_complain.h"
33 #include <pthread.h>
35 #include <mosquitto.h>
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"
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 typedef struct mqtt_client_conf mqtt_client_conf_t;
60 static char const *mosquitto_strerror (int code)
61 {
62 switch (code)
63 {
64 case MOSQ_ERR_SUCCESS: return "MOSQ_ERR_SUCCESS";
65 case MOSQ_ERR_NOMEM: return "MOSQ_ERR_NOMEM";
66 case MOSQ_ERR_PROTOCOL: return "MOSQ_ERR_PROTOCOL";
67 case MOSQ_ERR_INVAL: return "MOSQ_ERR_INVAL";
68 case MOSQ_ERR_NO_CONN: return "MOSQ_ERR_NO_CONN";
69 case MOSQ_ERR_CONN_REFUSED: return "MOSQ_ERR_CONN_REFUSED";
70 case MOSQ_ERR_NOT_FOUND: return "MOSQ_ERR_NOT_FOUND";
71 case MOSQ_ERR_CONN_LOST: return "MOSQ_ERR_CONN_LOST";
72 case MOSQ_ERR_SSL: return "MOSQ_ERR_SSL";
73 case MOSQ_ERR_PAYLOAD_SIZE: return "MOSQ_ERR_PAYLOAD_SIZE";
74 case MOSQ_ERR_NOT_SUPPORTED: return "MOSQ_ERR_NOT_SUPPORTED";
75 case MOSQ_ERR_AUTH: return "MOSQ_ERR_AUTH";
76 case MOSQ_ERR_ACL_DENIED: return "MOSQ_ERR_ACL_DENIED";
77 case MOSQ_ERR_UNKNOWN: return "MOSQ_ERR_UNKNOWN";
78 case MOSQ_ERR_ERRNO: return "MOSQ_ERR_ERRNO";
79 }
81 return "UNKNOWN ERROR CODE";
82 }
84 /*
85 * Functions
86 */
87 /* must hold conf->lock when calling. */
88 static int mqtt_reconnect_broker (mqtt_client_conf_t *conf)
89 {
90 int status;
92 if (conf->connected)
93 return (0);
95 status = mosquitto_reconnect (conf->mosq);
96 if (status != MOSQ_ERR_SUCCESS)
97 {
98 ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
99 (status == MOSQ_ERR_ERRNO ?
100 strerror(errno) : mosquitto_strerror (status)));
101 return (-1);
102 }
104 conf->connected = true;
106 c_release (LOG_INFO,
107 &conf->complaint_cantpublish,
108 "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
109 conf->host, conf->port);
111 return (0);
112 } /* mqtt_reconnect_broker */
114 static int mqtt_publish_message (mqtt_client_conf_t *conf, char *topic,
115 void const *payload, size_t payload_len)
116 {
117 char errbuf[1024];
118 int status;
120 pthread_mutex_lock (&conf->lock);
122 status = mqtt_reconnect_broker (conf);
123 if (status != 0) {
124 pthread_mutex_unlock (&conf->lock);
125 ERROR ("mqtt plugin: unable to reconnect to broker");
126 return (status);
127 }
129 status = mosquitto_publish(conf->mosq,
130 /* message id */ NULL,
131 topic,
132 (int) payload_len,
133 payload,
134 /* qos */ 0,
135 /* retain */ false);
136 if (status != MOSQ_ERR_SUCCESS)
137 {
138 c_complain (LOG_ERR,
139 &conf->complaint_cantpublish,
140 "plugin mqtt: mosquitto_publish failed: %s",
141 status == MOSQ_ERR_ERRNO ?
142 sstrerror(errno, errbuf, sizeof (errbuf)) :
143 mosquitto_strerror(status));
144 /*
145 Mark our connection "down" regardless of the error as a safety measure;
146 we will try to reconnect the next time we have to publish a message
147 */
148 conf->connected = false;
150 pthread_mutex_unlock (&conf->lock);
151 return (-1);
152 }
154 pthread_mutex_unlock (&conf->lock);
155 return (0);
156 } /* mqtt_publish_message */
158 static int mqtt_format_metric_value (char *buf, size_t buf_len,
159 const data_set_t *data_set, const value_list_t *vl, int ds_num)
160 {
161 gauge_t *rates = NULL;
162 gauge_t *value = NULL;
163 size_t metric_value_len;
164 int status = 0;
166 memset (buf, 0, buf_len);
168 if (data_set->ds[ds_num].type == DS_TYPE_GAUGE)
169 value = &vl->values[ds_num].gauge;
170 else {
171 rates = uc_get_rate (data_set, vl);
172 value = &rates[ds_num];
173 }
175 metric_value_len = ssnprintf (buf, buf_len, "%f", *value);
177 if (metric_value_len >= buf_len)
178 return (-ENOMEM);
180 if (rates)
181 sfree (rates);
183 return (status);
184 } /* mqtt_format_metric_value */
186 static int mqtt_format_message_topic (char *buf, size_t buf_len,
187 char const *prefix, const value_list_t *vl, const char *ds_name)
188 {
189 size_t topic_buf_len;
191 memset (buf, 0, buf_len);
193 /*
194 MQTT message topic format:
195 [<prefix>/]<hostname>/<plugin>/<plugin instance>/<type>/<type instance>/<ds>/
196 */
197 topic_buf_len = (size_t) ssnprintf (buf, buf_len,
198 "%s/%s/%s/%s/%s/%s/%s",
199 prefix,
200 vl->host,
201 vl->plugin,
202 vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "(null)",
203 vl->type,
204 vl->type_instance[0] != '\0' ? vl->type_instance : "(null)",
205 ds_name);
207 if (topic_buf_len >= buf_len)
208 {
209 ERROR ("mqtt_format_message_topic: topic buffer too small: "
210 "Need %zu bytes.", topic_buf_len + 1);
211 return (-ENOMEM);
212 }
214 return (0);
215 } /* mqtt_format_message_topic */
217 static int mqtt_format_payload (char *buf, size_t buf_len,
218 const data_set_t *data_set, const value_list_t *vl, int ds_num)
219 {
220 char metric_path[10 * DATA_MAX_NAME_LEN];
221 char metric_value[512];
222 size_t payload_buf_len;
223 int status = 0;
225 memset (buf, 0, buf_len);
227 ssnprintf (metric_path, sizeof (metric_path),
228 "%s.%s%s%s.%s%s%s%s%s",
229 vl->host,
230 vl->plugin,
231 vl->plugin_instance[0] != '\0' ? "." : "",
232 vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "",
233 vl->type,
234 vl->type_instance[0] != '\0' ? "." : "",
235 vl->type_instance[0] != '\0' ? vl->type_instance : "",
236 strcmp(data_set->ds[ds_num].name, "value") != 0 ? "." : "",
237 strcmp(data_set->ds[ds_num].name, "value") != 0 ?
238 data_set->ds[ds_num].name : "");
240 status = mqtt_format_metric_value (metric_value,
241 sizeof (metric_value),
242 data_set,
243 vl,
244 ds_num);
246 if (status != 0)
247 {
248 ERROR ("mqtt_format_payload: error with mqtt_format_metric_value");
249 return (status);
250 }
252 payload_buf_len = (size_t) ssnprintf (buf, buf_len,
253 "%s %s %u",
254 metric_path,
255 metric_value,
256 (unsigned int) CDTIME_T_TO_TIME_T (vl->time));
258 if (payload_buf_len >= buf_len)
259 {
260 ERROR ("mqtt_format_payload: payload buffer too small: "
261 "Need %zu bytes.", payload_buf_len + 1);
262 return (-ENOMEM);
263 }
265 return (status);
266 } /* mqtt_format_payload */
268 static int mqtt_write (const data_set_t *data_set, const value_list_t *vl,
269 user_data_t *user_data)
270 {
271 struct mqtt_client_conf *conf;
272 char msg_topic[MQTT_MAX_TOPIC_SIZE];
273 char msg_payload[MQTT_MAX_MESSAGE_SIZE];
274 int status = 0;
275 int i;
277 if (user_data == NULL)
278 return (EINVAL);
280 conf = user_data->data;
282 if (!conf->connected)
283 {
284 status = mqtt_reconnect_broker (conf);
286 if (status != 0) {
287 ERROR ("plugin mqtt: unable to reconnect to broker");
288 return (status);
289 }
290 }
292 for (i = 0; i < data_set->ds_num; i++)
293 {
294 status = mqtt_format_message_topic (msg_topic, sizeof (msg_topic),
295 conf->topic_prefix, vl, data_set->ds[i].name);
296 if (status != 0)
297 {
298 ERROR ("plugin mqtt: error with mqtt_format_message_topic");
299 return (status);
300 }
302 status = mqtt_format_payload (msg_payload,
303 sizeof (msg_payload),
304 data_set,
305 vl,
306 i);
308 if (status != 0)
309 {
310 ERROR ("mqtt_write: error with mqtt_format_payload");
311 return (status);
312 }
314 status = mqtt_publish_message (conf,
315 msg_topic,
316 msg_payload,
317 sizeof (msg_payload));
318 if (status != 0)
319 {
320 ERROR ("plugin mqtt: unable to publish message");
321 return (status);
322 }
324 DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_write[%02X]\x1B[0m "
325 "published message: topic=%s payload=%s",
326 (unsigned)pthread_self(),
327 msg_topic,
328 msg_payload);
329 }
331 return (status);
332 } /* mqtt_write */
334 static int mqtt_config (oconfig_item_t *ci)
335 {
336 struct mqtt_client_conf *conf;
337 user_data_t user_data;
338 char errbuf[1024];
339 int status;
341 DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_config[%02X]\x1B[0m ",
342 (unsigned)pthread_self());
344 conf = malloc (sizeof (*conf));
345 if (conf == NULL)
346 {
347 ERROR ("write_mqtt plugin: malloc failed.");
348 return (-1);
349 }
351 memset (conf, 0, sizeof (*conf));
353 conf->connected = false;
354 conf->host = MQTT_DEFAULT_HOST;
355 conf->port = MQTT_DEFAULT_PORT;
356 conf->client_id = MQTT_DEFAULT_CLIENT_ID;
357 conf->topic_prefix = MQTT_DEFAULT_TOPIC_PREFIX;
358 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
360 memset (&user_data, 0, sizeof (user_data));
361 user_data.data = conf;
363 conf->mosq = mosquitto_new (conf->client_id, /* user data = */ conf);
364 if (conf->mosq == NULL)
365 {
366 ERROR ("mqtt plugin: mosquitto_new failed");
367 free (conf);
368 return (-1);
369 }
371 status = mosquitto_connect (conf->mosq, conf->host, conf->port,
372 /* keepalive = */ 10, /* clean session = */ 1);
373 if (status != MOSQ_ERR_SUCCESS) {
374 ERROR ("mqtt_config: mosquitto_connect failed: %s",
375 (status == MOSQ_ERR_ERRNO ?
376 sstrerror(errno, errbuf, sizeof (errbuf)) :
377 mosquitto_strerror (status)));
378 return (-1);
379 }
381 DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"",
382 conf->host, conf->port);
384 conf->connected = true;
386 plugin_register_write ("mqtt", mqtt_write, &user_data);
388 return (0);
389 } /* mqtt_config */
391 static int mqtt_init (void)
392 {
393 mosquitto_lib_init();
395 return (0);
396 } /* mqtt_init */
398 void module_register (void)
399 {
400 plugin_register_complex_config ("mqtt", mqtt_config);
401 plugin_register_init ("mqtt", mqtt_init);
402 } /* void module_register */
404 /* vim: set sw=4 sts=4 et fdm=marker : */