Add queue_durability parameter to amqp_consumer input (influxdata#4628)

plugins/inputs/amqp_consumer/README.md

@@ -46,6 +46,10 @@ The following defaults are known to work with RabbitMQ:
 
  ## AMQP queue name
  queue = "telegraf"
+
+ ## AMQP queue durability can be "transient" or "durable".
+ queue_durability = "durable"
+
  ## Binding Key
  binding_key = "#"
 

plugins/inputs/amqp_consumer/amqp_consumer.go

@@ -30,7 +30,9 @@ type AMQPConsumer struct {
  ExchangeArguments map[string]string `toml:"exchange_arguments"`
 
  // Queue Name
- Queue string
+ Queue string `toml:"queue"`
+ QueueDurability string `toml:"queue_durability"`
+
  // Binding Key
  BindingKey string `toml:"binding_key"`
 
@@ -64,6 +66,8 @@ const (
  DefaultExchangeType = "topic"
  DefaultExchangeDurability = "durable"
 
+ DefaultQueueDurability = "durable"
+
  DefaultPrefetchCount = 50
 )
 
@@ -98,10 +102,13 @@ func (a *AMQPConsumer) SampleConfig() string {
  # exchange_arguments = { }
  # exchange_arguments = {"hash_propery" = "timestamp"}
 
- ## AMQP queue name
+ ## AMQP queue name.
  queue = "telegraf"
 
- ## Binding Key
+ ## AMQP queue durability can be "transient" or "durable".
+ queue_durability = "durable"
+
+ ## Binding Key.
  binding_key = "#"
 
  ## Maximum number of messages server should give to the worker.
@@ -260,13 +267,21 @@ func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, err
  return nil, err
  }
 
+ var queueDurable = true
+ switch a.QueueDurability {
+ case "transient":
+ queueDurable = false
+ default:
+ queueDurable = true
+ }
+
  q, err := ch.QueueDeclare(
- a.Queue, // queue
- true,  // durable
- false, // delete when unused
- false, // exclusive
- false, // no-wait
- nil, // arguments
+ a.Queue,  // queue
+ queueDurable, // durable
+ false,  // delete when unused
+ false,  // exclusive
+ false,  // no-wait
+ nil,  // arguments
  )
  if err != nil {
  return nil, fmt.Errorf("Failed to declare a queue: %s", err)
@@ -380,6 +395,7 @@ func init() {
  AuthMethod: DefaultAuthMethod,
  ExchangeType: DefaultExchangeType,
  ExchangeDurability: DefaultExchangeDurability,
+ QueueDurability: DefaultQueueDurability,
  PrefetchCount: DefaultPrefetchCount,
  }
  })