MQTTで受け取ったデータを可視化する (emqx、Telegraf、influxdb、grafana)

やっていること

さまざまなメトリック（データの指標や測定値）をMQTTプロトコルを使って送信し、それをリアルタイムでグラフ化するシステムを構築しています。MQTTは、IoTデバイスなど軽量なネットワーク環境でも効率的に通信できるプロトコルで、多くのメトリックを効率的に扱うのに最適です。
このシステムを持ち運び可能にするため、環境全体を仮想マシン上で構築しました。USBメモリさえあれば、どこでも同じ環境でメトリックの収集や可視化ができるようになっています。

※2024年10月26日 誤字を修正しました。
※2024年11月09日 gpasswdコマンドでログインしているユーザ名を指定するよう修正しました。

環境

• OS : Windows 11 Pro
• MQTTX 1.10.1
• QEMU 20240815版

仮想環境内のツール類は以下です。

• 仮想OS : Ubuntu 24.04
• Docker 25.0.4
• emqx 5.7.2
• telegraf 1.31.3
• InfluxDB 2.7.10
• Grafana 11.1.4

LinuxやWLS上のDockerで構築する場合は、QEMUの箇所は読み飛ばして問題ないです。

ツールの説明

各ツールがどのような役割を行っているか説明します。
私の理解なので、間違っていたら生暖く見守ってください。

• MQTTX
  MQTT Publisherとして動作する
  要は、MQTTクライアント
  MQTT Brokerへメッセージを送信する
• EMQX
  MQTT Brokerとして動作する
  MQTT Publisherからメッセージを受取り、MQTT Subscriberへメッセージを送信する
• telegraf
  MQTT Subscriberとして動作する
  MQTT Publisherからメッセージを受取り、InfluxDBへデータを送信する
• InfluxDB
  時系列データに特化したデータベース
• Grafana
  InfluxDBのデータを読み取り、可視化する

接続図

telegrafは、emqxへ接続してデータを取得しています。
Grafanaは、influxdbへ接続してデータを取得しています。
そのため矢印が上向きになっていますが、データは上から下へ流れています。

これからやること

1. QEMUをインストールして仮想環境を用意する
2. 仮想環境にUbuntuをインストールする
3. Ubuntu環境にDockerを構築する
4. Dockerでemqx、telegraf、InfluxDB、Grafanaを起動する
5. MQTTXでデータを送る
6. Grafanaで可視化、ワーイ

QEMUのインストール

下記URLからQEMUのインストールファイルをダウンロードします。

QEMU
アクセス後、「Download」をクリックし、「Windows」をクリックします。
「64-bit」のリンクをクリックします。
「2024」をクリックします。
今回は「qemu-w64-setup-20240815.exe」をダウンロードしました。

ダウンロードしたファイルを実行してインストールします。
インストールパスは、「F:\qemu」などを指定します。

※USBメモリを指定すると、USB上に環境を作成して、持ち運べるようになります。

Ubuntuのインストール

下記URLからUbuntuのISOファイルをダウンロードします。
Ubuntu
今回は「ubuntu-24.04-live-server-amd64.iso」をダウンロードしています。

QEMUを起動します。
その際にUbuntuのISOファイル指定とインストール用の10GBのディスク領域作成しています。
ダウンロードしたUbuntuのイメージをマウントして仮想マシンを起動しています。

コマンドプロンプトで下記のコマンドを実行します。

> f:
> cd qemu
> qemu-img create -f qcow2 ubuntu24.img 10G
> qemu-system-x86_64.exe -m 1G -smp 2 -boot order=dc -hda ubuntu24.img -cdrom "f:\ubuntu-24.04-live-server-amd64.iso"

QEMUが起動すると、Windowが表示され、Ubuntuのイメージがブートします。
そのままUbuntuをインストールします。

QEMUのWindowからカーソルを放したい場合は、「Ctrl]+[Alt]+[g]を押します

Ubuntuの場合、インストール中にsshのインストールの可否を聞かれます。
インストールするとTeraTerm等でアクセスが出来るようになり便利です。

インストールが完了したら、以下の手順で再起動します。

QEMUで [Ctrl]+[Alt]+[2]を押してQEMUモニタに入ります。
下記コマンドを実行して、isoをマウントしているデバイスを確認します。

(qemu) info block

確認したデバイスをアンマウントします。

(qemu) eject ide1-cd0
(qemu) info block  # マウントされていないことを確認する

QEMUで [Ctrl]+[Alt]+[1]を押て、元の画面に戻ります。
再起動します。

このままだと、仮想マシンへの22ポートが空いていないため、sshでアクセスできません。
ポートを開けた状態で起動するために、再起動後、ログインしたら一旦終了します。

sudo poweroff

起動用バッチファイルを作成します。
各ツールが使用するポートを開放しているため長くなっています。

qemu_start.bat

.\qemu-system-x86_64.exe -k ja -display sdl -rtc clock=vm,base=utc -smp 2 -m 2048 -netdev user,id=n1,hostfwd=tcp::22-:22,hostfwd=tcp::1883-:1883,hostfwd=tcp::8883-:8883,hostfwd=tcp::8083-:8083,hostfwd=tcp::8084-:8084,hostfwd=tcp::18083-:18083,hostfwd=tcp::8086-:8086,hostfwd=udp::8092-:8092,hostfwd=udp::8125-:8125,hostfwd=tcp::8094-:8094,hostfwd=tcp::3000-:3000 -device virtio-net-pci,netdev=n1 -hda ubuntu24.img

バッチファイルを実行して起動します。
起動後、sshで接続が可能になります。
接続情報は、下記になります。

接続先	localhost
ポート	TCP 22

Dockerのインストール

emqx、telegraf、InfluxDB、Grafanaは、Dockerコンテナで動作させます。
まずは、Dockerをインストールします。

UbuntuにDockerをインストール

Ubuntuにアクセスして、下記のコマンドを実行し、Dockerをインストールします。

sudo apt update
sudo apt install ca-certificates curl vim
sudo install -m 0755 -d /etc/apt/keyrings
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
sudo chmod a+r /etc/apt/keyrings/docker.asc
echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/ubuntu $(. /etc/os-release && echo "$VERSION_CODENAME") stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
sudo apt update
sudo apt install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
sudo gpasswd -a `logname` docker

ここで、一度ログインし直す必要があります。

Dockerのテスト

hello-worldを表示するコンテナを起動します。

docker run hello-world

下記のhello-worldが表示されたらOKです。

Unable to find image 'hello-world:latest' locally
latest: Pulling from library/hello-world
c1ec31eb5944: Pull complete
Digest: sha256:d211f485f2dd1dee407a80973c8f129f00d54604d2c90732e8e320e5038a0348
Status: Downloaded newer image for hello-world:latest

Hello from Docker!
This message shows that your installation appears to be working correctly.

To generate this message, Docker took the following steps:
 1. The Docker client contacted the Docker daemon.
 2. The Docker daemon pulled the "hello-world" image from the Docker Hub.
    (amd64)
 3. The Docker daemon created a new container from that image which runs the
    executable that produces the output you are currently reading.
 4. The Docker daemon streamed that output to the Docker client, which sent it
    to your terminal.

To try something more ambitious, you can run an Ubuntu container with:
 $ docker run -it ubuntu bash

Share images, automate workflows, and more with a free Docker ID:
 https://hub.docker.com/

For more examples and ideas, visit:
 https://docs.docker.com/get-started/

Docker ネットワークの作成

docker network create --subnet=172.16.1.0/24 mqtt-network

出来ていることを確認します。

docker network ls

準備

作成するディレクトリ構成は下記になります。

mqtt/
│   docker-compose.yml
│   telegraf.conf    
│
└── emqx/
│      data/
│      log/
│   
└── grafana/

必要なディレクトリを作成します。

mkdir mqtt
cd mqtt
mkdir -p emqx/{data,log}
mkdir grafana

telegraf用のコンフィグファイルを作成します。

telegraf.conf

[global_tags]
[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = "0s"
  hostname = "telegraf_001"
  omit_hostname = false
[[outputs.influxdb_v2]]
   urls = ["http://influxdb:8086"]
   token = "influx-admin-token"
   organization = "influxdb-org"
   bucket = "influxdb-bucket"
[[inputs.mqtt_consumer]]
   servers = ["tcp://emqx:1883"]
   topics = [
     "mqtt-topics/#",
   ]
    client_id = "telegraf_001"
    username = "emqx-user"
    password = "emqx-user-password"

Docker-composeの設定ファイルを作成します。

docker-compose.yml

services:
  influxdb:
    image: influxdb:2.7.10
    container_name: influxdb
    hostname: influxdb
    ports:
      - "8086:8086"
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: influx-user
      DOCKER_INFLUXDB_INIT_PASSWORD: influx-user-password
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: influx-admin-token
      DOCKER_INFLUXDB_INIT_ORG: influxdb-org
      DOCKER_INFLUXDB_INIT_BUCKET: influxdb-bucket
    volumes:
      - type: volume
        source: influxdb-data
        target: /var/lib/influxdb2
      - type: volume
        source: influxdb-config
        target: /etc/influxdb2
    networks:
      mqtt-network:
        ipv4_address: 172.16.1.30
        aliases:
          - influxdb.local.com

  grafana:
    image: grafana/grafana-oss:11.1.4
    container_name: grafana
    hostname: grafana
    ports:
      - "3000:3000"
    volumes:
      - ./grafana:/var/lib/grafana
    user: "0:0"
    networks:
      mqtt-network:
        ipv4_address: 172.16.1.40
        aliases:
          - grafana.local.com
    depends_on:
      - influxdb

  emqx:
    image: emqx:5.7.2
    container_name: emqx
    ports:
      - "1883:1883"
      - "8083:8083"
      - "8084:8084"
      - "8883:8883"
      - "18083:18083"
    volumes:
      - ./emqx/data:/opt/emqx/data
      - ./emqx/log:/opt/emqx/log
    networks:
      mqtt-network:
        ipv4_address: 172.16.1.10
        aliases:
          - emqx.local.com
    depends_on:
      - influxdb

  telegraf:
    image: telegraf:1.31.3
    container_name: telegraf
    ports:
      - "8092:8092/udp"
      - "8125:8125/udp"
      - "8094:8094/tcp"
    volumes:
      - ./telegraf.conf:/etc/telegraf/telegraf.conf
    networks:
      mqtt-network:
        ipv4_address: 172.16.1.20
        aliases:
          -  telegraf.local.com
    depends_on:
      - emqx

volumes:
  influxdb-data:
  influxdb-config:

networks:
  mqtt-network:
    name: mqtt-network
    external: true
    ipam:
      driver: default
      config:
        - subnet: 172.16.1.0/24

コンテナの起動

下記コマンドを実行します。

docker compose up -d

下記コマンドで起動を確認します。

$ docker compose ps

telegrafコンテナだけはログを見て接続を確認してください。
コンテナが起動していても、うまくアクセス出来ていないときがあります。

docker compose logs telegraf

正しくアクセスしているときは、下記のように表示されます。

[inputs.mqtt_consumer] Connected [tcp://emqx:1883]

アクセス出来ていないときは、下記のように表示されます。

[inputs.mqtt_consumer] Connected [tcp://emqx:1883]
[inputs.mqtt_consumer] Error in plugin: connection lost: EOF

アクセス出来ていないときは、コンテナを再起動します。

docker compose restart telegraf

EMQXの設定

EMQXのダッシュボードから設定を行います。

ブラウザから下記へアクセスします。
http://localhost:18083/

デフォルトのログイン情報です。

ユーザー名	admin
パスワード	public

初回ログイン時はパスワードの変更が必要です

左のメニューから「Access Control」-「Authentication」を選択します。
「Create」ボタンをクリックします。
「Password-Based」を選択した状態で、「Next」ボタンをクリックします。
「Built-in Database」を選択した状態で、「Next」ボタンをクリックします。
UserID Type、Password Hash、Salt Positionはデフォルトのまま、「Create」をクリックします。

「Paasword-Based」というバックエンドが作成されています。
作成されたバックエンドの「Users」をクリックします。
「+」ボタンをクリックします。

下記を設定します。

項目	設定値
Username	emqx-user
Password	emqx-user-password

「Save」ボタンをクリックします。
「emqx-user」が作成されていることを確認します。

InfluxDBの設定

InfluxDBのダッシュボードから設定を行います。

ブラウザから下記へアクセスします。
http://localhost:8086/

デフォルトのログイン情報です。

ユーザー名	influx-user
パスワード	influx-user-password

左のメニューから「Load Data」-「Bucket」を選択します。
「CREATE BUCKET」をクリックします。
下記を設定します。

項目	設定値
Name	influxdb-bucket

「CREATE」をクリックします。

左のメニューから「Load Data」-「Telegraf」を選択します。
「CREATE CONFIGURATION」をクリックします。
下記を設定します。

項目	設定値
Bucket	influxdb-bucket
source	Telegraf internal

検索に「teleg」まで入力すると、アイコンが1つだけになります

「CONTINUE CONFIGURING」をクリックします。

下記を設定します。

項目	設定値
Configuration Name	influx-001

「SAVE AND TEST」をクリックします。

「FINISH」をクリックします。

Grafanaの設定

Grafanaのダッシュボードから設定を行います。

ブラウザから下記へアクセスします。
http://localhost:3000/

デフォルトのログイン情報です。

ユーザー名	admin
パスワード	admin

初回ログイン時はパスワードの変更が必要です

左のメニューから「Connections」-「Data sources」を選択します。

メニューが表示されていない場合は、左上の「Open menu」ボタンをクリックします

「Add data sources」をクリックします。
「InfluxDB」をクリックします。
下記を設定します。

項目	設定値
Query language	Flux
URL	http://influxdb:8086
User	influx-001
Password	influx-user-password
Organization	influxdb-org
Token	influx-admin-token
Default Bucket	influxdb-bucket

「Save & test」をクリックします。
成功すると、「datasource is working」と表示されます。

左のメニューから「Dashboards 」を選択します。
「New」-「New Dashboard」をクリックします。
「Add visualization」をクリックします。
Select data sourceは、「influxdb」をクリックします。

画面下部のQuery欄に、下記のスクリプトを貼り付けます。

from(bucket: "influxdb-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "temp")
  |> filter(fn: (r) => r["_field"] == "cpu")
  |> filter(fn: (r) => r["host"] == "telegraf_001")
  |> filter(fn: (r) => r["topic"] == "mqtt-topics/cpu")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "mean")

「Save」をクリックします。

Titleに任意の名称を入力します。
「Save」をクリックします。
グラフが表示されます。

MQTTXのインストール

MQTTのデータ送信にMQTTXを使用します。
下記URLからQEMUのインストールファイルをダウンロードします。
MQTTX: オール・イン・ワンのMQTTクライアント
今回は「v1.10.1.win64.exe」をダウンロードしています。

MQTTXの設定

MQTTXを起動します。
「New Connection」をクリックします。

EMQXに接続するため下記を設定します。

項目	設定値
Name	mqttx001
Client ID	mqttx_001
host	mqtt://localhost
Username	emqx-user
Password	emqx-user-password

「Connect」をクリックします。

Connetに失敗する場合は、ブラウザでemqxのUsernameとPasswordを再度設定してみてください。
その際は、コピペではなく手入力してください。

「New Subscription」をクリックします。
下記を設定します。

項目	設定値
Topic	mqtt-topic/#

「Confirm」をクリックします。

下記を入力します。

設定値
mqtt-topics/cpu
temp cpu=40

右下の矢印ボタンをクリックします。

しばらくすると、Grafanaでデータがプロットされます。

mqttxで、値を変えて矢印ボタンをクリックすることで、グラフが表示されていきます。

接続確認

うまくプロットされない場合の確認についてです。

EMQXの確認

ブラウザから下記へアクセスします。
http://localhost:18083/

左のメニューから「Monitoring」-「Client」を選択します。
下記の用に表示されていることを確認します。

mqttx_001の表示がない・正しくない場合は、mqttxの設定を確認します。
telegraf_001の表示がない・正しくない場合は、telegraf.confとコンテナの設定を確認します。

Influxdbの確認

ブラウザから下記へアクセスします。
http://localhost:8086/

左のメニューから「Data Explorer」を選択します。

下図のように、「influxdb-bucket」-「temp」-「cpu」-「telegraf_001」-[mqtt-topics/cpu」にチェックを入れて、「SUBMIT」をクリックします。

mqttx入力した値がプロットされていることを確認します。
プロットされていない場合は、influxdbまでデータが届いていないことになります。

最後に

MQTTのPublisher・Broker・Subscriberの構成を利用したデータプロットの仕組みを構築できました。
Publisher(MQTTクライアント)を増やしても、EMQXで受け取れば即可視化が出来るようになっています。
今後は、Raspberry PiやM5Stackといった小型デバイスからのデータを送信して可視化を試したいです。

参考

• emqx - Official Image | Docker Hub
• telegraf - Official Image | Docker Hub
• influxdb - Official Image | Docker Hub
• Line protocol | InfluxDB Cloud (TSM) Documentation
• Grafana: The open observability platform | Grafana Labs
• Run Grafana Docker image | Grafana documentation
• Introduction - MQTTX Documentation