feat: allow group override global node connectivity check (#623)

Co-authored-by: mzz2017 <2017@duck.com>

component/outbound/dialer/connectivity_check.go

@@ -19,6 +19,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unsafe"
 
 	"github.com/daeuniverse/dae/common"
 
@@ -452,6 +453,18 @@ func (d *Dialer) aliveBackground() {
 			}
 		}
 	}()
+	var unused int
+	for _, opt := range CheckOpts {
+		if len(d.mustGetCollection(opt.networkType).AliveDialerSetSet) == 0 {
+			unused++
+		}
+	}
+	if unused == len(CheckOpts) {
+		d.Log.WithField("dialer", d.Property().Name).
+			WithField("p", unsafe.Pointer(d)).
+			Traceln("cleaned up due to unused")
+		return
+	}
 	var wg sync.WaitGroup
 	for range d.checkCh {
 		for _, opt := range CheckOpts {

component/outbound/dialer/dialer.go

@@ -10,7 +10,10 @@ import (
 	"fmt"
 	"sync"
 	"time"
+	"unsafe"
 
+	"github.com/daeuniverse/dae/common"
+	"github.com/daeuniverse/dae/config"
 	D "github.com/daeuniverse/outbound/dialer"
 	"github.com/daeuniverse/outbound/netproxy"
 	"github.com/sirupsen/logrus"
@@ -60,6 +63,21 @@ type Property struct {
 
 type AliveDialerSetSet map[*AliveDialerSet]int
 
+func NewGlobalOption(global *config.Global, log *logrus.Logger) *GlobalOption {
+	return &GlobalOption{
+		ExtraOption: D.ExtraOption{
+			AllowInsecure:     global.AllowInsecure,
+			TlsImplementation: global.TlsImplementation,
+			UtlsImitate:       global.UtlsImitate},
+		Log:               log,
+		TcpCheckOptionRaw: TcpCheckOptionRaw{Raw: global.TcpCheckUrl, Log: log, ResolverNetwork: common.MagicNetwork("udp", global.SoMarkFromDae, global.Mptcp), Method: global.TcpCheckHttpMethod},
+		CheckDnsOptionRaw: CheckDnsOptionRaw{Raw: global.UdpCheckDns, ResolverNetwork: common.MagicNetwork("udp", global.SoMarkFromDae, global.Mptcp), Somark: global.SoMarkFromDae},
+		CheckInterval:     global.CheckInterval,
+		CheckTolerance:    global.CheckTolerance,
+		CheckDnsTcp:       true,
+	}
+}
+
 // NewDialer is for register in general.
 func NewDialer(dialer netproxy.Dialer, option *GlobalOption, iOption InstanceOption, property *Property) *Dialer {
 	var collections [6]*collection
@@ -80,9 +98,16 @@ func NewDialer(dialer netproxy.Dialer, option *GlobalOption, iOption InstanceOpt
 		ctx:              ctx,
 		cancel:           cancel,
 	}
+	option.Log.WithField("dialer", d.Property().Name).
+		WithField("p", unsafe.Pointer(d)).
+		Traceln("NewDialer")
 	return d
 }
 
+func (d *Dialer) Clone() *Dialer {
+	return NewDialer(d.Dialer, d.GlobalOption, d.InstanceOption, d.property)
+}
+
 func (d *Dialer) Close() error {
 	d.cancel()
 	d.tickerMu.Lock()

config/config.go

@@ -89,6 +89,12 @@ type Group struct {
 	Filter           [][]*config_parser.Function `mapstructure:"filter" repeatable:""`
 	FilterAnnotation [][]*config_parser.Param    `mapstructure:"_"`
 	Policy           FunctionListOrString        `mapstructure:"policy" required:""`
+
+	TcpCheckUrl        []string      `mapstructure:"tcp_check_url"`
+	TcpCheckHttpMethod string        `mapstructure:"tcp_check_http_method"`
+	UdpCheckDns        []string      `mapstructure:"udp_check_dns"`
+	CheckInterval      time.Duration `mapstructure:"check_interval"`
+	CheckTolerance     time.Duration `mapstructure:"check_tolerance"`
 }
 
 type DnsRequestRouting struct {

config/desc.go

@@ -85,4 +85,9 @@ min: Select node by the latency of last check.
 min_avg10: Select node by the average of latencies of last 10 checks.
 min_moving_avg: Select node by the moving average of latencies of checks, which means more recent latencies have higher weight.
 `,
+	"tcp_check_url":         "Override global config.",
+	"tcp_check_http_method": "Override global config.",
+	"udp_check_dns":         "Override global config.",
+	"check_interval":        "Override global config.",
+	"check_tolerance":       "Override global config.",
 }

control/control_plane.go

@@ -33,7 +33,6 @@ import (
 	"github.com/daeuniverse/dae/config"
 	"github.com/daeuniverse/dae/pkg/config_parser"
 	internal "github.com/daeuniverse/dae/pkg/ebpf_internal"
-	D "github.com/daeuniverse/outbound/dialer"
 	"github.com/daeuniverse/outbound/pool"
 	"github.com/daeuniverse/outbound/protocol/direct"
 	"github.com/daeuniverse/outbound/transport/grpc"
@@ -256,18 +255,7 @@ func NewControlPlane(
 	if global.AllowInsecure {
 		log.Warnln("AllowInsecure is enabled, but it is not recommended. Please make sure you have to turn it on.")
 	}
-	option := &dialer.GlobalOption{
-		ExtraOption: D.ExtraOption{
-			AllowInsecure:     global.AllowInsecure,
-			TlsImplementation: global.TlsImplementation,
-			UtlsImitate:       global.UtlsImitate},
-		Log:               log,
-		TcpCheckOptionRaw: dialer.TcpCheckOptionRaw{Raw: global.TcpCheckUrl, Log: log, ResolverNetwork: common.MagicNetwork("udp", global.SoMarkFromDae, global.Mptcp), Method: global.TcpCheckHttpMethod},
-		CheckDnsOptionRaw: dialer.CheckDnsOptionRaw{Raw: global.UdpCheckDns, ResolverNetwork: common.MagicNetwork("udp", global.SoMarkFromDae, global.Mptcp), Somark: global.SoMarkFromDae},
-		CheckInterval:     global.CheckInterval,
-		CheckTolerance:    global.CheckTolerance,
-		CheckDnsTcp:       true,
-	}
+	option := dialer.NewGlobalOption(global, log)
 
 	// Dial mode.
 	dialMode, err := consts.ParseDialMode(global.DialMode)
@@ -323,8 +311,22 @@ func NewControlPlane(
 		if len(dialers) == 0 {
 			log.Infoln("\t<Empty>")
 		}
+		groupOption, err := ParseGroupOverrideOption(group, *global, log)
+		finalOption := option
+		if err == nil && groupOption != nil {
+			newDialers := make([]*dialer.Dialer, 0)
+			for _, d := range dialers {
+				newDialer := d.Clone()
+				deferFuncs = append(deferFuncs, newDialer.Close)
+				newDialer.GlobalOption = groupOption
+				newDialers = append(newDialers, newDialer)
+			}
+			log.Infof(`Group "%v"'s check option has been override.`, group.Name)
+			dialers = newDialers
+			finalOption = groupOption
+		}
 		// Create dialer group and append it to outbounds.
-		dialerGroup := outbound.NewDialerGroup(option, group.Name, dialers, annos, *policy,
+		dialerGroup := outbound.NewDialerGroup(finalOption, group.Name, dialers, annos, *policy,
 			core.outboundAliveChangeCallback(uint8(len(outbounds)), disableKernelAliveCallback))
 		outbounds = append(outbounds, dialerGroup)
 	}
@@ -515,6 +517,36 @@ func ParseFixedDomainTtl(ks []config.KeyableString) (map[string]int, error) {
 	return m, nil
 }
 
+func ParseGroupOverrideOption(group config.Group, global config.Global, log *logrus.Logger) (*dialer.GlobalOption, error) {
+	result := global
+	changed := false
+	if group.TcpCheckUrl != nil {
+		result.TcpCheckUrl = group.TcpCheckUrl
+		changed = true
+	}
+	if group.TcpCheckHttpMethod != "" {
+		result.TcpCheckHttpMethod = group.TcpCheckHttpMethod
+		changed = true
+	}
+	if group.UdpCheckDns != nil {
+		result.UdpCheckDns = group.UdpCheckDns
+		changed = true
+	}
+	if group.CheckInterval != 0 {
+		result.CheckInterval = group.CheckInterval
+		changed = true
+	}
+	if group.CheckTolerance != 0 {
+		result.CheckTolerance = group.CheckTolerance
+		changed = true
+	}
+	if changed {
+		option := dialer.NewGlobalOption(&result, log)
+		return option, nil
+	}
+	return nil, nil
+}
+
 // EjectBpf will resect bpf from destroying life-cycle of control plane.
 func (c *ControlPlane) EjectBpf() *bpfObjects {
 	return c.core.EjectBpf()

example.dae

@@ -40,6 +40,7 @@ global {
     auto_config_kernel_parameter: true
 
     ##### Node connectivity check.
+    # These options, as defaults, are effective when no definition is given in the group.
 
     # Host of URL should have both IPv4 and IPv6 if you have double stack in local.
     # First is URL, others are IP addresses if given.
@@ -213,6 +214,24 @@ group {
         # Select the node with min average of the last 10 latencies from the group for every connection.
         policy: min_avg10
     }
+
+    steam {
+        # Filter nodes from the global node pool defined by the subscription and node section above.
+        filter: subtag(my_sub) && !name(keyword: 'ExpireAt:')
+        # Select the node with min moving average of latencies from the group for every connection.
+        policy: min_moving_avg
+
+        # Override tcp_check_url in global.
+        tcp_check_url: 'http://test.steampowered.com'
+        # Override tcp_check_http_method in global
+        #tcp_check_http_method: HEAD
+        # Override udp_check_dns in global
+        #udp_check_dns: 'dns.google.com:53,8.8.8.8,2001:4860:4860::8888'
+        # Override check_interval in global
+        #check_interval: 30s
+        # Override check_tolerance in global
+        #check_tolerance: 50ms
+    }
 }
 
 # See https://github.com/daeuniverse/dae/blob/main/docs/en/configuration/routing.md for full examples.
@@ -238,6 +257,7 @@ routing {
     l4proto(udp) && dport(443) -> block
     dip(geoip:cn) -> direct
     domain(geosite:cn) -> direct
+    
 
     fallback: my_group
 }