はじめに
最近、過去に業務で設計・実装したAWS IotトピックのSubscriberとPublisherのインフラコードを整理してみました。これを機に、アーキテクチャや実装した内容について記事で紹介しようと思います。
ソースコードは下記repoをご参照ください。
※今回紹介する内容は汎用的なものであり、実際業務で実装したアーキテクチャやアプリのロジックの記載はありません。
やりたいこと
- Iotデバイス等からIotトピックに送られたメッセージをsubscribeし、DynamoDBに書き込みたい
- 特定のDynamoDBにアイテムが作成or更新or削除された際に、そのイベントの詳細をIotトピックにpublishし、Iotデバイス等で受信したい
アーキテクチャ
Subscriber
- （Iotデバイス等から）メッセージがJSON形式でIotトピック（iot/test/subscriber）に送られる
- subscriber-lambdaがトリガーされる
- subscriber-lambdaがeventからメッセージを取り出し、指定したフォーマットで指定したDynamoDBテーブル（subscriber-table）にデータを書き込む
Publisher
- （業務アプリ等が）指定したDynamoDBテーブル（publisher-table）にアイテムを作成（or更新or削除）する
- DynamoDB Streamsが変更データをキャプチャし、publisher-lambdaをトリガーする
- publisher-lambdaが、eventから変更データキャプチャを取り出し、指定したフォーマットで指定したIotトピック（iot/test/publisher）にメッセージをpublishする
動作確認
上記repoのTerraformコードをデプロイすれば、動作確認に必要なリソースが作成されます。動作確認の際にMQTTテストクライアントを利用します。
Subscriber
「AWS IoT」>「MQTT テストクライアント」の管理画面を開き、下記トピック名とメッセージペイロードを入力し、「発行」をクリックします。
iot/test/subscriber
{
"TenantID": "test01",
"Timestamp": "2024-04-12T12:53:12.195272",
"Message": "test"
}
DynamoDBの管理画面からsubscriber-tableを開き、項目のスキャンを行うと、上記手順で発行したメッセージをもとに作成されたアイテムが表示されます。
Publisher
「AWS IoT」>「MQTT テストクライアント」の管理画面を開き、下記トピック名を入力し、「サブスクライブ」をクリックします。
iot/test/publisher
DynamoDBの管理画面からpublisher-tableを開き、下記アイテムを新規作成します。
「MQTT テストクライアント」の管理画面に戻ってサブスクリプションを確認してみると、アイテムの変更データキャプチャをもとに発行されたメッセージが表示されます。
アイテムの更新と削除もやってみます。
実装時にハマっていたこと
インフラ
Terraformを使ってLambdaのトリガーにIot ruleを追加する際に、「aws_iot_topic_rule」のリソースブロックで「function_arn」を指定しただけでは、トリガーがうまく設定されませんでした。「aws_lambda_permission」のリソースも記述する必要がありました。
詳しくは過去記事をご参照ください。
アプリ
(今回公開したソースコードには記載されていませんが)Subscriberでは、Iot topicへのメッセージ発行をトリガーにLambda関数を実行し、そのLambda関数に渡すpayloadに、該当topicのtopic名を含める必要がありました。色々調査した結果、topic ruleのクエリでtopic()関数を使うことで実現可能と分かりました。（公式ドキュメントに記載ありました）
詳しくは過去記事をご参照ください。
その他
CICDやユニットテスト周りでも色々な試行錯誤がありましたが、機会があればまた記事にしようと思っています。