From 712678272a1dce80458a28168d983251c795e000 Mon Sep 17 00:00:00 2001 From: "David Muto (pseudomuto)" Date: Wed, 22 Jul 2026 09:30:38 -0400 Subject: [PATCH] [proxy]: Resolve upstreams per request from namespace and metadata The proxy could route to multiple named upstreams but only at fixed addresses. Migrations and multi-region setups may require one proxy to reach a different endpoint per namespace (e.g. Cloud per-namespace endpoints), where the dial address and TLS server name depend on the request's namespace and metadata. The router stamps the namespace it already extracts onto an internal metadata header (internal/transport/meta) so the typed per-upstream hop resolves without re-parsing payloads. connect.Conn wraps the connection pool as a grpc.ClientConnInterface whose target is chosen per call by a Resolver: a StaticResolver for fixed upstreams (dialed eagerly) or a DynamicResolver that renders hostPort/serverName and rebuilds credentials per request. The pool is now keyed by a logical key distinct from the dial target, so two connections to one address with different TLS server names don't collapse. A rendered address that is empty or malformed fails the request loudly instead of dialing garbage. For a templated mTLS upstream only the rendered server name varies per request, so rebuilding credentials would otherwise re-read and re-parse the certificate and CA files on every request. A shared creds.CertLoader loads that material once and caches it (safe to share concurrently, as it is immutable after the first load), so per-request resolution builds the dial options without touching disk. Fixed-address upstreams load once as before, so a rotated cert is picked up only on restart either way. This finalizes the upstream-routing RFC proposed in #32. --- e2e/harness_test.go | 14 +- e2e/templated_upstream_test.go | 83 ++++++++++ internal/config/config_test.go | 12 ++ internal/config/upstream.go | 11 +- internal/proxy/fx.go | 136 +++++++++++---- internal/proxy/fx_test.go | 23 ++- internal/proxy/resolver.go | 210 ++++++++++++++++++++++++ internal/proxy/resolver_test.go | 149 +++++++++++++++++ internal/proxy/server.go | 97 +++++------ internal/proxy/server_test.go | 76 ++++----- internal/proxy/translation_test.go | 30 +++- internal/router/fx.go | 6 +- internal/router/handler.go | 7 + internal/router/handler_test.go | 52 ++++++ internal/transport/connect/conn.go | 129 +++++++++++++++ internal/transport/connect/conn_test.go | 126 ++++++++++++++ internal/transport/connect/doc.go | 11 +- internal/transport/connect/pool.go | 67 ++++---- internal/transport/connect/pool_test.go | 104 ++++++++---- internal/transport/creds/mtls.go | 84 ++++++++-- internal/transport/creds/mtls_test.go | 51 ++++++ internal/transport/creds/tls.go | 19 ++- internal/transport/creds/tls_test.go | 24 ++- internal/transport/meta/meta.go | 47 ++++++ internal/transport/meta/meta_test.go | 44 +++++ rfc/routing.md | 116 ++++++++----- 26 files changed, 1431 insertions(+), 297 deletions(-) create mode 100644 e2e/templated_upstream_test.go create mode 100644 internal/proxy/resolver.go create mode 100644 internal/proxy/resolver_test.go create mode 100644 internal/transport/connect/conn.go create mode 100644 internal/transport/connect/conn_test.go create mode 100644 internal/transport/meta/meta.go create mode 100644 internal/transport/meta/meta_test.go diff --git a/e2e/harness_test.go b/e2e/harness_test.go index a2875e5..6ee3b35 100644 --- a/e2e/harness_test.go +++ b/e2e/harness_test.go @@ -115,17 +115,19 @@ func newFullApp(t *testing.T, cfg *config.Config) *fx.App { } // newProxyApp is a minimal fx app for the socket-level test in this package: it -// wires proxy.Module plus protoutil.Module (which provides the Translator -// proxy.Module requires). internal/proxy/fx_test.go has its own copy (with -// support for extra fx.Options) that its unit tests still depend on; this is a -// deliberate small duplication rather than an exported helper, so the two -// packages stay decoupled. +// wires proxy.Module plus protoutil.Module (which provides the Translator) and +// connect.Module (which provides the Pool), both required by proxy.Module. +// internal/proxy/fx_test.go has its own copy (with support for extra +// fx.Options) that its unit tests still depend on; this is a deliberate small +// duplication rather than an exported helper, so the two packages stay +// decoupled. func newProxyApp(t *testing.T, cfg *config.Config) *fx.App { t.Helper() return fx.New( fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), + connect.Module, protoutil.Module, proxy.Module, fx.NopLogger, @@ -191,7 +193,7 @@ func dialInbound(t *testing.T, addr string) *grpc.ClientConn { } // dialUnix returns a client connection to the proxy's unix socket for the -// given upstream host. The socket path matches what proxy.Start binds. +// given upstream host. The socket path matches what proxy.Listen binds. // internal/proxy/server_test.go has its own copy that its unit tests still // depend on; this is a deliberate small duplication for the socket-level test // in this package. diff --git a/e2e/templated_upstream_test.go b/e2e/templated_upstream_test.go new file mode 100644 index 0000000..7ae2c74 --- /dev/null +++ b/e2e/templated_upstream_test.go @@ -0,0 +1,83 @@ +package e2e + +import ( + "context" + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" + + "github.com/temporalio/temporal-proxy/internal/config" +) + +// TestEndToEndTemplatedUpstreamRoutesByRenderedAddress drives the full stack +// with a single templated upstream whose hostPort renders from request +// metadata, and proves each request reaches the upstream its metadata names. +// This exercises per-request address rendering end to end, and because two +// distinct rendered targets are both reached, it also proves the connection +// pool keys by rendered target rather than collapsing onto the first dial. +func TestEndToEndTemplatedUpstreamRoutesByRenderedAddress(t *testing.T) { + t.Parallel() + + svcA, addrA := newFakeUpstream(t) + svcB, addrB := newFakeUpstream(t) + + inboundAddr := freeTCPAddr(t) + app := newFullApp(t, &config.Config{ + Listen: config.ListenConfig{HostPort: inboundAddr}, + Routing: config.Routing{DefaultUpstream: "dynamic"}, + Upstreams: []config.Upstream{{ + Name: "dynamic", + Listen: config.ListenConfig{HostPort: `{{ index .Metadata "x-upstream" }}`}, + }}, + }) + require.NoError(t, app.Err()) + + startCtx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + require.NoError(t, app.Start(startCtx)) + t.Cleanup(func() { + stopCtx, stopCancel := context.WithTimeout(t.Context(), 5*time.Second) + defer stopCancel() + _ = app.Stop(stopCtx) + }) + + conn := dialInbound(t, inboundAddr) + defer func() { _ = conn.Close() }() + client := workflowservice.NewWorkflowServiceClient(conn) + + call := func(target string) { + ctx := metadata.AppendToOutgoingContext(startCtx, "x-upstream", target) + _, err := client.GetSystemInfo(ctx, &workflowservice.GetSystemInfoRequest{}, grpc.WaitForReady(true)) + require.NoError(t, err) + } + + call(addrA) + call(addrB) + + require.NotNil(t, svcA.received(), "the request naming upstream A must reach A") + require.NotNil(t, svcB.received(), "the request naming upstream B must reach B") + require.Equal(t, []string{addrA}, svcA.received().Get("x-upstream")) + require.Equal(t, []string{addrB}, svcB.received().Get("x-upstream")) +} + +// newFakeUpstream stands up a plaintext fake WorkflowService frontend and +// returns it with its dial address. +func newFakeUpstream(t *testing.T) (*capturingWorkflowService, string) { + t.Helper() + + svc := &capturingWorkflowService{} + srv := grpc.NewServer() + workflowservice.RegisterWorkflowServiceServer(srv, svc) + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + + return svc, lis.Addr().String() +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 59527a9..4f25529 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -359,6 +359,18 @@ func TestUpstream_IsTemplated(t *testing.T) { require.True(t, (&config.Upstream{Listen: config.ListenConfig{HostPort: "{{ .LocalNamespace }}.acme.cloud:7233"}}).IsTemplated()) require.False(t, (&config.Upstream{Listen: config.ListenConfig{HostPort: "127.0.0.1:7233"}}).IsTemplated()) + + // A templated TLS server name makes the upstream templated even when the + // hostPort is static. + require.True(t, (&config.Upstream{Listen: config.ListenConfig{ + HostPort: "127.0.0.1:7233", + TLS: &config.TLSConfig{ServerName: "{{ .RemoteNamespace }}.acme.cloud"}, + }}).IsTemplated()) + + require.False(t, (&config.Upstream{Listen: config.ListenConfig{ + HostPort: "127.0.0.1:7233", + TLS: &config.TLSConfig{ServerName: "static.acme.cloud"}, + }}).IsTemplated()) } func (e *errReader) Read(_ []byte) (int, error) { return 0, e.err } diff --git a/internal/config/upstream.go b/internal/config/upstream.go index abe1233..4c395a0 100644 --- a/internal/config/upstream.go +++ b/internal/config/upstream.go @@ -79,10 +79,15 @@ func (u *Upstream) Validate() error { ) } -// IsTemplated reports whether the upstream's hostPort contains a text/template -// action and so must be resolved per request. +// IsTemplated reports whether the upstream must be resolved per request because +// its hostPort, or its TLS server name when one is configured, contains a +// text/template action. func (u *Upstream) IsTemplated() bool { - return isTemplated(u.Listen.HostPort) + if isTemplated(u.Listen.HostPort) { + return true + } + + return u.Listen.TLS != nil && isTemplated(u.Listen.TLS.ServerName) } // Validate checks the namespace translation rules. diff --git a/internal/proxy/fx.go b/internal/proxy/fx.go index 02a588f..f8b9669 100644 --- a/internal/proxy/fx.go +++ b/internal/proxy/fx.go @@ -3,12 +3,15 @@ package proxy import ( "context" "fmt" + "slices" "go.uber.org/fx" + "google.golang.org/grpc" "github.com/temporalio/temporal-proxy/internal/auth" "github.com/temporalio/temporal-proxy/internal/config" "github.com/temporalio/temporal-proxy/internal/protoutil" + "github.com/temporalio/temporal-proxy/internal/transport/connect" "github.com/temporalio/temporal-proxy/internal/transport/creds" "github.com/temporalio/temporal-proxy/pkg/logger" ) @@ -17,48 +20,61 @@ import ( // and binds its lifecycle to the application. var Module = fx.Options(fx.Invoke(func(p ProxyParams) error { for i := range p.Config.Upstreams { - upstream := &p.Config.Upstreams[i] - - if err := upstream.Validate(); err != nil { + up := &p.Config.Upstreams[i] + if err := up.Validate(); err != nil { return fmt.Errorf("invalid upstream configuration: %w", err) } - if upstream.IsTemplated() { - return fmt.Errorf( - "upstream %q has a templated hostPort %q: templated upstreams are not yet supported", - upstream.Name, - upstream.Listen.HostPort, - ) - } - - opts := []Option{WithCredentials(upstreamCreds(upstream))} - - rules := &upstream.Namespaces.Rules + // Request-independent dial options: namespace translation and outbound + // credentials. Per-request credentials are added by the resolver. + var dialOpts []grpc.DialOption + rules := &up.Namespaces.Rules if rules.Configured() { - opts = append(opts, WithDialOptions(translationDialOptions(p.Translator, rules.Remote, rules.Local)...)) + dialOpts = append(dialOpts, translationDialOptions(p.Translator, rules.Remote, rules.Local)...) } - cp, err := auth.CredentialProviderFor(upstream.Credentials) + cp, err := auth.CredentialProviderFor(up.Credentials) if err != nil { - return fmt.Errorf("invalid credentials for upstream %q: %w", upstream.Name, err) + return fmt.Errorf("invalid credentials for upstream %q: %w", up.Name, err) } if cp != nil { - opts = append(opts, WithDialOptions(auth.DialOptions(cp)...)) + dialOpts = append(dialOpts, auth.DialOptions(cp)...) + } + + res, err := upstreamResolver(up, dialOpts) + if err != nil { + return err } + conn, err := connect.NewConn(p.Pool.ConnOrCreate, res) + if err != nil { + return err + } + + var opts []Option if p.Logger != nil { opts = append(opts, WithLogger(p.Logger)) } - svr, err := New(upstream.Listen.HostPort, opts...) + svr, err := New(up.Listen.HostPort, conn, opts...) if err != nil { - return fmt.Errorf("failed to create proxy for upstream %q: %w", upstream.Name, err) + return fmt.Errorf("failed to create proxy for upstream %q: %w", up.Name, err) } p.Lifecycle.Append(fx.Hook{ OnStart: func(context.Context) error { + // Bind synchronously so the socket is listening before the + // inbound server (whose OnStart runs after this one) starts + // routing requests to it; then serve in the background. + lis, err := svr.Listen(p.Context) + if err != nil { + return fmt.Errorf("failed to start proxy for upstream %q: %w", up.Name, err) + } + go func() { - if err := svr.Start(p.Context); err != nil { + defer func() { _ = lis.Close() }() + + if err := svr.Start(p.Context, lis); err != nil { // The proxy stopped serving unexpectedly. Bring the app // down rather than linger in a non-serving state; Start // has already logged the cause. @@ -76,9 +92,10 @@ var Module = fx.Options(fx.Invoke(func(p ProxyParams) error { })) // ProxyParams collects the fx-provided dependencies needed to construct and run -// the proxy [Server]. Context, Config, and Translator are required; Logger is -// optional and falls back to the default used by [New] when not supplied. -// [protoutil.Module] provides the Translator in the assembled application. +// the proxy [Server]. Context, Config, Translator, and Pool are required; +// Logger is optional and falls back to the default used by [New] when not +// supplied. [protoutil.Module] provides the Translator and [connect.Module] +// provides the Pool in the assembled application. type ProxyParams struct { fx.In Lifecycle fx.Lifecycle @@ -88,28 +105,83 @@ type ProxyParams struct { Context context.Context Config *config.Config Translator *protoutil.Translator + Pool *connect.Pool // Optional values Logger logger.Logger `optional:"true"` } +// upstreamResolver builds the [connect.Resolver] for an upstream. When neither +// the hostPort nor the TLS server name is templated it returns a static +// resolver whose connection is constructed eagerly (and reused for every +// request); otherwise it returns a DynamicResolver that renders the target and +// server name, and rebuilds credentials, per request. opts holds the +// request-independent dial options (namespace translation and outbound +// credentials). +func upstreamResolver(upstream *config.Upstream, opts []grpc.DialOption) (connect.Resolver, error) { + if upstream.IsTemplated() { + translator := func(s string) string { return s } + if upstream.Namespaces.Rules.Configured() { + translator = upstream.Namespaces.Rules.Remote + } + + // Share one loader across the per-request credentials so templated mTLS + // routing reads and parses the certificate and CA files once rather than + // on every request. Only ServerName varies per request. nil for non-mTLS + // upstreams, which have no files to load. + var loader *creds.CertLoader + if tls := upstream.Listen.TLS; tls != nil && tls.CA != "" { + loader = creds.NewCertLoader(tls.CA, tls.Cert, tls.Key) + } + + return NewDynamicResolver( + upstream, + WithRemoteNamespacer(translator), + WithOptionsFactory(func(data RouteData) ([]grpc.DialOption, error) { + creds, err := upstreamCreds(upstream, data.ResolvedServerName, loader).DialOption() + if err != nil { + return nil, err + } + + return append(slices.Clone(opts), creds), nil + }), + ) + } + + serverName := "" + if upstream.Listen.TLS != nil { + serverName = upstream.Listen.TLS.ServerName + } + + creds, err := upstreamCreds(upstream, serverName, nil).DialOption() + if err != nil { + return nil, fmt.Errorf("failed to build credentials for upstream %q: %w", upstream.Name, err) + } + + return connect.StaticResolver(upstream.Listen.HostPort, append(slices.Clone(opts), creds)...), nil +} + // upstreamCreds derives the credentials used to dial the upstream frontend from // the upstream TLS configuration: mutual TLS when a CA is set, server-verified // client TLS when TLS is configured without one, and insecure otherwise. -func upstreamCreds(upstream *config.Upstream) Credentials { +// serverName overrides the SNI/certificate-verification name; it is the +// upstream's static configured ServerName, or a per-request value rendered +// from a templated ServerName. loader, when non-nil, supplies the mTLS +// certificate material so it is loaded once and reused across the per-request +// credentials of a templated upstream; pass nil for a fixed-address upstream, +// whose credentials are built once. +func upstreamCreds(upstream *config.Upstream, serverName string, loader *creds.CertLoader) Credentials { tls := upstream.Listen.TLS if tls == nil { return creds.NewInsecure() } if tls.CA != "" { - return creds.NewMTLS( - tls.CA, - tls.Cert, - tls.Key, - creds.MTLSOptions{ServerName: tls.ServerName}, - ) + return creds.NewMTLS(tls.CA, tls.Cert, tls.Key, creds.MTLSOptions{ + ServerName: serverName, + Loader: loader, + }) } - return creds.NewClientTLS() + return creds.NewClientTLS(serverName) } diff --git a/internal/proxy/fx_test.go b/internal/proxy/fx_test.go index 6ce3006..2edc5e6 100644 --- a/internal/proxy/fx_test.go +++ b/internal/proxy/fx_test.go @@ -16,6 +16,7 @@ import ( "github.com/temporalio/temporal-proxy/internal/config" "github.com/temporalio/temporal-proxy/internal/protoutil" "github.com/temporalio/temporal-proxy/internal/proxy" + "github.com/temporalio/temporal-proxy/internal/transport/connect" "github.com/temporalio/temporal-proxy/pkg/logger" "github.com/temporalio/temporal-proxy/pkg/testutil" "github.com/temporalio/temporal-proxy/pkg/validation" @@ -155,7 +156,7 @@ func TestModuleWithUpstreamCredentials(t *testing.T) { require.NoError(t, app.Err()) } -func TestModuleRejectsTemplatedUpstream(t *testing.T) { +func TestModuleAcceptsTemplatedUpstream(t *testing.T) { t.Parallel() app := newProxyApp(t, &config.Config{ @@ -164,8 +165,19 @@ func TestModuleRejectsTemplatedUpstream(t *testing.T) { }, }) - require.Error(t, app.Err()) - require.ErrorContains(t, app.Err(), "templated") + // Previously this failed with "templated upstreams are not yet supported". + // A templated hostPort is now resolved per-request via a templatedPlan, so + // construction (and the full start/stop lifecycle) succeeds without ever + // dialing the upstream. + require.NoError(t, app.Err()) + + startCtx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + require.NoError(t, app.Start(startCtx)) + + stopCtx, stopCancel := context.WithTimeout(t.Context(), 5*time.Second) + defer stopCancel() + require.NoError(t, app.Stop(stopCtx)) } func TestModuleInstallsNamespaceTranslation(t *testing.T) { @@ -190,14 +202,12 @@ func TestModuleInstallsNamespaceTranslation(t *testing.T) { func TestModuleRequiresTranslator(t *testing.T) { t.Parallel() - // proxy.Module requires a *protoutil.Translator. Without protoutil.Module - // (or another provider) the app must fail to build rather than run without - // one. app := fx.New( fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(&config.Config{ Upstreams: []config.Upstream{{Name: "primary", Listen: config.ListenConfig{HostPort: "127.0.0.1:47242"}}}, }), + connect.Module, proxy.Module, fx.NopLogger, ) @@ -211,6 +221,7 @@ func newProxyApp(t *testing.T, cfg *config.Config, opts ...fx.Option) *fx.App { base := []fx.Option{ fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), + connect.Module, protoutil.Module, proxy.Module, fx.NopLogger, diff --git a/internal/proxy/resolver.go b/internal/proxy/resolver.go new file mode 100644 index 0000000..64c9cc3 --- /dev/null +++ b/internal/proxy/resolver.go @@ -0,0 +1,210 @@ +package proxy + +import ( + "context" + "fmt" + "net" + "slices" + "strings" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/template" + "github.com/temporalio/temporal-proxy/internal/transport/meta" +) + +type ( + // DynamicResolver is a [connect.Resolver] that renders an upstream's dial + // target (and optional TLS server name) per request from the local + // namespace and request metadata. It always reports IsStatic as false, so a + // [connect.Conn] built from it resolves lazily on every call. A non-templated + // hostPort renders to itself, so a DynamicResolver also serves upstreams with + // a fixed address. Construct one with NewDynamicResolver. + DynamicResolver struct { + name string + host *template.Template[template.UpstreamContext] + serverName *template.Template[template.UpstreamContext] + remote func(string) string + opts func(d RouteData) ([]grpc.DialOption, error) + } + + // RouteData is passed to the options factory once a request has been + // resolved. It carries the template context used for rendering plus the + // resolved TLS server name, so the factory can build dial options (e.g. + // credentials whose SNI depends on the rendered server name). + RouteData struct { + template.UpstreamContext + ResolvedServerName string + } + + // ResolverOption configures a DynamicResolver at construction. + ResolverOption func(*DynamicResolver) +) + +// NewDynamicResolver builds a DynamicResolver for up. It compiles the hostPort +// and TLS server-name templates (failing if either is malformed) and applies +// opts. By default the remote namespace equals the local one and no dial options +// are added; use WithRemoteNamespacer and WithOptionsFactory to change that. +func NewDynamicResolver(up *config.Upstream, opts ...ResolverOption) (*DynamicResolver, error) { + host, err := template.ParseUpstream(up.Listen.HostPort) + if err != nil { + return nil, fmt.Errorf("failed to parse upstream host template %q: %w", up.Name, err) + } + + tsn := "" + if up.Listen.TLS != nil { + tsn = up.Listen.TLS.ServerName + } + + serverName, err := template.ParseUpstream(tsn) + if err != nil { + return nil, fmt.Errorf("failed to parse upstream TLS server name template %q: %w", up.Name, err) + } + + r := &DynamicResolver{ + name: up.Name, + host: host, + serverName: serverName, + remote: func(s string) string { return s }, + opts: func(RouteData) ([]grpc.DialOption, error) { return nil, nil }, + } + + for _, opt := range opts { + opt(r) + } + + return r, nil +} + +// WithRemoteNamespacer sets the function that maps the local namespace to the +// remote one, making RemoteNamespace available to the templates. +func WithRemoteNamespacer(f func(string) string) ResolverOption { + return func(r *DynamicResolver) { r.remote = f } +} + +// WithOptionsFactory sets the function that produces the dial options for a +// resolved request. It receives the rendered host and server name via RouteData. +func WithOptionsFactory(f func(RouteData) ([]grpc.DialOption, error)) ResolverOption { + return func(r *DynamicResolver) { r.opts = f } +} + +// IsStatic reports that a DynamicResolver always resolves per request. +func (r *DynamicResolver) IsStatic() bool { + return false +} + +// Resolve renders the dial target and server name from ctx and returns the pool +// cache key, the dial target, and the dial options. The cache key combines the +// target and rendered server name so that two requests to the same address with +// different server names get distinct pooled connections. It fails with +// codes.Internal (naming the upstream and template) when a template fails to +// render, the rendered address is empty or malformed, or the options factory +// errors; nothing is dialed in those cases. +func (r *DynamicResolver) Resolve(ctx context.Context) (string, string, []grpc.DialOption, error) { + localNS := meta.NamespaceFrom(ctx) + data := template.UpstreamContext{ + LocalNamespace: localNS, + RemoteNamespace: r.remote(localNS), + Metadata: lastMetadataValues(ctx), + } + + target, err := r.host.Render(data) + if err != nil { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q failed to render hostPort %q: %v", + r.name, + r.host.String(), + err, + ) + } + + if invalidAddress(target) { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q rendered invalid hostPort %q from template %q", + r.name, + target, + r.host.String(), + ) + } + + svrName, err := r.serverName.Render(data) + if err != nil { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q failed to render serverName %q: %v", + r.name, + r.serverName.String(), + err, + ) + } + + opts, err := r.opts(RouteData{ + UpstreamContext: data, + ResolvedServerName: svrName, + }) + if err != nil { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q failed to build dial options: %v", + r.name, + err, + ) + } + + return target + "\x00" + svrName, target, opts, nil +} + +// lastMetadataValues flattens the request's outgoing metadata to one value per +// key, keeping the last (most recently added) value. gRPC lowercases metadata +// keys, so template lookups must use lowercase keys. +func lastMetadataValues(ctx context.Context) map[string]string { + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + return nil + } + + out := make(map[string]string, len(md)) + for k, v := range md { + if len(v) > 0 { + out[k] = v[len(v)-1] + } + } + + return out +} + +// invalidAddress reports whether addr is unusable as a dial target: empty, or +// missing a host label (e.g. ":7233" or ".acme:7233" produced when a referenced +// namespace or metadata value renders empty). It strips a leading gRPC scheme +// ("dns:///", "passthrough:///") and tolerates a single trailing dot so a +// fully-qualified name ("acme.example.:7233") is accepted. +func invalidAddress(addr string) bool { + if strings.TrimSpace(addr) == "" { + return true + } + + hp := addr + if i := strings.Index(hp, "://"); i >= 0 { + hp = hp[i+3:] + hp = strings.TrimPrefix(hp, "/") // handle the "dns:///" triple-slash form + } + + host, _, err := net.SplitHostPort(hp) + if err != nil { + host = hp // no port present; treat the whole remainder as the host + } + + host = strings.TrimSpace(host) + if host == "" { + return true + } + + host = strings.TrimSuffix(host, ".") + return slices.Contains(strings.Split(host, "."), "") +} diff --git a/internal/proxy/resolver_test.go b/internal/proxy/resolver_test.go new file mode 100644 index 0000000..f53b13c --- /dev/null +++ b/internal/proxy/resolver_test.go @@ -0,0 +1,149 @@ +package proxy_test + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" + "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" + + "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/proxy" + "github.com/temporalio/temporal-proxy/internal/transport/meta" +) + +func TestDynamicResolverIsStatic(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver(upstreamWith("u", "host:7233", "")) + require.NoError(t, err) + require.False(t, res.IsStatic()) +} + +func TestNewDynamicResolverRejectsBadTemplates(t *testing.T) { + t.Parallel() + + _, err := proxy.NewDynamicResolver(upstreamWith("u", "{{ .Nope }}:7233", "")) + require.Error(t, err) + require.ErrorContains(t, err, "host template") + + _, err = proxy.NewDynamicResolver(upstreamWith("u", "host:7233", "{{ .Nope }}")) + require.Error(t, err) + require.ErrorContains(t, err, "server name template") +} + +func TestDynamicResolverResolve(t *testing.T) { + t.Parallel() + + up := upstreamWith("cloud", "{{ .RemoteNamespace }}.acme.example:7233", "{{ .RemoteNamespace }}.sni.example") + + var got proxy.RouteData + res, err := proxy.NewDynamicResolver( + up, + proxy.WithRemoteNamespacer(func(s string) string { return s + "-remote" }), + proxy.WithOptionsFactory(func(d proxy.RouteData) ([]grpc.DialOption, error) { + got = d + return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil + }), + ) + require.NoError(t, err) + + key, target, opts, err := res.Resolve(meta.WithNamespace(t.Context(), "orders")) + require.NoError(t, err) + + require.Equal(t, "orders-remote.acme.example:7233", target) + require.Equal(t, "orders-remote.acme.example:7233\x00orders-remote.sni.example", key, + "cache key combines target and rendered server name") + require.Len(t, opts, 1) + + // The options factory sees the rendered server name and namespaces. + require.Equal(t, "orders-remote.sni.example", got.ResolvedServerName) + require.Equal(t, "orders", got.LocalNamespace) + require.Equal(t, "orders-remote", got.RemoteNamespace) +} + +func TestDynamicResolverUsesLastMetadataValue(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver(upstreamWith("m", `{{ index .Metadata "dc" }}.acme:7233`, "")) + require.NoError(t, err) + + md := metadata.MD{} + md.Append("dc", "old", "new") + ctx := metadata.NewOutgoingContext(t.Context(), md) + + _, target, _, err := res.Resolve(ctx) + require.NoError(t, err) + require.Equal(t, "new.acme:7233", target, "the last metadata value wins") +} + +func TestDynamicResolverFailsLoudOnOptionsError(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver( + upstreamWith("u", "host:7233", ""), + proxy.WithOptionsFactory(func(proxy.RouteData) ([]grpc.DialOption, error) { + return nil, errors.New("boom") + }), + ) + require.NoError(t, err) + + _, _, _, err = res.Resolve(meta.WithNamespace(t.Context(), "orders")) + require.Error(t, err) + require.Equal(t, codes.Internal, status.Code(err)) + require.ErrorContains(t, err, "failed to build dial options") +} + +func TestDynamicResolverRejectsInvalidAddresses(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + host string + wantErr bool + }{ + {"empty metadata leaves an empty label", `{{ index .Metadata "dc" }}.acme:7233`, true}, + {"leading dot", ".acme:7233", true}, + {"interior double dot", "acme..example:7233", true}, + {"missing host", ":7233", true}, + {"normal host", "orders.acme.example:7233", false}, + {"trailing-dot fqdn", "acme.example.:7233", false}, + {"ipv6 literal", "[::1]:7233", false}, + {"scheme prefixed", "dns:///host:7233", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver(upstreamWith("u", tt.host, "")) + require.NoError(t, err) + + _, _, _, err = res.Resolve(meta.WithNamespace(t.Context(), "orders")) + if tt.wantErr { + require.Error(t, err) + require.Equal(t, codes.Internal, status.Code(err)) + return + } + + require.NoError(t, err) + }) + } +} + +func upstreamWith(name, hostPort, serverName string) *config.Upstream { + up := &config.Upstream{ + Name: name, + Listen: config.ListenConfig{HostPort: hostPort}, + } + + if serverName != "" { + up.Listen.TLS = &config.TLSConfig{ServerName: serverName} + } + + return up +} diff --git a/internal/proxy/server.go b/internal/proxy/server.go index 3d678f0..b8e11c4 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -24,56 +24,35 @@ type ( DialOption() (grpc.DialOption, error) } - // Server proxies the Temporal WorkflowService. It dials an upstream frontend - // and re-serves it on a local unix socket, letting local workers connect - // without TLS while the upstream hop stays secured. + // Server proxies the Temporal WorkflowService. It re-serves an upstream + // frontend on a local unix socket, letting local workers connect without TLS + // while the upstream hop stays secured. The upstream connection(s) it + // forwards to are owned by the shared [connect.Pool], not by this Server. Server struct { svr *server.Server - conn *grpc.ClientConn - - host string // upstream hostPort path string // path to unix socket } // Options configures a [Server] at construction time. Options struct { - creds Credentials - logger logger.Logger - dialOpts []grpc.DialOption + logger logger.Logger } // Option configures a [Server] via [New]. Option func(*Options) ) -// New constructs a [Server] that forwards WorkflowService traffic to the -// upstream frontend at hostPort. The local listener is a unix socket whose path -// is derived from hostPort (see [Server.Start]). With no options it dials the -// upstream with insecure credentials and logs via a CLI logger. -func New(hostPort string, opts ...Option) (*Server, error) { - pops := &Options{ - creds: creds.NewInsecure(), - logger: logger.Default(), - } +// New constructs a Server that forwards WorkflowService traffic to the +// upstream reachable through cc. The local listener is a unix socket whose +// path is derived from hostPort. cc is typically a resolvingConn; the +// connection(s) it uses are owned by the shared pool, not by this Server. +func New(hostPort string, cc grpc.ClientConnInterface, opts ...Option) (*Server, error) { + pops := &Options{logger: logger.Default()} for _, opt := range opts { opt(pops) } - upstreamCreds, err := pops.creds.DialOption() - if err != nil { - return nil, fmt.Errorf("failed to generate outbound credentials: %w", err) - } - - dialOpts := append([]grpc.DialOption{upstreamCreds}, pops.dialOpts...) - - conn, err := grpc.NewClient(hostPort, dialOpts...) - if err != nil { - return nil, fmt.Errorf("failed to dial: %s, %w", hostPort, err) - } - - wfs, err := proxy.NewWorkflowServiceProxyServer(proxy.WorkflowServiceProxyOptions{ - Client: workflowservice.NewWorkflowServiceClient(conn), - }) + wfs, err := proxy.NewWorkflowServiceProxyServer(workflowProxyOptions(cc)) if err != nil { return nil, fmt.Errorf("failed to create workflowservice proxy: %w", err) } @@ -95,18 +74,7 @@ func New(hostPort string, opts ...Option) (*Server, error) { return nil, fmt.Errorf("failed to resolve socket path: %w", err) } - return &Server{ - svr: svr, - conn: conn, - host: hostPort, - path: path, - }, nil -} - -// WithCredentials sets the transport credentials used to dial the upstream -// frontend. -func WithCredentials(creds Credentials) Option { - return Option(func(o *Options) { o.creds = creds }) + return &Server{svr: svr, path: path}, nil } // WithLogger sets the logger used by the proxy. @@ -114,28 +82,30 @@ func WithLogger(log logger.Logger) Option { return Option(func(o *Options) { o.logger = log }) } -// WithDialOptions appends gRPC dial options applied to the outbound connection -// to the upstream frontend, in addition to the transport credentials. Outbound -// credentials and namespace translation are both wired this way. -func WithDialOptions(opts ...grpc.DialOption) Option { - return Option(func(o *Options) { o.dialOpts = append(o.dialOpts, opts...) }) -} - -// Start binds the local unix socket and serves until the proxy is stopped or -// ctx is cancelled. It first removes any socket left behind by a prior run. It -// blocks, so callers typically run it in its own goroutine. -func (s *Server) Start(ctx context.Context) error { +// Listen removes any socket left behind by a prior run and binds the proxy's +// local unix socket, returning the listener. Binding is separate from Start so +// callers can bind synchronously during startup (the socket is then listening, +// and the OS backlogs connections) before serving in the background, ensuring +// no request is routed to an unbound socket. +func (s *Server) Listen(ctx context.Context) (net.Listener, error) { // Remove any socket left behind by a prior run; otherwise the bind fails // with "address already in use". if err := os.Remove(s.path); err != nil && !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("failed to remove stale socket: unix://%s, %w", s.path, err) + return nil, fmt.Errorf("failed to remove stale socket: unix://%s, %w", s.path, err) } lis, err := (&net.ListenConfig{}).Listen(ctx, "unix", s.path) if err != nil { - return fmt.Errorf("failed to bind to socket: unix://%s, %w", s.path, err) + return nil, fmt.Errorf("failed to bind to socket: unix://%s, %w", s.path, err) } + return lis, nil +} + +// Start serves on lis until the proxy is stopped or ctx is cancelled. It +// blocks, so callers typically run it in its own goroutine after binding the +// listener with Listen. +func (s *Server) Start(ctx context.Context, lis net.Listener) error { return s.svr.Start(ctx, lis) } @@ -145,5 +115,14 @@ func (s *Server) Stop(ctx context.Context) error { return fmt.Errorf("failed to stop GRPC server: %w", err) } - return s.conn.Close() + return nil +} + +// workflowProxyOptions builds the options for the WorkflowService proxy +// server backed by cc. DisableHeaderForwarding is intentionally left false: +// the upstream proxy must forward incoming metadata (including the +// router-stamped namespace) onto the outbound call, since templated upstream +// resolution and namespace translation both depend on it. +func workflowProxyOptions(cc grpc.ClientConnInterface) proxy.WorkflowServiceProxyOptions { + return proxy.WorkflowServiceProxyOptions{Client: workflowservice.NewWorkflowServiceClient(cc)} } diff --git a/internal/proxy/server_test.go b/internal/proxy/server_test.go index 74c6694..e8fbdd5 100644 --- a/internal/proxy/server_test.go +++ b/internal/proxy/server_test.go @@ -2,7 +2,6 @@ package proxy_test import ( "context" - "errors" "os" "path/filepath" "testing" @@ -13,55 +12,21 @@ import ( "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/health/grpc_health_v1" - "github.com/temporalio/temporal-proxy/internal/auth" "github.com/temporalio/temporal-proxy/internal/proxy" - "github.com/temporalio/temporal-proxy/internal/transport/creds" "github.com/temporalio/temporal-proxy/internal/transport/socket" "github.com/temporalio/temporal-proxy/pkg/logger" ) -type failingCredentials struct { - err error -} - func TestNew(t *testing.T) { t.Parallel() t.Run("returns a server with default options", func(t *testing.T) { t.Parallel() - svr, err := proxy.New("127.0.0.1:7233") + svr, err := proxy.New("127.0.0.1:7233", upstreamConn(t, "127.0.0.1:7233")) require.NoError(t, err) require.NotNil(t, svr) }) - - t.Run("propagates credential errors", func(t *testing.T) { - t.Parallel() - - svr, err := proxy.New( - "127.0.0.1:7233", - proxy.WithCredentials(failingCredentials{err: errors.New("boom")}), - ) - require.Error(t, err) - require.Nil(t, svr) - require.ErrorContains(t, err, "outbound credentials") - require.ErrorContains(t, err, "boom") - }) -} - -func TestNewWithDialOptions(t *testing.T) { - t.Parallel() - - cp, err := auth.NewStaticCredentialProvider("k", "", "") - require.NoError(t, err) - - svr, err := proxy.New( - "127.0.0.1:7233", - proxy.WithCredentials(creds.NewClientTLS()), - proxy.WithDialOptions(auth.DialOptions(cp)...), - ) - require.NoError(t, err) - require.NotNil(t, svr) } func TestServerStartAndStop(t *testing.T) { @@ -73,14 +38,17 @@ func TestServerStartAndStop(t *testing.T) { const upstream = "127.0.0.1:17233" log := logger.NewTestLogger() - svr, err := proxy.New(upstream, proxy.WithLogger(log)) + svr, err := proxy.New(upstream, upstreamConn(t, upstream), proxy.WithLogger(log)) require.NoError(t, err) ctx, cancel := context.WithCancel(t.Context()) defer cancel() + lis, err := svr.Listen(ctx) + require.NoError(t, err) + errCh := make(chan error, 1) - go func() { errCh <- svr.Start(ctx) }() + go func() { errCh <- svr.Start(ctx, lis) }() conn := dialUnix(t, upstream) defer func() { _ = conn.Close() }() @@ -119,14 +87,17 @@ func TestStartRemovesStaleSocket(t *testing.T) { require.NoError(t, os.WriteFile(path, []byte("stale"), 0o600)) t.Cleanup(func() { _ = os.Remove(path) }) - svr, err := proxy.New(upstream) + svr, err := proxy.New(upstream, upstreamConn(t, upstream)) require.NoError(t, err) ctx, cancel := context.WithCancel(t.Context()) defer cancel() + lis, err := svr.Listen(ctx) + require.NoError(t, err) + errCh := make(chan error, 1) - go func() { errCh <- svr.Start(ctx) }() + go func() { errCh <- svr.Start(ctx, lis) }() conn := dialUnix(t, upstream) defer func() { _ = conn.Close() }() @@ -142,7 +113,7 @@ func TestStartRemovesStaleSocket(t *testing.T) { require.NoError(t, <-errCh) } -func TestStartReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { +func TestListenReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { t.Parallel() const upstream = "127.0.0.1:37233" @@ -150,26 +121,37 @@ func TestStartReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { path, err := socket.UnixPath(upstream) require.NoError(t, err) - // A non-empty directory at the socket path makes os.Remove fail, so Start + // A non-empty directory at the socket path makes os.Remove fail, so Listen // returns before it ever binds. require.NoError(t, os.Mkdir(path, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(path, "child"), nil, 0o600)) t.Cleanup(func() { _ = os.RemoveAll(path) }) - svr, err := proxy.New(upstream) + svr, err := proxy.New(upstream, upstreamConn(t, upstream)) require.NoError(t, err) - err = svr.Start(t.Context()) + _, err = svr.Listen(t.Context()) require.Error(t, err) require.ErrorContains(t, err, "failed to remove stale socket") } -func (f failingCredentials) DialOption() (grpc.DialOption, error) { - return nil, f.err +// upstreamConn returns a grpc.ClientConnInterface for New's cc argument. gRPC +// dials lazily, so this never opens a socket to upstream; the tests in this +// file never make an outbound RPC through it (they only exercise the local +// unix listener), so a plain client conn stands in for the pool-backed +// resolvingConn used in production. +func upstreamConn(t *testing.T, upstream string) grpc.ClientConnInterface { + t.Helper() + + conn, err := grpc.NewClient(upstream, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn } // dialUnix returns a client connection to the proxy's unix socket for the given -// upstream host. The socket path matches what proxy.Start binds. +// upstream host. The socket path matches what proxy.Listen binds. func dialUnix(t *testing.T, upstream string) *grpc.ClientConn { t.Helper() diff --git a/internal/proxy/translation_test.go b/internal/proxy/translation_test.go index f7ff37c..b21fc51 100644 --- a/internal/proxy/translation_test.go +++ b/internal/proxy/translation_test.go @@ -37,6 +37,15 @@ type ( } ) +func TestWorkflowProxyOptionsForwardsHeaders(t *testing.T) { + t.Parallel() + + // Guard: New must not disable header forwarding, or the upstream proxy would + // drop the router-stamped namespace header and templated resolution would + // break. + require.False(t, workflowProxyOptions(nil).DisableHeaderForwarding) +} + func TestUnaryClientInterceptorTranslatesBothDirections(t *testing.T) { t.Parallel() @@ -132,20 +141,29 @@ func TestOutboundNamespaceTranslation(t *testing.T) { go func() { _ = upstream.Serve(lis) }() t.Cleanup(upstream.Stop) - // Proxy dials the fake upstream, wrapping "local" -> "remote-local". + // Dial the fake upstream directly, wrapping "local" -> "remote-local"; this + // mirrors how fx folds translationDialOptions into a plan's dial options + // before the pool dials them. translator := protoutil.NewTranslator(protoregistry.GlobalFiles) - svr, err := New( - lis.Addr().String(), - WithDialOptions(translationDialOptions( + dialOpts := append( + []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, + translationDialOptions( translator, func(s string) string { return "remote-" + s }, func(s string) string { return "local-" + s }, - )...), + )..., ) + cc, err := grpc.NewClient(lis.Addr().String(), dialOpts...) + require.NoError(t, err) + t.Cleanup(func() { _ = cc.Close() }) + + svr, err := New(lis.Addr().String(), cc) require.NoError(t, err) ctx := t.Context() - go func() { _ = svr.Start(ctx) }() + srvLis, err := svr.Listen(ctx) + require.NoError(t, err) + go func() { _ = svr.Start(ctx, srvLis) }() t.Cleanup(func() { _ = svr.Stop(context.Background()) }) conn := dialUnixSocket(t, lis.Addr().String()) diff --git a/internal/router/fx.go b/internal/router/fx.go index 8c4897d..5a3e759 100644 --- a/internal/router/fx.go +++ b/internal/router/fx.go @@ -37,10 +37,8 @@ var Module = fx.Options(fx.Provide( return nil, fmt.Errorf("failed to resolve proxy socket path[%q]: %w", upstream.Name, err) } - conn, err := p.Pool.GetOrSet( - "unix://"+sockPath, - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) + sock := "unix://" + sockPath + conn, err := p.Pool.ConnOrCreate(sock, sock, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { return nil, fmt.Errorf("failed to create upstream client[%q]: %w", upstream.Name, err) } diff --git a/internal/router/handler.go b/internal/router/handler.go index 2e0acef..604ae32 100644 --- a/internal/router/handler.go +++ b/internal/router/handler.go @@ -9,6 +9,8 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" + + "github.com/temporalio/temporal-proxy/internal/transport/meta" ) type ( @@ -76,6 +78,11 @@ func Handler(d Director, r Reflector, rep *Reporter) grpc.StreamHandler { namespace = r.Namespace(method, first.payload) } + // Carry the extracted namespace to the upstream proxy so it can resolve a + // templated address without re-parsing the payload. Set (not append) so a + // client-supplied value cannot influence routing. + outCtx = meta.WithNamespace(outCtx, namespace) + target, err := d.Resolve(ctx, method, namespace, maps.Clone(md)) if err != nil { return err diff --git a/internal/router/handler_test.go b/internal/router/handler_test.go index 5d8563d..00f4080 100644 --- a/internal/router/handler_test.go +++ b/internal/router/handler_test.go @@ -9,6 +9,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" @@ -27,6 +28,7 @@ import ( "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/internal/router" "github.com/temporalio/temporal-proxy/internal/server" + "github.com/temporalio/temporal-proxy/internal/transport/meta" ) type ( @@ -448,6 +450,34 @@ tmprl_proxy_router_forwarding_errors_total{reason="stream_setup",upstream="prima `), "tmprl_proxy_router_forwarding_errors_total")) } +func TestHandlerStampsNamespace(t *testing.T) { + t.Parallel() + + upstream, got := capturingUpstream(t) + + m, _ := newTestReporter(t) + relayLis := bufconn.Listen(1024 * 1024) + relay := grpc.NewServer( + grpc.ForceServerCodecV2(router.Codec()), + grpc.UnknownServiceHandler(router.Handler(stubDirector{upstream: "u", cc: upstream}, &recordingReflector{ns: "orders"}, m)), + ) + serve(t, relay, relayLis) + + client := dialBufconn(t, relayLis) + + // The client sends a spoofed namespace header; the router must overwrite it + // with the value it extracted via the Reflector before forwarding. + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(meta.NamespaceHeader, "spoofed")) + _ = client.Invoke(ctx, "/test.v1.Echo/Ping", &grpc_health_v1.HealthCheckRequest{}, new(grpc_health_v1.HealthCheckResponse)) + + select { + case ns := <-got: + require.Equal(t, "orders", ns) + case <-time.After(2 * time.Second): + t.Fatal("upstream never received a request") + } +} + func TestStatusError(t *testing.T) { t.Parallel() @@ -514,6 +544,28 @@ func (d *recordingDirector) snapshot() (calls int, method, namespace string, md return d.calls, d.method, d.namespace, d.md } +// capturingUpstream is an in-process gRPC server that records the namespace +// header from the first stream it receives. +func capturingUpstream(t *testing.T) (*grpc.ClientConn, <-chan string) { + t.Helper() + + got := make(chan string, 1) + lis := bufconn.Listen(1024 * 1024) + srv := grpc.NewServer(grpc.UnknownServiceHandler(func(_ any, ss grpc.ServerStream) error { + md, _ := metadata.FromIncomingContext(ss.Context()) + vals := md.Get(meta.NamespaceHeader) + if len(vals) == 0 { + got <- "" + } else { + got <- vals[len(vals)-1] + } + return nil + })) + serve(t, srv, lis) + + return dialBufconn(t, lis), got +} + func dialBufconn(t *testing.T, lis *bufconn.Listener) *grpc.ClientConn { t.Helper() conn, err := grpc.NewClient( diff --git a/internal/transport/connect/conn.go b/internal/transport/connect/conn.go new file mode 100644 index 0000000..1b4d821 --- /dev/null +++ b/internal/transport/connect/conn.go @@ -0,0 +1,129 @@ +package connect + +import ( + "context" + "fmt" + + "google.golang.org/grpc" +) + +type ( + // Conn is a [grpc.ClientConnInterface] that resolves its dial target per + // call through a Resolver and fetches (lazily creating) the underlying + // pooled connection through a ConnFactory. With a dynamic Resolver a single + // Conn fronts many physical connections (e.g. one per namespace); with a + // static Resolver it always resolves to the same one. Construct one with + // NewConn. + Conn struct { + factory ConnFactory + resolver Resolver + } + + // ConnFactory returns the pooled connection for a (key, target) pair, + // creating it on first use. [Pool.ConnOrCreate] satisfies this signature: + // the first argument is the logical cache key and the second is the dial + // address. + ConnFactory func(string, string, ...grpc.DialOption) (*grpc.ClientConn, error) + + // Resolver decides, per request, which connection a Conn should use. Resolve + // returns the pool cache key, the dial target, and the dial options for that + // connection. IsStatic reports whether the resolution is fixed for the life + // of the Conn: a static resolver is dialed eagerly when the Conn is created; + // a dynamic one is resolved lazily on every call. + Resolver interface { + IsStatic() bool + Resolve(context.Context) (string, string, []grpc.DialOption, error) + } + + // staticResolver resolves to a fixed address and options, ignoring the + // request context. Its cache key equals its dial target, so two static + // resolvers for the same address with different options would share one + // pooled connection; that does not arise here because upstream hostPorts are + // unique (enforced by config). Contrast the dynamic resolver in + // internal/proxy, which folds the rendered server name into its key. + staticResolver struct { + addr string + opts []grpc.DialOption + } +) + +// NewConn returns a Conn that resolves through r and dials through f. When r is +// static the pooled connection is created eagerly here, so a malformed target +// or bad dial option surfaces at construction rather than on the first request; +// the socket itself is not opened until first use (gRPC connects lazily). A +// dynamic resolver defers creation to the first call that resolves a given +// target. +func NewConn(f ConnFactory, r Resolver) (*Conn, error) { + cc := &Conn{ + factory: f, + resolver: r, + } + + // Static resolvers are eager loaded + if r.IsStatic() { + if _, err := cc.conn(context.Background()); err != nil { + return nil, fmt.Errorf("failed to initialize connection: %w", err) + } + } + + return cc, nil +} + +// StaticResolver returns a Resolver that always resolves to hostPort with the +// given dial options. It reports IsStatic as true, so a Conn built from it is +// created eagerly and reuses a single pooled connection keyed by hostPort. +func StaticResolver(hostPort string, opts ...grpc.DialOption) Resolver { + return &staticResolver{ + addr: hostPort, + opts: opts, + } +} + +// Invoke resolves the connection for this call and forwards the unary RPC to +// it, satisfying [grpc.ClientConnInterface]. +func (c *Conn) Invoke(ctx context.Context, method string, args any, reply any, opts ...grpc.CallOption) error { + cc, err := c.conn(ctx) + if err != nil { + return err + } + + return cc.Invoke(ctx, method, args, reply, opts...) +} + +// NewStream resolves the connection for this call and opens the stream on it, +// satisfying [grpc.ClientConnInterface]. The target is resolved from ctx before +// any message is sent, so streaming and unary calls share one resolution path. +func (c *Conn) NewStream( + ctx context.Context, + desc *grpc.StreamDesc, + method string, + opts ...grpc.CallOption, +) (grpc.ClientStream, error) { + cc, err := c.conn(ctx) + if err != nil { + return nil, err + } + + return cc.NewStream(ctx, desc, method, opts...) +} + +// conn resolves the request and returns the pooled connection for it. +func (c *Conn) conn(ctx context.Context) (*grpc.ClientConn, error) { + key, target, opts, err := c.resolver.Resolve(ctx) + if err != nil { + return nil, err + } + + return c.factory(key, target, opts...) +} + +// IsStatic reports that a staticResolver never varies with the request. +func (r *staticResolver) IsStatic() bool { + return true +} + +// Resolve returns the fixed address as both the cache key and dial target, +// along with the configured dial options. +func (r *staticResolver) Resolve(context.Context) (string, string, []grpc.DialOption, error) { + return r.addr, r.addr, r.opts, nil +} diff --git a/internal/transport/connect/conn_test.go b/internal/transport/connect/conn_test.go new file mode 100644 index 0000000..180f2de --- /dev/null +++ b/internal/transport/connect/conn_test.go @@ -0,0 +1,126 @@ +package connect_test + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/temporalio/temporal-proxy/internal/transport/connect" +) + +type fakeResolver struct { + static bool + key, target string + opts []grpc.DialOption + err error +} + +func TestStaticResolver(t *testing.T) { + t.Parallel() + + r := connect.StaticResolver("host:7233", insecureOpt()) + require.True(t, r.IsStatic()) + + key, target, opts, err := r.Resolve(t.Context()) + require.NoError(t, err) + require.Equal(t, "host:7233", key, "cache key equals the address") + require.Equal(t, "host:7233", target, "dial target equals the address") + require.Len(t, opts, 1) +} + +func TestNewConnEagerLoadsStatic(t *testing.T) { + t.Parallel() + + pool := connect.NewPool() + t.Cleanup(func() { _ = pool.Close() }) + + _, err := connect.NewConn(pool.ConnOrCreate, connect.StaticResolver("passthrough:///eager", insecureOpt())) + require.NoError(t, err) + + // Eager load means the connection is already registered in the pool before + // any request is made. + conn, err := pool.Conn("passthrough:///eager") + require.NoError(t, err) + require.NotNil(t, conn) +} + +func TestNewConnDoesNotDialDynamic(t *testing.T) { + t.Parallel() + + calls := 0 + _, err := connect.NewConn(countingFactory(&calls, nil, nil), fakeResolver{static: false}) + require.NoError(t, err) + require.Zero(t, calls, "a dynamic resolver must not dial at construction") +} + +func TestNewConnStaticFactoryErrorFailsFast(t *testing.T) { + t.Parallel() + + calls := 0 + boom := errors.New("boom") + + _, err := connect.NewConn( + countingFactory(&calls, nil, boom), + connect.StaticResolver("passthrough:///x", insecureOpt()), + ) + require.Error(t, err) + require.ErrorContains(t, err, "failed to initialize connection") + require.ErrorIs(t, err, boom) + require.Equal(t, 1, calls) +} + +func TestConnResolveErrorPropagates(t *testing.T) { + t.Parallel() + + calls := 0 + boom := errors.New("resolve failed") + + c, err := connect.NewConn(countingFactory(&calls, nil, nil), fakeResolver{static: false, err: boom}) + require.NoError(t, err, "dynamic resolver is not resolved until a call is made") + require.ErrorIs(t, c.Invoke(t.Context(), "/svc/Method", nil, nil), boom) + + _, streamErr := c.NewStream(t.Context(), &grpc.StreamDesc{}, "/svc/Method") + require.ErrorIs(t, streamErr, boom) + require.Zero(t, calls, "the factory is never reached when resolution fails") +} + +func TestConnFactoryErrorPropagates(t *testing.T) { + t.Parallel() + + calls := 0 + boom := errors.New("dial failed") + + c, err := connect.NewConn( + countingFactory(&calls, nil, boom), + fakeResolver{static: false, key: "k", target: "t"}, + ) + require.NoError(t, err) + require.ErrorIs(t, c.Invoke(t.Context(), "/svc/Method", nil, nil), boom) + + _, streamErr := c.NewStream(t.Context(), &grpc.StreamDesc{}, "/svc/Method") + require.ErrorIs(t, streamErr, boom) + require.Equal(t, 2, calls, "both Invoke and NewStream resolve through the factory") +} + +func (f fakeResolver) IsStatic() bool { return f.static } + +func (f fakeResolver) Resolve(context.Context) (string, string, []grpc.DialOption, error) { + return f.key, f.target, f.opts, f.err +} + +// countingFactory returns a ConnFactory that records how many times it is +// invoked and always yields conn/err. +func countingFactory(calls *int, conn *grpc.ClientConn, err error) connect.ConnFactory { + return func(string, string, ...grpc.DialOption) (*grpc.ClientConn, error) { + *calls++ + return conn, err + } +} + +func insecureOpt() grpc.DialOption { + return grpc.WithTransportCredentials(insecure.NewCredentials()) +} diff --git a/internal/transport/connect/doc.go b/internal/transport/connect/doc.go index 291016d..b95a3c6 100644 --- a/internal/transport/connect/doc.go +++ b/internal/transport/connect/doc.go @@ -1,5 +1,10 @@ // Package connect manages a pool of reusable gRPC client connections keyed by -// host. Use NewPool to create a Pool, Set to register a connection for a host, -// Conn to retrieve it, GetOrSet to retrieve or lazily dial one, and Close to -// shut every connection down exactly once. +// a caller-supplied logical key, distinct from the dial target. Use NewPool to +// create a Pool, Set to register a connection for a key, Conn to retrieve it, +// ConnOrCreate to retrieve or create one on first use, and Close to shut every connection +// down exactly once. +// +// Conn wraps that pool as a [grpc.ClientConnInterface] whose dial target is +// chosen per call by a Resolver: StaticResolver for a fixed upstream, or a +// caller-supplied dynamic Resolver that varies the target with the request. package connect diff --git a/internal/transport/connect/pool.go b/internal/transport/connect/pool.go index 87fff25..1d96846 100644 --- a/internal/transport/connect/pool.go +++ b/internal/transport/connect/pool.go @@ -9,17 +9,21 @@ import ( ) var ( - // ErrDuplicateHost is returned by Pool.Set when a connection is already - // registered for the given host. - ErrDuplicateHost = errors.New("host already defined") + // ErrDuplicateKey is returned by Pool.Set when a connection is already + // registered for the given key. + ErrDuplicateKey = errors.New("key already defined") - // ErrHostNotFound is returned by Pool.Conn when no connection is registered - // for the given host. - ErrHostNotFound = errors.New("no connection for host") + // ErrKeyNotFound is returned by Pool.Conn when no connection is registered + // for the given key. + ErrKeyNotFound = errors.New("no connection for key") ) type ( - // Pool is a concurrency-safe set of gRPC client connections keyed by host. + // Pool is a concurrency-safe set of gRPC client connections keyed by a + // caller-supplied logical key, which is distinct from the dial target. + // Callers that need two connections to the same dial target (e.g. the same + // host:port with different TLS server names) must use different keys, or + // they will collapse onto whichever connection was dialed first. // The zero value is not usable; create one with NewPool. Pool struct { mu sync.RWMutex @@ -36,38 +40,41 @@ func NewPool() *Pool { } } -// Conn returns the connection registered for host. It returns ErrHostNotFound -// if no connection is registered for that host. -func (p *Pool) Conn(host string) (*grpc.ClientConn, error) { +// Conn returns the connection registered for key. It returns ErrKeyNotFound +// if no connection is registered for that key. +func (p *Pool) Conn(key string) (*grpc.ClientConn, error) { p.mu.RLock() - cn, ok := p.conns[host] + cn, ok := p.conns[key] p.mu.RUnlock() if !ok { - return nil, fmt.Errorf("%w: %s", ErrHostNotFound, host) + return nil, fmt.Errorf("%w: %s", ErrKeyNotFound, key) } return cn, nil } -// GetOrSet returns the connection registered for host, creating and registering -// one with grpc.NewClient(host, opts...) when none exists yet. If callers race to -// create the same host, each dials but only one connection is kept; the losers are -// closed and every caller receives the same *grpc.ClientConn. -func (p *Pool) GetOrSet(host string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { - if conn, _ := p.Conn(host); conn != nil { +// ConnOrCreate returns the connection registered for key, creating and registering +// one with grpc.NewClient(target, opts...) when none exists yet. key is the +// logical cache key and target is the dial address; callers that need distinct +// connections to the same target (e.g. identical host:port with different TLS +// server names) must pass distinct keys. If callers race to create the same +// key, each constructs a client but only one connection is kept; the losers are closed and +// every caller receives the same *grpc.ClientConn. +func (p *Pool) ConnOrCreate(key, target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { + if conn, _ := p.Conn(key); conn != nil { return conn, nil } - conn, err := grpc.NewClient(host, opts...) + conn, err := grpc.NewClient(target, opts...) if err != nil { - return nil, fmt.Errorf("failed to connect to %s, %w", host, err) + return nil, fmt.Errorf("failed to connect to %s, %w", target, err) } - if err := p.Set(host, conn); err != nil { - if errors.Is(err, ErrDuplicateHost) { - _ = conn.Close() // lost the race; don't leak our dial - return p.Conn(host) // return the connection that won + if err := p.Set(key, conn); err != nil { + if errors.Is(err, ErrDuplicateKey) { + _ = conn.Close() // lost the race; don't leak our dial + return p.Conn(key) // return the connection that won } return nil, err @@ -76,17 +83,17 @@ func (p *Pool) GetOrSet(host string, opts ...grpc.DialOption) (*grpc.ClientConn, return conn, nil } -// Set registers conn for host. It returns ErrDuplicateHost if a connection is -// already registered for that host, leaving the existing connection untouched. -func (p *Pool) Set(host string, conn *grpc.ClientConn) error { +// Set registers conn for key. It returns ErrDuplicateKey if a connection is +// already registered for that key, leaving the existing connection untouched. +func (p *Pool) Set(key string, conn *grpc.ClientConn) error { p.mu.Lock() defer p.mu.Unlock() - if _, ok := p.conns[host]; ok { - return fmt.Errorf("%w: %s", ErrDuplicateHost, host) + if _, ok := p.conns[key]; ok { + return fmt.Errorf("%w: %s", ErrDuplicateKey, key) } - p.conns[host] = conn + p.conns[key] = conn return nil } diff --git a/internal/transport/connect/pool_test.go b/internal/transport/connect/pool_test.go index d62cb34..a4f113f 100644 --- a/internal/transport/connect/pool_test.go +++ b/internal/transport/connect/pool_test.go @@ -15,24 +15,24 @@ import ( func TestPoolConn(t *testing.T) { t.Parallel() - t.Run("returns ErrHostNotFound when host is absent", func(t *testing.T) { + t.Run("returns ErrKeyNotFound when key is absent", func(t *testing.T) { t.Parallel() p := connect.NewPool() cn, err := p.Conn("missing") require.Nil(t, cn) - require.ErrorIs(t, err, connect.ErrHostNotFound) + require.ErrorIs(t, err, connect.ErrKeyNotFound) require.Contains(t, err.Error(), "missing") }) - t.Run("returns the connection when host is present", func(t *testing.T) { + t.Run("returns the connection when key is present", func(t *testing.T) { t.Parallel() p := connect.NewPool() conn := newConn(t) - require.NoError(t, p.Set("host", conn)) + require.NoError(t, p.Set("key", conn)) - cn, err := p.Conn("host") + cn, err := p.Conn("key") require.NoError(t, err) require.Same(t, conn, cn) }) @@ -45,18 +45,18 @@ func TestPoolSet(t *testing.T) { t.Parallel() p := connect.NewPool() - require.NoError(t, p.Set("host", newConn(t))) + require.NoError(t, p.Set("key", newConn(t))) }) - t.Run("rejects a duplicate host", func(t *testing.T) { + t.Run("rejects a duplicate key", func(t *testing.T) { t.Parallel() p := connect.NewPool() - require.NoError(t, p.Set("host", newConn(t))) + require.NoError(t, p.Set("key", newConn(t))) - err := p.Set("host", newConn(t)) - require.ErrorIs(t, err, connect.ErrDuplicateHost) - require.Contains(t, err.Error(), "host") + err := p.Set("key", newConn(t)) + require.ErrorIs(t, err, connect.ErrDuplicateKey) + require.Contains(t, err.Error(), "key") }) t.Run("keeps the original connection after a duplicate is rejected", func(t *testing.T) { @@ -64,10 +64,10 @@ func TestPoolSet(t *testing.T) { p := connect.NewPool() original := newConn(t) - require.NoError(t, p.Set("host", original)) - require.Error(t, p.Set("host", newConn(t))) + require.NoError(t, p.Set("key", original)) + require.Error(t, p.Set("key", newConn(t))) - cn, err := p.Conn("host") + cn, err := p.Conn("key") require.NoError(t, err) require.Same(t, original, cn) }) @@ -97,7 +97,7 @@ func TestPoolClose(t *testing.T) { p := connect.NewPool() conn := newConn(t) - require.NoError(t, p.Set("host", conn)) + require.NoError(t, p.Set("key", conn)) // First close shuts the underlying connection down. A second close on // the conn itself would error, so this proves Close short-circuits via @@ -111,7 +111,7 @@ func TestPoolClose(t *testing.T) { p := connect.NewPool() conn := newConn(t) - require.NoError(t, p.Set("host", conn)) + require.NoError(t, p.Set("key", conn)) // Close the connection out from under the pool so the pool's own // Close observes an "already closed" error and surfaces it. @@ -132,7 +132,7 @@ func TestPoolConcurrent(t *testing.T) { const ( workers = 50 - hosts = 8 + keys = 8 ) var wg sync.WaitGroup @@ -143,23 +143,23 @@ func TestPoolConcurrent(t *testing.T) { go func(i int) { defer wg.Done() - host := fmt.Sprintf("host-%d", i%hosts) + key := fmt.Sprintf("key-%d", i%keys) conn := newConn(t) <-start // line every goroutine up so the operations actually overlap - // Multiple goroutines fight over each host: one wins, the rest get - // ErrDuplicateHost. Close the connection we couldn't hand off so it + // Multiple goroutines fight over each key: one wins, the rest get + // ErrDuplicateKey. Close the connection we couldn't hand off so it // doesn't leak. - if err := p.Set(host, conn); err != nil { - require.ErrorIs(t, err, connect.ErrDuplicateHost) + if err := p.Set(key, conn); err != nil { + require.ErrorIs(t, err, connect.ErrDuplicateKey) require.NoError(t, conn.Close()) } - // Conn may or may not find the host yet depending on scheduling; + // Conn may or may not find the key yet depending on scheduling; // both outcomes are valid, we only care that the read is race-free. - if _, err := p.Conn(host); err != nil { - require.ErrorIs(t, err, connect.ErrHostNotFound) + if _, err := p.Conn(key); err != nil { + require.ErrorIs(t, err, connect.ErrKeyNotFound) } // Close racing against Set/Conn must also be safe. closeOnce means @@ -172,37 +172,37 @@ func TestPoolConcurrent(t *testing.T) { wg.Wait() } -func TestPoolGetOrSet(t *testing.T) { +func TestPoolConnOrCreate(t *testing.T) { t.Parallel() insecureOpt := grpc.WithTransportCredentials(insecure.NewCredentials()) - t.Run("creates and registers a connection when host is absent", func(t *testing.T) { + t.Run("creates and registers a connection when key is absent", func(t *testing.T) { t.Parallel() p := connect.NewPool() t.Cleanup(func() { require.NoError(t, p.Close()) }) - conn, err := p.GetOrSet("host", insecureOpt) + conn, err := p.ConnOrCreate("key", "target", insecureOpt) require.NoError(t, err) require.NotNil(t, conn) // The connection is registered, so a subsequent read returns the same one. - cn, err := p.Conn("host") + cn, err := p.Conn("key") require.NoError(t, err) require.Same(t, conn, cn) }) - t.Run("returns the existing connection when host is present", func(t *testing.T) { + t.Run("returns the existing connection when key is present", func(t *testing.T) { t.Parallel() p := connect.NewPool() t.Cleanup(func() { require.NoError(t, p.Close()) }) existing := newConn(t) - require.NoError(t, p.Set("host", existing)) + require.NoError(t, p.Set("key", existing)) - conn, err := p.GetOrSet("host", insecureOpt) + conn, err := p.ConnOrCreate("key", "target", insecureOpt) require.NoError(t, err) require.Same(t, existing, conn) }) @@ -214,15 +214,49 @@ func TestPoolGetOrSet(t *testing.T) { t.Cleanup(func() { require.NoError(t, p.Close()) }) // No transport credentials option, so grpc.NewClient fails synchronously. - conn, err := p.GetOrSet("host") + conn, err := p.ConnOrCreate("key", "target") require.Nil(t, conn) require.ErrorContains(t, err, "failed to connect") }) + t.Run("same target with different keys yields distinct connections", func(t *testing.T) { + t.Parallel() + + // A templated serverName can vary independently of the dial address. + // Two logical keys sharing a target must not collapse onto the same + // pooled connection, or one caller's TLS identity would silently leak + // onto another's channel. + p := connect.NewPool() + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + connA, err := p.ConnOrCreate("key-a", "shared-target", insecureOpt) + require.NoError(t, err) + + connB, err := p.ConnOrCreate("key-b", "shared-target", insecureOpt) + require.NoError(t, err) + + require.NotSame(t, connA, connB) + }) + + t.Run("same key returns the same connection regardless of target", func(t *testing.T) { + t.Parallel() + + p := connect.NewPool() + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + connA, err := p.ConnOrCreate("key", "target-a", insecureOpt) + require.NoError(t, err) + + connB, err := p.ConnOrCreate("key", "target-b", insecureOpt) + require.NoError(t, err) + + require.Same(t, connA, connB) + }) + t.Run("hands every concurrent caller the same connection", func(t *testing.T) { t.Parallel() - // Many goroutines race to create the same host. Only one dial can be + // Many goroutines race to create the same key. Only one dial can be // retained; the losers must be closed and every caller must receive the // winner. This only proves something under `go test -race`. p := connect.NewPool() @@ -238,7 +272,7 @@ func TestPoolGetOrSet(t *testing.T) { wg.Go(func() { <-start // line every goroutine up so the calls actually overlap - conn, err := p.GetOrSet("host", insecureOpt) + conn, err := p.ConnOrCreate("key", "target", insecureOpt) require.NoError(t, err) conns[i] = conn }) diff --git a/internal/transport/creds/mtls.go b/internal/transport/creds/mtls.go index 7c7cb0f..87b3d09 100644 --- a/internal/transport/creds/mtls.go +++ b/internal/transport/creds/mtls.go @@ -5,6 +5,7 @@ import ( "crypto/x509" "fmt" "os" + "sync" "google.golang.org/grpc" "google.golang.org/grpc/credentials" @@ -21,6 +22,7 @@ type ( certFile string keyFile string serverName string + loader *CertLoader } // MTLSOptions holds optional configuration for [MTLS] connections. @@ -29,6 +31,31 @@ type ( // certificate hostname. Useful when the server's certificate CN does not // match its dial address. ServerName string + + // Loader, when set, supplies the client certificate and CA pool used by + // [MTLS.DialOption]. Share one loader across the short-lived MTLS values + // built per request for a templated upstream so the certificate and CA + // files are read and parsed once rather than on every request. When nil, + // the MTLS loads (and caches) its own material on first use. + Loader *CertLoader + } + + // CertLoader reads and parses a client certificate/key pair and a CA + // certificate pool once, caching the result for reuse across [MTLS] values. + // It is safe for concurrent use: the parsed material is immutable after the + // first load, so a cached CA pool and certificate may be shared by many + // dial options. A rotated certificate on disk is not picked up until the + // process restarts, matching how a fixed-address upstream loads its material + // once at startup. + CertLoader struct { + caFile string + certFile string + keyFile string + + once sync.Once + cert tls.Certificate + caPool *x509.CertPool + err error } ) @@ -36,31 +63,33 @@ type ( // caFile and the client certificate/key pair from certFile and keyFile. opts // provides additional configuration; a zero-value [MTLSOptions] is valid. func NewMTLS(caFile, certFile, keyFile string, opts MTLSOptions) *MTLS { + loader := opts.Loader + if loader == nil { + loader = NewCertLoader(caFile, certFile, keyFile) + } + return &MTLS{ caFile: caFile, certFile: certFile, keyFile: keyFile, serverName: opts.ServerName, + loader: loader, } } +// NewCertLoader returns a [CertLoader] that loads the CA certificate from caFile +// and the client certificate/key pair from certFile and keyFile on first use. +func NewCertLoader(caFile, certFile, keyFile string) *CertLoader { + return &CertLoader{caFile: caFile, certFile: certFile, keyFile: keyFile} +} + // DialOption returns a [grpc.DialOption] that configures the outbound // connection with mutual TLS. The client presents its certificate/key pair and // verifies the server against the CA pool loaded from caFile. func (c *MTLS) DialOption() (grpc.DialOption, error) { - cert, err := tls.LoadX509KeyPair(c.certFile, c.keyFile) - if err != nil { - return nil, fmt.Errorf("failed to load client key pair: %w", err) - } - - ca := x509.NewCertPool() - caBytes, err := os.ReadFile(c.caFile) + cert, ca, err := c.loader.load() if err != nil { - return nil, fmt.Errorf("failed to load CA certificate: %w", err) - } - - if ok := ca.AppendCertsFromPEM(caBytes); !ok { - return nil, fmt.Errorf("failed to parse CA file: %s", c.caFile) + return nil, err } return grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{ @@ -137,3 +166,34 @@ func (c *MTLS) Validate() error { func (c *MTLS) Encrypted() bool { return true } + +// load reads and parses the client certificate/key pair and CA pool on the first +// call, caching the result (or the error) for every subsequent call. The +// returned certificate and CA pool are immutable and safe to share across dial +// options and goroutines. +func (l *CertLoader) load() (tls.Certificate, *x509.CertPool, error) { + l.once.Do(func() { + cert, err := tls.LoadX509KeyPair(l.certFile, l.keyFile) + if err != nil { + l.err = fmt.Errorf("failed to load client key pair: %w", err) + return + } + + ca := x509.NewCertPool() + caBytes, err := os.ReadFile(l.caFile) + if err != nil { + l.err = fmt.Errorf("failed to load CA certificate: %w", err) + return + } + + if ok := ca.AppendCertsFromPEM(caBytes); !ok { + l.err = fmt.Errorf("failed to parse CA file: %s", l.caFile) + return + } + + l.cert = cert + l.caPool = ca + }) + + return l.cert, l.caPool, l.err +} diff --git a/internal/transport/creds/mtls_test.go b/internal/transport/creds/mtls_test.go index fe1f442..35b3018 100644 --- a/internal/transport/creds/mtls_test.go +++ b/internal/transport/creds/mtls_test.go @@ -4,6 +4,7 @@ import ( "crypto/x509" "crypto/x509/pkix" "math/big" + "os" "path/filepath" "testing" "time" @@ -249,3 +250,53 @@ func TestMTLS_Encrypted(t *testing.T) { t.Parallel() require.True(t, creds.NewMTLS("", "", "", creds.MTLSOptions{}).Encrypted()) } + +func TestCertLoader(t *testing.T) { + t.Parallel() + + t.Run("shared loader reads material once", func(t *testing.T) { + t.Parallel() + + caFile, certFile, keyFile := testutil.GenerateMTLSCerts(t) + loader := creds.NewCertLoader(caFile, certFile, keyFile) + + // Simulate two per-request credentials for the same upstream sharing the + // loader; only ServerName differs. + opt, err := creds.NewMTLS(caFile, certFile, keyFile, creds.MTLSOptions{ + ServerName: "first.example.com", + Loader: loader, + }).DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + + // Remove the CA file. A re-read would fail at the CA step, so a successful + // second DialOption proves the loader cached the parsed material. + require.NoError(t, os.Remove(caFile)) + + opt, err = creds.NewMTLS(caFile, certFile, keyFile, creds.MTLSOptions{ + ServerName: "second.example.com", + Loader: loader, + }).DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + }) + + t.Run("load error is cached", func(t *testing.T) { + t.Parallel() + + _, certFile, keyFile := testutil.GenerateMTLSCerts(t) + missingCA := filepath.Join(t.TempDir(), "missing.pem") + loader := creds.NewCertLoader(missingCA, certFile, keyFile) + + m := creds.NewMTLS(missingCA, certFile, keyFile, creds.MTLSOptions{Loader: loader}) + + opt, err := m.DialOption() + require.ErrorContains(t, err, "failed to load CA certificate") + require.Nil(t, opt) + + // The cached error is returned again without another disk read. + opt, err = m.DialOption() + require.ErrorContains(t, err, "failed to load CA certificate") + require.Nil(t, opt) + }) +} diff --git a/internal/transport/creds/tls.go b/internal/transport/creds/tls.go index 7314f32..544c56d 100644 --- a/internal/transport/creds/tls.go +++ b/internal/transport/creds/tls.go @@ -28,16 +28,18 @@ var preferredCipherSuites = []uint16{ // presents a certificate to the client; the client is not required to present // one. Minimum TLS version is 1.2. type TLS struct { - certFile string - keyFile string + certFile string + keyFile string + serverName string } // NewClientTLS returns a [TLS] credential suitable for outbound (client-side) // connections. The server's certificate is verified against the system root CA // pool; no client certificate is presented. Use [NewMTLS] when mutual -// authentication is required. -func NewClientTLS() *TLS { - return NewServerTLS("", "") +// authentication is required. serverName overrides the name used for SNI and +// certificate verification; when empty it defaults to the dial target's host. +func NewClientTLS(serverName string) *TLS { + return &TLS{serverName: serverName} } // NewServerTLS returns a [TLS] credential that loads the server certificate and @@ -51,10 +53,11 @@ func NewServerTLS(certFile, keyFile string) *TLS { // DialOption returns a [grpc.DialOption] that configures the outbound // connection with TLS, verifying the server's certificate against the system -// root CA pool. No client certificate is presented; use [MTLS] when mutual -// authentication is required. +// root CA pool. The configured server name (when non-empty) is used for SNI and +// verification; otherwise the dial target's host is used. No client certificate +// is presented; use [MTLS] when mutual authentication is required. func (c *TLS) DialOption() (grpc.DialOption, error) { - return grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), nil + return grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, c.serverName)), nil } // ServerOption returns a [grpc.ServerOption] that configures the server with diff --git a/internal/transport/creds/tls_test.go b/internal/transport/creds/tls_test.go index fcc3e19..3a0b7a5 100644 --- a/internal/transport/creds/tls_test.go +++ b/internal/transport/creds/tls_test.go @@ -17,10 +17,22 @@ import ( func TestTLS_DialOption(t *testing.T) { t.Parallel() - // DialOption does not load any files; file paths are irrelevant. - opt, err := creds.NewClientTLS().DialOption() - require.NoError(t, err) - require.NotNil(t, opt) + t.Run("without a server name", func(t *testing.T) { + t.Parallel() + + // DialOption does not load any files; file paths are irrelevant. + opt, err := creds.NewClientTLS("").DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + }) + + t.Run("with a server name", func(t *testing.T) { + t.Parallel() + + opt, err := creds.NewClientTLS("example.com").DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + }) } func TestTLS_ServerOption(t *testing.T) { @@ -136,7 +148,7 @@ func TestTLS_Validate(t *testing.T) { // NewClientTLS leaves certFile empty; Validate is documented as // server-mode-only, but we still want to confirm it surfaces a clear // IO error rather than panicking. - err := creds.NewClientTLS().Validate() + err := creds.NewClientTLS("").Validate() require.ErrorContains(t, err, "failed to read PEM file") }) } @@ -145,6 +157,6 @@ func TestTLS_Encrypted(t *testing.T) { t.Parallel() // TLS encrypts the transport in both client and server modes. - require.True(t, creds.NewClientTLS().Encrypted()) + require.True(t, creds.NewClientTLS("").Encrypted()) require.True(t, creds.NewServerTLS("", "").Encrypted()) } diff --git a/internal/transport/meta/meta.go b/internal/transport/meta/meta.go new file mode 100644 index 0000000..9b37328 --- /dev/null +++ b/internal/transport/meta/meta.go @@ -0,0 +1,47 @@ +// Package meta defines the internal gRPC metadata contract carried between the +// router and the per-upstream proxy: the router stamps the request namespace it +// already extracted so the proxy can resolve a templated upstream address +// without parsing the payload again. It depends on no other internal packages. +package meta + +import ( + "context" + + "google.golang.org/grpc/metadata" +) + +// NamespaceHeader is the outgoing metadata key that carries the local (pre- +// translation) namespace from the router to the upstream proxy. +const NamespaceHeader = "x-temporal-proxy-namespace" + +// WithNamespace returns ctx with namespace set on its outgoing gRPC metadata, +// replacing any value already present for NamespaceHeader (so a client cannot +// influence routing by sending the header itself). +func WithNamespace(ctx context.Context, namespace string) context.Context { + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + md = metadata.MD{} + } else { + md = md.Copy() + } + + md.Set(NamespaceHeader, namespace) + return metadata.NewOutgoingContext(ctx, md) +} + +// NamespaceFrom returns the namespace carried on ctx's outgoing metadata, or "" +// when absent. When multiple values are present the last (most recently added) +// wins. +func NamespaceFrom(ctx context.Context) string { + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + return "" + } + + vals := md.Get(NamespaceHeader) + if len(vals) == 0 { + return "" + } + + return vals[len(vals)-1] +} diff --git a/internal/transport/meta/meta_test.go b/internal/transport/meta/meta_test.go new file mode 100644 index 0000000..958a626 --- /dev/null +++ b/internal/transport/meta/meta_test.go @@ -0,0 +1,44 @@ +package meta_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc/metadata" + + "github.com/temporalio/temporal-proxy/internal/transport/meta" +) + +func TestWithNamespaceRoundTrips(t *testing.T) { + t.Parallel() + + ctx := meta.WithNamespace(t.Context(), "orders") + require.Equal(t, "orders", meta.NamespaceFrom(ctx)) +} + +func TestWithNamespaceOverwritesExisting(t *testing.T) { + t.Parallel() + + // A spoofed value already present in outgoing metadata is replaced. + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(meta.NamespaceHeader, "spoofed")) + ctx = meta.WithNamespace(ctx, "orders") + + md, _ := metadata.FromOutgoingContext(ctx) + require.Equal(t, []string{"orders"}, md.Get(meta.NamespaceHeader)) +} + +func TestNamespaceFromReturnsLastValue(t *testing.T) { + t.Parallel() + + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs( + meta.NamespaceHeader, "old", + meta.NamespaceHeader, "new", + )) + require.Equal(t, "new", meta.NamespaceFrom(ctx)) +} + +func TestNamespaceFromAbsentIsEmpty(t *testing.T) { + t.Parallel() + + require.Equal(t, "", meta.NamespaceFrom(t.Context())) +} diff --git a/rfc/routing.md b/rfc/routing.md index cae60e0..ff1f816 100644 --- a/rfc/routing.md +++ b/rfc/routing.md @@ -2,83 +2,119 @@ How to handle routing to multiple upstream clusters based on namespace. -# Background +## Background -The proxy currently supports a single static upstream defined in config. However, we want a single proxy (tenant) to serve many namespaces, each potentially terminating at a different address/upstream. +The proxy currently supports a single static upstream defined in config. However, we want a single proxy (tenant) to +serve many namespaces, each potentially terminating at a different address/upstream. -This can be for a number of reasons, but notably, during migrations, some teams will want to move over namespace-by-namespace rather than cluster-by-cluster. They’d point all namespaces to the original upstream (typically the OSS frontend) and switch to another one as needed based on some routing rules. +This can be for a number of reasons, but notably, during migrations, some teams will want to move over +namespace-by-namespace rather than cluster-by-cluster. They’d point all namespaces to the original upstream (typically +the OSS frontend) and switch to another one as needed based on some routing rules. > [!NOTE] -> Cloud offers three relevant flavours of endpoint: per-namespace endpoints (`..tmprl.cloud:7233`), regional endpoints (`.aws.tmprl.cloud:7233`), and private-link endpoints (per-VPC hostnames). Operators also need to mix Cloud and self-hosted upstreams in a single proxy. > -> Ideally, we’d use the namespace endpoints since they’re already set up to handle multi-region traffic. However, there are times when that isn’t possible, so we should support all cases. +> Cloud offers three relevant flavours of endpoint: per-namespace endpoints (`..tmprl.cloud:7233`), +> regional endpoints (`.aws.tmprl.cloud:7233`), and private-link endpoints (per-VPC hostnames). Operators also +> need to mix Cloud and self-hosted upstreams in a single proxy. +> +> Ideally, we’d use the namespace endpoints since they’re already set up to handle multi-region traffic. However, there +> are times when that isn’t possible, so we should support all cases. -# Proposal +## Proposal -We need to support multiple named upstream configs in a single proxy and select among them (per request) using a set of routing rules. These rules will have access to the incoming namespace (before and after translation) and to gRPC metadata headers. +We need to support multiple named upstream configs in a single proxy and select among them (per request) using a set of +routing rules. These rules will have access to the incoming namespace (before and after translation) and to gRPC +metadata headers. -## Upstreams +### Upstreams -As part of this, we’ll need to exchange the single upstream we have today for multiple upstream definitions. Each upstream will have its `hostPort` and `tls.serverName` fields templated, allowing for the following replacements: +As part of this, we’ll need to exchange the single upstream we have today for multiple upstream definitions. Each +upstream will have its `hostPort` and `tls.serverName` fields templated, allowing for the following replacements: -* `{{ .LocalNamespace }}` \- The namespace before translation -* `{{ .RemoteNamespace }}` \- The namespace after translation -* `{{ .Metadata.DC }}` \- The `DC` metadata value (arbitrary) -* `{{ index .Metadata “x-cluster” }}` \- The `x-cluster` metadata value (arbitrary) +- `{{ .LocalNamespace }}` \- The namespace before translation +- `{{ .RemoteNamespace }}` \- The namespace after translation +- `{{ .Metadata.DC }}` \- The `DC` metadata value (arbitrary) +- `{{ index .Metadata “x-cluster” }}` \- The `x-cluster` metadata value (arbitrary) -To be clear, these are simply available, but not required. Static `hostPort/serverName` values are not a problem (see the example config [below](#configuration-schema)). +To be clear, these are simply available, but not required. Static `hostPort/serverName` values are not a problem (see +the example config [below](#configuration-schema)). -To minimize runtime issues, all static `hostPort` upstreams will establish eager connections at startup. Templated values are not predetermined and therefore cannot benefit from this. Those connections will be created lazily on first use. +To minimize runtime issues, all static `hostPort` upstreams will establish eager connections at startup. Templated +values are not predetermined and therefore cannot benefit from this. Those connections will be created lazily on first +use. -All connections will be maintained in a shared container (TBD) since connection establishment is expensive. Each connection will be a gRPC `ClientConn,` ensuring automatic redials and shared underlying channels via connection pooling. +All connections will be maintained in a shared container (TBD) since connection establishment is expensive. Each +connection will be a gRPC `ClientConn,` ensuring automatic redials and shared underlying channels via connection +pooling. -## Routing +### Routing -To match incoming requests to the appropriate upstream, we’ll need to define routing rules. In proxy config, we can define a set of rules that grant access to request information, which can be used to select an upstream connection. +To match incoming requests to the appropriate upstream, we’ll need to define routing rules. In proxy config, we can +define a set of rules that grant access to request information, which can be used to select an upstream connection. -These rules will be evaluated in the order in which they are defined. The first matching rule will be used to route the request, with one notable exception: requests without a namespace (system requests). These will be sent to a particular upstream defined as the `system` upstream. +These rules will be evaluated in the order in which they are defined. The first matching rule will be used to route the +request, with one notable exception: requests without a namespace (system requests). These will be sent to a particular +upstream defined as the `system` upstream. -The other special case is handling the absence of a match. It seems perfectly reasonable to reject the request with an error, but some operators may prefer to fall back to a specific upstream (e.g., receiving Temporal’s `NamespaceNotFound` rather than a proxy-specific error). +The other special case is handling the absence of a match. It seems perfectly reasonable to reject the request with an +error, but some operators may prefer to fall back to a specific upstream (e.g., receiving Temporal’s `NamespaceNotFound` +rather than a proxy-specific error). -### Rules +#### Rules -To route incoming requests to the appropriate upstream, we’ll need to define a mapping in the proxy config. The mapping can be done using the incoming namespace and/or gRPC metadata attached to the request. A rule is defined with a matcher that specifies the namespace and the metadata values to match, and an upstream to route to. +To route incoming requests to the appropriate upstream, we’ll need to define a mapping in the proxy config. The mapping +can be done using the incoming namespace and/or gRPC metadata attached to the request. A rule is defined with a matcher +that specifies the namespace and the metadata values to match, and an upstream to route to. -Rules use `AND` logic to match. This means that defining a rule for `namespace=test` that contains a metadata `key=value` pair requires both conditions to be true to match. While more complex logic could be added, it also introduces complexity and the risk of logic errors in the config. +Rules use `AND` logic to match. This means that defining a rule for `namespace=test` that contains a metadata +`key=value` pair requires both conditions to be true to match. While more complex logic could be added, it also +introduces complexity and the risk of logic errors in the config. It is an error to define a rule without a namespace or metadata values. This will fail validation on startup. -#### Namespace Rules +##### Namespace Rules -Since namespace translation is defined per upstream and we don’t yet know which upstream we’re using, the local namespace (before translation) is used to perform the match. When a namespace is not defined on the rule, no namespace-based matching will be performed. +Since namespace translation is defined per upstream and we don’t yet know which upstream we’re using, the local +namespace (before translation) is used to perform the match. When a namespace is not defined on the rule, no +namespace-based matching will be performed. -Namespace matches can be defined as string literals or simple globs. This supports exact-match, starts-with, ends-with, and contains semantics. Here are some valid examples: +Namespace matches can be defined as string literals or simple globs. This supports exact-match, starts-with, ends-with, +and contains semantics. Here are some valid examples: -* `my-namespace` -* `prod-*` -* `*-test` -* `*-test-*` +- `my-namespace` +- `prod-*` +- `*-test` +- `*-test-*` -#### Metadata Rules +##### Metadata Rules -Key/value pairs can be added to the match to validate incoming header metadata. When multiple conditions are defined, they are `AND`ed (i.e. all conditions must match). +Key/value pairs can be added to the match to validate incoming header metadata. When multiple conditions are defined, +they are `AND`ed (i.e. all conditions must match). -A header condition contains a map of keys and values. Keys are not case-sensitive (normalized to lowercase) but do not support wildcards. The intent here is to allow operators to make routing decisions based on known metadata (e.g. `X-Data-Centre`). Values follow the same convention as namespaces, with support for exact-match, starts-with, ends-with, and contains semantics. +A header condition contains a map of keys and values. Keys are not case-sensitive (normalized to lowercase) but do not +support wildcards. The intent here is to allow operators to make routing decisions based on known metadata (e.g. +`X-Data-Centre`). Values follow the same convention as namespaces, with support for exact-match, starts-with, ends-with, +and contains semantics. -### System Upstream +#### System Upstream -Not all RPCs have a namespace. For example, `GetClusterInfo` or `GetSystemInfo`. When these are called, we need to ensure they get sent to the appropriate upstream. To handle this, the routing config has a `system` field which can be set to a known upstream. +Not all RPCs have a namespace. For example, `GetClusterInfo` or `GetSystemInfo`. When these are called, we need to +ensure they get sent to the appropriate upstream. To handle this, the routing config has a `system` field which can be +set to a known upstream. When not defined, requests without a namespace fallback to the default upstream (see below). -### Default Upstream +#### Default Upstream -When a request cannot be matched, that is to say, no routing rules apply, the proxy will return an error (e.g. `RouteNotFound`). This isn’t always what operators desire. Sometimes they’d like a fallback that sends the request to a default upstream. +When a request cannot be matched, that is to say, no routing rules apply, the proxy will return an error (e.g. +`RouteNotFound`). This isn’t always what operators desire. Sometimes they’d like a fallback that sends the request to a +default upstream. -For these cases, a `default` upstream can be defined in the routing rules. This will ensure all requests are proxied regardless of namespace or metadata. +For these cases, a `default` upstream can be defined in the routing rules. This will ensure all requests are proxied +regardless of namespace or metadata. -## Configuration Schema +### Configuration Schema -Based on the information above, here’s an example of what the config could look like. +Based on the information above, here’s an example of what the config could look like. ```yaml hostPort: 0.0.0.0:7233