feat(inputs.amqp_consumer): Add support to rabbitmq stream queue (#13496)

plugins/inputs/amqp_consumer/README.md

@@ -73,6 +73,10 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## If true, queue will be passively declared.
   # queue_passive = false
 
+  ## Additional arguments when consuming from Queue
+  # queue_consume_arguments = { }
+  # queue_consume_arguments = {"x-stream-offset" = "first"}
+
   ## A binding between the exchange and queue using this binding key is
   ## created.  If unset, no binding is created.
   binding_key = "#"

plugins/inputs/amqp_consumer/amqp_consumer.go

@@ -43,6 +43,7 @@ type AMQPConsumer struct {
 	Queue                 string            `toml:"queue"`
 	QueueDurability       string            `toml:"queue_durability"`
 	QueuePassive          bool              `toml:"queue_passive"`
+	QueueConsumeArguments map[string]string `toml:"queue_consume_arguments"`
 
 	// Binding Key
 	BindingKey string `toml:"binding_key"`
@@ -284,6 +285,11 @@ func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, err
 		return nil, fmt.Errorf("failed to set QoS: %w", err)
 	}
 
+	consumeArgs := make(amqp.Table, len(a.QueueConsumeArguments))
+	for k, v := range a.QueueConsumeArguments {
+		consumeArgs[k] = v
+	}
+
 	msgs, err := ch.Consume(
 		q.Name,      // queue
 		"",          // consumer
@@ -291,7 +297,7 @@ func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, err
 		false,       // exclusive
 		false,       // no-local
 		false,       // no-wait
-		nil,    // arguments
+		consumeArgs, // arguments
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed establishing connection to queue: %w", err)

plugins/inputs/amqp_consumer/sample.conf

@@ -34,6 +34,10 @@
   ## If true, queue will be passively declared.
   # queue_passive = false
 
+  ## Additional arguments when consuming from Queue
+  # queue_consume_arguments = { }
+  # queue_consume_arguments = {"x-stream-offset" = "first"}
+
   ## A binding between the exchange and queue using this binding key is
   ## created.  If unset, no binding is created.
   binding_key = "#"