package outboundgroup import ( "context" "encoding/json" "errors" "fmt" "math/rand" "net" "time" "github.com/Dreamacro/clash/adapter/outbound" "github.com/Dreamacro/clash/common/murmur3" "github.com/Dreamacro/clash/component/dialer" C "github.com/Dreamacro/clash/constant" "github.com/Dreamacro/clash/constant/provider" "golang.org/x/net/publicsuffix" ) type strategyFn = func(proxies []C.Proxy, metadata *C.Metadata) C.Proxy type LoadBalance struct { *GroupBase disableUDP bool strategyFn strategyFn } var errStrategy = errors.New("unsupported strategy") func parseStrategy(config map[string]any) string { if elm, ok := config["strategy"]; ok { if strategy, ok := elm.(string); ok { return strategy } } return "consistent-hashing" } func getKey(metadata *C.Metadata) string { if metadata == nil { return "" } if metadata.Host != "" { // ip host if ip := net.ParseIP(metadata.Host); ip != nil { return metadata.Host } if etld, err := publicsuffix.EffectiveTLDPlusOne(metadata.Host); err == nil { return etld } } if !metadata.DstIP.IsValid() { return "" } return metadata.DstIP.String() } func jumpHash(key uint64, buckets int32) int32 { var b, j int64 for j < int64(buckets) { b = j key = key*2862933555777941757 + 1 j = int64(float64(b+1) * (float64(int64(1)<<31) / float64((key>>33)+1))) } return int32(b) } // DialContext implements C.ProxyAdapter func (lb *LoadBalance) DialContext(ctx context.Context, metadata *C.Metadata, opts ...dialer.Option) (c C.Conn, err error) { defer func() { if err == nil { c.AppendToChains(lb) lb.onDialSuccess() } else { lb.onDialFailed() } }() proxy := lb.Unwrap(metadata) c, err = proxy.DialContext(ctx, metadata, lb.Base.DialOptions(opts...)...) return } // ListenPacketContext implements C.ProxyAdapter func (lb *LoadBalance) ListenPacketContext(ctx context.Context, metadata *C.Metadata, opts ...dialer.Option) (pc C.PacketConn, err error) { defer func() { if err == nil { pc.AppendToChains(lb) } }() proxy := lb.Unwrap(metadata) return proxy.ListenPacketContext(ctx, metadata, lb.Base.DialOptions(opts...)...) } // SupportUDP implements C.ProxyAdapter func (lb *LoadBalance) SupportUDP() bool { return !lb.disableUDP } func strategyRoundRobin() strategyFn { idx := 0 return func(proxies []C.Proxy, metadata *C.Metadata) C.Proxy { length := len(proxies) for i := 0; i < length; i++ { idx = (idx + 1) % length proxy := proxies[idx] if proxy.Alive() { return proxy } } return proxies[0] } } func strategyConsistentHashing() strategyFn { maxRetry := 5 return func(proxies []C.Proxy, metadata *C.Metadata) C.Proxy { key := uint64(murmur3.Sum32([]byte(getKey(metadata)))) buckets := int32(len(proxies)) for i := 0; i < maxRetry; i, key = i+1, key+1 { idx := jumpHash(key, buckets) proxy := proxies[idx] if proxy.Alive() { return proxy } } return proxies[0] } } func strategyStickySessions() strategyFn { timeout := int64(600) type Session struct { idx int time time.Time } Sessions := make(map[string]map[string]Session) go func() { for { time.Sleep(time.Second * 60) now := time.Now().Unix() for _, subMap := range Sessions { for dest, session := range subMap { if now-session.time.Unix() > timeout { delete(subMap, dest) } } } } }() return func(proxies []C.Proxy, metadata *C.Metadata) C.Proxy { src := metadata.SrcIP.String() dest := getKey(metadata) now := time.Now() length := len(proxies) if Sessions[src] == nil { Sessions[src] = make(map[string]Session) } session, ok := Sessions[src][dest] if !ok || now.Unix()-session.time.Unix() > timeout { session.idx = rand.Intn(length) } session.time = now session.idx = 0 var res = proxies[0] for i := 0; i < length; i++ { idx := (session.idx + i) % length proxy := proxies[idx] if proxy.Alive() { session.idx = idx res = proxy break } } Sessions[src][dest] = session return res } } // Unwrap implements C.ProxyAdapter func (lb *LoadBalance) Unwrap(metadata *C.Metadata) C.Proxy { proxies := lb.GetProxies(true) return lb.strategyFn(proxies, metadata) } // MarshalJSON implements C.ProxyAdapter func (lb *LoadBalance) MarshalJSON() ([]byte, error) { var all []string for _, proxy := range lb.GetProxies(false) { all = append(all, proxy.Name()) } return json.Marshal(map[string]any{ "type": lb.Type().String(), "all": all, }) } func NewLoadBalance(option *GroupCommonOption, providers []provider.ProxyProvider, strategy string) (lb *LoadBalance, err error) { var strategyFn strategyFn switch strategy { case "consistent-hashing": strategyFn = strategyConsistentHashing() case "round-robin": strategyFn = strategyRoundRobin() case "sticky-sessions": strategyFn = strategyStickySessions() default: return nil, fmt.Errorf("%w: %s", errStrategy, strategy) } return &LoadBalance{ GroupBase: NewGroupBase(GroupBaseOption{ outbound.BaseOption{ Name: option.Name, Type: C.LoadBalance, Interface: option.Interface, RoutingMark: option.RoutingMark, }, option.Filter, providers, }), strategyFn: strategyFn, disableUDP: option.DisableUDP, }, nil }