mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-09-28 07:36:38 +08:00
feat: use probe target for tunnel quality checks
This commit is contained in:
@@ -234,8 +234,22 @@ func (h *Handler) monitorTunnelQualityHandler(w http.ResponseWriter, r *http.Req
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
targetsByTunnelID := map[int64]tunnelProbeTarget{}
|
||||||
|
if tunnels, listErr := h.repo.ListTunnels(); listErr == nil {
|
||||||
|
for _, item := range tunnels {
|
||||||
|
id := asInt64(item["id"], 0)
|
||||||
|
if id > 0 {
|
||||||
|
targetsByTunnelID[id] = effectiveTunnelProbeTargetValues(asString(item["probeTargetHost"]), asInt(item["probeTargetPort"], 0))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
snapshots := make([]tunnelQualitySnapshot, 0, len(qualities))
|
snapshots := make([]tunnelQualitySnapshot, 0, len(qualities))
|
||||||
for _, q := range qualities {
|
for _, q := range qualities {
|
||||||
|
target := targetsByTunnelID[q.TunnelID]
|
||||||
|
if target.Host == "" {
|
||||||
|
target = defaultTunnelProbeTarget()
|
||||||
|
}
|
||||||
snapshots = append(snapshots, tunnelQualitySnapshot{
|
snapshots = append(snapshots, tunnelQualitySnapshot{
|
||||||
TunnelID: q.TunnelID,
|
TunnelID: q.TunnelID,
|
||||||
EntryToExitLatency: q.EntryToExitLatency,
|
EntryToExitLatency: q.EntryToExitLatency,
|
||||||
@@ -246,6 +260,8 @@ func (h *Handler) monitorTunnelQualityHandler(w http.ResponseWriter, r *http.Req
|
|||||||
ErrorMessage: q.ErrorMessage,
|
ErrorMessage: q.ErrorMessage,
|
||||||
Timestamp: q.Timestamp,
|
Timestamp: q.Timestamp,
|
||||||
ChainDetails: q.ChainDetails,
|
ChainDetails: q.ChainDetails,
|
||||||
|
ProbeTargetHost: target.Host,
|
||||||
|
ProbeTargetPort: target.Port,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
response.WriteJSON(w, response.OK(snapshots))
|
response.WriteJSON(w, response.OK(snapshots))
|
||||||
|
|||||||
@@ -42,6 +42,8 @@ type tunnelQualitySnapshot struct {
|
|||||||
ErrorMessage string `json:"errorMessage,omitempty"`
|
ErrorMessage string `json:"errorMessage,omitempty"`
|
||||||
Timestamp int64 `json:"timestamp"`
|
Timestamp int64 `json:"timestamp"`
|
||||||
ChainDetails string `json:"chainDetails,omitempty"`
|
ChainDetails string `json:"chainDetails,omitempty"`
|
||||||
|
ProbeTargetHost string `json:"probeTargetHost,omitempty"`
|
||||||
|
ProbeTargetPort int `json:"probeTargetPort,omitempty"`
|
||||||
|
|
||||||
// internal fields for db reporting
|
// internal fields for db reporting
|
||||||
lastDBWrite int64 `json:"-"`
|
lastDBWrite int64 `json:"-"`
|
||||||
@@ -57,6 +59,7 @@ type tunnelQualityProber struct {
|
|||||||
interval time.Duration
|
interval time.Duration
|
||||||
lastPrune int64
|
lastPrune int64
|
||||||
probing int32 // atomic flag: 1 = probeAll running, 0 = idle
|
probing int32 // atomic flag: 1 = probeAll running, 0 = idle
|
||||||
|
probeNode bestExitProbeFunc
|
||||||
}
|
}
|
||||||
|
|
||||||
// newTunnelQualityProber creates a new prober (not yet running).
|
// newTunnelQualityProber creates a new prober (not yet running).
|
||||||
@@ -243,13 +246,19 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
|
|||||||
timeoutMessage: "探测超时",
|
timeoutMessage: "探测超时",
|
||||||
}
|
}
|
||||||
probeTarget := effectiveTunnelProbeTargetValues(tunnel.ProbeTargetHost, tunnel.ProbeTargetPort)
|
probeTarget := effectiveTunnelProbeTargetValues(tunnel.ProbeTargetHost, tunnel.ProbeTargetPort)
|
||||||
|
snap.ProbeTargetHost = probeTarget.Host
|
||||||
|
snap.ProbeTargetPort = probeTarget.Port
|
||||||
p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options, probeTarget)
|
p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options, probeTarget)
|
||||||
|
|
||||||
switch tunnel.Type {
|
switch tunnel.Type {
|
||||||
case 1:
|
case 1:
|
||||||
// Port forwarding: entry → Bing only
|
// Port forwarding: exit/public edge → public probe target when available.
|
||||||
if len(inNodes) > 0 {
|
publicProbeNodes := inNodes
|
||||||
lat, loss, err := p.tcpPingNode(inNodes[0].NodeID, "www.bing.com", 443, options)
|
if len(outNodes) > 0 {
|
||||||
|
publicProbeNodes = outNodes
|
||||||
|
}
|
||||||
|
if len(publicProbeNodes) > 0 {
|
||||||
|
lat, loss, err := p.pingNode(publicProbeNodes[0].NodeID, probeTarget.Host, probeTarget.Port, options)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
snap.ExitToBingLatency = lat
|
snap.ExitToBingLatency = lat
|
||||||
snap.ExitToBingLoss = loss
|
snap.ExitToBingLoss = loss
|
||||||
@@ -311,7 +320,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
|
|||||||
hop.TargetIP = targetIP
|
hop.TargetIP = targetIP
|
||||||
hop.TargetPort = targetPort
|
hop.TargetPort = targetPort
|
||||||
|
|
||||||
lat, loss, err := p.tcpPingNode(source.NodeID, targetIP, targetPort, options)
|
lat, loss, err := p.pingNode(source.NodeID, targetIP, targetPort, options)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
hop.Latency = lat
|
hop.Latency = lat
|
||||||
hop.Loss = loss
|
hop.Loss = loss
|
||||||
@@ -347,7 +356,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
|
|||||||
|
|
||||||
// Exit → Bing
|
// Exit → Bing
|
||||||
if len(outNodes) > 0 {
|
if len(outNodes) > 0 {
|
||||||
lat, loss, err := p.tcpPingNode(outNodes[0].NodeID, "www.bing.com", 443, options)
|
lat, loss, err := p.pingNode(outNodes[0].NodeID, probeTarget.Host, probeTarget.Port, options)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
snap.ExitToBingLatency = lat
|
snap.ExitToBingLatency = lat
|
||||||
snap.ExitToBingLoss = loss
|
snap.ExitToBingLoss = loss
|
||||||
@@ -361,9 +370,13 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
|
|||||||
|
|
||||||
snap.Success = probeOK
|
snap.Success = probeOK
|
||||||
default:
|
default:
|
||||||
// Unknown type: entry → Bing
|
// Unknown type: use exit/public edge when available, otherwise entry.
|
||||||
if len(inNodes) > 0 {
|
publicProbeNodes := inNodes
|
||||||
lat, loss, err := p.tcpPingNode(inNodes[0].NodeID, "www.bing.com", 443, options)
|
if len(outNodes) > 0 {
|
||||||
|
publicProbeNodes = outNodes
|
||||||
|
}
|
||||||
|
if len(publicProbeNodes) > 0 {
|
||||||
|
lat, loss, err := p.pingNode(publicProbeNodes[0].NodeID, probeTarget.Host, probeTarget.Port, options)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
snap.ExitToBingLatency = lat
|
snap.ExitToBingLatency = lat
|
||||||
snap.ExitToBingLoss = loss
|
snap.ExitToBingLoss = loss
|
||||||
@@ -401,7 +414,7 @@ func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chai
|
|||||||
}
|
}
|
||||||
// This best-exit decision cache is per decision round; the display-oriented
|
// This best-exit decision cache is per decision round; the display-oriented
|
||||||
// tunnel quality snapshot may still collect its own first-exit public probe.
|
// tunnel quality snapshot may still collect its own first-exit public probe.
|
||||||
roundPinger := newBestExitRoundPinger(p.tcpPingNode)
|
roundPinger := newBestExitRoundPinger(p.pingNode)
|
||||||
for _, owner := range owners {
|
for _, owner := range owners {
|
||||||
if nodeMap[owner.NodeID] == nil {
|
if nodeMap[owner.NodeID] == nil {
|
||||||
continue
|
continue
|
||||||
@@ -422,6 +435,13 @@ func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chai
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (p *tunnelQualityProber) pingNode(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) {
|
||||||
|
if p != nil && p.probeNode != nil {
|
||||||
|
return p.probeNode(nodeID, ip, port, options)
|
||||||
|
}
|
||||||
|
return p.tcpPingNode(nodeID, ip, port, options)
|
||||||
|
}
|
||||||
|
|
||||||
func (p *tunnelQualityProber) tcpPingNode(nodeID int64, ip string, port int, options diagnosisExecOptions) (latency float64, loss float64, err error) {
|
func (p *tunnelQualityProber) tcpPingNode(nodeID int64, ip string, port int, options diagnosisExecOptions) (latency float64, loss float64, err error) {
|
||||||
h := p.handler
|
h := p.handler
|
||||||
if h == nil {
|
if h == nil {
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
package handler
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"slices"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestTunnelQualityProberUsesConfiguredProbeTarget(t *testing.T) {
|
||||||
|
h := setupProbeTargetTunnelHandler(t)
|
||||||
|
seedProbeTargetTunnel(t, h, 77, "quality-target", "speed.example.com", 8443)
|
||||||
|
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(30, 'exit-a', 'exit-secret', '10.0.0.30', '10.0.0.30', '', '30000-30010', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 0)
|
||||||
|
`, time.Now().UnixMilli(), time.Now().UnixMilli()).Error; err != nil {
|
||||||
|
t.Fatalf("insert exit node: %v", err)
|
||||||
|
}
|
||||||
|
if err := h.repo.DB().Exec(`
|
||||||
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
||||||
|
VALUES(77, '3', 30, 30001, 'round', 1, 'tls')
|
||||||
|
`).Error; err != nil {
|
||||||
|
t.Fatalf("insert exit chain: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
p := newTunnelQualityProber(h)
|
||||||
|
var calls []string
|
||||||
|
p.probeNode = func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) {
|
||||||
|
calls = append(calls, fmt.Sprintf("%d|%s|%d", nodeID, ip, port))
|
||||||
|
return 10, 0, nil
|
||||||
|
}
|
||||||
|
p.probeTunnel(77)
|
||||||
|
|
||||||
|
if !slices.Contains(calls, "30|speed.example.com|8443") {
|
||||||
|
t.Fatalf("expected exit probe to configured target, calls=%+v", calls)
|
||||||
|
}
|
||||||
|
snaps := p.GetAll()
|
||||||
|
if len(snaps) != 1 {
|
||||||
|
t.Fatalf("expected one quality snapshot, got %+v", snaps)
|
||||||
|
}
|
||||||
|
if snaps[0].ProbeTargetHost != "speed.example.com" || snaps[0].ProbeTargetPort != 8443 {
|
||||||
|
t.Fatalf("unexpected snapshot target metadata: %+v", snaps[0])
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user