当存储后端触及硬性上限时,团队面对的往往不是一次普通重构,而是一场必须保持线上行为稳定的系统迁移。Datadog 工程师 Arnold Wakim 分享的案例值得关注:团队借助 Claude 和 Cursor 推进关键生产系统演进,用测试约束 AI 生成的改动,在突破存储限制的同时显著改善性能。
来源摘要没有披露具体存储引擎、数据模型和性能数字,因此下面不会猜测 Datadog 的内部实现,而是拆解这类实践中最值得复用的工程方法,并给出一个可以运行的简化迁移示例。
AI 的价值不只是多写几行代码
大型生产迁移通常包含大量机械但高风险的工作:梳理旧接口、补齐测试、实现新后端适配器、修改调用路径、生成迁移工具,以及检查错误处理。Claude 和 Cursor 可以加快这些任务,但它们不能替团队定义“正确行为”。
更稳妥的分工是:
- 工程师定义不变量,例如写入后必须可读、重复迁移必须幂等、旧数据不能丢失。
- 测试把这些不变量变成可执行契约。
- AI 根据契约生成或修改实现,并帮助定位失败原因。
- 人类审查并发、超时、资源消耗、可观测性和回滚路径。
这种顺序很关键。若直接让 AI “把旧存储迁移到新存储”,模型很容易产出表面完整、边界条件不足的代码。若先给它失败测试、接口约束和验收命令,任务就从开放式设计收敛成可验证的实现问题。
测试是迁移期间的行为护栏
存储迁移不能只验证几个正常读写请求。测试集合至少应覆盖四个层次:
- 兼容性测试:相同输入在旧后端和新后端得到等价结果。
- 迁移测试:历史数据能够复制,重复执行不会制造重复记录或错误状态。
- 故障测试:新后端超时、部分写入或不可用时,系统执行明确的降级策略。
- 性能测试:尾延迟、吞吐量、内存和存储增长符合上线门槛。
AI 特别适合根据已有接口批量生成参数化测试,但生成数量不等于覆盖风险。团队仍需主动提供空值、超大对象、乱序事件、重复请求、并发更新和损坏数据等案例。
还要警惕“测试与实现一起写错”。如果同一次提示让模型同时设计规则、实现代码和断言,错误假设可能在三处保持一致,测试依然全部通过。关键契约应来自现有生产行为、事故记录、指标和领域专家,而不是仅由模型推导。
可以这样实践:先搭一个可验证的迁移骨架
下面是一个简化的 Python 伪项目。假设系统要把键值数据从旧存储迁移到新存储,并在切换前比较两边的读取结果。示例不代表 Datadog 的内部代码,但可以直接运行,再替换成真实数据库客户端。
创建 migration.py:
from dataclasses import dataclass, field
from typing import Protocol
class Store(Protocol):
def get(self, key: str) -> str | None: ...
def put(self, key: str, value: str) -> None: ...
def items(self) -> list[tuple[str, str]]: ...
@dataclass
class MemoryStore:
data: dict[str, str] = field(default_factory=dict)
def get(self, key: str) -> str | None:
return self.data.get(key)
def put(self, key: str, value: str) -> None:
self.data[key] = value
def items(self) -> list[tuple[str, str]]:
return list(self.data.items())
def migrate(source: Store, target: Store) -> int:
changed = 0
for key, value in source.items():
if target.get(key) != value:
target.put(key, value)
changed += 1
return changed
def compare_reads(source: Store, target: Store, keys: list[str]) -> list[str]:
return [key for key in keys if source.get(key) != target.get(key)]
创建 test_migration.py:
from migration import MemoryStore, compare_reads, migrate
def test_migration_preserves_values():
old = MemoryStore({"a": "1", "b": "2"})
new = MemoryStore()
assert migrate(old, new) == 2
assert compare_reads(old, new, ["a", "b", "missing"]) == []
def test_migration_is_idempotent():
old = MemoryStore({"a": "1"})
new = MemoryStore()
assert migrate(old, new) == 1
assert migrate(old, new) == 0
assert new.get("a") == "1"
def test_migration_repairs_stale_target_value():
old = MemoryStore({"a": "current"})
new = MemoryStore({"a": "stale"})
assert migrate(old, new) == 1
assert compare_reads(old, new, ["a"]) == []
运行方式:
python -m venv .venv
. .venv/bin/activate
python -m pip install pytest
pytest -q
把这个骨架接入 Claude 或 Cursor 时,可以使用边界明确的提示,而不是只描述最终目标:
你正在修改一个生产存储迁移工具。
约束:
- 不修改 Store 公共接口。
- migrate 必须幂等。
- 目标端已有正确值时不得重复写入。
- 任何行为变更都必须先增加失败测试。
- 不删除现有测试。
任务:
为批量迁移增加可配置的 batch_size,并增加以下测试:
1. 空源存储不执行写入;
2. 最后一批不足 batch_size 时仍完整迁移;
3. 写入失败时返回非零状态且不吞掉异常。
完成后给出改动文件列表、风险点和 pytest 命令,不要假设测试已经通过。
这类提示把接口、不可破坏的行为和验证方式写清楚,也要求模型区分“生成了代码”和“代码已经通过验证”。
上线不是一次切换,而是一组可逆步骤
测试通过只说明代码满足当前测试集,并不代表可以直接接管生产流量。对于关键系统,可以这样组织发布过程:
- 影子读取:主请求仍读取旧后端,同时异步读取新后端,记录结果差异但不影响响应。
- 双写或日志回放:为新后端持续补充增量数据,并监控失败率与积压量。
- 小比例切流:按租户、分区或哈希范围逐步启用新读取路径。
- 独立开关:分别控制迁移任务、双写、新读取和回退,避免一个总开关承担所有状态。
- 明确回滚条件:提前规定错误率、P99 延迟、数据差异率或资源消耗达到什么阈值就暂停。
双写也有边界:两个后端之间不存在天然原子性,一边成功、一边失败时必须有补偿队列、重放日志或对账任务。AI 可以协助生成这些组件,但一致性策略必须由系统所有者明确决定。
采用这套方法前的检查清单
Datadog 案例传递出的核心经验,不是把生产迁移交给 AI,而是让 AI 在测试、审查和渐进发布构成的闭环中工作。准备落地时,应检查:
- 旧系统的关键行为是否已变成自动化测试。
- 性能目标是否包含 P95/P99,而不只是平均延迟。
- AI 是否能看到足够上下文,但接触不到生产密钥和敏感数据。
- 每个生成改动是否经过人工审查、静态检查和真实测试。
- 数据迁移是否幂等、可暂停、可恢复并可对账。
- 新旧后端差异是否有指标、日志和告警。
- 回滚路径是否经过演练,而不只是写在文档里。
Claude 和 Cursor 能显著压缩理解代码、补测试和实现适配层所需的时间,但生产迁移的可靠性仍来自明确契约、可观测的分阶段发布,以及由工程师负责的最终判断。