Influx Database Writer Service

The Influx DB database is an open-source time series database available under the MIT open source license or as a SaaS cloud offering. A commercially supported “hardened” version is available as an “enterprise” offering. The database is a time-series database that is optimised for storage and retrieval of data points in their associated time order. The database is written in Google’s open-source Go programming language.

Influx has gathered around it a number of other open-source implementations to create a little ecosystem - often referred to as the TICK stack from the initial letters of those components:-

Central to the communication between the mediation devices that monitor and control the elements of the microgrid (the EP Platform) and the cloud-based components that configure and report upon them are a series of MQTT Brokers. The brokers provide a link for JSON payloads with a well defined Quality of Service metric.

MQTT is a Client Server publish/subscribe messaging transport protocol. It is a light weight, open, simple, and designed so as to be easy to implement. The features were influenced by the features of IBM MQ Series but the protocols and service are original. These characteristics make it ideal for use in many situations, including constrained environments such as for communication in Machine to Machine (M2M) and Internet of Things (IoT) contexts where a small code footprint is required and/or network bandwidth is at a premium.

The device instarces which form part of the EP platform publish their status and data ‘signals’ to the cloud as published messages against a series of predefined topics that the cloud components can monitor as subscriptions to those topics.

The inflix-writer is a custom JavaScript module that is designed to run inside a NodeJS container and implements a “micro service” that receives messages from the MQTT broker, formats them into batches and writes them to the InfluxDB database. To do this it uses the standard InfluvDB APIs. In particular the write to DB API is used to create time series data in one of the “buckets” associated with the InfluxDB instance. The API protocol populates data values from by the influx-writer service which must supply:

The influx-writer is an event driven continuous service primarily run under a NodeJS instance that exposes no services other than an optional HealthCheck service but acts as a daemon reading published MQTT messages for specific topics and buffering them for write to the InfluxDB.

The influx DB is in the process of significant architectural change from versions 1.x to 2.x. Version 2.0 is currently at release candidate status and provides in a single binary multi-tenanted time series database, UI and dashboarding tools, background processing and monitoring agent, all behind a single API. It provides an entire TICK stack in one deployment with shareable dashboards, alerts and queries optimized for specific workloads and solutions.

Although elements of the Flux query language were available in InfluxDB version 1.x the newer release offers a language designed for data scripting, ETL, monitoring and alerting.

The influx-writer can support both versions of InfluxDB by configuring the INFLUX_DB_V2 environment variable to true.

The JavaScript code can be found in the BitBucket Repository at:-

git clone https://<YOUR_ID>@bitbucket.org/smc-dev/influx-writer.git

The influx-writer has a dependency on the following other JavaScript modules which also form part of the SMC Codebase:-

These dependencies create indirect dependencies on the following other SMC modules

The influx-writer makes use of a number of common JavaScript framework/library packages that will be installed into the NodeJS environment by the package manager (npm or yarn). The dependencies are illustrated in the structure diagram below. These external dependencies include:-

The i*nflux.js* file uses the Joi schema validation package to assemble and check a series of parameters that are needed to configure an instance of the influx-writer. The parameters are usually set via environment variables which make the mechanism suitable for both local implementations and those hosted in docker.

The queueConfig module evaluates environment variable to set up the key parameters for the buffered writes of data retrieved from the MQTT broker to the Influx DB. The recognised parameter variables are presented in the table below.

A number of environment variables are used by the*_ edge-service-wrapper _*modules that form the dependencies for the influx-writer. These variables tend to control access to key components such as the MQTT broker and the SMTP mail server. All services that incorporate the wrapper as a dependency have their behaviours influenced by the presence, or otherwise, of these environment variables.

Subscription setup happens during the start-up of the influx-writer. A short time sequence diagram below illustrates the steps in setting up the subscriptions. The process follows the following steps:-

We described above the MQTT topics which the influx-writer subscribes to. The mechanism used to respond to messages that are published on these topics is to use a set of asynchronous listeners that write to global property storage. The implementation of these listeners is wrapped in the DTO and DAO implementations provided in the @dgcsdev/edge-schemas module that forms a dependency for the influx-writer. The global storage uses simple Set() and Map() object instances to carry key/value pairs. A series of asynchronous subscriptions are set up for each the topics:-

Subscriptions listen for the events error, stop and data on each topic and attach a call back function or lambda to the event processing. For error and stop events the call back actions are broadly equivalent but the data function is specific to the topic data. For example, the function for the $online topic is mapped to a call back function on the DeviceSubscriber DTO Class which treats the values received on the topic as a tri-state “boolean” , true, false null. The list of subscribers

As the subscribers receive MQTT publish messages these are matched to the subscriber class in the table above and instances of the DTOs created and saved in a global properties store associated with the MQTT Broker client object.

The sequence diagram above illustrates the logical view of what happens in this main loop, in reality a loop not shown here is scheduling each of the callback functions off LIFO queues that hold the list of event handlers for asynchronous functions and timers.

Each point is represented by a JSON object the contains the following fields:-

The writable message is passed to the influxConnector when the listener calls writePoint()[*1.5]. The *influxConnector adds the data point to the pointsBuffer by alling its *addPoint() *function [1.6]

An Interval timer is set up with a periodicity that is configured using the WRITE_INTERVAL configuration variable (see #heading=h.slum7oe5qozb[Configuration]). When the timer interval has expired and there are no higher priority callbacks to service the NodeJS main loop will call influxConnector tick() function [1.7]. The function checks that the elapsed time expired is greater than the configured interval[1.8] and then calls the *requestBatch() *function supplied by the points buffer[1.9]

The data points are batched into a block size that is set by the BUFFER_SIZE configuration variable. The influxConnector function postPoints() is then used to create[1.10] a and HTTP request that holds:-

This HTTP Post will write the data to the Influx Datrabase[1.11]

For test purposes it is possible to configure the influxConnector to read MQTT incoming events, format them but prevent the write to the Influx DB. To achieve this set the NO_UPDATES configuration variable.

http://localhost:8444/