design-15 S7a: read-only sing-box stats collector (ADR-004)

README.md

@@ -305,6 +305,9 @@ missing checksum manifest aborts the install before the binary is written.
   with its control-plane `line_uuid` and declared downstream chain edge.
 - `LATTICE_PROXY_USAGE_FILE`, `LATTICE_PROXY_USAGE_URL`, and
   `LATTICE_PROXY_USAGE_XRAY_API` configure proxy usage reporting sources.
+- `LATTICE_SINGBOX_STATS_API=127.0.0.1:8080` enables the sing-box stats
+  collector (read-only loopback gRPC against the core's experimental API;
+  vendored proto, ADR-004).
 
 If a task-backed dashboard action reports `agent task execution disabled`, rerun
 the node detail page's generated reconfigure command with `allow_exec=true`

cmd/lattice-agent/main.go

@@ -136,6 +136,7 @@ type agentConfig struct {
 	ProxyUsageXrayAPI     string
 	ProxyUsageXrayBin     string
 	ProxyUsageXrayPattern string
+	SingBoxStatsAPI       string
 	SingBoxDiscover       bool
 	SingBoxBin            string
 	SingBoxMeta           string
@@ -159,6 +160,7 @@ type agentRuntimePayload struct {
 	ProxyUsageXrayAPI     string    `json:"proxy_usage_xray_api,omitempty"`
 	ProxyUsageXrayBin     string    `json:"proxy_usage_xray_bin,omitempty"`
 	ProxyUsageXrayPattern string    `json:"proxy_usage_xray_pattern,omitempty"`
+	SingBoxStatsAPI       string    `json:"singbox_stats_api,omitempty"`
 	ReportedAt            time.Time `json:"reported_at,omitempty"`
 }
 
@@ -222,6 +224,7 @@ func main() {
 	flag.StringVar(&cfg.ProxyUsageSecretFile, "proxy-usage-secret-file", os.Getenv("LATTICE_PROXY_USAGE_SECRET_FILE"), "optional file containing bearer secret for -proxy-usage-url")
 	flag.DurationVar(&cfg.ProxyUsageTimeout, "proxy-usage-timeout", envDuration("LATTICE_PROXY_USAGE_TIMEOUT", 3*time.Second), "timeout for -proxy-usage-url and -proxy-usage-xray-api")
 	flag.StringVar(&cfg.ProxyUsageXrayAPI, "proxy-usage-xray-api", os.Getenv("LATTICE_PROXY_USAGE_XRAY_API"), "optional loopback host:port of the xray API inbound; the agent runs `xray api statsquery` against it each interval (no new dependency; see ADR-003)")
+	flag.StringVar(&cfg.SingBoxStatsAPI, "singbox-stats-api", os.Getenv("LATTICE_SINGBOX_STATS_API"), "optional loopback host:port of the sing-box experimental stats API; the agent issues read-only gRPC QueryStats against it each interval (vendored proto; see ADR-004)")
 	flag.StringVar(&cfg.ProxyUsageXrayBin, "proxy-usage-xray-bin", os.Getenv("LATTICE_PROXY_USAGE_XRAY_BIN"), "xray binary for -proxy-usage-xray-api (default \"xray\" resolved on PATH)")
 	flag.StringVar(&cfg.ProxyUsageXrayPattern, "proxy-usage-xray-pattern", os.Getenv("LATTICE_PROXY_USAGE_XRAY_PATTERN"), "optional stat-name filter for -proxy-usage-xray-api (default \"user>>>\")")
 	flag.BoolVar(&cfg.SingBoxDiscover, "singbox-discover", os.Getenv("LATTICE_SINGBOX_DISCOVER") == "1", "report on-box sing-box nodes each interval by running read-only `sb --json list` (adoption bridge; read-only, no node mutation)")
@@ -629,6 +632,7 @@ func reportMetrics(cfg agentConfig) error {
 			ProxyUsageXrayAPI:     cfg.ProxyUsageXrayAPI,
 			ProxyUsageXrayBin:     cfg.ProxyUsageXrayBin,
 			ProxyUsageXrayPattern: cfg.ProxyUsageXrayPattern,
+			SingBoxStatsAPI:       cfg.SingBoxStatsAPI,
 			ReportedAt:            time.Now().UTC(),
 		},
 		"public_ip":     cfg.PublicIP,
@@ -810,8 +814,9 @@ func reportProxyUsage(cfg agentConfig) error {
 	hasFile := strings.TrimSpace(cfg.ProxyUsageFile) != ""
 	hasURL := strings.TrimSpace(cfg.ProxyUsageURL) != ""
 	hasXray := strings.TrimSpace(cfg.ProxyUsageXrayAPI) != ""
+	hasSingBox := strings.TrimSpace(cfg.SingBoxStatsAPI) != ""
 	configured := 0
-	for _, on := range []bool{hasFile, hasURL, hasXray} {
+	for _, on := range []bool{hasFile, hasURL, hasXray, hasSingBox} {
 		if on {
 			configured++
 		}
@@ -820,7 +825,7 @@ func reportProxyUsage(cfg agentConfig) error {
 		return nil
 	}
 	if configured > 1 {
-		return fmt.Errorf("configure only one of proxy-usage-file, proxy-usage-url, or proxy-usage-xray-api")
+		return fmt.Errorf("configure only one of proxy-usage-file, proxy-usage-url, proxy-usage-xray-api, or singbox-stats-api")
 	}
 	var (
 		snapshot model.ProxyUsageSnapshot
@@ -838,6 +843,12 @@ func reportProxyUsage(cfg agentConfig) error {
 			Secret:  cfg.ProxyUsageSecret,
 			Timeout: cfg.ProxyUsageTimeout,
 		}, cfg.NodeID)
+	case hasSingBox:
+		source = "singbox-stats"
+		snapshot, err = proxyusage.LoadSingBoxStats(context.Background(), proxyusage.SingBoxStatsSource{
+			APIAddr: cfg.SingBoxStatsAPI,
+			Timeout: cfg.ProxyUsageTimeout,
+		}, cfg.NodeID)
 	default:
 		source = "xray-cli"
 		snapshot, err = proxyusage.LoadXrayCLI(context.Background(), proxyusage.XrayCLISource{
@@ -931,15 +942,16 @@ func validateProxyUsageConfig(cfg agentConfig) error {
 	hasFile := strings.TrimSpace(cfg.ProxyUsageFile) != ""
 	hasURL := strings.TrimSpace(cfg.ProxyUsageURL) != ""
 	hasXray := strings.TrimSpace(cfg.ProxyUsageXrayAPI) != ""
+	hasSingBox := strings.TrimSpace(cfg.SingBoxStatsAPI) != ""
 	hasSecretFile := strings.TrimSpace(cfg.ProxyUsageSecretFile) != ""
 	configured := 0
-	for _, on := range []bool{hasFile, hasURL, hasXray} {
+	for _, on := range []bool{hasFile, hasURL, hasXray, hasSingBox} {
 		if on {
 			configured++
 		}
 	}
 	if configured > 1 {
-		return fmt.Errorf("configure only one of proxy-usage-file, proxy-usage-url, or proxy-usage-xray-api")
+		return fmt.Errorf("configure only one of proxy-usage-file, proxy-usage-url, proxy-usage-xray-api, or singbox-stats-api")
 	}
 	if strings.TrimSpace(cfg.ProxyUsageSecret) != "" && hasSecretFile {
 		return fmt.Errorf("configure either proxy-usage-secret or proxy-usage-secret-file, not both")
@@ -970,6 +982,13 @@ func validateProxyUsageConfig(cfg agentConfig) error {
 			return err
 		}
 	}
+	if hasSingBox {
+		if err := proxyusage.ValidateSingBoxStatsSource(proxyusage.SingBoxStatsSource{
+			APIAddr: cfg.SingBoxStatsAPI,
+		}); err != nil {
+			return err
+		}
+	}
 	if cfg.ProxyUsageTimeout < 0 {
 		return fmt.Errorf("proxy-usage-timeout cannot be negative")
 	}

cmd/lattice-agent/main_test.go

@@ -145,14 +145,19 @@ func TestReportMetricsUsesBearerAuthAndOmitsBodyToken(t *testing.T) {
 		if _, ok := body["host_facts"].(map[string]any); !ok {
 			return testResponse(http.StatusBadRequest, "missing host_facts"), nil
 		}
+		runtime, ok := body["agent_runtime"].(map[string]any)
+		if !ok || runtime["singbox_stats_api"] != "127.0.0.1:8080" {
+			return testResponse(http.StatusBadRequest, "missing singbox_stats_api"), nil
+		}
 		return testResponse(http.StatusOK, `{"ok":true}`), nil
 	})}
 
 	err := reportMetrics(agentConfig{
-		Server:   "http://lattice.test",
-		NodeID:   "node-a",
-		Token:    "node-secret",
-		Interval: time.Second,
+		Server:          "http://lattice.test",
+		NodeID:          "node-a",
+		Token:           "node-secret",
+		Interval:        time.Second,
+		SingBoxStatsAPI: "127.0.0.1:8080",
 	})
 	if err != nil {
 		t.Fatal(err)

go.mod

@@ -6,4 +6,15 @@ require github.com/LatticeNet/lattice-sdk v0.2.18-0.20260722123932-4a318f246d23
 
 require github.com/creack/pty v1.1.24
 
-require github.com/gorilla/websocket v1.5.3
+require (
+	github.com/gorilla/websocket v1.5.3
+	google.golang.org/grpc v1.72.2
+	google.golang.org/protobuf v1.36.6
+)
+
+require (
+	golang.org/x/net v0.35.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a // indirect
+)

go.sum

@@ -2,5 +2,39 @@ github.com/LatticeNet/lattice-sdk v0.2.18-0.20260722123932-4a318f246d23 h1:2qpbn
 github.com/LatticeNet/lattice-sdk v0.2.18-0.20260722123932-4a318f246d23/go.mod h1:7ENUQ4EoS/TSW/eNomCGfZGliUPJZ46uAvp7dVcEXoE=
 github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
 github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE=
+github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
+github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
+go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
+go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=
+go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI=
+go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ=
+go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE=
+go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A=
+go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU=
+go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce1EK0Gyvahk=
+go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w=
+go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k=
+go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
+golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8=
+golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a h1:51aaUVRocpvUOSQKM6Q7VuoaktNIaMCLuhZB6DKksq4=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a/go.mod h1:uRxBH1mhmO8PGhU89cMcHaXKZqO+OfakD8QQO0oYwlQ=
+google.golang.org/grpc v1.72.2 h1:TdbGzwb82ty4OusHWepvFWGLgIbNo1/SUynEN0ssqv8=
+google.golang.org/grpc v1.72.2/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM=
+google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
+google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=

internal/proxyusage/singbox.go

@@ -0,0 +1,139 @@
+package proxyusage
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/LatticeNet/lattice-node-agent/internal/proxyusage/singboxstats"
+	"github.com/LatticeNet/lattice-sdk/model"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+const (
+	defaultSingBoxStatsPattern = "user>>>"
+	defaultSingBoxStatsTimeout = 5 * time.Second
+)
+
+// SingBoxStatsSource collects per-user usage from sing-box's experimental
+// V2Ray Stats API over loopback gRPC (ADR-004). sing-box has no CLI stats
+// subcommand, so — unlike the xray path (ADR-003) — the gRPC call happens in
+// the agent itself, using the vendored proto in singboxstats (byte-identical
+// service/messages with upstream, only go_package adjusted).
+//
+// Queries are read-only (never reset): counters stay monotonic and the server
+// keeps ownership of successive-snapshot diffing, eligibility, and audit. The
+// API address must be loopback, mirroring the HTTP source's loopback rule —
+// a node must not be configured into dialing arbitrary networks.
+type SingBoxStatsSource struct {
+	APIAddr string        // loopback host:port of the sing-box experimental API
+	Pattern string        // optional substring stat-name filter; default "user>>>"
+	Timeout time.Duration // per-invocation timeout; default 5s
+	Now     func() time.Time
+
+	// query is a test seam; production uses grpcQueryStats.
+	query func(ctx context.Context, addr, pattern string) ([]nameValue, error)
+}
+
+type nameValue struct {
+	name  string
+	value int64
+}
+
+// ValidateSingBoxStatsSource checks the loopback API address and pattern
+// without dialing, so the agent can fail fast at startup.
+func ValidateSingBoxStatsSource(source SingBoxStatsSource) error {
+	if _, err := validateLoopbackHostPort(source.APIAddr); err != nil {
+		return err
+	}
+	if pattern := strings.TrimSpace(source.Pattern); pattern != "" {
+		if err := validateXrayStatsPattern(pattern); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// LoadSingBoxStats runs one read-only QueryStats call and normalizes the
+// counters into a server-ingestible snapshot. An empty counter set (a freshly
+// started core with no traffic yet) is a valid empty snapshot, not an error.
+// Counter names keep their on-box users[].name (design-15 §5 u_<hash>) — the
+// server reverses them into (user, line) pairs.
+func LoadSingBoxStats(ctx context.Context, source SingBoxStatsSource, nodeID string) (model.ProxyUsageSnapshot, error) {
+	addr, err := validateLoopbackHostPort(source.APIAddr)
+	if err != nil {
+		return model.ProxyUsageSnapshot{}, err
+	}
+	pattern := strings.TrimSpace(source.Pattern)
+	if pattern == "" {
+		pattern = defaultSingBoxStatsPattern
+	}
+	if err := validateXrayStatsPattern(pattern); err != nil {
+		return model.ProxyUsageSnapshot{}, err
+	}
+	timeout := source.Timeout
+	if timeout <= 0 {
+		timeout = defaultSingBoxStatsTimeout
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	query := source.query
+	if query == nil {
+		query = grpcQueryStats
+	}
+	stats, err := query(ctx, addr, pattern)
+	if err != nil {
+		return model.ProxyUsageSnapshot{}, fmt.Errorf("sing-box stats query: %w", err)
+	}
+	snapshot := model.ProxyUsageSnapshot{UserBytes: map[string]int64{}}
+	for _, stat := range stats {
+		if stat.value < 0 {
+			return model.ProxyUsageSnapshot{}, fmt.Errorf("sing-box stats counter %q cannot be negative", stat.name)
+		}
+		user, ok := v2rayUserFromStatName(stat.name)
+		if !ok {
+			continue
+		}
+		snapshot.UserBytes[user] += stat.value
+	}
+	return NormalizeSnapshot(snapshot, nodeID, now(source.Now))
+}
+
+// statsConn is the minimal connection surface the stats client needs.
+type statsConn interface {
+	grpc.ClientConnInterface
+	Close() error
+}
+
+// statsNewClientConn dials the loopback API; it is a test seam (bufconn) and
+// otherwise grpc.NewClient with the caller's options.
+var statsNewClientConn = func(addr string, opts ...grpc.DialOption) (statsConn, error) {
+	return grpc.NewClient(addr, opts...)
+}
+
+// grpcQueryStats dials the loopback experimental API (plaintext: loopback
+// only, exactly like xray's own statsquery) and issues a substring QueryStats.
+func grpcQueryStats(ctx context.Context, addr, pattern string) ([]nameValue, error) {
+	conn, err := statsNewClientConn(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = conn.Close() }()
+	resp, err := singboxstats.NewStatsServiceClient(conn).QueryStats(ctx, &singboxstats.QueryStatsRequest{
+		Patterns: []string{pattern},
+	})
+	if err != nil {
+		return nil, err
+	}
+	out := make([]nameValue, 0, len(resp.GetStat()))
+	for _, stat := range resp.GetStat() {
+		out = append(out, nameValue{name: stat.GetName(), value: stat.GetValue()})
+	}
+	return out, nil
+}