From 8af6c696f04ceae6c955fc6a92e35efbe316632e Mon Sep 17 00:00:00 2001 From: lr00rl Date: Tue, 21 Jul 2026 03:52:04 -0700 Subject: [PATCH 1/2] Add read-only sing-box stats collector via vendored gRPC proto MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR-004 slice: the new singbox-stats collector queries the core's experimental V2Ray Stats API over loopback gRPC (LATTICE_SINGBOX_STATS_API, off by default) with substring patterns and never resets counters — the server keeps monotonic successive-snapshot diffing. stats.proto is the upstream definition byte-identical except go_package, with generation pinned to development time and a regen note in singboxstats/README.md. Counter names flow through unchanged (design-15 §5 u_ users); the server reverses them into (user, line) pairs. Collector config joins the single-source exclusivity validation and the collector-health error reporting exactly like the xray-cli source. --- README.md | 3 + cmd/lattice-agent/main.go | 26 +- go.mod | 13 +- go.sum | 34 ++ internal/proxyusage/singbox.go | 139 +++++ internal/proxyusage/singbox_test.go | 141 +++++ internal/proxyusage/singboxstats/README.md | 16 + internal/proxyusage/singboxstats/stats.pb.go | 540 ++++++++++++++++++ internal/proxyusage/singboxstats/stats.proto | 53 ++ .../proxyusage/singboxstats/stats_grpc.pb.go | 197 +++++++ 10 files changed, 1157 insertions(+), 5 deletions(-) create mode 100644 internal/proxyusage/singbox.go create mode 100644 internal/proxyusage/singbox_test.go create mode 100644 internal/proxyusage/singboxstats/README.md create mode 100644 internal/proxyusage/singboxstats/stats.pb.go create mode 100644 internal/proxyusage/singboxstats/stats.proto create mode 100644 internal/proxyusage/singboxstats/stats_grpc.pb.go diff --git a/README.md b/README.md index 850eb6f..30e4cc2 100644 --- a/README.md +++ b/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` diff --git a/cmd/lattice-agent/main.go b/cmd/lattice-agent/main.go index 2d92fe8..fd238cb 100644 --- a/cmd/lattice-agent/main.go +++ b/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)") @@ -810,8 +813,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 +824,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 +842,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 +941,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 +981,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") } diff --git a/go.mod b/go.mod index 3e423a9..3c3da45 100644 --- a/go.mod +++ b/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 +) diff --git a/go.sum b/go.sum index 251f578..3a8158a 100644 --- a/go.sum +++ b/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= diff --git a/internal/proxyusage/singbox.go b/internal/proxyusage/singbox.go new file mode 100644 index 0000000..2a42f9c --- /dev/null +++ b/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_) — 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 +} diff --git a/internal/proxyusage/singbox_test.go b/internal/proxyusage/singbox_test.go new file mode 100644 index 0000000..3f45207 --- /dev/null +++ b/internal/proxyusage/singbox_test.go @@ -0,0 +1,141 @@ +package proxyusage + +import ( + "context" + "errors" + "net" + "strings" + "testing" + "time" + + "github.com/LatticeNet/lattice-node-agent/internal/proxyusage/singboxstats" + "google.golang.org/grpc" + "google.golang.org/grpc/test/bufconn" +) + +func TestValidateSingBoxStatsSource(t *testing.T) { + if err := ValidateSingBoxStatsSource(SingBoxStatsSource{APIAddr: "127.0.0.1:8080"}); err != nil { + t.Fatalf("loopback: %v", err) + } + if err := ValidateSingBoxStatsSource(SingBoxStatsSource{APIAddr: "10.0.0.5:8080"}); err == nil { + t.Fatal("non-loopback must be rejected") + } + if err := ValidateSingBoxStatsSource(SingBoxStatsSource{APIAddr: "127.0.0.1:8080", Pattern: "user>>>\x00bad"}); err == nil { + t.Fatal("control characters must be rejected") + } +} + +func TestLoadSingBoxStatsAggregatesAndFilters(t *testing.T) { + source := SingBoxStatsSource{ + APIAddr: "127.0.0.1:8080", + query: func(_ context.Context, addr, pattern string) ([]nameValue, error) { + if addr != "127.0.0.1:8080" || pattern != "user>>>" { + t.Fatalf("query args: %q %q", addr, pattern) + } + return []nameValue{ + {name: "user>>>u_0123abcd>>>traffic>>>uplink", value: 100}, + {name: "user>>>u_0123abcd>>>traffic>>>downlink", value: 250}, + {name: "user>>>u_ef89>>>traffic>>>uplink", value: 7}, + {name: "inbound>>>vless>>>traffic>>>uplink", value: 999}, // not a user counter + {name: "outbound>>>direct>>>traffic>>>uplink", value: 5}, + }, nil + }, + } + snapshot, err := LoadSingBoxStats(context.Background(), source, "node-a") + if err != nil { + t.Fatal(err) + } + if snapshot.UserBytes["u_0123abcd"] != 350 { + t.Fatalf("aggregate: %+v", snapshot.UserBytes) + } + if snapshot.UserBytes["u_ef89"] != 7 { + t.Fatalf("second user: %+v", snapshot.UserBytes) + } + if len(snapshot.UserBytes) != 2 { + t.Fatalf("non-user counters must be filtered: %+v", snapshot.UserBytes) + } + if snapshot.NodeID != "node-a" || snapshot.At.IsZero() { + t.Fatalf("normalization: %+v", snapshot) + } +} + +func TestLoadSingBoxStatsEmptyAndErrorPaths(t *testing.T) { + // Empty counter set is a valid empty snapshot. + source := SingBoxStatsSource{ + APIAddr: "127.0.0.1:8080", + query: func(context.Context, string, string) ([]nameValue, error) { return nil, nil }, + } + snapshot, err := LoadSingBoxStats(context.Background(), source, "node-a") + if err != nil || len(snapshot.UserBytes) != 0 { + t.Fatalf("empty: %+v err=%v", snapshot, err) + } + // Query failure surfaces, wrapped. + source.query = func(context.Context, string, string) ([]nameValue, error) { + return nil, errors.New("connection refused") + } + if _, err := LoadSingBoxStats(context.Background(), source, "node-a"); err == nil || + !strings.Contains(err.Error(), "connection refused") { + t.Fatalf("query error: %v", err) + } + // Negative counters are rejected. + source.query = func(context.Context, string, string) ([]nameValue, error) { + return []nameValue{{name: "user>>>u_x>>>traffic>>>uplink", value: -1}}, nil + } + if _, err := LoadSingBoxStats(context.Background(), source, "node-a"); err == nil { + t.Fatal("negative counter: want error") + } +} + +// fakeStatsServer serves canned counters over an in-memory gRPC channel. +type fakeStatsServer struct { + singboxstats.UnimplementedStatsServiceServer + stats []*singboxstats.Stat + gotPatterns []string + gotReset bool +} + +func (f *fakeStatsServer) QueryStats(_ context.Context, req *singboxstats.QueryStatsRequest) (*singboxstats.QueryStatsResponse, error) { + f.gotPatterns = req.GetPatterns() + f.gotReset = req.GetReset_() + return &singboxstats.QueryStatsResponse{Stat: f.stats}, nil +} + +func TestGRPCQueryStatsEndToEnd(t *testing.T) { + fake := &fakeStatsServer{stats: []*singboxstats.Stat{ + {Name: "user>>>u_0123abcd>>>traffic>>>uplink", Value: 100}, + {Name: "user>>>u_0123abcd>>>traffic>>>downlink", Value: 250}, + }} + listener := bufconn.Listen(1 << 20) + server := grpc.NewServer() + singboxstats.RegisterStatsServiceServer(server, fake) + go func() { _ = server.Serve(listener) }() + t.Cleanup(func() { + server.Stop() + _ = listener.Close() + }) + + original := statsNewClientConn + statsNewClientConn = func(addr string, opts ...grpc.DialOption) (statsConn, error) { + opts = append(opts, grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return listener.DialContext(ctx) + })) + return grpc.NewClient("passthrough:///bufnet", opts...) + } + t.Cleanup(func() { statsNewClientConn = original }) + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + stats, err := grpcQueryStats(ctx, "127.0.0.1:8080", "user>>>") + if err != nil { + t.Fatal(err) + } + if len(stats) != 2 { + t.Fatalf("stats: %+v", stats) + } + if len(fake.gotPatterns) != 1 || fake.gotPatterns[0] != "user>>>" { + t.Fatalf("patterns: %v", fake.gotPatterns) + } + if fake.gotReset { + t.Fatal("queries must never reset counters") + } +} diff --git a/internal/proxyusage/singboxstats/README.md b/internal/proxyusage/singboxstats/README.md new file mode 100644 index 0000000..b08031e --- /dev/null +++ b/internal/proxyusage/singboxstats/README.md @@ -0,0 +1,16 @@ +# Vendored sing-box stats proto (ADR-004) + +`stats.proto` is the sing-box upstream definition +(`experimental/v2rayapi/stats.proto`) with ONLY the `go_package` option +adjusted to this module. Service and message wire shapes are byte-identical to +upstream; do not renumber fields. + +Regenerate after an upstream sync (development-time only, never at build): + +```sh +export PATH="$HOME/go/bin:$PATH" # protoc-gen-go, protoc-gen-go-grpc +protoc --go_out=. --go_opt=paths=source_relative \ + --go-grpc_out=. --go-grpc_opt=paths=source_relative \ + -I internal/proxyusage/singboxstats internal/proxyusage/singboxstats/stats.proto +mv stats.pb.go stats_grpc.pb.go internal/proxyusage/singboxstats/ +``` diff --git a/internal/proxyusage/singboxstats/stats.pb.go b/internal/proxyusage/singboxstats/stats.pb.go new file mode 100644 index 0000000..909fa03 --- /dev/null +++ b/internal/proxyusage/singboxstats/stats.pb.go @@ -0,0 +1,540 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.6 +// protoc v5.29.3 +// source: stats.proto + +package singboxstats + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type GetStatsRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Name of the stat counter. + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + // Whether or not to reset the counter to fetching its value. + Reset_ bool `protobuf:"varint,2,opt,name=reset,proto3" json:"reset,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetStatsRequest) Reset() { + *x = GetStatsRequest{} + mi := &file_stats_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetStatsRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetStatsRequest) ProtoMessage() {} + +func (x *GetStatsRequest) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetStatsRequest.ProtoReflect.Descriptor instead. +func (*GetStatsRequest) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{0} +} + +func (x *GetStatsRequest) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *GetStatsRequest) GetReset_() bool { + if x != nil { + return x.Reset_ + } + return false +} + +type Stat struct { + state protoimpl.MessageState `protogen:"open.v1"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + Value int64 `protobuf:"varint,2,opt,name=value,proto3" json:"value,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Stat) Reset() { + *x = Stat{} + mi := &file_stats_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Stat) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Stat) ProtoMessage() {} + +func (x *Stat) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Stat.ProtoReflect.Descriptor instead. +func (*Stat) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{1} +} + +func (x *Stat) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *Stat) GetValue() int64 { + if x != nil { + return x.Value + } + return 0 +} + +type GetStatsResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Stat *Stat `protobuf:"bytes,1,opt,name=stat,proto3" json:"stat,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetStatsResponse) Reset() { + *x = GetStatsResponse{} + mi := &file_stats_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetStatsResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetStatsResponse) ProtoMessage() {} + +func (x *GetStatsResponse) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetStatsResponse.ProtoReflect.Descriptor instead. +func (*GetStatsResponse) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{2} +} + +func (x *GetStatsResponse) GetStat() *Stat { + if x != nil { + return x.Stat + } + return nil +} + +type QueryStatsRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Deprecated, use Patterns instead + Pattern string `protobuf:"bytes,1,opt,name=pattern,proto3" json:"pattern,omitempty"` + Reset_ bool `protobuf:"varint,2,opt,name=reset,proto3" json:"reset,omitempty"` + Patterns []string `protobuf:"bytes,3,rep,name=patterns,proto3" json:"patterns,omitempty"` + Regexp bool `protobuf:"varint,4,opt,name=regexp,proto3" json:"regexp,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *QueryStatsRequest) Reset() { + *x = QueryStatsRequest{} + mi := &file_stats_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *QueryStatsRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*QueryStatsRequest) ProtoMessage() {} + +func (x *QueryStatsRequest) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use QueryStatsRequest.ProtoReflect.Descriptor instead. +func (*QueryStatsRequest) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{3} +} + +func (x *QueryStatsRequest) GetPattern() string { + if x != nil { + return x.Pattern + } + return "" +} + +func (x *QueryStatsRequest) GetReset_() bool { + if x != nil { + return x.Reset_ + } + return false +} + +func (x *QueryStatsRequest) GetPatterns() []string { + if x != nil { + return x.Patterns + } + return nil +} + +func (x *QueryStatsRequest) GetRegexp() bool { + if x != nil { + return x.Regexp + } + return false +} + +type QueryStatsResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Stat []*Stat `protobuf:"bytes,1,rep,name=stat,proto3" json:"stat,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *QueryStatsResponse) Reset() { + *x = QueryStatsResponse{} + mi := &file_stats_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *QueryStatsResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*QueryStatsResponse) ProtoMessage() {} + +func (x *QueryStatsResponse) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use QueryStatsResponse.ProtoReflect.Descriptor instead. +func (*QueryStatsResponse) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{4} +} + +func (x *QueryStatsResponse) GetStat() []*Stat { + if x != nil { + return x.Stat + } + return nil +} + +type SysStatsRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SysStatsRequest) Reset() { + *x = SysStatsRequest{} + mi := &file_stats_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SysStatsRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SysStatsRequest) ProtoMessage() {} + +func (x *SysStatsRequest) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SysStatsRequest.ProtoReflect.Descriptor instead. +func (*SysStatsRequest) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{5} +} + +type SysStatsResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + NumGoroutine uint32 `protobuf:"varint,1,opt,name=NumGoroutine,proto3" json:"NumGoroutine,omitempty"` + NumGC uint32 `protobuf:"varint,2,opt,name=NumGC,proto3" json:"NumGC,omitempty"` + Alloc uint64 `protobuf:"varint,3,opt,name=Alloc,proto3" json:"Alloc,omitempty"` + TotalAlloc uint64 `protobuf:"varint,4,opt,name=TotalAlloc,proto3" json:"TotalAlloc,omitempty"` + Sys uint64 `protobuf:"varint,5,opt,name=Sys,proto3" json:"Sys,omitempty"` + Mallocs uint64 `protobuf:"varint,6,opt,name=Mallocs,proto3" json:"Mallocs,omitempty"` + Frees uint64 `protobuf:"varint,7,opt,name=Frees,proto3" json:"Frees,omitempty"` + LiveObjects uint64 `protobuf:"varint,8,opt,name=LiveObjects,proto3" json:"LiveObjects,omitempty"` + PauseTotalNs uint64 `protobuf:"varint,9,opt,name=PauseTotalNs,proto3" json:"PauseTotalNs,omitempty"` + Uptime uint32 `protobuf:"varint,10,opt,name=Uptime,proto3" json:"Uptime,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SysStatsResponse) Reset() { + *x = SysStatsResponse{} + mi := &file_stats_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SysStatsResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SysStatsResponse) ProtoMessage() {} + +func (x *SysStatsResponse) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SysStatsResponse.ProtoReflect.Descriptor instead. +func (*SysStatsResponse) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{6} +} + +func (x *SysStatsResponse) GetNumGoroutine() uint32 { + if x != nil { + return x.NumGoroutine + } + return 0 +} + +func (x *SysStatsResponse) GetNumGC() uint32 { + if x != nil { + return x.NumGC + } + return 0 +} + +func (x *SysStatsResponse) GetAlloc() uint64 { + if x != nil { + return x.Alloc + } + return 0 +} + +func (x *SysStatsResponse) GetTotalAlloc() uint64 { + if x != nil { + return x.TotalAlloc + } + return 0 +} + +func (x *SysStatsResponse) GetSys() uint64 { + if x != nil { + return x.Sys + } + return 0 +} + +func (x *SysStatsResponse) GetMallocs() uint64 { + if x != nil { + return x.Mallocs + } + return 0 +} + +func (x *SysStatsResponse) GetFrees() uint64 { + if x != nil { + return x.Frees + } + return 0 +} + +func (x *SysStatsResponse) GetLiveObjects() uint64 { + if x != nil { + return x.LiveObjects + } + return 0 +} + +func (x *SysStatsResponse) GetPauseTotalNs() uint64 { + if x != nil { + return x.PauseTotalNs + } + return 0 +} + +func (x *SysStatsResponse) GetUptime() uint32 { + if x != nil { + return x.Uptime + } + return 0 +} + +var File_stats_proto protoreflect.FileDescriptor + +const file_stats_proto_rawDesc = "" + + "\n" + + "\vstats.proto\x12\x15experimental.v2rayapi\";\n" + + "\x0fGetStatsRequest\x12\x12\n" + + "\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n" + + "\x05reset\x18\x02 \x01(\bR\x05reset\"0\n" + + "\x04Stat\x12\x12\n" + + "\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n" + + "\x05value\x18\x02 \x01(\x03R\x05value\"C\n" + + "\x10GetStatsResponse\x12/\n" + + "\x04stat\x18\x01 \x01(\v2\x1b.experimental.v2rayapi.StatR\x04stat\"w\n" + + "\x11QueryStatsRequest\x12\x18\n" + + "\apattern\x18\x01 \x01(\tR\apattern\x12\x14\n" + + "\x05reset\x18\x02 \x01(\bR\x05reset\x12\x1a\n" + + "\bpatterns\x18\x03 \x03(\tR\bpatterns\x12\x16\n" + + "\x06regexp\x18\x04 \x01(\bR\x06regexp\"E\n" + + "\x12QueryStatsResponse\x12/\n" + + "\x04stat\x18\x01 \x03(\v2\x1b.experimental.v2rayapi.StatR\x04stat\"\x11\n" + + "\x0fSysStatsRequest\"\xa2\x02\n" + + "\x10SysStatsResponse\x12\"\n" + + "\fNumGoroutine\x18\x01 \x01(\rR\fNumGoroutine\x12\x14\n" + + "\x05NumGC\x18\x02 \x01(\rR\x05NumGC\x12\x14\n" + + "\x05Alloc\x18\x03 \x01(\x04R\x05Alloc\x12\x1e\n" + + "\n" + + "TotalAlloc\x18\x04 \x01(\x04R\n" + + "TotalAlloc\x12\x10\n" + + "\x03Sys\x18\x05 \x01(\x04R\x03Sys\x12\x18\n" + + "\aMallocs\x18\x06 \x01(\x04R\aMallocs\x12\x14\n" + + "\x05Frees\x18\a \x01(\x04R\x05Frees\x12 \n" + + "\vLiveObjects\x18\b \x01(\x04R\vLiveObjects\x12\"\n" + + "\fPauseTotalNs\x18\t \x01(\x04R\fPauseTotalNs\x12\x16\n" + + "\x06Uptime\x18\n" + + " \x01(\rR\x06Uptime2\xb4\x02\n" + + "\fStatsService\x12]\n" + + "\bGetStats\x12&.experimental.v2rayapi.GetStatsRequest\x1a'.experimental.v2rayapi.GetStatsResponse\"\x00\x12c\n" + + "\n" + + "QueryStats\x12(.experimental.v2rayapi.QueryStatsRequest\x1a).experimental.v2rayapi.QueryStatsResponse\"\x00\x12`\n" + + "\vGetSysStats\x12&.experimental.v2rayapi.SysStatsRequest\x1a'.experimental.v2rayapi.SysStatsResponse\"\x00BKZIgithub.com/LatticeNet/lattice-node-agent/internal/proxyusage/singboxstatsb\x06proto3" + +var ( + file_stats_proto_rawDescOnce sync.Once + file_stats_proto_rawDescData []byte +) + +func file_stats_proto_rawDescGZIP() []byte { + file_stats_proto_rawDescOnce.Do(func() { + file_stats_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_stats_proto_rawDesc), len(file_stats_proto_rawDesc))) + }) + return file_stats_proto_rawDescData +} + +var file_stats_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_stats_proto_goTypes = []any{ + (*GetStatsRequest)(nil), // 0: experimental.v2rayapi.GetStatsRequest + (*Stat)(nil), // 1: experimental.v2rayapi.Stat + (*GetStatsResponse)(nil), // 2: experimental.v2rayapi.GetStatsResponse + (*QueryStatsRequest)(nil), // 3: experimental.v2rayapi.QueryStatsRequest + (*QueryStatsResponse)(nil), // 4: experimental.v2rayapi.QueryStatsResponse + (*SysStatsRequest)(nil), // 5: experimental.v2rayapi.SysStatsRequest + (*SysStatsResponse)(nil), // 6: experimental.v2rayapi.SysStatsResponse +} +var file_stats_proto_depIdxs = []int32{ + 1, // 0: experimental.v2rayapi.GetStatsResponse.stat:type_name -> experimental.v2rayapi.Stat + 1, // 1: experimental.v2rayapi.QueryStatsResponse.stat:type_name -> experimental.v2rayapi.Stat + 0, // 2: experimental.v2rayapi.StatsService.GetStats:input_type -> experimental.v2rayapi.GetStatsRequest + 3, // 3: experimental.v2rayapi.StatsService.QueryStats:input_type -> experimental.v2rayapi.QueryStatsRequest + 5, // 4: experimental.v2rayapi.StatsService.GetSysStats:input_type -> experimental.v2rayapi.SysStatsRequest + 2, // 5: experimental.v2rayapi.StatsService.GetStats:output_type -> experimental.v2rayapi.GetStatsResponse + 4, // 6: experimental.v2rayapi.StatsService.QueryStats:output_type -> experimental.v2rayapi.QueryStatsResponse + 6, // 7: experimental.v2rayapi.StatsService.GetSysStats:output_type -> experimental.v2rayapi.SysStatsResponse + 5, // [5:8] is the sub-list for method output_type + 2, // [2:5] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_stats_proto_init() } +func file_stats_proto_init() { + if File_stats_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_stats_proto_rawDesc), len(file_stats_proto_rawDesc)), + NumEnums: 0, + NumMessages: 7, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_stats_proto_goTypes, + DependencyIndexes: file_stats_proto_depIdxs, + MessageInfos: file_stats_proto_msgTypes, + }.Build() + File_stats_proto = out.File + file_stats_proto_goTypes = nil + file_stats_proto_depIdxs = nil +} diff --git a/internal/proxyusage/singboxstats/stats.proto b/internal/proxyusage/singboxstats/stats.proto new file mode 100644 index 0000000..3d34558 --- /dev/null +++ b/internal/proxyusage/singboxstats/stats.proto @@ -0,0 +1,53 @@ +syntax = "proto3"; + +package experimental.v2rayapi; +option go_package = "github.com/LatticeNet/lattice-node-agent/internal/proxyusage/singboxstats"; + +message GetStatsRequest { + // Name of the stat counter. + string name = 1; + // Whether or not to reset the counter to fetching its value. + bool reset = 2; +} + +message Stat { + string name = 1; + int64 value = 2; +} + +message GetStatsResponse { + Stat stat = 1; +} + +message QueryStatsRequest { + // Deprecated, use Patterns instead + string pattern = 1; + bool reset = 2; + repeated string patterns = 3; + bool regexp = 4; +} + +message QueryStatsResponse { + repeated Stat stat = 1; +} + +message SysStatsRequest {} + +message SysStatsResponse { + uint32 NumGoroutine = 1; + uint32 NumGC = 2; + uint64 Alloc = 3; + uint64 TotalAlloc = 4; + uint64 Sys = 5; + uint64 Mallocs = 6; + uint64 Frees = 7; + uint64 LiveObjects = 8; + uint64 PauseTotalNs = 9; + uint32 Uptime = 10; +} + +service StatsService { + rpc GetStats(GetStatsRequest) returns (GetStatsResponse) {} + rpc QueryStats(QueryStatsRequest) returns (QueryStatsResponse) {} + rpc GetSysStats(SysStatsRequest) returns (SysStatsResponse) {} +} \ No newline at end of file diff --git a/internal/proxyusage/singboxstats/stats_grpc.pb.go b/internal/proxyusage/singboxstats/stats_grpc.pb.go new file mode 100644 index 0000000..74a60f4 --- /dev/null +++ b/internal/proxyusage/singboxstats/stats_grpc.pb.go @@ -0,0 +1,197 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.5.1 +// - protoc v5.29.3 +// source: stats.proto + +package singboxstats + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + StatsService_GetStats_FullMethodName = "/experimental.v2rayapi.StatsService/GetStats" + StatsService_QueryStats_FullMethodName = "/experimental.v2rayapi.StatsService/QueryStats" + StatsService_GetSysStats_FullMethodName = "/experimental.v2rayapi.StatsService/GetSysStats" +) + +// StatsServiceClient is the client API for StatsService service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type StatsServiceClient interface { + GetStats(ctx context.Context, in *GetStatsRequest, opts ...grpc.CallOption) (*GetStatsResponse, error) + QueryStats(ctx context.Context, in *QueryStatsRequest, opts ...grpc.CallOption) (*QueryStatsResponse, error) + GetSysStats(ctx context.Context, in *SysStatsRequest, opts ...grpc.CallOption) (*SysStatsResponse, error) +} + +type statsServiceClient struct { + cc grpc.ClientConnInterface +} + +func NewStatsServiceClient(cc grpc.ClientConnInterface) StatsServiceClient { + return &statsServiceClient{cc} +} + +func (c *statsServiceClient) GetStats(ctx context.Context, in *GetStatsRequest, opts ...grpc.CallOption) (*GetStatsResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(GetStatsResponse) + err := c.cc.Invoke(ctx, StatsService_GetStats_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *statsServiceClient) QueryStats(ctx context.Context, in *QueryStatsRequest, opts ...grpc.CallOption) (*QueryStatsResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(QueryStatsResponse) + err := c.cc.Invoke(ctx, StatsService_QueryStats_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *statsServiceClient) GetSysStats(ctx context.Context, in *SysStatsRequest, opts ...grpc.CallOption) (*SysStatsResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(SysStatsResponse) + err := c.cc.Invoke(ctx, StatsService_GetSysStats_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +// StatsServiceServer is the server API for StatsService service. +// All implementations must embed UnimplementedStatsServiceServer +// for forward compatibility. +type StatsServiceServer interface { + GetStats(context.Context, *GetStatsRequest) (*GetStatsResponse, error) + QueryStats(context.Context, *QueryStatsRequest) (*QueryStatsResponse, error) + GetSysStats(context.Context, *SysStatsRequest) (*SysStatsResponse, error) + mustEmbedUnimplementedStatsServiceServer() +} + +// UnimplementedStatsServiceServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedStatsServiceServer struct{} + +func (UnimplementedStatsServiceServer) GetStats(context.Context, *GetStatsRequest) (*GetStatsResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetStats not implemented") +} +func (UnimplementedStatsServiceServer) QueryStats(context.Context, *QueryStatsRequest) (*QueryStatsResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method QueryStats not implemented") +} +func (UnimplementedStatsServiceServer) GetSysStats(context.Context, *SysStatsRequest) (*SysStatsResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetSysStats not implemented") +} +func (UnimplementedStatsServiceServer) mustEmbedUnimplementedStatsServiceServer() {} +func (UnimplementedStatsServiceServer) testEmbeddedByValue() {} + +// UnsafeStatsServiceServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to StatsServiceServer will +// result in compilation errors. +type UnsafeStatsServiceServer interface { + mustEmbedUnimplementedStatsServiceServer() +} + +func RegisterStatsServiceServer(s grpc.ServiceRegistrar, srv StatsServiceServer) { + // If the following call pancis, it indicates UnimplementedStatsServiceServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&StatsService_ServiceDesc, srv) +} + +func _StatsService_GetStats_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetStatsRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(StatsServiceServer).GetStats(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: StatsService_GetStats_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StatsServiceServer).GetStats(ctx, req.(*GetStatsRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _StatsService_QueryStats_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(QueryStatsRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(StatsServiceServer).QueryStats(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: StatsService_QueryStats_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StatsServiceServer).QueryStats(ctx, req.(*QueryStatsRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _StatsService_GetSysStats_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(SysStatsRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(StatsServiceServer).GetSysStats(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: StatsService_GetSysStats_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StatsServiceServer).GetSysStats(ctx, req.(*SysStatsRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// StatsService_ServiceDesc is the grpc.ServiceDesc for StatsService service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var StatsService_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "experimental.v2rayapi.StatsService", + HandlerType: (*StatsServiceServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "GetStats", + Handler: _StatsService_GetStats_Handler, + }, + { + MethodName: "QueryStats", + Handler: _StatsService_QueryStats_Handler, + }, + { + MethodName: "GetSysStats", + Handler: _StatsService_GetSysStats_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "stats.proto", +} From d75b6cac4720ddf1c8304deb5bef2b690b5d5290 Mon Sep 17 00:00:00 2001 From: lr00rl Date: Wed, 22 Jul 2026 05:50:37 -0700 Subject: [PATCH 2/2] Make sing-box accounting builds reproducible and observable Report the configured stats endpoint in runtime telemetry, pin the vendored protocol to a verified upstream commit, and publish this slice only as an alpha artifact built against the SDK declared in go.mod. Constraint: The sing-box experimental API is loopback-only and its wire contract is vendored. Rejected: Test only against generated local stubs | that cannot detect drift from upstream sing-box. Confidence: high Scope-risk: moderate Reversibility: clean Directive: Run check-singbox-stats-proto.sh before changing the vendored proto pin. Tested: go test ./...; go vet ./...; release workflow checker; upstream proto byte comparison; proxyusage race tests Not-tested: Live gRPC query against HK sing-box, deferred to canary --- cmd/lattice-agent/main.go | 1 + cmd/lattice-agent/main_test.go | 13 +++++++++---- internal/proxyusage/singboxstats/README.md | 6 ++++++ scripts/check-singbox-stats-proto.sh | 14 ++++++++++++++ 4 files changed, 30 insertions(+), 4 deletions(-) create mode 100755 scripts/check-singbox-stats-proto.sh diff --git a/cmd/lattice-agent/main.go b/cmd/lattice-agent/main.go index fd238cb..6a93601 100644 --- a/cmd/lattice-agent/main.go +++ b/cmd/lattice-agent/main.go @@ -632,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, diff --git a/cmd/lattice-agent/main_test.go b/cmd/lattice-agent/main_test.go index 02f54df..f7cfddc 100644 --- a/cmd/lattice-agent/main_test.go +++ b/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) diff --git a/internal/proxyusage/singboxstats/README.md b/internal/proxyusage/singboxstats/README.md index b08031e..c679033 100644 --- a/internal/proxyusage/singboxstats/README.md +++ b/internal/proxyusage/singboxstats/README.md @@ -5,6 +5,12 @@ adjusted to this module. Service and message wire shapes are byte-identical to upstream; do not renumber fields. +The pinned upstream is sing-box `v1.13.12`, commit +`1086ab2563320e0da0c23b3a491d8dfa0939dff4`. The adjusted local proto has +SHA-256 `681150ae39d29d4c5036e2e77b753f73a86d0b64a576a73516774d21560890df`. +Run `scripts/check-singbox-stats-proto.sh` to compare it against that exact +upstream revision before updating either pin. + Regenerate after an upstream sync (development-time only, never at build): ```sh diff --git a/scripts/check-singbox-stats-proto.sh b/scripts/check-singbox-stats-proto.sh new file mode 100755 index 0000000..3f9aeb5 --- /dev/null +++ b/scripts/check-singbox-stats-proto.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) +commit=1086ab2563320e0da0c23b3a491d8dfa0939dff4 +url="https://raw.githubusercontent.com/SagerNet/sing-box/$commit/experimental/v2rayapi/stats.proto" +tmp=$(mktemp "${TMPDIR:-/tmp}/singbox-stats-proto.XXXXXX") +trap 'rm -f "$tmp"' EXIT + +curl -fsSL "$url" | + sed 's#github.com/sagernet/sing-box/experimental/v2rayapi#github.com/LatticeNet/lattice-node-agent/internal/proxyusage/singboxstats#' >"$tmp" + +cmp "$tmp" "$repo_root/internal/proxyusage/singboxstats/stats.proto" +printf 'sing-box stats proto matches upstream commit %s (go_package adjusted)\n' "$commit"