diff --git a/etc/telegraf.conf b/etc/telegraf.conf index da826a1c70962..878a89164ad2b 100644 --- a/etc/telegraf.conf +++ b/etc/telegraf.conf @@ -167,8 +167,11 @@ # ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as # ## described here: https://www.rabbitmq.com/plugins.html # # auth_method = "PLAIN" +# ## Topic routing key +# # routing_key = "" # ## Telegraf tag to use as a routing key # ## ie, if this tag exists, its value will be used as the routing key +# ## and override routing_key config even if defined # routing_tag = "host" # ## Delivery Mode controls if a published message is persistent # ## Valid options are "transient" and "persistent". default: "transient" @@ -2176,7 +2179,7 @@ # reverse_metric_names = true -# # A plugin to collect stats from Opensmtpd - a validating, recursive, and caching DNS resolver +# # A plugin to collect stats from Opensmtpd - a validating, recursive, and caching DNS resolver # [[inputs.opensmtpd]] # ## If running as a restricted user you can prepend sudo for additional access: # #use_sudo = false @@ -3403,4 +3406,3 @@ # [[inputs.zipkin]] # # path = "/api/v1/spans" # URL path for span data # # port = 9411 # Port on which Telegraf listens - diff --git a/plugins/outputs/amqp/README.md b/plugins/outputs/amqp/README.md index 834074436073b..af85c104c7f7f 100644 --- a/plugins/outputs/amqp/README.md +++ b/plugins/outputs/amqp/README.md @@ -2,12 +2,18 @@ This plugin writes to a AMQP 0-9-1 Exchange, a promenent implementation of this protocol being [RabbitMQ](https://www.rabbitmq.com/). -Metrics are written to a topic exchange using tag, defined in configuration file as RoutingTag, as a routing key. +Metrics are written to a topic exchange using a routing key defined by: +1. The routing_key config defines a static value +2. The routing_tag config defines a metric tag with a dynamic value, overriding the static routing_key if found +3. If neither option is defined, or the tag is not found in a metric, then the empty routing key will be used -If RoutingTag is empty, then empty routing key will be used. -Metrics are grouped in batches by RoutingTag. +Metrics are grouped in batches by the final routing key. -This plugin doesn't bind exchange to a queue, so it should be done by consumer. +This plugin doesn't bind exchange to a queue, so it should be done by consumer. The exchange is always defined as type: topic. +To use it for distributing metrics equally among workers (type: direct), set the routing_key to a static value on the exchange, +declare and bind a single queue with the same routing_key, and consume from the same queue in each worker. +To use it to send metrics to many consumers at once (type: fanout), set the routing_key to "#" on the exchange, then declare, bind, +and consume from individual queues in each worker. For an introduction to AMQP see: - https://www.rabbitmq.com/tutorials/amqp-concepts.html @@ -26,8 +32,11 @@ For an introduction to AMQP see: ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as ## described here: https://www.rabbitmq.com/plugins.html # auth_method = "PLAIN" + ## Topic routing key + # routing_key = "" ## Telegraf tag to use as a routing key ## ie, if this tag exists, its value will be used as the routing key + ## and override routing_key config even if defined routing_tag = "host" ## Delivery Mode controls if a published message is persistent ## Valid options are "transient" and "persistent". default: "transient" diff --git a/plugins/outputs/amqp/amqp.go b/plugins/outputs/amqp/amqp.go index fed1edfe45338..4021ffbc25dd9 100644 --- a/plugins/outputs/amqp/amqp.go +++ b/plugins/outputs/amqp/amqp.go @@ -29,7 +29,9 @@ type AMQP struct { Exchange string // AMQP Auth method AuthMethod string - // Routing Key Tag + // Routing Key (static) + RoutingKey string `toml:"routing_key"` + // Routing Key from Tag RoutingTag string `toml:"routing_tag"` // InfluxDB database Database string @@ -83,8 +85,11 @@ var sampleConfig = ` ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as ## described here: https://www.rabbitmq.com/plugins.html # auth_method = "PLAIN" + ## Topic routing key + # routing_key = "" ## Telegraf tag to use as a routing key ## ie, if this tag exists, its value will be used as the routing key + ## and override routing_key config even if defined routing_tag = "host" ## Delivery Mode controls if a published message is persistent ## Valid options are "transient" and "persistent". default: "transient" @@ -241,6 +246,9 @@ func (q *AMQP) Write(metrics []telegraf.Metric) error { for _, metric := range metrics { var key string + if q.RoutingKey != "" { + key = q.RoutingKey + } if q.RoutingTag != "" { if h, ok := metric.Tags()[q.RoutingTag]; ok { key = h