Cloud Run worker pools を触ってみた

gcloud pubsub topics create worker-pools

gcloud pubsub subscriptions create worker-pools-sub --topic=worker-pools

gcloud storage buckets create gs://cm-hanzawa-yuya-worker-pools \
    --location='asia-northeast1' \
    --uniform-bucket-level-access

import os
import sys
from datetime import datetime
from google.cloud import pubsub_v1
from google.cloud import storage

def upload_to_storage(bucket_name, message_text):
    """メッセージをCloud Storageにアップロードする関数"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)

    # ファイル名を生成
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    filename = f"message_{timestamp}.txt"

    # Cloud Storageにメッセージを保存
    blob = bucket.blob(filename)
    blob.upload_from_string(message_text, content_type="text/plain; charset=utf-8")
    print(f"Message saved to gs://{bucket_name}/{filename}")

def pull_messages(project_id, subscription_id, bucket_name):
    """メッセージを StreamingPull する関数"""
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(project_id, subscription_id)

    def callback(message):
        """メッセージを受信したときのコールバック関数"""
        try:
            message_text = message.data.decode("utf-8")
            print(f"Received message: {message_text}")

            # Cloud Storageにメッセージをアップロード
            upload_to_storage(bucket_name, message_text)

            message.ack()
        except Exception as e:
            print(f"Error processing message: {e}")
            message.nack()

    streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
    print(f"Listening for messages on subscriptions/{subscription_id}...")

    try:
        # メッセージをPullし続ける
        streaming_pull_future.result()
    except Exception as e:
        streaming_pull_future.cancel()
        print(f"Stopped listening for messages due to error: {e}")
        raise

def main():
    # 環境変数を取得
    project_id = os.getenv("PROJECT_ID")
    subscription_id = os.getenv("SUBSCRIPTION_ID")
    bucket_name = os.getenv("BUCKET_NAME")

    if None in [project_id, subscription_id, bucket_name]:
        print("Error: PROJECT_ID, SUBSCRIPTION_ID, and BUCKET_NAME environment variables are required")
        sys.exit(1)

    try:
        pull_messages(project_id, subscription_id, bucket_name)
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

PROJECT_ID=<PROJECT_ID>
SUBSCRIPTION_ID="worker-pools-sub"
BUCKET_NAME="cm-hanzawa-yuya-worker-pools"

gcloud artifacts repositories create python-images \
    --repository-format docker \
    --location asia-northeast1

[[build.env]]
    name = "GOOGLE_ENTRYPOINT"
    value = "python main.py"

gcloud builds submit --pack image=asia-northeast1-docker.pkg.dev/<PROJECT_ID>/python-images/worker-pools

gcloud beta run worker-pools deploy worker-pools-test \
    --region=asia-northeast1 \
    --image=asia-northeast1-docker.pkg.dev/<PROJECT_ID>/python-images/worker-pools:latest \
    --env-vars-file=.local.env

$ gcloud storage cat gs://cm-hanzawa-yuya-worker-pools/message_20250710_091357_497317.txt
> こんにちわ