From e112d81697dcedad002f0012706294edeb1721b7 Mon Sep 17 00:00:00 2001 From: sagit <36596628+Sagit-chu@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:37:26 +0800 Subject: [PATCH] fix: bound and configure tunnel quality probes (#533) Closes #528 and #532. --- .../internal/http/handler/control_plane.go | 13 +- go-backend/internal/http/handler/handler.go | 14 ++ .../internal/http/handler/tunnel_best_exit.go | 6 +- .../http/handler/tunnel_best_exit_test.go | 36 +++-- .../http/handler/tunnel_quality_prober.go | 118 +++++++++++++--- .../handler/tunnel_quality_prober_test.go | 126 ++++++++++++++++++ .../internal/monitoring/tunnel_quality.go | 52 ++++++++ .../monitoring/tunnel_quality_test.go | 37 +++++ go-gost/x/socket/websocket_reporter.go | 32 ++++- go-gost/x/socket/websocket_reporter_test.go | 19 +++ vite-frontend/src/config/tunnel-quality.ts | 39 ++++++ vite-frontend/src/pages/config.tsx | 76 ++++++++++- .../src/pages/node/tunnel-monitor-view.tsx | 68 ++++++---- 13 files changed, 574 insertions(+), 62 deletions(-) create mode 100644 go-backend/internal/monitoring/tunnel_quality.go create mode 100644 go-backend/internal/monitoring/tunnel_quality_test.go create mode 100644 vite-frontend/src/config/tunnel-quality.ts diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 7b5ccb2..2498acd 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -59,6 +59,7 @@ type diagnosisWorkItem struct { type diagnosisExecOptions struct { commandTimeout time.Duration pingTimeoutMS int + pingCount int timeoutMessage string } @@ -1596,10 +1597,14 @@ func (h *Handler) tcpPingViaNode(nodeID int64, ip string, port int, options diag if options.pingTimeoutMS <= 0 { options.pingTimeoutMS = int(diagnosisCommandTimeout / time.Millisecond) } + pingCount := options.pingCount + if pingCount <= 0 { + pingCount = 4 + } res, err := h.sendNodeCommandWithTimeout(nodeID, "TcpPing", map[string]interface{}{ "ip": ip, "port": port, - "count": 4, + "count": pingCount, "timeout": options.pingTimeoutMS, }, options.commandTimeout, false, false) if err != nil { @@ -1626,12 +1631,16 @@ func (h *Handler) tcpPingViaRemoteNode(node *nodeRecord, ip string, port int, op if options.pingTimeoutMS <= 0 { options.pingTimeoutMS = int(diagnosisCommandTimeout / time.Millisecond) } + pingCount := options.pingCount + if pingCount <= 0 { + pingCount = 4 + } fc := client.NewFederationClientWithTimeout(options.commandTimeout) return fc.Diagnose(remoteURL, remoteToken, h.federationLocalDomain(), client.RuntimeDiagnoseRequest{ IP: strings.TrimSpace(ip), Port: port, - Count: 4, + Count: pingCount, Timeout: options.pingTimeoutMS, Protocol: "tcp", }) diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index 32ac730..043589b 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -1015,6 +1015,7 @@ func (h *Handler) updateConfigs(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.Err(-2, err.Error())) return } + h.notifyTunnelQualityConfigChanged(key) } response.WriteJSON(w, response.OKEmpty()) @@ -1062,6 +1063,7 @@ func (h *Handler) updateSingleConfig(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.Err(-2, err.Error())) return } + h.notifyTunnelQualityConfigChanged(name) response.WriteJSON(w, response.OKEmpty()) } @@ -1110,11 +1112,23 @@ func normalizeAndValidateConfigValue(key, value string) (string, error) { } case monitoring.ConfigMonitorRetentionDays: return monitoring.NormalizeMonitoringRetentionDays(value) + case monitoring.ConfigTunnelQualityProbeIntervalSec: + return monitoring.NormalizeTunnelQualityProbeIntervalSeconds(value) default: return value, nil } } +func (h *Handler) notifyTunnelQualityConfigChanged(key string) { + if h == nil || h.qualityProber == nil { + return + } + switch strings.TrimSpace(key) { + case monitorTunnelQualityEnabledConfigKey, monitoring.ConfigTunnelQualityProbeIntervalSec: + h.qualityProber.NotifyConfigChanged() + } +} + func (h *Handler) isTunnelQualityMonitoringEnabled() bool { if h == nil || h.repo == nil { return true diff --git a/go-backend/internal/http/handler/tunnel_best_exit.go b/go-backend/internal/http/handler/tunnel_best_exit.go index 4c2b2bc..e4885e5 100644 --- a/go-backend/internal/http/handler/tunnel_best_exit.go +++ b/go-backend/internal/http/handler/tunnel_best_exit.go @@ -152,7 +152,11 @@ func evaluateBestExitOwner(owner chainNodeRecord, exits []chainNodeRecord, nodes ownerNode := nodes[owner.NodeID] for _, exit := range exits { exitNode := nodes[exit.NodeID] - if exitNode == nil { + if !isTunnelProbeNodeOnline(ownerNode) { + scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, "owner node offline")) + continue + } + if !isTunnelProbeNodeOnline(exitNode) { scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, "exit node unavailable")) continue } diff --git a/go-backend/internal/http/handler/tunnel_best_exit_test.go b/go-backend/internal/http/handler/tunnel_best_exit_test.go index ca0f677..df6d291 100644 --- a/go-backend/internal/http/handler/tunnel_best_exit_test.go +++ b/go-backend/internal/http/handler/tunnel_best_exit_test.go @@ -358,9 +358,9 @@ func TestEvaluateBestExitOwnerScoresAllCandidates(t *testing.T) { {NodeID: 31, NodeName: "exit-b", Port: 30031}, } nodes := map[int64]*nodeRecord{ - 10: {ID: 10, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"}, - 30: {ID: 30, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"}, - 31: {ID: 31, ServerIP: "10.0.0.31", ServerIPv4: "10.0.0.31", TCPListenAddr: "[::]"}, + 10: {ID: 10, Status: 1, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"}, + 30: {ID: 30, Status: 1, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"}, + 31: {ID: 31, Status: 1, ServerIP: "10.0.0.31", ServerIPv4: "10.0.0.31", TCPListenAddr: "[::]"}, } pinger := func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) { switch { @@ -387,12 +387,30 @@ func TestEvaluateBestExitOwnerScoresAllCandidates(t *testing.T) { } } +func TestEvaluateBestExitOwnerSkipsOfflineCandidate(t *testing.T) { + owner := chainNodeRecord{NodeID: 10, NodeName: "entry"} + exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-a", Port: 30030}} + nodes := map[int64]*nodeRecord{ + 10: {ID: 10, Status: 1, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10"}, + 30: {ID: 30, Status: 0, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30"}, + } + ping := func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) { + t.Fatalf("offline best-exit candidate should not be probed: node=%d target=%s:%d", nodeID, ip, port) + return 0, 100, nil + } + + scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, defaultTunnelProbeTarget(), ping) + if len(scores) != 1 || scores[0].Success { + t.Fatalf("expected one failed offline candidate, got %+v", scores) + } +} + func TestEvaluateBestExitOwnerUsesConfiguredPublicProbeTarget(t *testing.T) { owner := chainNodeRecord{NodeID: 10, NodeName: "entry-a"} exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-a", Port: 30001}} nodes := map[int64]*nodeRecord{ - 10: {ID: 10, Name: "entry-a", ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10"}, - 30: {ID: 30, Name: "exit-a", ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30"}, + 10: {ID: 10, Name: "entry-a", Status: 1, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10"}, + 30: {ID: 30, Name: "exit-a", Status: 1, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30"}, } target := tunnelProbeTarget{Host: "speed.example.com", Port: 8443} var calls []string @@ -419,8 +437,8 @@ func TestEvaluateBestExitOwnerMarksCandidateFailedWhenOwnerToExitFails(t *testin owner := chainNodeRecord{NodeID: 10, NodeName: "entry"} exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-a", Port: 30030}} nodes := map[int64]*nodeRecord{ - 10: {ID: 10, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"}, - 30: {ID: 30, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"}, + 10: {ID: 10, Status: 1, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"}, + 30: {ID: 30, Status: 1, ServerIP: "10.0.0.30", ServerIPv4: "10.0.0.30", TCPListenAddr: "[::]"}, } pinger := func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) { return 0, 100, errBestExitProbeForTest @@ -436,8 +454,8 @@ func TestEvaluateBestExitOwnerMarksCandidateFailedWhenTargetResolutionFails(t *t owner := chainNodeRecord{NodeID: 10, NodeName: "entry"} exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-v6", Port: 30030}} nodes := map[int64]*nodeRecord{ - 10: {ID: 10, Name: "entry", ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"}, - 30: {ID: 30, Name: "exit-v6", ServerIP: "2001:db8::30", ServerIPv6: "2001:db8::30", TCPListenAddr: "[::]"}, + 10: {ID: 10, Name: "entry", Status: 1, ServerIP: "10.0.0.10", ServerIPv4: "10.0.0.10", TCPListenAddr: "[::]"}, + 30: {ID: 30, Name: "exit-v6", Status: 1, ServerIP: "2001:db8::30", ServerIPv6: "2001:db8::30", TCPListenAddr: "[::]"}, } pinger := func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) { t.Fatalf("ping should not be called when target resolution fails: node=%d ip=%s port=%d", nodeID, ip, port) diff --git a/go-backend/internal/http/handler/tunnel_quality_prober.go b/go-backend/internal/http/handler/tunnel_quality_prober.go index 0e2188f..4c18082 100644 --- a/go-backend/internal/http/handler/tunnel_quality_prober.go +++ b/go-backend/internal/http/handler/tunnel_quality_prober.go @@ -3,6 +3,7 @@ package handler import ( "context" "encoding/json" + "errors" "log" "sync" "sync/atomic" @@ -13,7 +14,6 @@ import ( ) const ( - tunnelQualityProbeInterval = 1 * time.Second tunnelQualityProbeTimeout = 8 * time.Second tunnelQualityPingTimeoutMs = 5000 tunnelQualityPruneInterval = 10 * time.Minute @@ -56,7 +56,7 @@ type tunnelQualityProber struct { cache sync.Map // tunnelID (int64) → *tunnelQualitySnapshot ctx context.Context cancel context.CancelFunc - interval time.Duration + wake chan struct{} lastPrune int64 probing int32 // atomic flag: 1 = probeAll running, 0 = idle probeNode bestExitProbeFunc @@ -65,8 +65,8 @@ type tunnelQualityProber struct { // newTunnelQualityProber creates a new prober (not yet running). func newTunnelQualityProber(h *Handler) *tunnelQualityProber { return &tunnelQualityProber{ - handler: h, - interval: tunnelQualityProbeInterval, + handler: h, + wake: make(chan struct{}, 1), } } @@ -86,6 +86,16 @@ func (p *tunnelQualityProber) Stop() { p.cancel() } +func (p *tunnelQualityProber) NotifyConfigChanged() { + if p == nil || p.wake == nil { + return + } + select { + case p.wake <- struct{}{}: + default: + } +} + // GetAll returns all cached quality snapshots (latest per tunnel). func (p *tunnelQualityProber) GetAll() []tunnelQualitySnapshot { var items []tunnelQualitySnapshot @@ -109,20 +119,44 @@ func (p *tunnelQualityProber) loop() { // Run once immediately p.probeAll() - ticker := time.NewTicker(p.interval) - defer ticker.Stop() - for { + timer := time.NewTimer(p.probeInterval()) select { case <-p.ctx.Done(): + stopAndDrainTunnelQualityTimer(timer) return - case <-ticker.C: + case <-p.wake: + stopAndDrainTunnelQualityTimer(timer) + continue + case <-timer.C: p.probeAll() p.maybePrune() } } } +func stopAndDrainTunnelQualityTimer(timer *time.Timer) { + if timer == nil || timer.Stop() { + return + } + select { + case <-timer.C: + default: + } +} + +func (p *tunnelQualityProber) probeInterval() time.Duration { + if p == nil || p.handler == nil || p.handler.repo == nil { + return time.Duration(monitoring.DefaultTunnelQualityProbeIntervalSec) * time.Second + } + cfg, err := p.handler.repo.GetConfigsByNames([]string{monitoring.ConfigTunnelQualityProbeIntervalSec}) + if err != nil { + return time.Duration(monitoring.DefaultTunnelQualityProbeIntervalSec) * time.Second + } + seconds := monitoring.TunnelQualityProbeIntervalSecondsFromConfigMap(cfg) + return time.Duration(seconds) * time.Second +} + func (p *tunnelQualityProber) isEnabled() bool { if p == nil || p.handler == nil { return true @@ -246,15 +280,19 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) { options := diagnosisExecOptions{ commandTimeout: tunnelQualityProbeTimeout, pingTimeoutMS: tunnelQualityPingTimeoutMs, + pingCount: 1, timeoutMessage: "探测超时", } p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options, probeTarget) + entry, _, entryOnline := p.firstOnlineChainNode(inNodes) + exit, _, exitOnline := p.firstOnlineChainNode(outNodes) + switch tunnel.Type { case 1: // Port forwarding: entry → public probe target only. - if len(inNodes) > 0 { - lat, loss, err := p.pingNode(inNodes[0].NodeID, probeTarget.Host, probeTarget.Port, options) + if entryOnline { + lat, loss, err := p.pingNode(entry.NodeID, probeTarget.Host, probeTarget.Port, options) if err == nil { snap.ExitToBingLatency = lat snap.ExitToBingLoss = loss @@ -262,24 +300,42 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) { } else { snap.ErrorMessage = err.Error() } + } else { + snap.ErrorMessage = "入口节点均不在线" } case 2: // Tunnel forwarding: entry → exit + exit → Bing probeOK := true - if len(inNodes) > 0 && len(outNodes) > 0 { + if !entryOnline { + probeOK = false + snap.ErrorMessage = "入口节点均不在线" + snap.EntryToExitLatency = -1 + snap.EntryToExitLoss = 100 + } else if !exitOnline { + probeOK = false + snap.ErrorMessage = "出口节点均不在线" + snap.EntryToExitLatency = -1 + snap.EntryToExitLoss = 100 + } else { var hops []TunnelQualityHop var totalLat float64 remainingSuccessProb := 1.0 nodesInPath := make([]chainNodeRecord, 0, 2+len(midNodesGrouped)) - nodesInPath = append(nodesInPath, inNodes[0]) + nodesInPath = append(nodesInPath, entry) for _, midGroup := range midNodesGrouped { - if len(midGroup) > 0 { - nodesInPath = append(nodesInPath, midGroup[0]) + mid, _, online := p.firstOnlineChainNode(midGroup) + if !online { + probeOK = false + snap.ErrorMessage = "中间节点组均不在线" + break } + nodesInPath = append(nodesInPath, mid) + } + if probeOK { + nodesInPath = append(nodesInPath, exit) } - nodesInPath = append(nodesInPath, outNodes[0]) for i := 0; i < len(nodesInPath)-1; i++ { source := nodesInPath[i] @@ -293,7 +349,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) { } targetNode, nodeErr := h.getNodeRecord(target.NodeID) - if nodeErr != nil || targetNode == nil { + if nodeErr != nil || !isTunnelProbeNodeOnline(targetNode) { snap.ErrorMessage = "节点 " + target.NodeName + " 不可用" probeOK = false hop.Latency = -1 @@ -351,8 +407,8 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) { } // Exit → Bing - if len(outNodes) > 0 { - lat, loss, err := p.pingNode(outNodes[0].NodeID, probeTarget.Host, probeTarget.Port, options) + if exitOnline { + lat, loss, err := p.pingNode(exit.NodeID, probeTarget.Host, probeTarget.Port, options) if err == nil { snap.ExitToBingLatency = lat snap.ExitToBingLoss = loss @@ -367,8 +423,8 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) { snap.Success = probeOK default: // Unknown type: entry → public probe target. - if len(inNodes) > 0 { - lat, loss, err := p.pingNode(inNodes[0].NodeID, probeTarget.Host, probeTarget.Port, options) + if entryOnline { + lat, loss, err := p.pingNode(entry.NodeID, probeTarget.Host, probeTarget.Port, options) if err == nil { snap.ExitToBingLatency = lat snap.ExitToBingLoss = loss @@ -376,12 +432,31 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) { } else { snap.ErrorMessage = err.Error() } + } else { + snap.ErrorMessage = "入口节点均不在线" } } p.storeResult(snap) } +func isTunnelProbeNodeOnline(node *nodeRecord) bool { + return node != nil && (node.IsRemote == 1 || node.Status == 1) +} + +func (p *tunnelQualityProber) firstOnlineChainNode(nodes []chainNodeRecord) (chainNodeRecord, *nodeRecord, bool) { + if p == nil || p.handler == nil { + return chainNodeRecord{}, nil, false + } + for _, candidate := range nodes { + node, err := p.handler.getNodeRecord(candidate.NodeID) + if err == nil && isTunnelProbeNodeOnline(node) { + return candidate, node, true + } + } + return chainNodeRecord{}, nil, false +} + func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chainNodeRecord, chainHops [][]chainNodeRecord, outNodes []chainNodeRecord, ipPreference string, options diagnosisExecOptions, probeTarget tunnelProbeTarget) { if p == nil || p.handler == nil || p.handler.bestExit == nil || len(outNodes) <= 1 { return @@ -444,6 +519,9 @@ func (p *tunnelQualityProber) tcpPingNode(nodeID int64, ip string, port int, opt if nodeErr != nil { return 0, 100, nodeErr } + if !isTunnelProbeNodeOnline(node) { + return 0, 100, errors.New("节点不在线") + } var pingData map[string]interface{} var pingErr error diff --git a/go-backend/internal/http/handler/tunnel_quality_prober_test.go b/go-backend/internal/http/handler/tunnel_quality_prober_test.go index 0b93a29..b66c20b 100644 --- a/go-backend/internal/http/handler/tunnel_quality_prober_test.go +++ b/go-backend/internal/http/handler/tunnel_quality_prober_test.go @@ -26,6 +26,9 @@ func TestTunnelQualityProberUsesConfiguredProbeTarget(t *testing.T) { p := newTunnelQualityProber(h) var calls []string p.probeNode = func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) { + if options.pingCount != 1 { + t.Fatalf("expected real-time quality probe count 1, got %d", options.pingCount) + } calls = append(calls, fmt.Sprintf("%d|%s|%d", nodeID, ip, port)) return 10, 0, nil } @@ -46,6 +49,63 @@ func TestTunnelQualityProberUsesConfiguredProbeTarget(t *testing.T) { } } +func TestTunnelQualityProberSkipsAllOfflineExits(t *testing.T) { + h := setupProbeTargetTunnelHandler(t) + seedQualityForwardTunnel(t, h, 81, []int{0, 0, 0}) + + p := newTunnelQualityProber(h) + probeCalls := 0 + p.probeNode = func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) { + probeCalls++ + return 0, 100, fmt.Errorf("unexpected probe node=%d target=%s:%d", nodeID, ip, port) + } + p.probeTunnel(81) + + if probeCalls != 0 { + t.Fatalf("expected no TCP probes when all exits are offline, got %d", probeCalls) + } + snaps := p.GetAll() + if len(snaps) != 1 { + t.Fatalf("expected one quality snapshot, got %+v", snaps) + } + if snaps[0].Success || snaps[0].ErrorMessage != "出口节点均不在线" { + t.Fatalf("expected offline exit snapshot, got %+v", snaps[0]) + } + if snaps[0].EntryToExitLoss != 100 { + t.Fatalf("expected 100%% entry-to-exit loss, got %+v", snaps[0]) + } +} + +func TestTunnelQualityProberUsesOnlineBackupExit(t *testing.T) { + h := setupProbeTargetTunnelHandler(t) + seedQualityForwardTunnel(t, h, 82, []int{0, 1}) + + p := newTunnelQualityProber(h) + var calls []string + p.probeNode = func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) { + if options.pingCount != 1 { + t.Fatalf("expected real-time quality probe count 1, got %d", options.pingCount) + } + calls = append(calls, fmt.Sprintf("%d|%s|%d", nodeID, ip, port)) + return 10, 0, nil + } + p.probeTunnel(82) + + if slices.Contains(calls, "10|10.0.0.30|30030") { + t.Fatalf("did not expect probe to offline primary exit, calls=%+v", calls) + } + if !slices.Contains(calls, "10|10.0.0.31|30031") { + t.Fatalf("expected entry probe to online backup exit, calls=%+v", calls) + } + if !slices.Contains(calls, "31|www.bing.com|443") { + t.Fatalf("expected public probe from online backup exit, calls=%+v", calls) + } + snaps := p.GetAll() + if len(snaps) != 1 || !snaps[0].Success { + t.Fatalf("expected successful backup exit snapshot, got %+v", snaps) + } +} + func TestTunnelQualityProberStoresProbeTargetWhenChainIncomplete(t *testing.T) { h := setupProbeTargetTunnelHandler(t) seedProbeTargetTunnel(t, h, 78, "quality-target-incomplete", "speed.example.com", 8443) @@ -67,3 +127,69 @@ func TestTunnelQualityProberStoresProbeTargetWhenChainIncomplete(t *testing.T) { t.Fatalf("unexpected snapshot target metadata: %+v", snaps[0]) } } + +func TestTunnelQualityProberUsesConfiguredInterval(t *testing.T) { + h := setupProbeTargetTunnelHandler(t) + if err := h.repo.UpsertConfig("monitor_tunnel_quality_interval_sec", "15", time.Now().UnixMilli()); err != nil { + t.Fatalf("upsert interval config: %v", err) + } + + p := newTunnelQualityProber(h) + if got := p.probeInterval(); got != 15*time.Second { + t.Fatalf("probe interval = %s, want 15s", got) + } +} + +func TestTunnelQualityProberConfigNotificationIsCoalesced(t *testing.T) { + p := newTunnelQualityProber(nil) + p.NotifyConfigChanged() + p.NotifyConfigChanged() + + if got := len(p.wake); got != 1 { + t.Fatalf("wake notifications = %d, want 1", got) + } +} + +func TestNormalizeTunnelQualityProbeIntervalConfigValue(t *testing.T) { + got, err := normalizeAndValidateConfigValue("monitor_tunnel_quality_interval_sec", " 15 ") + if err != nil || got != "15" { + t.Fatalf("normalize interval = %q, %v", got, err) + } + if _, err := normalizeAndValidateConfigValue("monitor_tunnel_quality_interval_sec", "0"); err == nil { + t.Fatalf("expected invalid interval to be rejected") + } +} + +func seedQualityForwardTunnel(t *testing.T, h *Handler, tunnelID int64, exitStatuses []int) { + t.Helper() + now := time.Now().UnixMilli() + if err := h.repo.DB().Exec(` + INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, inx, ip_preference, probe_target_host, probe_target_port) + VALUES(?, ?, 1, 2, 'tls', 1, ?, ?, 1, ?, '', '', 0) + `, tunnelID, fmt.Sprintf("quality-forward-%d", tunnelID), now, now, tunnelID).Error; err != nil { + t.Fatalf("insert forwarding tunnel: %v", err) + } + if err := h.repo.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(?, '1', 10, 30001, 'fifo', 1, 'tls') + `, tunnelID).Error; err != nil { + t.Fatalf("insert entry chain: %v", err) + } + for i, status := range exitStatuses { + nodeID := int64(30 + i) + port := 30030 + i + ip := fmt.Sprintf("10.0.0.%d", nodeID) + if err := h.repo.DB().Exec(` + INSERT INTO node(id, name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx) + VALUES(?, ?, ?, ?, ?, '', '30000-30100', '', 'v1', 1, 1, 1, ?, ?, ?, '[::]', '[::]', 0) + `, nodeID, fmt.Sprintf("exit-%d", i+1), fmt.Sprintf("exit-secret-%d", i+1), ip, ip, now, now, status).Error; err != nil { + t.Fatalf("insert exit node %d: %v", nodeID, err) + } + if err := h.repo.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(?, '3', ?, ?, 'fifo', ?, 'tls') + `, tunnelID, nodeID, port, i+1).Error; err != nil { + t.Fatalf("insert exit chain %d: %v", nodeID, err) + } + } +} diff --git a/go-backend/internal/monitoring/tunnel_quality.go b/go-backend/internal/monitoring/tunnel_quality.go new file mode 100644 index 0000000..4ab9528 --- /dev/null +++ b/go-backend/internal/monitoring/tunnel_quality.go @@ -0,0 +1,52 @@ +package monitoring + +import ( + "fmt" + "strconv" + "strings" +) + +const ( + ConfigTunnelQualityProbeIntervalSec = "monitor_tunnel_quality_interval_sec" + DefaultTunnelQualityProbeIntervalSec = 1 + MinTunnelQualityProbeIntervalSec = 1 + MaxTunnelQualityProbeIntervalSec = 3600 +) + +func TunnelQualityProbeIntervalSecondsFromConfigMap(cfg map[string]string) int { + if cfg == nil { + return DefaultTunnelQualityProbeIntervalSec + } + seconds, err := parseTunnelQualityProbeIntervalSeconds(cfg[ConfigTunnelQualityProbeIntervalSec]) + if err != nil { + return DefaultTunnelQualityProbeIntervalSec + } + return seconds +} + +func NormalizeTunnelQualityProbeIntervalSeconds(value string) (string, error) { + seconds, err := parseTunnelQualityProbeIntervalSeconds(value) + if err != nil { + return "", err + } + return strconv.Itoa(seconds), nil +} + +func parseTunnelQualityProbeIntervalSeconds(value string) (int, error) { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + return 0, fmt.Errorf("隧道质量探测间隔不能为空") + } + seconds, err := strconv.Atoi(trimmed) + if err != nil { + return 0, fmt.Errorf("隧道质量探测间隔必须是整数") + } + if seconds < MinTunnelQualityProbeIntervalSec || seconds > MaxTunnelQualityProbeIntervalSec { + return 0, fmt.Errorf( + "隧道质量探测间隔必须在 %d 到 %d 秒之间", + MinTunnelQualityProbeIntervalSec, + MaxTunnelQualityProbeIntervalSec, + ) + } + return seconds, nil +} diff --git a/go-backend/internal/monitoring/tunnel_quality_test.go b/go-backend/internal/monitoring/tunnel_quality_test.go new file mode 100644 index 0000000..0ca94b9 --- /dev/null +++ b/go-backend/internal/monitoring/tunnel_quality_test.go @@ -0,0 +1,37 @@ +package monitoring + +import "testing" + +func TestTunnelQualityProbeIntervalSecondsFromConfigMap(t *testing.T) { + tests := []struct { + name string + cfg map[string]string + want int + }{ + {name: "missing config", cfg: nil, want: DefaultTunnelQualityProbeIntervalSec}, + {name: "configured", cfg: map[string]string{ConfigTunnelQualityProbeIntervalSec: "15"}, want: 15}, + {name: "invalid", cfg: map[string]string{ConfigTunnelQualityProbeIntervalSec: "0"}, want: DefaultTunnelQualityProbeIntervalSec}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := TunnelQualityProbeIntervalSecondsFromConfigMap(tt.cfg); got != tt.want { + t.Fatalf("interval = %d, want %d", got, tt.want) + } + }) + } +} + +func TestNormalizeTunnelQualityProbeIntervalSeconds(t *testing.T) { + for _, value := range []string{"1", "15", "3600"} { + if got, err := NormalizeTunnelQualityProbeIntervalSeconds(value); err != nil || got != value { + t.Fatalf("normalize %q = %q, %v", value, got, err) + } + } + + for _, value := range []string{"", "0", "3601", "1.5", "abc"} { + if got, err := NormalizeTunnelQualityProbeIntervalSeconds(value); err == nil { + t.Fatalf("normalize %q unexpectedly succeeded with %q", value, got) + } + } +} diff --git a/go-gost/x/socket/websocket_reporter.go b/go-gost/x/socket/websocket_reporter.go index 4d4d5f1..0e5fa18 100644 --- a/go-gost/x/socket/websocket_reporter.go +++ b/go-gost/x/socket/websocket_reporter.go @@ -151,6 +151,7 @@ const ( initialBackoff = 2 * time.Second // 重连初始退避 maxBackoff = 2 * time.Minute // 重连最大退避 defaultMetricReportInterval = 5 * time.Second + maxConcurrentTCPPings = 8 ) type WebSocketReporter struct { @@ -172,6 +173,7 @@ type WebSocketReporter struct { connecting bool // 正在连接状态 connMutex sync.Mutex // 连接状态锁 aesCrypto *crypto.AESCrypto // AES加密器 + tcpPingSem chan struct{} // 限制诊断探测并发，避免离线目标耗尽连接 } var wsDial = func(dialer *websocket.Dialer, rawURL string) (*websocket.Conn, *http.Response, error) { @@ -201,6 +203,29 @@ func NewWebSocketReporter(serverURL string, secret string) *WebSocketReporter { connected: false, connecting: false, aesCrypto: aesCrypto, + tcpPingSem: make(chan struct{}, maxConcurrentTCPPings), + } +} + +func (w *WebSocketReporter) tryAcquireTCPPingSlot() bool { + if w == nil || w.tcpPingSem == nil { + return false + } + select { + case w.tcpPingSem <- struct{}{}: + return true + default: + return false + } +} + +func (w *WebSocketReporter) releaseTCPPingSlot() { + if w == nil || w.tcpPingSem == nil { + return + } + select { + case <-w.tcpPingSem: + default: } } @@ -840,9 +865,14 @@ func (w *WebSocketReporter) routeCommand(cmd CommandMessage) { // TCP Ping 诊断命令（只读，不需要保存配置） case "TcpPing": + response.Type = "TcpPingResponse" + if !w.tryAcquireTCPPingSlot() { + err = fmt.Errorf("TCP探测任务过多，请稍后重试") + break + } + defer w.releaseTCPPingSlot() var tcpPingResult TcpPingResponse tcpPingResult, err = w.handleTcpPing(cmd.Data) - response.Type = "TcpPingResponse" response.Data = tcpPingResult // needSaveConfig = false (默认值) diff --git a/go-gost/x/socket/websocket_reporter_test.go b/go-gost/x/socket/websocket_reporter_test.go index 2b43d5b..3e6c204 100644 --- a/go-gost/x/socket/websocket_reporter_test.go +++ b/go-gost/x/socket/websocket_reporter_test.go @@ -148,6 +148,25 @@ func TestNewWebSocketReporterUsesReducedMetricInterval(t *testing.T) { } } +func TestWebSocketReporterLimitsConcurrentTCPPings(t *testing.T) { + reporter := &WebSocketReporter{tcpPingSem: make(chan struct{}, maxConcurrentTCPPings)} + for i := 0; i < maxConcurrentTCPPings; i++ { + if !reporter.tryAcquireTCPPingSlot() { + t.Fatalf("expected TCP ping slot %d to be available", i) + } + } + if reporter.tryAcquireTCPPingSlot() { + t.Fatalf("expected TCP ping concurrency limit at %d", maxConcurrentTCPPings) + } + for i := 0; i < maxConcurrentTCPPings; i++ { + reporter.releaseTCPPingSlot() + } + if !reporter.tryAcquireTCPPingSlot() { + t.Fatalf("expected released TCP ping slot to be reusable") + } + reporter.releaseTCPPingSlot() +} + func TestFormatWebSocketDialErrorIncludesHTTPStatus(t *testing.T) { err := errors.New("websocket: bad handshake") resp := &http.Response{ diff --git a/vite-frontend/src/config/tunnel-quality.ts b/vite-frontend/src/config/tunnel-quality.ts new file mode 100644 index 0000000..907ee2b --- /dev/null +++ b/vite-frontend/src/config/tunnel-quality.ts @@ -0,0 +1,39 @@ +export const TUNNEL_QUALITY_INTERVAL_CONFIG_KEY = + "monitor_tunnel_quality_interval_sec"; +export const DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC = 1; +export const MIN_TUNNEL_QUALITY_INTERVAL_SEC = 1; +export const MAX_TUNNEL_QUALITY_INTERVAL_SEC = 3600; + +export const parseTunnelQualityIntervalSeconds = (value: unknown): number => { + const seconds = Number(value); + + return Number.isInteger(seconds) && + seconds >= MIN_TUNNEL_QUALITY_INTERVAL_SEC && + seconds <= MAX_TUNNEL_QUALITY_INTERVAL_SEC + ? seconds + : DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC; +}; + +export const validateTunnelQualityInterval = (value: string): string | null => { + const normalized = value.trim(); + + if (!normalized) { + return "请输入探测间隔"; + } + const seconds = Number(normalized); + + if (!Number.isInteger(seconds)) { + return "探测间隔必须是整数"; + } + if ( + seconds < MIN_TUNNEL_QUALITY_INTERVAL_SEC || + seconds > MAX_TUNNEL_QUALITY_INTERVAL_SEC + ) { + return `探测间隔必须在 ${MIN_TUNNEL_QUALITY_INTERVAL_SEC} 到 ${MAX_TUNNEL_QUALITY_INTERVAL_SEC} 秒之间`; + } + + return null; +}; + +export const tunnelQualityIntervalLabel = (seconds: number): string => + seconds === 1 ? "每秒" : `每 ${seconds} 秒`; diff --git a/vite-frontend/src/pages/config.tsx b/vite-frontend/src/pages/config.tsx index 9c36276..1360fab 100644 --- a/vite-frontend/src/pages/config.tsx +++ b/vite-frontend/src/pages/config.tsx @@ -43,6 +43,14 @@ import { BackIcon, SettingsIcon } from "@/components/icons"; import { ThemeSettings } from "@/components/theme-settings"; import { isAdmin } from "@/utils/auth"; import { getCachedConfigs, configCache, updateSiteConfig } from "@/config/site"; +import { + DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC, + MAX_TUNNEL_QUALITY_INTERVAL_SEC, + MIN_TUNNEL_QUALITY_INTERVAL_SEC, + parseTunnelQualityIntervalSeconds, + TUNNEL_QUALITY_INTERVAL_CONFIG_KEY, + validateTunnelQualityInterval, +} from "@/config/tunnel-quality"; import { type UpdateReleaseChannel, getUpdateReleaseChannel, @@ -157,6 +165,16 @@ const CONFIG_ITEMS: ConfigItem[] = [ "关闭后，前端停止自动刷新，后端停止实时隧道质量探测（全局配置）", type: "switch", }, + { + key: TUNNEL_QUALITY_INTERVAL_CONFIG_KEY, + label: "隧道质量探测间隔", + placeholder: String(DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC), + description: + "设置实时隧道质量检测的执行频率，单位为秒；允许 1–3600 秒，默认 1 秒。", + type: "input", + dependsOn: "monitor_tunnel_quality_enabled", + dependsValue: "true", + }, { key: "monitor_retention_days", label: "监控数据保留天数", @@ -239,6 +257,7 @@ const getInitialConfigs = (): Record => { "cloudflare_secret_key", "forward_compact_mode", "monitor_tunnel_quality_enabled", + TUNNEL_QUALITY_INTERVAL_CONFIG_KEY, "monitor_retention_days", "ip", "panel_domain", @@ -622,6 +641,19 @@ export default function ConfigPage() { // 保存配置 const handleSave = async () => { + const intervalValue = configs[TUNNEL_QUALITY_INTERVAL_CONFIG_KEY]; + const intervalChanged = + intervalValue !== originalConfigs[TUNNEL_QUALITY_INTERVAL_CONFIG_KEY]; + const intervalError = intervalChanged + ? validateTunnelQualityInterval(intervalValue || "") + : null; + + if (intervalError) { + toast.error(intervalError); + + return; + } + setSaving(true); try { const changedKeys = Object.keys(configs).filter( @@ -667,12 +699,22 @@ export default function ConfigPage() { }), ); - // 如果隧道质量检测开关变更，通知 tunnel-monitor-view - if (changedKeys.includes("monitor_tunnel_quality_enabled")) { + // 如果隧道质量检测配置变更，通知 tunnel-monitor-view + if ( + changedKeys.some((key) => + [ + "monitor_tunnel_quality_enabled", + TUNNEL_QUALITY_INTERVAL_CONFIG_KEY, + ].includes(key), + ) + ) { window.dispatchEvent( new CustomEvent("monitorTunnelQualityEnabledChanged", { detail: { enabled: configs["monitor_tunnel_quality_enabled"] === "true", + intervalSec: parseTunnelQualityIntervalSeconds( + configs[TUNNEL_QUALITY_INTERVAL_CONFIG_KEY], + ), }, }), ); @@ -1075,11 +1117,21 @@ export default function ConfigPage() { case "bg_image": return renderBgImageUploader(); - case "input": + case "input": { if (isBrandPreviewKey(item.key)) { return renderBrandAssetUploader(item.key, isChanged); } + const isTunnelQualityInterval = + item.key === TUNNEL_QUALITY_INTERVAL_CONFIG_KEY; + const intervalValue = isTunnelQualityInterval + ? (configs[item.key] ?? String(DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC)) + : (configs[item.key] ?? ""); + const intervalError = + isTunnelQualityInterval && configs[item.key] !== undefined + ? validateTunnelQualityInterval(intervalValue) + : null; + return ( handleConfigChange(item.key, e.target.value)} /> ); + } case "switch": return ( diff --git a/vite-frontend/src/pages/node/tunnel-monitor-view.tsx b/vite-frontend/src/pages/node/tunnel-monitor-view.tsx index b51aeeb..f54d2ce 100644 --- a/vite-frontend/src/pages/node/tunnel-monitor-view.tsx +++ b/vite-frontend/src/pages/node/tunnel-monitor-view.tsx @@ -53,12 +53,17 @@ import { TableRow, TableCell, } from "@/shadcn-bridge/heroui/table"; +import { + DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC, + parseTunnelQualityIntervalSeconds, + TUNNEL_QUALITY_INTERVAL_CONFIG_KEY, + tunnelQualityIntervalLabel, +} from "@/config/tunnel-quality"; interface TunnelMonitorViewProps { viewMode?: "list" | "grid"; } -const QUALITY_POLL_INTERVAL = 1_000; // 1 second const MONITOR_TUNNEL_QUALITY_ENABLED_CONFIG_KEY = "monitor_tunnel_quality_enabled"; const MONITOR_TUNNEL_QUALITY_ENABLED_EVENT = @@ -562,6 +567,9 @@ export function TunnelMonitorView({ const qualityTimerRef = useRef(null); const [monitorTunnelQualityEnabled, setMonitorTunnelQualityEnabled] = useState(true); + const [tunnelQualityIntervalSec, setTunnelQualityIntervalSec] = useState( + DEFAULT_TUNNEL_QUALITY_INTERVAL_SEC, + ); // Detail view state const [detailTunnelId, setDetailTunnelId] = useState(null); @@ -618,26 +626,28 @@ export function TunnelMonitorView({ } }, []); - const loadMonitorTunnelQualityEnabled = useCallback(async () => { - try { - const response = await getConfigByName( - MONITOR_TUNNEL_QUALITY_ENABLED_CONFIG_KEY, - ); + const loadTunnelQualityConfig = useCallback(async () => { + const [enabledResponse, intervalResponse] = await Promise.all([ + getConfigByName(MONITOR_TUNNEL_QUALITY_ENABLED_CONFIG_KEY).catch( + () => null, + ), + getConfigByName(TUNNEL_QUALITY_INTERVAL_CONFIG_KEY).catch(() => null), + ]); - setMonitorTunnelQualityEnabled( - typeof response.data?.value === "string" - ? response.data.value === "true" - : true, - ); - } catch { - setMonitorTunnelQualityEnabled(true); - } + setMonitorTunnelQualityEnabled( + typeof enabledResponse?.data?.value === "string" + ? enabledResponse.data.value === "true" + : true, + ); + setTunnelQualityIntervalSec( + parseTunnelQualityIntervalSeconds(intervalResponse?.data?.value), + ); }, []); useEffect(() => { void loadTunnels(); - void loadMonitorTunnelQualityEnabled(); - }, [loadMonitorTunnelQualityEnabled, loadTunnels]); + void loadTunnelQualityConfig(); + }, [loadTunnelQualityConfig, loadTunnels]); useEffect(() => { const timer = window.setInterval(() => { @@ -649,8 +659,11 @@ export function TunnelMonitorView({ useEffect(() => { const handleMonitorTunnelQualityEnabledChanged = (event: Event) => { - const enabled = (event as CustomEvent<{ enabled?: boolean }>).detail - ?.enabled; + const detail = ( + event as CustomEvent<{ enabled?: boolean; intervalSec?: number }> + ).detail; + const enabled = detail?.enabled; + const intervalSec = detail?.intervalSec; if (typeof enabled === "boolean") { setMonitorTunnelQualityEnabled(enabled); @@ -658,7 +671,12 @@ export function TunnelMonitorView({ setQualityLoading(false); } } else { - void loadMonitorTunnelQualityEnabled(); + void loadTunnelQualityConfig(); + } + if (typeof intervalSec === "number") { + setTunnelQualityIntervalSec( + parseTunnelQualityIntervalSeconds(String(intervalSec)), + ); } }; @@ -673,7 +691,7 @@ export function TunnelMonitorView({ handleMonitorTunnelQualityEnabledChanged as EventListener, ); }; - }, [loadMonitorTunnelQualityEnabled]); + }, [loadTunnelQualityConfig]); useEffect(() => { if (tunnels.length > 0 && !initialHistoryFetched.current) { @@ -726,7 +744,7 @@ export function TunnelMonitorView({ } }, [tunnels]); - // --- Load quality snapshots (auto-polling every 10s) --- + // --- Load quality snapshots using the configured probe interval --- const loadQuality = useCallback(async (options?: { silent?: boolean }) => { const silent = options?.silent ?? false; @@ -788,7 +806,7 @@ export function TunnelMonitorView({ qualityTimerRef.current = window.setInterval(() => { void loadQuality({ silent: true }); - }, QUALITY_POLL_INTERVAL); + }, tunnelQualityIntervalSec * 1000); return () => { if (qualityTimerRef.current) { @@ -796,7 +814,7 @@ export function TunnelMonitorView({ qualityTimerRef.current = null; } }; - }, [loadQuality, monitorTunnelQualityEnabled]); + }, [loadQuality, monitorTunnelQualityEnabled, tunnelQualityIntervalSec]); // --- Load quality history for detail chart --- const loadQualityHistory = useCallback( @@ -1065,7 +1083,7 @@ export function TunnelMonitorView({ {monitorTunnelQualityEnabled ? ( <> - 自动探测中（每秒测试，30秒上报） + {`自动探测中（${tunnelQualityIntervalLabel(tunnelQualityIntervalSec)}测试）`} ) : ( <> @@ -1129,7 +1147,7 @@ export function TunnelMonitorView({ {monitorTunnelQualityEnabled ? ( <> - 每秒探测 · 更新于 {lastQualityUpdate} + {`${tunnelQualityIntervalLabel(tunnelQualityIntervalSec)}探测 · 更新于 ${lastQualityUpdate}`} ) : ( <>