tcjlj/flvx
1
0
Fork 0
mirror of https://github.com/Sagit-chu/flvx.git synced 2026-09-28 23:56:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: use probe target for best exit scoring

Browse Source
This commit is contained in:
sagitchu
2026-05-01 23:20:50 +08:00
parent 58ef260755
commit cc0b8de2e1
3 changed files with 72 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go-backend/internal/http/handler/tunnel_best_exit.go
+15 -11
View File
@@ -57,6 +57,12 @@ type bestExitProbeResult struct {
err error
}
type bestExitProbeCacheKey struct {
NodeID int64
Host string
Port int
}
type bestExitDecision struct {
AppliedExitNodeID int64
PendingExitNodeID int64
@@ -138,7 +144,7 @@ func sortBestExitScores(scores []bestExitCandidateScore) {
})
}
func evaluateBestExitOwner(owner chainNodeRecord, exits []chainNodeRecord, nodes map[int64]*nodeRecord, ipPreference string, options diagnosisExecOptions, ping bestExitProbeFunc) []bestExitCandidateScore {
func evaluateBestExitOwner(owner chainNodeRecord, exits []chainNodeRecord, nodes map[int64]*nodeRecord, ipPreference string, options diagnosisExecOptions, target tunnelProbeTarget, ping bestExitProbeFunc) []bestExitCandidateScore {
scores := make([]bestExitCandidateScore, 0, len(exits))
if owner.NodeID <= 0 || len(exits) == 0 || ping == nil {
return scores
@@ -160,7 +166,7 @@ func evaluateBestExitOwner(owner chainNodeRecord, exits []chainNodeRecord, nodes
scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, ownerErr.Error()))
continue
}
publicLatency, publicLoss, publicErr := ping(exit.NodeID, bestExitPublicTargetHost, bestExitPublicTargetPort, options)
publicLatency, publicLoss, publicErr := ping(exit.NodeID, target.Host, target.Port, options)
if publicErr != nil {
scores = append(scores, failedBestExitCandidate(owner.NodeID, exit, publicErr.Error()))
continue
@@ -193,17 +199,15 @@ func resolveBestExitProbeTarget(fromNode, targetNode *nodeRecord, preferredPort
}
func newBestExitRoundPinger(base bestExitProbeFunc) bestExitProbeFunc {
cache := make(map[int64]bestExitProbeResult)
cache := make(map[bestExitProbeCacheKey]bestExitProbeResult)
return func(nodeID int64, ip string, port int, options diagnosisExecOptions) (float64, float64, error) {
if ip == bestExitPublicTargetHost && port == bestExitPublicTargetPort {
if cached, ok := cache[nodeID]; ok {
return cached.latency, cached.loss, cached.err
}
lat, loss, err := base(nodeID, ip, port, options)
cache[nodeID] = bestExitProbeResult{latency: lat, loss: loss, err: err}
return lat, loss, err
key := bestExitProbeCacheKey{NodeID: nodeID, Host: ip, Port: port}
if cached, ok := cache[key]; ok {
return cached.latency, cached.loss, cached.err
}
return base(nodeID, ip, port, options)
lat, loss, err := base(nodeID, ip, port, options)
cache[key] = bestExitProbeResult{latency: lat, loss: loss, err: err}
return lat, loss, err
}
}
go-backend/internal/http/handler/tunnel_best_exit_test.go
+37 -6
View File
@@ -2,6 +2,9 @@ package handler
import (
"errors"
"fmt"
"slices"
"strings"
"testing"
"time"
)
@@ -189,7 +192,7 @@ func TestBestExitEnsureAppliedDoesNotOverrideExistingAppliedExit(t *testing.T) {
}
}
func TestBestExitRoundPingerCachesPublicProbeOnly(t *testing.T) {
func TestBestExitRoundPingerCachesByNodeHostAndPort(t *testing.T) {
publicCalls := 0
ownerCalls := 0
pinger := newBestExitRoundPinger(func(nodeID int64, ip string, port int, _ diagnosisExecOptions) (float64, float64, error) {
@@ -220,8 +223,8 @@ func TestBestExitRoundPingerCachesPublicProbeOnly(t *testing.T) {
if _, _, err := pinger(10, "10.0.0.30", 30030, diagnosisExecOptions{}); err != nil {
t.Fatalf("unexpected repeated owner ping err=%v", err)
}
if ownerCalls != 2 {
t.Fatalf("expected owner-to-exit probes not cached, got %d calls", ownerCalls)
if ownerCalls != 1 {
t.Fatalf("expected owner-to-exit probes cached by target, got %d calls", ownerCalls)
}
}
@@ -375,7 +378,7 @@ func TestEvaluateBestExitOwnerScoresAllCandidates(t *testing.T) {
}
}
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, pinger)
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, defaultTunnelProbeTarget(), pinger)
if len(scores) != 2 {
t.Fatalf("expected two scores, got %+v", scores)
}
@@ -384,6 +387,34 @@ func TestEvaluateBestExitOwnerScoresAllCandidates(t *testing.T) {
}
}
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"},
}
target := tunnelProbeTarget{Host: "speed.example.com", Port: 8443}
var calls []string
ping := 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
}
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, target, ping)
if len(scores) != 1 || !scores[0].Success {
t.Fatalf("expected successful score, got %+v", scores)
}
if !slices.Contains(calls, "30|speed.example.com|8443") {
t.Fatalf("expected exit public probe to use configured target, calls=%+v", calls)
}
for _, call := range calls {
if strings.Contains(call, defaultTunnelProbeTargetHost) {
t.Fatalf("did not expect default target call when custom target configured: %+v", calls)
}
}
}
func TestEvaluateBestExitOwnerMarksCandidateFailedWhenOwnerToExitFails(t *testing.T) {
owner := chainNodeRecord{NodeID: 10, NodeName: "entry"}
exits := []chainNodeRecord{{NodeID: 30, NodeName: "exit-a", Port: 30030}}
@@ -395,7 +426,7 @@ func TestEvaluateBestExitOwnerMarksCandidateFailedWhenOwnerToExitFails(t *testin
return 0, 100, errBestExitProbeForTest
}
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, pinger)
scores := evaluateBestExitOwner(owner, exits, nodes, "", diagnosisExecOptions{}, defaultTunnelProbeTarget(), pinger)
if len(scores) != 1 || scores[0].Success {
t.Fatalf("expected failed candidate, got %+v", scores)
}
@@ -413,7 +444,7 @@ func TestEvaluateBestExitOwnerMarksCandidateFailedWhenTargetResolutionFails(t *t
return 0, 100, nil
}
scores := evaluateBestExitOwner(owner, exits, nodes, "v4", diagnosisExecOptions{}, pinger)
scores := evaluateBestExitOwner(owner, exits, nodes, "v4", diagnosisExecOptions{}, defaultTunnelProbeTarget(), pinger)
if len(scores) != 1 || scores[0].Success {
t.Fatalf("expected failed candidate, got %+v", scores)
}
go-backend/internal/http/handler/tunnel_quality_prober.go
+20 -3
View File
@@ -242,7 +242,8 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
pingTimeoutMS: tunnelQualityPingTimeoutMs,
timeoutMessage: "探测超时",
}
p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options)
probeTarget := p.probeTargetForTunnel(tunnelID)
p.probeBestExitOwners(tunnelID, inNodes, midNodesGrouped, outNodes, ipPreference, options, probeTarget)
switch tunnel.Type {
case 1:
@@ -376,7 +377,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
p.storeResult(snap)
}
func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chainNodeRecord, chainHops [][]chainNodeRecord, outNodes []chainNodeRecord, ipPreference string, options diagnosisExecOptions) {
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
}
@@ -407,7 +408,7 @@ func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chai
}
key := bestExitOwnerKey{TunnelID: tunnelID, OwnerNodeID: owner.NodeID}
p.handler.bestExit.ensureApplied(key, outNodes[0].NodeID, time.Now())
scores := evaluateBestExitOwner(owner, outNodes, nodeMap, ipPreference, options, roundPinger)
scores := evaluateBestExitOwner(owner, outNodes, nodeMap, ipPreference, options, probeTarget, roundPinger)
decision := p.handler.bestExit.observeScores(key, scores, time.Now())
if decision.Switch {
now := time.Now()
@@ -421,6 +422,22 @@ func (p *tunnelQualityProber) probeBestExitOwners(tunnelID int64, inNodes []chai
}
}
func (p *tunnelQualityProber) probeTargetForTunnel(tunnelID int64) tunnelProbeTarget {
if p == nil || p.handler == nil || p.handler.repo == nil {
return defaultTunnelProbeTarget()
}
tunnels, err := p.handler.repo.ListTunnels()
if err != nil {
return defaultTunnelProbeTarget()
}
for _, tunnel := range tunnels {
if asInt64(tunnel["id"], 0) == tunnelID {
return effectiveTunnelProbeTargetValues(asString(tunnel["probeTargetHost"]), asInt(tunnel["probeTargetPort"], 0))
}
}
return defaultTunnelProbeTarget()
}
func (p *tunnelQualityProber) tcpPingNode(nodeID int64, ip string, port int, options diagnosisExecOptions) (latency float64, loss float64, err error) {
h := p.handler
if h == nil {