MQTTフルリンク可観測性の構築：EMQXイベントトピックとテーブルに基づくイベント監査ソリューション

本番環境において、IoTチームは以下のような課題に直面することが多いです。

デバイスの頻繁な切断
クライアント認証失敗
トピックACL拒否
メッセージの正常な配信失敗
異常なメッセージドロップ

従来のログは通常、ブローカー、認証システム、業務システムに分散しています。問題をトラブルシューティングする際には複数システムのデータを関連付ける必要があり、非効率です。

EMQXイベントトピック、ルールエンジン、テーブルを活用することで、クライアント接続、認証、認可、サブスクリプション、メッセージフローのイベントを統一的に保存できます。これにより、MQTTフルリンクのイベント監査と問題の追跡が可能になります。

ソリューションの目的

EMQXイベントトピック、ルールエンジン、テーブルに基づく統一されたMQTTフルリンクイベントログプラットフォームを構築します。

本ソリューションは以下を実現します：

クライアントライフサイクルの追跡
MQTT認証監査
MQTT認可監査
MQTTメッセージフローのトレース
長期イベント保存
SQLによるクエリと分析

アーキテクチャ設計

text

EMQX Event Topics
        |
        v
    Rule Engine
        |
        v
    Single Action
        |
        v
    EMQX Tables
        |
        v
    SQL Query

特徴：

単一ルール
単一アクション
単一のワイドテーブル
長期保存
SQLクエリ対応

テーブルスキーマ

テーブル名：

text

emqx_full_link_events

テーブル作成文：

sql

CREATE TABLE emqx_full_link_events (
  "timestamp" TIMESTAMP TIME INDEX,
  "event" STRING,
  "clientid" STRING SKIPPING INDEX,
  "username" STRING SKIPPING INDEX,
  "peername" STRING SKIPPING INDEX,
  "sockname" STRING,
  "node" STRING,
  "topic" STRING,
  "id" STRING,
  "from_clientid" STRING,
  "from_username" STRING,
  "qos" BIGINT,
  "payload" STRING,
  "reason" STRING,
  "reason_code" STRING,
  "action" STRING,
  "result" STRING,
  "authz_source" STRING,
  "keepalive" BIGINT,
  "clean_start" BOOLEAN,
  "expiry_interval" BIGINT,
  "connected_at" BIGINT,
  "disconnected_at" BIGINT,
  "publish_received_at" BIGINT,
  "client_attrs" STRING
)
WITH (
  'append_mode'='true',
  'ttl'='7d'
);

テーブルパラメータの説明

本ソリューションでは以下のテーブルパラメータを使用しています：

sql

WITH (
  'append_mode'='true',
  'ttl'='7d'
);

`append_mode='true'`

MQTTイベントは典型的な時系列監査データです。クライアントの接続、認証、認可、サブスクリプション、メッセージ配信は継続的に新しいイベントレコードを生成します。

Append Modeを有効にすると：

すべてのイベントは追記モードで書き込まれます。
過去のレコードは更新されません。
完全なイベントフローが保持されます。
イベント監査と問題の追跡が可能です。

したがって、MQTTフルリンクイベント保存シナリオに適しています。

`ttl='7d'`

イベントトピックは大量のイベントデータを継続的に生成します。

無制限のデータ増加を防ぐため、本例では以下を設定しています：

sql

ttl='7d'

これは直近7日間のイベントデータのみを保持し、それを超えるデータは自動的にクリーンアップされることを意味します。

本番環境では業務要件に応じて保持期間を調整可能です。

テーブルコネクター設定

まず、EMQXテーブルにデータを書き込むためのコネクターを作成します。

テーブルコネクターの作成

データ統合 -> コネクター -> 新規コネクター -> EMQX Tables に進みます。

対応するテーブルデプロイメントの接続情報を入力し、作成を完了してください。

ルール設定

すべてのイベントを統一的に収集します：

sql

SELECT *
FROM "$events/#"

利点：

単一ルールで全イベントトピックをカバー
複数ルールの管理不要
統一された収集エントリポイント

アクション設定

text

emqx_full_link_events clientid=${clientid},event=${event},topic=${topic},username=${username},node=${node},peername=${peername},sockname=${sockname},qos=${qos}i,id=${id},payload=${payload},reason=${reason},reason_code=${reason_code},result=${result},from_clientid=${from_clientid},from_username=${from_username},action=${action},authz_source=${authz_source},keepalive=${keepalive}i,clean_start=${clean_start},expiry_interval=${expiry_interval}i,connected_at=${connected_at}i,disconnected_at=${disconnected_at}i,publish_received_at=${publish_received_at}i,client_attrs=${client_attrs} ${timestamp}

インデックス設計

sql

clientid STRING SKIPPING INDEX
username STRING SKIPPING INDEX
peername STRING SKIPPING INDEX

フィールド	用途
`clientid`	デバイスの完全なフローを検索
`username`	ユーザーの行動を検索
`peername`	発信元IPアドレスを検索

典型的なトラブルシューティングシナリオ

異なるイベントトピックは異なるフィールドを公開するため、一部のフィールドが空になることは正常です。これは問題の追跡やトラブルシューティングに影響しません。

本ソリューションは単一のワイドテーブルにクライアント接続、認証、認可、サブスクリプション、メッセージフローイベントを統一的に保存します。これにより単一のSQLクエリでフルリンクのトラブルシューティング分析が容易になります。

シナリオ1：クライアントライフサイクルのトラブルシューティング

対象の課題：

デバイスはブローカーに正常に接続できたか？
なぜデバイスがオフラインになったか？
デバイスは認証を完了したか？
デバイスはトピックを正常にサブスクライブしたか？

関連イベント：

イベント名	イベント	説明
認証イベント	`client.check_authn_complete`	クライアント認証完了
接続イベント	`client.connected`	クライアント接続成功
接続応答イベント	`client.connack`	ブローカーからの接続応答
サブスクリプションイベント	`session.subscribed`	クライアントのサブスクライブ成功
オフラインイベント	`client.disconnected`	クライアント切断

主要フィールド：

フィールド	説明
`clientid`	クライアントID
`username`	ユーザー名
`peername`	クライアント発信元アドレス
`connected_at`	オンライン時間
`disconnected_at`	オフライン時間
`reason`	オフライン理由
`reason_code`	認証結果

クエリ例：

sql

SELECT *
FROM emqx_full_link_events
WHERE clientid='device001'
AND event IN (
  'client.check_authn_complete',
  'client.connected',
  'client.connack',
  'session.subscribed',
  'client.disconnected'
)
ORDER BY timestamp ASC;

シナリオ2：認証および認可のトラブルシューティング

対象の課題：

なぜデバイスが接続に失敗したか？
ユーザー名とパスワードは正しいか？
なぜトピックのサブスクライブに失敗したか？
なぜACLがリクエストを拒否したか？

関連イベント：

イベント名	イベント	説明
認証イベント	`client.check_authn_complete`	ユーザー認証結果
認可イベント	`client.check_authz_complete`	ACLチェック結果

主要フィールド：

フィールド	説明
`reason_code`	認証結果
`action`	`publish` / `subscribe`
`result`	`allow` / `deny`
`authz_source`	ACLソース

一般的な結果説明：

フィールド	例値	説明
`reason_code`	`success`	認証成功
`reason_code`	`bad_username_or_password`	ユーザー名またはパスワードが誤り
`result`	`allow`	アクセス許可
`result`	`deny`	アクセス拒否

認証失敗のクエリ：

sql

SELECT *
FROM emqx_full_link_events
WHERE event='client.check_authn_complete'
AND reason_code<>'success'
ORDER BY timestamp DESC;

ACL拒否のクエリ：

sql

SELECT *
FROM emqx_full_link_events
WHERE event='client.check_authz_complete'
AND result='deny'
ORDER BY timestamp DESC;

シナリオ3：メッセージ配信の分析

対象の課題：

メッセージは正常に配信されたか？
どのクライアントがメッセージを受信したか？
ACKは受信されたか？

関連イベント：

イベント名	イベント	説明
メッセージ配信イベント	`message.delivered`	メッセージ配信成功
メッセージアックイベント	`message.acked`	ACK受信

主要フィールド：

フィールド	説明
`topic`	MQTTトピック
`payload`	メッセージ内容
`qos`	QoS（サービス品質）
`from_clientid`	パブリッシュ元クライアント
`clientid`	受信クライアント

クエリ例：

sql

SELECT *
FROM emqx_full_link_events
WHERE event IN (
  'message.delivered',
  'message.acked'
)
ORDER BY timestamp DESC;

シナリオ4：メッセージドロップの分析

対象の課題：

なぜメッセージが届かなかったか？
なぜブローカーでメッセージがドロップされたか？

関連イベント：

イベント名	イベント	説明
メッセージドロップイベント	`message.dropped`	転送中にメッセージがドロップ
メッセージ配信ドロップイベント	`message.delivery_dropped`	配信中にメッセージがドロップ

主要フィールド：

フィールド	説明
`topic`	トピック
`payload`	メッセージ内容
`clientid`	クライアント
`reason`	ドロップ理由

一般的な理由説明：

理由	説明
`no_subscribers`	サブスクライバーなし
`queue_full`	キューが満杯
`expired`	メッセージ期限切れ
`acl_denied`	ACL拒否
`disconnected`	クライアントオフライン

パブリッシュドロップのクエリ：

sql

SELECT *
FROM emqx_full_link_events
WHERE event='message.dropped'
ORDER BY timestamp DESC;

配信ドロップのクエリ：

sql

SELECT *
FROM emqx_full_link_events
WHERE event='message.delivery_dropped'
ORDER BY timestamp DESC;

よく使うクエリテンプレート

クライアントIDによる検索

指定したクライアントのすべてのイベントを検索：

sql

SELECT *
FROM emqx_full_link_events
WHERE clientid='device009'
ORDER BY timestamp DESC;

ユーザー名による検索

指定したユーザーのすべてのイベントを検索：

sql

SELECT *
FROM emqx_full_link_events
WHERE username='test'
ORDER BY timestamp DESC;

発信元IPアドレスによる検索

指定した発信元アドレスからクライアントの行動を分析：

sql

SELECT *
FROM emqx_full_link_events
WHERE peername LIKE '78.234.%'
ORDER BY timestamp DESC;

トピックによる検索

指定したトピックのメッセージおよび権限の動作を分析：

sql

SELECT *
FROM emqx_full_link_events
WHERE topic='factory/device/status'
ORDER BY timestamp DESC;

イベントタイプによる検索

指定したイベントを分析：

sql

SELECT *
FROM emqx_full_link_events
WHERE event='message.dropped'
ORDER BY timestamp DESC;

まとめ

イベントトピック、ルールエンジン、テーブルを組み合わせることで、軽量なMQTTフルリンクイベント可観測性プラットフォームを構築できます。

本ソリューションは単一ルール、単一アクション、単一のワイドテーブルでクライアント接続、認証、認可、サブスクリプション、メッセージフローイベントを統一的に保存します。これにより以下を実現します：

クライアントライフサイクル追跡
MQTT認証監査
MQTT認可監査
MQTTメッセージフロー分析
メッセージドロップの根本原因分析
発信元IP行動分析
長期イベント保持
SQLによるクエリと分析
Grafanaによる可視化

単一のワイドテーブル設計により、クライアント接続、認証、認可、サブスクリプション、メッセージイベントを統一されたテーブル構造で保存・検索可能です。これにより単一SQLクエリでフルリンクのトラブルシューティング分析が容易になります。

本ソリューションは以下に適しています：

IoTデバイスの運用保守
MQTTプラットフォーム運用
クライアント行動分析
トピック権限監査
メッセージ配信トラブルシューティング
長期イベント保持と追跡

運用、開発、業務チームが迅速に問題を特定し、過去イベントを追跡し、行動を分析するのに役立ち、トラブルシューティングコストを削減し、MQTTプラットフォームの運用効率と可観測性を向上させます。

専用フレックス

VPC ピアリング接続

PrivateLink の設定

拡張認証

拡張認可

ルール

LLMベースのMQTTデータ処理

サーバレスデプロイAPI

MQTTフルリンク可観測性の構築：EMQXイベントトピックとテーブルに基づくイベント監査ソリューション ​

ソリューションの目的 ​

アーキテクチャ設計 ​

テーブルスキーマ ​

テーブルパラメータの説明 ​

append_mode='true' ​

ttl='7d' ​

テーブルコネクター設定 ​

テーブルコネクターの作成 ​

ルール設定 ​

アクション設定 ​

インデックス設計 ​

典型的なトラブルシューティングシナリオ ​

シナリオ1：クライアントライフサイクルのトラブルシューティング ​

シナリオ2：認証および認可のトラブルシューティング ​

シナリオ3：メッセージ配信の分析 ​

シナリオ4：メッセージドロップの分析 ​

よく使うクエリテンプレート ​

クライアントIDによる検索 ​

ユーザー名による検索 ​

発信元IPアドレスによる検索 ​

トピックによる検索 ​

イベントタイプによる検索 ​

まとめ ​

MQTTフルリンク可観測性の構築：EMQXイベントトピックとテーブルに基づくイベント監査ソリューション

ソリューションの目的

アーキテクチャ設計

テーブルスキーマ

テーブルパラメータの説明

`append_mode='true'`

`ttl='7d'`

テーブルコネクター設定

テーブルコネクターの作成

ルール設定

アクション設定

インデックス設計

典型的なトラブルシューティングシナリオ

シナリオ1：クライアントライフサイクルのトラブルシューティング

シナリオ2：認証および認可のトラブルシューティング

シナリオ3：メッセージ配信の分析

シナリオ4：メッセージドロップの分析

よく使うクエリテンプレート

クライアントIDによる検索

ユーザー名による検索

発信元IPアドレスによる検索

トピックによる検索

イベントタイプによる検索

まとめ