スキーマ検証

EMQX は、指定したトピックからサブスクライバーにパブリッシュされるメッセージがあらかじめ定義されたデータ形式に準拠していることを保証するための組み込みスキーマ検証機能を備えています。スキーマ検証は、JSON Schema、Protobuf、Avro など複数のスキーマ形式および組み込みの SQL 文検証をサポートしています。本ページではスキーマ検証機能の概要と利用方法について説明します。

データ検証の必要性

クライアントはブローカーに非標準のメッセージをパブリッシュすることがあり、これがサブスクライバーやデータシステムで例外を引き起こしたり、セキュリティリスクとなる可能性があります。EMQX はデータ形式を早期に検証し、これらの非準拠メッセージを識別・ブロックすることでシステムの安定性と信頼性を確保します。スキーマ検証は以下の利点をもたらします。

データ整合性：MQTT メッセージの構造と形式を検証し、データの一貫性と正確性を保証します。
データ品質：欠落や無効なフィールド、データ型、フォーマットをチェックし、データの一貫性と品質を維持します。
統一データモデル：チーム全体やプロジェクトで統一されたデータモデルを使用することを保証し、不整合やエラーを減らします。
再利用と共有：チームメンバーがスキーマを再利用・共有でき、協力効率を向上させ、繰り返し作業やエラーを削減します。
セキュリティ：悪意あるまたは誤った形式のメッセージの処理を防ぎ、セキュリティ脆弱性のリスクを低減します。
相互運用性：メッセージが標準化された形式に準拠することを保証し、異なるデバイスやシステム間の通信を円滑にします。
デバッグ：無効または誤った形式のメッセージを容易に特定し、デバッグできます。

ワークフロー

メッセージがパブリッシュされると、あらかじめ定義されたルールに基づいて検証されます。検証が成功すれば処理は継続され、失敗した場合はユーザー設定のアクションが実行されます（例：メッセージ破棄や切断）。

メッセージがパブリッシュされると、まず EMQX の認可機構によりパブリッシュ権限がチェックされます。権限チェックを通過した後、ユーザーが設定した検証リストからパブリッシュされたトピックに基づいて検証ルールがマッチングされます。1つの検証器は複数のトピックやトピックフィルターに設定可能です。
検証ルールがマッチすると、メッセージはあらかじめ設定されたスキーマまたは SQL に対して検証されます。
- JSON Schema、Protobuf、Avro など複数のスキーマタイプをサポート。
- EMQX ルールエンジン構文に準拠した SQL 文をサポート。
- 1つのポリシーに複数のスキーマや SQL を追加し、それらの関係を指定可能：
  - All Pass：すべての検証が成功した場合のみ検証成功とみなす。
  - Any Pass：いずれかの検証が成功した時点で検証成功とみなす。
検証が成功すると、メッセージはルールエンジンのトリガーやサブスクライバーへの配信など次の処理に進みます。
検証が失敗した場合、以下のユーザー設定アクションが実行されます。
- メッセージ破棄：パブリッシュを終了しメッセージを破棄、QoS 1 および QoS 2 メッセージには PUBACK で特定の理由コード（131 - Implementation Specific Error）を返します。
- 切断してメッセージ破棄：メッセージを破棄し、パブリッシュクライアントを切断します。
- 無視：追加のアクションは行いません。
設定されたアクションにかかわらず、検証失敗時にログを出力可能で、ログの出力レベルはユーザーが設定でき、デフォルトは warning です。検証失敗はルールエンジンイベント $events/schema_validation/failed をトリガーすることもでき、ユーザーはこのイベントをキャッチして、誤ったメッセージを別トピックにパブリッシュしたり Kafka に送信して解析するなどのカスタム処理が可能です。

ユーザーガイド

このセクションではスキーマ検証機能の設定方法とテスト方法を説明します。

ダッシュボードでのスキーマ検証設定

ダッシュボードでスキーマ検証器を作成・設定する手順を示します。

ダッシュボード左ナビゲーションの Smart Data Hub -> Schema Validation をクリックします。
Schema Validation ページ右上の Create をクリックします。
Create Schema Validation ページで以下を設定します：
- Name：検証器の名前を入力します。
- Message Source Topic：検証対象とするメッセージのトピックを設定します。複数のトピックやトピックフィルターを設定可能です。
- Note（任意）：メモを入力します。
- Validation Method：
  - Validation Strategy：複数の検証方式間の関係を指定します。
    - All Pass（デフォルト）：すべての検証方式が成功した場合にのみ成功とみなします。
    - Any Pass：いずれかの検証方式が成功した時点で検証を終了し成功とみなします。
  - Validation List：Type ドロップダウンからスキーマを選択し、スキーマまたは SQL を追加します。スキーマの作成方法は Create Validation Schema を参照してください。
- Validation Failure Operation：
  - Action After Failure：検証失敗時に実行するアクションを選択します。
    - Drop Message：パブリッシュを終了しメッセージを破棄、QoS 1 および QoS 2 メッセージには PUBACK で特定の理由コードを返します。
    - Disconnect and Drop Message：メッセージを破棄し、パブリッシュクライアントを切断します。
    - Ignore：追加のアクションは行いません。
- Output Logs：検証失敗時にログを出力するか選択します。デフォルトは有効です。
- Logs Level：ログの出力レベルを設定します。デフォルトは warning です。
Create をクリックして設定を完了します。

これで Schema Validation ページのリストに有効な新しい検証器が表示されます。必要に応じて無効化できます。Actions 列の Settings をクリックして検証器の設定を更新したり、More から検証器の削除や順序変更も可能です。

設定ファイルでのスキーマ検証設定

設定の詳細は Configuration Manual を参照してください。

検証スキーマの作成

ここでは JSON Schema を例に検証スキーマの作成方法を示します。JSON Schema は以下の要件を満たす必要があります。

JSON オブジェクトに temp というプロパティを含むこと。
temp プロパティは整数であること。
temp プロパティは最低値が 101 であること。

json

{
  "$schema": "http://json-schema.org/draft-06/schema#",
  "type": "object",
  "properties": {
    "temp": {
      "type": "integer",
      "minimum": 101
    }
  },
  "required": ["temp"]
}