B システム設計/インフラ — Cloud Spanner デュアルリージョン（asia1）× PITR version_retention_period=7d × BurstyPool セッションプール × Argo Workflows CronWorkflow リトライ指数バックオフ × Spot+On-demand フォールバック × IAM 最小権限（spanner.databaseReader+User）× Workload Identity Federation（MOps バッチ集計基盤 Bad→Good）

📋 コピー# ksa.yaml — Kubernetes Service Account（Workload Identity 連携）
apiVersion: v1
kind: ServiceAccount
metadata:
  name: mops-spanner-batch
  namespace: mops
  annotations:
    # 修正⑦: KSA → GSA のバインド
    iam.gke.io/gcp-service-account: mops-spanner-batch@PROJECT_ID.iam.gserviceaccount.com
---
# cronworkflow.yaml — CronWorkflow（改善後）
apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: campaign-spanner-batch
  namespace: mops
  labels:
    app: campaign-spanner-batch
    team: mops
spec:
  schedule: "0 2 * * *"           # 毎日 JST 11:00 (UTC 02:00)
  timezone: "Asia/Tokyo"
  concurrencyPolicy: Forbid       # 修正: 前回実行中は新規をスキップ
  successfulJobsHistoryLimit: 5
  failedJobsHistoryLimit: 5
  startingDeadlineSeconds: 300    # 起動遅延許容5分

  workflowSpec:
    # 修正⑦: 専用 KSA（デフォルト SA を使わない）
    serviceAccountName: mops-spanner-batch

    entrypoint: main

    podMetadata:
      labels:
        tags.datadoghq.com/service: "campaign-spanner-batch"
        tags.datadoghq.com/env: "production"
        tags.datadoghq.com/version: "1.2.0"

    arguments:
      parameters:
        - name: report_date
          value: "{{=sprig.dateModify(\"-24h\", sprig.toDate(\"2006-01-02\", workflow.scheduledTime)) | sprig.date(\"2006-01-02\")}}"

    templates:
      - name: main
        steps:
          - - name: aggregate
              template: run-aggregate

      - name: run-aggregate
        inputs:
          parameters:
            - name: report_date
              value: "{{workflow.parameters.report_date}}"

        # 修正⑤: リトライ戦略（指数バックオフ）
        retryStrategy:
          limit: "3"
          retryPolicy: "OnFailure"
          backoff:
            duration: "30s"       # 初回リトライ待機
            factor: "2"           # 2倍ずつ延長（30s → 60s → 120s）
            maxDuration: "5m"     # 最大5分

        # 修正④: Spot 優先・On-demand フォールバック
        podSpecPatch: |
          tolerations:
            - key: cloud.google.com/gke-spot
              operator: Equal
              value: "true"
              effect: NoSchedule
          affinity:
            nodeAffinity:
              # Spot を「優先」するが必須ではない（On-demand にフォールバック）
              preferredDuringSchedulingIgnoredDuringExecution:
                - weight: 80
                  preference:
                    matchExpressions:
                      - key: cloud.google.com/gke-spot
                        operator: In
                        values: ["true"]

        container:
          # digest ピン留め（latest タグ禁止）
          image: asia-northeast1-docker.pkg.dev/PROJECT_ID/mops/spanner-batch@sha256:abc123
          imagePullPolicy: IfNotPresent

          env:
            - name: GOOGLE_CLOUD_PROJECT
              value: "PROJECT_ID"
            - name: SPANNER_INSTANCE_ID
              value: "campaign-spanner"
            - name: SPANNER_DATABASE_ID
              value: "campaign_db"
            - name: REPORT_DATE
              value: "{{inputs.parameters.report_date}}"
            - name: DD_AGENT_HOST
              valueFrom:
                fieldRef:
                  fieldPath: status.hostIP

          resources:
            requests:
              cpu: "500m"
              memory: "512Mi"
            limits:
              memory: "1Gi"

          securityContext:
            runAsNonRoot: true
            readOnlyRootFilesystem: true
            allowPrivilegeEscalation: false

    # バッチ全体タイムアウト（1時間）
    activeDeadlineSeconds: 3600

📋 コピー"""campaign_spanner_batch.py — Cloud Spanner 集計バッチ（改善後）"""
from __future__ import annotations

import logging
import os
from datetime import date
from typing import Final

import structlog
from google.cloud import spanner
from google.cloud.spanner_v1 import param_types

# ── 定数 ───────────────────────────────────────────────
SPANNER_PROJECT: Final[str] = os.environ["GOOGLE_CLOUD_PROJECT"]
SPANNER_INSTANCE: Final[str] = os.environ["SPANNER_INSTANCE_ID"]
SPANNER_DATABASE: Final[str] = os.environ["SPANNER_DATABASE_ID"]

# 修正③: BurstyPool パラメータ
POOL_TARGET_SIZE: Final[int] = 10  # 通常時のセッション数
POOL_MAX_SIZE:    Final[int] = 50  # スパイク時の最大セッション数

structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(),
)
logger = structlog.get_logger()


def build_spanner_client() -> tuple[spanner.Client, spanner.Database]:
    """Cloud Spanner クライアントを BurstyPool で初期化する。

    Workload Identity Federation により SA キーファイル不要。
    ADC (Application Default Credentials) が自動的に KSA → GSA として認証する。

    Returns:
        (client, database) のタプル
    """
    client = spanner.Client(project=SPANNER_PROJECT)
    instance = client.instance(SPANNER_INSTANCE)

    # 修正③: BurstyPool — 通常 target_size、スパイク時は max_size まで拡張
    pool = spanner.BurstyPool(
        target_size=POOL_TARGET_SIZE,
        max_size=POOL_MAX_SIZE,
    )
    database = instance.database(SPANNER_DATABASE, pool=pool)
    return client, database


def aggregate_campaign_orders(
    database: spanner.Database,
    report_date: date,
) -> list[dict]:
    """campaign_orders を日次集計する（読み取り専用トランザクション）。

    Args:
        database: Cloud Spanner データベースクライアント
        report_date: 集計対象日

    Returns:
        集計結果のリスト（campaign_id, total_amount, order_count）
    """
    log = logger.bind(report_date=str(report_date))
    log.info("aggregate_started")

    # 読み取り専用トランザクション: ロック不要でスループット高
    with database.snapshot() as snapshot:
        results = snapshot.execute_sql(
            """
            SELECT
                campaign_id,
                SUM(order_amount) AS total_amount,
                COUNT(*)          AS order_count
            FROM campaign_orders
            WHERE report_date = @report_date
            GROUP BY campaign_id
            """,
            params={"report_date": report_date},
            param_types={"report_date": param_types.DATE},
        )
        rows = list(results)

    log.info("aggregate_query_done", row_count=len(rows))
    return [
        {"campaign_id": row[0], "total_amount": row[1], "order_count": row[2]}
        for row in rows
    ]


def upsert_campaign_summary(
    database: spanner.Database,
    summaries: list[dict],
    report_date: date,
) -> None:
    """集計結果を campaign_summary テーブルに冪等 Upsert する。

    insert_or_update は既存行があれば更新、なければ挿入（冪等性確保）。
    リトライが発生しても二重書き込みにならない。

    Args:
        database: Cloud Spanner データベースクライアント
        summaries: aggregate_campaign_orders の返り値
        report_date: 集計対象日
    """
    log = logger.bind(report_date=str(report_date), row_count=len(summaries))

    with database.batch() as batch:
        for summary in summaries:
            batch.insert_or_update(
                table="campaign_summary",
                columns=(
                    "campaign_id", "report_date",
                    "total_amount", "order_count", "updated_at",
                ),
                values=[(
                    summary["campaign_id"],
                    report_date,
                    summary["total_amount"],
                    summary["order_count"],
                    spanner.COMMIT_TIMESTAMP,  # サーバーサイド TrueTime タイムスタンプ
                )],
            )

    log.info("upsert_done")


def main() -> None:
    """バッチエントリポイント。"""
    report_date_str = os.environ.get("REPORT_DATE", "")
    if not report_date_str:
        raise ValueError("REPORT_DATE 環境変数が未設定")

    report_date = date.fromisoformat(report_date_str)
    _client, database = build_spanner_client()

    summaries = aggregate_campaign_orders(database, report_date)
    upsert_campaign_summary(database, summaries, report_date)

    logger.info(
        "batch_completed",
        report_date=report_date_str,
        campaign_count=len(summaries),
    )


if __name__ == "__main__":
    main()