观察、决策、行动三个方框,旁边强调 TCP 保活探测对端而非应用健康。

TCP 保活探测空闲连接

TCP keepalive 询问的是:连接安静了一段时间后,对面的 TCP 端点是否仍然响应。它不询问 HTTP handler 是否正常、数据库查询是否推进、后台工作线程是否卡住。即使应用已经停顿,内核仍然可能确认收到的数据。因此,“传输对端能够回应”和“应用给出了有用结果”必须作为两个不同的观察项。 传输层探针不是应用心跳 保活有助于暴露远端连接状态已经消失,或者网络路径悄悄停止传递报文的情况。实现会在满足空闲条件后发送探针,并按照未响应探针的策略最终报告失败;但单个探针丢失不等于对端死亡,不能把网络中的临时丢包直接解释成确定性故障。 探针在应用字节流以下处理,不会变成 HTTP 请求、数据库查询或 WebSocket 心跳。如果连接终止在代理上,探测对象就是代理自身的 TCP 端点,不是代理后面的源站。某些中间设备的连接跟踪可能受探针影响,但不能由此承诺刷新应用的空闲计时,更不能保证所有 NAT 映射都会一直保留。 已经发出而尚未确认的数据还有自己的重传与失败规则,因此保活参数不是繁忙连接的通用总截止时间。正常 FIN、连接重置、应用读取超时和静默丢包是不同事件,恢复动作也未必相同。知道是哪一层报告了什么错误,比统一把它们写成“连接死了”更有用。 配置真正承载请求的 socket RFC 1122 对可选保活机制要求默认关闭,并规定默认空闲间隔不得短于两小时且应可配置。但应用和库可以显式启用并选择更短间隔,Go 的拨号器就有自己的设置。因此,查到操作系统默认值,并不能直接推出当前应用实际使用的策略;还需要沿着创建连接的代码检查有效配置。 Linux 的 tcp_keepalive_time、tcp_keepalive_intvl 和 tcp_keepalive_probes 分别表示探测前的空闲秒数、探针间隔和未响应探针数量,文档默认值为 7200、75、9。启用 SO_KEEPALIVE 后,还可以通过 TCP_KEEPIDLE、TCP_KEEPINTVL、TCP_KEEPCNT 为单个 socket 定制策略。这些名称与接口不是 macOS、Windows 通用的系统参数,不能跨平台照搬命令。 ...

十月 5, 2027 · 2 分钟 · XB Field Notes
客户端默认路由保持有效,服务器缺少回复路由仍会令通信失败。

默认路由匹配最宽范围,不会挽救所有失败路径

IPv4 默认路由是前缀 0.0.0.0/0,匹配所有 IPv4 目的地,但具体程度最低。在已经选择的路由表中,更长的匹配前缀通常优先。默认路由不是一个“其他具体路径失败后再重试”的过程。路由策略决定相关表,最终路由描述怎样转发,而不保证完整往返一定成功。匹配范围与路径健康是两类不同属性。 坏掉的样子 没有默认路由的机器,仍能访问直连或明确路由覆盖的目的地;反过来,添加默认路由也不证明能够访问互联网。下一跳需要可达,转发与过滤要允许流量,回复还需要返回路径。显示一个网关,只满足契约的一部分。真正配置是否共同工作,应由实际应用通信证明,而不是从路由文件中有一行 default 就推导全部能力。 更具体的拒绝路由会胜过可工作的默认路由。Linux 的 unreachable 等路由类型,会有意停止匹配目的地的转发。选定网关的邻居查询失败,也不意味着内核自动寻找更健康、却更宽泛的路径。静态路由优先级和 metric 不是服务健康监控或自动故障转移机制,不能把配置顺序当成逐个尝试可用出口的流程。 返回路径错误尤其容易误导。客户端选中预期网关并把流量送往监听中的服务器,服务器却可能无法把回复路由回客户端。反复修改客户端默认路由,可能破坏其他正常目的地,却没有修复缺失的反向路径。将所有连接超时归因于发送方网关之前,应查看两个方向和真实源地址。请求能到与响应能回,需要分别成立。 快速验证 将脚本保存为 check-default-route.py,准备 Docker 后运行 python3 check-default-route.py。Python 3.11.16 和 iproute2 6.1.0 在一次性 Linux 容器内建立客户端、路由器、目标拓扑。NET_ADMIN 与 SYS_ADMIN 支持嵌套命名空间和临时 proc 挂载。接口全是新建 veth,不向宿主发布端口或路由,也不修改现有网络设备: from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 iputils-ping=3:20221126-1+deb12u1 nftables=1.0.6-2+deb12u2 tcpdump=4.99.3-1 traceroute=1:2.1.2-1 procps=2:4.0.2-3 && rm -rf /var/lib/apt/lists/*\nCMD ["python3"]\n' program=r""" from contextlib import ExitStack,contextmanager from pathlib import Path from tempfile import TemporaryDirectory import json,os,select,signal,subprocess,time def run(*args,check=True,input=None,timeout=8): result=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout) if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr) return result def ns(name,*args,**kwargs):return run('ip','netns','exec',name,*args,**kwargs) def ip(name,*args,**kwargs):return ns(name,'ip',*args,**kwargs) def stop(process): if process.poll() is None: process.terminate() try:process.wait(timeout=2) except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2) def line(process,timeout=3): assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded' value=process.stdout.readline();assert value,'child exited without a response' return value def spawn(stack,command): process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) stack.callback(stop,process);return process @contextmanager def lab(names): with TemporaryDirectory(prefix='net-path-') as directory,ExitStack() as stack: root=Path(directory);proc=root/'proc';proc.mkdir();run('mount','-t','proc','proc',str(proc)) stack.callback(run,'umount',str(proc)) for name in names: run('ip','netns','add',name);stack.callback(run,'ip','netns','delete',name,check=False) ip(name,'link','set','lo','up') def setting(name,key,value): assert name in names path=proc/'sys'/key.replace('.','/') return ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r}); p.write_text({str(value)!r}); print(p.read_text().strip())').stdout.strip() yield root,stack,setting def wire(left,left_if,right,right_if,left_addresses,right_addresses): run('ip','link','add',left_if,'type','veth','peer','name',right_if) for name,interface,addresses in [(left,left_if,left_addresses),(right,right_if,right_addresses)]: run('ip','link','set',interface,'netns',name) for address in addresses:ip(name,'address','add',address,'dev',interface,*(['nodad'] if ':' in address else [])) ip(name,'link','set',interface,'up') def nft(name,script):return ns(name,'nft','-f','-',input=script) def counters(name): data=json.loads(ns(name,'nft','-j','list','ruleset').stdout);found={} for entry in data['nftables']: rule=entry.get('rule',{}) if 'comment' in rule: for expression in rule['expr']: if 'counter' in expression:found[rule['comment']]=expression['counter']['packets'] return found def server(stack,name,address,port=18080,label='server'): code=r''' from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer import json,socket,sys address,port,label=sys.argv[1],int(sys.argv[2]),sys.argv[3] class Handler(BaseHTTPRequestHandler): def do_GET(self): body=json.dumps({'server':label,'peer':self.client_address,'path':self.path}).encode() self.send_response(200);self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body) def log_message(self,*args):pass class Server(ThreadingHTTPServer): address_family=socket.AF_INET6 if ':' in address else socket.AF_INET def server_bind(self): if self.address_family==socket.AF_INET6:self.socket.setsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY,1) super().server_bind() service=Server((address,port),Handler);print('ready',flush=True);service.serve_forever() ''' process=spawn(stack,['ip','netns','exec',name,'python3','-u','-c',code,address,str(port),label]) assert line(process).strip()=='ready';return process def request(name,address,port=18080): code=r''' import http.client,json,sys client=http.client.HTTPConnection(sys.argv[1],int(sys.argv[2]),timeout=.7) try: client.request('GET','/lab-visible-message');response=client.getresponse();body=json.loads(response.read()) assert response.status==200;print(json.dumps({'ok':True,'status':response.status,'body':body})) except OSError as error:print(json.dumps({'ok':False,'type':type(error).__name__,'errno':error.errno})) finally:client.close() ''' return json.loads(ns(name,'python3','-c',code,address,str(port)).stdout) def routed(setting): wire('client','c0','router','r0',['10.201.1.2/30'],['10.201.1.1/30']) wire('router','r1','target','t0',['10.201.2.1/30'],['10.201.2.2/30']) setting('router','net.ipv4.ip_forward',1) ip('client','route','add','default','via','10.201.1.1') ip('target','route','add','default','via','10.201.2.1') with lab(['client','router','target']) as (root,stack,setting): routed(setting);ip('client','route','delete','default') ip('target','addr','add','198.51.100.9/32','dev','lo');ip('router','route','add','198.51.100.9/32','via','10.201.2.2') daemon=server(stack,'target','198.51.100.9') connected=ns('client','ping','-n','-c','1','-W','1','10.201.1.1');assert connected.returncode==0 missing=request('client','198.51.100.9');assert not missing['ok'] and missing['errno']==101 ip('client','route','add','default','via','10.201.1.1') chosen=json.loads(ip('client','-j','route','get','198.51.100.9').stdout)[0];assert chosen['gateway']=='10.201.1.1' healthy=request('client','198.51.100.9');assert healthy['ok'] ip('client','route','add','unreachable','198.51.100.9/32') specific=request('client','198.51.100.9');assert not specific['ok'] and specific['errno']==113 ip('client','route','delete','unreachable','198.51.100.9/32') recovered=request('client','198.51.100.9');assert recovered['ok'] ip('target','route','delete','default') bad_return=request('client','198.51.100.9');assert not bad_return['ok'] and bad_return['type']=='TimeoutError' assert json.loads(ip('client','-j','route','get','198.51.100.9').stdout)[0]['gateway']=='10.201.1.1' ip('target','route','add','default','via','10.201.2.1') final=request('client','198.51.100.9');assert final['ok'] print(json.dumps({'connected_ping_without_default':0,'no_default':missing,'chosen_route':chosen,'with_default':healthy, 'specific_unreachable_route':specific,'specific_removed':recovered,'missing_return_route':bad_return,'return_route_restored':final})) """ name='xbcatm-netpath-'+uuid.uuid4().hex[:10] def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'Dockerfile').write_text(recipe) build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN','--cap-add','SYS_ADMIN', '--memory','256m','--pids-limit','128',name,'python3','-',input=program) assert result.returncode==0,result.stderr print(json.dumps(json.loads(result.stdout),sort_keys=True)) finally: docker('rm','-f',name);docker('image','rm',name) 客户端使用 10.201.1.2,路由邻居为 10.201.1.1。目标位于路由器后的另一条链路,在回环接口拥有服务地址 198.51.100.9。路由器有到该服务的具体路由,目标起初具有返回默认路由。文档保留地址仅用于隔离拓扑,没有联系公网服务器。每项地址和下一跳都在脚本中明确配置,避免依赖宿主偶然存在的路由。 ...

九月 14, 2027 · 2 分钟 · XB Field Notes
接口地址不变,前缀变化仍可能改变内核选中的下一跳。

子网前缀影响直连路由,最终路径由路由表选择

IPv4 前缀长度表示地址中有多少前导位标识网络。对于 192.0.2.5/24,网段是 192.0.2.0/24,普通广播地址是 192.0.2.255。配置接口通常会建立相应直连路由,但前缀不是完整路由算法。内核还要依据路由策略与路由表,为具体目的地选择路径,不能只看地址是否位于某个网段就推断全部行为。 坏掉的样子 前缀太短,会让主机把原本远端的目的地当成直接相连。保留地址 192.0.2.5,却把 /24 改成 /16,就把 192.0.3.9 纳入直连范围。在真实以太网中,这种错误可能让主机尝试解析一个根本不在当前链路上的邻居。于是地址看起来正确,下一跳决策却已经改变;反复核对地址字符串本身无法发现问题。 “不在自己的子网就走默认网关”也不完整。更具体的路由可以选择其他网关或接口;在已经选择的路由表内,更长的匹配前缀优先。明确的主机路由甚至可以覆盖本来位于接口子网内的目的地。应检查实际查询结果,而不是仅凭接口地址重新猜测内核会怎么做。默认路由只处理没有更具体匹配的相应流量。 地址数量计算也需要上下文。普通 /24 的网络地址与定向广播地址通常不用于主机分配,但“减二”不是统一公式。RFC 3021 允许点到点 /31 链路的两个地址都用作端点,/32 则标识单个地址。这些规则解释地址语义,没有证明配置中的两端能够通过当前物理拓扑交换数据。可用地址集合和实际通信仍须区分。 快速验证 将脚本保存为 check-subnet.py,准备 Docker 后执行 python3 check-subnet.py。程序给固定 Python 镜像安装 iproute2 6.1.0,再只向一次性容器授予 NET_ADMIN;容器没有外部网络接入。脚本创建虚拟 dummy 接口并修改该容器的路由,不改变宿主接口、路由或生产网关。构建下载与运行时路由实验也处于不同阶段: from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid recipe='''FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84 RUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 && rm -rf /var/lib/apt/lists/* ''' program=r''' from ipaddress import IPv4Address,IPv4Interface,IPv4Network import json,subprocess def ip(*args): result=subprocess.run(['ip',*args],capture_output=True,text=True,timeout=3) assert result.returncode==0,result.stderr return result.stdout def route(address):return json.loads(ip('-j','route','get',address))[0] interface=IPv4Interface('192.0.2.5/24');network=interface.network assert str(network.network_address)=='192.0.2.0' and str(network.broadcast_address)=='192.0.2.255' assert IPv4Address('192.0.2.77') in network and IPv4Address('192.0.3.9') not in network try:IPv4Network('192.0.2.5/24');raise AssertionError('host bits accepted in strict network') except ValueError:strict_rejected=True point_to_point=[str(a) for a in IPv4Network('192.0.2.0/31').hosts()] assert point_to_point==['192.0.2.0','192.0.2.1'] ip('link','add','lab0','type','dummy');ip('link','set','lab0','up') ip('address','add','192.0.2.5/24','dev','lab0');ip('route','add','default','via','192.0.2.1','dev','lab0') ip('route','add','203.0.113.0/24','via','192.0.2.2','dev','lab0') direct=route('192.0.2.77');specific=route('203.0.113.9');default=route('198.51.100.9');before=route('192.0.3.9') assert direct['dev']=='lab0' and 'gateway' not in direct assert specific['gateway']=='192.0.2.2' and default['gateway']==before['gateway']=='192.0.2.1' ip('route','add','192.0.2.77/32','via','192.0.2.1','dev','lab0');overridden=route('192.0.2.77') assert overridden['gateway']=='192.0.2.1' ip('address','del','192.0.2.5/24','dev','lab0');ip('address','add','192.0.2.5/16','dev','lab0') ip('route','replace','default','via','192.0.2.1','dev','lab0');wrong=route('192.0.3.9') assert wrong['dev']=='lab0' and 'gateway' not in wrong print(json.dumps({'interface':str(interface),'network':str(network),'broadcast':str(network.broadcast_address), 'strict_host_bits_rejected':strict_rejected,'point_to_point_31_hosts':point_to_point, 'connected':direct,'more_specific_route':specific,'default':default,'host_route_overrides_subnet':overridden, 'destination_before_widening':before,'destination_after_wrong_16':wrong,'iproute_version':ip('-V').strip()})) ''' name='xbcatm-subnet-'+uuid.uuid4().hex[:10] def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'Dockerfile').write_text(recipe) build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN', '--memory','128m','--pids-limit','64',name,'python3','-',input=program) assert result.returncode==0,result.stderr print(json.dumps(json.loads(result.stdout),sort_keys=True)) finally: docker('rm','-f',name);docker('image','rm',name) Python 区分接口 192.0.2.5/24 与其网段 192.0.2.0/24。严格网络解析拒绝前者,因为主机位非零;/31 示例则列出 192.0.2.0 和 192.0.2.1 两个可用端点。这个计算确认地址集合,实验没有配置可工作的点到点链路,也没有测试这两个地址之间的转发。不能把库返回两个值,写成真实链路已经验证成功。 ...

九月 13, 2027 · 2 分钟 · XB Field Notes
跨独立网络的路由器交换可达性与策略通告。

BGP 组成了互联网的路由表

BGP 是唯一把几万个各自独立运营的网络缝成一张转发表的协议。它承载可达性,但它真正承载的是策略:每个网络说明自己起源哪些前缀、从哪个邻居学到它们,以及通过过滤器说明自己愿意把哪些继续传下去。 代价是这张表建立在彼此之间没有合同的信任之上。通告在扩散之前不会被任何东西校验,所以一处笔误或一条配错的过滤器,几分钟内就是全球性的,而且要等到有人注意到才会消失。 AS 路径同时是环路检测和策略信号 在出方向,eBGP 发言者会把自己的 AS 号前插,收到更新的一方如果发现 AS_PATH 里已经有自己的 AS 就直接丢弃。这就是环路防护的全部机制,也是路由绕成一圈后无法被重新引入的原因。iBGP 不做前插,所以一条 iBGP 学到的路由带着与把它带进本 AS 的那条 eBGP 路由相同的 AS_PATH;也正因如此,一个 AS 需要 iBGP 全互联或路由反射器才能把路由分发下去。 同一个字段还兼着另外两份工作。运营商用它做过滤——拒绝任何包含某个对等体 AS 的路径,只接受路径末端是客户 AS 的路由——而路径选择则比较它的长度。一个字段,三份职责,而且互相冲突:为了让一条路径显得不吸引人而做前插,同时也会让它看起来像是过滤器本该拒绝的路由。所以前插这类手段通常只在自家门口有效,一旦越过别人的过滤器边界就会失效。 为什么这张表大约是一百万条前缀 真实数字,取自本文核验时抓取的 CIDR Report 快照:IPv4 前缀 1,080,189 条;把能聚合的都塌缩掉则是 597,258 条。仅 Amazon 一家就起源 18,025 条前缀,其中 15,759 条是更大块的更具体前缀。整张表里大约有一半,是已经被通告过的地址空间上再切出来的更具体前缀。 ...

九月 12, 2027 · 1 分钟 · XB Field Notes
确认被过滤时,客户端可进入已建立状态,而服务器仍停留在 SYN-RECV。

TCP 握手建立序列号空间,不是同步两台机器的时钟

TCP 三次握手建立字节流的两个方向。每端提供自己的初始序列号,并确认另一端的序列号。这是两个独立序列空间,不是共享墙上时钟;应用建立连接不需要两台机器时钟同步。之后的序列号用于标识字节,支持连接内部的排序、重复处理、确认和重传。把序列号与时间混为一谈,会误解握手真正建立的协议状态。 坏掉的样子 客户端发送带初始序列号的 SYN,服务器回复 SYN-ACK,给出自己的初始序列号并确认客户端 SYN,客户端再确认服务器响应。即使没有应用数据,SYN 也占用序列空间中的一个位置,所以对应确认号增加一。将这些数值当成时间戳,会错误解释报文,还可能把排障引向无关的时钟同步问题,忽略真正缺失的确认或路由。 两端不会在完全相同的时刻改变状态。客户端收到 SYN-ACK 后,connect 可能成功返回,但最后一个确认在到达服务器前丢失。服务器仍可能停留在 SYN-RECEIVED,Linux 工具显示为 SYN-RECV。后续携带适当确认的数据也能完成交换,因此单独第三个报文缺失,并不必然导致永久失败,不能把抓包中的固定三包模式当成唯一合法过程。 TCP 建立成功也不代表应用健康或对端身份可信。服务可能接受连接却无法生成有效响应,代理可能完成客户端这一侧连接,却无法建立另一个独立上游连接。初始序列号选择有助于区分连接实例、抵御预测攻击,但不是身份认证协议,也不会加密字节流。传输状态、应用结果和安全身份需要各自对应的验证,不能由一项成功代替全部。 快速验证 将脚本保存为 check-tcp-handshake.py,准备 Docker 后运行 python3 check-tcp-handshake.py。一次性容器使用 Python 3.11.16、iproute2 6.1.0、nftables 1.0.6 和 tcpdump 4.99.3,在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套网络命名空间。客户端和目标由新建 veth 相连,不移动宿主接口或发布端口: from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 iputils-ping=3:20221126-1+deb12u1 nftables=1.0.6-2+deb12u2 tcpdump=4.99.3-1 traceroute=1:2.1.2-1 procps=2:4.0.2-3 && rm -rf /var/lib/apt/lists/*\nCMD ["python3"]\n' program=r""" from contextlib import ExitStack,contextmanager from pathlib import Path from tempfile import TemporaryDirectory import json,os,select,signal,subprocess,time def run(*args,check=True,input=None,timeout=8): result=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout) if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr) return result def ns(name,*args,**kwargs):return run('ip','netns','exec',name,*args,**kwargs) def ip(name,*args,**kwargs):return ns(name,'ip',*args,**kwargs) def stop(process): if process.poll() is None: process.terminate() try:process.wait(timeout=2) except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2) def line(process,timeout=3): assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded' value=process.stdout.readline();assert value,('child exited without a response',process.stderr.read()) return value def spawn(stack,command): process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) stack.callback(stop,process);return process @contextmanager def lab(names): with TemporaryDirectory(prefix='net-path-') as directory,ExitStack() as stack: root=Path(directory);proc=root/'proc';proc.mkdir();run('mount','-t','proc','proc',str(proc)) stack.callback(run,'umount',str(proc)) for name in names: run('ip','netns','add',name);stack.callback(run,'ip','netns','delete',name,check=False) ip(name,'link','set','lo','up') def setting(name,key,value): assert name in names path=proc/'sys'/key.replace('.','/') return ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r}); p.write_text({str(value)!r}); print(p.read_text().strip())').stdout.strip() yield root,stack,setting def wire(left,left_if,right,right_if,left_addresses,right_addresses): run('ip','link','add',left_if,'type','veth','peer','name',right_if) for name,interface,addresses in [(left,left_if,left_addresses),(right,right_if,right_addresses)]: run('ip','link','set',interface,'netns',name) for address in addresses:ip(name,'address','add',address,'dev',interface,*(['nodad'] if ':' in address else [])) ip(name,'link','set',interface,'up') def nft(name,script):return ns(name,'nft','-f','-',input=script) def counters(name): data=json.loads(ns(name,'nft','-j','list','ruleset').stdout);found={} for entry in data['nftables']: rule=entry.get('rule',{}) if 'comment' in rule: for expression in rule['expr']: if 'counter' in expression:found[rule['comment']]=expression['counter']['packets'] return found def capture(stack,root,name,interface,expression,label): path=root/(label+'.pcap') process=subprocess.Popen(['ip','netns','exec',name,'tcpdump','--immediate-mode','-Z','root','-U','-n','-i',interface,'-w',str(path),expression], stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,text=True) stack.callback(stop,process) deadline=time.monotonic()+3 while time.monotonic()<deadline: assert process.poll() is None,'tcpdump exited before listening' if select.select([process.stderr],[],[],.1)[0] and 'listening on' in process.stderr.readline():return process,path raise AssertionError('tcpdump did not become ready') def packets(process,path): import struct,socket process.send_signal(signal.SIGINT);_,diagnostic=process.communicate(timeout=3) assert process.returncode==0,diagnostic raw=path.read_bytes();assert raw[:4]==bytes.fromhex('d4c3b2a1') and struct.unpack_from('<I',raw,20)[0]==1 offset=24;result=[] while offset<len(raw): seconds,micros,size,original=struct.unpack_from('<IIII',raw,offset);offset+=16 frame=raw[offset:offset+size];offset+=size if frame[12:14]!=b'\x08\x00' or frame[23]!=6:continue ihl=(frame[14]&15)*4;length=struct.unpack_from('!H',frame,16)[0];tcp=14+ihl sport,dport,seq,ack=struct.unpack_from('!HHII',frame,tcp);header=(frame[tcp+12]>>4)*4 opts={};cursor=tcp+20 while cursor<tcp+header: kind=frame[cursor] if kind==0:break if kind==1:cursor+=1;continue option_length=frame[cursor+1];data=frame[cursor+2:cursor+option_length];assert option_length>=2 if kind==2:opts['mss']=struct.unpack('!H',data)[0] elif kind==3:opts['wscale']=data[0] elif kind==4:opts['sack_ok']=True elif kind==5:opts['sack']=[list(pair) for pair in struct.iter_unpack('!II',data)] cursor+=option_length body=frame[tcp+header:14+length] result.append({'t':seconds+micros/1e6,'src':socket.inet_ntoa(frame[26:30]),'dst':socket.inet_ntoa(frame[30:34]), 'sport':sport,'dport':dport,'seq':seq,'ack':ack,'flags':frame[tcp+13],'window':struct.unpack_from('!H',frame,tcp+14)[0], 'length':len(body),'options':opts,'preview':body[:24].hex()}) return result with lab(['client','target']) as (root,stack,setting): wire('client','c0','target','t0',['192.0.2.1/30'],['192.0.2.2/30']) server_code=r''' import socketserver class Handler(socketserver.StreamRequestHandler): def handle(self): data=self.rfile.readline();self.wfile.write(b'echo:'+data);self.wfile.flush() class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True server=Server(('192.0.2.2',18080),Handler);print('ready',flush=True);server.serve_forever() ''' service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='ready' tap,path=capture(stack,root,'client','c0','tcp port 18080','handshake') normal=ns('client','python3','-c', 'import socket; s=socket.socket();s.settimeout(2);s.bind(("192.0.2.1",40100));s.connect(("192.0.2.2",18080));s.sendall(b"normal\\n");print(s.makefile("rb").readline().decode().strip());s.close()') assert normal.stdout.strip()=='echo:normal' nft('client','''table ip missing_ack { chain output { type filter hook output priority 0; policy accept; tcp dport 18080 tcp flags & (syn|ack|rst|fin) == ack counter drop comment "missing_final_ack" } } ''') client_code=r''' import json,socket,sys s=socket.socket();s.settimeout(.25);s.bind(('192.0.2.1',40101));s.connect(('192.0.2.2',18080)) print(json.dumps({'connect_returned':True}),flush=True);s.sendall(b'deferred\n') try:s.recv(64);raise AssertionError('unexpected application response') except TimeoutError:print(json.dumps({'application_timeout':True}),flush=True) assert sys.stdin.readline().strip()=='restore';s.settimeout(5) received=b'' while not received.endswith(b'\n'): chunk=s.recv(64);assert chunk,'premature EOF';received+=chunk assert received==b'echo:deferred\n',received print(json.dumps({'recovered':received.decode().strip()}),flush=True);s.close() ''' client=spawn(stack,['ip','netns','exec','client','python3','-u','-c',client_code]) connected=json.loads(line(client));failure=json.loads(line(client));assert connected['connect_returned'] and failure['application_timeout'] pending=ns('target','ss','-nHt','state','syn-recv').stdout;assert '192.0.2.1:40101' in pending,pending blocked=counters('client');assert blocked['missing_final_ack']>0 nft('client','delete table ip missing_ack');client.stdin.write('restore\n');client.stdin.flush() recovery=json.loads(line(client,6));assert recovery['recovered']=='echo:deferred' client.wait(timeout=2);assert client.returncode==0 observed=packets(tap,path);flow=[p for p in observed if 40100 in (p['sport'],p['dport'])] syn=next(p for p in flow if p['flags']&0x12==0x02) synack=next(p for p in flow if p['flags']&0x12==0x12) third=next(p for p in flow if p['sport']==40100 and p['flags']&0x12==0x10) assert synack['ack']==(syn['seq']+1)%2**32 assert third['seq']==(syn['seq']+1)%2**32 and third['ack']==(synack['seq']+1)%2**32 print(json.dumps({'three_messages':[syn,synack,third],'normal_reply':normal.stdout.strip(), 'client_connect_returned':connected['connect_returned'],'application_timeout':failure['application_timeout'], 'server_still_syn_recv':pending.strip(),'drop_counters':blocked,'after_ack_filter_removed':recovery})) """ name='xbcatm-tcpflow-'+uuid.uuid4().hex[:10] def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'Dockerfile').write_text(recipe) build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN','--cap-add','SYS_ADMIN', '--memory','256m','--pids-limit','128',name,'python3','-',input=program) assert result.returncode==0,result.stderr print(json.dumps(json.loads(result.stdout),sort_keys=True)) finally: docker('rm','-f',name);docker('image','rm',name) 第一条连接使用客户端端口 40100 和目标端口 18080,发送换行结束的请求,收到预期回显。抓包提供真实 SYN、SYN-ACK 和最终 ACK 数值。断言检查服务器确认号等于客户端序列号加一,第三条消息确认号等于服务器序列号加一,并对三十二位序列空间使用取模运算。这样验证的是协议关系,而不是要求每次随机数相同。 ...

九月 10, 2027 · 3 分钟 · XB Field Notes
设备从中心服务器续租一个短命地址。

DHCP 只把地址租给你

DHCP 发的是带时长的租约而不是永久的地址。租期驱动续约机制,也让服务器收回设备已不再使用的地址。 借用而非拥有地址 DHCP 并不给设备一个永久的地址,而是把地址借给它一段有限时间。设备入网时广播去发现服务器,服务器用网络预先配置的租期给出一份租约。正因为分配是临时的,服务器随时知道哪些地址有效,可以当作资源池来统计。这份「临时」恰恰是大型、频繁变动的网络能管理起来的根本:地址是服务器可以分发、续期、收回的货币,而不是某个主机永远占着的财产。 续约让租约保持存活 在租约到期之前,客户端会主动请求续约——通常在租期过半时才提出,不够再续——于是健康设备可以在无人干预的情况下长期保有同一地址。这种「半衰期续租」让地址跨越重启与会话保持稳定,对语音、部分 VPN 这类地址一变就断的服务尤其重要。租期是一个治理旋钮:租期短,地址回收快,但续约流量更密;租期长,地址更稳定,却把早已离开的地址长期拴着不放。 过期即可回收 租约到期又没有续约时,服务器会收回该地址,再把它发给另一台主机。一台许多天前就离开大楼的笔记本,就因此不会一直占着一个新设备正需要的地址。放弃租约的设备下一次入网会拿到一个新地址——对临时上网没问题,对期望固定身份的服务却不友好。调整租约就是在换手率与可用性之间找平衡:拥挤或流动的网络用短租期,内部稳定、身份重要的场景用长租期。 坏掉的样子 地址池耗尽时最安静。新客户端广播 DISCOVER,却收不到 OFFER,或者 REQUEST 换来一个 NAK,笔记本只能拿到 169.254.x.x 的链路本地地址,哪儿也去不了。这时不要让客户端盲目重试:应该扩大地址池,在人员流动大的网络里缩短租期,让离开的设备尽快归还地址,同时清理服务器上因冲突被标记为废弃的绑定。 相反的错误,是把租期设得太短,短到变成风暴。一间有上千台主机的办公室,如果租期只有五分钟,就会有约一半设备每隔几分钟在 T1 时刻单播续约,掉队的还要在 T2 时刻广播重绑,服务器一旦重启,正好撞上续约洪峰。应当保持 RFC 2131 的默认值,T1 取租期一半,T2 取租期的八分之七,先把过短的租期调长,再把开机潮错峰,而不是把续约流量当成免费的。 流氓服务器靠抢答获胜。一台私接的家用路由器或者恶意主机,会抢在真正的服务器之前回答 DISCOVER,给出一个看起来合法、网关或 DNS 却被掉包的地址,把流量引到自己身上。症状会游走:有的租约正常,有的解析异常,受害的客户端随着每次竞速不断变化。对策是固定合法服务器的标识,在交换机上开启 DHCP 监听,只允许受信任端口发送 OFFER 和 ACK,并把流氓设备清走,而不是调定时器。 时钟偏斜破坏的是那份安静的契约。线上传输的是相对秒数,可一旦服务器用绝对时间戳保存过期时刻,或者笔记本一睡就睡过 T1 和 T2,就会提前过期或者超期占用。睡着的机器醒来时还配着旧地址,而服务器早已把它收回转租,两个主机就会争用同一个 IP。应当把长时间睡眠当作一次重绑,排查时把 NTP 和 DHCP 日志放在一起看,并保留冲突检测与 DECLINE、RELEASE 的处理,不要假设时钟永远不会跳变。 ...

九月 7, 2027 · 3 分钟 · XB Field Notes
一个稳定的名字,站在一组会变动的 Pod 前面。

Kubernetes Service 为 Pod 提供稳定入口

既然 Pod 会重建,就很难记住它的地址。Service 在这组 Pod 前面固定一个名字,让调用方不因 Pod 更换而受影响。 一个地址,一组 Pod 多副本带来了多个 Pod,而每个 Pod 都有自己的 IP,还会随时被重建。Service 正是它们前面那个稳定的家:你声明一个选择器,平台就分给它一个固定的集群 IP 和一段可被 DNS 解析的名字。集群 DNS 把名字解析到 Service 上,因此组件(Ingress、负载均衡、别的 Pod)只依赖这个名字,而从不关心具体某个 Pod 的地址。抽象的全部意义就在于:名字是持久的,背后站着的每一样东西都可以即丢即换。 选择规则决定由谁响应 Pod 靠「匹配」加入 Service,而不是靠被人点名登记:标签打中了选择器,它就进入;不再匹配或不再就绪,就被移出。平台从选择器推出后端这份视图,当一个 Pod 通过探针与就绪,就把它加进去,缩容或下线时就拿掉。所以后端列表是一份实时推导的视图,而不是从头到尾手打的名单。也正因为后端由选择器与就绪共同决定,一个没就绪的 Pod 会自己掉出去,流量绝不会被送到不该服务的 Pod 上。 流量总给当前匹配的 这层间接,正是让端点变化落不到调用方身上的原因。客户端只握着那个名字;它背后的集合被扩、被替、被滚动更新,都不需要改客户端设一个字。滚动更新一个接一个地替换掉又续上 Pod,Service 始终把流量导向当前匹配且就绪的那批,于是调用方察觉不到任何断层。你可以依赖这件事:只要选择器精准、Pod 与它匹配,Service 在每个时刻都把流量送进正确的那批。把选择器钉紧、探针管严,这个稳定的家便能在里面的 Pod 全是新面孔时依旧稳如磐石。 ...

八月 23, 2027 · 2 分钟 · XB Field Notes
观察、决策、行动三步示意图,旁边是限制每次等待并共享一个截止点的英文标题。

连接超时阻止无声的停摆

超时首先规定某个操作最多等待多久。它不会自动成为整个请求的总期限,也不能证明远端工作已经停止。选择数值或解释错误之前,要把连接建立、单次读取和调用方的整体预算分开;否则配置看起来都有限,整条调用链仍然可能等得远比预期久。 连上了,不代表能拿到有用的响应 TCP 建连可以成功,而对端应用始终不产生响应。此时问题属于读取阶段,不是 TCP 握手尚未完成。诊断信息应写出发生超时的阶段;含糊的“数据库连接超时”,可能指连接池获取借用失败,也可能指新建套接字失败,或已经连上后等不到查询结果。它们占用的资源、下一步检查和恢复语义并不相同。 Python 套接字上的超时作用于阻塞操作。循环执行多次成功的 recv(),每次都可能获得一份新的等待额度,因此对端只要持续发送小片段,就可能把整个读取循环拖到远超那个超时值。下面特意每次只读一个字节,便于观察单次操作成功与完整响应及时到达之间的差别,而不是把“一次没有超时”当成整个消息已经满足时限。 如果需要整个接收过程共用预算,应先用 time.monotonic() 建立一个截止点。每次开始下一项操作前,计算剩余时间,再取剩余时间与该操作上限中的较小值;预算耗尽后,就不能继续开始新的等待。单调时钟不会因校正系统日期而突然倒退或跳到另一段墙钟时间,但它也不能保证线程恰好在截止的那一刻获得调度,因此不能把它理解成硬实时执行器。 把这个模式扩展到真实客户端之前,仍要核对所用库的契约。域名解析、多个地址尝试、TLS 协商、发送请求以及从连接池排队获取连接,可能有各自的计时方式。本例只访问一个数字形式的回环地址,并使用普通 TCP;它没有证明 DNS 解析已经受限,也没有实现一个覆盖完整 HTTPS 请求流程的通用总超时。 坏掉的样子 已接受连接但保持沉默的对端,会让调用方读到超时;不断滴送少量数据的对端,则能让每次读取都成功,却使完整响应过慢。只有建连超时,并不能处理连接已经成功之后的这两种情况。反过来,设置得过短,也可能把正常的网络波动或线程调度延迟误判成故障,数值需要结合真实调用预算和工作负载来确定。 失败路径需要释放自己拥有的资源,并保留完整数据与部分数据的区别。套接字读取返回空字节串,表示 EOF,并不等于应用消息必然收齐。实验知道目标帧应该有六个字节,所以提前关闭会被当成失败;不能因为连接已经结束,就把收到的半份结果当作正常响应交给业务。 放弃等待也不等于取消远端工作。调用方超时时,对端可能已经执行了某个副作用。重试需要匹配的重放契约,以及原操作还剩下的预算,而不是每次重新获得一整套等待时间。这里的总期限是在操作边界协作检查的约束,仍有调度与清理开销,并不是对任意线程、系统调用或远端执行的瞬时中断保证。 快速验证 将代码保存为 connection_probe.py,执行 python3 connection_probe.py。脚本只依赖标准库,本次在 macOS 的 Python 3.14.4 上验证。每种情况都创建自己的回环监听端口和对端线程,结束时关闭套接字并等待线程退出;不会连接外部主机,也不修改防火墙或生产服务。 import socket import threading import time PAYLOAD = b"abcdef" PER_OPERATION = 0.5 def exercise(trickle, total=None, peer_payload=PAYLOAD): stop = threading.Event() failures = [] with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener: listener.bind(("127.0.0.1", 0)) listener.listen(1) listener.settimeout(2.0) def peer(): try: connection, _ = listener.accept() with connection: connection.settimeout(0.5) if not trickle: stop.wait(2.0) return for byte in peer_payload: if stop.wait(0.1): return connection.sendall(bytes([byte])) except (BrokenPipeError, ConnectionResetError): pass # The deadline case deliberately closes the client. except Exception as error: failures.append(error) worker = threading.Thread(target=peer) worker.start() started = time.monotonic() deadline = None if total is None else started + total received = bytearray() phase, outcome = "connect", "complete" def allowance(): if deadline is None: return PER_OPERATION remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("total budget expired") return min(PER_OPERATION, remaining) try: with socket.create_connection(listener.getsockname(), timeout=allowance()) as client: phase = "read" while len(received) < len(PAYLOAD): client.settimeout(allowance()) chunk = client.recv(1) if not chunk: raise EOFError(f"short frame: {len(received)}/{len(PAYLOAD)} bytes") received.extend(chunk) except TimeoutError: outcome = "total" if deadline is not None and time.monotonic() >= deadline else phase finally: elapsed = time.monotonic() - started stop.set() worker.join(2.0) assert not worker.is_alive(), "fixture peer failed to stop" assert not failures, failures return outcome, bytes(received), elapsed quiet = exercise(trickle=False) assert quiet[0] == "read" and quiet[1] == b"" trickle = exercise(trickle=True) assert trickle[0] == "complete" and trickle[1] == PAYLOAD assert trickle[2] >= 0.5 # Each successful recv got a fresh operation timeout. bounded = exercise(trickle=True, total=0.25) assert bounded[0] == "total" and bounded[1] != PAYLOAD try: exercise(trickle=True, peer_payload=b"ab") except EOFError as error: assert str(error) == "short frame: 2/6 bytes" else: raise AssertionError("a truncated frame must not become success") print("early EOF rejects a two-byte fragment of the six-byte frame") for name, result in (("quiet", quiet), ("trickle", trickle), ("bounded", bounded)): outcome, data, elapsed = result print(name, outcome, repr(data), "elapsed", round(elapsed, 3)) quiet 情况先成功建立连接,但没有收到任何数据,大约半秒后报告 read 超时。trickle 情况在发送六个字节之前分别暂停一小段时间,每次读取都落在单次额度内,完整帧却花了超过半秒才到齐。这里的 complete 是指六个预期字节全部到达,不是某一次读取返回了非空内容;这直接展示了逐次操作超时为什么不能代替整个读取循环的预算。 ...

八月 1, 2027 · 2 分钟 · XB Field Notes