src/amqp.c

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