メッセージ変換

メッセージ変換は、ユーザー定義のルールに基づいてメッセージを変更およびフォーマットし、その後の処理やサブスクライバーへの配信前に適用する機能です。この機能は非常にカスタマイズ可能で、複数のエンコーディングや高度な変換をサポートしています。

ワークフロー

メッセージがパブリッシュされると、以下のワークフローが実行されます。

1. スキーマ検証：メッセージがパブリッシュされ認可を通過すると、まずスキーマ検証が行われます。メッセージが検証に合格すると、次のステップに進みます。

2. メッセージ変換パイプライン：

   • 変換マッチング：メッセージは、そのトピックに基づいてユーザー定義の変換リストと照合されます。異なるトピックやトピックフィルターに対して複数の変換を設定可能です。
   • 変換実行：マッチした変換は設定された順序で実行されます。パイプラインはJSON、Protobuf、Avroなどの各種エンコーダー・デコーダーをサポートし、Variform式を使ってメッセージの拡張や修正が可能です。
   • 変換後処理：メッセージが変換パイプラインを正常に通過すると、ルールエンジンのトリガーやサブスクライバーへのメッセージ配信など、次の処理に進みます。

3. 失敗時の処理：変換が失敗した場合、ユーザー設定のアクションが実行されます。

   • メッセージ破棄：パブリッシュを終了しメッセージを破棄します。QoS 1およびQoS 2のメッセージにはPUBACKで特定の理由コード（131 - 実装固有のエラー）が返されます。
   • 切断してメッセージ破棄：メッセージを破棄し、パブリッシュしたクライアントを切断します。
   • 無視：追加の処理は行いません。

   変換失敗時には設定に関わらずログエントリを生成可能です。ログの出力レベルはユーザーが設定でき、デフォルトは warning です。さらに、変換失敗はルールエンジンイベント（$events/message_transformation/failed）をトリガーでき、誤ったメッセージを別トピックに再パブリッシュしたり、Kafkaに送信して解析するなどのカスタム処理を実装可能です。

ユーザーガイド

このセクションでは、メッセージ変換機能の設定方法とテスト方法を説明します。

ダッシュボードでのメッセージ変換設定

ダッシュボードでメッセージ変換を作成・設定する手順を示します。

1. ダッシュボードにアクセスし、左側のナビゲーションメニューから Smart Data Hub -> Message Transform をクリックします。

2. Message Transform ページ右上の Create をクリックします。

3. 「Create Message Transform」ページで以下の情報を設定します。

   • Name：変換の名前を入力します。

   • Message Source Topic：変換対象のメッセージが属するトピックを設定します。複数のトピックやトピックフィルターを設定可能です。

   • Note（任意）：任意のメモを入力します。

   • Message Format Transformation：

     • Source Format：変換パイプラインに入るメッセージのペイロードに適用するデコーダーを指定します。選択肢は以下の通りです。

       • None（デコードなし）
       • JSON
       • Avro
       • Protobuf
       • Custom (External HTTP)

       これらのデコーダーはバイナリの入力ペイロードを構造化マップに変換します。Avro、Protobuf、Custom (External HTTP)を選択する場合は、あらかじめスキーマレジストリで作成済みである必要があります。

       複数の変換が連なるパイプラインでは、各ステップでのデコードは必須ではありません。例えば、変換 T1 でペイロードが既にデコードされていれば、後続の変換 T2 はデコードをスキップし、正しい形式のペイロードを利用できます。

     • Target Format：変換パイプラインの最後にメッセージペイロードをバイナリ値としてエンコードするエンコーダーを指定します。エンコーダーの選択肢は Source Format と同じです。

       パイプラインの最後の変換のみがペイロードをバイナリ値にエンコードする必要があり、中間の変換はバイナリエンコードを処理する必要はありません。

   • Message Properties Transformation：

     • Properties：式の結果として得られた変換後の値を書き込む先を指定します。指定可能な宛先は payload、topic、qos、retain（対応するフラグを設定）、および user_property（MQTTのUser-Property）です。user_property を使用する場合は、このフィールドの下に正確に1つのキーを指定する必要があります（例：user_property.my_custom_prop）。payload はそのままメッセージペイロード全体を上書きするか、あるいはネストされたキーのパスを指定してJSONオブジェクトのように扱うことができます（例：payload.x.y）。
     • Target Value：設定したプロパティに書き込む値を定義します。この値は qos、retain、topic、payload、payload.x.y など他のフィールドからコピーするか、Variform式を使って生成できます。

   • Transformation Failure Operation：

     • Action After Failure：変換失敗時に実行するアクションを選択します。
       • Drop Message：パブリッシュを終了しメッセージを破棄、QoS 1およびQoS 2メッセージにはPUBACKで特定の理由コードを返します。
       • Disconnect and Drop Message：メッセージを破棄し、パブリッシュしたクライアントを切断します。
       • Ignore：追加の処理は行いません。

   • Output Logs：変換失敗時にログを生成するか選択します。ログはデフォルトで有効です。

   • Logs Level：ログの出力レベルを設定します。デフォルトは warning です。

4. Create をクリックして設定を完了します。

作成前に Preview をクリックすると、変換のテストが可能です。新しいペインが開き、QoS、ペイロード、リテインフラグの有無、パブリッシャーのユーザー名やクライアントIDなど、入力メッセージのコンテキストを指定できます。必要な情報を入力後、Execute Transformation をクリックすると指定したコンテキストで変換が実行され、結果を確認できます。

作成後は、メッセージ変換ページのリストに変換が表示され、デフォルトで有効になっています。必要に応じて無効化したり、Actions 列の Settings をクリックして設定を更新できます。削除や順序変更は More から行えます。

設定ファイルでのメッセージ変換設定

Avro形式でエンコードされたメッセージを受信し、JSONにデコードしたいとします。デコード後、パブリッシュクライアントのクライアント属性から取得した tenant 属性をトピックの先頭に付加し、ルールエンジンで処理する前にトピックを変更したい場合、以下の設定で実現可能です。

message_transformation {
  transformations = [
    {
      name = mytransformation
      topics = ["t"]
      failure_action = drop
      payload_decoder = {type = avro, schema = myschema}
      payload_encoder = {type = json}
      operations = [
        {key = "topic", value = "concat([client_attrs.tenant, '/', topic])"}
      ]
    }
  ]
}

この設定は、mytransformation という名前の変換を指定し、

• 指定したスキーマを使ってAvro形式のペイロードをデコード、
• ペイロードをJSON形式にエンコード、
• クライアント属性の tenant と元のトピックを連結してトピックを変更、

という処理を行います。

詳細な設定方法は設定マニュアルを参照してください。

REST API

REST APIを使ったメッセージ変換の詳細な利用方法はEMQX Enterprise APIをご覧ください。

デコード／エンコードスキーマの作成

デコーダーおよびエンコーダースキーマの作成方法については、スキーマレジストリのセクションを参照してください。

統計と指標

有効化すると、メッセージ変換はダッシュボード上で統計と指標を公開します。メッセージ変換ページで変換名をクリックすると、以下を確認できます。

統計情報：

• Total：システム起動以来のトリガー総数
• Success：成功したデータ変換の数
• Failed：失敗したデータ変換の数

レート指標：

• 現在の変換速度
• 過去5分間の速度
• 過去の最大速度

統計はリセット可能で、Prometheusの /prometheus/message_transformation からも取得可能です。