Skip to content

智能DNS 解析跑偏了怎么查?用 IP 归属地校验 GeoIP 数据的三个步骤(Python 脚本) ​

上周有个做站群的朋友发来一个问题:他给网站在云解析里配了智能线路,华东走 A 节点、华南走 B 节点,结果华南用户反馈"打开比之前慢"。查了半天 Nginx 日志、回源日志、CDN 报表,都没看出异常——因为问题不在服务器上,在解析上:华南的递归解析器把域名解到了华东的节点,用户跨了半个中国去拉资源。

这类问题有个共同点:症状在业务侧,根因在数据侧。智能 DNS、CDN 的节点调度,判断"用户在哪"靠的是 GeoIP 数据(IP 与地理位置的映射库)。这份数据不准、过期、或者跟你的预期不一致,调度就指错了地方,而你在监控上看不出任何红灯。

这篇讲怎么用 IP 归属地查询接口,给智能 DNS / CDN 的调度做一次数据校验,不用买任何东西,一个 Python 脚本就能出偏差报告。

一、先搞清楚解析为什么会跑偏 ​

排查之前得知道对手是谁,常见的跑偏原因有五类,后两类最容易被忽略:

  1. GeoIP 库版本旧。IP 段是会重新分配的,运营商之间也会互相划转。半年前的库把某个 /24 段标在华东,今天这个段已经拨给华南的移动宽带了。
  2. 数据源口径不同。同一批 IP,不同的 IP 库会给出不同结果——有的按注册机构(RIR)登记地,有的按 BGP 通告,有的用网络测量点反推。你在 A 家买的服务、用 B 家的数据验收,对不上很正常。
  3. 递归解析器的视角差异。智能 DNS 一般靠 EDNS Client Subnet(ECS)拿到用户网段再决策,如果递归解析器不支持 ECS、或者自己就被 CDN 就近放到了别的地方,DNS 只能按解析器出口 IP 给答案——给到的是解析器所在地的节点,不是用户所在地的节点。
  4. 你的查询链路本身被劫持或走了代理。在机房、云服务器上排查的时候特别常见:机器出口在境外或者被本地网络改写过 53 端口流量,你测到的"用户视角"根本是假的。
  5. 缓存和 TTL。线路切换后旧记录还在递归服务器上待着,TTL 没到点,切换看起来"没生效"。

第 4 点意味着:校验动作必须从多个解析器做,只看一台机器等于没测。

二、三步校验法 ​

整个流程拆成三步,每步都能单独拿出来用:

第一步,拉多源解析结果。 拿同一个域名,分别向阿里 DNS(223.5.5.5)、腾讯 DNS(119.29.29.29)、百度 DNS(180.76.76.76)、Google DNS(8.8.8.8)查 A 记录。相当于借这几个解析器的"眼睛"看域名解析到了哪台机器。

第二步,把解析出的 IP 拿去查归属地。 这里要做三件事:去重(多个解析器给同一个答案时只查一次)、缓存(进程内字典,跑批时不重复消耗额度)、限速(免费接口是 60 次/分钟的量级,脚本里每个 IP 之间隔 1.1 秒,稳一点)。

第三步,跟调度预期比对,输出偏差表。 预期是什么由你自己定义:按大区(华东/华北/华南)比、按省份比、按运营商比。比对不通过的行标出来,人工看一眼就知道该改 GeoIP 数据还是改调度规则。

为什么不直接 ping 或者 traceroute?因为 ICMP 在云环境里经常被禁,而你真正要确认的不是"能不能通",而是"分给你的地址属于哪个地区"。归属地是数据问题,不是连通性问题。

三、Python 实现 ​

脚本运行环境:Python 3 + dig 命令(没有 dig 的话 yum install bind-utils / apt install dnsutils,或者把解析部分换成 dnspython)。请求用标准库的 urllib,不需要额外装依赖。

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import re
import subprocess
import sys
import time
from urllib.parse import urlencode
from urllib.request import Request, urlopen

RESOLVERS = [
    ("223.5.5.5", "阿里DNS"),
    ("119.29.29.29", "腾讯DNS"),
    ("180.76.76.76", "百度DNS"),
    ("8.8.8.8", "GoogleDNS"),
]
API = "https://ip9.com.cn/get"
UA = "Mozilla/5.0 (X11; Linux x86_64) IP9-check/1.0"   # 必须带 UA,否则可能被前置防护拦掉
CACHE = {}


def dig(domain, server):
    """向指定 DNS 服务器查询 A 记录,返回 IP 列表"""
    try:
        out = subprocess.run(
            ["dig", "+short", "+time=3", "+tries=1", "@" + server, domain, "A"],
            capture_output=True, text=True, timeout=8).stdout
    except subprocess.TimeoutExpired:
        return []
    ips = []
    for line in out.splitlines():
        line = line.strip()
        if re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", line):
            ips.append(line)
    return ips


def geo(ip):
    """查 IP 归属地:内存缓存 + 限速(免费版 60 次/分钟/IP)"""
    if ip in CACHE:
        return CACHE[ip]
    url = API + "?" + urlencode({"ip": ip})
    req = Request(url, headers={"User-Agent": UA})
    try:
        with urlopen(req, timeout=5) as r:
            body = json.loads(r.read().decode("utf-8"))
    except Exception as e:                  # 超时 / 网络异常:不阻断整个巡检
        CACHE[ip] = {"error": str(e)}
        return CACHE[ip]
    if body.get("ret") != 200:
        info = {"error": "ret=%s" % body.get("ret")}
    else:
        d = body.get("data") or {}
        info = {
            "country": d.get("country", ""),
            "prov": d.get("prov", ""),
            "city": d.get("city", ""),
            "big_area": d.get("big_area", ""),
            "isp": d.get("isp", ""),
        }
    CACHE[ip] = info
    time.sleep(1.1)                         # 限速:每次查询至少隔 1.1 秒
    return info


def check(domain, expect):
    """expect 里给了哪个字段就比哪个字段,比如 {"big_area": "华南"}"""
    print("== %s(预期:%s)" % (domain, expect or "—"))
    seen = {}
    for server, label in RESOLVERS:
        for ip in dig(domain, server):
            seen.setdefault(ip, []).append(label)
    if not seen:
        print("   解析失败:多个解析器都没拿到结果")
        return
    for ip, who in seen.items():
        g = geo(ip)
        if g.get("error"):
            print("   %-16s %-14s 查询失败(%s)" % (ip, ",".join(who), g["error"]))
            continue
        verdict = "一致"
        if expect.get("big_area") and g["big_area"] and g["big_area"] != expect["big_area"]:
            verdict = "偏差:大区 %s != %s" % (g["big_area"], expect["big_area"])
        elif expect.get("prov") and g["prov"] and g["prov"] != expect["prov"]:
            verdict = "偏差:省份 %s != %s" % (g["prov"], expect["prov"])
        print("   %-16s %-18s %s%s %s [%s] %s" % (
            ip, ",".join(who), g["country"], g["prov"], g["city"], g["isp"], verdict))


if __name__ == "__main__":
    targets = sys.argv[1:] or ["www.example.com"]
    start = time.time()
    for t in targets:
        check(t, {"big_area": "华南"})      # 按自己的调度预期改
    print("\n共查询 %d 个不同 IP,耗时 %.1fs" % (len(CACHE), time.time() - start))

用法很简单:python3 dns_geo_check.py www.example.com shop.example.com,一次性把核心域名都过一遍。

四、我实跑了一遍,看到了什么 ​

在一台云服务器上跑上面这段脚本,查两个域名,结果挺说明问题:

== www.ip9.com.cn(预期:华南)
   125.24.249.195   阿里DNS              泰国曼谷 曼谷 [TOT] 一致
   43.132.80.155    腾讯DNS,GoogleDNS    新加坡  [腾讯云/CDN] 一致
   183.204.111.150  百度DNS              中国河南 新乡 [中国移动] 偏差:大区 华中 != 华南
   111.32.232.134   百度DNS              中国天津 天津 [中国移动] 偏差:大区 华北 != 华南

一个域名,三个答案:阿里 DNS 给的是泰国 TOT 的地址,腾讯 DNS 和 Google DNS 给的是新加坡腾讯云节点,百度 DNS 给的是河南、天津的移动地址。16 个不同 IP,脚本跑完 22.3 秒(基本都花在限速等待上)。

这里要分两层看:

  • 正常的那层:这台机器出口在境外,腾讯 DNS、Google DNS 按它所在的网络位置,把域名指向了新加坡、泰国的 CDN 节点——这是就近调度的常规结果,跟你"华东走 A、华南走 B"的国内线路是两码事。想校验国内线路,得从国内的机器或者国内的探测点发起。
  • 要警惕的那层:一台机器上向四个解析器发问,答案差这么远(同一家 CDN 的节点,阿里 DNS 给泰国 TOT、腾讯 DNS 给新加坡腾讯云、百度 DNS 给河南和天津的移动地址),得先怀疑自己的查询链路——这台机器的 53 端口流量有没有被本地网络改写?出口是不是在境外? 校验之前先确认这一点,否则你拿到的"用户视角"是伪造的。

更关键的是这句话:IP 归属地能证明"这个地址被分配给了哪个地区/运营商",但不能证明"这台机器物理上在哪儿"。 上面 43.132.80.155 归属地写新加坡、运营商写腾讯云,说的是这个 IP 段登记在新加坡的腾讯云手里;节点机架具体在哪个机房,归属地数据答不了。排序、调度这类决策要的是前者,别把它当后者用。

五、落地时的几个细节 ​

偏差不等于故障,先分类再动手。 anycast(同一 IP 在多地同时宣告)天生"归属地只有一个,服务点有多个",这类偏差属于正常;真正要修的是"某个大区的用户被稳定解析到跨大区节点"这种系统性偏差。

比大区,别比城市。 IP 归属地是城市级数据,城市边界附近、运营商双出口、动态 IP 都会让市名漂移。调度规则写到大区(华东/华南)是稳的,写到"广州市"就要准备接受误判。校验脚本也一样,{"big_area": "华南"} 比 {"city": "广州"} 靠谱得多。

海外的 IP 经常只有国家。 查 8.8.8.8 返回的是 country=美国、省和市都是空的,经纬度还是美国地理中心点。港澳台不一样,203.198.0.1 会返回 prov=香港、big_area=港澳台。写比对逻辑时这两类要分开处理,海外 IP 的偏差判断只能看国家。

内网和保留地址要单独排除。 查 192.168.1.1 会返回 ret=200,但 country=保留、prov/city 全空、isp 写着"内网地址"。如果脚本只判断 ret == 200 就当有效数据,这些空值会污染统计。

别让巡检把额度吃掉。 免费版 60 次/分钟是按调用方 IP 算的,一个 100 台机器的集群各跑一遍脚本很容易顶爆。做法是只在一台机器上跑(或者收敛到一次任务),结果写进文件让其他系统读;要覆盖更细的维度(区县、机房类型、AS 号)或者更高的调用量,就得上 VIP 版,那是按次计费的,18 万次/分钟的额度够大多数巡检用了。

把它做成定时任务。 校验的价值在"变的时候能发现"。每天凌晨跑一次,跟昨天的报告 diff 一下,出现新的偏差行就发通知——比每次出事再手工查省事得多,也能顺手发现自己的 GeoIP 数据源是不是该升级了。

总结 ​

智能 DNS、CDN 调度的正确性,最终都压在"IP → 地理位置"这份数据上。校验这件事不需要复杂的工具:多个解析器拉解析结果、批量查归属地、跟预期比对,三步就能把数据侧的偏差暴露出来。上面脚本的核心其实只有三点——去重缓存省额度、限速别打爆接口、比对只信大区,其余都是包装。

脚本里查归属地用的接口是 IP9 的免费接口 https://ip9.com.cn/get?ip=<IP>,不传 ip 参数就返回调用方自己的归属地,IPv4 和 IPv6 都支持,免费版 60 次/分钟、无需注册;返回里 prov/city/big_area/isp 这几个字段正好够做调度校验。字段含义、额度和 VIP/私有化部署的说明都在官网 https://www.ip9.com.cn 上。如果你的场景是日志侧的地域统计,可以顺手看看本站《用IP查询分析网站日志,做出用户地域分布图》;要对比本地离线库和在线接口的取舍,看《ip2region、qqzeng-ip、GeoIP2 和在线接口怎么选》。