FunctionalScaffold/CLAUDE.md

CLAUDE.md

本文件为 Claude Code (claude.ai/code) 在此代码仓库中工作时提供指导。

为了方便团队交流，项目的自然语言使用中文，包括代码注释和文档等

项目概述

FunctionalScaffold（函数式脚手架） 是一个算法工程化 Serverless 解决方案的脚手架生成器。

核心目标

解决三大痛点：

1. 不确定的算力需求 - 需要动态扩缩容能力
2. 算法同学工程化能力不足 - 降低工程化门槛
3. 后端同学集成难度过高 - 标准化接口规范

技术架构

采用 Docker 封装的 Serverless API 服务方案：

• 算法代码 + 运行环境打包为 Docker 镜像
• 部署到云厂商 Serverless 平台实现自动扩缩容
• FastAPI 作为 HTTP 接口层
• 算法逻辑保持独立和专注

架构流程

用户请求 → API网关 → 容器实例（冷/热启动）→ FastAPI → 算法程序 → 返回结果
                                              ↓
                                        外部服务（OSS/数据库）

代码架构

项目采用 src layout 结构（Python 最佳实践）：

src/functional_scaffold/
├── algorithms/          # 算法层 - 所有算法必须继承 BaseAlgorithm
│   ├── base.py         # 提供 execute() 包装器（埋点、错误处理）
│   └── prime_checker.py # 示例：质数判断算法
├── api/                # API 层 - FastAPI 路由和模型
│   ├── models.py       # Pydantic 数据模型（使用 ConfigDict）
│   ├── routes.py       # 路由定义（/invoke, /healthz, /readyz, /jobs）
│   └── dependencies.py # 依赖注入（request_id 生成）
├── core/               # 核心功能 - 横切关注点
│   ├── errors.py       # 异常类层次结构
│   ├── logging.py      # 结构化日志（JSON 格式）
│   ├── metrics.py      # Prometheus 指标和装饰器
│   └── tracing.py      # 分布式追踪（ContextVar）
├── utils/              # 工具函数
│   └── validators.py   # 输入验证
├── config.py           # 配置管理（pydantic-settings）
└── main.py             # FastAPI 应用入口

关键设计模式：

1. 算法抽象层：所有算法继承 BaseAlgorithm，只需实现 process() 方法。execute() 方法自动处理埋点、日志和错误包装。

2. 依赖注入：使用 FastAPI 的 Depends() 机制注入 request_id，通过 ContextVar 在异步上下文中传递。

3. 配置管理：使用 pydantic-settings 从环境变量或 .env 文件加载配置，支持类型验证。

4. 可观测性：

   • 日志：结构化 JSON 日志（pythonjsonlogger），自动包含 request_id
   • 指标：Prometheus 格式（request_counter, request_latency, algorithm_counter）
   • 追踪：request_id 关联所有日志和指标
   • 日志收集：Loki + Promtail 自动收集和查询日志

开发命令

环境设置

# 创建虚拟环境并安装依赖（开发模式）
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -e ".[dev]"

运行服务

# 方式1：使用辅助脚本（推荐）
./scripts/run_dev.sh

# 方式2：直接运行（开发模式，自动重载）
uvicorn functional_scaffold.main:app --reload --port 8000

# 方式3：生产模式
uvicorn functional_scaffold.main:app --host 0.0.0.0 --port 8000 --workers 4

访问地址：

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• Metrics: http://localhost:8000/metrics

测试

# 运行所有测试
pytest tests/ -v

# 运行单个测试文件
pytest tests/test_algorithms.py -v

# 运行单个测试类
pytest tests/test_algorithms.py::TestPrimeChecker -v

# 运行单个测试方法
pytest tests/test_algorithms.py::TestPrimeChecker::test_prime_numbers -v

# 生成覆盖率报告
pytest tests/ --cov=src/functional_scaffold --cov-report=html
# 查看报告：open htmlcov/index.html

# 使用辅助脚本（包含代码检查）
./scripts/run_tests.sh

代码质量

# 代码格式化（自动修复）
black src/ tests/

# 代码检查（不修改文件）
black --check src/ tests/

# 代码检查
ruff check src/ tests/

# 自动修复可修复的问题
ruff check --fix src/ tests/

配置说明：

• Black: 行长度 100，目标 Python 3.9+
• Ruff: 行长度 100，目标 Python 3.9+

Docker

# 构建镜像
docker build -f deployment/Dockerfile -t functional-scaffold:latest .

# 运行容器
docker run -p 8000:8000 functional-scaffold:latest

# 使用 docker-compose（包含 Prometheus + Grafana + Loki）
cd deployment
docker-compose up
# Grafana: http://localhost:3000 (admin/admin)
# Prometheus: http://localhost:9090
# Loki: http://localhost:3100

文档

# 导出 OpenAPI 规范到 docs/swagger/openapi.json
python scripts/export_openapi.py

添加新算法

1. 创建算法类（继承 BaseAlgorithm）

# src/functional_scaffold/algorithms/my_algorithm.py
from typing import Dict, Any
from .base import BaseAlgorithm

class MyAlgorithm(BaseAlgorithm):
    """我的算法类"""

    def process(self, input_data: Any) -> Dict[str, Any]:
        """
        算法处理逻辑

        Args:
            input_data: 输入数据

        Returns:
            Dict[str, Any]: 处理结果
        """
        # 实现算法逻辑
        result = do_something(input_data)
        return {"result": result}

2. 注册到 __init__.py

# src/functional_scaffold/algorithms/__init__.py
from .my_algorithm import MyAlgorithm
__all__ = [..., "MyAlgorithm"]

3. 添加 API 端点（在 api/routes.py）

@router.post("/my-endpoint")
async def my_endpoint(
    request: MyRequest,
    request_id: str = Depends(get_request_id)
):
    """我的算法端点"""
    algorithm = MyAlgorithm()
    result = algorithm.execute(request.data)
    return MyResponse(request_id=request_id, **result)

4. 定义数据模型（在 api/models.py）

class MyRequest(BaseModel):
    """我的请求模型"""

    model_config = ConfigDict(
        json_schema_extra={
            "example": {"data": "示例数据"}
        }
    )

    data: str = Field(..., description="输入数据")

5. 编写测试

# tests/test_my_algorithm.py
def test_my_algorithm():
    """测试我的算法"""
    algo = MyAlgorithm()
    result = algo.process("测试数据")
    assert result["result"] == expected

配置管理

配置通过 src/functional_scaffold/config.py 的 Settings 类管理：

• 从环境变量读取（不区分大小写）
• 支持 .env 文件
• 使用 pydantic-settings 进行类型验证

配置示例：

# .env 文件
APP_ENV=production
LOG_LEVEL=INFO
METRICS_ENABLED=true

访问配置：

from functional_scaffold.config import settings
print(settings.app_env)  # "production"

可观测性

日志

使用 core/logging.py 的 setup_logging()：

from functional_scaffold.core.logging import setup_logging

# 设置日志
logger = setup_logging(level="INFO", format_type="json")

# 记录日志（自动包含 request_id）
logger.info("处理请求", extra={"user_id": "123"})

日志特性：

• 结构化 JSON 格式
• 自动包含 request_id（从 ContextVar 中提取）
• 支持文件日志（可选，通过环境变量启用）
• 日志轮转（100MB，保留 5 个备份）

日志收集（Loki）

项目集成了 Grafana Loki 日志收集系统，支持两种收集模式：

模式 1: Docker stdio 收集（默认，推荐）

• 自动收集容器标准输出/错误
• 无需修改应用代码
• 性能影响极小

模式 2: 文件收集（备用）

• 日志持久化到文件
• 支持日志轮转
• 需要设置 LOG_FILE_ENABLED=true

查询日志：

# 使用 Loki API
curl -G -s "http://localhost:3100/loki/api/v1/query_range" \
  --data-urlencode 'query={job="functional-scaffold-app"}'

# 按 request_id 过滤
curl -G -s "http://localhost:3100/loki/api/v1/query_range" \
  --data-urlencode 'query={job="functional-scaffold-app"} |= "request-id-here"'

Grafana 仪表板：

• 访问 http://localhost:3000
• 进入 "日志监控" 仪表板
• 使用 Request ID 输入框过滤特定请求的日志

相关文档：

• 完整文档：docs/loki-integration.md
• 使用说明：docs/grafana-dashboard-usage.md
• 快速参考：docs/loki-quick-reference.md

指标

使用 core/metrics.py 的装饰器：

from functional_scaffold.core.metrics import track_algorithm_execution

@track_algorithm_execution("my_algorithm")
def my_function():
    """我的函数"""
    pass

可用指标：

• http_requests_total{method, endpoint, status} - HTTP 请求总数
• http_request_duration_seconds{method, endpoint} - HTTP 请求延迟
• algorithm_executions_total{algorithm, status} - 算法执行总数
• algorithm_execution_duration_seconds{algorithm} - 算法执行延迟

追踪

Request ID 自动注入到所有请求和日志：

from functional_scaffold.core.tracing import get_request_id

# 在请求上下文中获取 request_id
request_id = get_request_id()

Request ID 特性：

• 自动生成或从请求头 X-Request-ID 获取
• 通过 ContextVar 在异步上下文中传递
• 自动添加到所有日志记录中
• 可用于追踪单个请求的完整生命周期
• 在 Grafana 仪表板中可按 request_id 过滤日志

部署

Kubernetes

kubectl apply -f deployment/kubernetes/deployment.yaml
kubectl apply -f deployment/kubernetes/service.yaml

配置说明：

• 3 个副本
• 资源限制：256Mi-512Mi 内存，250m-500m CPU
• 健康检查：存活探针 (/healthz)，就绪探针 (/readyz)

阿里云函数计算（FC 3.0）

# 安装 Serverless Devs（如未安装）
npm install -g @serverless-devs/s

# 配置阿里云凭证（首次使用）
s config add

# 部署到阿里云函数计算
cd deployment/serverless && s deploy

# 验证配置语法
cd deployment/serverless && s plan

# 查看函数日志
cd deployment/serverless && s logs --tail

AWS Lambda

sam deploy --template-file deployment/serverless/aws-lambda.yaml

必须交付的三大组件

1. 接入规范

API 端点标准：

• /invoke - 同步调用接口
• /jobs - 异步任务接口（当前返回 501）
• /healthz - 存活检查
• /readyz - 就绪检查
• /metrics - Prometheus 指标

Schema 规范：

• 请求/响应 Schema（Pydantic 验证）
• 错误响应格式（统一的 ErrorResponse）
• 元数据和版本信息（每个响应包含 metadata）

2. Python SDK 运行时

已实现的能力：

• ✅ 参数校验（Pydantic + utils/validators.py）
• ✅ 错误包装和标准化（core/errors.py）
• ✅ 埋点（core/metrics.py - 延迟、失败率）
• ✅ 分布式追踪的关联 ID（core/tracing.py + RequestIdFilter）
• ✅ 日志收集和查询（Loki + Promtail）
• ⏳ Worker 运行时（重试、超时、DLQ - 待实现）

3. 脚手架生成器

已包含的模板：

• ✅ 示例算法函数（algorithms/prime_checker.py）
• ✅ Dockerfile（deployment/Dockerfile）
• ✅ CI/CD 流水线配置（.github/workflows/）
• ✅ Serverless 平台部署 YAML（deployment/serverless/）
• ✅ Grafana 仪表板模板（monitoring/grafana/dashboards/）
• ✅ 告警规则配置（monitoring/alerts/rules.yaml）
• ✅ Loki 日志收集配置（monitoring/loki.yaml, monitoring/promtail.yaml）

开发理念

算法同学只需修改核心算法函数。 所有基础设施、可观测性、部署相关的工作都由脚手架处理。

算法开发者只需：

1. 继承 BaseAlgorithm
2. 实现 process() 方法
3. 返回字典格式的结果

框架自动提供：

• HTTP 接口封装
• 参数验证
• 错误处理
• 日志记录
• 性能指标
• 健康检查
• 容器化部署

注意事项

1. Pydantic V2：使用 ConfigDict 而非 class Config，使用 model_config 而非 Config。

2. 异步上下文：request_id 使用 ContextVar 存储，在异步函数中自动传递。

3. 测试隔离：每个测试使用 TestClient，不需要启动真实服务器。

4. Docker 构建：Dockerfile 使用非 root 用户（appuser），包含健康检查。

5. 配置优先级：环境变量 > .env 文件 > 默认值。

6. Promtail 版本：使用 Promtail 3.0.0 或更高版本，以支持较新的 Docker API（1.44+）。如果遇到 "client version too old" 错误，需要升级 Promtail 版本。

日志收集系统（Loki）

项目集成了 Grafana Loki 日志收集系统，提供强大的日志查询和分析能力。

架构

应用容器 (stdout/stderr)
    ↓
Docker Engine
    ↓
Promtail (日志采集器)
    ↓
Loki (日志存储)
    ↓
Grafana (可视化)

服务组件

docker-compose 包含以下服务：

• app: 应用服务（端口 8111）
• loki: 日志存储服务（端口 3100）
• promtail: 日志采集服务（端口 9080）
• grafana: 可视化服务（端口 3000）
• prometheus: 指标收集服务（端口 9090）
• redis: 缓存服务（端口 6380）

日志收集模式

模式 1: Docker stdio 收集（默认）

特点：

• ✅ 无需修改应用代码
• ✅ 自动收集容器标准输出/错误
• ✅ 性能影响极小
• ✅ 推荐用于生产环境

配置： 应用容器需要添加标签（已配置）：

labels:
  logging: "promtail"
  logging_jobname: "functional-scaffold-app"

模式 2: 文件收集（备用）

特点：

• ✅ 日志持久化到文件
• ✅ 支持日志轮转（100MB，5个备份）
• ✅ 适合需要本地日志文件的场景

启用方式：

# docker-compose.yml
environment:
  - LOG_FILE_ENABLED=true
  - LOG_FILE_PATH=/var/log/app/app.log

日志格式

所有日志使用 JSON 格式，自动包含以下字段：

• asctime: 时间戳
• name: 日志器名称
• levelname: 日志级别（INFO, WARNING, ERROR）
• message: 日志消息
• request_id: 请求 ID（自动添加）
• timestamp: ISO 格式时间戳

查询日志

使用 Loki API

# 查询所有日志
curl -G -s "http://localhost:3100/loki/api/v1/query_range" \
  --data-urlencode 'query={job="functional-scaffold-app"}'

# 按 request_id 过滤
curl -G -s "http://localhost:3100/loki/api/v1/query_range" \
  --data-urlencode 'query={job="functional-scaffold-app"} |= "request-id-here"'

# 查询错误日志
curl -G -s "http://localhost:3100/loki/api/v1/query_range" \
  --data-urlencode 'query={job="functional-scaffold-app", level="ERROR"}'

使用 Grafana 仪表板

1. 访问 http://localhost:3000（admin/admin）
2. 进入 "日志监控" 仪表板
3. 使用 Request ID 输入框过滤特定请求的日志

仪表板面板：

• 日志流（实时）: 实时日志流
• 日志量趋势: 按时间和级别统计
• 日志级别分布: INFO/WARNING/ERROR 分布
• 错误日志: 只显示 ERROR 级别

使用 Grafana Explore

1. 访问 http://localhost:3000/explore
2. 选择 Loki 数据源
3. 使用 LogQL 查询语言

常用查询：

# 查询所有日志
{job="functional-scaffold-app"}

# 查询错误日志
{job="functional-scaffold-app", level="ERROR"}

# 按 request_id 过滤
{job="functional-scaffold-app"} |= "request-id-here"

# 使用 JSON 解析
{job="functional-scaffold-app"} | json | request_id="request-id-here"

# 统计日志量
sum by (level) (count_over_time({job="functional-scaffold-app"}[5m]))

验证和测试

# 验证 Loki 集成
./scripts/verify_loki.sh

# 测试 Request ID 过滤
./scripts/test_request_id_filter.sh

配置文件

• Loki 配置: monitoring/loki.yaml

  • 日志保留期: 7 天
  • 摄入速率限制: 10MB/s
  • 自动压缩和清理

• Promtail 配置: monitoring/promtail.yaml

  • Docker stdio 收集配置
  • 文件收集配置
  • JSON 日志解析规则

• Grafana Provisioning: monitoring/grafana/

  • 数据源自动配置（datasources/）
  • 仪表板自动加载（dashboards/）

故障排查

看不到日志：

1. 检查服务状态: docker-compose ps
2. 查看 Promtail 日志: docker-compose logs promtail
3. 验证容器标签: docker inspect <container> | grep Labels

Docker socket 权限问题：

sudo chmod 666 /var/run/docker.sock

日志延迟：

• Promtail 每 5 秒刷新一次
• 建议等待 5-10 秒后再查询

相关文档

• 完整文档: docs/loki-integration.md - 包含查询示例、故障排查、性能优化
• 快速参考: docs/loki-quick-reference.md - 常用命令和 LogQL 查询
• 仪表板使用: docs/grafana-dashboard-usage.md - Grafana 仪表板使用说明
• 实施总结: docs/loki-implementation-summary.md - 架构和实施细节
• 监控目录: monitoring/README.md - 配置文件说明