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 %" PRIsz " 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 =
 986         plugin_register_write(cbname, camqp_write,
 987                               &(user_data_t){
 988                                   .data = conf, .free_func = camqp_config_free,
 989                               });
 990     if (status != 0) {
 991       camqp_config_free(conf);
 992       return status;
 993     }
 994   } else {
 995     status = camqp_subscribe_init(conf);
 996     if (status != 0) {
 997       camqp_config_free(conf);
 998       return status;
 999     }
1000   }
1001
1002   return 0;
1003 } /* }}} int camqp_config_connection */
1004
1005 static int camqp_config(oconfig_item_t *ci) /* {{{ */
1006 {
1007   for (int i = 0; i < ci->children_num; i++) {
1008     oconfig_item_t *child = ci->children + i;
1009
1010     if (strcasecmp("Publish", child->key) == 0)
1011       camqp_config_connection(child, /* publish = */ 1);
1012     else if (strcasecmp("Subscribe", child->key) == 0)
1013       camqp_config_connection(child, /* publish = */ 0);
1014     else
1015       WARNING("amqp plugin: Ignoring unknown config option \"%s\".",
1016               child->key);
1017   } /* for (ci->children_num) */
1018
1019   return 0;
1020 } /* }}} int camqp_config */
1021
1022 void module_register(void) {
1023   plugin_register_complex_config("amqp", camqp_config);
1024   plugin_register_shutdown("amqp", camqp_shutdown);
1025 } /* void module_register */