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