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 */