feat(metrics): add endpoint server and phase1 gauge collectors

Author: MaurUppi

cmd/run.go

@@ -26,8 +26,6 @@ import (
 	"github.com/daeuniverse/outbound/protocol/direct"
 	"gopkg.in/natefinch/lumberjack.v2"
 
-	_ "net/http/pprof"
-
 	"github.com/daeuniverse/dae/cmd/internal"
 	"github.com/daeuniverse/dae/common"
 	"github.com/daeuniverse/dae/common/consts"
@@ -37,6 +35,7 @@ import (
 	"github.com/daeuniverse/dae/control"
 	"github.com/daeuniverse/dae/pkg/config_parser"
 	"github.com/daeuniverse/dae/pkg/logger"
+	"github.com/daeuniverse/dae/pkg/metrics"
 	"github.com/mohae/deepcopy"
 	"github.com/okzk/sdnotify"
 	"github.com/sirupsen/logrus"
@@ -134,12 +133,25 @@ func Run(log *logrus.Logger, conf *config.Config, externGeoDataDirs []string) (e
 		return err
 	}
 
-	var pprofServer *http.Server
-	if conf.Global.PprofPort != 0 {
-		pprofAddr := fmt.Sprintf("localhost:%d", conf.Global.PprofPort)
-		pprofServer = &http.Server{Addr: pprofAddr, Handler: nil}
-		go pprofServer.ListenAndServe()
+	metricsState := metrics.NewState()
+	metricsState.SetControlPlane(c)
+	metricsRegistry := metrics.NewRegistry(metricsState)
+
+	var endpointServer *http.Server
+	startEndpointServer := func(cfg metrics.EndpointConfig) {
+		if cfg.ListenAddress == "" {
+			endpointServer = nil
+			return
+		}
+		endpointServer = metrics.NewEndpointServer(cfg, metricsRegistry)
+		go func(server *http.Server, endpointCfg metrics.EndpointConfig) {
+			if e := metrics.StartEndpointServer(server, endpointCfg); e != nil && !errors.Is(e, http.ErrServerClosed) {
+				log.WithError(e).Errorln("Endpoint server stopped with error")
+			}
+		}(endpointServer, cfg)
 	}
+	endpointCfg := endpointConfigFromGlobal(conf, log)
+	startEndpointServer(endpointCfg)
 
 	// Serve tproxy TCP/UDP server util signals.
 	var listener *control.Listener
@@ -259,6 +271,10 @@ loop:
 				// Only keep dns cache when ip version preference not change.
 				dnsCache = c.CloneDnsCache()
 			}
+			// Stop old DNS listener before creating new one to avoid port conflicts
+			if err := c.StopDNSListener(); err != nil {
+				log.Warnf("[Reload] Failed to stop old DNS listener: %v", err)
+			}
 			log.Warnln("[Reload] Load new control plane")
 			newC, err := newControlPlane(log, obj, dnsCache, newConf, externGeoDataDirs)
 			if err != nil {
@@ -290,21 +306,21 @@ loop:
 			c = newC
 			conf = newConf
 			reloading = true
+			metricsState.SetControlPlane(newC)
 
 			// Ready to close.
 			if abortConnections {
 				oldC.AbortConnections()
 			}
 			oldC.Close()
 
-			if pprofServer != nil {
-				pprofServer.Shutdown(context.Background())
-				pprofServer = nil
-			}
-			if newConf.Global.PprofPort != 0 {
-				pprofAddr := fmt.Sprintf("localhost:%d", conf.Global.PprofPort)
-				pprofServer = &http.Server{Addr: pprofAddr, Handler: nil}
-				go pprofServer.ListenAndServe()
+			newEndpointCfg := endpointConfigFromGlobal(newConf, log)
+			if endpointConfigChanged(endpointCfg, newEndpointCfg) {
+				if endpointServer != nil {
+					_ = endpointServer.Shutdown(context.Background())
+				}
+				endpointCfg = newEndpointCfg
+				startEndpointServer(endpointCfg)
 			}
 		case syscall.SIGHUP:
 			// Ignore.
@@ -316,12 +332,37 @@ loop:
 	}
 	defer os.Remove(PidFilePath)
 	defer control.GetDaeNetns().Close()
+	if endpointServer != nil {
+		_ = endpointServer.Shutdown(context.Background())
+	}
 	if e := c.Close(); e != nil {
 		return fmt.Errorf("close control plane: %w", e)
 	}
 	return nil
 }
 
+func endpointConfigFromGlobal(conf *config.Config, log *logrus.Logger) metrics.EndpointConfig {
+	cfg := metrics.EndpointConfig{
+		ListenAddress:     conf.Global.EndpointListenAddress,
+		Username:          conf.Global.EndpointUsername,
+		Password:          conf.Global.EndpointPassword,
+		TlsCertificate:    conf.Global.EndpointTlsCertificate,
+		TlsKey:            conf.Global.EndpointTlsKey,
+		PrometheusEnabled: conf.Global.EndpointPrometheusEnabled,
+		PrometheusPath:    conf.Global.EndpointPrometheusPath,
+		PprofEnabled:      conf.Global.PprofPort != 0,
+	}
+	if cfg.ListenAddress == "" && conf.Global.PprofPort != 0 {
+		log.Warnln("pprof_port is deprecated, please use endpoint_listen_address instead")
+		cfg.ListenAddress = fmt.Sprintf("localhost:%d", conf.Global.PprofPort)
+	}
+	return cfg
+}
+
+func endpointConfigChanged(a, b metrics.EndpointConfig) bool {
+	return a != b
+}
+
 func newControlPlane(log *logrus.Logger, bpf interface{}, dnsCache map[string]*control.DnsCache, conf *config.Config, externGeoDataDirs []string) (c *control.ControlPlane, err error) {
 	// Deep copy to prevent modification.
 	conf = deepcopy.Copy(conf).(*config.Config)

component/outbound/dialer/alive_dialer_set.go

@@ -102,6 +102,12 @@ func (a *AliveDialerSet) GetRand() *Dialer {
 	return a.inorderedAliveDialerSet[ind]
 }
 
+func (a *AliveDialerSet) AliveCount() int {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return len(a.inorderedAliveDialerSet)
+}
+
 func (a *AliveDialerSet) SortingLatency(d *Dialer) time.Duration {
 	return a.dialerToLatency[d] + a.dialerToLatencyOffset[d]
 }

component/outbound/dialer/connectivity_check.go

@@ -503,6 +503,18 @@ func (d *Dialer) MustGetLatencies10(typ *NetworkType) *LatenciesN {
 	return d.mustGetCollection(typ).Latencies10
 }
 
+// GetCollectionState returns a snapshot of the dialer's health state for the given network type.
+func (d *Dialer) GetCollectionState(typ *NetworkType) (alive bool, lastLatency, avg10, movingAvg time.Duration) {
+	d.collectionFineMu.Lock()
+	col := d.mustGetCollection(typ)
+	alive = col.Alive
+	movingAvg = col.MovingAverage
+	d.collectionFineMu.Unlock()
+	lastLatency, _ = col.Latencies10.LastLatency()
+	avg10, _ = col.Latencies10.AvgLatency()
+	return
+}
+
 // RegisterAliveDialerSet is thread-safe.
 func (d *Dialer) RegisterAliveDialerSet(a *AliveDialerSet) {
 	if a == nil {

component/outbound/dialer_group.go

@@ -183,6 +183,17 @@ func (g *DialerGroup) GetSelectionPolicy() (policy consts.DialerSelectionPolicy)
 	return g.selectionPolicy.Policy
 }
 
+func (g *DialerGroup) AliveDialerSets() [6]*dialer.AliveDialerSet {
+	return g.aliveDialerSets
+}
+
+func (g *DialerGroup) SelectionPolicyName() string {
+	if g.selectionPolicy == nil {
+		return ""
+	}
+	return string(g.selectionPolicy.Policy)
+}
+
 func (d *DialerGroup) MustGetAliveDialerSet(typ *dialer.NetworkType) *dialer.AliveDialerSet {
 	if typ.IsDns {
 		switch typ.L4Proto {

config/config.go

@@ -46,13 +46,25 @@ type Global struct {
 	SniffingTimeout        time.Duration    `mapstructure:"sniffing_timeout" default:"100ms"`
 	TlsImplementation      string           `mapstructure:"tls_implementation" default:"tls"`
 	UtlsImitate            string           `mapstructure:"utls_imitate" default:"chrome_auto"`
+	TlsFragment            bool             `mapstructure:"tls_fragment" default:"false"`
+	TlsFragmentLength      string           `mapstructure:"tls_fragment_length" default:"50-100"`
+	TlsFragmentInterval    string           `mapstructure:"tls_fragment_interval" default:"10-20"`
 	PprofPort              uint16           `mapstructure:"pprof_port" default:"0"`
 	Mptcp                  bool             `mapstructure:"mptcp" default:"false"`
 	FallbackResolver       string           `mapstructure:"fallback_resolver" default:"8.8.8.8:53"`
 	BandwidthMaxTx         string           `mapstructure:"bandwidth_max_tx" default:"0"`
 	BandwidthMaxRx         string           `mapstructure:"bandwidth_max_rx" default:"0"`
+	UDPHopInterval         time.Duration    `mapstructure:"udphop_interval" default:"30s"`
 	DnsPerformanceLevel    string           `mapstructure:"dns_performance_level" default:"balanced"`
 	DnsIngressManual       DnsIngressManual `mapstructure:"dns_ingress_manual"`
+	// Endpoint (metrics + diagnostics)
+	EndpointListenAddress     string `mapstructure:"endpoint_listen_address" default:""`
+	EndpointUsername          string `mapstructure:"endpoint_username" default:""`
+	EndpointPassword          string `mapstructure:"endpoint_password" default:""`
+	EndpointTlsCertificate    string `mapstructure:"endpoint_tls_certificate" default:""`
+	EndpointTlsKey            string `mapstructure:"endpoint_tls_key" default:""`
+	EndpointPrometheusEnabled bool   `mapstructure:"endpoint_prometheus_enabled" default:"false"`
+	EndpointPrometheusPath    string `mapstructure:"endpoint_prometheus_path" default:"/metrics"`
 }
 
 type Utls struct {
@@ -121,10 +133,14 @@ type DnsRouting struct {
 }
 type KeyableString string
 type Dns struct {
-	IpVersionPrefer int             `mapstructure:"ipversion_prefer"`
-	FixedDomainTtl  []KeyableString `mapstructure:"fixed_domain_ttl"`
-	Upstream        []KeyableString `mapstructure:"upstream"`
-	Routing         DnsRouting      `mapstructure:"routing"`
+	IpVersionPrefer    int             `mapstructure:"ipversion_prefer"`
+	FixedDomainTtl     []KeyableString `mapstructure:"fixed_domain_ttl"`
+	Upstream           []KeyableString `mapstructure:"upstream"`
+	Routing            DnsRouting      `mapstructure:"routing"`
+	Bind               string          `mapstructure:"bind"`
+	OptimisticCache    bool            `mapstructure:"optimistic_cache" default:"true"`
+	OptimisticCacheTtl int             `mapstructure:"optimistic_cache_ttl" default:"60"`
+	MaxCacheSize       int             `mapstructure:"max_cache_size" default:"0"`
 }
 
 type Routing struct {

control/control_plane.go

@@ -616,6 +616,23 @@ func (c *ControlPlane) InjectBpf(bpf *bpfObjects) {
 	c.core.InjectBpf(bpf)
 }
 
+func (c *ControlPlane) Outbounds() []*outbound.DialerGroup {
+	return c.outbounds
+}
+
+func (c *ControlPlane) GetDnsController() *DnsController {
+	return c.dnsController
+}
+
+func (c *ControlPlane) CountTcpConnections() int {
+	count := 0
+	c.inConnections.Range(func(_, _ interface{}) bool {
+		count++
+		return true
+	})
+	return count
+}
+
 func (c *ControlPlane) CloneDnsCache() map[string]*DnsCache {
 	c.dnsController.dnsCacheMu.Lock()
 	defer c.dnsController.dnsCacheMu.Unlock()

control/dns_control.go

@@ -132,6 +132,27 @@ func (c *DnsController) cacheKey(qname string, qtype uint16) string {
 	return dnsmessage.CanonicalName(qname) + strconv.Itoa(int(qtype))
 }
 
+func (c *DnsController) CacheSize() int {
+	c.dnsCacheMu.Lock()
+	defer c.dnsCacheMu.Unlock()
+	return len(c.dnsCache)
+}
+
+// ConcurrencyInfo returns in-flight request count and limit.
+// This implementation uses handling map size as in-flight and has no hard limit.
+func (c *DnsController) ConcurrencyInfo() (inUse, limit int) {
+	c.handling.Range(func(_, _ interface{}) bool {
+		inUse++
+		return true
+	})
+	return inUse, 0
+}
+
+// ForwarderCacheInfo is empty on this branch because forwarders are not cached.
+func (c *DnsController) ForwarderCacheInfo() (count int, inFlightByUpstream map[string]int32) {
+	return 0, map[string]int32{}
+}
+
 func (c *DnsController) RemoveDnsRespCache(cacheKey string) {
 	c.dnsCacheMu.Lock()
 	_, ok := c.dnsCache[cacheKey]

control/udp_endpoint_pool.go

@@ -138,6 +138,15 @@ func (p *UdpEndpointPool) Get(lAddr netip.AddrPort) (udpEndpoint *UdpEndpoint, o
 	return _ue.(*UdpEndpoint), ok
 }
 
+func (p *UdpEndpointPool) Count() int {
+	count := 0
+	p.pool.Range(func(_, _ interface{}) bool {
+		count++
+		return true
+	})
+	return count
+}
+
 func (p *UdpEndpointPool) GetOrCreate(lAddr netip.AddrPort, createOption *UdpEndpointOptions) (udpEndpoint *UdpEndpoint, isNew bool, err error) {
 	_ue, ok := p.pool.Load(lAddr)
 begin:

control/udp_task_pool.go

@@ -57,6 +57,12 @@ func NewUdpTaskPool() *UdpTaskPool {
 	return p
 }
 
+func (p *UdpTaskPool) Count() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.m)
+}
+
 // EmitTask: Make sure packets with the same key (4 tuples) will be sent in order.
 func (p *UdpTaskPool) EmitTask(key string, task UdpTask) {
 	p.mu.Lock()

example.dae

@@ -12,6 +12,16 @@ global {
     # Set non-zero value to enable pprof.
     pprof_port: 0
 
+    # Endpoint configuration for metrics and diagnostics.
+    # Set a listen address to enable the endpoint server (disabled by default).
+    #endpoint_listen_address: '0.0.0.0:5556'
+    #endpoint_username: ''
+    #endpoint_password: ''
+    #endpoint_tls_certificate: ''
+    #endpoint_tls_key: ''
+    #endpoint_prometheus_enabled: true
+    #endpoint_prometheus_path: /metrics
+
     # DNS ingress performance level.
     # Options: lean, balanced (default), aggressive, manual.
     # - lean: for low-power/embedded devices (32 workers)