概要

複数の通知をそれぞれの対象者に向けてプッシュ通知する流れを実装しました。

使用したサービス

AWS Step Functions
Amazon Pinpoint
AWS Lambda

アーキテクチャ

AWS workshopにあったものを参考にしています。

処理の流れ

流れとしては先程のとおりなのですが

通知対象のリストを取得
(各通知に対して)対象ユーザーリストを取得、S3にファイル保存
S3に保存したファイルをもとにセグメント作成
セグメントが作成できたら
セグメントをもとにキャンペーンを作成
キャンペーンが成功したら
終了

という流れです。
今回は通知する件数が複数ある想定なのでmapを使用して処理を実装しています。

つまづいたところ

VPC lambda内でpinpointに接続するのは辛い

ユーザーの管理をRDSで実施していたのですが、セグメント作成するlambdaでターゲットユーザーの取得をしてセグメントを作成する、となるとVPC lambdaからpinpointへ接続する必要がありました。ただその場合、NATゲートウェイを用意してインターネット経由でアクセスしないといけない、とややこしい設定が必要でした。(9月10日に公式を確認したらpinpointもVPC endoint対応していたので不要かもです...)
なので今回はworkshopにあったアーキテクチャよりlambdaを増やしてRDSの操作をするlambdaとpinpointの処理をするlambdaを分けました。

S3に配置したファイルからセグメントを作成する時に使うのはcreate_segmentではない

今回のlambdaはpythonで実装しているのですが、セグメント作成では名前的にcreate_segmentを使うのだろうと勝手に思い込んで実装を進めていました。ところがどっこい、S3に配置したファイルを使ってセグメント作成するにはcreate_import_jobを使うのです。

StepFunctionsのmapとparallel内でのエラーハンドリングは大切

今回は複数の通知を処理する必要があったため、StepFunctions内でmapを使用しました。
実装をして動作確認をしている中で通知Aの処理中にエラーが発生してしまったことで、全体の実行が失敗に終わってしまい、並行で処理されている通知Bの処理が途中で強制終了されてしまう。ということがありました。
map処理では並行に複数の処理が走っている場合、どれか一つがエラーになると他の処理が強制終了されて失敗のステータスになってしまいます。そうすると、通知Bは正常に送信されるはずだったのに送信されなかった、、、ということが発生してしまいます。それは困るのでエラーハンドリングをきっちりする必要があります。
lambdaのtaskにcatch処理を追加するのです。lambda内でエラーが発生した場合に遷移する先なのですが、こちらの遷移先をfailにしてしまうとやはり全体が失敗となってしまうので、passのステートを遷移先に設定します。これにより、エラーが発生しても実行自体は失敗にならずに他の処理に影響を与えずに済みます。
ただし、今回はエラーが発生したことは把握したかったのでLambda内でput metric dataを使用してエラー監視をするようにしました。

悩んだところ

lambda部分をAPIを直接呼び出す形にするかどうか

StepFunctionsのいいところのひとつはLambdaを作成せずに直接APIを呼び出せることかなと思います。
そこで、セグメントやキャンペーンの作成をAPIで完結してしまおうかとも思いました。
しかし、以下の理由から今回はlambdaを使って実装することにしました。

StepFunctionsは途中のステートから実行することができない
API指定にすると各処理ごとでの動作確認ができず、毎度最初から実行する必要があります。Lambda実装にするとそれぞれの処理での動作確認ができてデバッグしやすいと思ったので今回はLambda実装にしました。
もっとシンプルな設計だとAPIでもいいかなぁと思います。
pinpointで.syncを使えなかった
stepfunctionsでAPI実行にすると、サービスによっては.syncを使用することができます。これを使用することで、例えばsagemakerで学習をする場合、stepfunctions内で定期的に学習ジョブのステータスを確認する処理を入れなくても.syncを設定することで学習ジョブが完了するまでステートがそこで待ってくれます。workshopの設計にもある通り、lambdaでセグメントやキャンペーンのステータスを確認する処理を入れているのですが.sync使ったらこのlambdaいらないじゃん！！と感動しました。しかしなんと、pinpointはこの.syncに対応していなかったのです...(参考)API使用を検討する前にLambdaで実装しはじめてしまっており、設計がシンプルになるならAPIに変更しようと思っていたのでシンプルにならないならLambdaのままでいいやーとなりました。
(API使用の設計に変更してからの動作確認が面倒だった)
前項にも記述したとおり、すでにLambdaで実装を進めていた状態だったのでAPIに変更した後に動作確認するのが面倒だなー()と思ってしまったので諦めましたw
検討タイミングが実装前でもっとシンプルな設計だったらAPI使用で実装したい！！