feat: outbound latency test

Author: M03ED

.env.example

@@ -22,6 +22,8 @@ API_KEY = xxxxxxxx-yyyy-zzzz-mmmm-aaaaaaaaaaa
 # STARTUP_LOG_TAIL_SIZE = 200
 # STATS_UPDATE_INTERVAL_SECONDS = 10
 # STATS_CLEANUP_INTERVAL_SECONDS = 300
+# LATENCY_TEST_URL = https://www.gstatic.com/generate_204
+# LATENCY_TIMEOUT_SECONDS = 5
 
 ### WireGuard host NAT
 ### Built-in routing enables runtime IPv4 forwarding and manages scoped nft NAT/forwarding rules.

backend/backend.go

@@ -18,6 +18,7 @@ type Backend interface {
 	UpdateUsersAndRestart(context.Context, []*common.User) error
 	GetSysStats(context.Context) (*common.BackendStatsResponse, error)
 	GetStats(context.Context, *common.StatRequest) (*common.StatResponse, error)
+	GetOutboundsLatency(context.Context, *common.LatencyRequest) (*common.LatencyResponse, error)
 	GetUserOnlineStats(context.Context, string) (*common.OnlineStatResponse, error)
 	GetUserOnlineIpListStats(context.Context, string) (*common.StatsOnlineIpListResponse, error)
 }

backend/wireguard/latency.go

@@ -0,0 +1,83 @@
+package wireguard
+
+import (
+	"context"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/pasarguard/node/common"
+)
+
+const wireGuardNATOutputInterfaceEnv = "PG_NODE_WG_NAT_OUTPUT_INTERFACE"
+
+func (wg *WireGuard) latencyProbeInterface() string {
+	if iface := strings.TrimSpace(os.Getenv(wireGuardNATOutputInterfaceEnv)); iface != "" {
+		return iface
+	}
+
+	wg.mu.RLock()
+	defer wg.mu.RUnlock()
+	if wg.config == nil {
+		return ""
+	}
+	return strings.TrimSpace(wg.config.InterfaceName)
+}
+
+func (wg *WireGuard) GetOutboundsLatency(ctx context.Context, request *common.LatencyRequest) (*common.LatencyResponse, error) {
+	wg.mu.RLock()
+	state := wg.state
+	testURL := wg.cfg.LatencyTestURL
+	timeoutSeconds := wg.cfg.LatencyTimeoutSeconds
+	wg.mu.RUnlock()
+
+	if state != lifecycleRunning {
+		return nil, errWireGuardNotStarted
+	}
+
+	iface := wg.latencyProbeInterface()
+	name := iface
+	if name == "" {
+		name = "wireguard"
+	}
+
+	requestName := strings.TrimSpace(request.GetName())
+	if requestName != "" && requestName != name {
+		return &common.LatencyResponse{Latencies: []*common.Latency{}}, nil
+	}
+
+	client := &http.Client{
+		Timeout:   time.Duration(timeoutSeconds) * time.Second,
+		Transport: latencyHTTPTransport(iface),
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodHead, testURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	start := time.Now()
+	resp, err := client.Do(req)
+	delay := time.Since(start).Milliseconds()
+	now := time.Now().Unix()
+
+	alive := err == nil
+	if resp != nil && resp.Body != nil {
+		resp.Body.Close()
+	}
+
+	latency := &common.Latency{
+		Name:        name,
+		Alive:       alive,
+		Delay:       delay,
+		Link:        testURL,
+		LastTryTime: now,
+		Source:      "wireguard-probe",
+	}
+	if alive {
+		latency.LastSeenTime = now
+	}
+
+	return &common.LatencyResponse{Latencies: []*common.Latency{latency}}, nil
+}

backend/wireguard/latency_linux.go

@@ -0,0 +1,32 @@
+//go:build linux
+
+package wireguard
+
+import (
+	"net"
+	"net/http"
+	"syscall"
+
+	"golang.org/x/sys/unix"
+)
+
+func latencyHTTPTransport(iface string) *http.Transport {
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	if iface == "" {
+		return transport
+	}
+
+	dialer := &net.Dialer{
+		Control: func(_, _ string, c syscall.RawConn) error {
+			var controlErr error
+			if err := c.Control(func(fd uintptr) {
+				controlErr = unix.SetsockoptString(int(fd), unix.SOL_SOCKET, unix.SO_BINDTODEVICE, iface)
+			}); err != nil {
+				return err
+			}
+			return controlErr
+		},
+	}
+	transport.DialContext = dialer.DialContext
+	return transport
+}

backend/wireguard/latency_other.go

@@ -0,0 +1,9 @@
+//go:build !linux
+
+package wireguard
+
+import "net/http"
+
+func latencyHTTPTransport(_ string) *http.Transport {
+	return http.DefaultTransport.(*http.Transport).Clone()
+}

backend/wireguard/latency_test.go

@@ -0,0 +1,27 @@
+package wireguard
+
+import "testing"
+
+func TestLatencyProbeInterfaceFallsBackToConfiguredInterface(t *testing.T) {
+	t.Setenv(wireGuardNATOutputInterfaceEnv, "")
+
+	wg := &WireGuard{
+		config: &Config{InterfaceName: "wg-test"},
+	}
+
+	if got := wg.latencyProbeInterface(); got != "wg-test" {
+		t.Fatalf("unexpected interface: got %s want wg-test", got)
+	}
+}
+
+func TestLatencyProbeInterfacePrefersNATEgressEnv(t *testing.T) {
+	t.Setenv(wireGuardNATOutputInterfaceEnv, "eth9")
+
+	wg := &WireGuard{
+		config: &Config{InterfaceName: "wg-test"},
+	}
+
+	if got := wg.latencyProbeInterface(); got != "eth9" {
+		t.Fatalf("unexpected interface: got %s want eth9", got)
+	}
+}

backend/xray/config.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"net/netip"
 	"slices"
 	"sort"
 	"strings"
@@ -575,7 +576,12 @@ func apiRuleSources() []string {
 	return sources
 }
 
-func (c *Config) ApplyAPI(apiPort int) (err error) {
+func loopbackListenAddress(port int) string {
+	addr := netip.AddrPortFrom(netip.MustParseAddr("127.0.0.1"), uint16(port))
+	return addr.String()
+}
+
+func (c *Config) ApplyAPI(apiPort, metricPort int) (err error) {
 	// Remove the existing inbound with the API_INBOUND tag
 	for i, inbound := range c.InboundConfigs {
 		if inbound.Tag == "API_INBOUND" {
@@ -590,6 +596,11 @@ func (c *Config) ApplyAPI(apiPort int) (err error) {
 		Tag:      apiTag,
 	}
 
+	c.Metrics = map[string]any{
+		"tag":    "metric",
+		"listen": loopbackListenAddress(metricPort),
+	}
+
 	if c.RouterConfig == nil {
 		c.RouterConfig = &conf.RouterConfig{}
 	}

backend/xray/latency.go

@@ -0,0 +1,93 @@
+package xray
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/netip"
+	"sort"
+	"time"
+
+	"github.com/pasarguard/node/common"
+)
+
+type observatoryEntry struct {
+	Alive        bool   `json:"alive"`
+	Delay        int64  `json:"delay"`
+	OutboundTag  string `json:"outbound_tag"`
+	LastSeenTime int64  `json:"last_seen_time"`
+	LastTryTime  int64  `json:"last_try_time"`
+}
+
+type debugVarsResponse struct {
+	Observatory map[string]observatoryEntry `json:"observatory"`
+}
+
+func (x *Xray) GetOutboundsLatency(ctx context.Context, request *common.LatencyRequest) (*common.LatencyResponse, error) {
+	x.mu.RLock()
+	started := x.core != nil && x.core.Started()
+	metricPort := x.metricPort
+	timeoutSeconds := x.cfg.LatencyTimeoutSeconds
+	x.mu.RUnlock()
+
+	if !started {
+		return nil, errors.New("xray not started")
+	}
+
+	client := &http.Client{Timeout: time.Duration(timeoutSeconds) * time.Second}
+	addr := netip.AddrPortFrom(netip.MustParseAddr("127.0.0.1"), uint16(metricPort))
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr.String()+"/debug/vars", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("read xray observatory metrics: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("read xray observatory metrics: unexpected status %d", resp.StatusCode)
+	}
+
+	var vars debugVarsResponse
+	if err := json.NewDecoder(resp.Body).Decode(&vars); err != nil {
+		return nil, fmt.Errorf("decode xray observatory metrics: %w", err)
+	}
+	if len(vars.Observatory) == 0 {
+		return nil, errors.New("xray outbound latency is not available")
+	}
+
+	name := request.GetName()
+	keys := make([]string, 0, len(vars.Observatory))
+	for key := range vars.Observatory {
+		if name != "" && key != name {
+			continue
+		}
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+
+	latencies := make([]*common.Latency, 0, len(keys))
+	for _, key := range keys {
+		entry := vars.Observatory[key]
+		linkName := entry.OutboundTag
+		if linkName == "" {
+			linkName = key
+		}
+		latencies = append(latencies, &common.Latency{
+			Name:         key,
+			Alive:        entry.Alive,
+			Delay:        entry.Delay,
+			Link:         linkName,
+			LastSeenTime: entry.LastSeenTime,
+			LastTryTime:  entry.LastTryTime,
+			Source:       "xray-observatory",
+		})
+	}
+
+	return &common.LatencyResponse{Latencies: latencies}, nil
+}

backend/xray/xray.go

@@ -17,11 +17,12 @@ type Xray struct {
 	cfg        *config.Config
 	core       *Core
 	handler    *api.XrayHandler
+	metricPort int
 	cancelFunc context.CancelFunc
 	mu         sync.RWMutex
 }
 
-func New(ctx context.Context, xrayConfig *Config, users []*common.User, port int, cfg *config.Config) (*Xray, error) {
+func New(ctx context.Context, xrayConfig *Config, users []*common.User, apiPort, metricPort int, cfg *config.Config) (*Xray, error) {
 	executableAbsolutePath, err := filepath.Abs(cfg.XrayExecutablePath)
 	if err != nil {
 		return nil, err
@@ -42,11 +43,12 @@ func New(ctx context.Context, xrayConfig *Config, users []*common.User, port int
 	xray := &Xray{
 		cancelFunc: xCancel,
 		cfg:        cfg,
+		metricPort: metricPort,
 	}
 
 	start := time.Now()
 
-	if err = xrayConfig.ApplyAPI(port); err != nil {
+	if err = xrayConfig.ApplyAPI(apiPort, metricPort); err != nil {
 		return nil, err
 	}
 
@@ -80,7 +82,7 @@ func New(ctx context.Context, xrayConfig *Config, users []*common.User, port int
 
 	xray.core = core
 
-	handler, err := api.NewXrayAPI(port)
+	handler, err := api.NewXrayAPI(apiPort)
 	if err != nil {
 		xray.Shutdown()
 		return nil, err