fix(node,agent): 转发链路两处慢失败补超时——①节点转发握手(告知目标+等agent回执)新增 HandshakeTimeout 默认5s, agent会话半死时观看端不再无限挂起到点即断、握手成功即清除不约束数据阶段; ②agent 拨内网目标新增 DialTimeout 默认5s(Agent 可配), 目标离线且防火墙丢包时不再裸 Dial 吊20s+, 到点回执失败观看端立即断开; 附 e2e 红绿双向证明(半死agent/黑洞拨号 两场景撤修复必红)
This commit is contained in:
@@ -28,6 +28,9 @@ import (
|
|||||||
const (
|
const (
|
||||||
defaultBackoff = 5 * time.Second
|
defaultBackoff = 5 * time.Second
|
||||||
defaultPing = 30 * time.Second
|
defaultPing = 30 * time.Second
|
||||||
|
// defaultDialTimeout 内网目标拨号上限:目标离线且防火墙丢包(不回 RST)时
|
||||||
|
// 裸 Dial 在 Windows 上要 SYN 重传 20s+, 观看端在节点侧只能干等。
|
||||||
|
defaultDialTimeout = 5 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
// Logger agent 日志接口。logrus.Logger 的 Debugf/Infof/Warnf/Errorf 签名
|
// Logger agent 日志接口。logrus.Logger 的 Debugf/Infof/Warnf/Errorf 签名
|
||||||
@@ -62,6 +65,7 @@ type Agent struct {
|
|||||||
AgentId string
|
AgentId string
|
||||||
Backoff time.Duration // 永久隧道重连退避(默认 5s)
|
Backoff time.Duration // 永久隧道重连退避(默认 5s)
|
||||||
PingEvery time.Duration // 各隧道控制流心跳间隔(默认 30s)
|
PingEvery time.Duration // 各隧道控制流心跳间隔(默认 30s)
|
||||||
|
DialTimeout time.Duration // 内网目标拨号上限(默认 5s; 超时向节点回执失败, 观看端立即断开)
|
||||||
|
|
||||||
// Logger 日志注入点。需要纳入宿主统一日志体系时设置
|
// Logger 日志注入点。需要纳入宿主统一日志体系时设置
|
||||||
// (logrus 直接注入 logrus.StandardLogger() 即可);未设置用 DefaultLogger。
|
// (logrus 直接注入 logrus.StandardLogger() 即可);未设置用 DefaultLogger。
|
||||||
@@ -85,6 +89,7 @@ func New(agentId string) *Agent {
|
|||||||
AgentId: agentId,
|
AgentId: agentId,
|
||||||
Backoff: defaultBackoff,
|
Backoff: defaultBackoff,
|
||||||
PingEvery: defaultPing,
|
PingEvery: defaultPing,
|
||||||
|
DialTimeout: defaultDialTimeout,
|
||||||
Logger: DefaultLogger{},
|
Logger: DefaultLogger{},
|
||||||
tunnels: make(map[string]*tunnelConn),
|
tunnels: make(map[string]*tunnelConn),
|
||||||
done: make(chan struct{}),
|
done: make(chan struct{}),
|
||||||
|
|||||||
+5
-1
@@ -22,7 +22,11 @@ func (a *Agent) handleData(stream net.Conn) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
conn, err := net.Dial("tcp", t.TargetAddr())
|
dialTimeout := a.DialTimeout
|
||||||
|
if dialTimeout <= 0 { // 零值 Agent 兜底
|
||||||
|
dialTimeout = defaultDialTimeout
|
||||||
|
}
|
||||||
|
conn, err := net.DialTimeout("tcp", t.TargetAddr(), dialTimeout)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
a.logger().Warnf("agent 拨号本地目标失败 target=%s err=%s", t.TargetAddr(), err.Error())
|
a.logger().Warnf("agent 拨号本地目标失败 target=%s err=%s", t.TargetAddr(), err.Error())
|
||||||
rf := NewFrame(FrameVersion, CmdTarget, f.StreamID())
|
rf := NewFrame(FrameVersion, CmdTarget, f.StreamID())
|
||||||
|
|||||||
@@ -0,0 +1,136 @@
|
|||||||
|
// 转发链路"慢失败"测试:两处历史缺陷的回归锚——
|
||||||
|
// 1. 节点转发握手无上限:agent 会话半死(收得到帧但永不应答)时观看端无限挂起
|
||||||
|
// → HandshakeTimeout 到点必须切断观看端, 让其快速失败可重试;
|
||||||
|
// 2. agent 拨内网目标无上限:目标离线且防火墙丢包(不回 RST)时裸 Dial 吊 20s+
|
||||||
|
// → DialTimeout 到点必须向节点回执失败, 观看端立即断开。
|
||||||
|
package e2e
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
|
"strconv"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.zeroonesoft.cn/golib/zonat/agent"
|
||||||
|
"git.zeroonesoft.cn/golib/zonat/internal/node"
|
||||||
|
|
||||||
|
"github.com/xtaci/smux"
|
||||||
|
)
|
||||||
|
|
||||||
|
// deafAgent 手工模拟"半死 agent":完成登录+注册隧道后, 对节点打开的数据流
|
||||||
|
// (CmdTarget)永不应答。返回节点为该隧道分配的公网端口。
|
||||||
|
func deafAgent(t *testing.T, n *node.Node, tunnelId, targetIp string, targetPort int) int {
|
||||||
|
t.Helper()
|
||||||
|
conn, err := net.DialTimeout("tcp", n.Addr(), 2*time.Second)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { conn.Close() })
|
||||||
|
|
||||||
|
tok, err := node.SignToken(n.JwtSecret, "deaf-agent", time.Hour)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
lf := agent.NewFrame(agent.FrameVersion, agent.CmdLogin, 0)
|
||||||
|
if err := lf.Marshal(agent.Login{AgentId: "deaf-agent", Token: tok}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := agent.WriteFrame(conn, lf); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
rf, err := agent.ReadFrame(conn)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
ret := agent.Ret{}
|
||||||
|
if err := rf.Unmarshal(&ret); err != nil || ret.Code != 0 {
|
||||||
|
t.Fatalf("登录失败: %+v", ret)
|
||||||
|
}
|
||||||
|
|
||||||
|
sess, err := smux.Client(conn, agent.SmuxConfig())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
control, err := sess.OpenStream()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
reg := agent.NewFrame(agent.FrameVersion, agent.CmdRegisterTunnel, 0)
|
||||||
|
if err := reg.Marshal(agent.Tunnel{Id: tunnelId, TargetIp: targetIp, TargetPort: targetPort, TTLSec: 0}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := agent.WriteFrame(control, reg); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
rr, err := agent.ReadFrame(control)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
rret := agent.Ret{}
|
||||||
|
if err := rr.Unmarshal(&rret); err != nil || rret.Code != 0 {
|
||||||
|
t.Fatalf("注册失败: %+v", rret)
|
||||||
|
}
|
||||||
|
return rret.Port
|
||||||
|
}
|
||||||
|
|
||||||
|
// assertViewerCutBy 断言观看端在时限内被节点断开(读到 EOF/重置, 而非超时)。
|
||||||
|
func assertViewerCutBy(t *testing.T, port int, within time.Duration) {
|
||||||
|
t.Helper()
|
||||||
|
viewer, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 2*time.Second)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer viewer.Close()
|
||||||
|
_ = viewer.SetReadDeadline(time.Now().Add(within))
|
||||||
|
buf := make([]byte, 16)
|
||||||
|
_, err = viewer.Read(buf)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("观看端读到了数据, 不应发生")
|
||||||
|
}
|
||||||
|
var netErr net.Error
|
||||||
|
if errors.As(err, &netErr) && netErr.Timeout() {
|
||||||
|
t.Fatalf("观看端 %v 内未被节点断开(超时切断未生效)", within)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHandshakeTimeoutCutsViewer 半死 agent: 节点必须在握手超时内切断观看端
|
||||||
|
// (修复前观看端无限挂起, 只能等 smux keepalive 判死整个会话 ~40s)。
|
||||||
|
func TestHandshakeTimeoutCutsViewer(t *testing.T) {
|
||||||
|
target := startEchoServer(t)
|
||||||
|
n := node.New()
|
||||||
|
n.JwtSecret = "test-jwt-secret"
|
||||||
|
n.BindTunnel = "127.0.0.1"
|
||||||
|
n.HandshakeTimeout = 500 * time.Millisecond
|
||||||
|
if err := n.Start("127.0.0.1:0"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Cleanup(n.Stop)
|
||||||
|
|
||||||
|
host, portStr, _ := net.SplitHostPort(target)
|
||||||
|
portNum, _ := strconv.Atoi(portStr)
|
||||||
|
listen := deafAgent(t, n, fmt.Sprintf("deaf-%d", time.Now().UnixNano()), host, portNum)
|
||||||
|
|
||||||
|
assertViewerCutBy(t, listen, 3*time.Second)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestAgentDialTimeoutCutsViewer 目标黑洞(不回 RST): agent 拨号到点回执失败,
|
||||||
|
// 观看端立即断开(修复前裸 Dial 在 Windows 吊 20s+, 观看端干等)。
|
||||||
|
func TestAgentDialTimeoutCutsViewer(t *testing.T) {
|
||||||
|
// 本机网络若把 TEST-NET-3 判定为可达(隧道/虚拟网卡), 黑洞前提不成立
|
||||||
|
if c, err := net.DialTimeout("tcp", "203.0.113.1:5900", 300*time.Millisecond); err == nil {
|
||||||
|
c.Close()
|
||||||
|
t.Skip("本机可将 TEST-NET-3 判定连通, 黑洞拨号场景不成立")
|
||||||
|
}
|
||||||
|
|
||||||
|
n, a := startEnv(t)
|
||||||
|
a.DialTimeout = 300 * time.Millisecond
|
||||||
|
id := fmt.Sprintf("blackhole-%d", time.Now().UnixNano())
|
||||||
|
port, err := a.RegisterTunnel(buildTunnel(t, n, a, id, "203.0.113.1:5900", 0))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
assertViewerCutBy(t, port, 3*time.Second)
|
||||||
|
}
|
||||||
@@ -27,6 +27,7 @@ type Node struct {
|
|||||||
JwtSecret string // agent 登录 JWT 校验密钥(HS256 系)
|
JwtSecret string // agent 登录 JWT 校验密钥(HS256 系)
|
||||||
BindTunnel string // 隧道监听绑定地址(默认 0.0.0.0)
|
BindTunnel string // 隧道监听绑定地址(默认 0.0.0.0)
|
||||||
SweepInterval time.Duration // TTL 清扫间隔(默认 30s)
|
SweepInterval time.Duration // TTL 清扫间隔(默认 30s)
|
||||||
|
HandshakeTimeout time.Duration // 单条观看连接的 agent 拨号应答上限(默认 5s; 超时立即断开观看端, 数据阶段不受限)
|
||||||
|
|
||||||
listener net.Listener
|
listener net.Listener
|
||||||
|
|
||||||
@@ -41,6 +42,7 @@ func New() *Node {
|
|||||||
return &Node{
|
return &Node{
|
||||||
BindTunnel: "0.0.0.0",
|
BindTunnel: "0.0.0.0",
|
||||||
SweepInterval: 30 * time.Second,
|
SweepInterval: 30 * time.Second,
|
||||||
|
HandshakeTimeout: 5 * time.Second,
|
||||||
sessions: make(map[*Session]struct{}),
|
sessions: make(map[*Session]struct{}),
|
||||||
tunnels: make(map[string]*TunnelServer),
|
tunnels: make(map[string]*TunnelServer),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -58,6 +58,11 @@ func (t *TunnelServer) handleConn(conn net.Conn) {
|
|||||||
}
|
}
|
||||||
defer stream.Close()
|
defer stream.Close()
|
||||||
|
|
||||||
|
// 转发握手(告知目标+等 agent 回执)限时: agent 会话半死/目标拨号不回时
|
||||||
|
// 观看端不再无限挂起, 到点立即断开让观看端快速失败可重试
|
||||||
|
if hs := t.session.node.HandshakeTimeout; hs > 0 {
|
||||||
|
_ = stream.SetDeadline(time.Now().Add(hs))
|
||||||
|
}
|
||||||
tf := agent.NewFrame(agent.FrameVersion, agent.CmdTarget, 0)
|
tf := agent.NewFrame(agent.FrameVersion, agent.CmdTarget, 0)
|
||||||
if err := tf.Marshal(t.tunnel); err != nil {
|
if err := tf.Marshal(t.tunnel); err != nil {
|
||||||
return
|
return
|
||||||
@@ -67,8 +72,10 @@ func (t *TunnelServer) handleConn(conn net.Conn) {
|
|||||||
}
|
}
|
||||||
rf, err := agent.ReadFrame(stream)
|
rf, err := agent.ReadFrame(stream)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
slog.Warn("节点 agent拨号应答失败", "id", t.tunnel.Id, "err", err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
_ = stream.SetDeadline(time.Time{}) // 数据阶段不受握手超时约束
|
||||||
ret := agent.Ret{}
|
ret := agent.Ret{}
|
||||||
if err := rf.Unmarshal(&ret); err != nil || ret.Code != 0 {
|
if err := rf.Unmarshal(&ret); err != nil || ret.Code != 0 {
|
||||||
slog.Warn("节点 agent拨号失败", "id", t.tunnel.Id, "msg", ret.Msg)
|
slog.Warn("节点 agent拨号失败", "id", t.tunnel.Id, "msg", ret.Msg)
|
||||||
|
|||||||
Reference in New Issue
Block a user