GCSからCloudRunを起動する

はじめに

みなさんはCloud Runをお使いでしょうか？
おそらくHTTPSリクエストによる呼び出しから何らかの処理を実行し、データベースなどに結果を格納するといった構成を組むことが多いのではないでしょうか？
Cloud Runは、EventarcやPub/Subを用いて様々なイベントをトリガーとして起動することも可能となっています。

今回は「GCSからCloudRunを起動する」と題しまして、Google Cloud Storage（GCS）にファイルがアップロードされたことをトリガーにしてCloud Runを起動し、CloudSQLにイベント情報を格納することを試します。

前提条件

前提条件として、GCPの各APIが有効になっている必要があります。
各APIの詳細は、参考にあるGoogleCloudのドキュメントをご覧ください。
それでは、はじめていきましょう。

GCSの準備

先ずは起動のトリガーとなるバケットを用意します。
後ほど設定するEventarcトリガーにより、Pub/Subを経由してCloud RunへGCSのイベントを送信します。

CloudSQLの準備

次にCloud Runの処理結果を格納する、適当なデータベースとテーブルを用意します。
今回は PostgreSQL を選択し、「ex-postgres」という名称のインスタンスを予め作成しています。

• データベース
  
• テーブル

create table gcs_events (
  id serial not null,
  subject varchar(50),
  date_time varchar(50),
  ce_type varchar(50),
  bucket_name varchar(50),
  primary key (id)
)

CloudRunの準備

最後に起動するCloud Runを準備します。
各プログラムはCloud Shellのエディタにて適当なフォルダ（ここではsample）を用意し、配下に作成します。
コードは後述しますが、出来上がったフォルダの中身は以下のようになります。

├── sample
│   ├── Dockerfile
│   ├── go.mod // モジュールの管理
│   ├── go.sum
│   └── main.go 
│   

レシーバーのプログラムを用意する

Cloud Run サービスに送信されたイベントは、HTTP リクエストの形式で受信されます。
そのためレシーバーとなるプログラムを下記に用意し、処理を呼び出すようにします。言語はGoです。また、Cloud SQLへの接続にはCloudSQL Auth Proxy を使用して Unix ソケット経由で接続します。

今回は行っていませんが、パスワードなどのSQL認証情報は、Secret Manager に格納することをおすすめします。Cloud Run を使用すると、シークレットを環境変数として渡すか、ボリュームとしてマウントできます。

main.go

package main

import (
	"database/sql"
	"fmt"
	_ "github.com/jackc/pgx/v4/stdlib"
	"log"
	"net/http"
	"os"
)

// connectUnixSocket initializes a Unix socket connection pool for
// a Cloud SQL instance of Postgres.
func connectUnixSocket() (*sql.DB, error) {
	mustGetenv := func(k string) string {
		v := os.Getenv(k)
		if v == "" {
			log.Fatalf("Warning: %s environment variable not set.\n", k)
		}
		return v
	}
	var (
		dbUser         = mustGetenv("DB_USER")              // e.g. 'my-db-user'
		dbPwd          = mustGetenv("DB_PASS")              // e.g. 'my-db-password'
		unixSocketPath = mustGetenv("INSTANCE_UNIX_SOCKET") // e.g. '/cloudsql/project:region:instance'
		dbName         = mustGetenv("DB_NAME")              // e.g. 'my-database'
	)
	dbURI := fmt.Sprintf("user=%s password=%s database=%s host=%s",
		dbUser, dbPwd, dbName, unixSocketPath)
	dbPool, err := sql.Open("pgx", dbURI)
	if err != nil {
		return nil, fmt.Errorf("sql.Open: %v", err)
	}
	return dbPool, nil
}

func SetEventsDB(s string, t string, ct string, b string) sql.Result {
	var (
		db  *sql.DB
		err error
	)

	db, err = connectUnixSocket()
	if err != nil {
		log.Fatalf("connectUnixSocket: unable to connect: %s", err)
	}

	upsert := fmt.Sprintf("INSERT INTO gcs_events( subject,date_time,ce_type,bucket_name) VALUES ('%s','%s','%s','%s');", s, t, ct, b)
	res, err := db.Exec(upsert)
	if err != nil {
		log.Fatalf("SQL ERROR: %s", err)
	}
	return res
}

// SetEventsStorage receives and processes a Cloud Audit Log event with Cloud Storage data.
func SetEventsStorage(w http.ResponseWriter, r *http.Request) {
	s := string(r.Header.Get("Ce-Subject"))
	t := string(r.Header.Get("Ce-Time"))
	ct := string(r.Header.Get("Ce-Type"))
	b := string(r.Header.Get("Ce-Bucket"))
	c := fmt.Sprintf("Detected Change in Cloud Storage: %s, %s, %s, %s", s, t, ct, b)
	log.Printf(c)
	fmt.Fprintln(w, c)

	SetEventsDB(s, t, ct, b)
}

func main() {
	http.HandleFunc("/", SetEventsStorage)

	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}
	log.Printf("Listening on port %s", port)
	if err := http.ListenAndServe(":"+port, nil); err != nil {
		log.Fatal(err)
	}
}

今回使用するEventarcトリガーのイベント情報は下記のようになっています。

• ce-subject：イベントが関連しているリソース
• ce-time：イベントが生成された時刻
• ce-type：イベントタイプ
• ce-bucket：イベントが発生したバケット

Dockerfileを用意する

Cloud RunへデプロイするコンテナのためのDockerfileを用意します。

Dockerfile

FROM golang:latest as builder
WORKDIR /app

COPY go.* ./
RUN go mod download
COPY . ./
RUN go build -v -o server

FROM debian:buster-slim
RUN set -x && apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
    ca-certificates && \
    rm -rf /var/lib/apt/lists/*
COPY --from=builder /app/server /server
CMD ["/server"]

コンテナをビルドする

Cloud Shell のターミナルを開き、sample フォルダへ移動します。
その後、Cloud Runで実行するサービスのコンテナをビルドします。
プロジェクトID、リージョン、サービス名は任意です。

build

export PROJECT_ID=$(gcloud config get-value project)
export SERVICE_NAME=gcs-cloudrun
export REGION=asia-east1
gcloud builds submit --tag gcr.io/${PROJECT_ID}/${SERVICE_NAME}

Cloud Runへデプロイする

続いて、ビルドしたコンテナをCloud Runへデプロイします。

deploy

gcloud run deploy ${SERVICE_NAME} \
  --image gcr.io/${PROJECT_ID}/${SERVICE_NAME} \
  --region=${REGION}

Cloud SQLへの接続設定を行う

こちらはデプロイ時にCLIにて設定可能なのですが、説明のため敢えてコンソールで再デプロイします。先ず Cloud Run の「新しいリビジョンの編集とデプロイ」リンクを選択し、環境変数の設定を行います。内容は下記のようになります。「INSTANCE_UNIX_SOCKET」には、Cloud SQL の接続名を設定します。接続名には、<プロジェクトID>:<リージョン>:<インスタンスID> の形式で入力する必要があります。

続いて、「接続」タブを選択し、Cloud RunがCloud SQLと通信できるように設定を行います。
先ほどと同じく、Cloud SQL の接続名を設定します。

設定ができたらデプロイを実行します。

トリガーの設定を行う

指定したイベントの通知を Cloud Run サービスが受信するように、Eventarc トリガーを設定します。Cloud Run の「トリガー」タブより、「Eventarc トリガー」リンクを選択します。
イベントの設定において、Cloud Storage バケットの更新などは直接イベントとして指定することができます。

動作確認

準備が完了できたら動作を確認します。

ファイルをGCSへアップロードします。

Cloud Runのログを見てみます。処理が動いたことを確認できます。

Cloud SQLのテーブルを見ると、データが格納されたことを確認できました。

さいごに

いかがでしたでしょうか？同じイベント駆動のサービスとしてCloud Functionsがあります。
今回紹介した処理はCloud Functionsでも可能ですが、「Cloud Functionsでは機能が不足する」、「コンテナを使いたい」など状況によってはCloud Runを検討すると良いでしょう。

参考

• Cloud SQL Auth Proxy を使用して接続する
• Cloud Run から接続する
• Eventarc を使用して Cloud Storage からイベントを受信する
• Cloud Shell から Cloud SQL for PostgreSQL に接続する