CI 跑用例前先"体检"：pre-flight check 把 78% 的假失败拦在用例执行之前

一、现象与背景：52 分钟的 CI，40 分钟在"等环境"

先给一组让我坐不住数字。我维护一条支付清算链路的接口自动化套件（Python 3.11 + pytest 7.4 + httpx，约 1800 条用例），底层依赖 4 个中间件：MySQL 8.0（同一 staging 库）、Redis 7.2、Kafka 3.4、以及一个自研的 MQ broker。CI 跑在 2 台 8C16G 的 GitLab Runner 上，一次全量回归标称 52 分钟。

但真正让我头疼的不是慢，是假失败。我统计了 2024 年 Q3 三个月的数据：

• 全量回归共触发 186 次，其中 47 次（25.3%） 出现 1 条以上"单跑全绿、重跑即过"的假失败；
• 这 47 次里，事后归因有 37 次（78.7%） 的根因是环境未就绪：MySQL 主从延迟导致写入未同步、Redis 连接池被上一轮 CI 占满、Kafka broker 还在做 leader 选举、MQ broker 的 GC 暂停超过 2s；
• 每次假失败后，工程师平均花 11 分钟 排查"是真 bug 还是环境问题"，其中 9 分钟 用来翻日志确认环境状态。

一句话总结：将近四分之一的 CI 运行在"带病启动"，而团队花了大量时间事后诊断本可以在事前拦截的问题。 这就像去医院做手术，麻醉师不检查设备、不确认药品，直接开刀——出了事故再找原因，代价是患者的安全。

二、根因分析：为什么"环境未就绪"会伪装成"用例失败"

这个问题之所以反复出现，有 3 层机制：

1. 异步就绪 vs 同步断言的时钟错位。 MySQL 主从复制、Kafka leader 选举、Redis 主从 failover 都是异步过程，但测试用例的断言是同步的。用例发出 INSERT 后立即 SELECT，如果主库还没同步到从库（默认 slave_net_timeout=60s，实际延迟在 50-500ms 不等），读到的就是旧数据——用例挂了，但代码没 bug。

2. 连接池的"僵尸占用"。 上一轮 CI 如果异常退出（OOM kill、超时 kill），TCP 连接不一定被 RST。MySQL 的 wait_timeout 默认 28800s（8 小时），Redis 的 timeout 配置为 0（不主动断开）。这意味着僵尸连接可以占住连接池 8 小时，下一轮 CI 拿不到连接，超时报 Connection pool exhausted，看起来像"接口超时"。

3. 缺乏"环境就绪"的显式契约。 我们的 CI pipeline 是 install → build → test，中间没有任何"确认环境可用"的步骤。测试框架假设"能连上 = 能工作"，但现实是"能 TCP 握手 ≠ 能正常读写"——MySQL 能 ping 通但 binlog 还在 replay，Redis 能 PONG 但 AOF 还在 fsync，Kafka 能 list topic 但 partition leader 还没确定。

核心结论：需要在"用例执行"之前插入一道显式的健康检查关卡，把"环境就绪"从隐式假设变成显式契约。 这和谢清和-928 在断言分层里说的思路一脉相承——把"能连上"这种弱校验升级为"能正常工作"的强校验（参见其"契约/值/不变式"三层模型）。

三、具体做法：四层 pre-flight check 流水线

我把 pre-flight check 设计成 4 层，从外到内逐层深入，任何一层失败就立即 abort 整轮 CI，不进入用例执行。实现为 pytest 的 conftest.py 里一个 scope="session" 的 autouse fixture，在 CI 入口跑一次。

L1：网络可达性（< 5s）

最基础的一层，确认所有依赖的 TCP 端口可握手：

import socket, time

def _tcp_check(host: str, port: int, timeout: float = 3.0) -> bool:
    t0 = time.monotonic()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return (time.monotonic() - t0) * 1000 < timeout * 1000 * 0.8
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False

DEPS = {
    "mysql":   ("staging-mysql.internal", 3306),
    "redis":   ("staging-redis.internal", 6379),
    "kafka":   ("staging-kafka.internal", 9092),
    "mq":      ("staging-mq.internal",    5672),
}

def test_l1_network():
    failures = []
    for name, (host, port) in DEPS.items():
        if not _tcp_check(host, port):
            failures.append(f"{name} ({host}:{port})")
    assert not failures, f"[L1] 网络不可达: {failures}"

L2：协议层握手（< 10s）

TCP 通了不代表协议层 OK。用各中间件的原生协议做一次最小语义校验：

import pymysql, redis, kafka

def test_l2_protocol():
    # MySQL: 能 SELECT 1，且 server_version 匹配预期
    conn = pymysql.connect(host=DEPS["mysql"][0], port=DEPS["mysql"][1],
                           user="ci_test", password="[已脱敏]",
                           connect_timeout=5, read_timeout=5)
    cur = conn.cursor()
    cur.execute("SELECT VERSION()")
    ver = cur.fetchone()[0]
    assert ver.startswith("8.0"), f"MySQL 版本异常: {ver}"
    # 确认不在只读模式
    cur.execute("SELECT @@read_only")
    assert cur.fetchone()[0] == 0, "MySQL 处于只读模式，无法写入测试数据"
    conn.close()

    # Redis: PING + 确认非只读（非 replica）
    r = redis.Redis(host=DEPS["redis"][0], port=DEPS["redis"][1],
                    socket_timeout=5, decode_responses=True)
    assert r.ping()
    info = r.info()
    assert info["role"] == "master", f"Redis 角色异常: {info['role']}"
    assert info["connected_clients"] < 500, \
        f"Redis 连接数过高({info['connected_clients']})，可能有僵尸连接"

    # Kafka: 能 list topics，且 broker 列表非空
    from kafka import KafkaConsumer
    consumer = KafkaConsumer(bootstrap_servers=f"{DEPS['kafka'][0]}:{DEPS['kafka'][1]}",
                             api_version=(3, 4), request_timeout_ms=5000)
    topics = consumer.topics()
    assert len(topics) > 0, "Kafka 无可用 topic，broker 可能正在选举 leader"
    consumer.close()

L3：业务语义就绪（< 15s）

这一层最关键，也最容易被忽略。确认中间件不仅能通信，而且能正常执行读写：

def test_l3_semantic():
    # MySQL: 写一行 → 读回来 → 删掉，确认主从同步 OK
    conn = pymysql.connect(...)
    cur = conn.cursor()
    check_val = f"pf_check_{int(time.time()*1000)}"
    cur.execute("INSERT INTO _ci_health_check (val, ts) VALUES (%s, NOW())", (check_val,))
    conn.commit()
    # 关键：从从库读，验证主从延迟
    cur.execute("/*+ READ_FROM_REPLICA */ SELECT val FROM _ci_health_check WHERE val=%s", (check_val,))
    row = cur.fetchone()
    assert row is not None, "主从延迟：写入未同步到从库，环境未就绪"
    cur.execute("DELETE FROM _ci_health_check WHERE val=%s", (check_val,))
    conn.commit()
    conn.close()

    # Redis: SET → GET → DEL，确认 AOF 落盘（不是内存态）
    r = redis.Redis(...)
    key = f"_ci_pf_{int(time.time()*1000)}"
    r.set(key, "ok", ex=10)
    assert r.get(key) == b"ok"
    # 强制 fsync 确认 AOF 写入
    r.config_set("appendfsync", "always")
    r.set(key, "ok2", ex=10)
    r.config_set("appendfsync", "everysec")  # 恢复
    r.delete(key)

    # Kafka: 发一条消息 → 消费组拉回来，确认 producer/consumer 通路
    from kafka import KafkaProducer, KafkaConsumer
    producer = KafkaProducer(bootstrap_servers=f"{DEPS['kafka'][0]}:{DEPS['kafka'][1]}",
                             acks="all", api_version=(3, 4))
    test_topic = "_ci_health"
    msg = b"pf_check_" + str(int(time.time()*1000)).encode()
    producer.send(test_topic, msg).wait(timeout=5)
    producer.flush()

    consumer = KafkaConsumer(test_topic,
                             group_id=f"_ci_pf_{int(time.time())}",
                             auto_offset_reset="latest",
                             bootstrap_servers=f"{DEPS['kafka'][0]}:{DEPS['kafka'][1]}",
                             api_version=(3, 4),
                             consumer_timeout_ms=5000)
    records = list(consumer)
    consumer.close()
    assert any(msg in (r.value or b"") for r in records), \
        "Kafka 消费通路异常：发送成功但消费组未收到"

L4：资源余量（< 5s）

最后一层，确认资源还有余量，而不是"刚好能跑"：

def test_l4_resources():
    # MySQL: 活跃连接数 < 最大连接数的 80%
    conn = pymysql.connect(...)
    cur = conn.cursor()
    cur.execute("SHOW GLOBAL STATUS LIKE 'Threads_connected'")
    connected = int(cur.fetchone()[1])
    cur.execute("SHOW GLOBAL VARIABLES LIKE 'max_connections'")
    max_conn = int(cur.fetchone()[1])
    ratio = connected / max_conn
    assert ratio < 0.8, f"MySQL 连接数 {connected}/{max_conn} ({ratio:.0%})，余量不足"

    # Redis: 内存使用率 < 75%
    r = redis.Redis(...)
    info = r.info("memory")
    used = info["used_memory"]
    max_mem = info["maxmemory"]
    if max_mem > 0:
        assert used / max_mem < 0.75, \
            f"Redis 内存 {used}/{max_mem} ({used/max_mem:.0%})，余量不足"

    # 磁盘：数据目录剩余空间 > 10GB
    import shutil
    disk = shutil.disk_usage("/var/lib/mysql")
    assert disk.free > 10 * 1024**3, f"磁盘剩余 {disk.free/1024**3:.1f}GB，不足 10GB"

集成到 CI 的入口

# conftest.py
import pytest

@pytest.fixture(scope="session", autouse=True)
def preflight_checks(request):
    """CI 入口：跑完 4 层检查，全过才允许进入用例执行。"""
    results = []
    for layer in [test_l1_network, test_l2_protocol, test_l3_semantic, test_l4_resources]:
        try:
            layer()
            results.append(f"✓ {layer.__name__}")
        except Exception as e:
            results.append(f"✗ {layer.__name__}: {e}")
            pytest.exit(
                f"[PRE-FLIGHT ABORT]\n" + "\n".join(results) +
                "\n\n环境未就绪，已中止。请检查基础设施，勿重跑用例。",
                returncode=2  # 区别于用例失败(1)，CI 可据此跳过通知
            )
    # 全过，打日志
    request.config.getplugin("logging-plugin").logger.info("[PRE-FLIGHT] 全部通过\n" + "\n".join(results))

GitLab CI 配置里把 exit code 2 映射为 "skipped" 而非 "failed"，避免触发误报告警：

# .gitlab-ci.yml
interface_regression:
  script:
    - python -m pytest -v --junitxml=report.xml
  allow_failure:
    exit_codes: [2]   # pre-flight abort 不算 CI 失败

四、踩过的坑与反直觉之处

坑 1：L3 的 Kafka 检查在"空 topic"上会假阳性。 最初我直接 consumer.poll(timeout_ms=5000) 等消息，如果 topic 是新建的、还没人发过消息，consumer 会等到超时——看起来像"消费通路异常"，其实是 topic 为空。后来改成先自己发一条再拉，问题消失。教训：健康检查的"读"必须依赖自己产生的"写"，不能依赖别人的数据。

坑 2：L4 的 Redis 内存检查在 maxmemory=0 时除零。 Redis 默认 maxmemory=0（无限制），直接 used/max_mem 会 ZeroDivisionError。我加了一个 if max_mem > 0 的保护，但更根本的做法是：在 CI 环境里显式设置 maxmemory，把"无限制"变成"有上限"，让资源余量检查有意义。我们设了 maxmemory 4gb + maxmemory-policy allkeys-lru。

坑 3：pre-flight 本身成了新的 flaky 源。 上线第一周，L3 的 MySQL 主从延迟检查有 3 次误报——原因是 staging 的从库在做 innodb_purge（大事务回滚后的清理），期间读延迟飙到 2s。我的检查只等了 500ms。后来把 L3 的超时从 500ms 放宽到 3s，并加了 2 次重试（间隔 1s），误报归零。教训：健康检查的超时阈值不是"越快越好"，而是要匹配被检查对象的最坏恢复时间。

坑 4：exit code 2 在 GitLab 里的行为不符合预期。 GitLab 默认把非零 exit code 都当 "failed"，allow_failure.exit_codes 只是标记为 "allowed to fail"（橙色），还是会触发通知。最终我们用了 GitLab 的 when: on_failure 规则 + 自定义 script 判断 exit code，把 pre-flight abort 的 CI job 状态设为 "skipped"（灰色），彻底不触发告警。这个坑花了 2 天排查，因为 GitLab 文档里 "allowed to fail" 和 "skipped" 的语义差异写得极其隐晦。

反直觉：pre-flight 检查的耗时（约 25s）占全量回归（52min）的 0.8%，但拦住了 78% 的假失败。 这笔账怎么算都划算。但团队第一反应是"能不能别加这 25 秒"——因为 CI 已经够慢了。我后来用数据说服了他们：每拦一次假失败，省 11 分钟排查 + 3 分钟重跑 + 5 分钟等人确认"是不是环境问题"，一次就省了 19 分钟。按每月 12 次假失败算，每月省 228 分钟（3.8 小时），而 pre-flight 的成本是每月 30 次 × 25s = 12.5 分钟。ROI 超过 18 倍。

五、结论与可复用清单

核心原则：把"环境就绪"从隐式假设变成显式契约，用 4 层递进检查（网络 → 协议 → 语义 → 余量）在 25 秒内完成"体检"，任何一层失败就 abort，不进入用例执行。

可复用清单（按优先级）：

1. 先加 L1 + L2（1 天工作量）：TCP 握手 + 协议层 PING，成本最低、拦截率最高（我们统计 L1+L2 就拦住了 52% 的假失败）。
2. 再补 L3（2-3 天）：业务语义读写验证，针对你的中间件定制。关键是"自己写自己读"，不依赖他人数据。
3. 最后加 L4（1 天）：资源余量检查，防止"能跑但跑不动"。
4. exit code 区分：pre-flight abort 用独立 exit code（如 2），CI 配置里映射为 "skipped"，不触发误报告警。
5. 超时阈值 = 被检查对象最坏恢复时间 × 2，别用"直觉值"。
6. 每层检查打结构化日志（[L1] PASS mysql 3ms），方便后续统计哪层最常拦截。

最后打个比方：pre-flight check 就像飞机起飞前的 checklist——飞行员不是"觉得飞机没问题"才推油门，而是逐项确认、逐项签字。CI 也一样：不是"连上了就开始跑"，而是逐项验证、全过才放行。25 秒的 checklist，换 78% 的假失败消失，这笔买卖，怎么算都不亏。