AWSを使っていて、なにか非同期な処理を実行したい場合、SQSからLambdaをキックする方法が最も一般的な方法の一つではないかと思います。
しかし、何らかの事情（15分以内に終わらないなど）により、LambdaではなくECSタスクとして実行したいケースも存在するのではないでしょうか。
そのようなケースでは、EventBridge Pipesを使うことで容易に実現することができますので、EventBridge PipesとSQSとECSを組み合わせた場合の基本的な構成について紹介します。
EventBridge Pipesは、Producer/Consumer型のメッセージング処理を提供するAWSのサービスで、SQSやDynamoDBなどの様々なサービスとLambda、ECS、API Gatewayなどの様々なサービスを結びつけることができます。
EventBridge Pipesを構成する要素としては、以下のようなものがあります。
EventBridge Pipesは、様々なソースからイベントデータを受け取り、必要に応じてそのデータをフィルタリングおよびエンリッチメントして、ターゲットに送信します。
フィルタリングとエンリッチメントについてはオプションとなっており、最低限、ソースとターゲットを指定する必要があります。
ソースは一連の処理の起点となるサービスのことです。
ソースとして指定できるサービスは以下のとおりです。
指定したフィルター条件を満たさないイベントデータを除外することができます。
フィルタリングで指定できるイベントパターンには以下のようなものがあります。
イベントパターンはJSON形式で指定します。例えば、detail.statusがstoppedおよびoverloade以外のイベントデータにマッチさせるイベントパターンの指定は以下のようになります。
{
"detail": {
"state": [{
"anything-but": ["stopped", "overloaded"]
}]
}
} ソースがSQSの場合、フィルター条件を満たさない場合は自動的にキューから削除されます。
また、メッセージのbodyの形式が文字列かJSONかによってイベントパターンの指定の仕方が変わるので、正しく指定しないと予期せぬ動作をする場合があります。
詳しくは公式ドキュメントをご覧ください。
エンリッチメントでは、ソースからのイベントデータをターゲットに送信する前に任意の処理を挟むことができます。
エンリッチメントとして指定できるサービスは以下のとおりです。
エンリッチメントは同期的に呼び出されます。また、レスポンスサイズは最大で6MBまでに制限されています。
ターゲットはイベントデータを最終的に処理するサービスで、以下のようなサービスを指定することができます（一部抜粋）
ソースがSQSのように順番を強制しない場合、EventBridgeでは、ターゲットを同期的、または非同期的に呼び出すことができます（デフォルトでは同期的に呼び出される）。
ECSタスクを起動する際にパラメータを渡したい場合、ContainerOverridesのCommandでコンテナに渡すことができます。
この設定に関しては、現時点ではコンソールからは行えないので、AWS CLIなどで設定を行います。
EventBridge Pipesを使うと、サービス間を疎に保つことができ、IAMで許可する権限を絞ることができます。
また、SQSとLambdaを組み合わせて非同期処理を実現する場合でも、EventBridge Pipesを使うことで、Lambdaの処理の中でSQSの操作をしなくて良くなる利点があります。
それでは、実際のコンソールを使って構築していきます。
AWSのコンソールでEventBridgeのページを開き、「パイプの作成」をクリックします。
また、サブネットとセキュリティグループについては、カンマ区切りで入力します。
先程説明した通り、SQSからのメッセージ中のパラメータをECSタスクに渡すには、ContainerOverridesのCommandにパラメータを入れる必要があります。これはコンソールからは設定できないので、以下のようにAWS CLIで設定を行います。
% aws pipes update-pipe --cli-input-json file://update.json
{
"Name": "ecs-pipe",
"RoleArn": "arn:aws:iam::048414909692:role/service-role/Amazon_EventBridge_Pipe_ecs-pipe_bd32b832",
"TargetParameters": {
"EcsTaskParameters": {
"EnableECSManagedTags": true,
"EnableExecuteCommand": false,
"LaunchType": "FARGATE",
"NetworkConfiguration": {
"awsvpcConfiguration": {
"AssignPublicIp": "ENABLED",
"SecurityGroups": [
"sg-08a80c3f9a0cfdd77"
],
"Subnets": [
"subnet-079116e8cbcd0a640",
"subnet-065b2b720b039523a"
]
}
},
"Overrides": {
"ContainerOverrides": [
{
"Command": [
"$.body"
],
"Name": "go"
}
]
},
"TaskCount": 1,
"TaskDefinitionArn": "arn:aws:ecs:ap-northeast-1:048414909692:task-definition/pipsSample"
},
"InputTemplate": "{\n \"body\": <$.body>\n}"
}
} それでは、実際にSQSにメッセージを送信して、ECSタスクが起動するか確認します。
SQSの画面を開いて、作成したキューの画面の「キューの送受信」をクリックします。
package main
import (
"fmt"
"os"
)
func main() {
args := os.Args[1:]
fmt.Println(args)
}
EventBridge PipesとSQSとECSを使ったメッセージング処理の基本的な構築について紹介しました。
実際のサービスでは、同時に起動するタスクの上限を設定したり、フィルタリングやエンリッチメントなどを活用するなど、より高度な使い方をすることになるのではと思います。
フィルタリングとエンリッチメントについても少し触ってみましたが、うまく行かないところが多く、なれるまでもう少し触ってみる必要がありそうです。
また、以前はEventBridge Pipesのログを直接出力することができなかったので、デバッグに苦労していたようですが、最近の更新でCloudWatch Logsに出力できるようになりました。
EventBridge Pipesは、まだまだ発展途上のサービスだと思いますので、これからもっと便利に使い易くなっていくと嬉しいです。