ljwx-children/PROMPT_BLUEPRINT_V2.md

Claude Code 驱动的生产级开发蓝图 v2.0

项目定位：基于 OrbStack K8s 的 FastAPI 生产级模板项目

核心理念：Claude Code 是主要开发伙伴，本蓝图是 Claude 的工作指南

适用场景：

• 使用本模板创建新的 FastAPI 微服务
• 复用已部署的 K8s 基础设施（Postgres/Redis/监控/日志/链路追踪）
• 通过 Argo CD 实现 GitOps 部署
• Claude Code 辅助完成从设计到上线的全流程

---

🎯 0. 项目目标

• 快速交付：在已有基础设施上快速启动新服务（<1天）
• 生产就绪：自带可观测性、健康检查、优雅停机
• Claude 友好：所有流程可由 Claude Code 辅助完成
• GitOps 驱动：代码即配置，Git 即发布入口

---

🏗️ 1. 基础设施现状（已就绪，无需搭建）

1.1 当前环境

• 集群：OrbStack K8s v1.33.5 (Mac Studio)
• 节点：单节点 orbstack
• 命名空间：27个，包括专用基础设施命名空间

1.2 可用的基础设施服务

服务	命名空间	访问地址	状态	用途
Argo CD	argocd	argocd-server.argocd.svc	✅ 运行中	GitOps 部署
Prometheus	monitoring	prometheus-kube-prometheus-prometheus.monitoring.svc:9090	✅ 运行中	Metrics 监控
Grafana	monitoring	kube-prometheus-stack-grafana.monitoring.svc:80	✅ 运行中	可视化 Dashboard
Loki	logging	loki.logging.svc:3100	✅ 运行中	日志聚合
Promtail	logging	自动采集	✅ 运行中	日志收集
Tempo	tracing	tempo.tracing.svc:3100	✅ 运行中	链路追踪
OpenTelemetry Collector	tracing	otel-collector-opentelemetry-collector.tracing.svc:4317	✅ 运行中	OTLP 接收
Postgres	infra	postgres-infra-postgresql.infra.svc:5432	✅ 运行中	共享数据库
MySQL	infra	mysql-infra.infra.svc:3306	✅ 运行中	共享数据库（备选）
Redis	infra	redis-infra-master.infra.svc:6379	✅ 运行中	缓存/队列
Nginx Ingress	infra	ingress-nginx-controller.infra.svc	✅ 运行中	流量入口

1.3 关键信息

# 查看所有基础设施状态
kubectl get pods -n argocd
kubectl get pods -n monitoring
kubectl get pods -n logging
kubectl get pods -n tracing
kubectl get pods -n infra

# 访问 Grafana（查看 Metrics + Logs）
kubectl port-forward -n monitoring svc/kube-prometheus-stack-grafana 3000:80
# 密码：kubectl get secret -n monitoring kube-prometheus-stack-grafana -o jsonpath="{.data.admin-password}" | base64 --decode

# 访问 Tempo（查看 Traces）
kubectl port-forward -n tracing svc/tempo 16686:16686

# 访问 Argo CD
kubectl port-forward -n argocd svc/argocd-server 8080:443
# 密码：kubectl get secret -n argocd argocd-initial-admin-secret -o jsonpath="{.data.password}" | base64 --decode

---

🤖 2. Claude Code 的角色与能力

2.1 Claude Code 在本项目中的定位

Claude 是你的全栈工程伙伴，负责：

1. 需求分析 → 理解业务需求，提出技术方案
2. 架构设计 → 基于已有基础设施设计服务架构
3. 代码实现 → 编写 FastAPI 代码、数据库模型、API 端点
4. 测试编写 → 生成单元测试、集成测试
5. 可观测性 → 集成 Prometheus Metrics、结构化日志、链路追踪
6. 部署配置 → 生成 Helm Values、Argo CD Application
7. 文档更新 → 更新 API 文档、部署文档

2.2 Claude Code 的工作流

需求输入 → Claude 分析 → 设计方案 → 用户确认 →
→ Claude 实现代码 → Claude 编写测试 → Claude 集成可观测性 →
→ Claude 生成部署配置 → 用户审查 → 部署上线 → Claude 协助验证

2.3 Claude Code 可用的工具和技能

内置工具：

• Read - 读取代码、配置文件
• Edit - 修改代码
• Write - 创建新文件
• Bash - 执行命令（kubectl、helm、git 等）
• Grep - 搜索代码
• Glob - 查找文件

项目自定义技能（.claude/skills/）：

• code-review - 代码审查（安全、质量、类型检查）
• test-generator - 自动生成 pytest 测试
• refactor-assistant - 代码重构建议
• api-designer - 生成 OpenAPI 规范

使用示例：

# 让 Claude 审查代码
"运行 code-review skill 审查 backend/ 目录的代码，重点关注安全性"

# 让 Claude 生成测试
"使用 test-generator skill 为 backend/main.py 生成测试用例"

# 让 Claude 检查重构机会
"运行 refactor-assistant skill 分析代码质量问题"

2.4 Claude Code 的开发模式

模式 A：探索式开发（Explore Agent）

用户："分析项目的数据库集成方式"
Claude：使用 Task tool (subagent_type=Explore) 探索代码库
→ 搜索数据库相关代码
→ 分析配置文件
→ 理解现有模式
→ 给出总结报告

模式 B：规划式开发（Plan Agent）

用户："添加用户认证功能"
Claude：使用 EnterPlanMode 进入规划模式
→ 分析现有架构
→ 设计认证方案（JWT/Session/OAuth2）
→ 列出需要修改的文件
→ 提供实施计划
→ 等待用户确认后执行

模式 C：执行式开发（Direct Execution）

用户："在 backend/main.py 添加 /health/ready 端点"
Claude：直接执行（小型、明确的任务）
→ Read backend/main.py
→ Edit 添加新端点
→ 更新测试
→ 验证语法

---

📐 3. 技术栈与工具链（已确定）

3.1 后端技术栈

语言: Python 3.11+
Web 框架: FastAPI + Uvicorn
ORM: SQLAlchemy 2.0 (async)
数据库迁移: Alembic
验证: Pydantic v2
异步 HTTP: httpx, aiohttp
认证: python-jose (JWT)
密码加密: passlib (bcrypt)

3.2 数据库选择

环境	数据库	连接方式	配置文件
本地开发	SQLite	本地文件	values-orbstack.yaml
生产环境	Postgres (infra namespace)	K8s Service DNS	values-infra-db.yaml

连接示例：

# 开发环境（SQLite）
DATABASE_URL = "sqlite+aiosqlite:///./gold-template.db"

# 生产环境（Postgres in infra namespace）
DATABASE_URL = "postgresql+asyncpg://app_user:${DB_PASSWORD}@postgres-infra-postgresql.infra.svc.cluster.local:5432/goldtemplate"

3.3 开发工具链

代码格式化: black
代码检查: flake8
类型检查: mypy
测试框架: pytest + pytest-asyncio + pytest-cov
安全扫描: bandit, safety
容器构建: Docker + BuildKit
部署工具: Helm + Argo CD

---

🔄 4. 工作方式（Claude Code 驱动）

4.1 开发流程（8步法）

1. 需求描述 → 用户提供功能需求
2. Claude 分析 → Claude 阅读相关代码，理解现有架构
3. 方案设计 → Claude 提出技术方案（含数据库、API、集成点）
4. 用户确认 → 用户审查方案，确认或调整
5. Claude 实现 → Claude 编写代码（后端、数据库迁移、配置）
6. Claude 测试 → Claude 编写测试，运行测试验证
7. Claude 集成 → Claude 添加 Metrics、日志、链路追踪
8. 部署验证 → Claude 生成部署配置，协助验证

4.2 与 Claude 协作的最佳实践

✅ 好的需求描述：

"添加用户认证功能：
- 支持 JWT token 认证
- 用户可以注册、登录、登出
- 密码使用 bcrypt 加密
- Token 有效期 7 天
- 需要数据库迁移脚本
- 需要单元测试和集成测试
- 集成 Prometheus metrics（登录成功/失败计数）
- 参考项目现有的数据库模式"

❌ 不好的需求描述：

"加个登录功能"

✅ 利用 Claude 的上下文能力：

第1轮："分析 backend/models.py 的数据库模型设计"
第2轮："基于现有模型，设计一个订单管理模块"
第3轮："实现订单创建 API，保持与现有代码风格一致"

✅ 使用项目技能：

"使用 code-review skill 检查安全问题"
"使用 test-generator skill 生成测试"

4.3 Claude 必须交付的内容（完整性检查）

对于任何功能需求，Claude 的输出必须包含：

✅ 1. 代码实现

• FastAPI 端点代码
• Pydantic 模型（请求/响应）
• SQLAlchemy 数据库模型（如需要）
• 业务逻辑代码

✅ 2. 数据库迁移（如涉及数据结构变更）

• Alembic 迁移脚本（upgrade/downgrade）
• 遵循 Expand/Contract 模式（生产环境）

✅ 3. 测试

• 至少一个单元测试或集成测试
• 测试覆盖核心逻辑

✅ 4. 可观测性

• Prometheus Metrics（Counter/Histogram/Gauge）
• 结构化日志（包含 trace_id）
• OpenTelemetry spans（如有外部调用）

✅ 5. 部署配置

• Helm values 变更（如需要）
• 环境变量配置
• Secret 管理（如需要）

✅ 6. 文档

• API 文档（FastAPI 自动生成）
• README 更新（如需要）
• 部署文档更新（如需要）

---

🗄️ 5. 数据库版本化与迁移（Alembic）

5.1 迁移工具

使用 Alembic（Python 生态标准）：

# 已安装：alembic==1.13.1

5.2 开发环境 vs 生产环境

环境	数据库	迁移策略	回滚策略
开发	SQLite	允许删除重建	可使用 downgrade
生产	Postgres	Expand/Contract	前滚修复 + PITR 兜底

5.3 生产环境迁移流程（Expand/Contract）

场景：添加新列 email 到 users 表

Phase 1: Expand（扩展）

# alembic/versions/001_add_email_column.py
"""Add email column to users table"""

def upgrade():
    # 新增列，允许 NULL（兼容旧版本应用）
    op.add_column('users', sa.Column('email', sa.String(255), nullable=True))

def downgrade():
    op.drop_column('users', 'email')

# 部署：先部署迁移，再部署应用
# 旧版本应用继续运行（不感知新列）
# 新版本应用可以读写新列

Phase 2: 数据回填（可选）

# alembic/versions/002_backfill_email.py
"""Backfill email for existing users"""

def upgrade():
    # 回填数据（如有默认值逻辑）
    op.execute("""
        UPDATE users
        SET email = username || '@example.com'
        WHERE email IS NULL
    """)

def downgrade():
    pass  # 回填不可逆

Phase 3: Contract（收缩）

# alembic/versions/003_make_email_required.py
"""Make email column required"""

def upgrade():
    # 添加约束（此时所有应用已升级，数据已回填）
    op.alter_column('users', 'email', nullable=False)

def downgrade():
    op.alter_column('users', 'email', nullable=True)

5.4 Claude Code 生成迁移脚本

用户请求：

"为 users 表添加 email 字段：
- 类型：String(255)
- 必填
- 唯一索引
- 需要遵循 Expand/Contract 模式
- 生成 3 个迁移脚本"

Claude 输出：

# 1. Expand - 添加可空列
# alembic/versions/20250101_001_add_email.py
# （完整代码）

# 2. 数据回填
# alembic/versions/20250101_002_backfill_email.py
# （完整代码）

# 3. Contract - 添加约束
# alembic/versions/20250101_003_email_constraints.py
# （完整代码）

5.5 PreSync Job（Argo CD 集成）

deployment/helm/gold-template/templates/migration-job.yaml：

{{- if .Values.database.migrations.enabled -}}
apiVersion: batch/v1
kind: Job
metadata:
  name: {{ include "gold-template.fullname" . }}-migration
  annotations:
    argocd.argoproj.io/hook: PreSync
    argocd.argoproj.io/hook-delete-policy: BeforeHookCreation
    argocd.argoproj.io/sync-wave: "1"
spec:
  ttlSecondsAfterFinished: 300
  backoffLimit: 3
  template:
    metadata:
      name: {{ include "gold-template.fullname" . }}-migration
    spec:
      restartPolicy: Never
      containers:
      - name: migration
        image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
        command: ["alembic", "upgrade", "head"]
        env:
        - name: DATABASE_URL
          value: "postgresql+asyncpg://$(DB_USER):$(DB_PASSWORD)@postgres-infra-postgresql.infra.svc.cluster.local:5432/goldtemplate"
        envFrom:
        - secretRef:
            name: {{ include "gold-template.fullname" . }}-db-secret
{{- end }}

启用方式（values-infra-db.yaml）：

database:
  migrations:
    enabled: true

---

📡 6. API 契约管理（FastAPI 自动生成）

6.1 FastAPI 的契约优先设计

核心理念：代码即文档（Code as Documentation）

# backend/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr

app = FastAPI(
    title="Gold Template API",
    version="1.0.0",
    description="Production-ready FastAPI template",
    docs_url="/docs",          # Swagger UI
    redoc_url="/redoc",        # ReDoc
    openapi_url="/openapi.json"  # OpenAPI Schema
)

class UserCreate(BaseModel):
    """用户创建请求"""
    username: str
    email: EmailStr
    password: str

class User(BaseModel):
    """用户响应"""
    id: int
    username: str
    email: str

    class Config:
        from_attributes = True

@app.post("/users", response_model=User, status_code=201, tags=["Users"])
async def create_user(user: UserCreate):
    """
    创建新用户

    - **username**: 用户名（唯一）
    - **email**: 邮箱地址
    - **password**: 密码（将被加密）

    Returns:
        User: 创建的用户信息（不包含密码）
    """
    # 实现逻辑
    ...

自动生成的文档：

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

6.2 Claude Code 生成 API 端点

用户请求：

"添加用户管理 API：
- POST /users - 创建用户
- GET /users/{id} - 获取用户详情
- GET /users - 列出用户（支持分页）
- PUT /users/{id} - 更新用户
- DELETE /users/{id} - 删除用户
- 需要请求/响应模型
- 需要错误处理
- 需要 API 文档注释"

Claude 输出：

1. Pydantic 模型（UserCreate, UserUpdate, User, UserList）
2. 路由处理函数（5个端点）
3. 错误处理（404, 409, 422）
4. API 文档注释（docstring）
5. 单元测试

6.3 OpenAPI Schema 版本控制

CI/CD 自动导出：

# .github/workflows/complete-cicd.yml
- name: Export OpenAPI Schema
  run: |
    python -c "
    from backend.main import app
    import json
    with open('openapi.json', 'w') as f:
        json.dump(app.openapi(), f, indent=2)
    "

- name: Detect Schema Changes
  run: |
    # 检查是否有破坏性变更
    npx openapi-diff openapi-previous.json openapi.json --format markdown > api-changes.md

- name: Comment PR with API Changes
  uses: actions/github-script@v6
  with:
    script: |
      const fs = require('fs');
      const changes = fs.readFileSync('api-changes.md', 'utf8');
      github.rest.issues.createComment({
        issue_number: context.issue.number,
        owner: context.repo.owner,
        repo: context.repo.repo,
        body: `## API Changes\n\n${changes}`
      });

6.4 前后端协作

生成前端 SDK：

# 生成 TypeScript SDK
npx @openapitools/openapi-generator-cli generate \
  -i http://localhost:8000/openapi.json \
  -g typescript-axios \
  -o frontend/src/api

# 生成 Python SDK（如需要）
openapi-generator-cli generate \
  -i http://localhost:8000/openapi.json \
  -g python \
  -o clients/python-client

---

📊 7. 可观测性集成（三大支柱）

7.1 架构图

                    ┌─────────────────┐
                    │   FastAPI App   │
                    └────────┬────────┘
                             │
            ┌────────────────┼────────────────┐
            │                │                │
            ▼                ▼                ▼
     ┌──────────┐     ┌──────────┐    ┌──────────┐
     │  Metrics │     │   Logs   │    │  Traces  │
     │ /metrics │     │  stdout  │    │   OTLP   │
     └─────┬────┘     └─────┬────┘    └─────┬────┘
           │                │               │
           ▼                ▼               ▼
    ┌────────────┐   ┌────────────┐  ┌────────────┐
    │ Prometheus │   │ Promtail   │  │ OTel       │
    │ (monitoring)   │ (logging)  │  │ Collector  │
    └─────┬────┘   └─────┬────┘  │ (tracing)  │
          │              │         └─────┬────┘
          │              ▼               │
          │        ┌────────────┐        │
          │        │    Loki    │        │
          │        │ (logging)  │        │
          │        └─────┬────┘        │
          │              │               │
          └──────────────┼───────────────┘
                         │
                         ▼
                  ┌────────────┐
                  │  Grafana   │
                  │ (Dashboard)│
                  └────────────┘
                         │
                         ▼
                  ┌────────────┐
                  │   Tempo    │
                  │ (tracing)  │
                  └────────────┘

7.2 Metrics（Prometheus）

backend/main.py - Metrics 集成：

from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from fastapi import FastAPI, Request, Response
import time

# 定义 Metrics
REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

REQUEST_DURATION = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration in seconds',
    ['method', 'endpoint']
)

DB_CONNECTIONS = Gauge(
    'database_connections_active',
    'Active database connections'
)

# Middleware 自动记录
@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start_time = time.time()

    response = await call_next(request)

    duration = time.time() - start_time

    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()

    REQUEST_DURATION.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(duration)

    return response

# Metrics 端点
@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

ServiceMonitor 配置（已创建）：

# deployment/helm/gold-template/templates/servicemonitor.yaml
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: {{ include "gold-template.fullname" . }}
spec:
  selector:
    matchLabels:
      {{- include "gold-template.selectorLabels" . | nindent 6 }}
  endpoints:
  - port: http
    path: /metrics
    interval: 30s
    scrapeTimeout: 10s

Grafana Dashboard 查询：

# RPS（每秒请求数）
rate(http_requests_total{namespace="gold-template"}[5m])

# 错误率
rate(http_requests_total{namespace="gold-template",status=~"5.."}[5m])
/ rate(http_requests_total{namespace="gold-template"}[5m])

# P95 延迟
histogram_quantile(0.95,
  rate(http_request_duration_seconds_bucket{namespace="gold-template"}[5m])
)

# 活跃数据库连接
database_connections_active{namespace="gold-template"}

7.3 Logs（Loki）

backend/main.py - 结构化日志：

import logging
import json
import os
from datetime import datetime
from opentelemetry import trace

class JSONFormatter(logging.Formatter):
    """结构化日志格式器"""

    def format(self, record):
        # 获取当前 span 的 trace_id
        span = trace.get_current_span()
        trace_id = format(span.get_span_context().trace_id, "032x") if span else "0" * 32

        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "trace_id": trace_id,                                    # ✅ 关联链路
            "service": "gold-template",                               # ✅ 服务名
            "env": os.getenv("APP_ENV", "development"),              # ✅ 环境
            "namespace": os.getenv("NAMESPACE", "gold-template"),    # ✅ 命名空间
            "pod": os.getenv("HOSTNAME", "local"),                   # ✅ Pod 名
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }

        # 添加异常堆栈
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        # 添加额外字段
        if hasattr(record, 'user_id'):
            log_data["user_id"] = record.user_id
        if hasattr(record, 'request_id'):
            log_data["request_id"] = record.request_id

        return json.dumps(log_data)

# 配置 logger
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
logger = logging.getLogger(__name__)

# 使用 JSON formatter
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.handlers = [handler]

# 使用示例
logger.info("User login successful", extra={"user_id": 123})
logger.error("Database connection failed", exc_info=True)

Loki 自动采集（无需配置，Promtail 已部署）：

• Promtail 自动采集所有 Pod 的 stdout/stderr
• 自动添加标签：namespace、pod、container

Grafana Loki 查询：

# 查看所有日志
{namespace="gold-template"}

# 过滤错误日志
{namespace="gold-template"} | json | level="ERROR"

# 通过 trace_id 查找相关日志
{namespace="gold-template"} | json | trace_id="abc123..."

# 统计错误数量（最近5分钟）
sum(count_over_time({namespace="gold-template"} | json | level="ERROR" [5m]))

7.4 Traces（Tempo + OpenTelemetry）

安装依赖：

# requirements.txt
opentelemetry-api==1.22.0
opentelemetry-sdk==1.22.0
opentelemetry-instrumentation-fastapi==0.43b0
opentelemetry-exporter-otlp==1.22.0
opentelemetry-instrumentation-sqlalchemy==0.43b0
opentelemetry-instrumentation-httpx==0.43b0

backend/main.py - OpenTelemetry 集成：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.sdk.resources import Resource
import os

# 配置 Resource（服务标识）
resource = Resource.create({
    "service.name": "gold-template",
    "service.version": "1.0.0",
    "deployment.environment": os.getenv("APP_ENV", "development"),
})

# 配置 Tracer Provider
trace_provider = TracerProvider(resource=resource)
trace.set_tracer_provider(trace_provider)

# 配置 OTLP Exporter（发送到 OpenTelemetry Collector）
otlp_exporter = OTLPSpanExporter(
    endpoint=os.getenv(
        "OTEL_EXPORTER_OTLP_ENDPOINT",
        "otel-collector-opentelemetry-collector.tracing.svc.cluster.local:4317"
    ),
    insecure=True  # 集群内通信
)

# 添加 Span Processor
trace_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

# 创建 FastAPI app
app = FastAPI()

# 自动 instrument FastAPI（自动创建 spans）
FastAPIInstrumentor.instrument_app(app)

# 自动 instrument SQLAlchemy（数据库查询 spans）
# SQLAlchemyInstrumentor().instrument(engine=engine)

# 自动 instrument HTTPX（外部 HTTP 调用 spans）
HTTPXClientInstrumentor().instrument()

# 手动创建 span 示例
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    tracer = trace.get_tracer(__name__)

    with tracer.start_as_current_span("get_user_from_db") as span:
        span.set_attribute("user.id", user_id)

        # 数据库查询（自动创建子 span）
        user = await db.get_user(user_id)

        if not user:
            span.set_attribute("user.found", False)
            span.set_status(Status(StatusCode.ERROR, "User not found"))
            raise HTTPException(status_code=404)

        span.set_attribute("user.found", True)
        return user

Helm Values 配置（values-infra-db.yaml）：

env:
  # OpenTelemetry 配置
  - name: OTEL_EXPORTER_OTLP_ENDPOINT
    value: "otel-collector-opentelemetry-collector.tracing.svc.cluster.local:4317"
  - name: OTEL_SERVICE_NAME
    value: "gold-template"
  - name: OTEL_TRACES_SAMPLER
    value: "parentbased_traceidratio"
  - name: OTEL_TRACES_SAMPLER_ARG
    value: "0.1"  # 10% 采样率（生产环境降低开销）
  - name: OTEL_PYTHON_LOG_CORRELATION
    value: "true"  # 日志中注入 trace_id

访问 Tempo：

kubectl port-forward -n tracing svc/tempo 16686:16686
# 访问 http://localhost:16686（Jaeger UI）

在 Grafana 中关联查询：

1. 在 Grafana 中配置 Tempo 数据源
2. 在 Loki 日志中点击 trace_id
3. 自动跳转到 Tempo 查看完整链路

7.5 Claude Code 集成可观测性

用户请求：

"为 /users API 添加可观测性：
- Prometheus metrics（请求计数、响应时间）
- 结构化日志（包含 user_id）
- OpenTelemetry trace（数据库查询、外部 API 调用）"

Claude 输出：

1. 添加 Metrics（Counter, Histogram）
2. 在日志中添加 extra={"user_id": ...}
3. 添加自定义 span 和 attributes
4. 更新 requirements.txt
5. 更新 Helm values（环境变量）

---

🚀 8. GitOps 部署（Argo CD）

8.1 GitOps 原则

✅ Git 是唯一的事实源（Single Source of Truth）
✅ 声明式配置（Helm Charts）
✅ 自动同步（Automated Sync）
✅ 可回滚（Git Revert）
✅ 审计追踪（Git History）
❌ 禁止直接 kubectl apply 到生产环境

8.2 Argo CD Application 配置

argocd/gold-template-app.yaml（待创建）：

apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: gold-template
  namespace: argocd
  finalizers:
    - resources-finalizer.argocd.argoproj.io
spec:
  project: default

  # Git 仓库配置
  source:
    repoURL: https://github.com/yourusername/gold-template.git
    targetRevision: main
    path: deployment/helm/gold-template

    # Helm 配置
    helm:
      valueFiles:
        - values.yaml
        - values-orbstack.yaml  # 开发环境
        # - values-infra-db.yaml  # 生产环境（使用 infra 数据库）

      # 参数覆盖（可选）
      parameters:
        - name: image.tag
          value: "latest"

  # 部署目标
  destination:
    server: https://kubernetes.default.svc
    namespace: gold-template

  # 同步策略
  syncPolicy:
    automated:
      prune: true      # 自动删除 Git 中不存在的资源
      selfHeal: true   # 自动修复集群中的手动变更
      allowEmpty: false

    syncOptions:
      - CreateNamespace=true
      - PrunePropagationPolicy=foreground
      - PruneLast=true

    retry:
      limit: 5
      backoff:
        duration: 5s
        factor: 2
        maxDuration: 3m

  # 健康检查
  ignoreDifferences:
    - group: apps
      kind: Deployment
      jsonPointers:
        - /spec/replicas  # 忽略 HPA 修改的 replicas

8.3 多环境配置

开发环境（argocd/gold-template-dev.yaml）：

apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: gold-template-dev
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/yourusername/gold-template.git
    targetRevision: develop  # 开发分支
    path: deployment/helm/gold-template
    helm:
      valueFiles:
        - values-orbstack.yaml
  destination:
    server: https://kubernetes.default.svc
    namespace: gold-template-dev
  syncPolicy:
    automated:
      prune: true
      selfHeal: true

生产环境（argocd/gold-template-prod.yaml）：

apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: gold-template-prod
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/yourusername/gold-template.git
    targetRevision: main  # 主分支
    path: deployment/helm/gold-template
    helm:
      valueFiles:
        - values.yaml
        - values-infra-db.yaml  # 使用 infra 数据库
  destination:
    server: https://kubernetes.default.svc
    namespace: gold-template-prod
  syncPolicy:
    automated:
      prune: true
      selfHeal: false  # 生产环境禁用自动修复，需要人工审查
    syncOptions:
      - CreateNamespace=true

8.4 CI/CD 集成 GitOps

修改后的 .github/workflows/complete-cicd.yml（待更新）：

# ❌ 删除：直接 kubectl apply
# deploy-prod:
#   steps:
#     - name: Deploy to K8s
#       run: kubectl apply -f deployment/k8s/

# ✅ 改为：更新 Git 触发 Argo CD 同步
deploy-prod:
  name: Deploy to Production (GitOps)
  runs-on: ubuntu-latest
  needs: [build, container-scan]
  if: github.ref == 'refs/heads/main'
  environment:
    name: production
    url: https://gold-template.example.com
  steps:
    - name: Checkout code
      uses: actions/checkout@v4

    - name: Update image tag in Helm values
      run: |
        # 更新 values-prod.yaml 中的镜像标签
        sed -i "s/tag: .*/tag: \"${{ github.sha }}\"/" deployment/helm/gold-template/values-prod.yaml

    - name: Commit and push
      run: |
        git config user.name "GitHub Actions"
        git config user.email "[email protected]"
        git add deployment/helm/gold-template/values-prod.yaml
        git commit -m "chore: update prod image to ${{ github.sha }}"
        git push

    # Argo CD 会自动检测 Git 变更并同步

    - name: Wait for Argo CD sync (optional)
      run: |
        # 使用 Argo CD CLI 等待同步完成
        argocd app sync gold-template-prod --prune
        argocd app wait gold-template-prod --health --timeout 600

8.5 部署流程

1. 开发者提交代码到 Git
   ↓
2. GitHub Actions 触发 CI
   ↓ (测试、构建、扫描)
3. 构建 Docker 镜像并推送到 Registry
   ↓
4. 更新 Helm values 中的镜像标签
   ↓
5. Commit & Push 到 Git
   ↓
6. Argo CD 检测到 Git 变更
   ↓
7. Argo CD 自动同步到 K8s 集群
   ↓
8. 执行 PreSync Hook（数据库迁移）
   ↓
9. 部署应用 Pod
   ↓
10. 健康检查
   ↓
11. 部署完成 ✅

8.6 回滚策略

方法 1：Argo CD 回滚

# 查看历史版本
argocd app history gold-template-prod

# 回滚到指定版本
argocd app rollback gold-template-prod <revision-id>

方法 2：Git Revert

# 查看提交历史
git log --oneline

# Revert 有问题的提交
git revert <commit-hash>
git push

# Argo CD 自动同步并回滚

---

📋 9. 输出格式规范（Claude Code 交付标准）

当用户给 Claude 一个需求时，Claude 的输出必须按以下结构：

9.1 变更概览

## 变更概览

**功能**：用户认证系统
**涉及模块**：
- 后端：backend/auth.py, backend/models.py, backend/main.py
- 数据库：新增 users 表
- 配置：新增 JWT_SECRET 环境变量
- 部署：更新 Helm values（Secret）
- 监控：新增 login_attempts_total metric

**影响范围**：
- 新增 API：POST /auth/register, POST /auth/login, POST /auth/logout
- 数据库迁移：1 个 Alembic 脚本
- 依赖变更：新增 python-jose, passlib

9.2 文件级变更清单

## 文件变更清单

**新增文件**：
- `backend/auth.py` - 认证逻辑
- `backend/models/user.py` - User 模型
- `alembic/versions/20250101_create_users_table.py` - 数据库迁移
- `tests/test_auth.py` - 认证测试

**修改文件**：
- `backend/main.py` - 添加认证路由
- `requirements.txt` - 添加依赖
- `deployment/helm/gold-template/values-infra-db.yaml` - 添加 Secret 配置

**删除文件**：
- 无

9.3 数据库迁移

# alembic/versions/20250101_create_users_table.py
"""Create users table"""

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(50), nullable=False),
        sa.Column('email', sa.String(255), nullable=False),
        sa.Column('hashed_password', sa.String(255), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('username'),
        sa.UniqueConstraint('email')
    )
    op.create_index('ix_users_email', 'users', ['email'])

def downgrade():
    op.drop_index('ix_users_email', table_name='users')
    op.drop_table('users')

9.4 后端代码

# backend/auth.py（完整代码）
# backend/models/user.py（完整代码）
# backend/main.py（变更部分）

9.5 测试

# tests/test_auth.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_register_user(client: AsyncClient):
    response = await client.post("/auth/register", json={
        "username": "testuser",
        "email": "[email protected]",
        "password": "securepassword123"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"
    assert "password" not in data

# 运行测试
# pytest tests/test_auth.py -v

9.6 可观测性

## 可观测性

**Metrics**：
- `login_attempts_total{status="success|failure"}` - 登录尝试计数
- `active_users` - 当前活跃用户数
- `token_generation_duration_seconds` - Token 生成耗时

**日志字段**：
- `user_id` - 用户 ID
- `username` - 用户名
- `action` - 操作类型（register/login/logout）

**Trace Spans**：
- `auth.register` - 用户注册
- `auth.login` - 用户登录
- `auth.verify_password` - 密码验证
- `database.query.users` - 数据库查询

9.7 部署配置

# deployment/helm/gold-template/values-infra-db.yaml

# 新增 Secret 配置
secretEnv:
  - name: JWT_SECRET
    secretName: gold-template-auth-secret
    secretKey: jwt-secret
  - name: JWT_ALGORITHM
    value: "HS256"
  - name: JWT_EXPIRATION_MINUTES
    value: "10080"  # 7 days

# 创建 Secret
kubectl create secret generic gold-template-auth-secret \
  --from-literal=jwt-secret=$(openssl rand -hex 32) \
  -n gold-template

9.8 验收步骤

## 验收步骤

### 本地验证
```bash
# 1. 运行测试
pytest tests/test_auth.py -v

# 2. 启动应用
uvicorn backend.main:app --reload

# 3. 测试 API
curl -X POST http://localhost:8000/auth/register \
  -H "Content-Type: application/json" \
  -d '{"username":"test","email":"test@example.com","password":"pass123"}'

curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username":"test","password":"pass123"}'

集群验证

# 1. 部署到集群
./scripts/deploy-orbstack.sh --mode infra-db

# 2. 检查 Pod 状态
kubectl get pods -n gold-template

# 3. 查看日志
kubectl logs -n gold-template -l app.kubernetes.io/name=gold-template -f

# 4. 测试 API
kubectl port-forward -n gold-template svc/gold-template 8000:80
curl -X POST http://localhost:8000/auth/register ...

可观测性验证

# 1. 检查 Metrics
curl http://localhost:8000/metrics | grep login_attempts_total

# 2. 查看日志（Grafana Loki）
# 访问 Grafana，查询：{namespace="gold-template"} | json | action="login"

# 3. 查看 Trace（Tempo）
# 访问 Tempo UI，搜索 service.name="gold-template"，查看 auth.login span

PromQL 验证

# 登录成功率
rate(login_attempts_total{status="success"}[5m])
/ rate(login_attempts_total[5m])

# 登录失败告警
rate(login_attempts_total{status="failure"}[5m]) > 10

---

## 🚫 10. 禁止事项（Claude 必须遵守）

### 10.1 代码质量

- ❌ **禁止**：不进行错误处理，直接抛出未捕获的异常
- ❌ **禁止**：硬编码敏感信息（密码、token、API key）
- ❌ **禁止**：使用同步 I/O（应使用 async/await）
- ❌ **禁止**：SQL 注入风险（使用 ORM 参数化查询）
- ❌ **禁止**：跳过类型检查（必须有类型注解）

### 10.2 数据库

- ❌ **禁止**：生产环境破坏性 DB 变更（删列、改类型）且未走 Expand/Contract
- ❌ **禁止**：修改已发布的 Alembic 迁移脚本
- ❌ **禁止**：不幂等的迁移脚本（应支持多次执行）
- ❌ **禁止**：生产环境使用 SQLite（必须用 Postgres）

### 10.3 部署

- ❌ **禁止**：在 CI 中直接 kubectl apply 到生产集群（必须走 GitOps）
- ❌ **禁止**：跳过 PreSync Hook（数据库迁移必须先于应用部署）
- ❌ **禁止**：把 Secret 写进 Git 仓库
- ❌ **禁止**：在容器镜像中硬编码配置（应使用环境变量）

### 10.4 可观测性

- ❌ **禁止**：高基数 Metrics labels（user_id、email 等）
  - ✅ 正确：`http_requests_total{method="POST",endpoint="/users",status="200"}`
  - ❌ 错误：`http_requests_total{user_id="12345"}` （会导致指标爆炸）

- ❌ **禁止**：日志中包含敏感信息（密码、token、信用卡号）
- ❌ **禁止**：不记录 trace_id（无法关联日志和链路）

### 10.5 交付

- ❌ **禁止**：输出"待补充 / TODO / 以后再说"
- ❌ **禁止**：缺少测试的代码
- ❌ **禁止**：没有可观测性的代码
- ❌ **禁止**：没有部署配置的新功能

---

## 🎯 11. Claude Code 任务执行模式

### 11.1 信息不足时的处理

**✅ 优先给出合理默认，并明确写出假设**

```markdown
**假设**：
1. 用户表使用自增 ID（如需 UUID，请告知）
2. 密码使用 bcrypt 加密（rounds=12）
3. JWT token 有效期 7 天
4. 未提供 refresh token 机制（如需要，请告知）

基于以上假设，我将实现...

❌ 不要反复提问拖慢节奏

错误示范：
"您希望使用哪种加密算法？bcrypt、argon2 还是 scrypt？"
"Token 有效期应该是多久？"
"是否需要 refresh token？"

正确做法：
使用业界最佳实践（bcrypt），在代码注释中说明可配置

11.2 最小闭环优先

先产出可落地的最小闭环，再做增强项

最小闭环（第1轮）：
✅ 基本登录/注册功能
✅ 数据库迁移
✅ 基础测试
✅ 基础可观测性
✅ 部署配置

增强项（第2轮，如用户需要）：
⭐ Refresh token
⭐ OAuth2 第三方登录
⭐ 多因素认证
⭐ 密码复杂度策略
⭐ 账号锁定机制

11.3 Claude 自我检查清单

在交付前，Claude 必须自检：

## 交付前自检

- [ ] 代码是否有类型注解？
- [ ] 是否有错误处理？
- [ ] 是否有单元测试或集成测试？
- [ ] 是否添加了 Prometheus metrics？
- [ ] 日志是否包含 trace_id？
- [ ] 是否有 OpenTelemetry spans？
- [ ] 数据库迁移是否幂等？
- [ ] 是否更新了 Helm values？
- [ ] 是否有验收步骤？
- [ ] 是否有安全风险（SQL 注入、XSS、硬编码密码）？

---

📚 12. 相关文档（必读）

本蓝图是高层规范，具体实施请参考以下文档：

核心文档

1. 数据库集成：docs/deployment/INFRA-DATABASE-INTEGRATION.md

   • 如何使用 infra namespace 的 Postgres/Redis
   • 跨 namespace 访问配置
   • 数据库迁移策略（Alembic）

2. 可观测性集成：docs/deployment/OBSERVABILITY-INTEGRATION.md

   • Prometheus metrics 接入
   • Loki 日志格式
   • Tempo 链路追踪配置

3. 部署指南：

   • 本地开发：docs/deployment/LOCAL-DEPLOYMENT-CONFIG.md
   • 镜像构建：docs/deployment/IMAGE-BUILD-GUIDE.md

4. 项目说明：

   • 项目概览：README.md
   • 快速开始：QUICKSTART.md
   • Claude 指南：.claude/CLAUDE.md

快速命令

# 本地部署（SQLite）
./scripts/deploy-orbstack.sh

# 使用 infra 数据库部署
./scripts/deploy-orbstack.sh --mode infra-db

# 推送到 Harbor
./scripts/deploy-orbstack.sh --push-harbor

# 查看可观测性数据
kubectl port-forward -n monitoring svc/kube-prometheus-stack-grafana 3000:80
kubectl port-forward -n tracing svc/tempo 16686:16686

---

🔄 13. 版本历史

• v2.0 (2025-12-30):

  • 基于实际 OrbStack K8s 环境重写
  • 强化 Claude Code 集成
  • 添加 FastAPI + Alembic 特定指导
  • 补充完整的可观测性实施细节
  • 添加 GitOps 部署流程
  • 提供具体的代码示例和配置

• v1.0 (2025-12-20):

  • 初始版本（通用规范）

---

📞 反馈与改进

本蓝图是活文档，欢迎改进建议！

反馈方式：

• 提交 Issue：gh issue create
• 提交 PR：改进蓝图内容
• 与 Claude 讨论：直接在对话中提出改进建议

下一步优化方向：

1. 添加性能优化指南（缓存、连接池、异步优化）
2. 添加安全加固指南（CORS、Rate Limiting、输入验证）
3. 添加多租户架构指南
4. 添加 gRPC 集成指南

---

现在，让我们开始用 Claude Code 构建生产级应用吧！ 🚀