Rebase metrics endpoint onto upstream runtime metrics

Preserve the new runtime traffic and latency sources from upstream main, then layer the old endpoint/Prometheus surface back on top with portable local tests for the HTTP/auth/config slices.

Constraint: Local macOS environment cannot perform Linux/eBPF package builds or generated-bpf compilation
Rejected: Re-apply old metrics branch wholesale | it deleted upstream runtime_stats and latency_probe
Rejected: Keep endpoint HTTP helpers inside pkg/metrics | prevented portable local tests because collectors pull in Linux-only control dependencies
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: Keep PR#968 runtime sources as the single source of truth and extend Prometheus collectors around them instead of duplicating runtime logic
Tested: go test ./common
Tested: go test ./pkg/metricshttp
Tested: (cd cmd && go test endpoint_config.go run_test.go)
Not-tested: Full ./control ./pkg/metrics ./cmd build on Linux with generated eBPF artifacts
Not-tested: GitHub Actions kernel/eBPF workflows

Author: Codex

cmd/endpoint_config.go

@@ -0,0 +1,74 @@
+/*
+ * SPDX-License-Identifier: AGPL-3.0-only
+ * Copyright (c) 2022-2025, daeuniverse Organization <dae@v2raya.org>
+ */
+
+package cmd
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/daeuniverse/dae/common"
+	"github.com/daeuniverse/dae/config"
+	"github.com/daeuniverse/dae/pkg/metricshttp"
+	"github.com/sirupsen/logrus"
+)
+
+func endpointConfigFromGlobal(conf *config.Config, log *logrus.Logger) metricshttp.EndpointConfig {
+	cfg := metricshttp.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 metricshttp.EndpointConfig) bool {
+	return a != b
+}
+
+func validateEndpointTLSFiles(cfg metricshttp.EndpointConfig) error {
+	if cfg.TlsCertificate == "" && cfg.TlsKey == "" {
+		return nil
+	}
+	if cfg.TlsCertificate == "" || cfg.TlsKey == "" {
+		return fmt.Errorf("endpoint_tls_certificate and endpoint_tls_key must be configured together")
+	}
+
+	certFile, err := os.Open(cfg.TlsCertificate)
+	if err != nil {
+		return fmt.Errorf("cannot open endpoint_tls_certificate '%s': %w", cfg.TlsCertificate, err)
+	}
+	certFi, err := certFile.Stat()
+	certFile.Close()
+	if err != nil {
+		return fmt.Errorf("cannot stat endpoint_tls_certificate '%s': %w", cfg.TlsCertificate, err)
+	}
+	if err = common.ValidateFilePermissionAllowed(cfg.TlsCertificate, certFi, 0o640, 0o644); err != nil {
+		return fmt.Errorf("invalid endpoint_tls_certificate: %w", err)
+	}
+
+	keyFile, err := os.Open(cfg.TlsKey)
+	if err != nil {
+		return fmt.Errorf("cannot open endpoint_tls_key '%s': %w", cfg.TlsKey, err)
+	}
+	keyFi, err := keyFile.Stat()
+	keyFile.Close()
+	if err != nil {
+		return fmt.Errorf("cannot stat endpoint_tls_key '%s': %w", cfg.TlsKey, err)
+	}
+	if err = common.ValidateFilePermissionAllowed(cfg.TlsKey, keyFi, 0o600); err != nil {
+		return fmt.Errorf("invalid endpoint_tls_key: %w", err)
+	}
+	return nil
+}

cmd/run.go

@@ -37,6 +37,8 @@ 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/daeuniverse/dae/pkg/metricshttp"
 	"github.com/mohae/deepcopy"
 	"github.com/okzk/sdnotify"
 	"github.com/sirupsen/logrus"
@@ -128,18 +130,35 @@ func Run(log *logrus.Logger, conf *config.Config, externGeoDataDirs []string) (e
 	// Remove AbortFile at beginning.
 	_ = os.Remove(AbortFile)
 
+	endpointCfg := endpointConfigFromGlobal(conf, log)
+	if err = validateEndpointTLSFiles(endpointCfg); err != nil {
+		return fmt.Errorf("invalid endpoint tls config: %w", err)
+	}
+
 	// New ControlPlane.
 	c, err := newControlPlane(log, nil, nil, conf, externGeoDataDirs)
 	if err != nil {
 		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 metricshttp.EndpointConfig) {
+		if cfg.ListenAddress == "" {
+			endpointServer = nil
+			return
+		}
+		endpointServer = metricshttp.NewEndpointServer(cfg, metricsRegistry)
+		go func(server *http.Server, endpointCfg metricshttp.EndpointConfig) {
+			if e := metricshttp.StartEndpointServer(server, endpointCfg); e != nil && !errors.Is(e, http.ErrServerClosed) {
+				log.WithError(e).Errorln("Endpoint server stopped with error")
+			}
+		}(endpointServer, cfg)
 	}
+	startEndpointServer(endpointCfg)
 
 	// Serve tproxy TCP/UDP server util signals.
 	var listener *control.Listener
@@ -252,6 +271,16 @@ loop:
 			log.SetOutput(oldLogOutput) // FIXME: THIS IS A HACK.
 			logrus.SetOutput(oldLogOutput)
 
+			newEndpointCfg := endpointConfigFromGlobal(newConf, log)
+			if err = validateEndpointTLSFiles(newEndpointCfg); err != nil {
+				log.WithFields(logrus.Fields{
+					"err": err,
+				}).Errorln("[Reload] Failed to reload")
+				sdnotify.Ready()
+				_ = os.WriteFile(SignalProgressFilePath, append([]byte{consts.ReloadError}, []byte("\n"+err.Error())...), 0644)
+				continue
+			}
+
 			// New control plane.
 			obj := c.EjectBpf()
 			var dnsCache map[string]*control.DnsCache
@@ -263,7 +292,7 @@ loop:
 			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 {
@@ -295,21 +324,20 @@ 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()
+			if endpointConfigChanged(endpointCfg, newEndpointCfg) {
+				if endpointServer != nil {
+					_ = endpointServer.Shutdown(context.Background())
+				}
+				endpointCfg = newEndpointCfg
+				startEndpointServer(endpointCfg)
 			}
 		case syscall.SIGHUP:
 			// Ignore.
@@ -321,6 +349,9 @@ 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)
 	}

cmd/run_test.go

@@ -0,0 +1,102 @@
+/*
+ * SPDX-License-Identifier: AGPL-3.0-only
+ * Copyright (c) 2022-2025, daeuniverse Organization <dae@v2raya.org>
+ */
+
+package cmd
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/daeuniverse/dae/config"
+	"github.com/daeuniverse/dae/pkg/metricshttp"
+	"github.com/sirupsen/logrus"
+)
+
+func writeEndpointFile(t *testing.T, name string, mode os.FileMode) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), name)
+	if err := os.WriteFile(path, []byte("x"), 0o600); err != nil {
+		t.Fatalf("write endpoint file: %v", err)
+	}
+	if err := os.Chmod(path, mode); err != nil {
+		t.Fatalf("chmod endpoint file: %v", err)
+	}
+	return path
+}
+
+func TestEndpointConfigFromGlobalUsesEndpointSettings(t *testing.T) {
+	conf := &config.Config{}
+	conf.Global.EndpointListenAddress = "127.0.0.1:5556"
+	conf.Global.EndpointUsername = "dae"
+	conf.Global.EndpointPassword = "secret"
+	conf.Global.EndpointTlsCertificate = "/tmp/cert.pem"
+	conf.Global.EndpointTlsKey = "/tmp/key.pem"
+	conf.Global.EndpointPrometheusEnabled = true
+	conf.Global.EndpointPrometheusPath = "/custom-metrics"
+	conf.Global.PprofPort = 0
+
+	cfg := endpointConfigFromGlobal(conf, logrus.New())
+	if cfg.ListenAddress != "127.0.0.1:5556" {
+		t.Fatalf("unexpected listen address: got=%q", cfg.ListenAddress)
+	}
+	if cfg.Username != "dae" || cfg.Password != "secret" {
+		t.Fatalf("unexpected credentials: got=%q/%q", cfg.Username, cfg.Password)
+	}
+	if !cfg.PrometheusEnabled || cfg.PrometheusPath != "/custom-metrics" {
+		t.Fatalf("unexpected prometheus settings: enabled=%v path=%q", cfg.PrometheusEnabled, cfg.PrometheusPath)
+	}
+	if cfg.PprofEnabled {
+		t.Fatal("pprof should be disabled when pprof_port is zero")
+	}
+}
+
+func TestEndpointConfigFromGlobalFallsBackToPprofPort(t *testing.T) {
+	conf := &config.Config{}
+	conf.Global.PprofPort = 6060
+
+	cfg := endpointConfigFromGlobal(conf, logrus.New())
+	if cfg.ListenAddress != "localhost:6060" {
+		t.Fatalf("unexpected pprof fallback address: got=%q want=%q", cfg.ListenAddress, "localhost:6060")
+	}
+	if !cfg.PprofEnabled {
+		t.Fatal("pprof should be enabled when pprof_port is non-zero")
+	}
+}
+
+func TestValidateEndpointTLSFilesRequiresPair(t *testing.T) {
+	err := validateEndpointTLSFiles(endpointConfigFromGlobal(&config.Config{}, logrus.New()))
+	if err != nil {
+		t.Fatalf("empty endpoint tls config should pass: %v", err)
+	}
+
+	err = validateEndpointTLSFiles(metricshttp.EndpointConfig{
+		TlsCertificate: "/tmp/cert.pem",
+	})
+	if err == nil {
+		t.Fatal("expected certificate-only config to fail")
+	}
+}
+
+func TestValidateEndpointTLSFilesChecksPermissions(t *testing.T) {
+	cert := writeEndpointFile(t, "cert.pem", 0o640)
+	key := writeEndpointFile(t, "key.pem", 0o600)
+
+	if err := validateEndpointTLSFiles(metricshttp.EndpointConfig{
+		TlsCertificate: cert,
+		TlsKey:         key,
+	}); err != nil {
+		t.Fatalf("expected valid certificate/key permissions to pass: %v", err)
+	}
+
+	tooOpenCert := writeEndpointFile(t, "cert-open.pem", 0o666)
+	err := validateEndpointTLSFiles(metricshttp.EndpointConfig{
+		TlsCertificate: tooOpenCert,
+		TlsKey:         key,
+	})
+	if err == nil {
+		t.Fatal("expected too-open certificate permissions to fail")
+	}
+}

common/file_permission.go

@@ -0,0 +1,44 @@
+/*
+ * SPDX-License-Identifier: AGPL-3.0-only
+ * Copyright (c) 2022-2025, daeuniverse Organization <dae@v2raya.org>
+ */
+
+package common
+
+import (
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+)
+
+func ValidateFilePermissionNotTooOpen(path string, fi os.FileInfo) error {
+	if fi.IsDir() {
+		return fmt.Errorf("cannot read a directory: %v", path)
+	}
+	if fi.Mode()&0o037 > 0 {
+		return fmt.Errorf("permissions %04o for '%v' are too open; requires the file is NOT writable by the same group and NOT accessible by others; suggest 0640 or 0600", fi.Mode()&0o777, path)
+	}
+	return nil
+}
+
+func ValidateFilePermissionAllowed(path string, fi os.FileInfo, allowedModes ...os.FileMode) error {
+	if fi.IsDir() {
+		return fmt.Errorf("cannot read a directory: %v", path)
+	}
+	perm := fi.Mode().Perm()
+	for _, mode := range allowedModes {
+		if perm == mode.Perm() {
+			return nil
+		}
+	}
+	if len(allowedModes) == 0 {
+		return fmt.Errorf("permissions %04o for '%v' are invalid", perm, path)
+	}
+	allowed := make([]string, 0, len(allowedModes))
+	for _, mode := range allowedModes {
+		allowed = append(allowed, fmt.Sprintf("%04o", mode.Perm()))
+	}
+	sort.Strings(allowed)
+	return fmt.Errorf("permissions %04o for '%v' are invalid; allowed: %s", perm, path, strings.Join(allowed, ", "))
+}