G-gen の杉村です。Pub/Sub の Cloud Storage サブスクリプション機能を実際に使ってみて、その細かい仕様を確認してみました。

はじめに

試したこと

当記事では Cloud Pub/Sub の Cloud Storage サブスクリプション機能を実際に使用してみた際の手順と挙動をご紹介します。

試した使い方はシンプルであり、以下の通りです。

検証用 Cloud Storage バケットを作成
Pub/Sub サービスエージェントに検証用バケットに対する書き込み権限を付与
検証用 Pub/Sub トピックを作成
トピックに所属する Cloud Storage サブスクリプションを作成
トピックにメッセージをパブリッシュ
Cloud Storage への出力ファイルを確認

前提知識

Cloud Pub/Sub (以下、Pub/Sub) は Google Cloud (旧称 GCP) のフルマネージドなメッセージキューイングサービスです。

Cloud Storage サブスクリプションとは、Pub/Sub のサブスクリプション (メッセージ出力先) の一種で、トピックが受け取ったメッセージを直接 Cloud Storage バケットに書き込む仕組みです。Amazon Web Services (AWS) の Amazon Kinesis Data Firehose に類似する機能であるとも言えます。扱ったデータサイズに応じて利用料金が発生します。

Cloud Storage バケットの作成

検証用のバケットとして my-test-bucket-for-pub-sub を作成します。

リージョンは us-central1 としましたが、今回の機能に特段のリージョン制限はありません。

作成手順の詳細は解説しません。以下のドキュメントをご参照ください。

参考 : バケットの作成

Pub/Sub サービスエージェントに権限付与

サービスエージェントとは

Pub/Sub サブスクリプションが Cloud Storage バケットにファイルを書き込むためには、適切な IAM 権限の付与が必要です。

Pub/Sub は、サービスエージェントと呼ばれる特殊なサービスアカウントの権限でファイルの書き込みを行います。サービスエージェントとは、Google Cloud のシステムが内部的に使用するサービスアカウントのことです。詳細は以下の記事をご参照ください。

blog.g-gen.co.jp

権限の付与

Pub/Sub のサービスエージェントは service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com という名称です (${PROJECT_NUMBER} は自身のプロジェクト番号に置き換え)。

このサービスエージェントには、書き込み先バケットに対する Storage オブジェクト作成者 (roles/storage.objectCreator) ロールと Storage レガシーバケット読み取り (roles/storage.legacyBucketReader ロールの2つが必要です。

対象バケット単体に対してのみ IAM ロールを付与することもできますし、プロジェクト全体に IAM ロールを付与することもできます。後者の場合はプロジェクト内の全てのバケットに対して、Pub/Sub からファイルを書き込めるようになります。

手順は以下をご参照ください。

参考 : Pub/Sub サービスアカウントに Cloud Storage のロールを割り当てる

トピックの作成

まず検証用のトピックを作成します。名称は my-test-topic としました。

こちらも作成手順の詳細は解説しませんので、以下のドキュメントをご参照ください。

参考 : トピックの作成と管理

Cloud Storage サブスクリプションの作成

概要

先程のトピックにぶらさげる形でサブスクリプションを作成します。my-test-subscription という名称にします。

サブスクリプションは pull push BigQuery Cloud Storage の4種類から選択できます。今回は Cloud Storage を選びます。

Cloud Storage サブスクリプションに固有なパラメータは「書き込み先バケット」「書き込み時のファイル形式」「接頭辞・接尾辞」「メタデータ書き込み有無」「ストレージバッチの最長時間」「ストレージバッチの最大バイト数」などです。

ファイル形式はシンプルに内容が確認できる text としました。その他には Avro (バイナリエンコードされスキーマ定義されたファイル形式。スキーマハンドリングにメリット) が選択可能です。「メタデータ書き込み有無」は、Avro でのみ利用可能です。text 形式の場合は、message_id などのメタデータやパブリッシュ時に付与したプロパティは失われることに注意が必要です。

「ストレージバッチの最長時間」「ストレージバッチの最大バイト数」は、受け取ったメッセージをどのくらいまで保持してからまとめてバケットに書き込むかを定義する値です。例えばそれぞれ「5分」「1024 KB」とすると、5分経過するか受け取り済みメッセージの合計サイズが 1024 KBを超えるかのいずれかを満たした場合に、まとめて書き込みが実行されます。なお「ストレージバッチの最大バイト数」はオプションであり、未設定のままにすることもできます。

参考 : Cloud Storage サブスクリプションのプロパティ

パラメータ

今回はパラメータを以下のように設定しました。メッセージ有効期限など他のサブスクリプションタイプにも共通な値は省略しています。

パラメータ名	値
配信タイプ	Cloud Storage への書き込み
Cloud Storage バケット	my-test-bucket-for-pub-sub
ストレージファイル形式	text
ストレージファイル接頭辞	test-file-
ストレージファイル接尾辞	.txt
ストレージバッチの最長時間	5 分
ストレージバッチの最大バイト数	未設定

スクリーンショット

検証

メッセージの発行

実際にメッセージを発行してみます。

以下のコマンドで Hello world. It’s ${現在の日付と日時}" という文字列をトピックにパブリッシュ (発行) してみます。これを数秒間隔で繰り返して実行します。

gcloud pubsub topics publish my-test-topic --message="Hello world. It’s `date '+%Y-%m-%d %H:%M:%S'`"

ファイルの確認

メッセージのパブリッシュから5分程度待つと、Cloud Storage にファイルが吐き出されました。ファイル名は <ファイル接頭辞><日時 (UTC)>_<UUID><ファイル接尾辞> の形式になります。

参考 : ファイル名の接頭辞と接尾辞

バケットに出力されたオブジェクト名、作成日時、ファイルの内容 (パブリッシュされた日時) は以下の通りです。

出力順	オブジェクト名	オブジェクト作成時刻	ファイルの内容
1	test-file-2023-08-15T 02_35_08+00_00_d1c49b.txt	11:40:08	Hello world. It’s 2023-08-15 11:35:01
2	test-file-2023-08-15T 02_35_10+00_00_f504c5.txt	11:40:10	Hello world. It’s 2023-08-15 11:35:09
3	test-file-2023-08-15T 02_35_11+00_00_3b783f.txt	11:40:11	Hello world. It’s 2023-08-15 11:35:07
4	test-file-2023-08-15T 02_35_14+00_00_ba1f59.txt	11:40:14	Hello world. It’s 2023-08-15 11:35:10
5	test-file-2023-08-15T 02_48_05+00_00_13cd63.txt	11:53:05	Hello world. It’s 2023-08-15 11:48:02
6	test-file-2023-08-15T 02_51_05+00_00_23d695.txt	11:56:05	Hello world. It’s 2023-08-15 11:51:04
7	test-file-2023-08-15T 03_01_11+00_00_53375c.txt	12:06:11	Hello world. It’s 2023-08-15 12:01:03 Hello world. It’s 2023-08-15 12:01:06
8	test-file-2023-08-15T 03_01_11+00_00_94e77b.txt	12:06:11	Hello world. It’s 2023-08-15 12:01:08

分かったことは、以下のとおりです。

必ずしもメッセージをパブリッシュした順にファイルに出力されるわけではない (No 2と3が逆転している)
メッセージのパブリッシュから概ね「ストレージバッチの最長時間」で設定した時間が経過したあとにオブジェクトが生成される
1メッセージにつき1ファイルが生成されるわけではなく、同時間帯に複数メッセージを受け取った場合は、改行区切りで1ファイル内に複数メッセージが書き込まれる
オブジェクト名に含まれる UTC 日時は必ずしもオブジェクト生成時刻と一致せず、パブリッシュ日時とも一致しない
- オブジェクト名に含まれる日時はパブリッシュ時刻の数秒後となっている。サブスクリプションがトピックからメッセージを受け取って処理した時刻である可能性がある

オブジェクトのメタデータ

Cloud Storage サブスクリプションによって生成されたオブジェクトは、どのようなものでしょうか。着目すべき点だけを挙げます。

content_type は application/octet-stream
- タイプ不明、もしくは特に指定しないことを意味する (参考 : IANA - Media Types)
ストレージクラスは Standard (バケットの設定と同じ)
公開アクセスは 無効化 (バケットの設定と同じ)

スクリーンショット

改行の扱い

同時間帯に受け取ったメッセージは単一ファイルに改行区切りで書き込まれることが分かりました。では、改行を含むメッセージの場合はどのように処理されるのかを確認するため、以下のようなコマンドを短い間隔で複数回実行しました。

date '+%Y-%m-%d %H:%M:%S'; gcloud pubsub topics publish my-test-topic --message="Line 1: `date '+%Y-%m-%d %H:%M:%S'`
Line 2
Line 3"

結果は以下のようになりました。

出力順	オブジェクト名	オブジェクト作成時刻	ファイルの内容
1	test-file- 2023-08-15T03:36:13 +00:00_1046df.txt	12:41:14	Line 1: 2023-08-15 12:36:10 Line 2 Line 3 Line 1: 2023-08-15 12:36:11 Line 2 Line 3
2	test-file- 2023-08-15T03:47:01 +00:00_fd1b17.txt	12:52:01	Line 1: 2023-08-15 12:46:57 Line 2 Line 3 Line 1: 2023-08-15 12:46:57 Line 2 Line 3 Line 1: 2023-08-15 12:47:47 Line 2 Line 3

複数メッセージが改行区切りで1個のテキストファイルに入っています。一つ一つのメッセージにパースする際は、メッセージに明確なデリミタが含まれていないと苦労することになりそうです。あるいは本来的に当機能では、改行を前提としないデータや JSON のような構造を持ったデータが適していると言えそうです。

参考 : ファイル形式

複数サブスクリプション

ひとつのトピックに複数の Cloud Storage サブスクリプションをぶらさげた場合、2つのバケットに同時にオブジェクトが生成されます。

トラブルシューティング

Cloud Storage サブスクリプションを作成する際、以下のようなエラーメッセージが表示される場合があります。

エラーメッセージ "Cloud Pub/Sub did not have the necessary permissions..."

API がエラー「Cloud Pub/Sub did not have the necessary permissions configured to access the provided bucket ${BUCKET_NAME} (or the bucket may not exist). Please verify that the service account service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com was granted the Storage Legacy Bucket Reader and Storage Object Creator roles for the provided bucket. These roles must be granted at the bucket level, not the project level. If you must grant roles at the project level, you may instead grant the Storage Admin role for the project containing the provided bucket.」を返しました

エラー文中の ${BUCKET_NAME} は自身のバケット名に、${PROJECT_NUMBER} はプロジェクト番号に置き換えてお読みください。

これは Pub/Sub のサービスエージェントが対象の Cloud Storage バケットに対して十分な権限を持っていないことを意味しています。

当記事の Pub/Sub サービスエージェントに権限付与 を参考に、IAM 権限を付与すれば解決します。