固定出口IP质量探测与健康检查的工程实践

线上业务依赖固定出口 IP 的场景不少：对端接口鉴权按 IP 白名单放行、账号登录地要长期一致、数据回传链路要对账可追溯。这类出口一旦出现延迟抖动、丢包或归属漂移，表现就是接口超时、登录被判定异常、回传数据缺失。被动等业务报障再排查，定位链条长、时效差，更靠谱的做法是主动探测、量化出口质量，把"感觉不稳"变成可比较的指标。

本文给出一套可在生产落地的出口质量探测与健康检查方案：从探测维度设计、单点基础探测、并发探测脚本、健康评分模型到定时调度与告警，整条链路配齐代码和数据表。

一、先把探测维度想清楚

出口质量不是一个"通不通"能概括的，至少要拆成五个维度，每个维度对应一个可量化的指标和采集方式。

其中，“归属一致性”常常最容易被忽略，但它偏偏是固定出口最核心的特征之一：如果一条出口的 ASN 查出来今天属于运营商 A，明天却变成数据中心 B，那就说明它本质上是会漂移的动态出口。哪怕延迟再低，业务侧的登录地异常判定也还是绕不过去。把这一项纳入探测，才能提前揪出那种“延迟没涨，但归属已经变了”的隐性故障。

二、单点基础探测

先把最基础的三件事用 bash 跑通：延迟丢包、TCP 握手耗时、路径与归属。这几条命令是后面并发脚本的"原子操作"，单跑也能快速判断一条出口的大致状态。

# 连续 100 包，统计丢包率与延迟抖动（mdev 越小越稳）
ping -c 100 -i 0.2 -q 203.0.113.10 | tail -2
# rtt min/a vg/max/mdev = 38.1/44.5/82.3/6.2 ms
# 100 packets transmitted, 100 received, 0% packet loss

# 拆解 DNS / TCP 握手 / 总耗时，重点看 tcp 这一档
curl -o /dev/null -s -w "dns=%{time_namelookup}s tcp=%{time_connect}s total=%{time_total}sn" 
  https://203.0.113.10/health --max-time 5
# dns=0.001s tcp=0.045s total=0.210s

# 路径跳数与稳定性，重点看跳数会不会抖、有没有绕远
traceroute -n -q 1 -w 2 203.0.113.10

# 归属一致性：ASN 与 ISP，定期跑一次和基线对比
whois 203.0.113.10 | grep -iE "^(origin|originas|netname|orgname|country):"
# Origin:     AS7922
# NetName:     COMCAST-7922
# Country:     US

单点命令能判断单条出口，但生产环境往往有多条出口要同时盯，而且要高频采样统计 P50/P99。手跑 ping 显然不够，得写脚本。

三、Python 并发探测脚本

把上面的 TCP 握手探测封装成异步任务，对一组出口端点并发探测，连续采样 N 轮，汇总成功率、丢包率、P50/P99 延迟。用 asyncio 是因为握手是 IO 密集型，并发能把一轮采样的墙钟时间压到接近最慢的那一条。

import asyncio
import time
import json
from collections import defaultdict


class ExitProbe:
    """对一组固定出口端点做 TCP 握手探测，汇总延迟、丢包、成功率。"""

    def __init__(self, targets, timeout=3.0):
        # targets: [(name, host, port), ...]
        self.targets = targets
        self.timeout = timeout

    async def _tcp_handshake(self, host, port):
        loop = asyncio.get_event_loop()
        start = time.monotonic()
        try:
            fut = loop.create_connection(lambda: asyncio.Protocol(), host, port)
            await asyncio.wait_for(fut, timeout=self.timeout)
            return True, (time.monotonic() - start) * 1000
        except (asyncio.TimeoutError, OSError):
            return False, None

    async def probe_one(self, name, host, port):
        ok, rtt = await self._tcp_handshake(host, port)
        return name, {"ok": ok, "rtt_ms": round(rtt, 1) if rtt else None}

    async def run(self, rounds=20, interval=1.0):
        stats = defaultdict(lambda: {"ok": 0, "fail": 0, "rtts": []})
        for _ in range(rounds):
            tasks = [self.probe_one(n, h, p) for n, h, p in self.targets]
            for name, r in await asyncio.gather(*tasks):
                if r["ok"]:
                    stats[name]["ok"] += 1
                    stats[name]["rtts"].append(r["rtt_ms"])
                else:
                    stats[name]["fail"] += 1
            await asyncio.sleep(interval)
        return self._summary(stats)

    @staticmethod
    def _summary(stats):
        out = {}
        for name, v in stats.items():
            total = v["ok"] + v["fail"]
            rtts = sorted(v["rtts"])
            out[name] = {
                "success_rate": round(v["ok"] / total * 100, 2),
                "loss_rate": round(v["fail"] / total * 100, 2),
                "p50_ms": round(rtts[len(rtts) // 2], 1) if rtts else None,
                "p99_ms": round(rtts[int(len(rtts) * 0.99)], 1) if rtts else None,
            }
        return out


if __name__ == "__main__":
    targets = [
        ("exit-a", "203.0.113.10", 443),
        ("exit-b", "198.51.100.20", 443),
    ]
    probe = ExitProbe(targets, timeout=3.0)
    summary = asyncio.run(probe.run(rounds=20, interval=1.0))
    print(json.dumps(summary, indent=2, ensure_ascii=False))

一次 20 轮采样的输出大概长这样：

{
  "exit-a": {"success_rate": 100.0, "loss_rate": 0.0, "p50_ms": 45.2, "p99_ms": 78.6},
  "exit-b": {"success_rate": 95.0, "loss_rate": 5.0, "p50_ms": 120.5, "p99_ms": 410.3}
}

光有原始指标还不够，两条出口谁好谁坏、该不该告警，需要一个统一的评分把多维指标压成一个数。

四、健康评分与判定

把四个维度按权重加权成 0–100 的健康分，再按分数区间判一个状态。权重不是死的，连通和延迟权重高是因为它们直接决定业务可用性，丢包次之，归属权重低但设了"一票否决"的语义——归属一变，哪怕延迟再低也该告警。

def health_score(metrics, baseline):
    """按四维加权计算健康分（满分 100）。"""
    score = 0.0
    # 连通成功率（40）
    score += min(40, 40 * metrics["success_rate"] / 99.5)
    # 延迟 P99（30）：落在 good 以内满分，超过 max 零分，中间线性
    p99 = metrics["p99_ms"] or baseline["p99_max"]
    if p99 <= baseline["p99_good"]:
        score += 30
    else:
        score += max(0, 30 * (baseline["p99_max"] - p99)
                         / (baseline["p99_max"] - baseline["p99_good"]))
    # 丢包率（20）：<1% 满分，≥5% 零分
    loss = metrics["loss_rate"]
    score += 20 if loss < 1 else max(0, 20 * (5 - loss) / 4)
    # 归属一致性（10）
    score += 10 if metrics.get("asn_match") else 0
    return round(score, 1)


def judge(score):
    if score >= 90:
        return "健康"
    if score >= 75:
        return "降级"
    return "异常"
if __name__ == "__main__":
    baseline = {"p99_good": 150, "p99_max": 400}
    sample = {"success_rate": 100.0, "loss_rate": 0.0,
              "p99_ms": 78.6, "asn_match": True}
    s = health_score(sample, baseline)
    print(f"score={s} state={judge(s)}")
    # score=100.0 state=健康

分数落到哪个区间，对应不同处置：

五、定时调度与告警

探测脚本要持续跑才有意义，挂到 cron 里按分钟采样，结果落日志。告警别一抖就发，出口偶尔一次握手失败很正常，连续 N 次低于阈值才告，加冷却时间避免抖动误报。

# crontab：每分钟探测一次，结果追加日志
* * * * * /opt/exit-probe/probe.py >> /var/log/exit-probe.log 2>&1

# 告警冷却：连续 3 次评分 < 75 才触发，恢复后冷却 10 分钟

日志里每行是一条 JSON 摘要，配合 jq 或日志平台就能直接出趋势曲线，比单独搭一套监控轻得多。

六、探测对象与基线参照

探测脚本本身对任何可达 IP 通用，换成哪条出口都能跑。做基线标定时，需要一条归属稳定、地址长期不变的出口当参照基准——我用的一条 LinkStatic 静态 ISP 固定出口，whois 查出来 Origin 是 AS7922、归属为真实住宅运营商，连续 72 小时探测 P99 延迟稳定在 80ms 以内、丢包 0%，地址长期不变，刚好拿来做健康基线。这组数据只说明这一条出口在这一次测试周期的表现，换地区换节点都要重新跑，脚本不挑出口来源。

小结

整条链路可以拆成这样几个环节：维度设计 → 基础探测 → 并发采样 → 评分判定 → 调度告警。它真正有价值的地方，是把过去那种“总觉得出口不太稳”的模糊判断，变成一组可以横向比较、持续跟踪的数字。比如延迟 P99、丢包率、归属一致性，这些指标一旦被量化，出了问题就能直接顺着数据去定位，不必再等业务侧先投诉、再倒推原因。尤其是归属一致性，特别适合做常驻探测，因为它盯住的正是那类最容易被忽略的隐性故障：延迟看着没涨，但出口其实已经悄悄漂了。