[translation]: Add a library for substituting one gRPC method for ano…

internal/translation/doc.go

@@ -0,0 +1,20 @@
+// Package translation rewrites one gRPC method call into another on the hop to
+// the upstream.
+//
+// A [Translation] pairs an inbound method with the upstream method that stands
+// in for it, plus the conversions between their request and response types. A
+// [Registry] holds the set of them, and [UnaryClientInterceptor] applies it:
+// a call whose method is registered is converted, sent under the upstream
+// method, and converted back before the caller sees a reply. Everything else is
+// passed straight through.
+//
+// This is distinct from the namespace translation in internal/proxy, which
+// rewrites names inside a message but leaves the method alone. The two compose:
+// install method translation as the innermost interceptor so namespace
+// translation, payload codecs, and the reflective forwarder all keep seeing the
+// method and message types the caller actually asked for.
+//
+// The package is the mechanism only; it ships no translations of its own. A
+// [Translation] names the two methods and carries the conversions between their
+// message types, so what is translated lives with the domain that needs it.
+package translation

internal/translation/interceptor.go

@@ -0,0 +1,115 @@
+package translation
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+// Option configures the interceptor built by [UnaryClientInterceptor].
+type Option func(*options)
+
+type options struct {
+	via grpc.ClientConnInterface
+}
+
+// Via sends a translated call over cc instead of continuing down the chain to
+// the connection the interceptor is installed on.
+//
+// The upstream method belongs to a different service, which the connection that
+// received the call does not serve: the caller asked a Temporal frontend for
+// ListNamespaces, and only Temporal Cloud's control plane can answer it. Where
+// that service lives is as fixed as the conversions themselves, so the
+// translation carries the connection rather than the request being routed to it,
+// and a request reaching any upstream is answered the same way.
+func Via(cc grpc.ClientConnInterface) Option {
+	return func(o *options) { o.via = cc }
+}
+
+// DialOptions returns the dial options that install method translation on an
+// outbound connection. Callers fold them into the dial options for the upstream
+// connection, last, so translation is the innermost interceptor: every other
+// interceptor on the chain then sees the method and message types the caller
+// asked for rather than the substitute sent upstream.
+func DialOptions(r *Registry, opts ...Option) []grpc.DialOption {
+	return []grpc.DialOption{grpc.WithChainUnaryInterceptor(UnaryClientInterceptor(r, opts...))}
+}
+
+// UnaryClientInterceptor returns a unary client interceptor that replaces a call
+// to a method r translates with a call to the method it translates onto: the
+// request is converted, invoked under the upstream method, and the upstream's
+// reply is folded into the reply the caller allocated. A method r does not
+// translate is invoked unchanged, as is a call whose request or reply is not a
+// proto message. Any headers the translation declares are stamped on the
+// substituted call only.
+//
+// An upstream error is returned as it arrived, so the caller sees the upstream's
+// status rather than a translated one. A conversion that fails is Internal: the
+// mapping is compiled in, so a failure there is a proxy bug and not something
+// the caller did.
+func UnaryClientInterceptor(r *Registry, opts ...Option) grpc.UnaryClientInterceptor {
+	o := &options{}
+	for _, opt := range opts {
+		opt(o)
+	}
+
+	return func(
+		ctx context.Context,
+		method string,
+		req, reply any,
+		cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		callOpts ...grpc.CallOption,
+	) error {
+		t, ok := r.Lookup(method)
+		if !ok {
+			return invoker(ctx, method, req, reply, cc, callOpts...)
+		}
+
+		in, inOK := req.(proto.Message)
+		out, outOK := reply.(proto.Message)
+		if !inOK || !outOK {
+			// Nothing to convert against. The forwarder types every message from the
+			// proto registry, so this only happens on a hand-rolled call; forwarding
+			// it unchanged is closer to right than failing it.
+			return invoker(ctx, method, req, reply, cc, callOpts...)
+		}
+
+		upReq, err := t.request(in)
+		if err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+
+		upReply := t.reply()
+		if err := o.invoke(t.stamp(ctx), t, upReq, upReply, cc, invoker, callOpts...); err != nil {
+			return err
+		}
+
+		if err := t.response(in, upReply, out); err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+
+		return nil
+	}
+}
+
+// invoke sends the substituted call: over the connection Via named when there is
+// one, and otherwise down the chain to the connection the interceptor was
+// installed on.
+func (o *options) invoke(
+	ctx context.Context,
+	t *Translation,
+	req, reply any,
+	cc *grpc.ClientConn,
+	invoker grpc.UnaryInvoker,
+	callOpts ...grpc.CallOption,
+) error {
+	if o.via != nil {
+		return o.via.Invoke(ctx, t.to, req, reply, callOpts...)
+	}
+
+	return invoker(ctx, t.to, req, reply, cc, callOpts...)
+}

internal/translation/interceptor_test.go

@@ -0,0 +1,257 @@
+package translation
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	workflowservice "go.temporal.io/api/workflowservice/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+)
+
+// systemInfoService is a fake upstream serving the method the test translation
+// substitutes onto, recording the metadata it was called with.
+type systemInfoService struct {
+	workflowservice.UnimplementedWorkflowServiceServer
+
+	lis     net.Listener
+	version string
+
+	// mu guards md, which the serving goroutine writes while the test reads.
+	mu sync.Mutex
+	md metadata.MD
+}
+
+func testRegistry(t *testing.T, opts ...func(*Translation)) *Registry {
+	t.Helper()
+
+	tr := Adapt(fromMethod, toMethod, okRequest, okResponse)
+	for _, opt := range opts {
+		opt(tr)
+	}
+
+	r, err := NewRegistry(tr)
+	require.NoError(t, err)
+
+	return r
+}
+
+func TestUnaryClientInterceptorPassesUntranslatedMethodsThrough(t *testing.T) {
+	t.Parallel()
+
+	var gotMethod string
+	invoker := func(_ context.Context, method string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error {
+		gotMethod = method
+		return nil
+	}
+
+	const other = "/temporal.api.workflowservice.v1.WorkflowService/ListNamespaces"
+	err := UnaryClientInterceptor(testRegistry(t))(
+		t.Context(), other,
+		&workflowservice.ListNamespacesRequest{}, &workflowservice.ListNamespacesResponse{}, nil, invoker,
+	)
+	require.NoError(t, err)
+	require.Equal(t, other, gotMethod)
+}
+
+func TestUnaryClientInterceptorSubstitutesTheUpstreamCall(t *testing.T) {
+	t.Parallel()
+
+	var (
+		gotMethod string
+		gotReq    *workflowservice.GetSystemInfoRequest
+	)
+
+	invoker := func(_ context.Context, method string, req, reply any, _ *grpc.ClientConn, _ ...grpc.CallOption) error {
+		gotMethod = method
+		gotReq = mustProto[*workflowservice.GetSystemInfoRequest](t, req)
+		mustProto[*workflowservice.GetSystemInfoResponse](t, reply).ServerVersion = "1.2.3"
+
+		return nil
+	}
+
+	reply := &workflowservice.DescribeNamespaceResponse{}
+	err := UnaryClientInterceptor(testRegistry(t))(
+		t.Context(), fromMethod,
+		&workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply, nil, invoker,
+	)
+	require.NoError(t, err)
+
+	require.Equal(t, toMethod, gotMethod, "the upstream sees the substituted method")
+	require.NotNil(t, gotReq, "and the converted request")
+	require.Equal(t, "payments@1.2.3", reply.GetNamespaceInfo().GetName())
+}
+
+func TestUnaryClientInterceptorReturnsTheUpstreamError(t *testing.T) {
+	t.Parallel()
+
+	want := status.Error(codes.PermissionDenied, "no")
+	invoker := func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error {
+		return want
+	}
+
+	err := UnaryClientInterceptor(testRegistry(t))(
+		t.Context(), fromMethod,
+		&workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, nil, invoker,
+	)
+	require.Equal(t, want, err, "the caller sees the upstream's status, untranslated")
+}
+
+func TestUnaryClientInterceptorReportsConversionFailureAsInternal(t *testing.T) {
+	t.Parallel()
+
+	// A reply of the wrong type for the registered translation: the request
+	// converts, the call succeeds, and folding the reply back fails.
+	invoker := func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error {
+		return nil
+	}
+
+	err := UnaryClientInterceptor(testRegistry(t))(
+		t.Context(), fromMethod,
+		&workflowservice.DescribeNamespaceRequest{}, &workflowservice.ListNamespacesResponse{}, nil, invoker,
+	)
+	require.Equal(t, codes.Internal, status.Code(err))
+}
+
+func TestUnaryClientInterceptorForwardsNonProtoCallsUnchanged(t *testing.T) {
+	t.Parallel()
+
+	var gotMethod string
+	invoker := func(_ context.Context, method string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error {
+		gotMethod = method
+		return nil
+	}
+
+	err := UnaryClientInterceptor(testRegistry(t))(t.Context(), fromMethod, "req", "reply", nil, invoker)
+	require.NoError(t, err)
+	require.Equal(t, fromMethod, gotMethod, "nothing to convert, so nothing is substituted")
+}
+
+func TestUnaryClientInterceptorStampsHeadersOnTheSubstitutedCallOnly(t *testing.T) {
+	t.Parallel()
+
+	r := testRegistry(t, func(tr *Translation) { tr.WithHeader("x-api-version", "v1") })
+
+	var gotMD metadata.MD
+	invoker := func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error {
+		gotMD, _ = metadata.FromOutgoingContext(ctx)
+		return nil
+	}
+
+	err := UnaryClientInterceptor(r)(
+		t.Context(), fromMethod,
+		&workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, nil, invoker,
+	)
+	require.NoError(t, err)
+	require.Equal(t, []string{"v1"}, gotMD.Get("x-api-version"))
+
+	// An untranslated call keeps whatever the caller sent.
+	ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs("x-api-version", "caller"))
+	err = UnaryClientInterceptor(r)(
+		ctx, "/pkg.Other/Untranslated",
+		&workflowservice.DescribeNamespaceRequest{}, &workflowservice.DescribeNamespaceResponse{}, nil, invoker,
+	)
+	require.NoError(t, err)
+	require.Equal(t, []string{"caller"}, gotMD.Get("x-api-version"))
+}
+
+func TestViaSendsTheSubstitutedCallElsewhere(t *testing.T) {
+	t.Parallel()
+
+	// Via is what lets a translation answer over a connection other than the one
+	// it is installed on, for an upstream method the original connection's service
+	// does not serve.
+	elsewhere := newSystemInfoService(t, "9.9.9")
+
+	cc, err := grpc.NewClient(elsewhere.addr(t), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = cc.Close() })
+
+	var chained bool
+	invoker := func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error {
+		chained = true
+		return nil
+	}
+
+	reply := &workflowservice.DescribeNamespaceResponse{}
+	err = UnaryClientInterceptor(testRegistry(t), Via(cc))(
+		t.Context(), fromMethod,
+		&workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply, nil, invoker,
+	)
+	require.NoError(t, err)
+
+	require.False(t, chained, "the call must leave the chain rather than continue down it")
+	require.Equal(t, "payments@9.9.9", reply.GetNamespaceInfo().GetName())
+}
+
+func TestDialOptionsTranslateOverARealConnection(t *testing.T) {
+	t.Parallel()
+
+	upstream := newSystemInfoService(t, "1.2.3")
+
+	dialOpts := append(
+		[]grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())},
+		DialOptions(testRegistry(t, func(tr *Translation) { tr.WithHeader("x-api-version", "v1") }))...,
+	)
+	cc, err := grpc.NewClient(upstream.addr(t), dialOpts...)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = cc.Close() })
+
+	// Invoke the inbound method by name, the way the reflective forwarder does:
+	// the interceptor is what turns it into the substituted call.
+	reply := &workflowservice.DescribeNamespaceResponse{}
+	err = cc.Invoke(t.Context(), fromMethod, &workflowservice.DescribeNamespaceRequest{Namespace: "payments"}, reply)
+	require.NoError(t, err)
+
+	require.Equal(t, "payments@1.2.3", reply.GetNamespaceInfo().GetName())
+	require.Equal(t, []string{"v1"}, upstream.metadata().Get("x-api-version"), "the header reached the wire")
+}
+
+func (s *systemInfoService) GetSystemInfo(
+	ctx context.Context, _ *workflowservice.GetSystemInfoRequest,
+) (*workflowservice.GetSystemInfoResponse, error) {
+	md, _ := metadata.FromIncomingContext(ctx)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.md = md
+
+	return &workflowservice.GetSystemInfoResponse{ServerVersion: s.version}, nil
+}
+
+// newSystemInfoService starts the fake on a loopback port and stops it when the
+// test ends.
+func newSystemInfoService(t *testing.T, version string) *systemInfoService {
+	t.Helper()
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	svc := &systemInfoService{lis: lis, version: version}
+
+	svr := grpc.NewServer()
+	workflowservice.RegisterWorkflowServiceServer(svr, svc)
+	go func() { _ = svr.Serve(lis) }()
+	t.Cleanup(svr.Stop)
+
+	return svc
+}
+
+func (s *systemInfoService) addr(t *testing.T) string {
+	t.Helper()
+	return s.lis.Addr().String()
+}
+
+func (s *systemInfoService) metadata() metadata.MD {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.md.Copy()
+}