セクション4：Google Cloud サービスとの統合

公式試験ガイド 042426版 — Section 4: Integrating applications with Google Cloud services

4.1 データ・ストレージサービスとの統合

Cloud SQL 接続：Auth Proxy vs Connector

Cloud Run / GKE / GCE から Cloud SQL に接続する方法には、Auth Proxy（プロセス分離）とConnector ライブラリ（プロセス内）の2つがある。

Cloud Run + Auth Proxy（推奨）

# Cloud Run へのデプロイで Cloud SQL を関連付け
gcloud run deploy my-service \
  --image=us-central1-docker.pkg.dev/PROJECT/repo/app:v1 \
  --add-cloudsql-instances=PROJECT:us-central1:my-instance \
  --set-env-vars="INSTANCE_CONNECTION_NAME=PROJECT:us-central1:my-instance" \
  --region=us-central1

# アプリ側: Unix Socket で接続
import psycopg2
conn = psycopg2.connect(
    host=f"/cloudsql/{os.environ['INSTANCE_CONNECTION_NAME']}",
    user="my-user",
    password=db_password,    # Secret Manager から取得
    database="my-db"
)

Cloud SQL Connector（Python の例）

from google.cloud.sql.connector import Connector, IPTypes
import sqlalchemy

connector = Connector(ip_type=IPTypes.PRIVATE)   # Private IP 推奨

def getconn():
    return connector.connect(
        "PROJECT:us-central1:my-instance",
        "pg8000",
        user="my-user",
        password=db_password,
        db="my-db",
        enable_iam_auth=True   # IAM 認証
    )

engine = sqlalchemy.create_engine("postgresql+pg8000://", creator=getconn)

選択の指針

Firestore クライアントの典型パターン

from google.cloud import firestore

db = firestore.Client()

# 書き込み（自動 created/updated）
db.collection("users").document("u1").set({
    "name": "Alice",
    "age": 30,
    "createdAt": firestore.SERVER_TIMESTAMP,
})

# トランザクション（単一文書 ACID）
@firestore.transactional
def update_balance(transaction, user_ref, amount):
    snapshot = user_ref.get(transaction=transaction)
    new_balance = snapshot.get("balance") + amount
    transaction.update(user_ref, {"balance": new_balance})

user_ref = db.collection("users").document("u1")
update_balance(db.transaction(), user_ref, 1000)

# クエリ + ページング
query = db.collection("users").where("age", ">", 18).order_by("createdAt").limit(100)
docs = query.stream()
for doc in docs:
    print(doc.id, doc.to_dict())

Pub/Sub クライアントライブラリの典型パターン

# Publish（バッチ + 非同期）
from google.cloud import pubsub_v1
from concurrent import futures

publisher = pubsub_v1.PublisherClient(
    publisher_options=pubsub_v1.types.PublisherOptions(
        enable_message_ordering=False,
        flow_control=pubsub_v1.types.PublishFlowControl(
            message_limit=1000,
            byte_limit=10 * 1024 * 1024,
            limit_exceeded_behavior=pubsub_v1.types.LimitExceededBehavior.BLOCK,
        ),
    ),
    batch_settings=pubsub_v1.types.BatchSettings(
        max_messages=100,
        max_bytes=1024 * 1024,
        max_latency=0.1,
    ),
)

topic_path = publisher.topic_path("PROJECT", "my-topic")
publish_futures = []
for msg in messages:
    future = publisher.publish(topic_path, msg.encode("utf-8"), attribute_key="value")
    publish_futures.append(future)
# 全部の publish の完了待ち
futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)

# Subscribe（streaming pull）
from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path("PROJECT", "my-sub")

flow_control = pubsub_v1.types.FlowControl(
    max_messages=100,           # 1度に受け取る最大メッセージ数
)

def callback(message):
    try:
        process(message.data)
        message.ack()           # 処理成功
    except TransientError:
        message.nack()          # 即座に再配信
    except Exception as e:
        # 処理失敗 → ack せずデッドラインで自動再配信、または DLT へ

streaming_pull_future = subscriber.subscribe(
    subscription_path, callback=callback, flow_control=flow_control)

with subscriber:
    try:
        streaming_pull_future.result(timeout=None)
    except KeyboardInterrupt:
        streaming_pull_future.cancel()

4.2 Google Cloud API の利用

Application Default Credentials (ADC) の優先順位

サービスごとのデフォルト SA

リトライと指数バックオフ — 公式推奨パターン

HTTP / gRPC コードとリトライ可否

指数バックオフ + ジッターの実装

from google.api_core import retry

# Cloud Client Libraries の組み込みリトライ
my_retry = retry.Retry(
    predicate=retry.if_exception_type(
        google.api_core.exceptions.ServiceUnavailable,      # 503
        google.api_core.exceptions.DeadlineExceeded,        # 504 / DEADLINE_EXCEEDED
        google.api_core.exceptions.ResourceExhausted,       # 429 / RESOURCE_EXHAUSTED
        google.api_core.exceptions.Aborted,                  # 409 ABORTED
    ),
    initial=1.0,                # 初回 1秒
    maximum=60.0,               # 最大 60秒
    multiplier=2.0,             # 2倍ずつ
    deadline=600.0,             # 全体 10分まで
)

# クライアントメソッドに渡す
result = client.some_operation(request, retry=my_retry, timeout=30)

Full Jitter の自前実装（参考）

import random
import time
from typing import Callable, TypeVar

T = TypeVar("T")

def retry_with_jitter(
    fn: Callable[[], T],
    max_retries: int = 5,
    initial_delay: float = 1.0,
    max_delay: float = 60.0,
) -> T:
    for attempt in range(max_retries):
        try:
            return fn()
        except TransientError:
            if attempt == max_retries - 1:
                raise
            # Full Jitter: random(0, min(max_delay, initial * 2^attempt))
            delay = min(max_delay, initial_delay * (2 ** attempt))
            sleep_time = random.uniform(0, delay)
            time.sleep(sleep_time)

ジッター戦略の比較

ページネーション・バッチング・キャッシング

ページネーションの 3 スタイル

SELECT * FROM users
LIMIT 20 OFFSET 100

response = client.list(
    page_size=100, page_token=token)
token = response.next_page_token

SELECT * FROM users
WHERE id > $last_id
ORDER BY id LIMIT 20

Cloud Client Library の自動ページネーション

from google.cloud import storage

client = storage.Client()
bucket = client.bucket("my-bucket")

# パターン A: iterator が自動でページング（メモリ効率○）
for blob in bucket.list_blobs(page_size=100):
    process(blob)

# パターン B: ページ単位で明示的に扱う
pages = bucket.list_blobs(page_size=100).pages
for page in pages:
    print(f"Got a page of {len(list(page))} blobs")
    for blob in page:
        process(blob)

バッチング（Firestore の例）

# Firestore のバッチ書き込み（上限 500 文書）
batch = db.batch()
for i, doc in enumerate(docs):
    if i > 0 and i % 500 == 0:
        batch.commit()         # 500 件ごとにコミット
        batch = db.batch()
    batch.set(db.collection("c").document(doc["id"]), doc)
batch.commit()

キャッシング (Cache-Aside パターン)

import json
import redis

cache = redis.Redis(host="MEMORYSTORE_IP", decode_responses=True)
TTL = 300   # 5分

def get_user(user_id: str) -> dict:
    # 1. キャッシュを覗く
    cached = cache.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    # 2. なければ DB から取得
    user = db.collection("users").document(user_id).get().to_dict()

    # 3. キャッシュに書く
    cache.setex(f"user:{user_id}", TTL, json.dumps(user))
    return user

4.3 トラブルシューティングと可観測性

OpenTelemetry → Cloud Trace 計装

OpenTelemetry (OTel) はベンダーニュートラルな計装フレームワーク。書いたコードを Cloud Trace、Datadog、NewRelic 等にエクスポート可能。

Python での OTel 自動計装

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.propagators.cloud_trace_propagator import CloudTraceFormatPropagator
from opentelemetry import propagate

# Tracer Provider
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(CloudTraceSpanExporter(project_id="PROJECT_ID"))
)

# W3C Trace Context（標準）+ GCP X-Cloud-Trace-Context（互換）
propagate.set_global_textmap(CloudTraceFormatPropagator())

# Flask / requests を自動計装
app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)   # 各エンドポイントが自動 span
RequestsInstrumentor().instrument()        # requests ライブラリの呼び出しも自動 span

# 手動計装
tracer = trace.get_tracer(__name__)

@app.route("/orders/")
def get_order(id):
    with tracer.start_as_current_span("get_order_details") as span:
        span.set_attribute("order.id", id)
        order = db.collection("orders").document(id).get()
        span.set_attribute("order.total", order.get("total"))
        return order.to_dict()

W3C Trace Context によるサービス境界の伝播

traceparent ヘッダの構造

traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01
            ^^  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^ ^^
            |   |                                |                |
            |   |                                |                +-- flags (01 = sampled)
            |   |                                +-- parent-id (span-id, 16 hex)
            |   +-- trace-id (32 hex, リクエスト全体で同じ)
            +-- version (00)

構造化ログ（Cloud Logging）

JSON で stdout に出力するだけで、Cloud Logging が severity / trace / labels / httpRequest を自動認識する。

特殊フィールド一覧

Python での実装例（OTel と統合）

import json
import sys
from opentelemetry import trace

PROJECT_ID = "my-project"

def log_structured(severity: str, message: str, **kwargs):
    span = trace.get_current_span()
    ctx = span.get_span_context() if span else None

    entry = {
        "severity": severity,
        "message": message,
        **kwargs,
    }
    if ctx and ctx.is_valid:
        entry["logging.googleapis.com/trace"] = f"projects/{PROJECT_ID}/traces/{ctx.trace_id:032x}"
        entry["logging.googleapis.com/spanId"] = f"{ctx.span_id:016x}"
        entry["logging.googleapis.com/trace_sampled"] = bool(ctx.trace_flags & 0x01)

    print(json.dumps(entry), file=sys.stdout, flush=True)

# 使い方
log_structured("INFO", "User logged in", user_id="u1", method="oauth")
log_structured("ERROR", "Payment failed", order_id="o123",
               error_code="CARD_DECLINED")

# 特定ユーザーの ERROR 以上を検索
severity>=ERROR
resource.type="cloud_run_revision"
resource.labels.service_name="my-service"
timestamp >= "2026-05-28T00:00:00Z"
jsonPayload.user_id = "u1"

# trace ID で全 logs を見る
trace="projects/my-project/traces/abc123..."

ログから Cloud Trace への遷移

構造化ログに logging.googleapis.com/trace を含めると、Cloud Logging UI から「View in Trace」ボタンが出現。1クリックで該当 trace の全 spans を可視化できる。

Cloud Logging のコスト最適化

Error Reporting

同じスタックトレースを1つのインシデントにグルーピングし、メール / Slack / Pub/Sub に通知。

import logging
import sys
import traceback
import json
import google.cloud.logging

# Cloud Logging ハンドラ
client = google.cloud.logging.Client()
client.setup_logging()

# 未捕捉例外を自動キャプチャ
def excepthook(exc_type, exc_value, exc_tb):
    logging.error(
        "Uncaught exception",
        extra={"json_fields": {"stack_trace": "".join(traceback.format_exception(exc_type, exc_value, exc_tb))}}
    )
sys.excepthook = excepthook

# ERROR ログ + stack_trace が Error Reporting で集約される
try:
    risky_operation()
except Exception:
    logging.error(
        "Payment processing failed",
        exc_info=True,    # 自動でスタックトレース付与
        extra={"json_fields": {"order_id": "o123"}}
    )

Cloud Profiler との関係

🚨 事故ケース集（postmortem 形式）

セクション4 試験戦略まとめ