amqp_consumer

README

AMQP Consumer Input Plugin

This plugin provides a consumer for use with AMQP 0-9-1, a prominent implementation of this protocol being RabbitMQ.

Metrics are read from a topic exchange using the configured queue and binding_key.

Message payload should be formatted in one of the Telegraf Data Formats.

For an introduction to AMQP see:

• amqp - concepts
• rabbitmq: getting started

Service Input

This plugin is a service input. Normal plugins gather metrics determined by the interval setting. Service plugins start a service to listens and waits for metrics or events to occur. Service plugins have two key differences from normal plugins:

1. The global or plugin specific interval setting may not apply
2. The CLI options of --test, --test-wait, and --once may not produce output for this plugin

Global configuration options

In addition to the plugin-specific configuration settings, plugins support additional global and plugin configuration settings. These settings are used to modify metrics, tags, and field or create aliases and configure ordering, etc. See the CONFIGURATION.md for more details.

Secret-store support

This plugin supports secrets from secret-stores for the username and password option. See the secret-store documentation for more details on how to use them.

Configuration

# AMQP consumer plugin
[[inputs.amqp_consumer]]
  ## Brokers to consume from.  If multiple brokers are specified a random broker
  ## will be selected anytime a connection is established.  This can be
  ## helpful for load balancing when not using a dedicated load balancer.
  brokers = ["amqp://localhost:5672/influxdb"]

  ## Authentication credentials for the PLAIN auth_method.
  # username = ""
  # password = ""

  ## 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".
  # exchange_type = "topic"

  ## If true, exchange will be passively declared.
  # exchange_passive = false

  ## Exchange durability can be either "transient" or "durable".
  # exchange_durability = "durable"

  ## Additional exchange arguments.
  # exchange_arguments = { }
  # exchange_arguments = {"hash_property" = "timestamp"}

  ## AMQP queue name.
  queue = "telegraf"

  ## AMQP queue durability can be "transient" or "durable".
  queue_durability = "durable"

  ## 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 = "#"

  ## Maximum number of messages server should give to the worker.
  # prefetch_count = 50

  ## Max undelivered messages
  ## This plugin uses tracking metrics, which ensure messages are read to
  ## outputs before acknowledging them to the original broker to ensure data
  ## is not lost. This option sets the maximum messages to read from the
  ## broker that have not been written by an output.
  ##
  ## This value needs to be picked with awareness of the agent's
  ## metric_batch_size value as well. Setting max undelivered messages too high
  ## can result in a constant stream of data batches to the output. While
  ## setting it too low may never flush the broker's messages.
  # max_undelivered_messages = 1000

  ## Auth method. PLAIN and EXTERNAL are supported
  ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as
  ## described here: https://www.rabbitmq.com/plugins.html
  # auth_method = "PLAIN"

  ## Optional TLS Config
  # tls_ca = "/etc/telegraf/ca.pem"
  # tls_cert = "/etc/telegraf/cert.pem"
  # tls_key = "/etc/telegraf/key.pem"
  ## Use TLS but skip chain & host verification
  # insecure_skip_verify = false

  ## Content encoding for message payloads, can be set to
  ## "gzip", "identity" or "auto"
  ## - Use "gzip" to decode gzip
  ## - Use "identity" to apply no encoding
  ## - Use "auto" determine the encoding using the ContentEncoding header
  # content_encoding = "identity"

  ## Maximum size of decoded message.
  ## Acceptable units are B, KiB, KB, MiB, MB...
  ## Without quotes and units, interpreted as size in bytes.
  # max_decompression_size = "500MB"

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"

Metrics

TODO

Example Output

TODO

Documentation

Index

• type AMQPConsumer
  • func (a *AMQPConsumer) Gather(_ telegraf.Accumulator) error
  • func (a *AMQPConsumer) Init() error
  • func (*AMQPConsumer) SampleConfig() string
  • func (a *AMQPConsumer) SetParser(parser telegraf.Parser)
  • func (a *AMQPConsumer) Start(acc telegraf.Accumulator) error
  • func (a *AMQPConsumer) Stop()

Types

type AMQPConsumer

type AMQPConsumer struct {
	URL                    string            `toml:"url" deprecated:"1.7.0;use 'brokers' instead"`
	Brokers                []string          `toml:"brokers"`
	Username               config.Secret     `toml:"username"`
	Password               config.Secret     `toml:"password"`
	Exchange               string            `toml:"exchange"`
	ExchangeType           string            `toml:"exchange_type"`
	ExchangeDurability     string            `toml:"exchange_durability"`
	ExchangePassive        bool              `toml:"exchange_passive"`
	ExchangeArguments      map[string]string `toml:"exchange_arguments"`
	MaxUndeliveredMessages int               `toml:"max_undelivered_messages"`

	// Queue Name
	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"`

	// Controls how many messages the server will try to keep on the network
	// for consumers before receiving delivery acks.
	PrefetchCount int

	// AMQP Auth method
	AuthMethod string
	tls.ClientConfig

	ContentEncoding      string      `toml:"content_encoding"`
	MaxDecompressionSize config.Size `toml:"max_decompression_size"`
	Log                  telegraf.Logger
	// contains filtered or unexported fields
}

AMQPConsumer is the top level struct for this plugin

func (*AMQPConsumer) Gather

func (a *AMQPConsumer) Gather(_ telegraf.Accumulator) error

All gathering is done in the Start function

func (*AMQPConsumer) Init

func (a *AMQPConsumer) Init() error

func (*AMQPConsumer) SampleConfig

func (*AMQPConsumer) SampleConfig() string

func (*AMQPConsumer) SetParser

func (a *AMQPConsumer) SetParser(parser telegraf.Parser)

func (*AMQPConsumer) Start

func (a *AMQPConsumer) Start(acc telegraf.Accumulator) error

Start satisfies the telegraf.ServiceInput interface

func (*AMQPConsumer) Stop

func (a *AMQPConsumer) Stop()