Harmonic Society
オンライン予約 お問い合わせ
プログラミング

Google Cloud Functionsの実践ガイド｜イベント駆動型処理の構築と活用パターン

kento_morota 21分で読めます

「定期的なデータ集計処理を自動化したい」「ファイルがアップロードされたら自動で加工処理を走らせたい」「APIのエンドポイントを手軽に作りたい」――こうした「ちょっとした処理の自動化」ニーズは、あらゆるシステム開発の現場で発生します。

Google Cloud Functions（第2世代）は、サーバーの構築や管理なしに、関数単位でコードを実行できるサーバーレスサービスです。HTTPリクエストやGCPサービスのイベントをトリガーにして、必要な時だけ処理が実行されます。

この記事では、Cloud Functionsの基本概念から実践的な実装パターン、運用のポイントまで、実務で使える知識を体系的に解説します。コンテナベースのサーバーレスに興味がある方は、Cloud Runの実践ガイドもあわせてご覧ください。

Cloud Functionsの基本概念と第2世代の特徴

Cloud Functionsは、GCPが提供するFaaS（Function as a Service）です。インフラの管理は一切不要で、関数のコードを書いてデプロイするだけで、自動スケーリングされるエンドポイントが手に入ります。

第2世代（Gen 2）の進化点

現在推奨されている第2世代（Gen 2）は、内部的にCloud Runの基盤で動作しており、第1世代に比べて大幅に機能が強化されています。

  • 実行時間の延長：最大60分（第1世代は最大9分）
  • インスタンスサイズの拡大：最大16GB RAM、4 vCPU（第1世代は8GB/2 vCPU）
  • 同時実行：1インスタンスで最大1,000の同時リクエスト処理が可能
  • トラフィック分割：カナリアデプロイが可能
  • Eventarcとの統合：90以上のGCPサービスからのイベントをトリガーにできる

対応ランタイム

Cloud Functionsは以下のランタイムに対応しています。

  • Node.js：18、20、22
  • Python：3.10、3.11、3.12
  • Go：1.21、1.22
  • Java：11、17、21
  • Ruby：3.2、3.3
  • .NET：6、8
  • PHP：8.2、8.3

料金体系と無料枠

Cloud Functionsは使った分だけ支払う従量課金制です。リクエストがない時間帯にはコストが発生しません。

課金の仕組み

1. 呼び出し回数
関数が実行された回数に対して課金されます。200万回/月まで無料です。

2. コンピューティング時間
関数の実行時間 × 割り当てられたリソース（CPU・メモリ）で課金されます。

3. ネットワーキング
関数からの送信（下り）トラフィックに対して課金されます。5GB/月まで無料です。

無料枠の詳細

  • 呼び出し：200万回/月
  • コンピューティング：400,000 GB秒 + 200,000 GHz秒/月
  • ネットワーキング（下り）：5GB/月

具体例として、256MBメモリの関数を平均200ミリ秒で実行する場合、月間約200万回まで無料枠で収まります。クラウドコスト全体の管理については、クラウドコスト削減の方法も参考にしてください。

HTTPトリガー関数の実装

最も基本的なHTTPトリガー関数の実装方法を、Python、Node.js、Goの3つの言語で解説します。

Pythonでの実装

# main.py
import functions_framework
import json

@functions_framework.http
def hello_http(request):
    """HTTP Cloud Function."""
    request_json = request.get_json(silent=True)
    request_args = request.args

    name = "World"
    if request_json and "name" in request_json:
        name = request_json["name"]
    elif request_args and "name" in request_args:
        name = request_args["name"]

    return json.dumps({"message": f"Hello, {name}!"}, ensure_ascii=False)

# requirements.txt
functions-framework==3.*

Node.jsでの実装

// index.js
const functions = require('@google-cloud/functions-framework');

functions.http('helloHttp', (req, res) => {
  const name = req.body.name || req.query.name || 'World';
  res.json({ message: `Hello, ${name}!` });
});

デプロイ手順

# Python関数のデプロイ（第2世代）
gcloud functions deploy hello-function \
  --gen2 \
  --runtime python312 \
  --trigger-http \
  --allow-unauthenticated \
  --entry-point hello_http \
  --region asia-northeast1 \
  --memory 256Mi \
  --timeout 60s

# デプロイ後のテスト
curl https://asia-northeast1-PROJECT_ID.cloudfunctions.net/hello-function?name=Tokyo

--allow-unauthenticatedフラグを外すと、認証なしでのアクセスが拒否されます。本番環境ではIAMによる認証を設定してアクセスを制御しましょう。

イベントトリガー関数の実装パターン

Cloud Functionsの真価は、GCPサービスのイベントに反応して自動的に処理を実行できるイベント駆動型のアーキテクチャにあります。

Cloud Storageトリガー：ファイルアップロード時の自動処理

Cloud Storageへのファイルアップロードをトリガーにした処理は、最もよく使われるパターンの1つです。

# main.py - 画像アップロード時にサムネイルを生成
import functions_framework
from google.cloud import storage
from PIL import Image
import io

@functions_framework.cloud_event
def generate_thumbnail(cloud_event):
    """Cloud Storageのオブジェクト作成イベントで発火"""
    data = cloud_event.data
    bucket_name = data["bucket"]
    file_name = data["name"]

    # 画像ファイル以外はスキップ
    if not file_name.lower().endswith(('.png', '.jpg', '.jpeg')):
        print(f"Skipping non-image file: {file_name}")
        return

    # サムネイル生成済みファイルはスキップ（無限ループ防止）
    if file_name.startswith("thumbnails/"):
        return

    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(file_name)

    # 画像をダウンロードしてリサイズ
    image_data = blob.download_as_bytes()
    image = Image.open(io.BytesIO(image_data))
    image.thumbnail((200, 200))

    # サムネイルをアップロード
    output = io.BytesIO()
    image.save(output, format=image.format or 'JPEG')
    output.seek(0)

    thumb_blob = bucket.blob(f"thumbnails/{file_name}")
    thumb_blob.upload_from_file(output, content_type=blob.content_type)
    print(f"Thumbnail created: thumbnails/{file_name}")

# Cloud Storageトリガーでデプロイ
gcloud functions deploy generate-thumbnail \
  --gen2 \
  --runtime python312 \
  --trigger-event-filters="type=google.cloud.storage.object.v1.finalized" \
  --trigger-event-filters="bucket=my-images-bucket" \
  --entry-point generate_thumbnail \
  --region asia-northeast1 \
  --memory 512Mi \
  --timeout 120s

Pub/Subトリガー：非同期メッセージ処理

Pub/Subメッセージをトリガーにした非同期処理パターンです。メール送信や通知処理に適しています。

# main.py - Pub/Subメッセージを処理
import functions_framework
import base64
import json

@functions_framework.cloud_event
def process_message(cloud_event):
    """Pub/Subメッセージの処理"""
    message_data = base64.b64decode(cloud_event.data["message"]["data"]).decode()
    payload = json.loads(message_data)

    event_type = payload.get("type")
    if event_type == "user_signup":
        send_welcome_email(payload["email"])
    elif event_type == "order_completed":
        generate_invoice(payload["order_id"])

    print(f"Processed event: {event_type}")

# Pub/Subトリガーでデプロイ
gcloud functions deploy process-message \
  --gen2 \
  --runtime python312 \
  --trigger-topic=my-events-topic \
  --entry-point process_message \
  --region asia-northeast1

Cloud Schedulerとの連携：定期実行ジョブ

cron形式のスケジュールで関数を定期実行するパターンです。日次の集計処理やレポート生成に使えます。

# main.py - 毎日の売上集計
import functions_framework
from google.cloud import bigquery

@functions_framework.http
def daily_aggregation(request):
    """日次の売上データ集計"""
    client = bigquery.Client()

    query = """
    INSERT INTO analytics.daily_summary (date, total_sales, order_count)
    SELECT
        DATE(created_at) as date,
        SUM(amount) as total_sales,
        COUNT(*) as order_count
    FROM orders.transactions
    WHERE DATE(created_at) = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
    GROUP BY date
    """

    job = client.query(query)
    job.result()  # 完了を待つ

    return {"status": "success", "rows_affected": job.num_dml_affected_rows}

# HTTP関数をデプロイ
gcloud functions deploy daily-aggregation \
  --gen2 \
  --runtime python312 \
  --trigger-http \
  --no-allow-unauthenticated \
  --entry-point daily_aggregation \
  --region asia-northeast1 \
  --timeout 300s

# Cloud Schedulerジョブを作成（毎朝6時に実行）
gcloud scheduler jobs create http daily-aggregation-job \
  --schedule="0 6 * * *" \
  --time-zone="Asia/Tokyo" \
  --uri="https://asia-northeast1-PROJECT_ID.cloudfunctions.net/daily-aggregation" \
  --http-method=POST \
  --oidc-service-account-email=scheduler-sa@PROJECT_ID.iam.gserviceaccount.com

この例ではHTTPトリガー関数にOIDC認証を設定し、Cloud Schedulerから認証付きリクエストを送信しています。BigQueryとの連携は日次集計の典型的なユースケースです。

エラーハンドリングとリトライ設計

イベント駆動型の処理では、エラーハンドリングとリトライの設計が非常に重要です。

冪等性（べきとうせい）の確保

Cloud Functionsのイベントトリガーは、同じイベントが複数回配信される可能性があります（at-least-once delivery）。同じ処理が複数回実行されても結果が変わらないよう、冪等性を確保しましょう。

from google.cloud import firestore

db = firestore.Client()

@functions_framework.cloud_event
def idempotent_handler(cloud_event):
    """冪等なイベントハンドラー"""
    event_id = cloud_event["id"]

    # 処理済みイベントのチェック
    doc_ref = db.collection("processed_events").document(event_id)
    if doc_ref.get().exists:
        print(f"Event {event_id} already processed, skipping")
        return

    try:
        # メインの処理
        result = process_event(cloud_event.data)

        # 処理済みとして記録
        doc_ref.set({
            "processed_at": firestore.SERVER_TIMESTAMP,
            "result": result
        })
    except Exception as e:
        print(f"Error processing event {event_id}: {e}")
        raise  # リトライのために例外を再送出

リトライの設定

# リトライを有効にしてデプロイ
gcloud functions deploy my-function \
  --gen2 \
  --runtime python312 \
  --trigger-topic=my-topic \
  --retry \
  --region asia-northeast1

デッドレターキューの設定

リトライを繰り返しても処理できないメッセージは、デッドレターキューに送って後から分析できるようにします。

# デッドレタートピックの作成
gcloud pubsub topics create dead-letter-topic

# サブスクリプションにデッドレターポリシーを設定
gcloud pubsub subscriptions update my-function-sub \
  --dead-letter-topic=dead-letter-topic \
  --max-delivery-attempts=5

セキュリティとネットワーク設定

Cloud Functionsのセキュリティ設定は、IAMとセキュリティ設計の原則に基づいて行います。

関数の認証設定

本番環境のHTTPトリガー関数では、必ず認証を有効にします。

# 認証必須でデプロイ
gcloud functions deploy my-api \
  --gen2 \
  --runtime python312 \
  --trigger-http \
  --no-allow-unauthenticated \
  --region asia-northeast1

# 特定のサービスアカウントに呼び出し権限を付与
gcloud functions add-invoker-policy-binding my-api \
  --region asia-northeast1 \
  --member="serviceAccount:caller-sa@PROJECT_ID.iam.gserviceaccount.com"

VPCコネクタによるプライベートネットワーク接続

Cloud FunctionsからVPC内のリソース（Cloud SQLのプライベートIPなど）にアクセスする場合は、VPCコネクタを設定します。

# VPCコネクタの作成
gcloud compute networks vpc-access connectors create my-connector \
  --network default \
  --region asia-northeast1 \
  --range 10.8.0.0/28

# VPCコネクタを指定してデプロイ
gcloud functions deploy my-function \
  --gen2 \
  --runtime python312 \
  --trigger-http \
  --vpc-connector my-connector \
  --egress-settings all \
  --region asia-northeast1

環境変数とシークレット管理

# 環境変数を設定してデプロイ
gcloud functions deploy my-function \
  --gen2 \
  --set-env-vars "ENV=production,LOG_LEVEL=info" \
  --set-secrets "API_KEY=api-key-secret:latest,DB_PASSWORD=db-password:latest" \
  --region asia-northeast1

Secret Managerに格納したシークレットを--set-secretsフラグで安全に参照できます。環境変数に機密情報を直接記述するのは避けてください。

テストとデバッグの手法

Cloud Functionsの開発効率を上げるためのテスト・デバッグ手法を紹介します。

ローカルでのテスト実行

Functions Frameworkを使うと、ローカル環境で関数を実行してテストできます。

# Functions Frameworkのインストール
pip install functions-framework

# ローカルで関数を起動
functions-framework --target hello_http --port 8080 --debug

# 別のターミナルからテスト
curl http://localhost:8080 -d '{"name": "Test"}'

単体テストの作成

# test_main.py
import pytest
from unittest.mock import Mock
from main import hello_http

def test_hello_http_with_name():
    req = Mock()
    req.get_json.return_value = {"name": "Tokyo"}
    req.args = {}

    response = hello_http(req)
    assert "Tokyo" in response

def test_hello_http_without_name():
    req = Mock()
    req.get_json.return_value = None
    req.args = {}

    response = hello_http(req)
    assert "World" in response

# テストの実行
pytest test_main.py -v

ログによるデバッグ

# 関数のログをリアルタイムで確認
gcloud functions logs read my-function \
  --gen2 \
  --region asia-northeast1 \
  --limit 50

# Cloud Loggingでフィルタリング
gcloud logging read 'resource.type="cloud_run_revision" AND resource.labels.service_name="my-function" AND severity>=ERROR' \
  --limit 20

まとめ：Cloud Functionsで実現するイベント駆動アーキテクチャ

Cloud Functionsは、サーバー管理なしに軽量な処理を自動実行できる、実務に直結するサービスです。本記事のポイントを振り返ります。

  • 基本：第2世代を使い、最大60分の実行時間と1,000の同時接続を活用する
  • HTTPトリガー：APIエンドポイントやWebhookの受け口として活用する
  • イベントトリガー：Cloud StorageのファイルアップロードやPub/Subメッセージを起点に自動処理を構築する
  • 定期実行：Cloud Schedulerとの連携で日次集計やBigQueryへのデータ投入を自動化する
  • 信頼性：冪等性の確保、リトライ設計、デッドレターキューでエラーに強い設計にする
  • セキュリティ：IAMによる認証とSecret Managerによる機密情報管理を徹底する

Cloud Functionsは単体の小さな処理から始めて、徐々にイベント駆動型のアーキテクチャに発展させていくのがおすすめです。コンテナベースでより複雑なアプリケーションを動かしたい場合は、Cloud Runへの移行も検討してください。インフラ全体をコード化して管理するには、TerraformによるIaC入門が役立ちます。

#GCP#Cloud Functions#サーバーレス
共有:
無料メルマガ

週1回、最新の技術記事をお届け

AI・クラウド・開発の最新記事を毎週月曜にメールでお届けします。登録は無料、いつでも解除できます。

※ プライバシーポリシーに基づき管理します

関連記事

AWS CloudFrontでサイト高速化｜CDN設定からキャッシュ戦略まで実践解説

AWS CloudFrontでサイト高速化｜CDN設定からキャッシュ戦略まで実践解説
AWS CloudWatchで監視・アラート設定｜運用担当者のための実践ガイド

AWS CloudWatchで監視・アラート設定｜運用担当者のための実践ガイド
AWS CodePipelineでCI/CD構築｜コード変更から本番デプロイまでの自動化

AWS CodePipelineでCI/CD構築｜コード変更から本番デプロイまでの自動化
AWS Cost Explorerでコスト可視化｜ムダを見つけて月額費用を削減する実践術

AWS Cost Explorerでコスト可視化｜ムダを見つけて月額費用を削減する実践術
AWS ECS/Fargateでコンテナ運用｜Docker→本番デプロイの実践ガイド

AWS ECS/Fargateでコンテナ運用｜Docker→本番デプロイの実践ガイド
AWS IAMのベストプラクティス｜最小権限の原則を実務で実装する方法

AWS IAMのベストプラクティス｜最小権限の原則を実務で実装する方法
AWS RDSの実務ガイド｜データベース構築・バックアップ・パフォーマンスチューニング

AWS RDSの実務ガイド｜データベース構築・バックアップ・パフォーマンスチューニング
AWS S3の実務活用ガイド｜バケット設計・アクセス制御・コスト最適化の実践

AWS S3の実務活用ガイド｜バケット設計・アクセス制御・コスト最適化の実践
AWS VPCのネットワーク設計入門｜サブネット・セキュリティグループの実践構成

AWS VPCのネットワーク設計入門｜サブネット・セキュリティグループの実践構成

AI活用のヒントをお探しですか？お気軽にご相談ください。

まずは話だけ聞いてもらう