Support passive queue declaration in amqp_consumer (#5831)

This commit is contained in:
Daniel Nelson 2019-05-20 14:38:35 -07:00 committed by GitHub
parent b5cd9a9ff2
commit e141518cf0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 100 additions and 52 deletions

View File

@ -27,7 +27,7 @@ The following defaults are known to work with RabbitMQ:
# username = ""
# password = ""
## Exchange to declare and consume from.
## Name of the exchange to declare. If unset, no exchange will be declared.
exchange = "telegraf"
## Exchange type; common types are "direct", "fanout", "topic", "header", "x-consistent-hash".
@ -49,7 +49,11 @@ The following defaults are known to work with RabbitMQ:
## AMQP queue durability can be "transient" or "durable".
queue_durability = "durable"
## Binding Key
## If true, queue will be passively declared.
# queue_passive = false
## A binding between the exchange and queue using this binding key is
## created. If unset, no binding is created.
binding_key = "#"
## Maximum number of messages server should give to the worker.

View File

@ -41,6 +41,7 @@ type AMQPConsumer struct {
// Queue Name
Queue string `toml:"queue"`
QueueDurability string `toml:"queue_durability"`
QueuePassive bool `toml:"queue_passive"`
// Binding Key
BindingKey string `toml:"binding_key"`
@ -101,7 +102,7 @@ func (a *AMQPConsumer) SampleConfig() string {
# username = ""
# password = ""
## Exchange to declare and consume from.
## Name of the exchange to declare. If unset, no exchange will be declared.
exchange = "telegraf"
## Exchange type; common types are "direct", "fanout", "topic", "header", "x-consistent-hash".
@ -123,7 +124,11 @@ func (a *AMQPConsumer) SampleConfig() string {
## AMQP queue durability can be "transient" or "durable".
queue_durability = "durable"
## Binding Key.
## If true, queue will be passively declared.
# queue_passive = false
## A binding between the exchange and queue using this binding key is
## created. If unset, no binding is created.
binding_key = "#"
## Maximum number of messages server should give to the worker.
@ -286,6 +291,7 @@ func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, err
return nil, fmt.Errorf("Failed to open a channel: %s", err)
}
if a.Exchange != "" {
var exchangeDurable = true
switch a.ExchangeDurability {
case "transient":
@ -309,27 +315,18 @@ func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, err
if err != nil {
return nil, err
}
var queueDurable = true
switch a.QueueDurability {
case "transient":
queueDurable = false
default:
queueDurable = true
}
q, err := ch.QueueDeclare(
a.Queue, // queue
queueDurable, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
q, err := declareQueue(
ch,
a.Queue,
a.QueueDurability,
a.QueuePassive)
if err != nil {
return nil, fmt.Errorf("Failed to declare a queue: %s", err)
return nil, err
}
if a.BindingKey != "" {
err = ch.QueueBind(
q.Name, // queue
a.BindingKey, // binding-key
@ -340,6 +337,7 @@ func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, err
if err != nil {
return nil, fmt.Errorf("Failed to bind a queue: %s", err)
}
}
err = ch.Qos(
a.PrefetchCount,
@ -402,6 +400,48 @@ func declareExchange(
return nil
}
func declareQueue(
channel *amqp.Channel,
queueName string,
queueDurability string,
queuePassive bool,
) (*amqp.Queue, error) {
var queue amqp.Queue
var err error
var queueDurable = true
switch queueDurability {
case "transient":
queueDurable = false
default:
queueDurable = true
}
if queuePassive {
queue, err = channel.QueueDeclarePassive(
queueName, // queue
queueDurable, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
} else {
queue, err = channel.QueueDeclare(
queueName, // queue
queueDurable, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
}
if err != nil {
return nil, fmt.Errorf("error declaring queue: %v", err)
}
return &queue, nil
}
// Read messages from queue and add them to the Accumulator
func (a *AMQPConsumer) process(ctx context.Context, msgs <-chan amqp.Delivery, ac telegraf.Accumulator) {
a.deliveries = make(map[telegraf.TrackingID]amqp.Delivery)

View File

@ -33,7 +33,7 @@ For an introduction to AMQP see:
# exchange_type = "topic"
## If true, exchange will be passively declared.
# exchange_declare_passive = false
# exchange_passive = false
## Exchange durability can be either "transient" or "durable".
# exchange_durability = "durable"

View File

@ -92,7 +92,7 @@ var sampleConfig = `
# exchange_type = "topic"
## If true, exchange will be passively declared.
# exchange_declare_passive = false
# exchange_passive = false
## Exchange durability can be either "transient" or "durable".
# exchange_durability = "durable"

View File

@ -78,6 +78,10 @@ func Connect(config *ClientConfig) (*client, error) {
}
func (c *client) DeclareExchange() error {
if c.config.exchange == "" {
return nil
}
var err error
if c.config.exchangePassive {
err = c.channel.ExchangeDeclarePassive(