こんちは、最近全然ブログ書いてないなーと思ってましたが、たった2ヶ月でしたね。。
今回は、slackみたいなメールの受信およびチャットアプリへの転送基盤をAWS上に構築する機会があったので、やってみた記録を残します。
やりたいこと
みなさんslackは使ってますか？slackには便利な機能がいろいろあり、その中の1つにメール受信機能があります。
これは、slackの特定のチャンネルから払い出されたメールアドレスに対してメールが送信されるとそのチャンネルにメールの内容が転送される、という仕組みです。
slackの公式サイトにも記載があります。
今回はslackではなく代わりに自前のチャットアプリケーションがあるとし、特定のドメインに対してメールが送信されると、受信メールアドレスに応じて、そのチャットアプリケーションの指定のチャンネルに対してメッセージを表示します。
こんなイメージですね
| 受信メールアドレス | メッセージ転送先 |
|---|---|
| chatapp+gewu932032@morifuji-is.ninja | #チャンネルA |
| chatapp+1p30czw2f0@morifuji-is.ninja | #チャンネルB |
| chatapp+10zokr4jfe@morifuji-is.ninja | #チャンネルC |
これをAWS上で構築していきます。
構成案
AWSでメールを受信する仕組みを構築する際には、Amazon SESが鉄板だと思うのでこれを軸に考えていきます。また、イベント失敗時のイベント再実行の仕組みも必要なので、SNSを経由してSQSキューにメッセージを溜めていく構成として、いくつか構成案を洗い出してみました。
- 1.SES → SNS → SQS → EventBridge Pipes → Fargate
- 2.SES → SNS → SQS → AutoscalingGroup(Fargate)
- 3.SES → SNS → SQS → Lambda → 既存APIサーバー（LambdaからHTTPリクエストで既存APIサーバーに処理させる）
- 4.SES → SNS → SQS → Lambda
- 5.SES → SNS → SQS、 CloudWatchEventsで定期的にFargate起動
今回は、すでにwebアプリケーションがコンテナで管理されておりFargateでAPIサーバーとして常時起動しています。したがって、チャットに転送する処理はそのコンテナを利用できるようにしたいです。
なので4は省きます。
また、受信するべきメール数は時間帯によってばらつきがあり、1時間あたり1000件を超え場合もあれば0件の場合もあります。無駄なコンテナの起動はコストに跳ね返ってくるため避けたく、5も省きます。
AutoscalingGroupの構築は割と手間に感じるので、構築の容易さから2と3も省きます。
なので 1.SES → SNS → SQS → EventBridge Pipes → Fargate という構成で組んでみます。
1.SES → SNS → SQS → EventBridge Pipes → Fargate
これではやりたいことができませんでした。Fargate側でメッセージのバッチ処理ができなかったためです。
確かに、AWSのAmazon EventBridge Pipes のバッチ処理と同時実行でも「サポートされているバッチ処理可能なターゲット」の中にFargateはありませんでした。
バッチ処理ができなければメッセージごと（=メールごと）にECSのタスクが立ち上がってしまい、コストの面から望ましくありません。
なので、コンテナで処理させるのを諦めてシンプルに 4.SES → SNS → SQS → Lambda という構成で組んでみました。
4.SES → SNS → SQS → Lambda
結論から言うと、この構成でやりたいことができました！SQSとLambda連携時のバッチ処理、および、SQSへの部分応答は以下の設定を行うことで実現できました
1.LambdaのトリガーとしてSQSを登録する際に、Batch Size / Batch window / Report batch item failures を設定する
2.Lambda関数で失敗したイベントIDを以下のフォーマットで返却する
return {
"batchItemFailures": [
{
"itemIdentifier": "messageId1"
},
{
"itemIdentifier": "messageId2"
},
]
}
https://dev.classmethod.jp/articles/update-aws-lambda-partial-failures/
また、Lambdaの中で受信したメールをハンドリングしましたが、思ったより大変でした。
メールの構造をAWS側では特にハンドリングしてくれないので自前でヘッダーとボディーに分割する必要があり、さらにSQSやSNSを経由するとその分メールのデータが入れ子構造になるのでメールの本文を取り出すためにいくつもデコード処理が必要になりました。（SNSのサブスクリプションの設定でraw メッセージ配信の有効化を無効化すると1つ省くことができるはずです）
以下にテスト用で書いたLambda関数(Node.js)を記載します。複数のメッセージが含まれると奇数番目のメッセージは失敗として扱っています。
export const handler = async (event) => {
for (const { messageId, body } of event.Records) {
console.log(messageId);
const messageFromSqs = JSON.parse(body)
const emailMessage = JSON.parse(messageFromSqs.Message)
const content = Buffer.from(emailMessage.content, 'base64').toString()
const [emailHeader, emailBody] = content.split("\r\n\r\n")
const formattedBody = Buffer.from(emailBody, 'base64').toString()
console.log({messageFromSqs});
console.log({emailMessage})
console.log({content})
console.log({emailHeader})
console.log(formattedBody)
}
const failures = event.Records.filter((_v, index) => {
return index%2===1
}).map(({messageId}) => {
return {
itemIdentifier: messageId
}
})
console.log({failures});
return {
"batchItemFailures": failures
}
};
また、SQSにはDLQが設定されているため、Lambda関数の実行に失敗したとしても再実行が手軽にできるようになっています。
予期しないフォーマットのメールを受け取ったとしても、これで簡単に原因調査ができそうです
まとめ
これで、メール受信基盤のベースを作成することができました。メールアドレスごとに処理を振り分ける場合は、SESのルールセットに記述しても良いですし、Lambda関数側で振り分けても良さそうです。
ただし、同一メールのToに同ドメインのメールアドレスを複数紐づけて（例えばtest001@example.comとtest002@example.com）メール送信すると、SESでの受信は1つにまとめられてしまいます。その場合はLambda側でパースしてそれぞれのToに対して処理を行う必要がありそうです。
また、今回はGmailからのplainなメール送信に対して正常に処理できるLambda関数を作成しましたが、よくあるwebサービスから送信されるHTMLメールや画像付きメールを適切に扱うには、Content-Typeヘッダーを見てデコードする処理が必要そうです。