RabbitMQ and Databricks Integration

Input and output integration overview

This plugin reads metrics from RabbitMQ servers, providing essential insights into the performance and state of the messaging system.

Use Telegraf’s HTTP output plugin to push metrics straight into a Databricks Lakehouse by calling the SQL Statement Execution API with a JSON-wrapped INSERT or volume PUT command.

Integration details

RabbitMQ

The RabbitMQ plugin for Telegraf allows users to gather metrics from RabbitMQ servers via the RabbitMQ Management Plugin. This capability is crucial for monitoring the performance and health of RabbitMQ instances, which are widely utilized for message queuing and processing in various applications. The plugin provides comprehensive insights into key RabbitMQ metrics, including message rates, queue depths, and node health statistics, thereby enabling operators to maintain optimal performance and robustness of their messaging infrastructure. Additionally, it supports secret-stores for managing sensitive credentials securely, making integration with existing systems smoother. Configuration options allow for flexibility in specifying the nodes, queues, and exchanges to monitor, providing valuable adaptability for diverse deployment scenarios.

Databricks

This configuration turns Telegraf into a lightweight ingestion agent for the Databricks Lakehouse. It leverages the Databricks SQL Statement Execution API 2.0, which accepts authenticated POST requests containing a JSON payload with a statement field. Each Telegraf flush dynamically renders a SQL INSERT (or, for file-based workflows, a PUT ... INTO /Volumes/... command) that lands the metrics into a Unity Catalog table or volume governed by Lakehouse security. Under the hood Databricks stores successful inserts as Delta Lake transactions, enabling ACID guarantees, time-travel, and scalable analytics. Operators can point the warehouse_id at any serverless or classic SQL warehouse, and all authentication is handled with a PAT or service-principal token—no agents or JDBC drivers required. Because Telegraf’s HTTP output supports custom headers, batching, TLS, and proxy settings, the same pattern scales from edge IoT gateways to container sidecars, consolidating infrastructure telemetry, application logs, or business KPIs directly into the Lakehouse for BI, ML, and Lakehouse Monitoring. Unity Catalog volumes provide a governed staging layer when file uploads and COPY INTO are preferred, and the approach aligns with Databricks’ recommended ingestion practices for partners and ISVs.

Configuration

RabbitMQ

[[inputs.rabbitmq]]
  ## Management Plugin url. (default: http://localhost:15672)
  # url = "http://localhost:15672"
  ## Tag added to rabbitmq_overview series; deprecated: use tags
  # name = "rmq-server-1"
  ## Credentials
  # username = "guest"
  # password = "guest"

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

  ## Optional request timeouts
  ## ResponseHeaderTimeout, if non-zero, specifies the amount of time to wait
  ## for a server's response headers after fully writing the request.
  # header_timeout = "3s"
  ##
  ## client_timeout specifies a time limit for requests made by this client.
  ## Includes connection time, any redirects, and reading the response body.
  # client_timeout = "4s"

  ## A list of nodes to gather as the rabbitmq_node measurement. If not
  ## specified, metrics for all nodes are gathered.
  # nodes = ["rabbit@node1", "rabbit@node2"]

  ## A list of queues to gather as the rabbitmq_queue measurement. If not
  ## specified, metrics for all queues are gathered.
  ## Deprecated in 1.6: Use queue_name_include instead.
  # queues = ["telegraf"]

  ## A list of exchanges to gather as the rabbitmq_exchange measurement. If not
  ## specified, metrics for all exchanges are gathered.
  # exchanges = ["telegraf"]

  ## Metrics to include and exclude. Globs accepted.
  ## Note that an empty array for both will include all metrics
  ## Currently the following metrics are supported: "exchange", "federation", "node", "overview", "queue"
  # metric_include = []
  # metric_exclude = []

  ## Queues to include and exclude. Globs accepted.
  ## Note that an empty array for both will include all queues
  # queue_name_include = []
  # queue_name_exclude = []

  ## Federation upstreams to include and exclude specified as an array of glob
  ## pattern strings.  Federation links can also be limited by the queue and
  ## exchange filters.
  # federation_upstream_include = []
  # federation_upstream_exclude = []

Databricks

[[outputs.http]]
  ## Databricks SQL Statement Execution API endpoint
  url = "https://{{ env "DATABRICKS_HOST" }}/api/2.0/sql/statements"

  ## Use POST to submit each Telegraf batch as a SQL request
  method = "POST"

  ## Personal-access token (PAT) for workspace or service principal
  headers = { Authorization = "Bearer {{ env "DATABRICKS_TOKEN" }}" }

  ## Send JSON that wraps the metrics batch in a SQL INSERT (or PUT into a Volume)
  content_type = "application/json"

  ## Serialize metrics as JSON so they can be embedded in the SQL statement
  data_format = "json"
  json_timestamp_units = "1ms"

  ## Build the request body.  Telegraf replaces the template variables at runtime.
  ## Example inserts a row per metric into a Unity-Catalog table.
  body_template = """
  {
    \"statement\": \"INSERT INTO ${TARGET_TABLE} VALUES {{range .Metrics}}(from_unixtime({{.timestamp}}/1000), {{.fields.usage}}, '{{.tags.host}}'){{end}}\",
    \"warehouse_id\": \"${WAREHOUSE_ID}\"
  }
  """

  ## Optional: add batching limits or TLS settings
  # batch_size = 500
  # timeout     = "10s"

Input and output integration examples

RabbitMQ

1. Monitoring Queue Performance Metrics: Use the RabbitMQ plugin to keep track of queue performance over time. This involves setting up monitoring dashboards that visualize crucial queue metrics such as message rates, the number of consumers, and message delivery rates. With this information, teams can proactively address any bottlenecks or performance issues by analyzing trends and making data-informed decisions about scaling or optimizing their RabbitMQ configuration.

2. Alerting on System Health: Integrate the RabbitMQ plugin with an alerting system to notify operational teams of potential issues within RabbitMQ instances. For example, if the number of unacknowledged messages reaches a critical threshold or if queues become overwhelmed, alerts can trigger, allowing for immediate investigation and swift remedial action to maintain the health of message flows.

3. Analyzing Message Processing Metrics: Employ the plugin to gather detailed metrics on message processing performance, such as the rates of messages published, acknowledged, and redelivered. By analyzing these metrics, teams can evaluate the efficiency of their message consumer applications and make adjustments to configuration or code where necessary, thereby enhancing overall system throughput and resilience.

4. Cross-System Data Integration: Leverage the metrics collected by the RabbitMQ plugin to integrate data flows between RabbitMQ and other systems or services. For example, use the gathered metrics to drive automated workflows or analytics pipelines that utilize messages processed in RabbitMQ, enabling organizations to optimize workflows and enhance data agility across their ecosystems.

Databricks

1. Edge-to-Lakehouse Telemetry Pipe: Deploy Telegraf on factory PLCs to sample vibration metrics and post them every second to a serverless SQL warehouse. Delta tables power PowerBI dashboards that alert engineers when thresholds drift.
2. Blue-Green CI/CD Rollout Metrics: Attach a Telegraf sidecar to each Kubernetes canary pod; it inserts container stats into a Unity Catalog table tagged by deployment_id, letting Databricks SQL compare error-rate percentiles and auto-rollback underperforming versions.
3. SaaS Usage Metering: Insert per-tenant API-call counters via the HTTP plugin; a nightly Lakehouse query aggregates usage into invoices, eliminating custom metering micro-services.
4. Security Forensics Lake: Upload JSON batches of Suricata IDS events to a Unity Catalog volume using PUT commands, then run COPY INTO for near-real-time enrichment with Delta Live Tables, producing a searchable threat-intel lake that joins network logs with user session data.

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.