39575b93e68d52e2ba98e25716eac78b57d35e0f
[collectd.git] / src / amqp.c
1 /**
2  * collectd - src/amqp.c
3  * Copyright (C) 2009       Sebastien Pahl
4  * Copyright (C) 2010-2012  Florian Forster
5  *
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:
12  *
13  * The above copyright notice and this permission notice shall be included in
14  * all copies or substantial portions of the Software.
15  *
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.
23  *
24  * Authors:
25  *   Sebastien Pahl <sebastien.pahl at dotcloud.com>
26  *   Florian Forster <octo at collectd.org>
27  **/
28
29 #include "collectd.h"
30
31 #include "plugin.h"
32 #include "utils/cmds/putval.h"
33 #include "utils/common/common.h"
34 #include "utils/format_graphite/format_graphite.h"
35 #include "utils/format_json/format_json.h"
36
37 #include <amqp.h>
38 #include <amqp_framing.h>
39
40 #ifdef HAVE_AMQP_TCP_SOCKET_H
41 #include <amqp_tcp_socket.h>
42 #endif
43 #ifdef HAVE_AMQP_SOCKET_H
44 #include <amqp_socket.h>
45 #endif
46 #ifdef HAVE_AMQP_TCP_SOCKET
47 #if defined HAVE_DECL_AMQP_SOCKET_CLOSE && !HAVE_DECL_AMQP_SOCKET_CLOSE
48 /* rabbitmq-c does not currently ship amqp_socket.h
49  * and, thus, does not define this function. */
50 int amqp_socket_close(amqp_socket_t *);
51 #endif
52 #endif
53
54 /* Defines for the delivery mode. I have no idea why they're not defined by the
55  * library.. */
56 #define CAMQP_DM_VOLATILE 1
57 #define CAMQP_DM_PERSISTENT 2
58
59 #define CAMQP_FORMAT_COMMAND 1
60 #define CAMQP_FORMAT_JSON 2
61 #define CAMQP_FORMAT_GRAPHITE 3
62
63 #define CAMQP_CHANNEL 1
64
65 /*
66  * Data types
67  */
68 struct camqp_config_s {
69   bool publish;
70   char *name;
71
72   char *host;
73   int port;
74   char *vhost;
75   char *user;
76   char *password;
77
78   char *exchange;
79   char *routing_key;
80
81   /* Number of seconds to wait before connection is retried */
82   int connection_retry_delay;
83
84   /* publish only */
85   uint8_t delivery_mode;
86   bool store_rates;
87   int format;
88   /* publish & graphite format only */
89   char *prefix;
90   char *postfix;
91   char escape_char;
92   unsigned int graphite_flags;
93
94   /* subscribe only */
95   char *exchange_type;
96   char *queue;
97   bool queue_durable;
98   bool queue_auto_delete;
99
100   amqp_connection_state_t connection;
101   pthread_mutex_t lock;
102 };
103 typedef struct camqp_config_s camqp_config_t;
104
105 /*
106  * Global variables
107  */
108 static const char *def_host = "localhost";
109 static const char *def_vhost = "/";
110 static const char *def_user = "guest";
111 static const char *def_password = "guest";
112 static const char *def_exchange = "amq.fanout";
113
114 static pthread_t *subscriber_threads;
115 static size_t subscriber_threads_num;
116 static bool subscriber_threads_running = true;
117
118 #define CONF(c, f) (((c)->f != NULL) ? (c)->f : def_##f)
119
120 /*
121  * Functions
122  */
123 static void camqp_close_connection(camqp_config_t *conf) /* {{{ */
124 {
125   int sockfd;
126
127   if ((conf == NULL) || (conf->connection == NULL))
128     return;
129
130   sockfd = amqp_get_sockfd(conf->connection);
131   amqp_channel_close(conf->connection, CAMQP_CHANNEL, AMQP_REPLY_SUCCESS);
132   amqp_connection_close(conf->connection, AMQP_REPLY_SUCCESS);
133   amqp_destroy_connection(conf->connection);
134   close(sockfd);
135   conf->connection = NULL;
136 } /* }}} void camqp_close_connection */
137
138 static void camqp_config_free(void *ptr) /* {{{ */
139 {
140   camqp_config_t *conf = ptr;
141
142   if (conf == NULL)
143     return;
144
145   camqp_close_connection(conf);
146
147   sfree(conf->name);
148   sfree(conf->host);
149   sfree(conf->vhost);
150   sfree(conf->user);
151   sfree(conf->password);
152   sfree(conf->exchange);
153   sfree(conf->exchange_type);
154   sfree(conf->queue);
155   sfree(conf->routing_key);
156   sfree(conf->prefix);
157   sfree(conf->postfix);
158
159   sfree(conf);
160 } /* }}} void camqp_config_free */
161
162 static char *camqp_bytes_cstring(amqp_bytes_t *in) /* {{{ */
163 {
164   char *ret;
165
166   if ((in == NULL) || (in->bytes == NULL))
167     return NULL;
168
169   ret = malloc(in->len + 1);
170   if (ret == NULL)
171     return NULL;
172
173   memcpy(ret, in->bytes, in->len);
174   ret[in->len] = 0;
175
176   return ret;
177 } /* }}} char *camqp_bytes_cstring */
178
179 static bool camqp_is_error(camqp_config_t *conf) /* {{{ */
180 {
181   amqp_rpc_reply_t r;
182
183   r = amqp_get_rpc_reply(conf->connection);
184   if (r.reply_type == AMQP_RESPONSE_NORMAL)
185     return false;
186
187   return true;
188 } /* }}} bool camqp_is_error */
189
190 static char *camqp_strerror(camqp_config_t *conf, /* {{{ */
191                             char *buffer, size_t buffer_size) {
192   amqp_rpc_reply_t r;
193
194   r = amqp_get_rpc_reply(conf->connection);
195   switch (r.reply_type) {
196   case AMQP_RESPONSE_NORMAL:
197     sstrncpy(buffer, "Success", buffer_size);
198     break;
199
200   case AMQP_RESPONSE_NONE:
201     sstrncpy(buffer, "Missing RPC reply type", buffer_size);
202     break;
203
204   case AMQP_RESPONSE_LIBRARY_EXCEPTION:
205 #if HAVE_AMQP_RPC_REPLY_T_LIBRARY_ERRNO
206     if (r.library_errno)
207       return sstrerror(r.library_errno, buffer, buffer_size);
208 #else
209     if (r.library_error)
210       return sstrerror(r.library_error, buffer, buffer_size);
211 #endif
212     else
213       sstrncpy(buffer, "End of stream", buffer_size);
214     break;
215
216   case AMQP_RESPONSE_SERVER_EXCEPTION:
217     if (r.reply.id == AMQP_CONNECTION_CLOSE_METHOD) {
218       amqp_connection_close_t *m = r.reply.decoded;
219       char *tmp = camqp_bytes_cstring(&m->reply_text);
220       ssnprintf(buffer, buffer_size, "Server connection error %d: %s",
221                 m->reply_code, tmp);
222       sfree(tmp);
223     } else if (r.reply.id == AMQP_CHANNEL_CLOSE_METHOD) {
224       amqp_channel_close_t *m = r.reply.decoded;
225       char *tmp = camqp_bytes_cstring(&m->reply_text);
226       ssnprintf(buffer, buffer_size, "Server channel error %d: %s",
227                 m->reply_code, tmp);
228       sfree(tmp);
229     } else {
230       ssnprintf(buffer, buffer_size, "Server error method %#" PRIx32,
231                 r.reply.id);
232     }
233     break;
234
235   default:
236     ssnprintf(buffer, buffer_size, "Unknown reply type %i", (int)r.reply_type);
237   }
238
239   return buffer;
240 } /* }}} char *camqp_strerror */
241
242 #if HAVE_AMQP_RPC_REPLY_T_LIBRARY_ERRNO
243 static int camqp_create_exchange(camqp_config_t *conf) /* {{{ */
244 {
245   amqp_exchange_declare_ok_t *ed_ret;
246
247   if (conf->exchange_type == NULL)
248     return 0;
249
250   ed_ret = amqp_exchange_declare(
251       conf->connection,
252       /* channel     = */ CAMQP_CHANNEL,
253       /* exchange    = */ amqp_cstring_bytes(conf->exchange),
254       /* type        = */ amqp_cstring_bytes(conf->exchange_type),
255       /* passive     = */ 0,
256       /* durable     = */ 0,
257       /* auto_delete = */ 1,
258       /* arguments   = */ AMQP_EMPTY_TABLE);
259   if ((ed_ret == NULL) && camqp_is_error(conf)) {
260     char errbuf[1024];
261     ERROR("amqp plugin: amqp_exchange_declare failed: %s",
262           camqp_strerror(conf, errbuf, sizeof(errbuf)));
263     camqp_close_connection(conf);
264     return -1;
265   }
266
267   INFO("amqp plugin: Successfully created exchange \"%s\" "
268        "with type \"%s\".",
269        conf->exchange, conf->exchange_type);
270
271   return 0;
272 } /* }}} int camqp_create_exchange */
273 #else
274 static int camqp_create_exchange(camqp_config_t *conf) /* {{{ */
275 {
276   amqp_exchange_declare_ok_t *ed_ret;
277   amqp_table_t argument_table;
278   struct amqp_table_entry_t_ argument_table_entries[1];
279
280   if (conf->exchange_type == NULL)
281     return 0;
282
283   /* Valid arguments: "auto_delete", "internal" */
284   argument_table.num_entries = STATIC_ARRAY_SIZE(argument_table_entries);
285   argument_table.entries = argument_table_entries;
286   argument_table_entries[0].key = amqp_cstring_bytes("auto_delete");
287   argument_table_entries[0].value.kind = AMQP_FIELD_KIND_BOOLEAN;
288   argument_table_entries[0].value.value.boolean = 1;
289
290   ed_ret = amqp_exchange_declare(
291       conf->connection,
292       /* channel     = */ CAMQP_CHANNEL,
293       /* exchange    = */ amqp_cstring_bytes(conf->exchange),
294       /* type        = */ amqp_cstring_bytes(conf->exchange_type),
295       /* passive     = */ 0,
296       /* durable     = */ 0,
297 #if defined(AMQP_VERSION) && AMQP_VERSION >= 0x00060000
298       /* auto delete = */ 0,
299       /* internal    = */ 0,
300 #endif
301       /* arguments   = */ argument_table);
302   if ((ed_ret == NULL) && camqp_is_error(conf)) {
303     char errbuf[1024];
304     ERROR("amqp plugin: amqp_exchange_declare failed: %s",
305           camqp_strerror(conf, errbuf, sizeof(errbuf)));
306     camqp_close_connection(conf);
307     return -1;
308   }
309
310   INFO("amqp plugin: Successfully created exchange \"%s\" "
311        "with type \"%s\".",
312        conf->exchange, conf->exchange_type);
313
314   return 0;
315 } /* }}} int camqp_create_exchange */
316 #endif
317
318 static int camqp_setup_queue(camqp_config_t *conf) /* {{{ */
319 {
320   amqp_queue_declare_ok_t *qd_ret;
321   amqp_basic_consume_ok_t *cm_ret;
322
323   qd_ret = amqp_queue_declare(conf->connection,
324                               /* channel     = */ CAMQP_CHANNEL,
325                               /* queue       = */
326                                   (conf->queue != NULL)
327                                   ? amqp_cstring_bytes(conf->queue)
328                                   : AMQP_EMPTY_BYTES,
329                               /* passive     = */ 0,
330                               /* durable     = */ conf->queue_durable,
331                               /* exclusive   = */ 0,
332                               /* auto_delete = */ conf->queue_auto_delete,
333                               /* arguments   = */ AMQP_EMPTY_TABLE);
334   if (qd_ret == NULL) {
335     ERROR("amqp plugin: amqp_queue_declare failed.");
336     camqp_close_connection(conf);
337     return -1;
338   }
339
340   if (conf->queue == NULL) {
341     conf->queue = camqp_bytes_cstring(&qd_ret->queue);
342     if (conf->queue == NULL) {
343       ERROR("amqp plugin: camqp_bytes_cstring failed.");
344       camqp_close_connection(conf);
345       return -1;
346     }
347
348     INFO("amqp plugin: Created queue \"%s\".", conf->queue);
349   }
350   DEBUG("amqp plugin: Successfully created queue \"%s\".", conf->queue);
351
352   /* bind to an exchange */
353   if (conf->exchange != NULL) {
354     amqp_queue_bind_ok_t *qb_ret;
355
356     assert(conf->queue != NULL);
357     qb_ret = amqp_queue_bind(
358         conf->connection,
359         /* channel     = */ CAMQP_CHANNEL,
360         /* queue       = */ amqp_cstring_bytes(conf->queue),
361         /* exchange    = */ amqp_cstring_bytes(conf->exchange),
362         /* routing_key = */
363             (conf->routing_key != NULL) ? amqp_cstring_bytes(conf->routing_key)
364                                         : AMQP_EMPTY_BYTES,
365         /* arguments   = */ AMQP_EMPTY_TABLE);
366     if ((qb_ret == NULL) && camqp_is_error(conf)) {
367       char errbuf[1024];
368       ERROR("amqp plugin: amqp_queue_bind failed: %s",
369             camqp_strerror(conf, errbuf, sizeof(errbuf)));
370       camqp_close_connection(conf);
371       return -1;
372     }
373
374     DEBUG("amqp plugin: Successfully bound queue \"%s\" to exchange \"%s\".",
375           conf->queue, conf->exchange);
376   } /* if (conf->exchange != NULL) */
377
378   cm_ret =
379       amqp_basic_consume(conf->connection,
380                          /* channel      = */ CAMQP_CHANNEL,
381                          /* queue        = */ amqp_cstring_bytes(conf->queue),
382                          /* consumer_tag = */ AMQP_EMPTY_BYTES,
383                          /* no_local     = */ 0,
384                          /* no_ack       = */ 1,
385                          /* exclusive    = */ 0,
386                          /* arguments    = */ AMQP_EMPTY_TABLE);
387   if ((cm_ret == NULL) && camqp_is_error(conf)) {
388     char errbuf[1024];
389     ERROR("amqp plugin: amqp_basic_consume failed: %s",
390           camqp_strerror(conf, errbuf, sizeof(errbuf)));
391     camqp_close_connection(conf);
392     return -1;
393   }
394
395   return 0;
396 } /* }}} int camqp_setup_queue */
397
398 static int camqp_connect(camqp_config_t *conf) /* {{{ */
399 {
400   static time_t last_connect_time;
401
402   amqp_rpc_reply_t reply;
403   int status;
404 #ifdef HAVE_AMQP_TCP_SOCKET
405   amqp_socket_t *socket;
406 #else
407   int sockfd;
408 #endif
409
410   if (conf->connection != NULL)
411     return 0;
412
413   time_t now = time(NULL);
414   if (now < (last_connect_time + conf->connection_retry_delay)) {
415     DEBUG("amqp plugin: skipping connection retry, "
416           "ConnectionRetryDelay: %d",
417           conf->connection_retry_delay);
418     return 1;
419   } else {
420     DEBUG("amqp plugin: retrying connection");
421     last_connect_time = now;
422   }
423
424   conf->connection = amqp_new_connection();
425   if (conf->connection == NULL) {
426     ERROR("amqp plugin: amqp_new_connection failed.");
427     return ENOMEM;
428   }
429
430 #ifdef HAVE_AMQP_TCP_SOCKET
431 #define CLOSE_SOCKET() /* amqp_destroy_connection() closes the socket for us   \
432                         */
433   /* TODO: add support for SSL using amqp_ssl_socket_new
434    *       and related functions */
435   socket = amqp_tcp_socket_new(conf->connection);
436   if (!socket) {
437     ERROR("amqp plugin: amqp_tcp_socket_new failed.");
438     amqp_destroy_connection(conf->connection);
439     conf->connection = NULL;
440     return ENOMEM;
441   }
442
443   status = amqp_socket_open(socket, CONF(conf, host), conf->port);
444   if (status < 0) {
445     status *= -1;
446     ERROR("amqp plugin: amqp_socket_open failed: %s", STRERROR(status));
447     amqp_destroy_connection(conf->connection);
448     conf->connection = NULL;
449     return status;
450   }
451 #else /* HAVE_AMQP_TCP_SOCKET */
452 #define CLOSE_SOCKET() close(sockfd)
453   /* this interface is deprecated as of rabbitmq-c 0.4 */
454   sockfd = amqp_open_socket(CONF(conf, host), conf->port);
455   if (sockfd < 0) {
456     status = (-1) * sockfd;
457     ERROR("amqp plugin: amqp_open_socket failed: %s", STRERROR(status));
458     amqp_destroy_connection(conf->connection);
459     conf->connection = NULL;
460     return status;
461   }
462   amqp_set_sockfd(conf->connection, sockfd);
463 #endif
464
465   reply = amqp_login(conf->connection, CONF(conf, vhost),
466                      /* channel max = */ 0,
467                      /* frame max   = */ 131072,
468                      /* heartbeat   = */ 0,
469                      /* authentication = */ AMQP_SASL_METHOD_PLAIN,
470                      CONF(conf, user), CONF(conf, password));
471   if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
472     ERROR("amqp plugin: amqp_login (vhost = %s, user = %s) failed.",
473           CONF(conf, vhost), CONF(conf, user));
474     amqp_destroy_connection(conf->connection);
475     CLOSE_SOCKET();
476     conf->connection = NULL;
477     return 1;
478   }
479
480   amqp_channel_open(conf->connection, /* channel = */ 1);
481   /* FIXME: Is checking "reply.reply_type" really correct here? How does
482    * it get set? --octo */
483   if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
484     ERROR("amqp plugin: amqp_channel_open failed.");
485     amqp_connection_close(conf->connection, AMQP_REPLY_SUCCESS);
486     amqp_destroy_connection(conf->connection);
487     CLOSE_SOCKET();
488     conf->connection = NULL;
489     return 1;
490   }
491
492   INFO("amqp plugin: Successfully opened connection to vhost \"%s\" "
493        "on %s:%i.",
494        CONF(conf, vhost), CONF(conf, host), conf->port);
495
496   status = camqp_create_exchange(conf);
497   if (status != 0)
498     return status;
499
500   if (!conf->publish)
501     return camqp_setup_queue(conf);
502   return 0;
503 } /* }}} int camqp_connect */
504
505 static int camqp_shutdown(void) /* {{{ */
506 {
507   DEBUG("amqp plugin: Shutting down %" PRIsz " subscriber threads.",
508         subscriber_threads_num);
509
510   subscriber_threads_running = 0;
511   for (size_t i = 0; i < subscriber_threads_num; i++) {
512     /* FIXME: Sending a signal is not very elegant here. Maybe find out how
513      * to use a timeout in the thread and check for the variable in regular
514      * intervals. */
515     pthread_kill(subscriber_threads[i], SIGTERM);
516     pthread_join(subscriber_threads[i], /* retval = */ NULL);
517   }
518
519   subscriber_threads_num = 0;
520   sfree(subscriber_threads);
521
522   DEBUG("amqp plugin: All subscriber threads exited.");
523
524   return 0;
525 } /* }}} int camqp_shutdown */
526
527 /*
528  * Subscribing code
529  */
530 static int camqp_read_body(camqp_config_t *conf, /* {{{ */
531                            size_t body_size, const char *content_type) {
532   char body[body_size + 1];
533   char *body_ptr;
534   size_t received;
535   amqp_frame_t frame;
536   int status;
537
538   memset(body, 0, sizeof(body));
539   body_ptr = &body[0];
540   received = 0;
541
542   while (received < body_size) {
543     status = amqp_simple_wait_frame(conf->connection, &frame);
544     if (status < 0) {
545       status = (-1) * status;
546       ERROR("amqp plugin: amqp_simple_wait_frame failed: %s", STRERROR(status));
547       camqp_close_connection(conf);
548       return status;
549     }
550
551     if (frame.frame_type != AMQP_FRAME_BODY) {
552       NOTICE("amqp plugin: Unexpected frame type: %#" PRIx8, frame.frame_type);
553       return -1;
554     }
555
556     if ((body_size - received) < frame.payload.body_fragment.len) {
557       WARNING("amqp plugin: Body is larger than indicated by header.");
558       return -1;
559     }
560
561     memcpy(body_ptr, frame.payload.body_fragment.bytes,
562            frame.payload.body_fragment.len);
563     body_ptr += frame.payload.body_fragment.len;
564     received += frame.payload.body_fragment.len;
565   } /* while (received < body_size) */
566
567   if (strcasecmp("text/collectd", content_type) == 0) {
568     status = cmd_handle_putval(stderr, body);
569     if (status != 0)
570       ERROR("amqp plugin: cmd_handle_putval failed with status %i.", status);
571     return status;
572   } else if (strcasecmp("application/json", content_type) == 0) {
573     ERROR("amqp plugin: camqp_read_body: Parsing JSON data has not "
574           "been implemented yet. FIXME!");
575     return 0;
576   } else {
577     ERROR("amqp plugin: camqp_read_body: Unknown content type \"%s\".",
578           content_type);
579     return EINVAL;
580   }
581
582   /* not reached */
583   return 0;
584 } /* }}} int camqp_read_body */
585
586 static int camqp_read_header(camqp_config_t *conf) /* {{{ */
587 {
588   int status;
589   amqp_frame_t frame;
590   amqp_basic_properties_t *properties;
591   char *content_type;
592
593   status = amqp_simple_wait_frame(conf->connection, &frame);
594   if (status < 0) {
595     status = (-1) * status;
596     ERROR("amqp plugin: amqp_simple_wait_frame failed: %s", STRERROR(status));
597     camqp_close_connection(conf);
598     return status;
599   }
600
601   if (frame.frame_type != AMQP_FRAME_HEADER) {
602     NOTICE("amqp plugin: Unexpected frame type: %#" PRIx8, frame.frame_type);
603     return -1;
604   }
605
606   properties = frame.payload.properties.decoded;
607   content_type = camqp_bytes_cstring(&properties->content_type);
608   if (content_type == NULL) {
609     ERROR("amqp plugin: Unable to determine content type.");
610     return -1;
611   }
612
613   status = camqp_read_body(conf, (size_t)frame.payload.properties.body_size,
614                            content_type);
615
616   sfree(content_type);
617   return status;
618 } /* }}} int camqp_read_header */
619
620 static void *camqp_subscribe_thread(void *user_data) /* {{{ */
621 {
622   camqp_config_t *conf = user_data;
623   int status;
624
625   cdtime_t interval = plugin_get_interval();
626
627   while (subscriber_threads_running) {
628     amqp_frame_t frame;
629
630     status = camqp_connect(conf);
631     if (status != 0) {
632       ERROR("amqp plugin: camqp_connect failed. "
633             "Will sleep for %.3f seconds.",
634             CDTIME_T_TO_DOUBLE(interval));
635       nanosleep(&CDTIME_T_TO_TIMESPEC(interval), /* remaining = */ NULL);
636       continue;
637     }
638
639     status = amqp_simple_wait_frame(conf->connection, &frame);
640     if (status < 0) {
641       ERROR("amqp plugin: amqp_simple_wait_frame failed. "
642             "Will sleep for %.3f seconds.",
643             CDTIME_T_TO_DOUBLE(interval));
644       camqp_close_connection(conf);
645       nanosleep(&CDTIME_T_TO_TIMESPEC(interval), /* remaining = */ NULL);
646       continue;
647     }
648
649     if (frame.frame_type != AMQP_FRAME_METHOD) {
650       DEBUG("amqp plugin: Unexpected frame type: %#" PRIx8, frame.frame_type);
651       continue;
652     }
653
654     if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) {
655       DEBUG("amqp plugin: Unexpected method id: %#" PRIx32,
656             frame.payload.method.id);
657       continue;
658     }
659
660     camqp_read_header(conf);
661
662     amqp_maybe_release_buffers(conf->connection);
663   } /* while (subscriber_threads_running) */
664
665   camqp_config_free(conf);
666   pthread_exit(NULL);
667   return NULL;
668 } /* }}} void *camqp_subscribe_thread */
669
670 static int camqp_subscribe_init(camqp_config_t *conf) /* {{{ */
671 {
672   int status;
673   pthread_t *tmp;
674
675   tmp = realloc(subscriber_threads,
676                 sizeof(*subscriber_threads) * (subscriber_threads_num + 1));
677   if (tmp == NULL) {
678     ERROR("amqp plugin: realloc failed.");
679     sfree(subscriber_threads);
680     return ENOMEM;
681   }
682   subscriber_threads = tmp;
683   tmp = subscriber_threads + subscriber_threads_num;
684   memset(tmp, 0, sizeof(*tmp));
685
686   status = plugin_thread_create(tmp, /* attr = */ NULL, camqp_subscribe_thread,
687                                 conf, "amqp subscribe");
688   if (status != 0) {
689     ERROR("amqp plugin: pthread_create failed: %s", STRERROR(status));
690     return status;
691   }
692
693   subscriber_threads_num++;
694
695   return 0;
696 } /* }}} int camqp_subscribe_init */
697
698 /*
699  * Publishing code
700  */
701 /* XXX: You must hold "conf->lock" when calling this function! */
702 static int camqp_write_locked(camqp_config_t *conf, /* {{{ */
703                               const char *buffer, const char *routing_key) {
704   int status;
705
706   status = camqp_connect(conf);
707   if (status != 0)
708     return status;
709
710   amqp_basic_properties_t props = {._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |
711                                              AMQP_BASIC_DELIVERY_MODE_FLAG |
712                                              AMQP_BASIC_APP_ID_FLAG,
713                                    .delivery_mode = conf->delivery_mode,
714                                    .app_id = amqp_cstring_bytes("collectd")};
715
716   if (conf->format == CAMQP_FORMAT_COMMAND)
717     props.content_type = amqp_cstring_bytes("text/collectd");
718   else if (conf->format == CAMQP_FORMAT_JSON)
719     props.content_type = amqp_cstring_bytes("application/json");
720   else if (conf->format == CAMQP_FORMAT_GRAPHITE)
721     props.content_type = amqp_cstring_bytes("text/graphite");
722   else
723     assert(23 == 42);
724
725   status = amqp_basic_publish(
726       conf->connection,
727       /* channel = */ 1, amqp_cstring_bytes(CONF(conf, exchange)),
728       amqp_cstring_bytes(routing_key),
729       /* mandatory = */ 0,
730       /* immediate = */ 0, &props, amqp_cstring_bytes(buffer));
731   if (status != 0) {
732     ERROR("amqp plugin: amqp_basic_publish failed with status %i.", status);
733     camqp_close_connection(conf);
734   }
735
736   return status;
737 } /* }}} int camqp_write_locked */
738
739 static int camqp_write(const data_set_t *ds, const value_list_t *vl, /* {{{ */
740                        user_data_t *user_data) {
741   camqp_config_t *conf = user_data->data;
742   char routing_key[6 * DATA_MAX_NAME_LEN];
743   char buffer[8192];
744   int status;
745
746   if ((ds == NULL) || (vl == NULL) || (conf == NULL))
747     return EINVAL;
748
749   if (conf->routing_key != NULL) {
750     sstrncpy(routing_key, conf->routing_key, sizeof(routing_key));
751   } else {
752     ssnprintf(routing_key, sizeof(routing_key), "collectd/%s/%s/%s/%s/%s",
753               vl->host, vl->plugin, vl->plugin_instance, vl->type,
754               vl->type_instance);
755
756     /* Switch slashes (the only character forbidden by collectd) and dots
757      * (the separation character used by AMQP). */
758     for (size_t i = 0; routing_key[i] != 0; i++) {
759       if (routing_key[i] == '.')
760         routing_key[i] = '/';
761       else if (routing_key[i] == '/')
762         routing_key[i] = '.';
763     }
764   }
765
766   if (conf->format == CAMQP_FORMAT_COMMAND) {
767     status = cmd_create_putval(buffer, sizeof(buffer), ds, vl);
768     if (status != 0) {
769       ERROR("amqp plugin: cmd_create_putval failed with status %i.", status);
770       return status;
771     }
772   } else if (conf->format == CAMQP_FORMAT_JSON) {
773     size_t bfree = sizeof(buffer);
774     size_t bfill = 0;
775
776     format_json_initialize(buffer, &bfill, &bfree);
777     format_json_value_list(buffer, &bfill, &bfree, ds, vl, conf->store_rates);
778     format_json_finalize(buffer, &bfill, &bfree);
779   } else if (conf->format == CAMQP_FORMAT_GRAPHITE) {
780     status =
781         format_graphite(buffer, sizeof(buffer), ds, vl, conf->prefix,
782                         conf->postfix, conf->escape_char, conf->graphite_flags);
783     if (status != 0) {
784       ERROR("amqp plugin: format_graphite failed with status %i.", status);
785       return status;
786     }
787   } else {
788     ERROR("amqp plugin: Invalid format (%i).", conf->format);
789     return -1;
790   }
791
792   pthread_mutex_lock(&conf->lock);
793   status = camqp_write_locked(conf, buffer, routing_key);
794   pthread_mutex_unlock(&conf->lock);
795
796   return status;
797 } /* }}} int camqp_write */
798
799 /*
800  * Config handling
801  */
802 static int camqp_config_set_format(oconfig_item_t *ci, /* {{{ */
803                                    camqp_config_t *conf) {
804   char *string;
805   int status;
806
807   string = NULL;
808   status = cf_util_get_string(ci, &string);
809   if (status != 0)
810     return status;
811
812   assert(string != NULL);
813   if (strcasecmp("Command", string) == 0)
814     conf->format = CAMQP_FORMAT_COMMAND;
815   else if (strcasecmp("JSON", string) == 0)
816     conf->format = CAMQP_FORMAT_JSON;
817   else if (strcasecmp("Graphite", string) == 0)
818     conf->format = CAMQP_FORMAT_GRAPHITE;
819   else {
820     WARNING("amqp plugin: Invalid format string: %s", string);
821   }
822
823   free(string);
824
825   return 0;
826 } /* }}} int config_set_string */
827
828 static int camqp_config_connection(oconfig_item_t *ci, /* {{{ */
829                                    bool publish) {
830   camqp_config_t *conf;
831   int status;
832
833   conf = calloc(1, sizeof(*conf));
834   if (conf == NULL) {
835     ERROR("amqp plugin: calloc failed.");
836     return ENOMEM;
837   }
838
839   /* Initialize "conf" {{{ */
840   conf->publish = publish;
841   conf->name = NULL;
842   conf->format = CAMQP_FORMAT_COMMAND;
843   conf->host = NULL;
844   conf->port = 5672;
845   conf->vhost = NULL;
846   conf->user = NULL;
847   conf->password = NULL;
848   conf->exchange = NULL;
849   conf->routing_key = NULL;
850   conf->connection_retry_delay = 0;
851
852   /* publish only */
853   conf->delivery_mode = CAMQP_DM_VOLATILE;
854   conf->store_rates = false;
855   conf->graphite_flags = 0;
856   /* publish & graphite only */
857   conf->prefix = NULL;
858   conf->postfix = NULL;
859   conf->escape_char = '_';
860   /* subscribe only */
861   conf->exchange_type = NULL;
862   conf->queue = NULL;
863   conf->queue_durable = false;
864   conf->queue_auto_delete = true;
865   /* general */
866   conf->connection = NULL;
867   pthread_mutex_init(&conf->lock, /* attr = */ NULL);
868   /* }}} */
869
870   status = cf_util_get_string(ci, &conf->name);
871   if (status != 0) {
872     sfree(conf);
873     return status;
874   }
875
876   for (int i = 0; i < ci->children_num; i++) {
877     oconfig_item_t *child = ci->children + i;
878
879     if (strcasecmp("Host", child->key) == 0)
880       status = cf_util_get_string(child, &conf->host);
881     else if (strcasecmp("Port", child->key) == 0) {
882       status = cf_util_get_port_number(child);
883       if (status > 0) {
884         conf->port = status;
885         status = 0;
886       }
887     } else if (strcasecmp("VHost", child->key) == 0)
888       status = cf_util_get_string(child, &conf->vhost);
889     else if (strcasecmp("User", child->key) == 0)
890       status = cf_util_get_string(child, &conf->user);
891     else if (strcasecmp("Password", child->key) == 0)
892       status = cf_util_get_string(child, &conf->password);
893     else if (strcasecmp("Exchange", child->key) == 0)
894       status = cf_util_get_string(child, &conf->exchange);
895     else if (strcasecmp("ExchangeType", child->key) == 0)
896       status = cf_util_get_string(child, &conf->exchange_type);
897     else if ((strcasecmp("Queue", child->key) == 0) && !publish)
898       status = cf_util_get_string(child, &conf->queue);
899     else if ((strcasecmp("QueueDurable", child->key) == 0) && !publish)
900       status = cf_util_get_boolean(child, &conf->queue_durable);
901     else if ((strcasecmp("QueueAutoDelete", child->key) == 0) && !publish)
902       status = cf_util_get_boolean(child, &conf->queue_auto_delete);
903     else if (strcasecmp("RoutingKey", child->key) == 0)
904       status = cf_util_get_string(child, &conf->routing_key);
905     else if ((strcasecmp("Persistent", child->key) == 0) && publish) {
906       bool tmp = 0;
907       status = cf_util_get_boolean(child, &tmp);
908       if (tmp)
909         conf->delivery_mode = CAMQP_DM_PERSISTENT;
910       else
911         conf->delivery_mode = CAMQP_DM_VOLATILE;
912     } else if ((strcasecmp("StoreRates", child->key) == 0) && publish) {
913       status = cf_util_get_boolean(child, &conf->store_rates);
914       (void)cf_util_get_flag(child, &conf->graphite_flags,
915                              GRAPHITE_STORE_RATES);
916     } else if ((strcasecmp("Format", child->key) == 0) && publish)
917       status = camqp_config_set_format(child, conf);
918     else if ((strcasecmp("GraphiteSeparateInstances", child->key) == 0) &&
919              publish)
920       status = cf_util_get_flag(child, &conf->graphite_flags,
921                                 GRAPHITE_SEPARATE_INSTANCES);
922     else if ((strcasecmp("GraphiteAlwaysAppendDS", child->key) == 0) && publish)
923       status = cf_util_get_flag(child, &conf->graphite_flags,
924                                 GRAPHITE_ALWAYS_APPEND_DS);
925     else if ((strcasecmp("GraphitePreserveSeparator", child->key) == 0) &&
926              publish)
927       status = cf_util_get_flag(child, &conf->graphite_flags,
928                                 GRAPHITE_PRESERVE_SEPARATOR);
929     else if ((strcasecmp("GraphitePrefix", child->key) == 0) && publish)
930       status = cf_util_get_string(child, &conf->prefix);
931     else if ((strcasecmp("GraphitePostfix", child->key) == 0) && publish)
932       status = cf_util_get_string(child, &conf->postfix);
933     else if ((strcasecmp("GraphiteEscapeChar", child->key) == 0) && publish) {
934       char *tmp_buff = NULL;
935       status = cf_util_get_string(child, &tmp_buff);
936       if (strlen(tmp_buff) > 1)
937         WARNING("amqp plugin: The option \"GraphiteEscapeChar\" handles "
938                 "only one character. Others will be ignored.");
939       conf->escape_char = tmp_buff[0];
940       sfree(tmp_buff);
941     } else if (strcasecmp("ConnectionRetryDelay", child->key) == 0)
942       status = cf_util_get_int(child, &conf->connection_retry_delay);
943     else
944       WARNING("amqp plugin: Ignoring unknown "
945               "configuration option \"%s\".",
946               child->key);
947
948     if (status != 0)
949       break;
950   } /* for (i = 0; i < ci->children_num; i++) */
951
952   if ((status == 0) && (conf->exchange == NULL)) {
953     if (conf->exchange_type != NULL)
954       WARNING("amqp plugin: The option \"ExchangeType\" was given "
955               "without the \"Exchange\" option. It will be ignored.");
956
957     if (!publish && (conf->routing_key != NULL))
958       WARNING("amqp plugin: The option \"RoutingKey\" was given "
959               "without the \"Exchange\" option. It will be ignored.");
960   }
961
962   if (status != 0) {
963     camqp_config_free(conf);
964     return status;
965   }
966
967   if (conf->exchange != NULL) {
968     DEBUG("amqp plugin: camqp_config_connection: exchange = %s;",
969           conf->exchange);
970   }
971
972   if (publish) {
973     char cbname[128];
974     ssnprintf(cbname, sizeof(cbname), "amqp/%s", conf->name);
975
976     status = plugin_register_write(cbname, camqp_write,
977                                    &(user_data_t){
978                                        .data = conf,
979                                        .free_func = camqp_config_free,
980                                    });
981     if (status != 0) {
982       camqp_config_free(conf);
983       return status;
984     }
985   } else {
986     status = camqp_subscribe_init(conf);
987     if (status != 0) {
988       camqp_config_free(conf);
989       return status;
990     }
991   }
992
993   return 0;
994 } /* }}} int camqp_config_connection */
995
996 static int camqp_config(oconfig_item_t *ci) /* {{{ */
997 {
998   for (int i = 0; i < ci->children_num; i++) {
999     oconfig_item_t *child = ci->children + i;
1000
1001     if (strcasecmp("Publish", child->key) == 0)
1002       camqp_config_connection(child, /* publish = */ true);
1003     else if (strcasecmp("Subscribe", child->key) == 0)
1004       camqp_config_connection(child, /* publish = */ false);
1005     else
1006       WARNING("amqp plugin: Ignoring unknown config option \"%s\".",
1007               child->key);
1008   } /* for (ci->children_num) */
1009
1010   return 0;
1011 } /* }}} int camqp_config */
1012
1013 void module_register(void) {
1014   plugin_register_complex_config("amqp", camqp_config);
1015   plugin_register_shutdown("amqp", camqp_shutdown);
1016 } /* void module_register */