InfluxDB

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 familiar MQTT topics.

InfluxDB organizes data into measurements (similar to tables), where each measurement contains data points. Each point includes a timestamp, a value, fields, and tags that help you organize and query your data effectively.

This protocol implementation supports the Flux query language, giving you powerful querying capabilities for your time series data.

Use the configuration parameters and examples below to integrate InfluxDB with your Connectware services.

Connection Properties

Endpoint Properties

Reading Data

You can read data from InfluxDB in two 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 field. When you subscribe to data, specify a polling interval. Your query runs automatically at that frequency. Query results are delivered as JSON to your MQTT broker.

Dynamic queries: Use value interpolation in your Flux queries with tags like @tag to make your queries 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, the system automatically generates a complete Flux query by replacing the @ placeholders with your values.

Output Format on Read

Query results are automatically published to MQTT topics:

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

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

You will 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, create 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. If you need flexibility, override this default by including a measurement property in your individual data messages.

How to write: Send an MQTT message to your endpoint's /set topic. Here is an example that includes a measurement name in the message:

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

What you can include:

• Tags and fields: Both are fully supported for organizing your data.

• Timestamp: Optional. If you don't provide one, InfluxDB automatically 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 to optimize performance. Data is buffered and written 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 will receive a confirmation message on the endpoint's /res topic. This response contains:

• timestamp: Unix timestamp (in milliseconds) when the write was executed.

• value: Set to true when the write succeeds.

Service Commissioning File Example

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

3KB

influxdb-example.yml

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