实战示例 #
本页按使用场景分类，每个示例都可以直接复制运行。
1. 批量写入压测 #
目标： 评估写入吞吐量、集群负载与队列压力。
# variables: [
# {name: "id", type: "sequence"},
# {name: "uuid", type: "uuid"},
# {name: "suffix", type: "range", from: 1, to: 100},
# {name: "ip", type: "file", path: "dict/ip.txt"},
# {name: "now_unix", type: "now_unix"},
# ],
POST $[[env.ES_ENDPOINT]]/_bulk
{"index": {"_index": "bench-$[[suffix]]", "_id": "$[[uuid]]", "routing": "$[[routing_no]]"}}
{"id": "$[[id]]", "ip": "$[[ip]]", "batch": "$[[batch_no]]", "ts": "$[[now_unix]]"}
# request: {
# basic_auth: {
# username: "$[[env.ES_USERNAME]]",
# password: "$[[env.ES_PASSWORD]]",
# },
# runtime_variables: {
# batch_no: "uuid",
# },
# runtime_body_line_variables: {
# routing_no: "uuid",
# },
# body_repeat_times: 1000,
# },
运行命令：
# 持续 60 秒，100 并发
loadgen -d 60 -c 100
# 固定 QPS
loadgen -d 60 -c 100 -r 2000
# 固定请求总量
loadgen -d 600 -c 100 -l 50000
参考样例： testing/setup/loadgen/cases/bulk/create.yml
2. 搜索性能测试 #
目标： 评估查询延迟与吞吐，观察不同查询模式的差异。
# variables: [
# {name: "user", type: "file", path: "dict/user.txt"},
# {name: "ip", type: "file", path: "dict/ip.txt"},
# ],
GET $[[env.ES_ENDPOINT]]/my-index/_search?q=user:$[[user]]
# request: {
# basic_auth: {
# username: "$[[env.ES_USERNAME]]",
# password: "$[[env.ES_PASSWORD]]",
# },
# },
运行命令：
loadgen -d 30 -c 10
参考样例： testing/setup/loadgen/cases/search/random_search.yml
3. 混合请求压测 #
目标： 模拟真实流量中的 create + update + delete 混合负载。
一个 DSL 文件中写多个请求，Loadgen 会按顺序循环执行：
# variables: [
# {name: "id", type: "sequence"},
# {name: "uuid", type: "uuid"},
# {name: "user", type: "list", data: ["alice", "bob", "test"]},
# ],
POST $[[env.ES_ENDPOINT]]/_bulk
{"index": {"_index": "mixed-test", "_id": "$[[uuid]]"}}
{"id": "$[[id]]", "user": "$[[user]]"}
# request: {
# body_repeat_times: 10,
# },
POST $[[env.ES_ENDPOINT]]/_bulk
{"delete": {"_index": "mixed-test", "_id": "$[[uuid]]"}}
# request: {
# body_repeat_times: 5,
# },
参考样例： testing/setup/loadgen/cases/bulk/mixed.yml
4. 功能验证与 CI 集成 #
目标： 一次性执行所有请求，断言全部通过后退出。
# runner: {
# total_rounds: 1,
# no_warm: true,
# assert_invalid: true,
# assert_error: true,
# },
GET $[[env.ES_ENDPOINT]]/_cluster/health
# assert: {
# _ctx.response.status: 200,
# },
GET $[[env.ES_ENDPOINT]]/_cat/indices
# assert: {
# _ctx.response.status: 200,
# },
运行命令：
loadgen -run health_check.dsl
echo $? # 0=通过, 1=断言失败, 2=请求错误
适合放入 CI 流水线，任何断言不通过即失败退出。
参考样例： testing/setup/loadgen/cases/dummy/loadgen.yml
5. 动态变量注册（链路式测试） #
目标： 从前序请求提取值，供后续请求使用。
# runner: {
# total_rounds: 1,
# no_warm: true,
# assert_invalid: true,
# },
# 创建文档，提取生成的 _id
POST $[[env.ES_ENDPOINT]]/my-index/_doc
{"title": "test document", "user": "loadgen"}
# assert: {
# _ctx.response.status: 201,
# },
# register: [
# { doc_id: "_ctx.response.body_json._id" },
# ]
# 用注册的 doc_id 读取该文档
GET $[[env.ES_ENDPOINT]]/my-index/_doc/$[[doc_id]]
# assert: {
# _ctx.response.status: 200,
# _ctx.response.body_json._source.user: "loadgen",
# },
# 删除该文档
DELETE $[[env.ES_ENDPOINT]]/my-index/_doc/$[[doc_id]]
# assert: {
# _ctx.response.status: 200,
# },
参考样例： testing/setup/easysearch/cases/crud/create_without_id/loadgen.yml
6. JSON 响应断言 #
目标： 对 JSON 响应体中特定字段做精确校验。
GET $[[env.ES_ENDPOINT]]/my-index/_search
{"query": {"match_all": {}}}
# assert: {
# _ctx.response.status: 200,
# _ctx.response.body_json.hits.total.value: 100,
# _ctx.response.body_json.timed_out: false,
# },
简写断言格式：
GET /api/status
# assert: (200, {status: "ok", version: "1.0"}),
参考样例： testing/setup/gateway/cases/filters/set_response/loadgen.yml
7. 带鉴权的 API 测试（Token 模式） #
目标： 先登录获取 token，后续请求通过 Header 鉴权。
# runner: {
# total_rounds: 1,
# no_warm: true,
# assert_invalid: true,
# default_endpoint: "$[[env.API_HOST]]",
# },
# 登录
POST /account/login
{"email": "admin@example.com", "password": "$[[env.ADMIN_PASSWORD]]"}
# assert: (200, {}),
# register: [
# { token: "_ctx.response.body_json.access_token" },
# ]
# 使用 token 调用业务接口
GET /api/resources?size=10
# request: {
# headers: [
# {Authorization: "Bearer $[[token]]"},
# ],
# disable_header_names_normalizing: true,
# },
# assert: {
# _ctx.response.status: 200,
# },
# 创建资源
POST /api/resources
{"name": "test-resource", "type": "demo"}
# request: {
# headers: [
# {Authorization: "Bearer $[[token]]"},
# ],
# disable_header_names_normalizing: true,
# },
# assert: (200, {}),
# register: [
# { resource_id: "_ctx.response.body_json.id" },
# ]
# 删除资源
DELETE /api/resources/$[[resource_id]]
# request: {
# headers: [
# {Authorization: "Bearer $[[token]]"},
# ],
# disable_header_names_normalizing: true,
# },
# assert: {
# _ctx.response.status: 200,
# },
8. Coco AI 集成测试实践 #
我们在 Coco Server 项目中用 Loadgen 实现了完整的集成测试：
项目结构 #
tests/
├── loadgen.yml # 全局配置（环境变量、runner）
├── assets/
│ ├── run_integration_tests.py # 自动执行所有场景
│ ├── start_coco.sh
│ ├── stop_coco.sh
│ └── reset_coco_indices.sh
├── llm/
│ ├── scenario1.dsl
│ ├── scenario2.dsl
│ └── scenario3.dsl
├── assistant/
│ └── scenario1.dsl
└── datasource/
└── scenario1.dsl
全局配置 #
# tests/loadgen.yml
runner:
total_rounds: 1
no_warm: true
reset_context: true
assert_invalid: true
default_endpoint: "$[[env.COCO_SERVER]]"
env:
COCO_SERVER: http://127.0.0.1:9000
ADMIN_PASSWORD: PASSword_123
单场景调试 #
cd coco
loadgen -config ./tests/loadgen.yml -run ./tests/llm/scenario3.dsl -debug
全量回归测试 #
cd coco
python3 ./tests/assets/run_integration_tests.py
该脚本自动执行：停止环境 → 重置数据 → 启动服务 → 逐个执行 DSL → 清理。
相关链接 #
9. 项目组织建议 #
推荐将测试工程按以下方式组织：
project/
├── loadgen.yml # 全局配置
├── dict/ # 语料文件
│ ├── ip.txt
│ ├── user.txt
│ └── keywords.txt
├── cases/ # 按功能分组的测试用例
│ ├── bulk/
│ │ ├── create.dsl
│ │ └── mixed.dsl
│ ├── search/
│ │ └── random.dsl
│ └── crud/
│ └── lifecycle.dsl
└── suites/ # 套件配置（批量运行多用例）
└── regression.yml
语料文件示例（dict/ip.txt）：
192.168.1.1
10.0.0.1
172.16.0.100
每行一个值，变量每次请求随机取一行。
参考仓库：
10. 常用命令速查 #
# 基础压测（60 秒、100 并发）
loadgen -run bench.dsl -d 60 -c 100
# 固定 QPS
loadgen -run bench.dsl -d 60 -c 100 -r 1000
# 固定总请求数
loadgen -run bench.dsl -d 600 -c 100 -l 50000
# 启用 gzip 压缩
loadgen -run bench.dsl -d 60 -c 100 -compress
# 功能验证（单次执行）
loadgen -run test.dsl
# 指定配置文件
loadgen -run test.dsl -config env/production.yml
# 调试模式
loadgen -run test.dsl -debug