受信したイベントを変換する

Eventarc Advanced

CEL を使用して変換式を記述することで、イベントデータを変換できます。たとえば、イベントペイロードを変更して、宛先の特定の API 契約を満たすことができます。

メッセージバインディングを指定しない限り、イベントは常にバイナリコンテンツモードで HTTP リクエストを使用して CloudEvents 形式で配信されます。

入力データ形式と出力データ形式を設定する

CEL で変換式を記述するだけでなく、受信イベントデータのデータ形式を指定することもできます。これにより、Eventarc Advanced はイベントのペイロードを解析する方法を認識できます。また、データの形式を変換することもできます。

サポートされている形式は、Avro、JSON、Protobuf です。詳細については、受信イベントの形式を設定するをご覧ください。

変換式

イベントを変換する場合、すべてのイベント属性には、事前定義された message オブジェクトを介して CEL 式で変数としてアクセスできます。これらの変数には、実行時にイベントデータに基づいて値が設定されます。次に例を示します。

message.id はイベントの id 属性を返します
message.data はイベントペイロードの CEL 値表現を返します
message.data.some-key はイベントペイロードから some-key という名前のフィールドのコンテンツを返します

message.data のフィールドは常に String 型として表され、入力データ形式の設定時に指定されたスキーマを使用して、値が元のイベントからマッピングされます。

変換式は、イベントコンテキスト属性とイベントデータペイロードを含む完全なイベントを表す必要があります。式は JSON で記述されますが、事前定義された CEL 関数、マクロ、演算子、正規表現は RE2 を使用してサポートされています。Eventarc Advanced は、イベントデータの変換に使用できる特定の拡張関数もサポートしています。

CEL 式を使用してイベントデータを変換する例を 2 つ示します。その他のユースケースと例については、変換の例をご覧ください。

例: 属性値をフォーマットする

次の例では、phone_number 属性値を正規表現関数を使用してフォーマットします。（他の属性は省略されています）。

  // Input:
  // {
  //   "data":
  //   {
  //     "email_address": "charlie@altostrat.com",
  //     "phone_number": "8005550100",
  //   }
  // }
  // Output:
  // {
  //    "data":
  //    {
  //      "email_domain": "altostrat.com",
  //      "phone_number": "(800) 555-0100",
  //      "area_code": "800",
  //      "local_number": "5550100",
  //    }
  // }

  {
    "data":
    {
      "email_domain": re.capture(
                        message.data.email_address,
                        "\\S+@(\\S+)"),

      "phone_number": re.extract(
                        message.data.phone_number,
                        "^(\\d{3})(\\d{3})(\\d{4})", "(\\1) \\2-\\3"
                      ),

    }.merge ( re.captureN(message.data.phone_number,
                        "^(?P\d{3})[\w\-)(]*(?P\d{7})"
                      )
    )
  }

前の例で使用されている正規表現関数は次のとおりです。

re.capture: 名前なしまたは名前付きの最初のグループ値をキャプチャします。引数は次のとおりです。
- target: 解析する文字列
- regex: 値のキャプチャに使用する正規表現
最初にキャプチャされたグループ値の文字列を返します。
re.captureN: 指定された文字列と正規表現に対して完全一致を行います。引数は次のとおりです。
- target: 解析する文字列
- regex: 値のキャプチャに使用する正規表現
名前付きグループ（グループ名、キャプチャされた文字列）または名前なしグループ（グループインデックス、キャプチャされた文字列）の Key-Value ペアを含むマップを返します。
re.extract: 指定されたターゲット文字列からグループ値を照合し、文字列を書き換えます。引数は次のとおりです。
- target: 解析する文字列
- regex: 正規表現値の抽出に使用
- rewrite: 結果のフォーマット方法を示す正規表現
rewrite 引数に基づいてフォーマットされた、抽出された値の文字列を返します。

例: 配列をオブジェクトの配列にマッピングする

次の例では、整数の配列をオブジェクトの配列にマッピングします。（他の属性は省略されています）。

  // Input:
  // {
  //   "data":
  //   {
  //        "product_ids": [1, 2, 3]
  //   }
  // }
  // Output:
  // {
  //    "data":
  //    {
  //             "products": [
  //                {
  //                   "name": "apple",
  //                   "price": 70
  //                },
  //                {
  //                    "name": "orange",
  //                    "price":  80
  //                },
  //                {
  //                    "name": "Product(3)",
  //                    "price": 0
  //                },
  //                {
  //                     "name": "apple",
  //                     "price": 70
  //                }
  //            ]
  //    }
  // }

  {
    "data":
    {
      "products":  message.data.product_ids.map(product_id,
              product_id == 1?
              {
                "name": "apple",
                "price": 70
              } :
              product_id == 2?
              {
                "name": "orange",
                "price":  80
              } :
              // Default:
              {
                "name": "Product(" + string(product_id) + ")",
                "price": 0
              }
          )
    }
  }

イベントを変換するようにパイプラインを構成する

コンソールまたは gcloud CLI を使用して、イベントデータを変換するようにパイプラインを構成できます。 Google Cloud

パイプラインごとに 1 つのメディエーションのみがサポートされています。

コンソール

コンソールで、[**Eventarc**] > [**パイプライン**] ページに移動します。 Google Cloud

Pipelines に移動
パイプラインを作成できます。パイプラインを更新する場合は、パイプラインの名前をクリックします。
[**パイプラインの詳細**] ページで、 [**編集**] をクリックします。
[イベントメディエーション] ペインで、次の操作を行います。
1. [変換を適用する] チェックボックスをオンにします。
2. [受信形式] リストで、該当する形式を選択します。
  
  詳細については、受信イベントの形式を設定するをご覧ください。
3. [CEL 式] フィールドに、JSON で変換式を記述します。事前定義された CEL 関数、マクロ、演算子、正規表現がサポートされています。次に例を示します。
```
{
"id": message.id,
"datacontenttype": "application/json",
"data": "{ \"scrubbed\": \"true\" }"
}
```
  上記の例では、次の処理が行われます。
  - 元のイベントから id 以外のすべての属性を削除する
  - datacontenttype 属性を application/json に設定する
  - イベントペイロードを静的な JSON 文字列に置き換える
4. [続行] をクリックします。
[宛先] ペインで、次の操作を行います。
1. 該当する場合は、[送信形式] リストで形式を選択します。
  
  詳細については、受信イベントの形式を設定するをご覧ください。
2. 必要に応じて、メッセージバインディング を適用します。詳細については、このドキュメントのメッセージバインディングを定義するをご覧ください。
[保存] をクリックします。

パイプラインの更新には数分かかることがあります。

gcloud

ターミナルを開きます。
パイプラインを作成することも、 gcloud eventarc pipelines update コマンドを使用してパイプラインを更新することもできます。
```
gcloud eventarc pipelines update PIPELINE_NAME \
    --location=REGION \
    --mediations=transformation_template=\
'
 {
    TRANSFORMATION_EXPRESSION
 }
'
```
注: --mediations フラグは、デフォルトの区切り文字としてハッシュ記号（#）を使用します。式にハッシュが含まれている場合は、解析の競合を避けるために別の区切り文字を指定してください。詳細については、 gcloud トピックのエスケープをご覧ください。たとえば、区切り文字としてセミコロン（;）を使用するには、次のようにします。
--mediations=^;^transformation_template=TRANSFORMATION_EXPRESSION

次のように置き換えます。
- PIPELINE_NAME: パイプラインの ID または完全修飾された名前
- REGION: サポートされている Eventarc Advanced のロケーション
  
  または、gcloud CLI のロケーションプロパティを設定することもできます。
```
gcloud config set eventarc/location REGION
```
- TRANSFORMATION_EXPRESSION: JSON で記述された式。事前定義された CEL 関数、マクロ、演算子、正規表現がサポートされています。mediations フラグは、transformation_template キーを適用するために使用されます。
パイプラインの更新には数分かかることがあります。

例:
```
gcloud eventarc pipelines update my-pipeline \
    --location=us-central1 \
    --mediations=transformation_template=\
'
 {
    "id": message.id,
    "datacontenttype": "application/json",
    "data": "{ \"scrubbed\": \"true\" }"
 }
'
```
上記の例では、次の処理が行われます。
- 元のイベントから id 以外のすべての属性を削除する
- datacontenttype 属性を application/json に設定する
- イベントペイロードを静的な JSON 文字列に置き換える

拡張関数

Eventarc Advanced は、バス経由で受信したイベントデータの変換に使用できる次の拡張関数をサポートしています。

関数	説明
`denormalize`	冗長データを追加して読み取りパフォーマンスを向上させることで、マップまたはリストを非正規化します。結果のマップのフィールド名はピリオド（`.`）で区切られます。リストインデックスは、 `0`から始まる文字列キーに変換されます。 Avro と Protobuf のフィールド名ではピリオド（`.`）を使用できないため、この関数は JSON データをターゲットにするときにのみ使用してください。例: `map.() -> map(string, dyn)` または `list() -> map(string, dyn)`
`merge`	2 つのフィールドを結合し、結合されたフィールドを返します。名前が重複するフィールドはマージされます。例: `message.(message) -> message`
`removeFields`	イベントから特定のフィールドを削除します。フィールド名はパスとして解決されます。ピリオド文字（`.`）は区切り文字として使用されます。未加工の JSON が必要です。JSON をマーシャリングすると、変換が JSON 文字列に適用され、エラーが発生する可能性があります。例: `message.(list(string)) -> message`
`setField`	イベントのフィールドを、指定されたキーで追加または置き換えます。フィールド名はパスとして解決されます。ピリオド文字（`.`）は区切り文字として使用されます。例: `message.(string, dyn) -> message`

例: 他のデータを変更せずにイベントペイロードに属性を追加する

// Input:
// {
//   "data": 
//   {
//        "credit_card_number": "XXXX-XXXX-XXXX-XXXX"
//   }
// }
// Output:
// {
//    "data":
//    {
//        "credit_card_number": "XXXX-XXXX-XXXX-XXXX",
//        "card_type": "credit"
//    }
// }
{
  "data": message.data.merge(
    {
      "card_type": "credit"
    }
  )
}

例: イベントペイロードからアイテムのリストを非正規化する

// Input:
//{
//"data": 
//   {
//        "products": [
//          {
//            "number": 021774,
//            "type": "perishable",
//            "price": 2.00
//          },
//          {
//            "number": 95602,
//            "type": "diy",
//            "price": 120.00
//          },
//          {
//            "number": 568302,
//            "type": "toys",
//            "price": 12.00
//          }
//        ]
//   }
//}
//
// Output:
//{
//"data":
//    {
//        "products": {
//            "0.number": 021774,
//            "0.type": "perishable",
//            "0.price": 2.00,
//            "1.number": 95602,
//            "1.type": "diy",
//            "1.price": 120.00,
//            "2.number": 568302,
//            "2.type": "toys",
//            "2.price": 12.00
//          }
//   }
//}
//
//
message.setField("data.products", message.data.products.denormalize())

例: イベントペイロードからフィールドを削除する

// Input:
// {
//   "data": 
//   {
//     "payment": {
//       "card_number": "XXXX-XXXX-XXXX-XXXX",
//       "card_type": "credit",
//     }
//   }
// }
// Output:
// {
//   "data":
//   {
//     "payment": {
//       "card_type": "credit"
//     }
//   }
// }
message.removeFields(["data.payment.card_number"])

メッセージバインディングを定義する

デフォルトでは、バイナリコンテンツモードで HTTP リクエストを使用して、イベントは常に CloudEvents 形式で宛先に配信されます。必要に応じて、メッセージバインディングを定義して新しい HTTP リクエストを作成することで、この動作をオーバーライドできます。

他のポリシーまたは制御（OAuth トークンや OIDC トークンなど）によって導入された HTTP ヘッダーは保持され、バインディング式から生成されたヘッダーとマージされます。

注: ジョブ構成をオーバーライドする Cloud Run ジョブを実行するには、Cloud Run デベロッパー （roles/run.developer）またはオーバーライド付き Cloud Run ジョブ実行者 （roles/run.jobsExecutorWithOverrides）IAM ロールを付与する必要があります。詳細については、特定の実行のジョブ構成をオーバーライドするをご覧ください。

メッセージバインディングは、 Google Cloud コンソールでパイプラインを構成するとき、または gcloud CLI を使用して定義できます。

コンソール

コンソールで、[**Eventarc**] > [**パイプライン**] ページに移動します。 Google Cloud

Pipelines に移動
パイプラインを作成できます。パイプラインを更新する場合は、パイプラインの名前をクリックします。

パイプラインの更新には 10 分以上かかる場合があります。
[**パイプラインの詳細**] ページで、 [**編集**] をクリックします。
[宛先] ペインで、JSON で記述された CEL 式であるメッセージバインディング を適用します。これにより、新しく作成された HTTP リクエストがパイプラインの宛先に送信されます。

詳細については、このドキュメントの受信メッセージにアクセスすると HTTP リクエストを作成するをご覧ください。
[保存] をクリックします。

gcloud

ターミナルを開きます。
パイプラインを作成することも、 gcloud eventarc pipelines update コマンドを使用してパイプラインを更新することもできます。
```
gcloud eventarc pipelines update PIPELINE_NAME \
    --location=REGION \
    --destinations=http_endpoint_message_binding_template='MESSAGE_BINDING'
```
注: --destinations フラグは、デフォルトの区切り文字としてハッシュ記号 (#) を使用します。式にハッシュが含まれている場合は、解析の競合を避けるために別の区切り文字を指定してください。詳細については、 gcloud トピックのエスケープをご覧ください。たとえば、区切り文字としてセミコロン（;）を使用するには、次のようにします。
--destinations=^;^http_endpoint_message_binding_template='MESSAGE_BINDING'

次のように置き換えます。
- PIPELINE_NAME: パイプラインの ID または完全修飾された名前
- REGION: サポートされている Eventarc Advanced のロケーション
  
  または、gcloud CLI のロケーションプロパティを設定することもできます。
```
gcloud config set eventarc/location REGION
```
- MESSAGE_BINDING: JSON で記述された CEL 式。これにより、新しく作成された HTTP リクエストがパイプラインの宛先に送信されます。
  
  詳細については、このドキュメントの受信メッセージにアクセスすると HTTP リクエストを作成するをご覧ください。
例:
```
gcloud eventarc pipelines create my-pipeline \
    --location=us-central1 \
    --destinations=http_endpoint_uri='https://example-endpoint.com', \
http_endpoint_message_binding_template='{"headers":{"new-header-key": "new-header-value"}}'
```
http_endpoint_message_binding_template キーを使用する場合は、http_endpoint_uri キーも設定する必要があります。

受信メッセージにアクセスする

CEL 式を使用して、次のように受信 CloudEvents メッセージにアクセスできます。

message.data 値を使用して、受信メッセージの data フィールドにアクセスします。
message.key 値（ key は属性の名前）を使用して、受信メッセージの属性にアクセスします。
headers 変数を使用して、処理チェーンの以前のメディエーションによって HTTP リクエストに追加されたヘッダーにアクセスします。この変数は、追加の HTTP ヘッダーに対応する Key-Value ペアのマップを定義します。最初の受信リクエストの元のヘッダーには対応しません。

たとえば、次の CEL 式を使用して、以前のパイプラインメディエーションで追加されたヘッダーに追加のヘッダーを追加することで、ヘッダーのみの HTTP リクエストを作成できます。

{"headers": headers.merge({"new-header-key": "new-header-value"})}

HTTP リクエストを作成する

CEL 式の結果は、Key-Value ペアのマップである必要があります。このマップの headers フィールドと body フィールドを使用して、次のように HTTP リクエストを作成します。

headers フィールドの場合:

CEL 式の結果として headers マップが存在する場合、その Key-Value ペアは HTTP リクエストヘッダーに直接マッピングされ、その値は対応するデータ型の正規文字列エンコードを使用して作成されます。
headers フィールドが存在しない場合、結果の HTTP リクエストは、以前のパイプラインメディエーションで追加されたヘッダーを保持します。

body フィールドの場合:

CEL 式の結果として body フィールドが存在する場合、その値は HTTP リクエスト本文に直接マッピングされます。
body フィールドの値が bytes 型または string 型の場合、そのまま HTTP リクエスト本文として使用されます。それ以外の場合は、JSON 文字列に変換されます。
body フィールドが存在しない場合、結果の HTTP リクエスト本文は、バイナリコンテンツモードの最後の CloudEvents HTTP メッセージバインディングの本文になります。

CEL 式の結果として他のフィールドがある場合は無視されます。

拡張関数

Eventarc Advanced は、メッセージバインディングを指定するときにイベントデータの変換に使用できる次の拡張関数をサポートしています。

関数	説明
`merge`	渡された CEL マップを、関数が適用される CEL マップにマージします。両方のマップに同じキーが存在する場合、またはキーの値が型 `map`の場合、両方のマップがマージされます。それ以外の場合は、渡されたマップの値が使用されます。例: `map1.merge(map2) -> map3`
`toBase64`	CEL 値を base64 URL エンコードされた文字列に変換します。例: `map.toBase64() -> string`
`toCloudEventJsonWithPayloadFormat`	メッセージを CloudEvents メッセージの JSON 表現に対応する CEL マップに変換し、メッセージデータに `toDestinationPayloadFormat` を適用します。また、イベントの `datacontenttype` を指定された送信形式（`output_payload_format_*`）に設定します。送信形式が設定されていない場合は、既存の `datacontenttype` が使用されます。それ以外の場合、 `datacontenttype` は設定されません。メッセージが CloudEvents 仕様に準拠していない場合、関数は失敗します。データを JSON 文字列に変換するには、`toJsonString` を使用します。例: `message.toCloudEventJsonWithPayloadFormat() -> map.toJsonString() -> string`
`toDestinationPayloadFormat`	`message.data` を指定された送信形式（`output_payload_format_*`）に変換します。送信形式が設定されていない場合、 `message.data` は変更されずに返されます。例: `message.data.toDestinationPayloadFormat() -> string or bytes`
`toJsonString`	CEL 値を JSON 文字列に変換します。例: `map.toJsonString() -> string`
`toMap`	CEL マップの CEL リストを 1 つの CEL マップに変換します。例: `list(map).toMap() -> map`

例: ヘッダーを保持し、新しいヘッダーを追加して、本文を宛先形式に設定する

gcloud eventarc pipelines create my-pipeline \
    --location=us-central1 \
    --input-payload-format-json='{}' \
    --destinations=http_endpoint_uri='https://example-endpoint.com',http_endpoint_message_binding_template='{"headers": headers.merge({"content-type":"application/avro"}), "body": message.data.toDestinationPayloadFormat()}',output_payload_format_avro_schema_definition='{"schema_definition": "{"type":"record","name":"myrecord","fields":[{"name":"name","type":"string"},{"name":"account_late","type":"boolean"}]}"}'

受信したイベントを変換する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

入力データ形式と出力データ形式を設定する

変換式

例: 属性値をフォーマットする

例: 配列をオブジェクトの配列にマッピングする

イベントを変換するようにパイプラインを構成する

コンソール

gcloud

拡張関数

例: 他のデータを変更せずにイベント ペイロードに属性を追加する

例: イベント ペイロードからアイテムのリストを非正規化する

例: イベント ペイロードからフィールドを削除する

メッセージ バインディングを定義する

コンソール

gcloud

受信メッセージにアクセスする

HTTP リクエストを作成する

拡張関数

例: ヘッダーを保持し、新しいヘッダーを追加して、本文を 宛先形式に設定する

次のステップ

受信したイベントを変換する

例: 他のデータを変更せずにイベントペイロードに属性を追加する

例: イベントペイロードからアイテムのリストを非正規化する

例: イベントペイロードからフィールドを削除する

メッセージバインディングを定義する

例: ヘッダーを保持し、新しいヘッダーを追加して、本文を宛先形式に設定する