春江暮客

春江暮客的个人学习分享网站

用 Python 排查域名解析失败:getaddrinfo、IPv4/IPv6 与等待上限

2026-10-19 技术
用 Python 排查域名解析失败:getaddrinfo、IPv4/IPv6 与等待上限

后台任务报 Temporary failure in name resolution,直接增加 HTTP 重试次数,往往只会延长等待。本文用 Python 标准库做一个独立的主机名解析检查:输出 IPv4、IPv6、错误分类和退出码,再把它放进有等待上限的自动化入口。

适合初学者到中级读者,衔接已有的 Python TCP 端口探测。先确认名字能解析,再检查端口和应用响应,排错顺序更清楚。

1. 方法:检查系统解析结果

socket.getaddrinfo() 官方文档说明它封装系统调用。本工具限定 TCP 类型,按系统返回顺序列出并去重地址,保留 IPv6 的 scope_id,但不建立连接。

在使用 glibc 的 Linux 上,名字来源和查询顺序由 NSS 配置参与决定。因此成功结果可能来自 /etc/hosts、缓存或 DNS,不能证明本次一定访问了 DNS 服务器。

按 Linux getaddrinfo 错误说明区分:

status 含义与下一步
ok 获得至少一个地址;继续验证连接。
temporary_failure EAI_AGAIN;检查当前解析环境,稍后有限重试。
name_or_service_unknown EAI_NONAME;检查拼写、名称来源和地址族。它不是 DNS 报文中的 NXDOMAIN 证据。
resolver_error 其他解析错误;保留实际错误码与文字。
system_error 其他系统错误;按错误码排查。
no_addresses 没有可用地址;作为失败处理。

2. 准备:不安装第三方包

需要 Linux 和 Python 3.9 或更新版本。示例在 Linux、Python 3.11.4 上测试,不修改系统解析配置。

  1. 建立演示目录:
python3 --version
mkdir dns-check-demo
cd dns-check-demo
  1. 将下一节完整代码保存为 check_dns.py。工具接收 ASCII 主机名,例如 localhost 或 www.bobobk.com,支持末尾的点;不接收 URL、端口后缀或 IPv6 字面量。国际化域名先使用对应的 ASCII/Punycode 名称。

3. 实现:一行 JSON 和明确退出码

"""Inspect system hostname resolution, without connecting. Python 3.9+."""
import argparse
import json
import re
import socket
import time

FAMILIES = {"any": socket.AF_UNSPEC, "ipv4": socket.AF_INET,
            "ipv6": socket.AF_INET6}


def hostname(value):
    name = value[:-1] if value.endswith(".") else value
    label = r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
    if len(name) > 253 or not re.fullmatch(label + r"(?:\." + label + r")*", name):
        raise argparse.ArgumentTypeError("use an ASCII hostname, not a URL")
    return value


def resolve(host, port, family):
    report = {"host": host, "port": port, "family": family,
              "status": "ok", "error_code": None, "error": "", "addresses": []}
    started = time.monotonic()
    try:
        rows = socket.getaddrinfo(host, port, FAMILIES[family],
                                  socket.SOCK_STREAM, socket.IPPROTO_TCP)
        seen = set()
        for af, _, _, _, address in rows:
            if af not in (socket.AF_INET, socket.AF_INET6):
                continue
            scope = address[3] if af == socket.AF_INET6 else 0
            key = (af, address[0], scope)
            if key in seen:
                continue
            seen.add(key)
            report["addresses"].append({
                "family": "ipv4" if af == socket.AF_INET else "ipv6",
                "address": address[0], "scope_id": scope,
            })
        if not report["addresses"]:
            report["status"] = "no_addresses"
    except socket.gaierror as exc:
        status = {socket.EAI_AGAIN: "temporary_failure",
                  socket.EAI_NONAME: "name_or_service_unknown"}.get(
                      exc.errno, "resolver_error")
        report.update(status=status, error_code=exc.errno, error=str(exc))
    except OSError as exc:
        report.update(status="system_error", error_code=exc.errno, error=str(exc))
    report["elapsed_ms"] = round((time.monotonic() - started) * 1000, 3)
    return report


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("host", type=hostname)
    parser.add_argument("--port", type=int, default=443)
    parser.add_argument("--family", choices=FAMILIES, default="any")
    args = parser.parse_args()
    if not 1 <= args.port <= 65535:
        parser.error("--port must be between 1 and 65535")
    report = resolve(args.host, args.port, args.family)
    print(json.dumps(report, ensure_ascii=True))
    return 0 if report["status"] == "ok" else 1


if __name__ == "__main__":
    raise SystemExit(main())

--port 只为返回的候选 socket 地址指定数字端口,不会连接该端口。error_code 可能是 EAI_*,也可能是系统 errno,要结合 status 解释;不要硬编码某个平台的错误数字。

4. 使用:先本地,再目标环境

  1. 查看帮助并检查本地 IPv4 解析:
python3 check_dns.py --help
python3 check_dns.py localhost --family ipv4

常见输出如下,地址列表与耗时以本机实际结果为准:

{"host":"localhost","port":443,"family":"ipv4","status":"ok","error_code":null,"error":"","addresses":[{"family":"ipv4","address":"127.0.0.1","scope_id":0}],"elapsed_ms":0.2}
  1. 在发生故障的主机或容器内检查真实域名;这里用本站作演示:
python3 check_dns.py www.bobobk.com
python3 check_dns.py www.bobobk.com --family ipv4
python3 check_dns.py www.bobobk.com --family ipv6

没有 IPv6 记录、系统不支持 IPv6 或地址族受限时,最后一条可能失败。这不自动表示 IPv4 业务故障。记录可能变化,不要把示例网站的 IP 写死在测试里。

退出码 0 表示本次获得地址,1 表示检查失败,2 表示参数错误。解析成功仍需 curl 检查 HTTP 状态与正文。

5. 自动化:用子进程限制等待

getaddrinfo() 没有本工具可设置的超时参数,socket.setdefaulttimeout() 也不是它的解析超时开关。不要把测得的 elapsed_ms 当成等待上限。

  1. 将下面代码保存为同目录的 run_check.py:
"""Run check_dns.py in a child process with a five-second wait budget."""
import json
from pathlib import Path
import subprocess
import sys

script = Path(__file__).resolve().with_name("check_dns.py")
try:
    result = subprocess.run(
        [sys.executable, str(script), *sys.argv[1:]],
        capture_output=True, text=True, timeout=5, check=False,
    )
except subprocess.TimeoutExpired:
    print(json.dumps({"status": "wrapper_timeout", "wait_budget_seconds": 5}))
    raise SystemExit(124)
sys.stdout.write(result.stdout)
sys.stderr.write(result.stderr)
raise SystemExit(result.returncode)

subprocess.run(timeout=...) 文档说明超时后会终止并等待子进程。这里把解析留在独立子进程中,超出等待预算时输出 wrapper_timeout 并退出 124;它与解析器返回 temporary_failure 是两种观察。进程创建和终止本身仍可能增加耗时,因此不承诺严格在五秒内返回。

  1. 在同目录执行;保存退出码后再输出提示,避免掩盖失败:
if python3 run_check.py localhost --family ipv4; then
  printf 'Resolution passed; continue with connection checks\n'
else
  dns_exit=$?
  printf 'Resolution check failed: exit=%s\n' "$dns_exit" >&2
  exit "$dns_exit"
fi

定时任务中使用实际解释器与 run_check.py 的绝对路径,并按 cron 路径与运行环境排查核对账号、工作目录和容器环境。这个预检只反映当时结果,后续客户端仍要处理自己的解析和连接失败。

6. 验证:不依赖公网故障

将下面代码保存为 test_dns.py。localhost 使用真实系统解析;双栈、重复地址、临时失败和其他错误由注入数据验证,不需要改 DNS、断网或等待真实超时。

import errno
import json
from pathlib import Path
import socket
import subprocess
import sys
from unittest.mock import patch
import check_dns

root = Path(__file__).resolve().parent


def run(*args):
    return subprocess.run([sys.executable, str(root / "check_dns.py"), *args],
                          capture_output=True, text=True, timeout=5)


# Actual system resolution; no public hostname or remote connection required.
result = run("localhost", "--family", "ipv4")
report = json.loads(result.stdout)
assert result.returncode == 0 and report["status"] == "ok", report
assert report["addresses"] and report["elapsed_ms"] >= 0
assert all(item["family"] == "ipv4" for item in report["addresses"])

rows = [
    (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.10", 443)),
    (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.10", 443)),
    (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2001:db8::10", 443, 0, 0)),
]
with patch("check_dns.socket.getaddrinfo", return_value=rows) as lookup:
    report = check_dns.resolve("example.test", 443, "any")
    assert report["status"] == "ok" and len(report["addresses"]) == 2, report
    assert [item["family"] for item in report["addresses"]] == ["ipv4", "ipv6"]
    lookup.assert_called_once_with("example.test", 443, socket.AF_UNSPEC,
                                   socket.SOCK_STREAM, socket.IPPROTO_TCP)

for failure, expected in [
    (socket.gaierror(socket.EAI_AGAIN, "injected temporary failure"), "temporary_failure"),
    (socket.gaierror(socket.EAI_NONAME, "injected unknown name"), "name_or_service_unknown"),
    (socket.gaierror(socket.EAI_FAIL, "injected failure"), "resolver_error"),
    (OSError(errno.EACCES, "injected denial"), "system_error"),
]:
    with patch("check_dns.socket.getaddrinfo", side_effect=failure):
        report = check_dns.resolve("example.test", 443, "any")
        assert report["status"] == expected and report["addresses"] == [], report
        assert report["error_code"] == failure.errno, report

with patch("check_dns.socket.getaddrinfo", return_value=[]):
    assert check_dns.resolve("example.test", 443, "any")["status"] == "no_addresses"

for arguments in [("https://example.com",), ("bad host",), ("",),
                  ("localhost", "--port", "0"), ("localhost", "--port", "65536"),
                  ("localhost", "--family", "wrong")]:
    result = run(*arguments)
    assert result.returncode == 2 and not result.stdout and result.stderr, result

print("OK: localhost, dual stack, deduplication, resolver errors, empty result, CLI errors")
python3 test_dns.py
python3 run_check.py localhost --family ipv4

测试脚本应输出:

OK: localhost, dual stack, deduplication, resolver errors, empty result, CLI errors

注入错误只验证分类逻辑,不证明真实 DNS 服务器会返回哪个错误。若本地测试失败,先检查 localhost 的系统解析结果,不要直接改测试为忽略失败。

7. 排错:在实际调用方查看配置

在使用 glibc 且已有 getent 的 Linux 上,可只读对照以下信息:

getent ahosts localhost
getent ahosts www.bobobk.com
cat /etc/nsswitch.conf
cat /etc/resolv.conf

getent ahosts 手册说明它同样调用 getaddrinfo()。它的标志与本工具不完全一致,输出数量和顺序不要求相同;缺少 getent 时仍可运行 Python 工具。

现象 具体动作
temporary_failure 在同一容器内对照解析器配置和网络可达性;记录错误后再决定是否有限重试。
name_or_service_unknown 核对域名拼写,去掉 https:// 和路径;分开检查 IPv4 与 IPv6。
主机成功,容器失败 对照容器自己的 /etc/resolv.conf、NSS 与 hosts 来源;主机结果不能替代容器结果。
解析成功仍连接失败 继续检查 TCP、TLS 和 HTTP,不把解析地址列表当成服务健康证明。
wrapper_timeout 检查解析等待与进程状态;保留退出码 124,不输出伪造的成功结果。

8. 小结

把主机名解析单独检查,可以先确定故障发生在连接之前还是之后。用 JSON 和退出码保留观察结果,再用子进程控制自动化等待。解析成功后继续验证端口与响应内容,才能判断服务是否可用。

标签

1024 12306 ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn chatgpt checkpoint cli cloudflare codex cofoldarena copy cpu监控 cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging deployment desktop-app devtools disk-space disown dns docker dovecot download electron encoding esm esm2 esm3 esmc esmfold2 faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io flashppi flask folium frontend game generator git github-actions google google-research grep gzip harness hls html http hugo imports indexnow javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics mutation mysql nanobert nanobodies networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 phaser pillow pip postfix preprocessing print protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security selenium seo sequence-identity sha256 shell signals sklearn socket solana somaticsignatures spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tcp tensorflow tkinter tls toml tron tronpy troubleshooting turtle typesafe-ai ubuntu usdt uv vhhbert vite webp wordcloud wordpress workflow yaml zipfile zoneinfo 后台 寓言 概率 经济 贸易 迅雷解析 钱包

友情链接

其它