AWS IoT CoreからInfluxDBへ送信できるようになったので試してみた
こんにちは。製造ビジネステクノロジー部のakkyです。
最近、AWS IoT CoreにJSONデータをInfluxDBへ直接送信できるルールアクションが追加されました。InfluxDBはTimestream for InfluxDBだけではなく、セルフホスティングしているInfluxDBも送信先として使用可能なようです。 今まで、IoT CoreからInfluxDBにデータを送信するためには、Lambdaなどを使ってInfluxDBへ転送する必要がありましたが、この機能を使用するとLambdaが不要になります。
今回はこの新機能を試してみましたのでご紹介します。ただし現在は制限もありますので併せて紹介します。
構成
今まで
IoT CoreルールからLambdaを使用してラインプロトコルに変換してInfluxDBへ送信する必要がありました。
VPC Lambdaを使用することで、Private Subnetに置いたInfluxDBへ接続できます。
今回サポートされた構成
下図のようにIoT Coreルールから直接InfluxDBへアクセスできるようになったため、Lambdaを使用する必要がなくなりました。
他のIoT Ruleと同様にSQLを書く必要はありますが、マネコンからの設定だけでInfluxDBへデータを送ることができるようになりました。
ただし、InfluxDBへインターネット経由でアクセスできる必要があり、プライベートサブネットに配置されたInfluxDBクラスターへは接続できません。これが現在の最大の制約です。
JSONフォーマットについて
クライアントからMQTTで送られてくる次のような形式のJSONデータをInfluxDBへ保存できるか試しました。
messageTimeはUNIX時間、dataに保存したい測定データを格納する方式としました。
{"messageTime": 1234567890, "data": {"D0": 1, "D1": 2}}
この形式をこのままIoT Coreで取り扱うには2つ問題があります。とはいえ軽微なものです。
1. 時刻列はtimestamp固定
InfluxDBアクションが参照する時刻はtimestampというプロパティでなければなりません。
SELECT messageTime AS timestampとすれば解決できます。
なお、今回は秒単位ですがミリ秒なども選択できます。
2. ペイロード形式の問題
投入するデータがdataにネストされて入っている点が少し問題です。
InfluxDBルールでは次のような形を想定していますので、データを変換する必要があります。
{"timestamp": 1234567890, "D0": 1, "D1": 2}
素朴なのはSQLで次のように指定する方法です。
SELECT messageTime AS timestamp, data.D0 AS D0, data.D1 AS D1 FROM 'topic'
ただ、この方法ではdataが増えた際にSQLも随時更新する必要があります。
そこで、transform()関数を使用してデータを配列に変換する方法を使うとdataの内容が変わっても問題なくなります。
SELECT VALUE transform('enrichArray', {'timestamp': messageTime}, [data]) FROM 'topic'
この場合、出力は配列になりますが、幸いなことにInfluxDBルールのドキュメントによるとClient-side batchingでは配列を受け入れられるので問題ありません。
検証
サービス
InfluxDB V2環境をCDKで作成しました。
InfluxDBへのアクセスにはインターネットからのアクセスが必要ですが、IoT Coreのアクセス元IPアドレスリストは公開されていないようです。
しかしだからといって0.0.0.0/0で公開するのはリスクがあるので、せめてAWSからのアクセスには制限したいところです。
そのため代替として、InfluxDBにつけるセキュリティグループにip-ranges.jsonからap-northeast-1リージョンかつEC2プレフィックスを抜き出し、指定したところ、うまくいきました。
なお、セキュリティグループのクオータ(L-0EA8095F)がデフォルトでは60個なので、デプロイ前に引き上げ申請をしておく必要があります。現在、上記条件のルールは160件程度なので200件で引き上げ申請をしました。
IoT Coreルールの設定
VPCとInfluxDBの作成後、IoT Core Ruleを作成します。InfluxDBアクションの送信先はInfluxDBのアドレスとします。バケット名、測定名、組織名はお好みの値に設定し、タイムスタンプの単位は送信するUNIX時間に合わせます。
バッチに関しては今回は設定しませんでした。
ラインプロトコルタグについてはclientid()が入るようにしました。ただし今回の検証ではaws cliから送信したのでN/Aとなりました。
動作確認
上記フォーマットでランダムな値を送信するシェルスクリプトを作成し、InfluxDB WebUIへ接続すると、値が記録されていることが確認できました。きちんと_fieldにdataのプロパティ名が設定されていました。
まとめ
AWS IoT CoreのInfluxDBルールについて検証しました。
Lambda不要でInfluxDBへデータ送信することができるようになりましたので、よりInfluxDBの用途が広がったと思います。
しかし、Publicサブネットへ配置しないといけないなどの制限もありますので、本番システムへ採用するには少々検証が必要かと思います。
以上
