feat: metrics including pprof, expvars

Author: yichya

app/metrics/config.pb.go

@@ -0,0 +1,13 @@
+syntax = "proto3";
+
+package xray.app.metrics;
+option csharp_namespace = "Xray.App.Metrics";
+option go_package = "github.com/xtls/xray-core/app/metrics";
+option java_package = "com.xray.app.metrics";
+option java_multiple_files = true;
+
+// Config is the settings for metrics.
+message Config {
+  // Tag of the outbound handler that handles metrics http connections.
+  string tag = 1;
+}

app/metrics/config.proto

@@ -0,0 +1,9 @@
+package metrics
+
+import "github.com/xtls/xray-core/common/errors"
+
+type errPathObjHolder struct{}
+
+func newError(values ...interface{}) *errors.Error {
+	return errors.New(values...).WithPathObj(errPathObjHolder{})
+}

app/metrics/errors.generated.go

@@ -0,0 +1,115 @@
+package metrics
+
+import (
+	"context"
+	"expvar"
+	"net/http"
+	_ "net/http/pprof"
+	"strings"
+
+	"github.com/xtls/xray-core/app/observatory"
+	"github.com/xtls/xray-core/app/stats"
+	"github.com/xtls/xray-core/common"
+	"github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/common/signal/done"
+	"github.com/xtls/xray-core/core"
+	"github.com/xtls/xray-core/features/extension"
+	"github.com/xtls/xray-core/features/outbound"
+	feature_stats "github.com/xtls/xray-core/features/stats"
+)
+
+type MetricsHandler struct {
+	ohm          outbound.Manager
+	statsManager feature_stats.Manager
+	observatory  extension.Observatory
+	tag          string
+}
+
+// NewMetricsHandler creates a new MetricsHandler based on the given config.
+func NewMetricsHandler(ctx context.Context, config *Config) (*MetricsHandler, error) {
+	c := &MetricsHandler{
+		tag: config.Tag,
+	}
+	common.Must(core.RequireFeatures(ctx, func(om outbound.Manager, sm feature_stats.Manager) {
+		c.statsManager = sm
+		c.ohm = om
+	}))
+	expvar.Publish("stats", expvar.Func(func() interface{} {
+		manager, ok := c.statsManager.(*stats.Manager)
+		if !ok {
+			return nil
+		}
+		var resp = map[string]map[string]map[string]int64{
+			"inbound":  {},
+			"outbound": {},
+			"user":     {},
+		}
+		manager.VisitCounters(func(name string, counter feature_stats.Counter) bool {
+			nameSplit := strings.Split(name, ">>>")
+			typeName, tagOrUser, direction := nameSplit[0], nameSplit[1], nameSplit[3]
+			if item, found := resp[typeName][tagOrUser]; found {
+				item[direction] = counter.Value()
+			} else {
+				resp[typeName][tagOrUser] = map[string]int64{
+					direction: counter.Value(),
+				}
+			}
+			return true
+		})
+		return resp
+	}))
+	expvar.Publish("observatory", expvar.Func(func() interface{} {
+		if c.observatory == nil {
+			common.Must(core.RequireFeatures(ctx, func(observatory extension.Observatory) error {
+				c.observatory = observatory
+				return nil
+			}))
+		}
+		var resp = map[string]*observatory.OutboundStatus{}
+		if o, err := c.observatory.GetObservation(context.Background()); err != nil {
+			return err
+		} else {
+			for _, x := range o.(*observatory.ObservationResult).GetStatus() {
+				resp[x.OutboundTag] = x
+			}
+		}
+		return resp
+	}))
+	return c, nil
+}
+
+func (p *MetricsHandler) Type() interface{} {
+	return (*MetricsHandler)(nil)
+}
+
+func (p *MetricsHandler) Start() error {
+	listener := &OutboundListener{
+		buffer: make(chan net.Conn, 4),
+		done:   done.New(),
+	}
+
+	go func() {
+		if err := http.Serve(listener, http.DefaultServeMux); err != nil {
+			newError("failed to start metrics server").Base(err).AtError().WriteToLog()
+		}
+	}()
+
+	if err := p.ohm.RemoveHandler(context.Background(), p.tag); err != nil {
+		newError("failed to remove existing handler").WriteToLog()
+	}
+
+	return p.ohm.AddHandler(context.Background(), &Outbound{
+		tag:      p.tag,
+		listener: listener,
+	})
+}
+
+func (p *MetricsHandler) Close() error {
+	return nil
+}
+
+func init() {
+	common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, cfg interface{}) (interface{}, error) {
+		return NewMetricsHandler(ctx, cfg.(*Config))
+	}))
+}

app/metrics/metrics.go

@@ -0,0 +1,109 @@
+package metrics
+
+import (
+	"context"
+	"sync"
+
+	"github.com/xtls/xray-core/common"
+	"github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/common/net/cnc"
+	"github.com/xtls/xray-core/common/signal/done"
+	"github.com/xtls/xray-core/transport"
+)
+
+// OutboundListener is a net.Listener for listening pprof http connections.
+type OutboundListener struct {
+	buffer chan net.Conn
+	done   *done.Instance
+}
+
+func (l *OutboundListener) add(conn net.Conn) {
+	select {
+	case l.buffer <- conn:
+	case <-l.done.Wait():
+		conn.Close()
+	default:
+		conn.Close()
+	}
+}
+
+// Accept implements net.Listener.
+func (l *OutboundListener) Accept() (net.Conn, error) {
+	select {
+	case <-l.done.Wait():
+		return nil, newError("listen closed")
+	case c := <-l.buffer:
+		return c, nil
+	}
+}
+
+// Close implement net.Listener.
+func (l *OutboundListener) Close() error {
+	common.Must(l.done.Close())
+L:
+	for {
+		select {
+		case c := <-l.buffer:
+			c.Close()
+		default:
+			break L
+		}
+	}
+	return nil
+}
+
+// Addr implements net.Listener.
+func (l *OutboundListener) Addr() net.Addr {
+	return &net.TCPAddr{
+		IP:   net.IP{0, 0, 0, 0},
+		Port: 0,
+	}
+}
+
+// Outbound is a outbound.Handler that handles pprof http connections.
+type Outbound struct {
+	tag      string
+	listener *OutboundListener
+	access   sync.RWMutex
+	closed   bool
+}
+
+// Dispatch implements outbound.Handler.
+func (co *Outbound) Dispatch(ctx context.Context, link *transport.Link) {
+	co.access.RLock()
+
+	if co.closed {
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
+		co.access.RUnlock()
+		return
+	}
+
+	closeSignal := done.New()
+	c := cnc.NewConnection(cnc.ConnectionInputMulti(link.Writer), cnc.ConnectionOutputMulti(link.Reader), cnc.ConnectionOnClose(closeSignal))
+	co.listener.add(c)
+	co.access.RUnlock()
+	<-closeSignal.Wait()
+}
+
+// Tag implements outbound.Handler.
+func (co *Outbound) Tag() string {
+	return co.tag
+}
+
+// Start implements common.Runnable.
+func (co *Outbound) Start() error {
+	co.access.Lock()
+	co.closed = false
+	co.access.Unlock()
+	return nil
+}
+
+// Close implements common.Closable.
+func (co *Outbound) Close() error {
+	co.access.Lock()
+	defer co.access.Unlock()
+
+	co.closed = true
+	return co.listener.Close()
+}