Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -29,3 +29,7 @@ docs/
# build things
./proxy
dist/

# bench artifacts
bench-*.txt
benchstat-*.txt
50 changes: 50 additions & 0 deletions internal/protoutil/extractor_bench_test.go
Original file line number Diff line number Diff line change
@@ -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")
}
}
}
39 changes: 39 additions & 0 deletions internal/protoutil/translate_bench_test.go
Original file line number Diff line number Diff line change
@@ -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" })
}
}
2 changes: 1 addition & 1 deletion internal/router/compile_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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": {}}
Expand Down
89 changes: 89 additions & 0 deletions internal/router/director_bench_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
})
}
12 changes: 6 additions & 6 deletions internal/router/handler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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)
Expand All @@ -705,15 +705,15 @@ 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...)
}

// 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) {
Expand All @@ -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(
Expand All @@ -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,
Expand Down
Loading
Loading