# 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プラットフォームの運用効率と可観測性を向上させます。