# ❌ 従来のGCE構成での問題点
# インスタンス: n1-standard-2 × 3台（常時稼働）
# 月額費用: ¥180,000
# CPU使用率: 平均15%（大部分がアイドル状態）
# スケーリング: 手動設定、遅延あり
# 運用工数: 週20時間（パッチ適用、監視等）
# 典型的な処理フロー
1. ユーザーが画像をアップロード
2. GCEインスタンスで画像処理実行
3. 処理結果をCloud Storageに保存
4. 結果URLをレスポンス
# 問題：処理が集中する時間帯以外はリソースが無駄

# ✅ Cloud Functionsでの解決
import functions_framework
from google.cloud import storage
from PIL import Image
import io
@functions_framework.http
def process_image(request):
    """画像処理を行うCloud Function"""
# リクエストから画像データを取得
image_data = request.files['image'].read()
# PIL で画像処理
image = Image.open(io.BytesIO(image_data))
# リサイズ処理（複数サイズ生成）
sizes = [(800, 600), (400, 300), (200, 150)]
processed_images = {}
for width, height in sizes:
resized = image.resize((width, height), Image.Resampling.LANCZOS)
# バイナリデータに変換
img_byte_arr = io.BytesIO()
resized.save(img_byte_arr, format='JPEG', quality=85)
processed_images[f"{width}x{height}"] = img_byte_arr.getvalue()
# Cloud Storage に保存
client = storage.Client()
bucket = client.bucket('processed-images-bucket')
urls = {}
for size, image_data in processed_images.items():
blob_name = f"processed/{size}/{request.form['filename']}"
blob = bucket.blob(blob_name)
blob.upload_from_string(image_data, content_type='image/jpeg')
urls[size] = f"https://storage.googleapis.com/processed-images-bucket/{blob_name}"
return {
'status': 'success',
'processed_urls': urls,
'processing_time': '0.8s'
}
# requirements.txt
functions-framework==3.4.0
google-cloud-storage==2.10.0
Pillow==10.0.0

# 実際の使用量（プロジェクトA）
リクエスト数: 1,000,000回/月
平均実行時間: 800ms
メモリ使用量: 512MB
# 料金計算
リクエスト料金: (1,000,000 - 2,000,000無料) × $0.0000004 = $0
実行時間料金: 800,000秒 × $0.0000025 = $2.00
メモリ料金: 800,000 × 0.5GB × $0.0000025 = $1.00
月額合計: $3.00 (約¥450)

# 同等の使用量での比較
リクエスト数: 1,000,000回/月
平均実行時間: 800ms
メモリ使用量: 512MB
# 料金計算
リクエスト料金: (1,000,000 - 1,000,000無料) × $0.0000002 = $0
実行時間料金: 800,000秒 × $0.0000166667 = $13.33
月額合計: $13.33 (約¥2,000)

# ✅ シンプルなデプロイ
# 1. 関数作成
gcloud functions deploy process-image \
--runtime python39 \
--trigger-http \
--allow-unauthenticated \
--memory 512MB \
--timeout 60s
# 2. 環境変数設定
gcloud functions deploy process-image \
--set-env-vars BUCKET_NAME=my-bucket,API_KEY=secret
# 3. 即座にデプロイ完了（約30秒）

# ❌ より複雑な設定が必要
# 1. IAMロール作成
# 2. デプロイパッケージ作成
# 3. Lambda関数作成
# 4. API Gateway設定
# 5. 権限設定
# 合計設定時間: 約15分

graph TD
    A[ユーザー] --> B[Cloud Load Balancer]
    B --> C[Cloud Functions]
    C --> D[Cloud Storage]
    C --> E[Cloud Vision API]
    C --> F[Firestore]
    F --> G[リアルタイム通知]

# main.py - メイン処理
import functions_framework
from google.cloud import storage, firestore, vision
import json
import uuid
from datetime import datetime
@functions_framework.http
def analyze_image(request):
    """画像分析を行うCloud Function"""
try:
# リクエスト検証
if 'image' not in request.files:
return {'error': 'No image provided'}, 400
image_file = request.files['image']
user_id = request.form.get('user_id')
# 一意のファイル名生成
file_id = str(uuid.uuid4())
filename = f"{user_id}/{file_id}.jpg"
# Cloud Storage にアップロード
storage_client = storage.Client()
bucket = storage_client.bucket('image-analysis-bucket')
blob = bucket.blob(filename)
blob.upload_from_file(image_file, content_type='image/jpeg')
# Cloud Vision API で画像分析
vision_client = vision.ImageAnnotatorClient()
image = vision.Image(source=vision.ImageSource(
image_uri=f"gs://image-analysis-bucket/{filename}"
))
# 複数の分析を並列実行
responses = {}
# ラベル検出
label_response = vision_client.label_detection(image=image)
responses['labels'] = [
{'description': label.description, 'score': label.score}
for label in label_response.label_annotations[:10]
]
# テキスト検出
text_response = vision_client.text_detection(image=image)
responses['text'] = text_response.full_text_annotation.text if text_response.full_text_annotation else ""
# 顔検出
face_response = vision_client.face_detection(image=image)
responses['faces'] = len(face_response.face_annotations)
# 結果をFirestoreに保存
db = firestore.Client()
doc_ref = db.collection('image_analysis').document(file_id)
doc_ref.set({
'user_id': user_id,
'filename': filename,
'analysis_results': responses,
'created_at': datetime.utcnow(),
'status': 'completed'
})
return {
'status': 'success',
'file_id': file_id,
'analysis_results': responses,
'image_url': f"https://storage.googleapis.com/image-analysis-bucket/{filename}"
}
except Exception as e:
# エラーログ記録
print(f"Error processing image: {str(e)}")
return {'error': 'Internal server error'}, 500
# Firestore トリガー関数
@functions_framework.cloud_event
def notify_analysis_complete(cloud_event):
    """分析完了時の通知処理"""
# Firestore の変更イベントを処理
data = cloud_event.data
if data.get('value', {}).get('fields', {}).get('status', {}).get('stringValue') == 'completed':
user_id = data['value']['fields']['user_id']['stringValue']
file_id = data['value']['fields']['filename']['stringValue']
# プッシュ通知やメール送信などの処理
send_notification(user_id, f"画像分析が完了しました: {file_id}")
def send_notification(user_id, message):
    """通知送信処理"""
# 実際の通知ロジック
pass

# ETL処理のCloud Function
import functions_framework
from google.cloud import bigquery, storage
import pandas as pd
import io
@functions_framework.cloud_event
def process_data_file(cloud_event):
    """Cloud Storage にファイルがアップロードされた時の処理"""
# イベントデータから情報取得
bucket_name = cloud_event.data['bucket']
file_name = cloud_event.data['name']
# CSVファイルのみ処理
if not file_name.endswith('.csv'):
return
# Cloud Storage からファイル読み込み
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(file_name)
# pandas でデータ処理
csv_data = blob.download_as_text()
df = pd.read_csv(io.StringIO(csv_data))
# データクレンジング
df = df.dropna()  # 欠損値除去
df['processed_at'] = pd.Timestamp.now()  # 処理時刻追加
# BigQuery にデータ投入
bigquery_client = bigquery.Client()
table_id = "my-project.my_dataset.processed_data"
job_config = bigquery.LoadJobConfig(
write_disposition="WRITE_APPEND",
source_format=bigquery.SourceFormat.CSV,
autodetect=True,
)
# DataFrameを直接BigQueryに投入
job = bigquery_client.load_table_from_dataframe(
df, table_id, job_config=job_config
)
job.result()  # 完了まで待機
print(f"Processed {len(df)} rows from {file_name}")

# グローバル変数でクライアント初期化（再利用）
from google.cloud import storage, firestore
import functions_framework
# 関数外で初期化（コンテナ再利用時に効果的）
storage_client = storage.Client()
db_client = firestore.Client()
@functions_framework.http
def optimized_function(request):
    """最適化されたCloud Function"""
# クライアントは既に初期化済み
bucket = storage_client.bucket('my-bucket')
# 処理実行
result = process_data(request.json)
# Firestore に結果保存
db_client.collection('results').add(result)
return {'status': 'success'}
# 接続プールの設定
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# セッション設定（グローバル）
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)

# メモリ効率的な大容量ファイル処理
import functions_framework
from google.cloud import storage
import csv
import io
@functions_framework.http
def process_large_file(request):
    """大容量ファイルをストリーミング処理"""
bucket_name = request.json['bucket']
file_name = request.json['file']
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(file_name)
# ストリーミング読み込み（メモリ効率的）
processed_count = 0
batch_size = 1000
batch_data = []
with blob.open('r') as file:
csv_reader = csv.DictReader(file)
for row in csv_reader:
# データ処理
processed_row = process_row(row)
batch_data.append(processed_row)
# バッチサイズに達したら処理
if len(batch_data) >= batch_size:
save_batch(batch_data)
batch_data = []
processed_count += batch_size
# 残りのデータを処理
if batch_data:
save_batch(batch_data)
processed_count += len(batch_data)
return {'processed_rows': processed_count}
def process_row(row):
    """行データの処理"""
# 実際の処理ロジック
return row
def save_batch(batch_data):
    """バッチデータの保存"""
# BigQuery や Firestore への保存
pass

import functions_framework
import json
import logging
from datetime import datetime
# 構造化ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@functions_framework.http
def logged_function(request):
    """構造化ログを出力するCloud Function"""
# リクエスト開始ログ
request_id = request.headers.get('X-Cloud-Trace-Context', 'unknown')
logger.info(json.dumps({
'severity': 'INFO',
'message': 'Function execution started',
'request_id': request_id,
'timestamp': datetime.utcnow().isoformat(),
'user_agent': request.headers.get('User-Agent'),
'method': request.method
}))
try:
# メイン処理
result = main_processing(request.json)
# 成功ログ
logger.info(json.dumps({
'severity': 'INFO',
'message': 'Function execution completed',
'request_id': request_id,
'processing_time': result.get('processing_time'),
'records_processed': result.get('count')
}))
return result
except Exception as e:
# エラーログ
logger.error(json.dumps({
'severity': 'ERROR',
'message': 'Function execution failed',
'request_id': request_id,
'error': str(e),
'error_type': type(e).__name__
}))
return {'error': 'Internal server error'}, 500
def main_processing(data):
    """メイン処理"""
# 実際の処理ロジック
return {'status': 'success', 'count': 100, 'processing_time': '1.2s'}

from google.cloud import monitoring_v3
import time
# メトリクス送信
def send_custom_metrics(metric_name, value, labels=None):
    """カスタムメトリクスをCloud Monitoringに送信"""
client = monitoring_v3.MetricServiceClient()
project_name = f"projects/{PROJECT_ID}"
series = monitoring_v3.TimeSeries()
series.metric.type = f"custom.googleapis.com/{metric_name}"
series.resource.type = "cloud_function"
series.resource.labels["function_name"] = FUNCTION_NAME
series.resource.labels["region"] = REGION
if labels:
for key, value in labels.items():
series.metric.labels[key] = value
now = time.time()
seconds = int(now)
nanos = int((now - seconds) * 10 ** 9)
interval = monitoring_v3.TimeInterval(
{"end_time": {"seconds": seconds, "nanos": nanos}}
)
point = monitoring_v3.Point(
{"interval": interval, "value": {"double_value": value}}
)
series.points = [point]
client.create_time_series(name=project_name, time_series=[series])
# 使用例
@functions_framework.http
def monitored_function(request):
start_time = time.time()
try:
# 処理実行
result = process_request(request)
# 成功メトリクス送信
processing_time = time.time() - start_time
send_custom_metrics("function_success_count", 1)
send_custom_metrics("function_processing_time", processing_time)
return result
except Exception as e:
# エラーメトリクス送信
send_custom_metrics("function_error_count", 1, {"error_type": type(e).__name__})
raise

# 最小権限の原則に基づくIAM設定
# 1. カスタムサービスアカウント作成
gcloud iam service-accounts create cloud-function-sa \
    --display-name="Cloud Function Service Account"
# 2. 必要最小限の権限付与
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:cloud-function-sa@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/storage.objectViewer"
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:cloud-function-sa@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/firestore.user"
# 3. Cloud Function にサービスアカウント適用
gcloud functions deploy my-function \
    --service-account=cloud-function-sa@PROJECT_ID.iam.gserviceaccount.com

import functions_framework
from google.auth.transport import requests
from google.oauth2 import id_token
import jwt
@functions_framework.http
def authenticated_function(request):
    """JWT認証付きCloud Function"""
# Authorization ヘッダーから JWT 取得
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return {'error': 'Missing or invalid authorization header'}, 401
token = auth_header.split(' ')[1]
try:
# JWT検証（Google ID Token の場合）
idinfo = id_token.verify_oauth2_token(
token, requests.Request(), GOOGLE_CLIENT_ID
)
# ユーザー情報取得
user_id = idinfo['sub']
email = idinfo['email']
# 認可チェック
if not is_authorized(user_id, request.path):
return {'error': 'Insufficient permissions'}, 403
# メイン処理実行
result = main_processing(request.json, user_id)
return result
except ValueError as e:
return {'error': 'Invalid token'}, 401
def is_authorized(user_id, resource_path):
    """ユーザーの認可チェック"""
# 実際の認可ロジック
return True
def main_processing(data, user_id):
    """認証済みユーザーの処理"""
return {'status': 'success', 'user_id': user_id}