Refactor the polling and log

wzshiming · wzshiming · commit 146a7d32bf79 · 2025-03-04T17:42:13.000+08:00
diff --git a/chain/bridge.go b/chain/bridge.go
@@ -181,7 +181,6 @@ func (b *Bridge) bridgeStream(ctx context.Context, listenConfig bridge.ListenCon
 					raw = idle.NewIdleConn(raw, idleTimeout)
 				}
 				backoff = time.Second / 10
-				b.logger.Info("Connect", "remote_address", raw.RemoteAddr().String())
 				go b.stepIgnoreErr(ctx, dialer, raw, dials)
 			}
 		}(i, l)
@@ -289,7 +288,6 @@ func (b *Bridge) bridgeProxy(ctx context.Context, listenConfig bridge.ListenConf
 					raw = idle.NewIdleConn(raw, idleTimeout)
 				}
 				backoff = time.Second / 10
-				b.logger.Info("Connect", "remote_address", raw.RemoteAddr().String())
 				go h.ServeConn(raw)
 			}
 		}(i, host)
diff --git a/chain/chain.go b/chain/chain.go
@@ -2,30 +2,35 @@ package chain
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"math/rand"
+	"math"
+	"net"
 	"strings"
+	"sync"
 
 	"github.com/wzshiming/bridge"
 	"github.com/wzshiming/bridge/config"
 	"github.com/wzshiming/bridge/internal/scheme"
-	"github.com/wzshiming/schedialer"
-	"github.com/wzshiming/schedialer/plugins/probe"
-	"github.com/wzshiming/schedialer/plugins/roundrobin"
+	"github.com/wzshiming/bridge/logger"
 )
 
 // BridgeChain is a bridger that supports multiple crossing of bridger.
 type BridgeChain struct {
 	DialerFunc   func(dialer bridge.Dialer) bridge.Dialer
 	proto        map[string]bridge.Bridger
 	defaultProto bridge.Bridger
+
+	backoffCount map[string]uint64
+	mutex        sync.Mutex
 }
 
 // NewBridgeChain create a new BridgeChain.
 func NewBridgeChain() *BridgeChain {
 	return &BridgeChain{
-		proto:      map[string]bridge.Bridger{},
-		DialerFunc: NewEnvDialer,
+		proto:        map[string]bridge.Bridger{},
+		DialerFunc:   NewEnvDialer,
+		backoffCount: map[string]uint64{},
 	}
 }
 
@@ -35,7 +40,7 @@ func (b *BridgeChain) BridgeChain(ctx context.Context, dialer bridge.Dialer, add
 		return dialer, nil
 	}
 	address := addresses[len(addresses)-1]
-	d, err := b.Dial(ctx, dialer, strings.Split(address, "|"), "")
+	d, err := b.multiDial(dialer, strings.Split(address, "|"))
 	if err != nil {
 		return nil, err
 	}
@@ -65,7 +70,7 @@ func (b *BridgeChain) bridgeChainWithConfig(ctx context.Context, dialer bridge.D
 		return dialer, nil
 	}
 	address := addresses[len(addresses)-1]
-	d, err := b.Dial(ctx, dialer, address.LB, address.Probe)
+	d, err := b.multiDial(dialer, address.LB)
 	if err != nil {
 		return nil, err
 	}
@@ -76,32 +81,16 @@ func (b *BridgeChain) bridgeChainWithConfig(ctx context.Context, dialer bridge.D
 	return b.bridgeChainWithConfig(ctx, d, addresses...)
 }
 
-func (b *BridgeChain) Dial(ctx context.Context, dialer bridge.Dialer, addresses []string, probeUrl string) (bridge.Dialer, error) {
-	if len(addresses) == 1 {
-		return b.dialOne(ctx, dialer, addresses[0])
-	}
-	plugins := []schedialer.Plugin{
-		roundrobin.NewRoundRobinWithIndex(100, rand.Uint64()%uint64(len(addresses))),
-	}
-	if probeUrl != "" {
-		plugins = append(plugins, probe.NewProbe(100, probeUrl))
-	}
-	plugin := schedialer.NewPlugins(plugins...)
-	for _, address := range addresses {
-		dial, err := b.dialOne(ctx, dialer, address)
-		if err != nil {
-			return nil, err
-		}
-		proxy := schedialer.Proxy{
-			Name:   address,
-			Dialer: dial,
-		}
-		plugin.AddProxy(ctx, &proxy)
+func (b *BridgeChain) multiDial(dialer bridge.Dialer, addresses []string) (bridge.Dialer, error) {
+	useCount := &backoffManager{
+		addresses: addresses,
+		dialer:    dialer,
+		bc:        b,
 	}
-	return schedialer.NewSchedialer(plugin), nil
+	return useCount, nil
 }
 
-func (b *BridgeChain) dialOne(ctx context.Context, dialer bridge.Dialer, address string) (bridge.Dialer, error) {
+func (b *BridgeChain) singleDial(ctx context.Context, dialer bridge.Dialer, address string) (bridge.Dialer, error) {
 	sch, _, ok := scheme.SplitSchemeAddr(address)
 	if !ok {
 		return nil, fmt.Errorf("unsupported protocol format %q", address)
@@ -126,3 +115,63 @@ func (b *BridgeChain) Register(name string, bridger bridge.Bridger) error {
 func (b *BridgeChain) RegisterDefault(bridger bridge.Bridger) {
 	b.defaultProto = bridger
 }
+
+type backoffManager struct {
+	addresses []string
+	dialer    bridge.Dialer
+
+	bc *BridgeChain
+}
+
+func (u *backoffManager) useLeastAndCount(addresses []string) string {
+	if len(addresses) == 1 {
+		return addresses[0]
+	}
+	min := uint64(math.MaxUint64)
+
+	u.bc.mutex.Lock()
+	defer u.bc.mutex.Unlock()
+
+	var minAddress string
+	for _, address := range addresses {
+		if u.bc.backoffCount[address] < min {
+			min = u.bc.backoffCount[address]
+			minAddress = address
+		}
+	}
+	u.bc.backoffCount[minAddress]++
+	return minAddress
+}
+
+func (u *backoffManager) backoff(address string, count uint64) {
+	u.bc.mutex.Lock()
+	defer u.bc.mutex.Unlock()
+	u.bc.backoffCount[address] += count
+}
+
+func (u *backoffManager) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	var errs []error
+
+	tryTimes := len(u.addresses)
+	for i := 0; i < tryTimes; i++ {
+		addr := u.useLeastAndCount(u.addresses)
+		dialer, err := u.bc.singleDial(ctx, u.dialer, addr)
+		if err != nil {
+			errs = append(errs, err)
+			logger.Std.Warn("failed dial", "err", err, "previous", addr)
+			u.backoff(addr, 16)
+			continue
+		}
+		conn, err := dialer.DialContext(ctx, network, address)
+		if err != nil {
+			errs = append(errs, err)
+			logger.Std.Warn("failed dial target", "err", err, "previous", addr, "target", address)
+			u.backoff(addr, 8)
+			continue
+		}
+
+		logger.Std.Info("success dial target", "previous", addr, "target", address)
+		return conn, nil
+	}
+	return nil, fmt.Errorf("all addresses are failed: %w", errors.Join(errs...))
+}
diff --git a/cmd/bridge/main.go b/cmd/bridge/main.go
@@ -98,14 +98,14 @@ func main() {
 		tasks, err = config.LoadConfig(configs...)
 		if err != nil {
 			printDefaults()
-			slog.Error("LoadConfig", "err", err)
+			logger.Std.Error("LoadConfig", "err", err)
 			return
 		}
 	} else {
 		tasks, err = config.LoadConfigWithArgs(listens, dials)
 		if err != nil {
 			printDefaults()
-			slog.Error("LoadConfigWithArgs", "err", err)
+			logger.Std.Error("LoadConfigWithArgs", "err", err)
 			return
 		}
 	}
@@ -136,7 +136,6 @@ func run(ctx context.Context, log *slog.Logger, tasks []config.Chain) {
 		}
 		go func(task config.Chain) {
 			defer wg.Done()
-			log := log.With("chains", task)
 			log.Info(chain.ShowChainWithConfig(task))
 			b := chain.NewBridge(log, dump)
 			err := b.BridgeWithConfig(ctx, task)
diff --git a/cmd/bridge/main_other.go b/cmd/bridge/main_other.go
@@ -63,7 +63,6 @@ func runWithReload(ctx context.Context, log *slog.Logger, tasks []config.Chain,
 			wg.Add(1)
 			go func(ctx context.Context, task config.Chain) {
 				defer wg.Done()
-				log := log.With("chains", task)
 				log.Info(chain.ShowChainWithConfig(task))
 				for ctx.Err() == nil {
 					b := chain.NewBridge(log, dump)
diff --git a/config/config.go b/config/config.go
@@ -3,13 +3,13 @@ package config
 import (
 	"encoding/json"
 	"fmt"
-	"log"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/wzshiming/bridge/internal/scheme"
+	"github.com/wzshiming/bridge/logger"
 )
 
 func LoadConfigWithArgs(listens []string, dials []string) ([]Chain, error) {
@@ -55,7 +55,7 @@ func LoadConfig(configs ...string) ([]Chain, error) {
 	for _, confPath := range configs {
 		data, err := os.ReadFile(confPath)
 		if err != nil {
-			log.Println(err)
+			logger.Std.Error("LoadConfig", "err", err, "path", confPath)
 			continue
 		}
 		conf := Config{}
@@ -100,8 +100,7 @@ func (c Chain) Unique() string {
 }
 
 type Node struct {
-	Probe string   `json:"probe"`
-	LB    []string `json:"lb"`
+	LB []string `json:"lb"`
 }
 
 func (m Node) MarshalJSON() ([]byte, error) {
diff --git a/go.mod b/go.mod
@@ -14,7 +14,6 @@ require (
 	github.com/wzshiming/httpproxy v0.5.6
 	github.com/wzshiming/notify v0.1.1
 	github.com/wzshiming/permuteproxy v0.0.2
-	github.com/wzshiming/schedialer v0.6.1
 	github.com/wzshiming/shadowsocks v0.4.1
 	github.com/wzshiming/socks4 v0.3.2
 	github.com/wzshiming/socks5 v0.5.1
diff --git a/go.sum b/go.sum
@@ -26,8 +26,6 @@ github.com/wzshiming/notify v0.1.1 h1:rJXoszrkNglhCVyn/IfW500f5cW03q1q7YzL8hsLch
 github.com/wzshiming/notify v0.1.1/go.mod h1:SFhsQKZJznzsDcj/Qfo9A65k5IRcpUrpgbLRzZEa/DI=
 github.com/wzshiming/permuteproxy v0.0.2 h1:svedMueotlxJk9oJfA0gs8WzRYOdgd0DER9XvKpjwlY=
 github.com/wzshiming/permuteproxy v0.0.2/go.mod h1:Ny08A1JbuljB8FeJAOiB7dfvRGCVD8PB9hwrALIvYI8=
-github.com/wzshiming/schedialer v0.6.1 h1:4VwtIjVF3uMoWqjbyw3oqYi7WGOEYvDu3L9OYT8sbGY=
-github.com/wzshiming/schedialer v0.6.1/go.mod h1:TvVxg4QZIBTJzRfmL/G7g6CzynFQKPmtXtSeJ2c4Lus=
 github.com/wzshiming/shadowsocks v0.4.1 h1:tyLYtLSQs90jpMIkD+T8KuZH5foXwOH0ZjxSOb45orI=
 github.com/wzshiming/shadowsocks v0.4.1/go.mod h1:CfKm/Keclli2sPGfjskGVH+F3gpF0YPVdcf5t4krypY=
 github.com/wzshiming/socks4 v0.3.2 h1:w87nwfgRWteVwIH39nqTur8c+2dcODeLgLrWspcUkSc=
diff --git a/logger/logger.go b/logger/logger.go
@@ -1,11 +1,15 @@
 package logger
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
+	"os"
 )
 
-var Std = slog.Default()
+var Std = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
+	Level: slog.LevelInfo,
+}))
 
 func Wrap(logger *slog.Logger, name string) *wrap {
 	return &wrap{
@@ -18,5 +22,5 @@ type wrap struct {
 }
 
 func (w wrap) Println(v ...interface{}) {
-	w.Logger.Info(fmt.Sprintln(v...))
+	w.Logger.Log(context.Background(), slog.LevelWarn, "print", "message", fmt.Sprint(v...))
 }
diff --git a/protocols/local/local.go b/protocols/local/local.go
@@ -2,13 +2,13 @@ package local
 
 import (
 	"context"
-	"log/slog"
 	"net"
 	"os"
 	"strings"
 
 	"github.com/wzshiming/bridge"
 	"github.com/wzshiming/bridge/internal/netutils"
+	"github.com/wzshiming/bridge/logger"
 	"github.com/wzshiming/commandproxy"
 )
 
@@ -25,17 +25,17 @@ type Local struct {
 }
 
 func (l *Local) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
-	slog.Debug("Dial", "network", network, "address", address)
+	logger.Std.Debug("Dial", "network", network, "address", address)
 	return l.Dialer.DialContext(ctx, network, address)
 }
 
 func (l *Local) Listen(ctx context.Context, network, address string) (net.Listener, error) {
-	slog.Debug("Listen", "network", network, "address", address)
+	logger.Std.Debug("Listen", "network", network, "address", address)
 	return l.ListenConfig.Listen(ctx, network, address)
 }
 
 func (l *Local) CommandDialContext(ctx context.Context, name string, args ...string) (net.Conn, error) {
-	slog.Debug("CommandDial", "name", name, "args", args)
+	logger.Std.Debug("CommandDial", "name", name, "args", args)
 	proxy := commandproxy.ProxyCommand(ctx, name, args...)
 	proxy.Stderr = os.Stderr
 	conn, err := proxy.Stdio()
@@ -48,7 +48,7 @@ func (l *Local) CommandDialContext(ctx context.Context, name string, args ...str
 }
 
 func (l *Local) CommandListen(ctx context.Context, name string, args ...string) (net.Listener, error) {
-	slog.Debug("CommandListen", "name", name, "args", args)
+	logger.Std.Debug("CommandListen", "name", name, "args", args)
 	proxy := append([]string{name}, args...)
 	remoteAddr := netutils.NewNetAddr("cmd", strings.Join(proxy, " "))
 	return netutils.NewCommandListener(ctx, l, l.LocalAddr, remoteAddr, proxy)

Original file line number	Diff line number	Diff line change
`@@ -181,7 +181,6 @@ func (b *Bridge) bridgeStream(ctx context.Context, listenConfig bridge.ListenCon`
`181`	`181`	`raw = idle.NewIdleConn(raw, idleTimeout)`
`182`	`182`	`}`
`183`	`183`	`backoff = time.Second / 10`
`184`		`- b.logger.Info("Connect", "remote_address", raw.RemoteAddr().String())`
`185`	`184`	`go b.stepIgnoreErr(ctx, dialer, raw, dials)`
`186`	`185`	`}`
`187`	`186`	`}(i, l)`
`@@ -289,7 +288,6 @@ func (b *Bridge) bridgeProxy(ctx context.Context, listenConfig bridge.ListenConf`
`289`	`288`	`raw = idle.NewIdleConn(raw, idleTimeout)`
`290`	`289`	`}`
`291`	`290`	`backoff = time.Second / 10`
`292`		`- b.logger.Info("Connect", "remote_address", raw.RemoteAddr().String())`
`293`	`291`	`go h.ServeConn(raw)`
`294`	`292`	`}`
`295`	`293`	`}(i, host)`
Original file line number	Diff line number	Diff line change
`@@ -98,14 +98,14 @@ func main() {`
`98`	`98`	`tasks, err = config.LoadConfig(configs...)`
`99`	`99`	`if err != nil {`
`100`	`100`	`printDefaults()`
`101`		`- slog.Error("LoadConfig", "err", err)`
	`101`	`+ logger.Std.Error("LoadConfig", "err", err)`
`102`	`102`	`return`
`103`	`103`	`}`
`104`	`104`	`} else {`
`105`	`105`	`tasks, err = config.LoadConfigWithArgs(listens, dials)`
`106`	`106`	`if err != nil {`
`107`	`107`	`printDefaults()`
`108`		`- slog.Error("LoadConfigWithArgs", "err", err)`
	`108`	`+ logger.Std.Error("LoadConfigWithArgs", "err", err)`
`109`	`109`	`return`
`110`	`110`	`}`
`111`	`111`	`}`
`@@ -136,7 +136,6 @@ func run(ctx context.Context, log *slog.Logger, tasks []config.Chain) {`
`136`	`136`	`}`
`137`	`137`	`go func(task config.Chain) {`
`138`	`138`	`defer wg.Done()`
`139`		`- log := log.With("chains", task)`
`140`	`139`	`log.Info(chain.ShowChainWithConfig(task))`
`141`	`140`	`b := chain.NewBridge(log, dump)`
`142`	`141`	`err := b.BridgeWithConfig(ctx, task)`
Original file line number	Diff line number	Diff line change
`@@ -1,11 +1,15 @@`
`1`	`1`	`package logger`
`2`	`2`
`3`	`3`	`import (`
	`4`	`+ "context"`
`4`	`5`	`"fmt"`
`5`	`6`	`"log/slog"`
	`7`	`+ "os"`
`6`	`8`	`)`
`7`	`9`
`8`		`-var Std = slog.Default()`
	`10`	`+var Std = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{`
	`11`	`+ Level: slog.LevelInfo,`
	`12`	`+}))`
`9`	`13`
`10`	`14`	`func Wrap(logger slog.Logger, name string) wrap {`
`11`	`15`	`return &wrap{`
`@@ -18,5 +22,5 @@ type wrap struct {`
`18`	`22`	`}`
`19`	`23`
`20`	`24`	`func (w wrap) Println(v ...interface{}) {`
`21`		`- w.Logger.Info(fmt.Sprintln(v...))`
	`25`	`+ w.Logger.Log(context.Background(), slog.LevelWarn, "print", "message", fmt.Sprint(v...))`
`22`	`26`	`}`