diff --git a/.gitignore b/.gitignore index 1198189..63f180e 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,7 @@ docs/ # build things ./proxy dist/ + +# bench artifacts +bench-*.txt +benchstat-*.txt diff --git a/internal/protoutil/extractor_bench_test.go b/internal/protoutil/extractor_bench_test.go new file mode 100644 index 0000000..592ad2f --- /dev/null +++ b/internal/protoutil/extractor_bench_test.go @@ -0,0 +1,50 @@ +package protoutil_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoregistry" + + "go.temporal.io/proxy/internal/protoutil" + "go.temporal.io/proxy/internal/testutils" +) + +func benchMarshal(b *testing.B, m proto.Message) []byte { + b.Helper() + + payload, err := proto.Marshal(m) + require.NoError(b, err) + return payload +} + +func BenchmarkExtractor_NamespacedRequest(b *testing.B) { + payload := benchMarshal(b, testutils.StartWorkflowRequest()) + ex := protoutil.NewExtractor(protoregistry.GlobalFiles, protoregistry.GlobalTypes) + require.Equal(b, "payments-production", ex.Namespace(startMethod, payload), "warm-up call must extract the namespace") + + b.ReportAllocs() + for b.Loop() { + if ns := ex.Namespace(startMethod, payload); ns == "" { + b.Fatal("namespace was not extracted") + } + } +} + +// BenchmarkExtractor_SystemRequest covers the request the SDK sends on every +// connect: no namespace accessor, so the full unmarshal runs and the cast +// misses. +func BenchmarkExtractor_SystemRequest(b *testing.B) { + payload := benchMarshal(b, &workflowservice.GetSystemInfoRequest{}) + ex := protoutil.NewExtractor(protoregistry.GlobalFiles, protoregistry.GlobalTypes) + require.Equal(b, "", ex.Namespace(systemMethod, payload)) + + b.ReportAllocs() + for b.Loop() { + if ns := ex.Namespace(systemMethod, payload); ns != "" { + b.Fatal("system request unexpectedly carried a namespace") + } + } +} diff --git a/internal/protoutil/translate_bench_test.go b/internal/protoutil/translate_bench_test.go new file mode 100644 index 0000000..505dd97 --- /dev/null +++ b/internal/protoutil/translate_bench_test.go @@ -0,0 +1,39 @@ +package protoutil_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/namespace/v1" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/protobuf/reflect/protoregistry" + + "go.temporal.io/proxy/internal/protoutil" + "go.temporal.io/proxy/internal/testutils" +) + +func BenchmarkTranslator_StartWorkflowRequest(b *testing.B) { + tr := protoutil.NewTranslator(protoregistry.GlobalFiles) + require.NoError(b, tr.WarmService(wfService)) + req := testutils.StartWorkflowRequest() + + b.ReportAllocs() + for b.Loop() { + // The mapper returns a fixed name so repeated in-place rewrites stay + // stable and the benchmark measures the walk, not string growth. + tr.Translate(req, func(string) string { return "payments-production-remote" }) + } +} + +func BenchmarkTranslator_DescribeNamespaceResponse(b *testing.B) { + tr := protoutil.NewTranslator(protoregistry.GlobalFiles) + require.NoError(b, tr.WarmService(wfService)) + resp := &workflowservice.DescribeNamespaceResponse{ + NamespaceInfo: &namespace.NamespaceInfo{Name: "payments-production"}, + } + + b.ReportAllocs() + for b.Loop() { + tr.Translate(resp, func(string) string { return "payments-production-local" }) + } +} diff --git a/internal/router/compile_test.go b/internal/router/compile_test.go index 42e04e9..a60d34e 100644 --- a/internal/router/compile_test.go +++ b/internal/router/compile_test.go @@ -112,7 +112,7 @@ func TestMuxFor(t *testing.T) { // prepared returns r after preparing it inside a minimal config with one // upstream per name r references. -func prepared(t *testing.T, r config.Routing) config.Routing { +func prepared(t testing.TB, r config.Routing) config.Routing { t.Helper() names := map[string]struct{}{"primary": {}} diff --git a/internal/router/director_bench_test.go b/internal/router/director_bench_test.go new file mode 100644 index 0000000..f7f2fa0 --- /dev/null +++ b/internal/router/director_bench_test.go @@ -0,0 +1,89 @@ +package router_test + +import ( + "testing" + + "google.golang.org/grpc" + + "go.temporal.io/proxy/internal/config" + "go.temporal.io/proxy/internal/router" +) + +// BenchmarkDirector_Resolve measures the routing decision including the +// decision metric: the real MuxFor mux and a Prometheus-backed Reporter on a +// fresh registry, with the handlers map the dataplane would hold. The shapes +// cover a namespace rule that matches, a metadata-keyed rule, the system +// upstream, and the default fall-through. The logger is nil so Debug entries +// take the same disabled path as production's default log level. +func BenchmarkDirector_Resolve(b *testing.B) { + mux := router.MuxFor(prepared(b, config.Routing{ + DefaultUpstream: "primary", + SystemUpstream: "system", + Rules: []config.RoutingRule{ + {Upstream: "prod", Match: config.RoutingMatch{Namespace: "prod-*"}}, + {Upstream: "cluster-3", Match: config.RoutingMatch{Metadata: map[string]string{"x-cluster": "3"}}}, + }, + })) + + rep, _ := newTestReporter(b, "primary", "prod", "cluster-3", "system") + handlers := map[string]grpc.StreamHandler{ + "primary": func(any, grpc.ServerStream) error { return nil }, + "prod": func(any, grpc.ServerStream) error { return nil }, + "cluster-3": func(any, grpc.ServerStream) error { return nil }, + "system": func(any, grpc.ServerStream) error { return nil }, + } + d := router.NewDirector(mux, handlers, rep, nil) + + b.Run("namespace-match", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + target, err := d.Resolve(b.Context(), "/svc/Method", "prod-eu-1", nil) + if err != nil { + b.Fatal(err) + } + if target.Upstream != "prod" { + b.Fatalf("routed to %q", target.Upstream) + } + } + }) + + b.Run("metadata-match", func(b *testing.B) { + md := map[string][]string{"x-cluster": {"3"}} + b.ReportAllocs() + for b.Loop() { + target, err := d.Resolve(b.Context(), "/svc/Method", "other-ns", md) + if err != nil { + b.Fatal(err) + } + if target.Upstream != "cluster-3" { + b.Fatalf("routed to %q", target.Upstream) + } + } + }) + + b.Run("system", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + target, err := d.Resolve(b.Context(), "/svc/Method", "", nil) + if err != nil { + b.Fatal(err) + } + if target.Upstream != "system" { + b.Fatalf("routed to %q", target.Upstream) + } + } + }) + + b.Run("default", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + target, err := d.Resolve(b.Context(), "/svc/Method", "tenant-x", nil) + if err != nil { + b.Fatal(err) + } + if target.Upstream != "primary" { + b.Fatalf("routed to %q", target.Upstream) + } + } + }) +} diff --git a/internal/router/handler_test.go b/internal/router/handler_test.go index 6c34db6..9021fad 100644 --- a/internal/router/handler_test.go +++ b/internal/router/handler_test.go @@ -681,7 +681,7 @@ func capturingUpstream(t *testing.T) (*grpc.ClientConn, <-chan string) { return dialBufconn(t, lis), got } -func dialBufconn(t *testing.T, lis *bufconn.Listener) *grpc.ClientConn { +func dialBufconn(t testing.TB, lis *bufconn.Listener) *grpc.ClientConn { t.Helper() conn, err := grpc.NewClient( "passthrough:///bufnet", @@ -695,7 +695,7 @@ func dialBufconn(t *testing.T, lis *bufconn.Listener) *grpc.ClientConn { return conn } -func serve(t *testing.T, srv *grpc.Server, lis *bufconn.Listener) { +func serve(t testing.TB, srv *grpc.Server, lis *bufconn.Listener) { t.Helper() go func() { _ = srv.Serve(lis) }() t.Cleanup(srv.Stop) @@ -705,7 +705,7 @@ func serve(t *testing.T, srv *grpc.Server, lis *bufconn.Listener) { // production namespace and subsystem so emitted series match the // tmprl_proxy_router_* names, for tests that need to observe (or discard) what // the handler records. -func newTestReporter(t *testing.T, upstreams ...string) (*router.Reporter, *prometheus.Registry) { +func newTestReporter(t testing.TB, upstreams ...string) (*router.Reporter, *prometheus.Registry) { t.Helper() return newLabeledTestReporter(t, metrics.MetadataLabels{}, upstreams...) } @@ -713,7 +713,7 @@ func newTestReporter(t *testing.T, upstreams ...string) (*router.Reporter, *prom // newLabeledTestReporter is newTestReporter for the cases that care what // metadata labels do to the label set and to pre-resolution. func newLabeledTestReporter( - t *testing.T, + t testing.TB, labels metrics.MetadataLabels, upstreams ...string, ) (*router.Reporter, *prometheus.Registry) { @@ -726,7 +726,7 @@ func newLabeledTestReporter( // newRelayToUpstream stands up a fake upstream (configured by registerUpstream), // then a bare relay server that forwards all methods to it via router.Handler // using pass-through routing. Returns a client conn pointed at the relay. -func newRelayToUpstream(t *testing.T, registerUpstream func(*grpc.Server)) *grpc.ClientConn { +func newRelayToUpstream(t testing.TB, registerUpstream func(*grpc.Server)) *grpc.ClientConn { t.Helper() return newRelayWith( @@ -740,7 +740,7 @@ func newRelayToUpstream(t *testing.T, registerUpstream func(*grpc.Server)) *grpc // Reflector, and Reporter, so tests can observe how the handler routes and what // it records. makeDirector receives the connection to the fake upstream. func newRelayWith( - t *testing.T, + t testing.TB, registerUpstream func(*grpc.Server), makeDirector func(cc *grpc.ClientConn) router.Director, reflector router.Reflector, diff --git a/internal/router/relay_bench_test.go b/internal/router/relay_bench_test.go new file mode 100644 index 0000000..7022c10 --- /dev/null +++ b/internal/router/relay_bench_test.go @@ -0,0 +1,152 @@ +package router_test + +import ( + "context" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/stretchr/testify/require" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/reflect/protoregistry" + + "go.temporal.io/proxy/internal/config" + "go.temporal.io/proxy/internal/metrics" + "go.temporal.io/proxy/internal/protoutil" + "go.temporal.io/proxy/internal/proxy" + "go.temporal.io/proxy/internal/router" + "go.temporal.io/proxy/internal/server" + "go.temporal.io/proxy/internal/services" + "go.temporal.io/proxy/internal/testutils" +) + +const ( + benchService = "temporal.api.workflowservice.v1.WorkflowService" + benchMethod = "/" + benchService + "/StartWorkflowExecution" + benchRunID = "0198f4a2-c0de-7f31-a5b2-9e11d1c2a3b4" +) + +// benchUpstreamServer returns a gRPC server that answers StartWorkflowExecution +// under the real service name, so the forwarder resolves the method against the +// production descriptor and type registries. The handler decodes the request and +// returns a run, which is the observable part of the response SDKs rely on. +func benchUpstreamServer() *grpc.Server { + desc := grpc.ServiceDesc{ + ServiceName: benchService, + HandlerType: (*any)(nil), + Methods: []grpc.MethodDesc{{ + MethodName: "StartWorkflowExecution", + Handler: func(_ any, _ context.Context, dec func(any) error, _ grpc.UnaryServerInterceptor) (any, error) { + req := new(workflowservice.StartWorkflowExecutionRequest) + if err := dec(req); err != nil { + return nil, err + } + + return &workflowservice.StartWorkflowExecutionResponse{RunId: benchRunID}, nil + }, + }}, + Streams: []grpc.StreamDesc{}, + } + + srv := grpc.NewServer() + srv.RegisterService(&desc, nil) + return srv +} + +// benchInvoke runs one unary call and checks the response survived the round +// trip, so a codec or forwarder regression fails the benchmark instead of +// feeding it garbage. +func benchInvoke(b *testing.B, conn *grpc.ClientConn, req *workflowservice.StartWorkflowExecutionRequest) { + b.Helper() + + resp := new(workflowservice.StartWorkflowExecutionResponse) + if err := conn.Invoke(b.Context(), benchMethod, req, resp); err != nil { + b.Fatal(err) + } + if got := resp.GetRunId(); got != benchRunID { + b.Fatalf("response carried run id %q", got) + } +} + +// benchRelayConn stands up the production gateway shape: the hybrid codec, the +// server metrics interceptor, PeekInterceptor backed by the real extractor, and +// Handler backed by a real Director whose single upstream is a Forwarder over a +// bufconn connection to the fake upstream. Authentication (off by default) and +// panic recovery are not installed. +func benchRelayConn(b *testing.B) *grpc.ClientConn { + b.Helper() + + upstreamLis := bufconn.Listen(1024 * 1024) + upstream := benchUpstreamServer() + serve(b, upstream, upstreamLis) + upstreamConn := dialBufconn(b, upstreamLis) + b.Cleanup(func() { _ = upstreamConn.Close() }) + + fw, err := proxy.NewForwarder(upstreamConn) + require.NoError(b, err) + + mux := router.MuxFor(prepared(b, config.Routing{DefaultUpstream: "primary"})) + rep, _ := newTestReporter(b, "primary") + d := router.NewDirector(mux, map[string]grpc.StreamHandler{"primary": fw.Handle}, rep, nil) + + extractor := protoutil.NewExtractor(protoregistry.GlobalFiles, protoregistry.GlobalTypes) + allowlist := services.NewAllowlist(services.Default()) + + reg := prometheus.NewRegistry() + serverRep := server.NewReporter( + metrics.New("tmprl_proxy", promauto.With(reg)).ForSubsystem("server"), + metrics.MetadataLabels{}, + ) + + relayLis := bufconn.Listen(1024 * 1024) + relay := grpc.NewServer( + grpc.ForceServerCodecV2(router.Codec()), + grpc.ChainStreamInterceptor( + serverRep.StreamInterceptor(), + router.PeekInterceptor(extractor, allowlist), + ), + grpc.UnknownServiceHandler(router.Handler(d, allowlist)), + ) + serve(b, relay, relayLis) + + conn := dialBufconn(b, relayLis) + b.Cleanup(func() { _ = conn.Close() }) + return conn +} + +// BenchmarkUnaryThroughProxy measures one StartWorkflowExecution round trip as +// an SDK client observes it: client -> gateway (codec, metrics, peek, routing) +// -> forwarder (typed codec hop) -> upstream and back. Compare with +// BenchmarkUnaryDirect: the difference is what the proxy adds per unary call. +func BenchmarkUnaryThroughProxy(b *testing.B) { + conn := benchRelayConn(b) + req := testutils.StartWorkflowRequest() + benchInvoke(b, conn, req) + + b.ReportAllocs() + for b.Loop() { + benchInvoke(b, conn, req) + } +} + +// BenchmarkUnaryDirect is the no-proxy baseline for BenchmarkUnaryThroughProxy: +// the same request to the same fake upstream over bufconn, with the standard +// proto codec, as an SDK client would speak to a Temporal Service directly. +func BenchmarkUnaryDirect(b *testing.B) { + upstreamLis := bufconn.Listen(1024 * 1024) + upstream := benchUpstreamServer() + serve(b, upstream, upstreamLis) + + conn := dialBufconn(b, upstreamLis) + b.Cleanup(func() { _ = conn.Close() }) + + req := testutils.StartWorkflowRequest() + benchInvoke(b, conn, req) + + b.ReportAllocs() + for b.Loop() { + benchInvoke(b, conn, req) + } +} diff --git a/internal/testutils/request.go b/internal/testutils/request.go new file mode 100644 index 0000000..13374d9 --- /dev/null +++ b/internal/testutils/request.go @@ -0,0 +1,32 @@ +// Package testutils holds helpers shared between the internal packages' test +// suites. +package testutils + +import ( + "go.temporal.io/api/common/v1" + "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" +) + +// StartWorkflowRequest builds a StartWorkflowExecutionRequest shaped like the +// ones SDK clients send: identity fields, a typed workflow and task queue, and +// a 1 KiB input payload, so extraction and translation measure a realistic +// unmarshal rather than a two-field toy message. +func StartWorkflowRequest() *workflowservice.StartWorkflowExecutionRequest { + input := make([]byte, 1024) + for i := range input { + input[i] = byte('a' + i%26) + } + + return &workflowservice.StartWorkflowExecutionRequest{ + Namespace: "payments-production", + WorkflowId: "billing-run-2026-10-10-abc123", + WorkflowType: &common.WorkflowType{Name: "billing.daily-run"}, + TaskQueue: &taskqueue.TaskQueue{Name: "billing-tq"}, + Input: &common.Payloads{Payloads: []*common.Payload{{ + Metadata: map[string][]byte{"encoding": []byte("json/plain")}, + Data: input, + }}}, + Identity: "1@worker-7", + } +} diff --git a/mise-tasks/test/bench b/mise-tasks/test/bench new file mode 100755 index 0000000..16b173c --- /dev/null +++ b/mise-tasks/test/bench @@ -0,0 +1,38 @@ +#!/usr/bin/env bash +#MISE description="Run benchmarks, optionally comparing against another revision" +#USAGE flag "-b --base " help="Compare the working tree against this revision with benchstat" +#USAGE flag "-c --count " default="10" +#USAGE flag "-t --time