Grafana Data Source for The Things Network

📝 27 Sep 2021

UPDATE: There’s a better way to connect Grafana to The Things Network, see this

The Things Network is a public global wireless network for IoT devices…

(And it’s free for fair use!)

Grafana is a open source tool for visualising all kinds of real-time data…

(Works on Linux, macOS and Windows)

Can we connect Grafana to The Things Network…

And instantly visualise the Sensor Data from our IoT Devices?

Visualising The Things Network Sensor Data with Grafana

Today we shall experiment with a custom MQTT Data Source for Grafana that will stream real-time Sensor Data from The Things Network.

Wait… We’re streaming the Sensor Data without storing it?

Yep this streaming setup for Grafana requires fewer components because it doesn’t store the data.

But it has limitations, which we’ll discuss shortly.

(This is work-in-progress, some spot may get rough. And please pardon my ghastly GoLang 🙏)

Grafana visualising Sensor Data from The Things Network

Grafana visualising Sensor Data from The Things Network

§1 Configure The Things Network MQTT

Previously we have configured our IoT Device in The Things Network…

“The Things Network on PineDio Stack BL604 RISC-V Board”

Now we enable the MQTT Server in The Things Network by clicking…

Applications → (Your Application) → Integrations → MQTT

Configure The Things Network MQTT Server

Click “Generate New API Key” and copy the values for…

Public Address

(We won’t be using the Public TLS Address since our Data Source doesn’t support TLS)
Username
Password

(This is the only time we can see the password. Don’t forget to copy it!)

We’ll use the values in the next chapter.

§1.1 Test The Things Network MQTT

To test the MQTT Server at The Things Network, enter this command…

## Change au1.cloud.thethings.network to our MQTT Public Address
## Change luppy-application@ttn to our MQTT Username
mosquitto_sub -h au1.cloud.thethings.network -t "#" -u "luppy-application@ttn" -P "YOUR_API_KEY" -d

MQTT JSON Messages will appear whenever our IoT Device joins the network or transmits data.

(See sample MQTT Log)

(More about The Things Network MQTT)

§2 Configure Grafana Data Source

Let’s add and configure our Grafana Data Source for The Things Network…

lupyuen/the-things-network-datasource

Follow these instructions to install Grafana and the Data Source for The Things Network…

“Install Grafana and Data Source”

In Grafana, click the left menu bar…

Configuration → Data Sources

Click “Add Data Source”

Add Data Source

Look for “The Things Network” and click “Select”

Data Source for The Things Network

Fill in the values copied from our MQTT Server at The Things Network…

Name: Use the default
Host: Public Address of our MQTT Server
Port: 1883
Username: Username for our MQTT Server
Password: Password for our MQTT Server

Configuring the Grafana Data Source for The Things Network

Click “Save & Test”

We should see the message “MQTT Connected”…

MQTT Connected

§3 Grafana Dashboard

Let’s render the Sensor Data from The Things Network in Grafana!

Click “Add Panel” (top right)

Click “Add An Empty Panel”

Add Panel

Set the Data Source to “The Things Network”

Set the Topic to “all”

Set Data Source and Topic

Click “Apply” (top right)

Our Sensor Data appears on the Grafana Dashboard!

The Things Network Dashboard

§3.1 View Raw Data

To see the Raw Sensor Data as a table…

Click “Panel Title” and “Edit”

Click “Table View”

Table View

§3.2 Filter Data

To filter the Sensor Data that will be rendered in the dashboard…

Click the “Transform” Tab

Select “Filter Data By Values”

Set the Conditions and click “Apply”

Filter Data

The above filter matches the Device ID with the Regular Expression…

eui-70b3.*

Which means that only Device IDs starting with “eui-70b3” will be rendered.

§4 CBOR: Concise Binary Object Representation

The Sensor Data we’ve seen earlier… Was it encoded as JSON?

Not quite. If we encode the Sensor Data as JSON…

{ 
  "t": 1234, 
  "l": 2345 
}

We’ll need 19 bytes to transmit the Sensor Data.

Which might be too much for The Things Network.

What are the message limits for The Things Network?

The Things Network is Free for Fair Use, with limits on the size of messages and how often we may send them.

If we expect to send 10 messages per hour, our Message Payload should not exceed 12 bytes.

Thus our Message Payload is too small for JSON! We’ll encode with CBOR instead.

(More about The Things Network limits)

What is CBOR?

Concise Binary Object Representation (CBOR) works like a binary, compressed form of JSON…

Sensor Data encoded as CBOR

Our Data Source for The Things Network assumes that the Message Payload is encoded with CBOR.

To experiment with CBOR, try the CBOR Playground…

CBOR Playground

(More about CBOR implementations)

§4.1 Decode CBOR in Go

Our Data Source calls this Go Library to decode CBOR Message Payloads…

fxamacker/cbor

We call the library like so…

import "github.com/fxamacker/cbor/v2"

//  Encoded CBOR payload for { "t": 1234, "l": 2345 }
payload := []byte{0xa2, 0x61, 0x74, 0x19, 0x04, 0xd2, 0x61, 0x6c, 0x19, 0x09, 0x29}

//  Decode CBOR payload to a map of string → interface{}
var body map[string]interface{}
err := cbor.Unmarshal(payload, &body)

//  Shows: map[l:2345 t:1234]
if err == nil {
  fmt.Printf("%v\n", body)
}

Later we’ll see the decoding logic in our Data Source.

§4.2 Message Payload

Where is the Message Payload?

Our CBOR Message Payload is embedded deep inside the MQTT Message from The Things Network…

{
  "end_device_ids": {
    "device_id": "eui-YOUR_DEVICE_EUI",
    "application_ids": {
      "application_id": "YOUR_APPLICATION_ID"
    },
    "dev_eui":  "YOUR_DEVICE_EUI",
    "join_eui": "0000000000000000",
    "dev_addr": "YOUR_DEV_ADDR"
  },
  "correlation_ids": [ ... ],
  "received_at":     "2021-09-25T13:46:17.083379844Z",
  "uplink_message":  {
    "session_key_id":  "YOUR_SESSION_KEY_ID",
    "f_port":          2,
    "frm_payload":     "omF0GQTSYWwZA+g=",

(Source)

frm_payload contains our CBOR Message Payload, encoded with Base64.

We’ll watch the extraction of the Message Payload in a while.

§5 MQTT Integration

Let’s look at the Go Source Code for our Data Source…

lupyuen/the-things-network-datasource

Our Data Source is modded from the MQTT Data Source for Grafana…

grafana/mqtt-datasource

Here’s how our Data Source subscribes to MQTT Topics: pkg/mqtt/client.go

//  Name of our default topic
//  TODO: Support other topics
const defaultTopicName = "all"

//  We will subscribe to all MQTT topics
//  TODO: Support other topics
const defaultTopicMQTT = "#"

//  Subscribe to the topic.
//  We assume that the Topic Name is "all".
func (c *Client) Subscribe(t string) {
  //  If Topic Name already exists, quit
  if _, ok := c.topics.Load(t); ok {
    return
  }

  //  Create the topic: "all"
  topic := Topic{
    path: t,
  }
  c.topics.Store(&topic)

  //  Subscribe to all MQTT Topics: "#". TODO: Support other topics.
  //  Previously: c.client.Subscribe(t, 0, c.HandleMessage)
  c.client.Subscribe(defaultTopicMQTT, 0, c.HandleMessage)
}

Sorry this code looks wonky…

(Because I haven’t decided which topics we should support)

We subscribe to all MQTT Topics: “#”

This means we will receive MQTT Messages for all our devices.

Also we will receive all types of messages, including the “Join Network” messages.
However “#” is not a valid Topic Name in our Grafana Data Source. (Not sure why it fails)

Hence we use “all” as the substitute Topic Name for the “#” MQTT Topic.

§5.2 Receive MQTT Messages

Incoming MQTT Messages are handled by this function: pkg/mqtt/client.go

//  Handle incoming MQTT messages
func (c *Client) HandleMessage(_ paho.Client, msg paho.Message) {
  //  Assume that the topic is "all". TODO: Support other topics.
  //  Previously: topic, ok := c.topics.Load(msg.Topic())
  topic, ok := c.topics.Load(defaultTopicName)
  if !ok {  //  Topic not found, quit
    return
  }

  //  Compose message
  message := Message{
    Timestamp: time.Now(),
    Value:     string(msg.Payload()),
  }

We begin by composing a Message object that will be processed later.

Sorry again for this horrible hack: We reject messages without a valid CBOR Base64 Payload…

  //  TODO: Fix this hack to reject messages without a valid CBOR Base64 Payload.
  //  CBOR Payloads must begin with a CBOR Map: 0xA1 or 0xA2 or 0xA3 or ...
  //  So the Base64 Encoding must begin with "o" or "p" or "q" or ...
  //  We stop at 0xB1 (Base64 "s") because we assume LoRaWAN Payloads will be under 50 bytes.
  //  Join Messages don't have a payload and will also be rejected.
  const frm_payload = "\"frm_payload\":\""
  if !strings.Contains(message.Value, frm_payload+"o") &&
    !strings.Contains(message.Value, frm_payload+"p") &&
    !strings.Contains(message.Value, frm_payload+"q") &&
    !strings.Contains(message.Value, frm_payload+"r") &&
    !strings.Contains(message.Value, frm_payload+"s") {
    log.DefaultLogger.Debug(fmt.Sprintf("Missing or invalid payload: %s", message.Value))
    return
  }

Next we store the Message object (keeping the most recent 1,000 messages)…

  //  Store message for query
  topic.messages = append(topic.messages, message)

  //  Limit the size of the retained messages
  if len(topic.messages) > 1000 {
    topic.messages = topic.messages[1:]
  }

  //  Update the topic messages
  c.topics.Store(topic)

Then we stream the message to a channel…

  //  Stream message to topic "all". TODO: Support other topics.
  //  Previously: streamMessage := StreamMessage{Topic: msg.Topic(), Value: string(msg.Payload())}
  streamMessage := StreamMessage{Topic: defaultTopicName, Value: string(msg.Payload())}
  select {
  case c.stream <- streamMessage:
  default:
    //  Don't block if nothing is reading from the channel
  }
}

What happens next? Read on and find out…

§6 Transform MQTT Messages

Our Data Source calls this function to transform the Received MQTT Message into a Grafana Data Frame: pkg/plugin/message.go

//  Transform the array of MQTT Messages (JSON encoded) 
//  into a Grafana Data Frame.
func ToFrame(topic string, messages []mqtt.Message) *data.Frame {
  count := len(messages)
  if count > 0 {
    first := messages[0].Value

    //  JSON Message must begin with "{"
    if strings.HasPrefix(first, "{") {
      return jsonMessagesToFrame(topic, messages)
    }
  }
  //  Omitted: Handle non-JSON messages

The code above forwards the received messages (JSON format) to this function: pkg/plugin/message.go

//  Transform the array of MQTT Messages (JSON encoded) 
//  into a Grafana Data Frame. See sample messages: https://github.com/lupyuen/the-things-network-datasource#mqtt-log
func jsonMessagesToFrame(topic string, messages []mqtt.Message) *data.Frame {
  //  Quit if no messages to transform
  count := len(messages)
  if count == 0 {
    return nil
  }

  //  Transform the first message
  msg := messages[0]

  //  Decode the CBOR payload
  body, err := decodeCborPayload(msg.Value)
  if err != nil {
    return set_error(data.NewFrame(topic), err)
  }

We begin by decoding the CBOR payload for the first message.

(More about this in the next chapter)

Next we construct the Timestamp Field…

  //  Construct the Timestamp field
  timeField := data.NewFieldFromFieldType(data.FieldTypeTime, count)
  timeField.Name = "Time"
  timeField.SetConcrete(0, msg.Timestamp)

We compose the Data Fields (like “t” and “l”) for the first message…

  //  Create a field for each key and set the first value
  keys := make([]string, 0, len(body))
  fields := make(map[string]*data.Field, len(body))

  //  Compose the fields for the first row of the Data Frame
  for key, val := range body {
    //  Get the Data Frame Type for the field
    typ := get_type(val)

    //  Create the field for the first row
    field := data.NewFieldFromFieldType(typ, count)
    field.Name = key
    field.SetConcrete(0, val)
    fields[key] = field
    keys = append(keys, key)
  }
  sort.Strings(keys) // keys stable field order.

(We’ll see “get_type” in the next chapter)

Now we do the same for the remaining messages…

  //  Transform the messages after the first one
  for row, m := range messages {
    //  Skip the first message
    if row == 0 {
      continue
    }

    //  Decode the CBOR payload
    body, err := decodeCborPayload(m.Value)
    if err != nil {  //  Ignore decode errors
      continue
    }

    //  Set the Timestamp for the transformed row
    timeField.SetConcrete(row, m.Timestamp)

    //  Set the fields for the transformed row
    for key, val := range body {
      field, ok := fields[key]
      if ok {
        field.SetConcrete(row, val)
      }
    }
  }

Finally we pack all the Data Fields into a Data Frame…

  //  Construct the Data Frame
  frame := data.NewFrame(topic, timeField)

  //  Append the fields to the Data Frame
  for _, key := range keys {
    frame.Fields = append(frame.Fields, fields[key])
  }
  return frame
}

And we return the transformed Data Frame to Grafana.

Transformation Complete!

How do we handle transformation errors?

We call this function to set the error on the Data Frame, and return it to Grafana: pkg/plugin/message.go

//  Return the Data Frame set to the given error
func set_error(frame *data.Frame, err error) *data.Frame {
  frame.AppendNotices(data.Notice{
    Severity: data.NoticeSeverityError,
    Text:     err.Error(),
  })
  log.DefaultLogger.Debug(err.Error())
  return frame
}

§7 Decode CBOR Payload

We’ve seen the Message Transformation Logic, now we decode the CBOR payload in the MQTT Message: pkg/plugin/message.go

//  Decode the CBOR payload in the JSON message.
//  See sample messages: https://github.com/lupyuen/the-things-network-datasource#mqtt-log
func decodeCborPayload(msg string) (map[string]interface{}, error) {
  //  Deserialise the message doc to a map of string → interface{}
  var doc map[string]interface{}
  err := json.Unmarshal([]byte(msg), &doc)
  if err != nil {
    return nil, err
  }

We start by deserialising the JSON message.

Remember that our Message Payload is located at…

uplink_message → frm_payload

We extract the Message Payload like so…

  //  Get the Uplink Message
  uplink_message, ok := doc["uplink_message"].(map[string]interface{})
  if !ok {
    return nil, errors.New("uplink_message missing")
  }

  //  Get the Payload
  frm_payload, ok := uplink_message["frm_payload"].(string)
  if !ok {
    return nil, errors.New("frm_payload missing")
  }

Message Payload is encoded with Base64, thus we decode it…

  //  Base64 decode the Payload
  payload, err := base64.StdEncoding.DecodeString(frm_payload)
  if err != nil {
    return nil, err
  }

Next we decode the CBOR payload…

  //  Decode CBOR payload to a map of String → interface{}
  var body map[string]interface{}
  err = cbor.Unmarshal(payload, &body)
  if err != nil {
    return nil, err
  }

(Yep we’ve seen this earlier)

To support filtering by Device ID, we extract the Device ID from the MQTT Message…

  //  Add the Device ID to the body: end_device_ids → device_id
  end_device_ids, ok := doc["end_device_ids"].(map[string]interface{})
  if ok {
    device_id, ok := end_device_ids["device_id"].(string)
    if ok {
      body["device_id"] = device_id
    }
  }
  return body, nil
}

Finally we return the decoded CBOR payload.

(Containing “t”, “l” and the Device ID)

§7.1 Convert CBOR Type

Note that we need to specify the types of Data Fields when populating a Grafana Data Frame.

We call this function to map CBOR Types to Grafana Data Field Types: pkg/plugin/message.go

//  Return the Data Frame Type for the CBOR decoded value
func get_type(val interface{}) data.FieldType {
  //  Based on https://github.com/fxamacker/cbor/blob/master/decode.go#L43-L53
  switch v := val.(type) {
  //  CBOR booleans decode to bool.
  case bool:
    return data.FieldTypeBool

  //  CBOR positive integers decode to uint64.
  case uint64:
    return data.FieldTypeNullableUint64

  //  CBOR floating points decode to float64.
  case float64:
    return data.FieldTypeNullableFloat64

  //  CBOR text strings decode to string.
  case string:
    return data.FieldTypeNullableString

§7.2 Testing the Data Source

Testing the MQTT Message Transformation looks painful!

Indeed. That’s why we wrote another Go program to test the transformation by calling jsonMessagesToFrame…

Testing the MQTT Message Transformation

§8 Troubleshooting

If we have problems with the Data Source, enabling Debug Logging might help.

Edit the Grafana Configuration File…

## For Linux:
/usr/share/grafana/conf/defaults.ini

## For macOS:
/usr/local/etc/grafana/grafana.ini

## For Windows:
C:\Program Files\GrafanaLabs\grafana\conf\defaults.ini

Set the Log Level…

[log]
level = debug

Save the file and restart the Grafana Server.

Check the Grafana Log at…

## For Linux:
/var/log/grafana/grafana.log

## For macOS:
/usr/local/var/log/grafana/grafana.log

## For Windows:
C:\Program Files\GrafanaLabs\grafana\data\log\grafana.log

(See sample Grafana Log)

§9 Store Data with Prometheus

Today we have streamed Sensor Data from The Things Network to Grafana over MQTT…

Visualising The Things Network Sensor Data with Grafana

Wait… We’re streaming the Sensor Data without storing it?

Yep this streaming setup for Grafana requires fewer components because it doesn’t store the data.

But this streaming setup has limitations…

What happens when we restart our Grafana Server?

All our Sensor Data is lost!
We’re remembering only the last 1,000 messages.

Which is OK for checking real-time Sensor Data transmitted by our devices…

But not OK for observing Sensor Data trends over time!

Can we store the Sensor Data?

We can store the Sensor Data in Prometheus, the open source Time-Series Data Store…

Storing The Things Network Sensor Data with Prometheus

Grafana supports Prometheus as a Data Source, so pushing our Sensor Data from Prometheus to Grafana is easy.

To ingest MQTT Messages from The Things Network into Prometheus, we need an MQTT Gateway like this…

hikhvar/mqtt2prometheus

This Grafana setup looks more complicated, but it works well for visualising historical Sensor Data. See this…

“Monitor IoT Devices in The Things Network with Prometheus and Grafana”

More about this in the next chapter…

§10 Shortcomings

We’ve seen two ways of connecting Grafana to The Things Network…

Via Grafana Data Source for The Things Network

(As explained in this article)
Via Prometheus with an MQTT Gateway

(See this)

How do they compare?

Well the approach described in this article has some serious shortcomings…

Fixed Payload Format: Only CBOR Payloads are supported

(Prometheus uses a configurable Payload Formatter that supports multiple payload formats)
Fixed MQTT Topic: Only one MQTT Topic is supported (“#”)

(Prometheus supports configurable MQTT Topics)
Fixed Fields: Only specific fields are extracted (Device ID and Sensor Data)

(Prometheus extracts multiple Metrics based on configurable JSON Paths)
No TLS Encryption: Our MQTT Password is transmitted in the clear

(Prometheus supports TLS Encryption for MQTT)
No Persistence: Our Sensor Data disappears when we restart Grafana Server

(Prometheus stores Sensor Data in a Time Series Database)
Custom Coding: Our Data Source requires custom coding. And it’s incomplete.

(Prometheus works without custom coding, except for the Custom Payload Formatter. And a minor patch for TLS Encryption)

Thus Prometheus is the recommended way to connect Grafana to The Things Network…

“Monitor IoT Devices in The Things Network with Prometheus and Grafana”

§11 What’s Next

I hope you enjoyed our exploration today: Streaming Sensor Data from The Things Network to Grafana over MQTT.

In the next article we shall head back to PineDio Stack BL604 and transmit actual Sensor Data to The Things Network, encoded with CBOR.

Many Thanks to my GitHub Sponsors for supporting my work! This article wouldn’t have been possible without your support.

Got a question, comment or suggestion? Create an Issue or submit a Pull Request here…

lupyuen.github.io/src/grafana.md

§12 Notes

This article is the expanded version of this Twitter Thread
What exactly are “t” and “l” in our Sensor Data?
```
{ 
  "t": 1234, 
  "l": 2345 
}
```
“t” and “l” represent our (imaginary) Temperature Sensor and Light Sensor.

We shortened the Field Names to fit the Sensor Data into 11 bytes of CBOR.

With Grafana we can map “t” and “l” to their full names for display.
Why is the temperature transmitted as an integer: 1234?

That’s because floating-point numbers compress poorly with CBOR unless we select the proper encoding.

(Either 3 bytes, 5 bytes or 9 bytes per float. See the next note)

Instead we assume that our integer data has been scaled up 100 times.

(So 1234 actually means 12.34 ºC)

We may configure Grafana to divide our integer data by 100 when rendering the values.
If we’re actually encoding floats in CBOR, how do we select the proper encoding?

The CBOR spec says that there are 3 ways to encode floats…
- IEEE 754 Half-Precision Float (16 bits)
  
  (3.3 significant decimal digits)
- IEEE 754 Single-Precision Float (32 bits)
  
  (6 to 9 significant decimal digits)
- IEEE 754 Double-Precision Float (64 bits)
  
  (15 to 17 significant decimal digits)
What would be the proper encoding for a float (like 12.34) that could range from 0.00 to 99.99?

This means that we need 4 significant decimal digits.

Which is too many for a Half-Precision Float (16 bits), but OK for a Single-Precision Float (32 bits).

Thus we need 5 bytes to encode the float. (Including the CBOR Initial Byte)

(Thanks to @chrysn for highlighting this!)
Is it meaningful to record temperatures that are accurate to 0.01 ºC?

How much accuracy do we need for Sensor Data anyway?

The accuracy for our Sensor Data depends on…
1. Our monitoring requirements, and
2. Accuracy of our sensors
Learn more about Accuracy and Precision of Sensor Data…

“IoT’s Lesser Known Power: “Good Enough” Data Accuracy”
It’s possible to decode CBOR in The Things Network servers with a Payload Formatter.

If we do this, we can skip the CBOR decoding in our Grafana Data Source. See this…

“CBOR Payload Formatter for The Things Network”

§13 Appendix: Install Grafana and Data Source

Here are the steps to install Grafana and our Data Source for The Things Network.

§13.1 Install Grafana

Browse to grafana.com/oss/grafana

Click “Get Grafana → Self-Managed → Download Grafana”
For “Edition” select “OSS”
Click Linux, macOS, Windows, Arm or Docker

(Grafana for Linux works on WSL too)
Follow the instructions to download and install Grafana

For Linux and macOS: Start the Grafana Server

## For Ubuntu and WSL
sudo service grafana-server restart
sudo service grafana-server status

## For macOS
brew services start grafana

To test Grafana, browse to

http://localhost:3000

Username: admin

Password: admin

§13.2 Build Data Source

(Note: Our Data Source uses the Grafana Live Streaming API, please use Grafana version 8.0 or later)

For Windows: Grant Full Control permission to the Users group for the Grafana Plugins Folder…
```
C:\Program Files\GrafanaLabs\grafana\data\plugins
```

Download the Data Source into the Grafana Plugins Folder…

##  For Linux: Need "sudo" to access this folder
cd /var/lib/grafana/plugins

##  For macOS: 
cd /usr/local/var/lib/grafana/plugins

##  For Windows: Need to grant "Full Control" permission to "Users" group for this folder
cd C:\Program Files\GrafanaLabs\grafana\data\plugins

##  Download source files for The Things Network Data Source
git clone --recursive https://github.com/lupyuen/the-things-network-datasource

Install the Build Tools…

Build Tools for Linux (Ubuntu)

Build Tools for macOS

Build Tools for Windows

(More details here)

Build the Data Source…

##  Install the dependencies
cd the-things-network-datasource
yarn install

##  Build the Data Source (React + Go)
yarn build

##  If "mage" is not found, set the PATH
export PATH=$PATH:$GOPATH/bin

(See the Build Log)

If “yarn build” fails on Windows, edit package.json and replace “rm -rf” by “rimraf”

Restart the Grafana Service for the Data Source to load

## For Ubuntu and WSL:
sudo service grafana-server restart
sudo service grafana-server status

## For macOS:
brew services restart grafana

## For Windows: Run this as Administrator
net stop grafana
net start grafana

§13.3 Enable Data Source