3bed0a5e32e537b9898c4bb7cb3b63085f529ce7
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 };
59 /*
60 * Functions
61 */
62 static int mqtt_reconnect_broker (struct mqtt_client_conf *conf)
63 {
64 int status;
66 if (conf->connected)
67 return (0);
69 pthread_mutex_lock (&conf->lock);
71 status = mosquitto_reconnect (conf->mosq);
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 }
81 conf->connected = true;
83 c_release (LOG_INFO,
84 &conf->complaint_cantpublish,
85 "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
86 conf->host, conf->port);
88 pthread_mutex_unlock (&conf->lock);
90 return (0);
91 } /* mqtt_reconnect_broker */
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;
99 status = mosquitto_publish(conf->mosq,
100 /* message id */ NULL,
101 topic,
102 (int) payload_len,
103 payload,
104 /* qos */ 0,
105 /* retain */ false);
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;
121 return (-1);
122 }
124 return (0);
125 } /* mqtt_publish_message */
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;
135 memset (buf, 0, buf_len);
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 }
144 metric_value_len = ssnprintf (buf, buf_len, "%f", *value);
146 if (metric_value_len >= buf_len)
147 return (-ENOMEM);
149 if (rates)
150 sfree (rates);
152 return (status);
153 } /* mqtt_format_metric_value */
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;
160 memset (buf, 0, buf_len);
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);
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 }
183 return (0);
184 } /* mqtt_format_message_topic */
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;
194 memset (buf, 0, buf_len);
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 : "");
209 status = mqtt_format_metric_value (metric_value,
210 sizeof (metric_value),
211 data_set,
212 vl,
213 ds_num);
215 if (status != 0)
216 {
217 ERROR ("mqtt_format_payload: error with mqtt_format_metric_value");
218 return (status);
219 }
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));
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 }
234 return (status);
235 } /* mqtt_format_payload */
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;
246 if (user_data == NULL)
247 return (EINVAL);
249 conf = user_data->data;
251 if (!conf->connected)
252 {
253 status = mqtt_reconnect_broker (conf);
255 if (status != 0) {
256 ERROR ("plugin mqtt: unable to reconnect to broker");
257 return (status);
258 }
259 }
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 }
271 status = mqtt_format_payload (msg_payload,
272 sizeof (msg_payload),
273 data_set,
274 vl,
275 i);
277 if (status != 0)
278 {
279 ERROR ("mqtt_write: error with mqtt_format_payload");
280 return (status);
281 }
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 }
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 }
300 return (status);
301 } /* mqtt_write */
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;
310 DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_config[%02X]\x1B[0m ",
311 (unsigned)pthread_self());
313 conf = malloc (sizeof (*conf));
314 if (conf == NULL)
315 {
316 ERROR ("write_mqtt plugin: malloc failed.");
317 return (-1);
318 }
320 memset (conf, 0, sizeof (*conf));
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);
329 memset (&user_data, 0, sizeof (user_data));
330 user_data.data = conf;
332 if ((conf->mosq = mosquitto_new (conf->client_id, true, NULL)) == NULL) {
333 ERROR ("mqtt_config: mosquitto_new failed");
334 return (-1);
335 }
337 status = mosquitto_connect (conf->mosq, conf->host, conf->port, 10);
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 }
347 DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"",
348 conf->host, conf->port);
350 conf->connected = true;
352 plugin_register_write ("mqtt", mqtt_write, &user_data);
354 return (0);
355 } /* mqtt_config */
357 static int mqtt_init (void)
358 {
359 mosquitto_lib_init();
361 return (0);
362 } /* mqtt_init */
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 */
370 /* vim: set sw=4 sts=4 et fdm=marker : */