🌊

[Google Cloud]EventarcでStorageトリガーのCloud Runを作る

2024/05/19に公開

Cloud Run

Google Cloud

Eventarc

tech

※2024年5月19日現在はプレビュー版の機能になります
Eventarcを利用することでCloud Storage(GCS)をトリガーにCloud Runを実行できます。
Cloud FunctionsのStorageトリガーみたいなものです。
Cloud Functionsの第二世代もCloud Runで起動しておりEventarcでトリガーを設定しているためやっていることは似ています。
しかしCloud Functionsの第二世代は512Mbまでのメモリ制限があるため、メモリが必要な際になど利用できるかもしれません。

構成図

GCSの特定のバケットにファイルが配置された時に、Cloud Runを起動できるようになります。

やったこと

チュートリアル形式で書いています。

・サービスの処理

1. バケットにファイルが配置されるとCloud Runのサービスが起動
2. 配置されたバケット名・ファイル名を取得
3. 配置されたファイルを別のバケットにコピー

・作成手順

作成手順

1. サービスアカウントの作成・APIの有効化
2. サービスのデプロイ
3. バケット・Eventarcの作成

ソースコード

からソースコードをダウンロードできます。
01_run_all_deployment.shを実行すればサービスを作成できます。（変数は設定してください）

変数の設定

変数を設定します。
値は適宜修正してください。

変数

PROJECT_ID=$(gcloud config get-value project)
SERVICE_ACCOUNT_NAME=gcs-trigger-cloud-run
SERVICE_ACCOUNT=gcs-trigger-cloud-run@${PROJECT_ID}.iam.gserviceaccount.com
SERVICE_NAME=gcs-trigger-cloud-run
REPOSITORY_NAME=gcs-trigger-cloud-run
REGION=asia-northeast1
SOURCE_BUCKET_NAME=source-gcs-trigger-cloud-run
DESTINATION_BUCKET_NAME=destination-gcs-trigger-cloud-run

1. サービスアカウントの作成・APIの有効化

EventarcとCloud Runの両方で利用するサービスアカウントを作成しています。

サービスアカウント作成とロール付与・API有効化

# サービスアカウント作成
gcloud iam service-accounts create ${SERVICE_ACCOUNT_NAME} \
  --project=${PROJECT_ID} \
  --description="test gcs trigger for cloud run" \
  --display-name="${SERVICE_ACCOUNT_NAME}"

# サービスのAPI有効化
gcloud services enable artifactregistry.googleapis.com \
    cloudbuild.googleapis.com \
    eventarc.googleapis.com \
    run.googleapis.com \
    storage.googleapis.com \
    --project=${PROJECT_ID}

# IAMロールの付与
while read role ; do
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
    --member=serviceAccount:${SERVICE_ACCOUNT} \
    --role=${role}
done < roles.txt

roles.txt

roles/artifactregistry.serviceAgent
roles/run.serviceAgent
roles/run.developer
roles/run.invoker
roles/eventarc.eventReceiver
roles/storage.objectUser

2. サービスのデプロイ

ディレクトリ構成はこのようになっています。

ディレクトリ構成図

/
├ app/
│　└ main.py
├ Dockerfile
└ requirements.txt

・main.py
FASTで構成しています。
バケットのファイルを別のバケットにコピーする処理です。
配置先のバケットとコピー先のバケットが必要です。

main.py

import os
import logging
from typing import Dict

from fastapi import FastAPI
from google.cloud import storage

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)

app = FastAPI()

def transfer_file(source_bucket_name, source_blob_name, destination_bucket_name):
    # GCS クライアントを初期化
    storage_client = storage.Client()

    # ソースバケットとブロブを取得
    source_bucket = storage_client.bucket(source_bucket_name)
    source_blob = source_bucket.blob(source_blob_name)

    # デスティネーションバケットを取得
    destination_bucket = storage_client.bucket(destination_bucket_name)
    destination_blob_name = source_blob_name

    # ファイルをコピー
    source_bucket.copy_blob(source_blob, destination_bucket, destination_blob_name)

    result = f"ファイル {source_blob_name} を {source_bucket_name} から {destination_bucket_name} に転送しました。"
    logger.info(result)
    return result

@app.post("/")
async def on_event(event:Dict):
    source_bucket = event.get("bucket")
    source_file = event.get("name")

    if os.getenv('DESTINATION_BUCKET_NAME'):
        destination_bucket = os.getenv('DESTINATION_BUCKET_NAME')
    else:
        error_no_bucket = "コピー先のバケット名がありません"
        logger.error(error_no_bucket)
        return error_no_bucket

    result = transfer_file(source_bucket, source_file, destination_bucket)
    return result

Dockerfile

FROM python:3.11-slim

WORKDIR /mnt/

COPY requirements.txt /mnt/

# Install production dependencies.
RUN pip install --upgrade pip
RUN pip install -r requirements.txt

COPY . /mnt/

CMD uvicorn app.main:app --host 0.0.0.0 --port $PORT

requirements.txt

uvicorn==0.29.0
fastapi[all]==0.111.0
google-cloud-storage==2.16.0
cloudevents==1.10.1

ファイルを作成したらデプロイします。
変数は適当に入力

デプロイ

gcloud artifacts repositories create ${REPOSITORY_NAME} \
    --repository-format=docker \
    --location=${REGION} \
    --project=${PROJECT_ID}

gcloud builds submit --tag ${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY_NAME}/${SERVICE_NAME}:v1

gcloud run deploy ${SERVICE_NAME} \
    --image ${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY_NAME}/${SERVICE_NAME}:v1 \
    --region ${REGION} \
    --service-account=${SERVICE_ACCOUNT} \
    --project=${PROJECT_ID} \
    --set-env-vars "DESTINATION_BUCKET_NAME=${DESTINATION_BUCKET_NAME}"

3. バケット・Eventarcの作成

--event-filters="type=google.cloud.storage.object.v1.finalized"がトリガーの種類です。ファイルの作成・更新をトリガーにしています。
ファイル削除時などの設定もできます。

バケット・Eventarcの作成

gsutil mb -p ${PROJECT_ID} -c standard -l asia-northeast1 gs://${SOURCE_BUCKET_NAME}
gsutil mb -p ${PROJECT_ID} -c standard -l asia-northeast1 gs://${DESTINATION_BUCKET_NAME}

#トリガー作成
gcloud eventarc triggers create ${SERVICE_NAME} \
    --destination-run-service=${SERVICE_NAME} \
    --destination-run-region=${REGION} \
    --location=${REGION} \
    --event-filters="type=google.cloud.storage.object.v1.finalized" \
    --event-filters="bucket=${SOURCE_BUCKET_NAME}" \
    --service-account=${SERVICE_ACCOUNT}

トリガーの作成には２分ほど時間がかかる可能性があります。
トリガーを確認する場合は少し時間を空けてください。

トリガー確認

% gcloud eventarc triggers list --location=${REGION}

NAME                          TYPE                                      DESTINATION                               ACTIVE  LOCATION
gcs-trigger-cloud-run         google.cloud.storage.object.v1.finalized  Cloud Run service: gcs-trigger-cloud-run  Yes     asia-northeast1

以上で完了です。

結果の確認

動作確認としてtxtファイルをバケットに配置してみます。

実行テスト

# バケットにデータを配置
% echo "hello" > test.txt 
% gsutil cp test.txt gs://source-gcs-trigger-cloud-run
Copying file://test.txt [Content-Type=text/plain]...
/ [1 files][    6.0 B/    6.0 B]                                                
Operation completed over 1 objects/6.0 B.

# 実行結果の確認
% gsutil ls gs://destination-gcs-trigger-cloud-run
gs://destination-gcs-trigger-cloud-run/test.txt
% gcloud logging read "resource.labels.service_name=$SERVICE_NAME AND textPayload:test.txt" --format=json
[
  {
    "insertId": "664775af0003aeb61779f6a9",
    "labels": {
      "instanceId": "00f46b928540c2fa147f31833db421a9b44c2c3992de24bc3641d298e498d7993b4377a5afa9d7efd81638e837a6c3125faf94853ae68a331620520c4de26fde35deb5"
    },
    "logName": "projects/プロジェクトID/logs/run.googleapis.com%2Fstderr",
    "receiveTimestamp": "2024-05-17T15:20:15.288023824Z",
    "resource": {
      "labels": {
        "configuration_name": "gcs-trigger-cloud-run",
        "location": "asia-northeast1",
        "project_id": "プロジェクトID",
        "revision_name": "gcs-trigger-cloud-run-00014-hqj",
        "service_name": "gcs-trigger-cloud-run"
      },
      "type": "cloud_run_revision"
    },
    "textPayload": "2024-05-17 15:20:15,242 - app.main - INFO - ファイル test.txt を source-gcs-trigger-cloud-run から destination-gcs-trigger-cloud-run に転送しました。",
    "timestamp": "2024-05-17T15:20:15.241334Z"
  }
]

その他

Cloud Runに転送されるリクエストデータ

こちらからデータを確認できます。

イベントトリガーについて

ファイルの配置・更新以外にもファイル削除などでトリガーすることもできます。
こちらから確認できます。