AMQP and MongoDB Integration

Input and output integration overview

The AMQP Consumer Input Plugin allows you to ingest data from an AMQP 0-9-1 compliant message broker, such as RabbitMQ, enabling seamless data collection for monitoring and analytics purposes.

The MongoDB Telegraf Plugin enables users to send metrics to a MongoDB database, automatically managing time series collections.

Integration details

AMQP

This plugin provides a consumer for use with AMQP 0-9-1, a prominent implementation of which is RabbitMQ. AMQP, or Advanced Message Queuing Protocol, was originally developed to enable reliable, interoperable messaging between diverse systems in a network. The plugin reads metrics from a topic exchange using a configured queue and binding key, delivering a flexible and efficient means of collecting data from AMQP-compliant messaging systems. This enables users to leverage existing RabbitMQ implementations to monitor their applications effectively by capturing detailed metrics for analysis and alerting.

MongoDB

This plugin sends metrics to MongoDB and seamlessly integrates with its time series functionality, allowing for automatic creation of collections as time series when they don’t already exist. It requires MongoDB version 5.0 or higher to utilize the time series collections feature, which is vital for efficiently storing and querying time-based data. This plugin enhances the monitoring capabilities by ensuring that all relevant metrics are stored and organized correctly within MongoDB, providing users the ability to leverage MongoDB’s powerful querying and aggregation features for time series analysis.

Configuration

AMQP

[[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

  ## Timeout for establishing the connection to a broker
  # timeout = "30s"

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

MongoDB

[[outputs.mongodb]]
              # connection string examples for mongodb
              dsn = "mongodb://localhost:27017"
              # dsn = "mongodb://mongod1:27017,mongod2:27017,mongod3:27017/admin&replicaSet=myReplSet&w=1"

              # overrides serverSelectionTimeoutMS in dsn if set
              # timeout = "30s"

              # default authentication, optional
              # authentication = "NONE"

              # for SCRAM-SHA-256 authentication
              # authentication = "SCRAM"
              # username = "root"
              # password = "***"

              # for x509 certificate authentication
              # authentication = "X509"
              # tls_ca = "ca.pem"
              # tls_key = "client.pem"
              # # tls_key_pwd = "changeme" # required for encrypted tls_key
              # insecure_skip_verify = false

              # database to store measurements and time series collections
              # database = "telegraf"

              # granularity can be seconds, minutes, or hours.
              # configuring this value will be based on your input collection frequency.
              # see https://docs.mongodb.com/manual/core/timeseries-collections/#create-a-time-series-collection
              # granularity = "seconds"

              # optionally set a TTL to automatically expire documents from the measurement collections.
              # ttl = "360h"

Input and output integration examples

AMQP

1. Integrating Application Metrics with AMQP: Use the AMQP Consumer plugin to gather application metrics that are published to a RabbitMQ exchange. By configuring the plugin to listen to specific queues, teams can gain insights into application performance, track request rates, error counts, and latency metrics, all in real-time. This setup not only aids in anomaly detection but also provides valuable data for capacity planning and system optimization.

2. Event-Driven Monitoring: Configure the AMQP Consumer to trigger specific monitoring events whenever certain conditions are met within an application. For instance, if a message indicating a high error rate is received, the plugin can feed this data into monitoring tools, generating alerts or scaling events. This integration can improve responsiveness to issues and automate parts of the operations workflow.

3. Cross-Platform Data Aggregation: Leverage the AMQP Consumer plugin to consolidate metrics from various applications distributed across different platforms. By utilizing RabbitMQ as a centralized message broker, organizations can unify their monitoring data, allowing for comprehensive analysis and dashboarding through Telegraf, thus maintaining visibility across heterogeneous environments.

4. Real-Time Log Processing: Extend the use of the AMQP Consumer to capture log data sent to a RabbitMQ exchange, processing logs in real time for monitoring and alerting purposes. This application ensures that operational issues are detected and addressed swiftly by analyzing log patterns, trends, and anomalies as they occur.

MongoDB

1. Dynamic Logging to MongoDB for IoT Devices: Utilize this plugin to collect and store metrics from a fleet of IoT devices in real-time. By sending device logs directly to MongoDB, you can create a centralized database that allows for easy access and querying of health metrics and performance data, enabling proactive maintenance and troubleshooting based on historical trends.

2. Time Series Analysis of Web Traffic: Use the MongoDB Telegraf Plugin to gather and analyze web traffic metrics over time. This application can help you understand peak usage times, user interactions, and behavior patterns, which can guide marketing strategies and infrastructure scaling decisions for improved user experience.

3. Automated Monitoring and Alerting System: Integrate the MongoDB plugin into an automated monitoring system that tracks application performance metrics. With time series collections, you can set up alerts based on specific thresholds, allowing your team to respond to potential issues before they affect users. This proactive management can enhance service reliability and overall performance.

4. Data Retention and TTL Management in Metrics Storage: Leverage the TTL feature for documents within MongoDB collections to auto-expire outdated metrics. This is particularly useful for environments where only recent performance data is relevant, preventing your MongoDB database from becoming cluttered with old metrics and ensuring efficient data management.

Feedback

Thank you for being part of our community! If you have any general feedback or found any bugs on these pages, we welcome and encourage your input. Please submit your feedback in the InfluxDB community Slack.