InfluxDB

Connect Connectware to InfluxDB 2 or InfluxDB 3 to write and query time-series data using MQTT topics.

Connect Connectware to InfluxDB, a time series database designed for storing and querying time-stamped data. This integration lets you read from and write to your InfluxDB instance using MQTT topics.

InfluxDB organizes data into measurements (similar to tables), where each measurement contains data points. Each point includes a timestamp, fields that hold the data values, and tags that you can use to organize and filter your data.

Connectware supports InfluxDB 2 and InfluxDB 3:

• The Influxdb protocol targets InfluxDB 2 and uses the Flux query language. See InfluxDB 2 Support.

• The Influxdb3 protocol targets InfluxDB 3 (Core, Enterprise, and Cloud) and uses SQL or InfluxQL. See InfluxDB 3 Support.

InfluxDB 2 Support

Connectware integrates with InfluxDB 2 using the Influxdb protocol.

• Connection Properties

• Endpoint Properties

Set the org and bucket connection properties to select the InfluxDB organization and bucket that Connectware reads from and writes to.

Reading Data

You can read data from InfluxDB in the following ways:

• One-time read: Define an endpoint with a read property to fetch data on demand.

• Continuous subscription: Define an endpoint with a subscribe property to poll data at regular intervals.

Both methods require a valid Flux query in the endpoint's query property. When you subscribe to data, specify a polling interval. The query runs automatically at that frequency.

Dynamic queries: Use @ placeholders in your Flux queries to make them flexible and reusable.

from(bucket:"@bucket") |> range(start: @startMeasurementTime) |> filter(fn: (r) => r._measurement == "@measurement")

In this example, when you request a read and provide values for bucket, measurement, and startMeasurementTime, Connectware generates the complete Flux query by replacing the @ placeholders with your values.

Output Format on Read

Connectware publishes query results as JSON to MQTT topics:

• One-time reads: Results appear on the endpoint's /res topic.

• Subscriptions: Results appear on the endpoint's default topic.

You receive the data as a JSON array containing your InfluxDB query results.

[
  {
    "result": "_result",
    "table": 0,
    "_start": "2021-02-14T09:29:24.514083303Z",
    "_stop": "2021-02-15T09:29:24.514083303Z",
    "_time": "2021-02-15T09:29:06.059Z",
    "_value": 19.7,
    "_field": "value",
    "_measurement": "temperature"
  },
  {
    "result": "_result",
    "table": 0,
    "_start": "2021-02-14T09:29:24.514083303Z",
    "_stop": "2021-02-15T09:29:24.514083303Z",
    "_time": "2021-02-15T09:29:06.059623817Z",
    "_value": 21.3,
    "_field": "value",
    "_measurement": "temperature"
  }
]

Writing Data

To write data to InfluxDB, define an endpoint with a write property.

Setting the measurement name: You can set a default measurement property in your endpoint configuration. This applies to all data points you send. To override this default, include a measurement property in individual data messages.

How to write: Send an MQTT message to the endpoint's /set topic. The following example includes a measurement name in the message:

{
  "tags": { "rpm": "8000", "oil_temp": "250" },
  "value": 91,
  "fields": { "engine_number": 1 },
  "timestamp": 1719129600000,
  "measurement": "temperature"
}

What you can include:

• Tags and fields: Use both to organize your data.

• Timestamp: Optional. If you do not provide one, InfluxDB assigns the current time.

Writing multiple points: Send an array of data points in a single message to write multiple values at once.

Write behavior: Writes are asynchronous. Connectware buffers data points and writes them to InfluxDB at intervals defined by the flushInterval connection property (default: 10 seconds). This follows InfluxDB's recommended client design pattern.

Output Format on Write

After writing data, you receive a confirmation message on the endpoint's /res topic. This response contains:

• timestamp: Unix timestamp (in milliseconds) of the write.

• value: Set to true when the write succeeds.

Service Commissioning File Example

This example demonstrates a complete InfluxDB 2 integration with write endpoints and a subscription for reading data.

influxdb-example.yml

3KB

Open

influxdb-example.yml

description: |
  Sample InfluxDB commissioning file

metadata:
  name: Cybus InfluxDB Example
  provider: cybus
  homepage: https://cybus.io
  version: 1.0.0

#------------------------------------------------------------------------------
# Parameters
#------------------------------------------------------------------------------

parameters:
  influxHost:
    type: string
    description: 'HTTP address of InfluxDB server'
    default: 'influxdbhost'

  influxPort:
    type: integer
    description: 'Influx Port'
    default: 8086

  influxScheme:
    type: 'string'
    description: 'Either use http or https for the server url'
    default: 'http'

#------------------------------------------------------------------------------
# Resources
#------------------------------------------------------------------------------
resources:
  #----------------------------------------------------------------------------
  # Connections
  #----------------------------------------------------------------------------

  influxdbConnection:
    type: Cybus::Connection
    properties:
      protocol: Influxdb
      connection:
        host: !ref influxHost
        token: '-an-influx-db-jwt-token-'
        port: !ref influxPort
        bucket: turbine
        scheme: !ref influxScheme
        flushInterval: 5000

  #----------------------------------------------------------------------------
  # Endpoints
  #----------------------------------------------------------------------------

  turbineWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Influxdb
      connection: !ref influxdbConnection
      write:
        measurement: 'turbine'

  rotaryEncoderWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Influxdb
      connection: !ref influxdbConnection
      write:
        measurement: 'rotary_encoder'

  rotary_encoder_angle:
    type: Cybus::Endpoint
    properties:
      protocol: Influxdb
      connection: !ref influxdbConnection
      subscribe:
        interval: 6000
        query: >
          from(bucket:"turbine") |> range(start: -1d) |> filter(fn: (r) => r._measurement == "rotary_encoder")

  #----------------------------------------------------------------------------
  # Mappings
  #----------------------------------------------------------------------------

  # A mapping that overrides the measurement value with the name of the MQTT topic
  mappings:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: 'turbine/#'
          publish:
            endpoint: !ref turbineWrite
          rules:
            - transform:
                # Add the topic as measurement
                expression: '$merge([$,{"measurement": $context.topic}])'
        # A mapping that will pass data from the MQTT topic to the write endpoint
        # allowing overriding the measurement by providing it in the topic
        - subscribe:
            topic: 'encoder/#'
          publish:
            endpoint: !ref rotaryEncoderWrite

InfluxDB 3 Support

Connectware integrates with InfluxDB 3 (Core, Enterprise, and Cloud) using the Influxdb3 protocol.

• Connection Properties

• Endpoint Properties

Set the database connection property to the InfluxDB 3 database that Connectware reads from and writes to. In InfluxDB 3, a database is the equivalent of a bucket in InfluxDB 2.

Authentication

InfluxDB 3 uses token-based authentication. Set the token connection property. Use a token that is scoped to the specific database and permission level you need, for example a write-only token, instead of an admin token.

The authScheme connection property selects the token format:

• Leave authScheme unset for InfluxDB Cloud.

• Set authScheme to Bearer for InfluxDB 3 Core and Enterprise.

To encrypt the connection with TLS, set scheme to https and make sure that the Connectware agent or pod trusts the certificate of the InfluxDB 3 server. For advanced TLS options, use the transportOptions connection property.

Writing Data

Define a Cybus::Endpoint with a write property to send data to InfluxDB 3. Publishing an MQTT message to the endpoint's /set topic writes one or more points. A message contains the fields to write and can also contain tags, a timestamp, and a measurement name that overrides the endpoint's default. Fields and tags that you define on the endpoint are merged with the values from the message. If a message does not contain a timestamp, the connector uses the current system time. Timestamps use millisecond precision (precision: ms).

{
  "measurement": "exhaust_temperature",
  "timestamp": 1752650000000,
  "tags": { "machine": "press_1", "line": "A" },
  "fields": {
    "temperature": 81.5,
    "running": true,
    "status": "ok",
    "cycle_count": 1500
  }
}

Writing multiple points: Send an array of data points in a single message to write multiple values at once.

Writes are asynchronous. The connector buffers incoming points and flushes them as a single batch when either of two thresholds is met, whichever comes first:

• The buffer reaches the batchSize limit (default: 1000 points).

• The flushInterval elapses (default: 3000 milliseconds).

If a write fails, the connector retries the data when the next batch is written.

Field data types. InfluxDB 3 fixes each field column's data type on the first write to a measurement. By default, the connector infers numeric types from the payload — whole numbers become integer, other numbers become float. If a field can arrive as either, the type inferred from the first message may be rejected by later writes.

To avoid this, declare types explicitly using the fieldTypes endpoint property. Allowed values are float, integer, uinteger, string, and boolean.

Output Format on Write

After writing data, you receive a confirmation message on the endpoint's /res topic:

{ "value": true, "timestamp": 1784204563072 }

• timestamp: Unix timestamp (in milliseconds) of the write.

• value: Set to true when the write succeeds.

Reading Data

Define an endpoint with a read property for one-time queries, or a subscribe property for continuous polling. For subscriptions, set an interval in milliseconds or a cronExpression to control the polling schedule. Provide the query as the endpoint's query value.

InfluxDB 3 uses SQL as its default query language. InfluxQL is also supported for compatibility with existing InfluxQL queries. Set queryType to sql (default) or influxql, either at the connection level or per endpoint.

To bind values into a query, prefer the params endpoint property over string interpolation. Parameters supplied in a read request payload override the endpoint's defaults.

Output Format on Read

Connectware publishes query results as JSON to MQTT topics:

• One-time reads: Results appear on the endpoint's /res topic.

• Subscriptions: Results appear on the endpoint's default topic.

You receive the data as a JSON array containing one object per result row. Each object contains the queried columns and the time value as a Unix timestamp in milliseconds.

[
  {
    "cycle_count": 1500,
    "line": "A",
    "machine": "press_1",
    "running": true,
    "status": "ok",
    "temperature": 81.5,
    "time": 1752650000000,
    "value": null
  }
]

InfluxDB 3 Service Commissioning File Example

This example connects to an InfluxDB 3 server and writes data to the machine_metrics measurement, adding a constant tag to every point.

influxdb3-example.yml

914B

Open

influxdb3-example.yml

description: |
  Sample InfluxDB 3 service commissioning file

metadata:
  name: Cybus InfluxDB 3 Example
  provider: cybus
  homepage: https://cybus.io
  version: 1.0.0

parameters:
  influxHost:
    type: string
    description: 'Hostname or IP address of the InfluxDB 3 server'
    default: 'influxdb3'

  influxToken:
    type: string
    description: 'Authentication token for the InfluxDB 3 server'

resources:
  influxdb3Connection:
    type: Cybus::Connection
    properties:
      protocol: Influxdb3
      connection:
        host: !ref influxHost
        port: 8181
        scheme: http
        authScheme: Bearer
        token: !ref influxToken
        database: machine_data

  machineMetricsWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Influxdb3
      connection: !ref influxdb3Connection
      write:
        measurement: machine_metrics
        tags:
          line: assembly-1