Merge pull request #2796 from elfiesmelfie/feat_ipmi_SEL_ignore_list
[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 "common.h"
32 #include "plugin.h"
33 #include "utils_cmd_putval.h"
34 #include "utils_format_graphite.h"
35 #include "utils_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       snprintf(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       snprintf(buffer, buffer_size, "Server channel error %d: %s",
227                m->reply_code, tmp);
228       sfree(tmp);
229     } else {
230       snprintf(buffer, buffer_size, "Server error method %#" PRIx32,
231                r.reply.id);
232     }
233     break;
234
235   default:
236     snprintf(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       = */ (conf->queue != NULL)
326                                   ? amqp_cstring_bytes(conf->queue)
327                                   : AMQP_EMPTY_BYTES,
328                               /* passive     = */ 0,
329                               /* durable     = */ conf->queue_durable,
330                               /* exclusive   = */ 0,
331                               /* auto_delete = */ conf->queue_auto_delete,
332                               /* arguments   = */ AMQP_EMPTY_TABLE);
333   if (qd_ret == NULL) {
334     ERROR("amqp plugin: amqp_queue_declare failed.");
335     camqp_close_connection(conf);
336     return -1;
337   }
338
339   if (conf->queue == NULL) {
340     conf->queue = camqp_bytes_cstring(&qd_ret->queue);
341     if (conf->queue == NULL) {
342       ERROR("amqp plugin: camqp_bytes_cstring failed.");
343       camqp_close_connection(conf);
344       return -1;
345     }
346
347     INFO("amqp plugin: Created queue \"%s\".", conf->queue);
348   }
349   DEBUG("amqp plugin: Successfully created queue \"%s\".", conf->queue);
350
351   /* bind to an exchange */
352   if (conf->exchange != NULL) {
353     amqp_queue_bind_ok_t *qb_ret;
354
355     assert(conf->queue != NULL);
356     qb_ret =
357         amqp_queue_bind(conf->connection,
358                         /* channel     = */ CAMQP_CHANNEL,
359                         /* queue       = */ amqp_cstring_bytes(conf->queue),
360                         /* exchange    = */ amqp_cstring_bytes(conf->exchange),
361                         /* routing_key = */ (conf->routing_key != NULL)
362                             ? amqp_cstring_bytes(conf->routing_key)
363                             : AMQP_EMPTY_BYTES,
364                         /* arguments   = */ AMQP_EMPTY_TABLE);
365     if ((qb_ret == NULL) && camqp_is_error(conf)) {
366       char errbuf[1024];
367       ERROR("amqp plugin: amqp_queue_bind failed: %s",
368             camqp_strerror(conf, errbuf, sizeof(errbuf)));
369       camqp_close_connection(conf);
370       return -1;
371     }
372
373     DEBUG("amqp plugin: Successfully bound queue \"%s\" to exchange \"%s\".",
374           conf->queue, conf->exchange);
375   } /* if (conf->exchange != NULL) */
376
377   cm_ret =
378       amqp_basic_consume(conf->connection,
379                          /* channel      = */ CAMQP_CHANNEL,
380                          /* queue        = */ amqp_cstring_bytes(conf->queue),
381                          /* consumer_tag = */ AMQP_EMPTY_BYTES,
382                          /* no_local     = */ 0,
383                          /* no_ack       = */ 1,
384                          /* exclusive    = */ 0,
385                          /* arguments    = */ AMQP_EMPTY_TABLE);
386   if ((cm_ret == NULL) && camqp_is_error(conf)) {
387     char errbuf[1024];
388     ERROR("amqp plugin: amqp_basic_consume failed: %s",
389           camqp_strerror(conf, errbuf, sizeof(errbuf)));
390     camqp_close_connection(conf);
391     return -1;
392   }
393
394   return 0;
395 } /* }}} int camqp_setup_queue */
396
397 static int camqp_connect(camqp_config_t *conf) /* {{{ */
398 {
399   static time_t last_connect_time;
400
401   amqp_rpc_reply_t reply;
402   int status;
403 #ifdef HAVE_AMQP_TCP_SOCKET
404   amqp_socket_t *socket;
405 #else
406   int sockfd;
407 #endif
408
409   if (conf->connection != NULL)
410     return 0;
411
412   time_t now = time(NULL);
413   if (now < (last_connect_time + conf->connection_retry_delay)) {
414     DEBUG("amqp plugin: skipping connection retry, "
415           "ConnectionRetryDelay: %d",
416           conf->connection_retry_delay);
417     return 1;
418   } else {
419     DEBUG("amqp plugin: retrying connection");
420     last_connect_time = now;
421   }
422
423   conf->connection = amqp_new_connection();
424   if (conf->connection == NULL) {
425     ERROR("amqp plugin: amqp_new_connection failed.");
426     return ENOMEM;
427   }
428
429 #ifdef HAVE_AMQP_TCP_SOCKET
430 #define CLOSE_SOCKET() /* amqp_destroy_connection() closes the socket for us   \
431                           */
432   /* TODO: add support for SSL using amqp_ssl_socket_new
433    *       and related functions */
434   socket = amqp_tcp_socket_new(conf->connection);
435   if (!socket) {
436     ERROR("amqp plugin: amqp_tcp_socket_new failed.");
437     amqp_destroy_connection(conf->connection);
438     conf->connection = NULL;
439     return ENOMEM;
440   }
441
442   status = amqp_socket_open(socket, CONF(conf, host), conf->port);
443   if (status < 0) {
444     status *= -1;
445     ERROR("amqp plugin: amqp_socket_open failed: %s", STRERROR(status));
446     amqp_destroy_connection(conf->connection);
447     conf->connection = NULL;
448     return status;
449   }
450 #else /* HAVE_AMQP_TCP_SOCKET */
451 #define CLOSE_SOCKET() close(sockfd)
452   /* this interface is deprecated as of rabbitmq-c 0.4 */
453   sockfd = amqp_open_socket(CONF(conf, host), conf->port);
454   if (sockfd < 0) {
455     status = (-1) * sockfd;
456     ERROR("amqp plugin: amqp_open_socket failed: %s", STRERROR(status));
457     amqp_destroy_connection(conf->connection);
458     conf->connection = NULL;
459     return status;
460   }
461   amqp_set_sockfd(conf->connection, sockfd);
462 #endif
463
464   reply = amqp_login(conf->connection, CONF(conf, vhost),
465                      /* channel max = */ 0,
466                      /* frame max   = */ 131072,
467                      /* heartbeat   = */ 0,
468                      /* authentication = */ AMQP_SASL_METHOD_PLAIN,
469                      CONF(conf, user), CONF(conf, password));
470   if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
471     ERROR("amqp plugin: amqp_login (vhost = %s, user = %s) failed.",
472           CONF(conf, vhost), CONF(conf, user));
473     amqp_destroy_connection(conf->connection);
474     CLOSE_SOCKET();
475     conf->connection = NULL;
476     return 1;
477   }
478
479   amqp_channel_open(conf->connection, /* channel = */ 1);
480   /* FIXME: Is checking "reply.reply_type" really correct here? How does
481    * it get set? --octo */
482   if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
483     ERROR("amqp plugin: amqp_channel_open failed.");
484     amqp_connection_close(conf->connection, AMQP_REPLY_SUCCESS);
485     amqp_destroy_connection(conf->connection);
486     CLOSE_SOCKET();
487     conf->connection = NULL;
488     return 1;
489   }
490
491   INFO("amqp plugin: Successfully opened connection to vhost \"%s\" "
492        "on %s:%i.",
493        CONF(conf, vhost), CONF(conf, host), conf->port);
494
495   status = camqp_create_exchange(conf);
496   if (status != 0)
497     return status;
498
499   if (!conf->publish)
500     return camqp_setup_queue(conf);
501   return 0;
502 } /* }}} int camqp_connect */
503
504 static int camqp_shutdown(void) /* {{{ */
505 {
506   DEBUG("amqp plugin: Shutting down %" PRIsz " subscriber threads.",
507         subscriber_threads_num);
508
509   subscriber_threads_running = 0;
510   for (size_t i = 0; i < subscriber_threads_num; i++) {
511     /* FIXME: Sending a signal is not very elegant here. Maybe find out how
512      * to use a timeout in the thread and check for the variable in regular
513      * intervals. */
514     pthread_kill(subscriber_threads[i], SIGTERM);
515     pthread_join(subscriber_threads[i], /* retval = */ NULL);
516   }
517
518   subscriber_threads_num = 0;
519   sfree(subscriber_threads);
520
521   DEBUG("amqp plugin: All subscriber threads exited.");
522
523   return 0;
524 } /* }}} int camqp_shutdown */
525
526 /*
527  * Subscribing code
528  */
529 static int camqp_read_body(camqp_config_t *conf, /* {{{ */
530                            size_t body_size, const char *content_type) {
531   char body[body_size + 1];
532   char *body_ptr;
533   size_t received;
534   amqp_frame_t frame;
535   int status;
536
537   memset(body, 0, sizeof(body));
538   body_ptr = &body[0];
539   received = 0;
540
541   while (received < body_size) {
542     status = amqp_simple_wait_frame(conf->connection, &frame);
543     if (status < 0) {
544       status = (-1) * status;
545       ERROR("amqp plugin: amqp_simple_wait_frame failed: %s", STRERROR(status));
546       camqp_close_connection(conf);
547       return status;
548     }
549
550     if (frame.frame_type != AMQP_FRAME_BODY) {
551       NOTICE("amqp plugin: Unexpected frame type: %#" PRIx8, frame.frame_type);
552       return -1;
553     }
554
555     if ((body_size - received) < frame.payload.body_fragment.len) {
556       WARNING("amqp plugin: Body is larger than indicated by header.");
557       return -1;
558     }
559
560     memcpy(body_ptr, frame.payload.body_fragment.bytes,
561            frame.payload.body_fragment.len);
562     body_ptr += frame.payload.body_fragment.len;
563     received += frame.payload.body_fragment.len;
564   } /* while (received < body_size) */
565
566   if (strcasecmp("text/collectd", content_type) == 0) {
567     status = cmd_handle_putval(stderr, body);
568     if (status != 0)
569       ERROR("amqp plugin: cmd_handle_putval failed with status %i.", status);
570     return status;
571   } else if (strcasecmp("application/json", content_type) == 0) {
572     ERROR("amqp plugin: camqp_read_body: Parsing JSON data has not "
573           "been implemented yet. FIXME!");
574     return 0;
575   } else {
576     ERROR("amqp plugin: camqp_read_body: Unknown content type \"%s\".",
577           content_type);
578     return EINVAL;
579   }
580
581   /* not reached */
582   return 0;
583 } /* }}} int camqp_read_body */
584
585 static int camqp_read_header(camqp_config_t *conf) /* {{{ */
586 {
587   int status;
588   amqp_frame_t frame;
589   amqp_basic_properties_t *properties;
590   char *content_type;
591
592   status = amqp_simple_wait_frame(conf->connection, &frame);
593   if (status < 0) {
594     status = (-1) * status;
595     ERROR("amqp plugin: amqp_simple_wait_frame failed: %s", STRERROR(status));
596     camqp_close_connection(conf);
597     return status;
598   }
599
600   if (frame.frame_type != AMQP_FRAME_HEADER) {
601     NOTICE("amqp plugin: Unexpected frame type: %#" PRIx8, frame.frame_type);
602     return -1;
603   }
604
605   properties = frame.payload.properties.decoded;
606   content_type = camqp_bytes_cstring(&properties->content_type);
607   if (content_type == NULL) {
608     ERROR("amqp plugin: Unable to determine content type.");
609     return -1;
610   }
611
612   status = camqp_read_body(conf, (size_t)frame.payload.properties.body_size,
613                            content_type);
614
615   sfree(content_type);
616   return status;
617 } /* }}} int camqp_read_header */
618
619 static void *camqp_subscribe_thread(void *user_data) /* {{{ */
620 {
621   camqp_config_t *conf = user_data;
622   int status;
623
624   cdtime_t interval = plugin_get_interval();
625
626   while (subscriber_threads_running) {
627     amqp_frame_t frame;
628
629     status = camqp_connect(conf);
630     if (status != 0) {
631       ERROR("amqp plugin: camqp_connect failed. "
632             "Will sleep for %.3f seconds.",
633             CDTIME_T_TO_DOUBLE(interval));
634       nanosleep(&CDTIME_T_TO_TIMESPEC(interval), /* remaining = */ NULL);
635       continue;
636     }
637
638     status = amqp_simple_wait_frame(conf->connection, &frame);
639     if (status < 0) {
640       ERROR("amqp plugin: amqp_simple_wait_frame failed. "
641             "Will sleep for %.3f seconds.",
642             CDTIME_T_TO_DOUBLE(interval));
643       camqp_close_connection(conf);
644       nanosleep(&CDTIME_T_TO_TIMESPEC(interval), /* remaining = */ NULL);
645       continue;
646     }
647
648     if (frame.frame_type != AMQP_FRAME_METHOD) {
649       DEBUG("amqp plugin: Unexpected frame type: %#" PRIx8, frame.frame_type);
650       continue;
651     }
652
653     if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) {
654       DEBUG("amqp plugin: Unexpected method id: %#" PRIx32,
655             frame.payload.method.id);
656       continue;
657     }
658
659     camqp_read_header(conf);
660
661     amqp_maybe_release_buffers(conf->connection);
662   } /* while (subscriber_threads_running) */
663
664   camqp_config_free(conf);
665   pthread_exit(NULL);
666   return NULL;
667 } /* }}} void *camqp_subscribe_thread */
668
669 static int camqp_subscribe_init(camqp_config_t *conf) /* {{{ */
670 {
671   int status;
672   pthread_t *tmp;
673
674   tmp = realloc(subscriber_threads,
675                 sizeof(*subscriber_threads) * (subscriber_threads_num + 1));
676   if (tmp == NULL) {
677     ERROR("amqp plugin: realloc failed.");
678     sfree(subscriber_threads);
679     return ENOMEM;
680   }
681   subscriber_threads = tmp;
682   tmp = subscriber_threads + subscriber_threads_num;
683   memset(tmp, 0, sizeof(*tmp));
684
685   status = plugin_thread_create(tmp, /* attr = */ NULL, camqp_subscribe_thread,
686                                 conf, "amqp subscribe");
687   if (status != 0) {
688     ERROR("amqp plugin: pthread_create failed: %s", STRERROR(status));
689     return status;
690   }
691
692   subscriber_threads_num++;
693
694   return 0;
695 } /* }}} int camqp_subscribe_init */
696
697 /*
698  * Publishing code
699  */
700 /* XXX: You must hold "conf->lock" when calling this function! */
701 static int camqp_write_locked(camqp_config_t *conf, /* {{{ */
702                               const char *buffer, const char *routing_key) {
703   int status;
704
705   status = camqp_connect(conf);
706   if (status != 0)
707     return status;
708
709   amqp_basic_properties_t props = {._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |
710                                              AMQP_BASIC_DELIVERY_MODE_FLAG |
711                                              AMQP_BASIC_APP_ID_FLAG,
712                                    .delivery_mode = conf->delivery_mode,
713                                    .app_id = amqp_cstring_bytes("collectd")};
714
715   if (conf->format == CAMQP_FORMAT_COMMAND)
716     props.content_type = amqp_cstring_bytes("text/collectd");
717   else if (conf->format == CAMQP_FORMAT_JSON)
718     props.content_type = amqp_cstring_bytes("application/json");
719   else if (conf->format == CAMQP_FORMAT_GRAPHITE)
720     props.content_type = amqp_cstring_bytes("text/graphite");
721   else
722     assert(23 == 42);
723
724   status = amqp_basic_publish(
725       conf->connection,
726       /* channel = */ 1, amqp_cstring_bytes(CONF(conf, exchange)),
727       amqp_cstring_bytes(routing_key),
728       /* mandatory = */ 0,
729       /* immediate = */ 0, &props, amqp_cstring_bytes(buffer));
730   if (status != 0) {
731     ERROR("amqp plugin: amqp_basic_publish failed with status %i.", status);
732     camqp_close_connection(conf);
733   }
734
735   return status;
736 } /* }}} int camqp_write_locked */
737
738 static int camqp_write(const data_set_t *ds, const value_list_t *vl, /* {{{ */
739                        user_data_t *user_data) {
740   camqp_config_t *conf = user_data->data;
741   char routing_key[6 * DATA_MAX_NAME_LEN];
742   char buffer[8192];
743   int status;
744
745   if ((ds == NULL) || (vl == NULL) || (conf == NULL))
746     return EINVAL;
747
748   if (conf->routing_key != NULL) {
749     sstrncpy(routing_key, conf->routing_key, sizeof(routing_key));
750   } else {
751     snprintf(routing_key, sizeof(routing_key), "collectd/%s/%s/%s/%s/%s",
752              vl->host, vl->plugin, vl->plugin_instance, vl->type,
753              vl->type_instance);
754
755     /* Switch slashes (the only character forbidden by collectd) and dots
756      * (the separation character used by AMQP). */
757     for (size_t i = 0; routing_key[i] != 0; i++) {
758       if (routing_key[i] == '.')
759         routing_key[i] = '/';
760       else if (routing_key[i] == '/')
761         routing_key[i] = '.';
762     }
763   }
764
765   if (conf->format == CAMQP_FORMAT_COMMAND) {
766     status = cmd_create_putval(buffer, sizeof(buffer), ds, vl);
767     if (status != 0) {
768       ERROR("amqp plugin: cmd_create_putval failed with status %i.", status);
769       return status;
770     }
771   } else if (conf->format == CAMQP_FORMAT_JSON) {
772     size_t bfree = sizeof(buffer);
773     size_t bfill = 0;
774
775     format_json_initialize(buffer, &bfill, &bfree);
776     format_json_value_list(buffer, &bfill, &bfree, ds, vl, conf->store_rates);
777     format_json_finalize(buffer, &bfill, &bfree);
778   } else if (conf->format == CAMQP_FORMAT_GRAPHITE) {
779     status =
780         format_graphite(buffer, sizeof(buffer), ds, vl, conf->prefix,
781                         conf->postfix, conf->escape_char, conf->graphite_flags);
782     if (status != 0) {
783       ERROR("amqp plugin: format_graphite failed with status %i.", status);
784       return status;
785     }
786   } else {
787     ERROR("amqp plugin: Invalid format (%i).", conf->format);
788     return -1;
789   }
790
791   pthread_mutex_lock(&conf->lock);
792   status = camqp_write_locked(conf, buffer, routing_key);
793   pthread_mutex_unlock(&conf->lock);
794
795   return status;
796 } /* }}} int camqp_write */
797
798 /*
799  * Config handling
800  */
801 static int camqp_config_set_format(oconfig_item_t *ci, /* {{{ */
802                                    camqp_config_t *conf) {
803   char *string;
804   int status;
805
806   string = NULL;
807   status = cf_util_get_string(ci, &string);
808   if (status != 0)
809     return status;
810
811   assert(string != NULL);
812   if (strcasecmp("Command", string) == 0)
813     conf->format = CAMQP_FORMAT_COMMAND;
814   else if (strcasecmp("JSON", string) == 0)
815     conf->format = CAMQP_FORMAT_JSON;
816   else if (strcasecmp("Graphite", string) == 0)
817     conf->format = CAMQP_FORMAT_GRAPHITE;
818   else {
819     WARNING("amqp plugin: Invalid format string: %s", string);
820   }
821
822   free(string);
823
824   return 0;
825 } /* }}} int config_set_string */
826
827 static int camqp_config_connection(oconfig_item_t *ci, /* {{{ */
828                                    bool publish) {
829   camqp_config_t *conf;
830   int status;
831
832   conf = calloc(1, sizeof(*conf));
833   if (conf == NULL) {
834     ERROR("amqp plugin: calloc failed.");
835     return ENOMEM;
836   }
837
838   /* Initialize "conf" {{{ */
839   conf->publish = publish;
840   conf->name = NULL;
841   conf->format = CAMQP_FORMAT_COMMAND;
842   conf->host = NULL;
843   conf->port = 5672;
844   conf->vhost = NULL;
845   conf->user = NULL;
846   conf->password = NULL;
847   conf->exchange = NULL;
848   conf->routing_key = NULL;
849   conf->connection_retry_delay = 0;
850
851   /* publish only */
852   conf->delivery_mode = CAMQP_DM_VOLATILE;
853   conf->store_rates = false;
854   conf->graphite_flags = 0;
855   /* publish & graphite only */
856   conf->prefix = NULL;
857   conf->postfix = NULL;
858   conf->escape_char = '_';
859   /* subscribe only */
860   conf->exchange_type = NULL;
861   conf->queue = NULL;
862   conf->queue_durable = false;
863   conf->queue_auto_delete = true;
864   /* general */
865   conf->connection = NULL;
866   pthread_mutex_init(&conf->lock, /* attr = */ NULL);
867   /* }}} */
868
869   status = cf_util_get_string(ci, &conf->name);
870   if (status != 0) {
871     sfree(conf);
872     return status;
873   }
874
875   for (int i = 0; i < ci->children_num; i++) {
876     oconfig_item_t *child = ci->children + i;
877
878     if (strcasecmp("Host", child->key) == 0)
879       status = cf_util_get_string(child, &conf->host);
880     else if (strcasecmp("Port", child->key) == 0) {
881       status = cf_util_get_port_number(child);
882       if (status > 0) {
883         conf->port = status;
884         status = 0;
885       }
886     } else if (strcasecmp("VHost", child->key) == 0)
887       status = cf_util_get_string(child, &conf->vhost);
888     else if (strcasecmp("User", child->key) == 0)
889       status = cf_util_get_string(child, &conf->user);
890     else if (strcasecmp("Password", child->key) == 0)
891       status = cf_util_get_string(child, &conf->password);
892     else if (strcasecmp("Exchange", child->key) == 0)
893       status = cf_util_get_string(child, &conf->exchange);
894     else if (strcasecmp("ExchangeType", child->key) == 0)
895       status = cf_util_get_string(child, &conf->exchange_type);
896     else if ((strcasecmp("Queue", child->key) == 0) && !publish)
897       status = cf_util_get_string(child, &conf->queue);
898     else if ((strcasecmp("QueueDurable", child->key) == 0) && !publish)
899       status = cf_util_get_boolean(child, &conf->queue_durable);
900     else if ((strcasecmp("QueueAutoDelete", child->key) == 0) && !publish)
901       status = cf_util_get_boolean(child, &conf->queue_auto_delete);
902     else if (strcasecmp("RoutingKey", child->key) == 0)
903       status = cf_util_get_string(child, &conf->routing_key);
904     else if ((strcasecmp("Persistent", child->key) == 0) && publish) {
905       bool tmp = 0;
906       status = cf_util_get_boolean(child, &tmp);
907       if (tmp)
908         conf->delivery_mode = CAMQP_DM_PERSISTENT;
909       else
910         conf->delivery_mode = CAMQP_DM_VOLATILE;
911     } else if ((strcasecmp("StoreRates", child->key) == 0) && publish) {
912       status = cf_util_get_boolean(child, &conf->store_rates);
913       (void)cf_util_get_flag(child, &conf->graphite_flags,
914                              GRAPHITE_STORE_RATES);
915     } else if ((strcasecmp("Format", child->key) == 0) && publish)
916       status = camqp_config_set_format(child, conf);
917     else if ((strcasecmp("GraphiteSeparateInstances", child->key) == 0) &&
918              publish)
919       status = cf_util_get_flag(child, &conf->graphite_flags,
920                                 GRAPHITE_SEPARATE_INSTANCES);
921     else if ((strcasecmp("GraphiteAlwaysAppendDS", child->key) == 0) && publish)
922       status = cf_util_get_flag(child, &conf->graphite_flags,
923                                 GRAPHITE_ALWAYS_APPEND_DS);
924     else if ((strcasecmp("GraphitePreserveSeparator", child->key) == 0) &&
925              publish)
926       status = cf_util_get_flag(child, &conf->graphite_flags,
927                                 GRAPHITE_PRESERVE_SEPARATOR);
928     else if ((strcasecmp("GraphitePrefix", child->key) == 0) && publish)
929       status = cf_util_get_string(child, &conf->prefix);
930     else if ((strcasecmp("GraphitePostfix", child->key) == 0) && publish)
931       status = cf_util_get_string(child, &conf->postfix);
932     else if ((strcasecmp("GraphiteEscapeChar", child->key) == 0) && publish) {
933       char *tmp_buff = NULL;
934       status = cf_util_get_string(child, &tmp_buff);
935       if (strlen(tmp_buff) > 1)
936         WARNING("amqp plugin: The option \"GraphiteEscapeChar\" handles "
937                 "only one character. Others will be ignored.");
938       conf->escape_char = tmp_buff[0];
939       sfree(tmp_buff);
940     } else if (strcasecmp("ConnectionRetryDelay", child->key) == 0)
941       status = cf_util_get_int(child, &conf->connection_retry_delay);
942     else
943       WARNING("amqp plugin: Ignoring unknown "
944               "configuration option \"%s\".",
945               child->key);
946
947     if (status != 0)
948       break;
949   } /* for (i = 0; i < ci->children_num; i++) */
950
951   if ((status == 0) && (conf->exchange == NULL)) {
952     if (conf->exchange_type != NULL)
953       WARNING("amqp plugin: The option \"ExchangeType\" was given "
954               "without the \"Exchange\" option. It will be ignored.");
955
956     if (!publish && (conf->routing_key != NULL))
957       WARNING("amqp plugin: The option \"RoutingKey\" was given "
958               "without the \"Exchange\" option. It will be ignored.");
959   }
960
961   if (status != 0) {
962     camqp_config_free(conf);
963     return status;
964   }
965
966   if (conf->exchange != NULL) {
967     DEBUG("amqp plugin: camqp_config_connection: exchange = %s;",
968           conf->exchange);
969   }
970
971   if (publish) {
972     char cbname[128];
973     snprintf(cbname, sizeof(cbname), "amqp/%s", conf->name);
974
975     status =
976         plugin_register_write(cbname, camqp_write,
977                               &(user_data_t){
978                                   .data = conf, .free_func = camqp_config_free,
979                               });
980     if (status != 0) {
981       camqp_config_free(conf);
982       return status;
983     }
984   } else {
985     status = camqp_subscribe_init(conf);
986     if (status != 0) {
987       camqp_config_free(conf);
988       return status;
989     }
990   }
991
992   return 0;
993 } /* }}} int camqp_config_connection */
994
995 static int camqp_config(oconfig_item_t *ci) /* {{{ */
996 {
997   for (int i = 0; i < ci->children_num; i++) {
998     oconfig_item_t *child = ci->children + i;
999
1000     if (strcasecmp("Publish", child->key) == 0)
1001       camqp_config_connection(child, /* publish = */ true);
1002     else if (strcasecmp("Subscribe", child->key) == 0)
1003       camqp_config_connection(child, /* publish = */ false);
1004     else
1005       WARNING("amqp plugin: Ignoring unknown config option \"%s\".",
1006               child->key);
1007   } /* for (ci->children_num) */
1008
1009   return 0;
1010 } /* }}} int camqp_config */
1011
1012 void module_register(void) {
1013   plugin_register_complex_config("amqp", camqp_config);
1014   plugin_register_shutdown("amqp", camqp_shutdown);
1015 } /* void module_register */