Storage

Magistrala currently supports these storage databases for messages:

• PostgreSQL
• Timescale

These storages are activated via docker-compose add-ons.

The <project_root>/docker folder contains an addons directory. This directory is used for various services that are not core to the Magistrala platform but could be used for providing additional features.

In order to run these services, core services, as well as the network from the core composition, should be already running.

Writers

Writers provide an implementation of various message writers. Message writers are services that consume Magistrala messages, transform them to desired format and store them in specific data store. The path of the configuration file can be set using the following environment variables: MG_CASSANDRA_WRITER_CONFIG_PATH, MG_POSTGRES_WRITER_CONFIG_PATH, MG_INFLUX_WRITER_CONFIG_PATH, MG_MONGO_WRITER_CONFIG_PATH and MG_TIMESCALE_WRITER_CONFIG_PATH.

Subscriber config

Each writer can filter messages based on a topics list that is set in the config.toml configuration file's [subscriber] section. This is a different topic namespace from the channel pub/sub topics covered in the Subtopics Section: writers consume from the broker's internal writers/# queue (every message a rule or writer-bound publish routes there), not directly from channels.<channel_id>.... If you want to listen on every writer topic, leave topics at its default ["writers/#"]; otherwise pass a list of more specific filters. Filters are slash-delimited and support the MQTT-style wildcards + (one level) and # (a variable-length trailing suffix), for both NATS and FluxMQ builds. Here is an example:

[subscriber]
topics = ["writers/<domain_id>/c/<channel_id>/bedroom/temperature", "writers/<domain_id>/c/<channel_id>/bedroom/humidity"]

Regarding the Subtopics Section in the messaging page, a message published to m/<domain_id>/c/<channel_id>/bedroom/temperature is republished to writers under writers/<domain_id>/c/<channel_id>/bedroom/temperature - substitute your actual domain and channel IDs, or use + in their place to match across all of them (e.g. writers/+/c/+/bedroom/temperature).

Transformer config

There are two types of transformers: SenML and JSON. The transformer type is set in configuration file.

For SenML transformer, supported message payload formats are SenML+CBOR and SenML+JSON. They are configurable over content_type field in the [transformer] section and expect application/senml+json or application/senml+cbor formats. Here is an example:

[transformer]
format = "senml"
content_type = "application/senml+json"

Usually, the payload of the IoT message contains message time. It can be in different formats (like base time and record time in the case of SenML) and the message field can be under the arbitrary key. Usually, we would want to map that time to the Magistrala Message field Created and for that reason, we need to configure the Transformer to be able to read the field, parse it using proper format and location (if devices time is different than the service time), and map it to Magistrala Message.

For JSON transformer you can configure time_fields in the [transformer] section to use arbitrary fields from the JSON message payload as timestamp. time_fields is represented by an array of objects with fields field_name, field_format and location that represent respectively the name of the JSON key to use as timestamp, the time format to use for the field value and the time location. Here is an example:

[transformer]
format = "json"
time_fields = [{ field_name = "seconds_key", field_format = "unix",    location = "UTC"},
               { field_name = "millis_key",  field_format = "unix_ms", location = "UTC"},
               { field_name = "micros_key",  field_format = "unix_us", location = "UTC"},
               { field_name = "nanos_key",   field_format = "unix_ns", location = "UTC"}]

JSON transformer can be used for any JSON payload. For the messages that contain JSON array as the root element, JSON Transformer does normalization of the data: it creates a separate JSON message for each JSON object in the root. In order to be processed and stored properly, JSON messages need to contain message format information. For the sake of simplicity, nested JSON objects are flatten to a single JSON object in InfluxDB, using composite keys separated by the / separator. This implies that the separator character (/) is not allowed in the JSON object key while using InfluxDB. Apart from InfluxDB, separator character (/) usage in the JSON object key is permitted, since other Writer types do not flat the nested JSON objects. For example, the following JSON object:

{
  "name": "name",
  "id": 8659456789564231564,
  "in": 3.145,
  "alarm": true,
  "ts": 1571259850000,
  "d": {
    "tmp": 2.564,
    "hmd": 87,
    "loc": {
      "x": 1,
      "y": 2
    }
  }
}

for InfluxDB will be transformed to:

{
  "name": "name",
  "id": 8659456789564231564,
  "in": 3.145,
  "alarm": true,
  "ts": 1571259850000,
  "d/tmp": 2.564,
  "d/hmd": 87,
  "d/loc/x": 1,
  "d/loc/y": 2
}

while for other Writers it will preserve its original format.

The message format is stored in the subtopic. It's the last part of the subtopic. In the example:

http://localhost:8008/channels/<channelID>/messages/home/temperature/myFormat

the message format is myFormat. It can be any valid subtopic name, JSON transformer is format-agnostic. The format is used by the JSON message consumers so that they can process the message properly. If the format is not present (i.e. message subtopic is empty), JSON Transformer will report an error. Message writers will store the message(s) in the table/collection/measurement (depending on the underlying database) with the name of the format (which in the example is myFormat). Magistrala writers will try to save any format received (whether it will be successful depends on the writer implementation and the underlying database), but it's recommended that publishers don't send different formats to the same subtopic.

InfluxDB, InfluxDB Writer

From the project root execute the following command:

docker-compose -f docker/addons/influxdb-writer/docker-compose.yml up -d

This will install and start:

• InfluxDB - time series database
• InfluxDB writer - message repository implementation for InfluxDB

Those new services will take some additional ports:

• 8086 by InfluxDB
• 9006 by InfluxDB writer service

To access Influx-UI, navigate to http://localhost:8086 and login with: magistrala, password: magistrala

Cassandra and Cassandra Writer

./docker/addons/cassandra-writer/init.sh

Please note that Cassandra may not be suitable for your testing environment because of its high system requirements.

MongoDB and MongoDB Writer

docker-compose -f docker/addons/mongodb-writer/docker-compose.yml up -d

MongoDB default port (27017) is exposed, so you can use various tools for database inspection and data visualization.

PostgreSQL and PostgreSQL Writer

docker-compose -f docker/addons/postgres-writer/docker-compose.yml up -d

Postgres default port (5432) is exposed, so you can use various tools for database inspection and data visualization.

Timescale and Timescale Writer

docker-compose -f docker/addons/timescale-writer/docker-compose.yml up -d

Timescale default port (5432) is exposed, so you can use various tools for database inspection and data visualization.

By default, the timescale-writer add-on's config.toml has no [transformer] section at all, which means it silently falls back to the SenML transformer (see Transformer config above) - it will not persist JSON messages, including ones saved via the Rules Engine's Internal DB (JSON) output.

Running SenML and JSON Writers Together

A single writer process is locked to whichever transformer format its config.toml specifies - it can't switch between SenML and JSON per message. To store both formats at once, run a second timescale-writer instance dedicated to JSON, alongside the default SenML one:

1. Copy the timescale-writer add-on's compose service (or the MG_TIMESCALE_WRITER_CONFIG_PATH config it loads) to a second instance, e.g. timescale-writer-json.

2. Give the new instance a config.toml with the JSON transformer. The [subscriber] section's topics key takes slash-delimited MQTT-style filters (+ for one level, # for a variable-length trailing suffix) - the same syntax the writer add-ons already use, for both NATS and FluxMQ builds:

   [subscriber]
   topics = ["writers/+/c/+/json"]
   
   [transformer]
   format = "JSON"

   This particular filter matches messages whose subtopic is exactly json - the case where the rule's input channel itself has no subtopic, which the Rules Engine's Internal DB (JSON) output produces by default. If your rule's input channel does have a subtopic, the JSON output appends /json to it instead (e.g. an input subtopic of devices/dev-1 becomes devices/dev-1/json), which needs an extra + per subtopic level to match, e.g. writers/+/c/+/+/json.

3. Leave the existing SenML timescale-writer instance's config.toml and topics filter unchanged (it defaults to ["writers/#"], matching every writer-bound message regardless of format).

# is a trailing wildcard only - it can't express "anything ending in .../json" regardless of depth in a single filter, so a topic tree with subtopics of varying depth needs either one +-based filter per depth you expect, or falling back to the broader ["writers/#"] catch-all. With that catch-all, the JSON writer attempts to parse every message it receives as JSON: messages whose subtopic doesn't end in a json segment still get stored, just under a table named after whatever their own last subtopic segment is, rather than the json table - harmless clutter, not data corruption, but worth scoping down with the +-based filter above once you know your deployment's topic structure.

As explained above, the JSON transformer derives its destination table from the last segment of the message's subtopic - so messages published (or, from a rule, saved) with a subtopic ending in json land in a table literally named json, matching what a channel's Message Views JSON-format view queries for.

Readers

Readers provide an implementation of various message readers. Message readers are services that consume normalized (in SenML format) Magistrala messages from data storage and opens HTTP API for message consumption. Installing corresponding writer before reader is implied.

Each of the Reader services exposes the same HTTP API for fetching messages on its default port.

To read sent messages on channel with id channel_id you should send GET request to /channels/<channel_id>/messages with client access token in Authorization header. That client must be connected to channel with channel_id

Response should look like this:

HTTP/1.1 200 OK
Content-Type: application/json
Date: Tue, 18 Sep 2018 18:56:19 GMT
Content-Length: 228

{
    "messages": [
        {
            "Channel": 1,
            "Publisher": 2,
            "Protocol": "mqtt",
            "Name": "name:voltage",
            "Unit": "V",
            "Value": 5.6,
            "Time": 48.56
        },
        {
            "Channel": 1,
            "Publisher": 2,
            "Protocol": "mqtt",
            "Name": "name:temperature",
            "Unit": "C",
            "Value": 24.3,
            "Time": 48.56
        }
    ]
}

Note that you will receive only those messages that were sent by authorization token's owner. You can specify offset and limit parameters in order to fetch specific group of messages. An example of HTTP request looks like:

curl -s -S -i  -H "Authorization: Client <client_secret>" http://localhost:<service_port>/channels/<channel_id>/messages?offset=0&limit=5&format=<subtopic>

If you don't provide offset and limit parameters, default values will be used instead: 0 for offset and 10 for limit. The format parameter indicates the last subtopic of the message. As indicated under the Writers section, the message format is stored in the subtopic as the last part of the subtopic. In the example:

http://localhost:<service_port>/channels/<channelID>/messages/home/temperature/myFormat

the message format is myFormat and the value for format=<subtopic> is format=myFormat.

InfluxDB Reader

To start InfluxDB reader, execute the following command:

docker-compose -f docker/addons/influxdb-reader/docker-compose.yml up -d

Cassandra Reader

To start Cassandra reader, execute the following command:

docker-compose -f docker/addons/cassandra-reader/docker-compose.yml up -d

MongoDB Reader

To start MongoDB reader, execute the following command:

docker-compose -f docker/addons/mongodb-reader/docker-compose.yml up -d

PostgreSQL Reader

To start PostgreSQL reader, execute the following command:

docker-compose -f docker/addons/postgres-reader/docker-compose.yml up -d

Timescale Reader

To start Timescale reader, execute the following command:

docker-compose -f docker/addons/timescale-reader/docker-compose.yml up -d