Cloud Pub/Subを使ってCloud Storageへのアップロードの検知をする

先日ハッカソンに出た際に、動画像処理をする必要があったので、Cloud Functionsを使い処理を実装し、アプリを開発しました。

実際に使ってみると、Cloud Functionsでは関数実行のトリガーを簡単に設定できて便利なのですが、

• ローカルでデバッグできない
• デプロイに時間がかかる

という理由でデプロイしてデバッグを繰り返す…という方法でやっていたので、
単純な処理の実装にかなり時間がかかってしまいました。
エミュレータを使えばローカルでもできそうですが、今回は諦めました。

そこでCloud Run + Cloud Pub/Subの構成で
Cloud Storageに画像をアップロードしたら関数を実行できる環境を作りました。
この記事ではCloud Runの説明はあまり書いていません。

Cloud Pub/Subとは?

メッセージングサービスで、これを用いることによって別々のアプリケーション間の通信を実現することができます。
PublisherがTopicに向けてメッセージを送信し、そのトピックに対してSubscriptionを複数作成でき、そこから最終的にSubscriberに配信するというような仕組みになっています。

出典: Cloud Pub/Subドキュメント より引用

Cloud Pub/Subの特性

at-least-once配信

公開するメッセージを少なくとも1回以上配信する仕組みで、Subscriberが確認応答するまで何度も配信し続けます。
通常はメッセージは公開された順に1回配信するのですが、メッセージが順不同で配信されたり複数回配信される場合があります。
複数配信に対応するためには、Subscriberが冪等(ある操作を何度行なっても結果が同じ)である必要があります。
こちらの記事でどのくらい重複するのかを実際に検証していました。

また、メッセージの順序指定がされていない理由としては、厳密に順序指定が必要な使用例があまりなく、重要視されていないからです。
仮にメッセージの順序を保証するとなると、メッセージ①の確認応答を認識するまでメッセージ②の配信を待機しなければなりません。
この順序を妥協することで、スループットが低くなることを防ぎます。

きちんと順序指定をしたい場合はこちらに手法が載ってありました。

pullサブスクリプションとpushサブスクリプションについて

pullサブスクリプション

サブスクライバーアプリケーションがPub/Subサーバーに対して定期的にサーバーにメッセージを取りに行く方式で、
受信したらacknowledgeメソッドを呼び出すことで受信確認をする方法。

pushサブスクリプション

Pub/Subサーバーが指定したエンドポイントに対しメッセージをPOSTし、ステータスコード(200,201,202,204,102)をみて、確認応答をする方法。

今回はStorageにアップロード時にだけイベントが発火すれば良いということで、push型を選択しました。

Cloud Runの環境設定

Cloud Runではコンテナをデプロイすることで簡単に環境を作ることができるサービスです。
環境設定の方法は公式ドキュメントを見て行なっています。

私はGoで実装しましたが、好きな言語でエンドポイント作れば良いと思います。

// server.go
// これをmain.goでserver.Init()で呼び出しています。

package server

import "github.com/gin-gonic/gin"

func Init() {
	r := router()
	err := r.Run()
	if err != nil {
		panic("Can't start server.")
	}
}

func router() *gin.Engine {
	r := gin.Default()

	u := r.Group("/api")
	{
          u.POST("/pubsub_sample", func(c *gin.Context) {
			fmt.Println("message receive!")
             })
	}
	return r
}

これはContainer RepositoryにデプロイしてCloud Runにもデプロイする。

ちなみにCloud Runの設定画面で起動元の役割/メンバーをallUsersにすると誰でもアクセスできるようになります。

Pub/Subの設定

トピックの作成

Pub/Subコンソール上でトピックを作成をクリックして適当な名前(pubsub-sample)を入力。

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

トピックIDをpubsub-sampleのものに設定して作成する。
配信タイプをpushにして、エンドポイントをCloud Runにデプロイしたサービスのエンドポイントを指定する。

こうすることでイベントが発生してメッセージが送信されると、設定したエンドポイントにPOSTされるようになる。

Storageの変更通知を設定

Cloud Storageのドキュメントを参照した。
gsutilコマンドで設定できます。

$ gsutil notification create -t [TOPIC_NAME] -f json gs://[BUCKET_NAME]

結果

試しに設定したstorageに画像をアップロードしてみると…

できたーーーー。
あとはよしなに処理をかけばOK👌

間違ってたりアドバイスがあればコメントお願いします。