Datadog 如何用 Claude、Cursor 与测试驱动方法迁移关键生产系统

2026-07-10 23 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

关键生产系统遇到存储后端的硬性限制时,迁移往往不是简单地替换客户端或复制数据。接口语义、历史边界条件、并发行为和性能特征都可能在切换过程中发生变化。Datadog 工程师 Arnold Wakim 分享的案例值得关注,原因不只是团队使用了 Claude 和 Cursor,而是他们把 AI 放进了一个由测试、性能数据和渐进式发布约束的生产迁移流程。

AI 加速的是验证闭环,不是替团队承担风险

从来源摘要能够确认的是,这次工作涉及一个关键生产系统、存储后端的硬限制,以及显著的性能改善。文章同时讨论了哪些做法有效、哪些无效和团队得到的经验。至于具体存储产品、数据模型和性能数字,摘要没有提供,因此不宜自行补全。

在这类迁移中,Claude、Cursor 等工具比较适合承担高频、可验证的工作:

  • 阅读旧实现和测试,整理隐含的数据契约。
  • 为边界条件生成候选测试,再由工程师审查断言是否正确。
  • 将一个改动拆成小补丁,持续运行单元测试和集成测试。
  • 分析失败日志、调用链和性能剖析结果,提出可复现的排查方向。
  • 生成迁移脚本、监控查询或回滚步骤的初稿。

它们不适合独立决定数据一致性策略、删除旧数据,或者根据一组通过的单元测试宣布迁移完成。AI 能快速生成看似合理的代码,但它并不知道生产系统真正允许多大的数据偏差、延迟回退或恢复时间。

先把旧系统的行为变成可执行契约

存储迁移最危险的部分通常不是新实现写不出来,而是团队没有完整描述旧系统已经形成的行为。例如,缺失记录究竟返回空值还是错误,重复写入是否幂等,同一时间戳的数据如何排序,过期数据何时不可见,这些细节可能没有出现在设计文档中,却已经被上游服务依赖。

测试驱动迁移可以把工作拆成三个阶段:

  1. 用特征测试记录旧实现当前的可观察行为,即使其中有些行为并不理想。
  2. 让新旧实现运行同一套契约测试,消除语义差异。
  3. 通过影子读取、双写或离线回放比较真实流量下的结果,再逐步切换读取路径。

AI 在第一阶段尤其有用:它可以扫描调用点,列出可能遗漏的输入组合,并生成测试骨架。但“正确答案”必须来自旧系统的实际行为、明确的产品规则或团队确认,不能来自模型猜测。

可以这样实践:为新旧存储实现建立契约测试

下面是一个可直接运行的最小示例。它不代表 Datadog 的具体实现,而是演示如何让旧后端和候选后端共享同一组行为测试,并在迁移期间做影子比较。

创建 test_storage_contract.py

from dataclasses import dataclass
from typing import Protocol

import pytest


@dataclass(frozen=True)
class Event:
    event_id: str
    timestamp: int
    payload: str


class EventStore(Protocol):
    def put(self, event: Event) -> None: ...
    def get(self, event_id: str) -> Event | None: ...
    def list_since(self, timestamp: int) -> list[Event]: ...


class LegacyStore:
    def __init__(self) -> None:
        self._events: dict[str, Event] = {}

    def put(self, event: Event) -> None:
        self._events[event.event_id] = event

    def get(self, event_id: str) -> Event | None:
        return self._events.get(event_id)

    def list_since(self, timestamp: int) -> list[Event]:
        return sorted(
            (event for event in self._events.values() if event.timestamp >= timestamp),
            key=lambda event: (event.timestamp, event.event_id),
        )


class CandidateStore(LegacyStore):
    """Replace this class with an adapter for the new storage backend."""


@pytest.fixture(params=[LegacyStore, CandidateStore])
def store(request) -> EventStore:
    return request.param()


def test_missing_event_returns_none(store: EventStore) -> None:
    assert store.get("missing") is None


def test_put_is_idempotent_by_event_id(store: EventStore) -> None:
    event = Event("evt-1", 100, "created")
    store.put(event)
    store.put(event)
    assert store.list_since(0) == [event]


def test_list_since_is_inclusive_and_stably_sorted(store: EventStore) -> None:
    store.put(Event("evt-b", 100, "second"))
    store.put(Event("evt-a", 100, "first"))
    store.put(Event("evt-old", 99, "old"))

    assert store.list_since(100) == [
        Event("evt-a", 100, "first"),
        Event("evt-b", 100, "second"),
    ]


def shadow_compare(primary: EventStore, candidate: EventStore, since: int) -> None:
    expected = primary.list_since(since)
    actual = candidate.list_since(since)
    assert actual == expected, {
        "since": since,
        "expected": expected,
        "actual": actual,
    }


def test_shadow_read_matches_legacy_store() -> None:
    legacy = LegacyStore()
    candidate = CandidateStore()
    events = [
        Event("evt-1", 100, "created"),
        Event("evt-2", 101, "updated"),
    ]
    for event in events:
        legacy.put(event)
        candidate.put(event)

    shadow_compare(legacy, candidate, since=100)

运行方式:

python -m venv .venv
. .venv/bin/activate
python -m pip install pytest
pytest -q test_storage_contract.py

接入真实系统时,可以保留 EventStore 契约,把 CandidateStore 替换成新后端适配器。不要让测试直接依赖后端特有字段,否则契约测试会退化为实现测试。涉及真实数据的影子比较也不应只使用 assert:应记录差异类型、请求维度和样本,并设置采样率,避免给生产链路增加不可控负载。

给 Claude 或 Cursor 的任务要足够窄

比起“把这个存储系统迁移掉”,更有效的提示应包含明确边界、验收命令和禁止事项。可以这样改造后交给编码助手:

目标:让 CandidateStore 满足 test_storage_contract.py 中的全部契约。

约束:
- 不修改 LegacyStore。
- 不删除或放宽现有断言。
- 每次只处理一个失败测试。
- 保持 list_since 的边界包含语义和稳定排序。
- 不引入测试中未证明必要的新抽象。

工作流程:
1. 先解释当前失败反映的行为差异。
2. 提交最小代码改动。
3. 运行 pytest -q test_storage_contract.py。
4. 报告修改内容、测试结果和仍未覆盖的风险。

这种提示把模型限制在一个可观察的反馈循环里。相反,范围过大的重写任务容易产生三个问题:模型同时改变接口与实现、为了通过测试而削弱断言,以及在缺少生产上下文时“补齐”并不存在的业务规则。

上线前还需要测试之外的护栏

单元测试通过只说明已编码的契约成立,并不能证明新后端在真实数据量、并发和故障条件下表现正确。迁移关键路径时,至少应检查:

  • 新旧路径是否能按租户、分区或流量比例独立切换。
  • 影子读取的差异是否按缺失、顺序、精度和内容分类。
  • 是否观察延迟分位数、错误率、吞吐量和后端资源消耗,而不只看平均延迟。
  • 双写失败时以哪个系统为准,补偿队列如何重放。
  • 回滚是否只需要配置变更,旧后端又需要保留多久。
  • 数据校验是否覆盖空数据、大对象、重复请求、乱序和时间边界。
  • AI 生成的代码是否经过人工审查、静态检查和依赖安全检查。

Datadog 这类案例真正可复用的部分,不是选中某个 AI 工具就能自动完成迁移,而是把模型置于严格的工程控制面内:测试定义行为,真实流量验证兼容性,指标判断性能,渐进发布限制影响范围,工程师负责最终决策。对于关键生产系统,这套边界比生成代码的速度更重要。


相关推荐