【re:Invent】Amazon EventBridge Pipesの検証と今後について

この記事は、「[Relic Advent Calendar 2022](https://qiita.com/advent-calendar/2022/relic)」の3日目の記事です！

# 前置き
re:Invent 2022が無事終わりましたね。



今年はコロナ禍もあってなんと3年ぶりにラスベガス現地で開催され、Twitterのタイムラインからは現地組の様子を伺えたこともあって現地の雰囲気を感じながら私もkeynoteを中心にオンラインで参戦しました。夜はre:Invent、昼はAWSのシニアエバンジェリスト亀田さんによるre:Capで補完、最終日昼にBlackBelt Onlineで総復習とキャッチアップに勤しんだ1週間だったと思います。

今年の**re:Inventで発表されたアップデートは126個**と、毎年の事ながらですが大量に投下されました。

何処が発祥かは不明ですがJAWS-UGには「**アウトプットしないのは知的な便秘**」という言葉が流行っており、このままでは体調不良を起こしてしまうので私も気になったサービスや新機能を自分で実際に検証して、その内容を書いていこうと思います。

他所さんが後々整理した物を書いてくれるでしょって心の声も聞こえたりしますが、**自分で一次情報拾って試していく事に一定の価値がある**と思っているのと、**多くの人が手を動かしてアウトプットする事で、知見だけじゃなく熱量も繋いでいける**と思うので、多少内容被っても気にしないスタイルで書いていきます!

# Amazon EventBridge Pipesとは
さて、第一弾として「Amazon EventBridge Pipes」について書いていこうと思います。

![スクリーンショット 2022-12-02 2.09.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/82a4fd5f-487a-0266-4739-ae1771be60c7.png)

**EventBridge Pipes**とはEventBridgeの新機能で、**イベントプロデューサー**と呼ばれるイベントのソース元から、**イベント情報をフィルタリングや強化(Enrichment)で変換**した後に、**コンシューマー**と呼ばれるターゲットに**配信**してくれる仕組みです。

また、それらの**順序性を維持**して処理できるというのが大きなポイントになります。
プロデューサーがDynamoDBStreamやKinesis、SQS、MQ、Apache Kafkaなど限定されているのはこの為かもしれません。

![スクリーンショット 2022-12-03 18.10.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/6b7ec6d5-3d08-c23f-21f6-985bf1b161f9.png)


## プロデューサーに対応しているサービス
- Amazon DynamoDB　Stream
- Amazon Kinesis Data Streams
- Amazon SQS
- Amazon Managed Streaming for Apache Kafka
- Amazon MQ
- セルフマネージド Apache Kafka

## コンシューマーに対応しているサービス
- Amazon Step Functions
- Kinesis Data Streams
- Kinesis Data Firehose
- AWS Lambda
- EventBridge
- API Gateway
- CloudWatch logs
- ECS Cluster
- Inspector
- Redshift Cluster
- SageMaker pipeline
- Amazon SQS
- Amazon SNS
- AWS Batch

## EventBridge Pipes上の「強化（Enrichment）」に対応しているサービス
- EventBridge
- AWS Lambda
- API Gateway
- Amazon Step Functions

## EventBridgeイメージ
これまでのEventBridgeとはどう違うのか理解するため、EventBridgeについて改めて復習しますと、EventBridgeはかなりざっくり言うと以下の3つのフローで構成されています。

1. イベントソースからイベントの受信
1. フィルタリング
1. ターゲットへ配信

![Product-Page-Diagram_Amazon-EventBridge_a.32aad7666cc3f644676c494e7abd75051b267c47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/8ef9504a-9df8-3df8-787a-3042e399912f.png)

EventBridgeが対象とするイベントソースは以下を見て分かる通り、AWSサービスだけでも多岐に渡ります。その他にも、DataDogやAuth0、Shopify、Stripeなどの**30を超えるSaaS連携にも対応**しています。

|対応サービス|配信タイプ|
|:----|----|
|Amazon AppFlow|ベストエフォート|
|Application Auto Scaling|ベストエフォート|
|AWS Application Cost Profiler|ベストエフォート|
|Amazon Athena|ベストエフォート|
|AWS Backup|ベストエフォート|
|AWS Batch|保証付き|
|Amazon Braket|保証付き|
|AWS Private Certificate Authority|ベストエフォート|
|Amazon Chime|ベストエフォート|
|AWS CloudFormation|ベストエフォート|
|AWS CloudTrail からのイベント|ベストエフォート|
|Amazon CloudWatch|保証付き|
|AWS CodeArtifact|保証付き|
|AWS CodeBuild|ベストエフォート|
|AWS CodeCommit|ベストエフォート|
|AWS CodeDeploy|ベストエフォート|
|AWS Config|ベストエフォート|
|Amazon Connect|ベストエフォート|
|AWS Control Tower|ベストエフォート|
|AWS Data Exchange|ベストエフォート|
|Amazon Data Lifecycle Manager|ベストエフォート|
|AWS DataSync|ベストエフォート|
|Amazon DevOps Guru|ベストエフォート|
|AWS Elastic Beanstalk|ベストエフォート|
|Amazon Elastic Block Store|ベストエフォート|
|Amazon Elastic Block Store ボリュームへの変更|ベストエフォート|
|Amazon Elastic Compute Cloud (Amazon EC2)|ベストエフォート|
|Amazon EC2 Auto Scaling|ベストエフォート|
|Amazon EC2 フリート|ベストエフォート|
|Amazon EC2 スポットインスタンスの中断|ベストエフォート|
|Amazon Elastic Container Registry|ベストエフォート|
|Amazon Elastic Container Service|保証付き|
|AWS Elemental MediaConvert|保証付き|
|AWS Elemental MediaLive|ベストエフォート|
|AWS Elemental MediaPackage|ベストエフォート|
|AWS Elemental MediaStore|保証付き|
|Amazon EMR|ベストエフォート|
|Amazon EMR Serverless|ベストエフォート|
|Amazon EventBridge のスケジュールされたルール|保証付き|
|Amazon GameLift|ベストエフォート|
|AWS Glue|ベストエフォート|
|AWS Glue DataBrew|ベストエフォート|
|AWS Ground Station|ベストエフォート|
|AWS Health|ベストエフォート|
|Amazon Inspector|ベストエフォート|
|Amazon Interactive Video Service|ベストエフォート|
|AWS IoT Analytics|保証付き|
|AWS IoT Greengrass V1|ベストエフォート|
|AWS IoT Greengrass V2|ベストエフォート|
|AWS Key Management Service CMK の削除|保証付き|
|AWS Key Management Service CMK のローテーション|ベストエフォート|
|AWS Key Management Service でインポートされたキーマテリアルの有効期限切れ|ベストエフォート|
|Amazon Location Service|保証付き|
|Amazon Macie|ベストエフォート|
|Amazon Managed Blockchain|ベストエフォート|
|AWS Managed Services|ベストエフォート|
|Amazon OpenSearch Service|ベストエフォート|
|AWS OpsWorks|保証付き|
|AWS Proton|ベストエフォート|
|Amazon QLDB|保証付き|
|Amazon RDS|ベストエフォート|
|Amazon Redshift|保証付き|
|AWS Resource Access Manager|ベストエフォート|
|Amazon SageMaker|ベストエフォート|
|Savings Plans|ベストエフォート|
|AWS Security Hub|保証付き|
|AWS Signer|保証付き|
|Amazon Simple Storage Service (Amazon S3)|保証付き|
|Amazon Simple Workflow Service|ベストエフォート|
|AWS Step Functions|ベストエフォート|
|AWS Storage Gateway|保証付き|
|AWS Support|ベストエフォート|
|AWS Systems Manager|ベストエフォート|
|リソースのタグ変更|ベストエフォート|
|AWS Transit Gateway|ベストエフォート|
|Amazon Translate|保証付き|
|AWS Trusted Advisor|ベストエフォート|
|Amazon WorkSpaces|ベストエフォート|


## EventBridgeとの違い
一方で、EventBridge Piplesにおけるプロデューサーは**一部のAWSサービスに限定**され、フィルタリングだけでなく間に強化(Enrichment)ステップを挟んでLambdaやStepFunctionなどで処理した後に、コンシューマーへの配信を可能することで、イベントドリブンなアプリケーション開発を支援してくれます。

また、それら**順序立った一連の流れをコードを書くことなく構築**する事が可能なので、コード記述の開発時間を短縮できるのも大きいポイントになります。

# EventBridge Pipesの構築方法
EventBridge Pipesは以下の方法で構築できます。
- AWSコンソール
- CLoudFormation
- CDK

※CDKに対応しているのは嬉しいですね！

# 対応リージョン(2022年現在)
以下の2リージョン以外の商用リージョンで利用可能です。

## 非対応リージョン(2022年現在)
- アジアパシフィック(ハイデラバード)
- ヨーロッパ(チューリッヒ)

# 料金(東京リージョン)
|条件|価格|
|---|---|
|フィルタリング後リクエスト数|100万回で$0.50|

※ペイロードの64KBで1つの要求として課金される。
例えば、256KBのペイロードを持つイベントバッチは4つの要求として課金

## 例
送信されるイベント数: 20,000,000
フィルタリング条件に合致するイベント: 50%
リクエスト毎に5つのイベントとして処理
フィルタリング後のリクエスト数: 20,000,000 * 0.5/5 = 2,000,000
コスト: 2M * 0.5ドル = 1ドル

# EventBridge Pipesを構築
ドキュメントにはCloudFormationやCDKに対応と明記されていましたが、まだリリースされていないようだったのでコンソールから実際に触ってみました。

今回は扱いに慣れるため簡単な、DynamoDB→フィルタリング→CloudWatchlogsのパターンでやってみようと思います。慣れてきたら強化（Enrichment）ステップも挟んだ処理を試してみようと思います。

## DynamoDB(プロデューサー)の用意
DynamoDBを以下の設定値で用意
|項目|設定値|
|---|---|
|パーティションキー|id|
|ソートキー|Message|
|DynamoDBストリーム|有効<br>※必須|

## EventBridge Pipe構築
- コンソールより「パイプの作成」を押下
![スクリーンショット 2022-12-03 19.12.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/a8810030-955c-dcaa-9491-9abaf225c434.png)
- パイプ構築画面の「イベントソース」をクリック
- 詳細のイベントソース選択より「DynamoDB」を選択
![スクリーンショット_2022-12-03_22_39_51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/66b6e379-ce60-dbd4-1539-cfafb9d76312.png)
- DynamoDBストリーム選択より事前に用意した物を選択
![スクリーンショット_2022-12-03_22_42_34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/61755b66-b005-0a30-c9d7-9b343b97781a.png)
- イベントのフィルタリングを選択
![スクリーンショット_2022-12-03_22_43_40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/f5ce0fb6-ce9a-7ebf-295f-101cde074c91.png)
- サンプルイベントタイプで「Enter my own」を選択して以下を入力
![スクリーンショット_2022-12-03_22_56_19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/755dc5af-2cfb-1e18-ee1a-1a8e690c8687.png)

```json
{
  "eventID": "1",
  "eventVersion": "1.0",
  "dynamodb": {
    "Keys": {
      "Id": {
        "N": "101"
      }
    },
    "NewImage": {
      "Message": {
        "S": "Sample"
      },
      "Id": {
        "N": "101"
      }
    },
    "StreamViewType": "NEW_AND_OLD_IMAGES",
    "SequenceNumber": "111",
    "SizeBytes": 26
  },
  "awsRegion": "us-east-1",
  "eventName": "INSERT",
  "eventSourceARN": "arn:aws:dynamodb:us-east-1:111122223333:table/EventSourceTable",
  "eventSource": "aws:dynamodb"
}
```
- イベントパターンで「プレフィックスマッチング」を選択して以下を入力後、テストパターンをクリック
- イベントパターンに一致したサンプルイベントと画面に表示される事を確認
![スクリーンショット_2022-12-03_22_56_34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/5063c61a-7c24-8a96-2576-9dacd9d01e14.png)

```json
{
  "dynamodb": {
    "NewImage": {
      "Message": {
        "S": [{
          "prefix": "Sample"
        }]
      }
    }
  }
}
```
- ターゲットを選択して、ターゲットサービスで「CloudWatch logs」を選択
![スクリーンショット_2022-12-03_22_52_39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/09aab0c9-d717-bdb1-4f85-062efa660d05.png)
- ロググループに任意のロググープを選択
- 追加設定で任意のログストリーム名を入力
![スクリーンショット_2022-12-03_22_54_36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/7728932f-2695-13ef-9f2b-a92f58ca6db6.png)
- 画面上部にてパイプ名を入力して「パイプを作成」をクリック
![スクリーンショット_2022-12-03_23_00_31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/c8c8d8cd-cf87-0596-c239-a15ccfe39685.png)

# 動作確認
- プロデューサーとなるDynamoDBにて項目を以下の値で作成
    - Sampleではログが出力され、Exampleではログが出力されない

|パーティションキー|値|ソートキー|値|
|---|---|---|---|
|id|1|Message|Sample|
|id|2|Message|Example|
- CloudWatchlogsでログ出力確認
![スクリーンショット_2022-12-03_23_11_45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/8b8fbd1a-7e18-df23-c946-af4e53ae28b5.png)

今回、一先ず一連の流れを汲んだPipeが動作されたことが確認できました。

## おまけ1
EventBridge Pipeの削除は「削除」
![スクリーンショット 2022-12-03 21.03.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/638e45aa-22bd-4f45-65b0-4e566d444d1e.png)

EventBridge Ruleの削除は「リソース名」
![スクリーンショット 2022-12-03 21.03.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/354022/b3309abc-d1b2-18b0-a7d1-a6c8760de4e1.png)

どうして…

## おまけ2
まだ日本語の公式ドキュメントが更新されていないため、EventBridgePipesについて調べたい時は[English Ver](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html)を見に行く必要があります。

# 今後の予定
re:inventのkeynoteではEventBridge Pipesの今後の予定についても触れられていましたのでご紹介します。

:::note info
- CloudWatch logsへのサポート
    - 現状CloudWatchメトリクスでパイプの呼び出しと失敗のみ確認可能
- VPCエンドポイント/プライベートリンクへのサポート
- AWS SDKへのサービス統合
- アカウント間連携
- アーカイブとリプレイのサポート
- SAMとCDK(L2)/IDEへのサポート
:::

どれも待ち遠しいですね！！

# まとめ
EventBridgeやStepFunctionなどの対応サービス拡張や機能追加などここ1年で特に進化しており、keynoteで発表されたこともあってイベントドリブンなサーバーレスアプリケーション開発がAWSとって大きなテーマの1つでトレンドなんだなと感じました。

引き続きアップデートされたサービスや機能について気になったものを検証して書いていきたいと思います！

# 参照
- [Create Point-to-Point Integrations Between Event Producers and Consumers with Amazon EventBridge Pipes](https://aws.amazon.com/jp/blogs/aws/new-create-point-to-point-integrations-between-event-producers-and-consumers-with-amazon-eventbridge-pipes/)