AMQP and TimescaleDB Integration
Powerful performance with an easy integration, powered by Telegraf, the open source data connector built by InfluxData.
5B+
Telegraf downloads
#1
Time series database
Source: DB Engines
1B+
Downloads of InfluxDB
2,800+
Contributors
Table of Contents
Powerful Performance, Limitless Scale
Collect, organize, and act on massive volumes of high-velocity data. Any data is more valuable when you think of it as time series data. with InfluxDB, the #1 time series platform built to scale with Telegraf.
See Ways to Get Started
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.
This output plugin delivers a reliable and efficient mechanism for routing Telegraf collected metrics directly into TimescaleDB. By leveraging PostgreSQL’s robust ecosystem combined with TimescaleDB’s time series optimizations, it supports high-performance data ingestion and advanced querying capabilities.
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.
TimescaleDB
TimescaleDB is an open source time series database built as an extension to PostgreSQL, designed to handle large scale, time-oriented data efficiently. Launched in 2017, TimescaleDB emerged in response to the growing need for a robust, scalable solution that could manage vast volumes of data with high insert rates and complex queries. By leveraging PostgreSQL’s familiar SQL interface and enhancing it with specialized time series capabilities, TimescaleDB quickly gained popularity among developers looking to integrate time series functionality into existing relational databases. Its hybrid approach allows users to benefit from PostgreSQL’s flexibility, reliability, and ecosystem while providing optimized performance for time series data.
The database is particularly effective in environments that demand fast ingestion of data points combined with sophisticated analytical queries over historical periods. TimescaleDB has a number of innovative features like hypertables which transparently partition data into manageable chunks and built-in continuous aggregation. These allow for significantly improved query speed and resource efficiency.
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"
TimescaleDB
# Publishes metrics to a TimescaleDB database
[[outputs.postgresql]]
## Specify connection address via the standard libpq connection string:
## host=... user=... password=... sslmode=... dbname=...
## Or a URL:
## postgres://[user[:password]]@localhost[/dbname]?sslmode=[disable|verify-ca|verify-full]
## See https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING
##
## All connection parameters are optional. Environment vars are also supported.
## e.g. PGPASSWORD, PGHOST, PGUSER, PGDATABASE
## All supported vars can be found here:
## https://www.postgresql.org/docs/current/libpq-envars.html
##
## Non-standard parameters:
## pool_max_conns (default: 1) - Maximum size of connection pool for parallel (per-batch per-table) inserts.
## pool_min_conns (default: 0) - Minimum size of connection pool.
## pool_max_conn_lifetime (default: 0s) - Maximum connection age before closing.
## pool_max_conn_idle_time (default: 0s) - Maximum idle time of a connection before closing.
## pool_health_check_period (default: 0s) - Duration between health checks on idle connections.
# connection = ""
## Postgres schema to use.
# schema = "public"
## Store tags as foreign keys in the metrics table. Default is false.
# tags_as_foreign_keys = false
## Suffix to append to table name (measurement name) for the foreign tag table.
# tag_table_suffix = "_tag"
## Deny inserting metrics if the foreign tag can't be inserted.
# foreign_tag_constraint = false
## Store all tags as a JSONB object in a single 'tags' column.
# tags_as_jsonb = false
## Store all fields as a JSONB object in a single 'fields' column.
# fields_as_jsonb = false
## Name of the timestamp column
## NOTE: Some tools (e.g. Grafana) require the default name so be careful!
# timestamp_column_name = "time"
## Type of the timestamp column
## Currently, "timestamp without time zone" and "timestamp with time zone"
## are supported
# timestamp_column_type = "timestamp without time zone"
## Templated statements to execute when creating a new table.
# create_templates = [
# '''CREATE TABLE {{ .table }} ({{ .columns }})''',
# ]
## Templated statements to execute when adding columns to a table.
## Set to an empty list to disable. Points containing tags for which there is
## no column will be skipped. Points containing fields for which there is no
## column will have the field omitted.
# add_column_templates = [
# '''ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join ", ADD COLUMN IF NOT EXISTS " }}''',
# ]
## Templated statements to execute when creating a new tag table.
# tag_table_create_templates = [
# '''CREATE TABLE {{ .table }} ({{ .columns }}, PRIMARY KEY (tag_id))''',
# ]
## Templated statements to execute when adding columns to a tag table.
## Set to an empty list to disable. Points containing tags for which there is
## no column will be skipped.
# tag_table_add_column_templates = [
# '''ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join ", ADD COLUMN IF NOT EXISTS " }}''',
# ]
## The postgres data type to use for storing unsigned 64-bit integer values
## (Postgres does not have a native unsigned 64-bit integer type).
## The value can be one of:
## numeric - Uses the PostgreSQL "numeric" data type.
## uint8 - Requires pguint extension (https://github.com/petere/pguint)
# uint64_type = "numeric"
## When using pool_max_conns > 1, and a temporary error occurs, the query is
## retried with an incremental backoff. This controls the maximum duration.
# retry_max_backoff = "15s"
## Approximate number of tag IDs to store in in-memory cache (when using
## tags_as_foreign_keys). This is an optimization to skip inserting known
## tag IDs. Each entry consumes approximately 34 bytes of memory.
# tag_cache_size = 100000
## Cut column names at the given length to not exceed PostgreSQL's
## 'identifier length' limit (default: no limit)
## (see https://www.postgresql.org/docs/current/limits.html)
## Be careful to not create duplicate column names!
# column_name_length_limit = 0
## Enable & set the log level for the Postgres driver.
# log_level = "warn" # trace, debug, info, warn, error, none
Input and output integration examples
AMQP
-
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.
-
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.
-
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.
-
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.
TimescaleDB
-
Real-Time IoT Data Ingestion: Use the plugin to collect and store sensor data from thousands of IoT devices in real time. This setup facilitates immediate analysis, helping organizations monitor operational efficiency and respond quickly to changing conditions.
-
Cloud Application Performance Monitoring: Leverage the plugin to feed detailed performance metrics from distributed cloud applications into TimescaleDB. This integration supports real-time dashboards and alerts, enabling teams to swiftly identify and mitigate performance bottlenecks.
-
Historical Data Analysis and Reporting: Implement a system where long-term metrics are stored in TimescaleDB for comprehensive historical analysis. This approach allows businesses to perform trend analysis, generate detailed reports, and make data-driven decisions based on archived time-series data.
-
Adaptive Alerting and Anomaly Detection: Integrate the plugin with automated anomaly detection workflows. By continuously streaming metrics to TimescaleDB, machine learning models can analyze data patterns and trigger alerts when anomalies occur, enhancing system reliability and proactive maintenance.
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.
Powerful Performance, Limitless Scale
Collect, organize, and act on massive volumes of high-velocity data. Any data is more valuable when you think of it as time series data. with InfluxDB, the #1 time series platform built to scale with Telegraf.
See Ways to Get Started
Related Integrations
Related Integrations
HTTP and InfluxDB Integration
The HTTP plugin collects metrics from one or more HTTP(S) endpoints. It supports various authentication methods and configuration options for data formats.
View IntegrationKafka and InfluxDB Integration
This plugin reads messages from Kafka and allows the creation of metrics based on those messages. It supports various configurations including different Kafka settings and message processing options.
View IntegrationKinesis and InfluxDB Integration
The Kinesis plugin allows for reading metrics from AWS Kinesis streams. It supports multiple input data formats and offers checkpointing features with DynamoDB for reliable message processing.
View Integration