InfluxDB and Telegraf

This page describes a service (subsystem) that is used to store information from RTC components to an InfluxDB database.

InfluxDB

InfluxDB is an open-source MIT-licensed time-series database suitable for storing and querying metrics from RTC components. It supports extremely high write throughput, nanosecond-level timestamp precision, and fast analytical queries.

InfluxDB 3 introduces major architectural changes compared to earlier versions. It uses a SQL-based query engine, supports unlimited cardinality, auto-creates databases and tables on write, and exposes its API through Apache Arrow Flight SQL (gRPC). A token-based access system is used here for the communication.

InfluxDB 3 exposes data via:

SQL and InfluxQL query languages
Apache Arrow Flight SQL for high-performance reads and writes
Official client libraries for C#, Go, Java, Node.js, Python

For writing data is used the Influx line protocol. When receiving new measurements or columns, the database will automatically extend its schema.

Here is provided example communication with InfluxDB using client library for Python langauge:

import random
import time
from influxdb_client_3 import InfluxDBClient3, Point

def writing(database):
    """Creates points in infinite loop separated by 1 second from each other"""

    while True:
        # create point
        p = (Point(measurement_name="metrics")
            .tag("component", "rtc_component_1")
            .tag("aggregation", random.choice(["mean", "min", "max"]))
            .tag("metric_type", "buffer_occupancy")
            .field("value", round(random.uniform(13.0, 16.0), 2)))

        # write it to influx
        client.write(database=database, record=p)

        # wait 1s
        time.sleep(1)

if __name__ == "__main__":
    try:
        writing(DATABASE)
    finally:
        client.close()

Terminology

Fields
- key–value pairs containing actual metric values (e.g., temperature = 23.5)
- fields are not indexed, so filtering by them is slower on large datasets
Tags
- string-string key–value metadata (e.g., host = server1, region = eu)
- tags are indexed, making filtering and grouping efficient
- they are fully optional, you don’t need to provide any
Points
- single datapoints defined by table, tags, fields and timestamp
Series
- a group of points with the same measurement and tag set
- each series has a unique series key and unique timestamps per point
Tables
- formally known as measurements
- structured tables grouping related data
- columns (field/tags) and data types are derived automatically from incoming data
Databases
- formally known as buckets
- named locations for storing time-series data and retention policies
- InfluxDB 3 automatically creates them when needed

Telegraf

Telegraf is a plugin-based agent for collecting, processing, and forwarding data. It runs as a system daemon and is configured entirely through a single configuration file.

It can gather data from hardware, operating systems, applications, or sockets, transform it using processors/aggregators, and forward it to multiple destinations.

One of these destinations can be InfluxDB. Since no native InfluxDB 3 output plugin exists yet, Telegraf writes to InfluxDB 3 using the InfluxDB v2 output plugin, which is compatible with the InfluxDB 3 write API.

Plugins

Input plugins – collect data (CPU, memory, disks, processes, GPUs, network, socket listeners, etc.).
Output plugins – deliver data to storage systems (including InfluxDB via outputs.influxdb_v2).
Processor and aggregator plugins – allow transformations before writing.
External plugins – telegraf can run standalone programs via execd.

Multiple plugins can be enabled simultaneously. Custom plugins can be written in Go language.

TelegrafAdapter

RTC toolkit components do not communicate with InfluxDB directly. Instead, they contact the Telegraf. This type of communication has two causes: first, it avoids the need to add access tokens for communication into the configuration of RTC components. The second cause is that Telegraf can collect system metrics data as well. So it is possible to store system data alongside with RTC data and later correlate them.

The RTC toolkit includes a service called TelegrafAdapter. This service mediates the communication between an RTC component and Telegraf. The way this communication works is shown in the diagram below.

../../_images/InfluxDB_Telegraf_Diagram.png

Communication is handled via sockets. It uses the UDP protocol and the data is sent in line-protocol format (in Telegraf is this format called influx). TelegrafAdapter is part of the ServiceContainer, and its instance can be obtained as follows:

auto telegraf_adapter = m_services.Get<TelegrafAdapter>();

With this instance, one can then send data to InfluxDB using the SendPoints() method. This method expects a vector of InfluxPoint structures to be provided.

auto point = InfluxPoint{
    .table_name = "point",
    .tags = InfluxTagMap(),
    .fields = {{"value", false}}
};

telegraf_adapter.SendPoints({ point });

InfluxPoint

An InfluxPoint represents a single atomic record in the InfluxDB. It contains four different properties that reflect how InfluxDB stores data. The names of these properties follow the terminology introduced in InfluxDB version 3. These properties are:

table_name – mandatory; specifies the table in which the record will be stored
tags – optional; can remain empty; tags are indexed in InfluxDB and are used for searching or filtering records; it can be set either using a map<std::string, std::string> or through the InfluxTagMap class
fields – mandatory; the InfluxPoint must contain at least one field, otherwise it has no meaning
timestamp – optional; user can set it, but if left empty, InfluxPoint will insert the current time when converting to line protocol

InfluxTagMap

This is a helper class that behaves like map<std::string, std::string>. Compared to a regular map, InfluxTagMap additionally provides the Extend() method, which allows existing maps to be extended in a chained manner with additional tags. The method can be used as follows:

auto tags = InfluxTagMap(telegraf_adapter.BaseTags())
    .Extend({{"name", "status"}})
    .Extend({{"is_reduced", "true"}});

Integration in RTC toolkit

TelegrafAdapter is a completely optional subsystem. It retrieves information about the Telegraf agent’s endpoint from the Service Discovery, and if this endpoint contains the scheme null://, TelegrafAdapter will be entirely disabled. In that case, no data will be sent at all.

However, as mentioned above, the communication is implemented using UDP. So if TelegrafAdapter should be used, the expected format of the Telegraf endpoint is following: udp://127.0.0.1:8081.

It is expected to be active by default, and therefore it is integrated in several places. Currently, TelegrafAdapter is used to send three different types of data to InfluxDB.

Metrics

TelegrafAdapter is used directly in metrics, so any metric intended to be stored in OLDB is also stored in InfluxDB. Provided metrics are stored in the table called metrics. Metrics stored in InfluxDB can be customised in many ways - from column names to the definition of tags that should also be stored.

States

RTC Component states are stored whenever they change. At the moment the event of a component’s state change occurs, this information is also sent to InfluxDB. All such changes are stored in the state table.

Alerts

For storing alert information, the AlertTelegrafPublisher class was created, which inherits from AlertObserverIf. This mechanism is event-driven: when an alert transitions to a different state (true/false), the corresponding record is emitted and written to the alert table in InfluxDB.