综合项目：异步订单处理平台

操作者 → ID token → Cloud Run API → Pub/Sub → 鉴权 push → Cloud Run Worker
                         ↓ GET ├→ Cloud SQL
                    Cloud Storage ←─────────────────────────┘ Secret Manager
                         ↓ 批量加载
                      BigQuery

GitHub OIDC → 发布 SA → Cloud Run revision
Cloud Build → Artifact Registry → 同一镜像的 api / worker 两种运行模式
日志 → Cloud Logging → 指标 / 告警 → Cloud Monitoring

source lab.env
export PROJECT_NUMBER=$(gcloud projects describe "$PROJECT_ID" --format='value(projectNumber)')
export HUMAN_EMAIL=$(gcloud config get-value account)
export ORDERS_BUCKET="${PROJECT_ID}-orders"
export SOURCE_BUCKET="${PROJECT_ID}-orders-source"
export API_SA="orders-api@$PROJECT_ID.iam.gserviceaccount.com"
export WORKER_SA="orders-worker@$PROJECT_ID.iam.gserviceaccount.com"
export PUSH_SA="orders-push@$PROJECT_ID.iam.gserviceaccount.com"
export BUILD_SA="orders-build@$PROJECT_ID.iam.gserviceaccount.com"
export CALLER_SA="orders-caller@$PROJECT_ID.iam.gserviceaccount.com"
gcloud services enable run.googleapis.com artifactregistry.googleapis.com cloudbuild.googleapis.com storage.googleapis.com sqladmin.googleapis.com secretmanager.googleapis.com pubsub.googleapis.com bigquery.googleapis.com logging.googleapis.com monitoring.googleapis.com
for name in orders-api orders-worker orders-push orders-build orders-caller; do
  gcloud iam service-accounts create "$name"
done
# 操作者可挂载这四个 SA；caller 只用于 通过 impersonation 调用 。
for sa in "$API_SA" "$WORKER_SA" "$PUSH_SA" "$BUILD_SA"; do
  gcloud iam service-accounts add-iam-policy-binding "$sa" --member="user:$HUMAN_EMAIL" --role=roles/iam.serviceAccountUser
done
gcloud iam service-accounts add-iam-policy-binding "$CALLER_SA" --member="user:$HUMAN_EMAIL" --role=roles/iam.serviceAccountTokenCreator
gcloud storage buckets create "gs://$ORDERS_BUCKET" --location="$REGION" --uniform-bucket-level-access --public-access-prevention
gcloud storage buckets create "gs://$SOURCE_BUCKET" --location="$REGION" --uniform-bucket-level-access --public-access-prevention
gcloud artifacts repositories create orders-images --location="$REGION" --repository-format=docker
gcloud pubsub topics create orders-created
gcloud pubsub topics create orders-dead
gcloud pubsub subscriptions create orders-dead-inspect --topic=orders-dead

gcloud pubsub topics add-iam-policy-binding orders-created --member="serviceAccount:$API_SA" --role=roles/pubsub.publisher
gcloud storage buckets add-iam-policy-binding "gs://$ORDERS_BUCKET" --member="serviceAccount:$API_SA" --role=roles/storage.objectViewer
gcloud storage buckets add-iam-policy-binding "gs://$ORDERS_BUCKET" --member="serviceAccount:$WORKER_SA" --role=roles/storage.objectCreator
gcloud storage buckets add-iam-policy-binding "gs://$ORDERS_BUCKET" --member="serviceAccount:$WORKER_SA" --role=roles/storage.objectViewer
gcloud projects add-iam-policy-binding "$PROJECT_ID" --member="serviceAccount:$WORKER_SA" --role=roles/cloudsql.client
gcloud artifacts repositories add-iam-policy-binding orders-images --location="$REGION" --member="serviceAccount:$BUILD_SA" --role=roles/artifactregistry.writer
gcloud storage buckets add-iam-policy-binding "gs://$SOURCE_BUCKET" --member="serviceAccount:$BUILD_SA" --role=roles/storage.objectViewer
gcloud projects add-iam-policy-binding "$PROJECT_ID" --member="serviceAccount:$BUILD_SA" --role=roles/logging.logWriter

gcloud sql instances create orders-pg --database-version=POSTGRES_16 --edition=ENTERPRISE --tier=db-f1-micro --region="$REGION" --storage-size=10GB --availability-type=zonal --backup-start-time=20:00
gcloud sql databases create orders --instance=orders-pg
DB_PASSWORD=$(python3 -c 'import secrets; print(secrets.token_urlsafe(32))')
gcloud sql users create orders --instance=orders-pg --password="$DB_PASSWORD"
printf '%s' "$DB_PASSWORD" | gcloud secrets create orders-db-password --data-file=- --replication-policy=automatic
unset DB_PASSWORD
gcloud secrets add-iam-policy-binding orders-db-password --member="serviceAccount:$WORKER_SA" --role=roles/secretmanager.secretAccessor
export CONNECTION_NAME=$(gcloud sql instances describe orders-pg --format='value(connectionName)')

import base64
import json
import os
import uuid
from datetime import datetime, timezone

import psycopg
from flask import Flask, jsonify, request
from google.api_core.exceptions import NotFound, PreconditionFailed
from google.cloud import pubsub_v1, storage

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 65536
MODE = os.environ.get('MODE', 'api')
PROJECT = os.environ['PROJECT_ID']
BUCKET = os.environ['BUCKET']
store = storage.Client()
publisher = pubsub_v1.PublisherClient() if MODE == 'api' else None


def validate(value):
    if not isinstance(value, dict):
        raise ValueError('object required')
    order_id = str(uuid.UUID(str(value['id'])))
    amount = value['amount']
    if type(amount) is not int or not 0 < amount <= 100000000:
        raise ValueError('amount must be positive integer cents')
    return order_id, amount


@app.get('/healthz')
def health():
    return {'status': 'ok', 'mode': MODE}


@app.post('/orders')
def create():
    if MODE != 'api':
        return '', 404
    try:
        order_id, amount = validate(request.get_json())
    except (ValueError, KeyError, TypeError):
        return {'error': 'valid UUID id and integer amount required'}, 400
    event = {'id': order_id, 'amount': amount,
             'created_at': datetime.now(timezone.utc).isoformat()}
    topic = publisher.topic_path(PROJECT, os.environ['TOPIC'])
    publisher.publish(topic, json.dumps(event).encode()).result(timeout=20)
    print(json.dumps({'severity': 'INFO', 'event': 'accepted', 'order_id': order_id}), flush=True)
    return {'id': order_id, 'status': 'accepted'}, 202


@app.get('/orders/<order_id>')
def get_order(order_id):
    if MODE != 'api':
        return '', 404
    try:
        order_id = str(uuid.UUID(order_id))
    except ValueError:
        return {'error': 'invalid UUID'}, 400
    try:
        data = store.bucket(BUCKET).blob(f'orders/{order_id}.json').download_as_text()
        return jsonify(json.loads(data))
    except NotFound:
        return {'status': 'not_ready_or_unknown'}, 404


@app.post('/push')
def consume():
    if MODE != 'worker':
        return '', 404
    order_id = None
    try:
        envelope = request.get_json()
        event = json.loads(base64.b64decode(envelope['message']['data'], validate=True))
        order_id, amount = validate(event)
        created_at = datetime.fromisoformat(event['created_at'])
        if created_at.tzinfo is None:
            raise ValueError('timestamp needs timezone')
        # 教学简化：首次请求建表。生产由独立 migration 身份预建表。
        with psycopg.connect(host=os.environ['DB_HOST'], dbname='orders',
                             user=os.environ.get('DB_USER', 'orders'), password=os.environ['DB_PASSWORD'],
                             connect_timeout=10,
                             sslmode=os.environ.get('DB_SSLMODE', 'prefer')) as conn:
            with conn.cursor() as cur:
                if os.environ.get('AUTO_MIGRATE', 'true') == 'true':
                    cur.execute('''CREATE TABLE IF NOT EXISTS orders (
                        id uuid PRIMARY KEY, amount bigint NOT NULL,
                        created_at timestamptz NOT NULL)''')
                cur.execute('''INSERT INTO orders VALUES (%s, %s, %s)
                    ON CONFLICT (id) DO NOTHING''', (order_id, amount, created_at))
                cur.execute('SELECT amount, created_at FROM orders WHERE id=%s', (order_id,))
                saved_amount, saved_time = cur.fetchone()
                if saved_amount != amount:
                    raise ValueError('same id with different amount')
        # SQL 已提交。重投会重新尝试归档，不会因数据库已有行而跳过归档。
        record = {'id': order_id, 'amount': saved_amount, 'created_at': saved_time.isoformat()}
        blob = store.bucket(BUCKET).blob(f'orders/{order_id}.json')
        try:
            blob.upload_from_string(json.dumps(record) + '\n',
                                    content_type='application/json', if_generation_match=0)
        except PreconditionFailed:
            if json.loads(blob.download_as_text()) != record:
                raise ValueError('archive conflict')
        print(json.dumps({'severity': 'INFO', 'event': 'completed', 'order_id': order_id}), flush=True)
        return '', 204
    except Exception as error:
        # 不输出密码、完整事件或可能包含凭据的连接字符串。
        print(json.dumps({'severity': 'ERROR', 'event': 'processing_failed',
                          'order_id': order_id, 'error_type': type(error).__name__}), flush=True)
        return {'status': 'retry'}, 500

Flask>=3.1,<4
gunicorn>=23,<24
google-cloud-storage>=3,<4
google-cloud-pubsub>=2.28,<3
psycopg[binary]>=3.2,<4

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
ENV PYTHONUNBUFFERED=1
CMD ["sh", "-c", "exec gunicorn --bind 0.0.0.0:${PORT:-8080} --workers 1 --threads 4 --timeout 60 app:app"]

steps:
- name: gcr.io/cloud-builders/docker
  args: ['build', '-t', '${_IMAGE}', '.']
images: ['${_IMAGE}']
options:
  logging: CLOUD_LOGGING_ONLY

export IMAGE="${REGION}-docker.pkg.dev/${PROJECT_ID}/orders-images/app:v1"
gcloud builds submit orders-app --region="$REGION" --config=orders-app/cloudbuild.yaml --service-account="projects/$PROJECT_ID/serviceAccounts/$BUILD_SA" --gcs-source-staging-dir="gs://$SOURCE_BUCKET/source" --substitutions="_IMAGE=$IMAGE"
export DIGEST=$(gcloud artifacts docker images describe "$IMAGE" --format='value(image_summary.digest)')
export RELEASE_IMAGE="${REGION}-docker.pkg.dev/${PROJECT_ID}/orders-images/app@$DIGEST"
gcloud run deploy orders-worker --region="$REGION" --image="$RELEASE_IMAGE" --service-account="$WORKER_SA" --no-allow-unauthenticated --execution-environment=gen2 --add-cloudsql-instances="$CONNECTION_NAME" --set-secrets=DB_PASSWORD=orders-db-password:1 --set-env-vars="MODE=worker,PROJECT_ID=$PROJECT_ID,BUCKET=$ORDERS_BUCKET,DB_HOST=/cloudsql/$CONNECTION_NAME" --concurrency=4 --min=0 --max=2 --timeout=60
gcloud run deploy orders-api --region="$REGION" --image="$RELEASE_IMAGE" --service-account="$API_SA" --no-allow-unauthenticated --set-env-vars="MODE=api,PROJECT_ID=$PROJECT_ID,BUCKET=$ORDERS_BUCKET,TOPIC=orders-created" --min=0 --max=2 --timeout=60
export API_URL=$(gcloud run services describe orders-api --region="$REGION" --format='value(status.url)')
export WORKER_URL=$(gcloud run services describe orders-worker --region="$REGION" --format='value(status.url)')
gcloud run services add-iam-policy-binding orders-api --region="$REGION" --member="serviceAccount:$CALLER_SA" --role=roles/run.invoker
gcloud run services add-iam-policy-binding orders-worker --region="$REGION" --member="serviceAccount:$PUSH_SA" --role=roles/run.invoker

gcloud beta services identity create --service=pubsub.googleapis.com --project="$PROJECT_ID"
export PUBSUB_AGENT="service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com"
# Pub/Sub service agent 可为指定 push SA 生成 OIDC token。
gcloud iam service-accounts add-iam-policy-binding "$PUSH_SA" --member="serviceAccount:$PUBSUB_AGENT" --role=roles/iam.serviceAccountTokenCreator
gcloud pubsub topics add-iam-policy-binding orders-dead --member="serviceAccount:$PUBSUB_AGENT" --role=roles/pubsub.publisher
gcloud pubsub subscriptions create orders-worker-sub --topic=orders-created --push-endpoint="$WORKER_URL/push" --push-auth-service-account="$PUSH_SA" --push-auth-token-audience="$WORKER_URL" --ack-deadline=60 --min-retry-delay=10s --max-retry-delay=60s --dead-letter-topic="projects/$PROJECT_ID/topics/orders-dead" --max-delivery-attempts=5
gcloud pubsub subscriptions add-iam-policy-binding orders-worker-sub --member="serviceAccount:$PUBSUB_AGENT" --role=roles/pubsub.subscriber

TOKEN=$(gcloud auth print-identity-token --impersonate-service-account="$CALLER_SA" --audiences="$API_URL")
export ORDER_ID=$(python3 -c 'import uuid; print(uuid.uuid4())')
curl --fail-with-body -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' -d "{\"id\":\"$ORDER_ID\",\"amount\":1990}" "$API_URL/orders"
# 等待处理，再查询；未完成时 404，稍后重试。
curl --fail-with-body -H "Authorization: Bearer $TOKEN" "$API_URL/orders/$ORDER_ID"
# 重投相同 id 与金额
curl --fail-with-body -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' -d "{\"id\":\"$ORDER_ID\",\"amount\":1990}" "$API_URL/orders"
unset TOKEN
gcloud storage ls "gs://$ORDERS_BUCKET/orders/$ORDER_ID.json"
gcloud logging read 'resource.type="cloud_run_revision" AND jsonPayload.order_id="'"$ORDER_ID"'"' --freshness=1h --limit=20 --format=json

SELECT id, count(*) FROM orders GROUP BY id;

bq --location="$REGION" mk --dataset "$PROJECT_ID:orders_analytics"
# replace 让重复全量加载不产生重复报表行；仅用于该小型实验表。
bq --location="$REGION" load --replace --source_format=NEWLINE_DELIMITED_JSON --time_partitioning_field=created_at "$PROJECT_ID:orders_analytics.orders" "gs://$ORDERS_BUCKET/orders/*.json" id:STRING,amount:INTEGER,created_at:TIMESTAMP
bq --location="$REGION" query --use_legacy_sql=false --maximum_bytes_billed=100000000 "SELECT COUNT(*) orders, SUM(amount) cents FROM \`$PROJECT_ID.orders_analytics.orders\`"
gcloud logging metrics create orders_failures --log-filter='resource.type="cloud_run_revision" AND resource.labels.service_name="orders-worker" AND jsonPayload.event="processing_failed"'

gcloud run services describe orders-api --region="$REGION" --format='yaml(status.traffic)'
# Copy the currently serving 100% revision; latestReadyRevisionName can be an unused candidate.
export OLD_REV='REPLACE_CURRENT_SERVING_REVISION'
gcloud run deploy orders-api --region="$REGION" --image="$RELEASE_IMAGE" --no-traffic --tag=candidate
gcloud run services describe orders-api --region="$REGION" --format='yaml(status.traffic)'

gcloud run services update-traffic orders-api --region="$REGION" --to-latest

gcloud run services update-traffic orders-api --region="$REGION" --to-revisions="$OLD_REV=100"

gcloud run services update orders-api --region="$REGION" --ingress=internal-and-cloud-load-balancing
TOKEN=$(gcloud auth print-identity-token --impersonate-service-account="$CALLER_SA" --audiences="$API_URL")
curl --fail -H "Authorization: Bearer $TOKEN" "https://$DOMAIN/healthz"
unset TOKEN

export PROJECT_NUMBER=$(gcloud projects describe "$PROJECT_ID" --format='value(projectNumber)')
export KSA_PRINCIPAL="principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${PROJECT_ID}.svc.id.goog/subject/ns/lab/sa/app"
gcloud artifacts repositories add-iam-policy-binding orders-images --location="$REGION" --member="serviceAccount:lab-gke-node@$PROJECT_ID.iam.gserviceaccount.com" --role=roles/artifactregistry.reader
gcloud pubsub topics add-iam-policy-binding orders-created --member="$KSA_PRINCIPAL" --role=roles/pubsub.publisher
gcloud storage buckets add-iam-policy-binding "gs://$ORDERS_BUCKET" --member="$KSA_PRINCIPAL" --role=roles/storage.objectViewer
kubectl -n lab create deployment orders-api --image="$RELEASE_IMAGE" --dry-run=client -o yaml > orders-gke.yaml
# 先生成清单再启动，避免应用缺环境变量启动失败。
python3 - <<'GKEPY'
import os, yaml
path = 'orders-gke.yaml'
data = yaml.safe_load(open(path))
spec = data['spec']['template']['spec']
spec['serviceAccountName'] = 'app'
spec['containers'][0]['env'] = [{'name': k, 'value': v} for k, v in {
    'MODE': 'api', 'PROJECT_ID': os.environ['PROJECT_ID'],
    'BUCKET': os.environ['ORDERS_BUCKET'], 'TOPIC': 'orders-created'}.items()]
spec['containers'][0]['resources'] = {'requests': {'cpu': '250m', 'memory': '512Mi'},
                                    'limits': {'cpu': '1', 'memory': '512Mi'}}
with open(path, 'w') as f:
    yaml.safe_dump(data, f)
GKEPY
kubectl apply -f orders-gke.yaml
kubectl -n lab rollout status deployment/orders-api --timeout=600s
kubectl -n lab port-forward deployment/orders-api 8081:8080

terraform import google_storage_bucket.archive "$ORDERS_BUCKET"
terraform plan

gcloud pubsub subscriptions delete orders-worker-sub orders-dead-inspect
gcloud run services delete orders-api --region="$REGION"
gcloud run services delete orders-worker --region="$REGION"
gcloud pubsub topics delete orders-created orders-dead
gcloud sql instances delete orders-pg
gcloud secrets delete orders-db-password
bq rm -r -f -d "$PROJECT_ID:orders_analytics"
gcloud artifacts repositories delete orders-images --location="$REGION"
gcloud storage rm --recursive "gs://$SOURCE_BUCKET"

gcloud logging metrics delete orders_failures
gcloud projects remove-iam-policy-binding "$PROJECT_ID" --member="serviceAccount:$WORKER_SA" --role=roles/cloudsql.client
gcloud projects remove-iam-policy-binding "$PROJECT_ID" --member="serviceAccount:$BUILD_SA" --role=roles/logging.logWriter
for sa in "$API_SA" "$WORKER_SA" "$PUSH_SA" "$BUILD_SA" "$CALLER_SA"; do
  gcloud iam service-accounts delete "$sa"
done