スキーマ検証

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

なぜデータを検証するのか

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

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

ワークフロー

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

メッセージがパブリッシュされると、まずEMQXの認可機構を通じてパブリッシュ権限がチェックされます。権限チェックを通過した後、ユーザー設定の検証リストからパブリッシュされたトピックに基づいて検証ルールがマッチングされます。1つのバリデーターは複数のトピックまたはトピックフィルターに設定可能です。
検証ルールがマッチすると、メッセージはあらかじめ設定されたスキーマまたはSQLに対して検証されます。
- JSON Schema、Protobuf、Avroなど複数のスキーマタイプをサポート。
- EMQXルールエンジンの構文に準拠したSQLステートメントをサポート。
- 1つのポリシーに複数のスキーマやSQLを追加し、その関係性を指定可能：
  - すべて合格：すべての検証が合格した場合のみ検証成功とみなす。
  - いずれか合格：いずれかの検証が合格した時点で検証成功とみなす。
検証に成功すると、ルールエンジンのトリガーやサブスクライバーへの配送など次の処理に進みます。
検証に失敗した場合、以下のユーザー設定アクションが実行されます。
- メッセージ破棄：パブリッシュを終了しメッセージを破棄、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を追加します。スキーマの作成方法はスキーマ作成を参照してください。
- 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からはバリデーターの削除や順序変更も可能です。

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

設定の詳細については設定マニュアルを参照してください。

スキーマの作成

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

EMQX 6.0.4以降、スキーマレジストリはdraft-03、draft-04、draft-06に加え、JSON Schema draft 2019-09およびdraft 2020-12をサポートしています。$schemaが省略された場合はdraft-06が使用されます。対応範囲や制限、完全な例についてはスキーマレジストリの例 - 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"]
}