Ingesting Real-Time Data to InfluxDB with PubSub+

You’ve probably heard about Solace following open standards, but why does it matter to your business/applications? There are typically rules to ensure that your products and applications maintain interoperability. How does that apply to event brokers? Often customers ask if Solace can integrate with other systems, and specifically in capital markets and IoT domains, if Solace has real-time integration with time series databases like InfluxDB.

Since Solace offers a wide range of connectors and is based on open standards, you don’t usually need to build processes to integrate with applications — all you need to do is spin up a connector to build the integration pipeline. Similarly, InfluxData provides a wide range of plugins which can act as a consumer to Solace and ingest data directly into InfluxDB.

InfluxDB is an open-source time series database (TSDB) developed by the company InfluxData. It is used for storage and retrieval of time series data in fields such as operations monitoring, application metrics, Internet of Things sensor data, and real-time analytics. InfluxDB is capable of ingesting millions of data points per second providing high level performance.  The classic real-world example of a time series is stock exchange currency price data or IoT data emitted by sensors.

In this blog post I will illustrate the steps required to ingest the data from Solace to InfluxDB at real time via Telegraf MQTT plugin.

Solace facilitates integration with applications, other middleware, and data movement technology with support for open APIs like JMS and OpenMAMA, and open wireline protocols such as AMQP, MQTT and REST. This allows you to get real-time data flowing between diverse applications, big data systems, cloud services and IoT devices.

Telegraf input plugins are used with the InfluxData time series platform to collect metrics from the system, services, or third-party APIs. The MQTT consumer plugin reads from the specified MQTT topics and creates metrics using one of the supported input data formats. The output of our work today will create a pipeline flowing from Solace to InfluxDB

1. Start instance of influxDB, one can even use sign up for the free edition of influx db at https://cloud2.influxdata.com/signup
   
2. Create a bucket in influxDB to hold the IoT data:
   
3. Create a Telegraf configuration for MQTT source plugin to consume messages from the Solace broker via MQTT and insert them into influxDB.
   
4. Update the Telegraf configuration with the Solace broker details-
   1. Broker details
   2. Change the QoS to 1 to consume messages from queue.
   3. Topics for the Telegraf plugin to consume messages published on Solace and insert into influxDB. A queue will be created with the topic subscription specified in the topic section.

      [[inputs.mqtt_consumer]]
        ## Broker URLs for the MQTT server or cluster.  To connect to multiple
        ## clusters or standalone servers, use a separate plugin instance.
        ##   example: servers = ["tcp://localhost:1883"]
        ##            servers = ["ssl://localhost:1883"]
        ##            servers = ["ws://localhost:1883"]
        servers = ["tcp://mr-connection-tx0q1aux4ms.messaging.solace.cloud:1883"]
      
        ## Topics that will be subscribed to.
        topics = [
        
          "telegraf/iot/solace"
        ]
      
        ## The message topic will be stored in a tag specified by this value.  If set
        ## to the empty string no topic tag will be created.
        # topic_tag = "topic"
      
        ## QoS policy for messages
        ##   0 = at most once
        #   1 = at least once
        ##   2 = exactly once
        ##
        ## When using a QoS of 1 or 2, you should enable persistent_session to allow
        ## resuming unacknowledged messages.
        qos = 1
      

5. Update the Solace credentials and data format. For this example, data format used is json. Save changes.

   ## Username and password to connect MQTT server.
      username = "admin"
      password = "admin"
   
     ## 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
   
     ## 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 = "json"
   

6. Click on setup instructions, this will generate the API token, configuration and the command to run the Telegraf plugin.
   
7. Execute the command to start the consumer:
   
8. The consumers should be visible on the Solace queue; the Solace queue will have a topic subscription as specified in Step 4.
   
9. Publish data on topic  telegraf/iot/solace
   Get sample data here.
10. Query the influxDB on the bucket created in step 2.
    

Conclusion

With open standards, Solace can integrate with most other products and applications, preventing you from getting locked in to one technology. Mere configurations as in this case is enough to wiretap data from Solace to InfluxDB, saving you the hassle and headaches associated with building a connector using no-code /low-code iPaas.